"""Acceso y lógica de los artefactos de EJES de ataque (por liga-temporada). Los artefactos (chicos, ~KB-MB) se generan offline con scripts/ejes_build_artifacts.py del repo Racing y viven en Azure bajo ``{report_artifacts_root}/ejes///``: team_profiles.parquet equipo × eje × lado(gen/con): mix_z + real/esperado en eje-alto match_team_ejes.parquet matchId × equipo: medias de eje + tiros real/esperado + goles + fecha match_team_vars.parquet matchId × equipo: medias de variables (drill-down "qué se movió") scales.json σ partido-a-partido, medias/σ de liga, normas por equipo (gen y con) predictor.json por eje: coefs {gen, con, intercept} (gen propia + concesión rival) En dev, ``EJES_ARTIFACTS_DIR`` apunta al directorio local y evita Azure. """ from __future__ import annotations import json import os import time from functools import lru_cache from pathlib import Path import pandas as pd from racing_reports.config import DEFAULT_SETTINGS from racing_reports.datastore import DataStore EJE_LABELS = { "verticalidad": "Verticalidad", "elaboracion": "Elaboración", "individual": "Juego individual", "rotura_lineas": "Rotura de líneas", "amplitud_centro": "Amplitud / centros", "profundidad_espalda": "Profundidad / espalda", "robo_alto": "Robo alto", "aereo_segunda": "Aéreo / 2ª jugada", } EJES = list(EJE_LABELS) # Nombres legibles de las variables de secuencia (estilo del diccionario_variables.xlsx; # las que existen ahí usan su nombre, el resto se completó en el mismo estilo). VAR_LABELS = { "atk_l": "Largo del bloque propio (m)", "atk_w": "Ancho del bloque propio (m)", "ball_distance": "Distancia recorrida por el balón (m)", "behind_line_ok": "Pases a la espalda completados", "broke_last_line": "Roturas de la última línea", "broke_second_last": "Roturas de la penúltima línea", "cross_pvadded_max": "Peligro del mejor centro (PV)", "dang_runs": "Corridas peligrosas", "deep_completions": "Pases profundos completados", "def_h": "Altura de la línea defensiva rival", "dribbles_box": "Regates ganados en zona de área", "dribbles_last_third": "Regates en último tercio", "duration_s": "Duración de la jugada (s)", "hold_max": "Máxima tenencia individual (s)", "hold_mean": "Tenencia media por toque (s)", "hold_min": "Mínima tenencia individual (s)", "line_breaks": "Acciones que rompen líneas", "max_x": "Punto más alto alcanzado (x)", "n_centros": "Centros", "n_events": "Acciones de la jugada", "n_involved": "Jugadores involucrados", "n_passes": "Pases de la jugada", "opt_quality_mean": "Calidad media de opciones de pase", "paredes": "Paredes (uno-dos)", "pass_options_mean": "Receptores disponibles por pase", "passes_behind_line": "Pases a la espalda intentados", "pct_adentro": "% pases hacia el centro", "pct_afuera": "% pases hacia afuera", "pct_cortos": "% pases cortos", "pct_headpass": "% pases de cabeza", "pct_largos": "% pases largos", "pct_launch": "% pelotazos (launch)", "pct_layoff": "% descargas (lay-off)", "pct_medios": "% pases medios", "pct_switch": "% cambios de orientación", "pct_through": "% pases entre líneas", "pct_verticales": "% pases verticales", "pelotazos_espalda": "Pelotazos a la espalda", "press_final": "Presión recibida en último tercio", "press_media": "Presión recibida media", "prog_x": "Progresión de la jugada (x)", "start_x": "Altura de inicio de la jugada (x)", "starts_from_dispossess": "Jugadas nacidas de robo", "takeons_ok": "Regates ganados (1v1)", "tempo": "Ritmo (pases por segundo)", "transition_speed": "Velocidad de transición", "verticalidad_media": "Verticalidad media del pase", } def var_label(var: str) -> str: return VAR_LABELS.get(var, var) _AVAIL_CACHE: dict[str, object] = {"ts": 0.0, "value": None} _AVAIL_TTL = 1800.0 def _slug(league: str) -> str: return "".join(c.lower() if c.isalnum() else "-" for c in league).strip("-") def _local_dir() -> Path | None: raw = os.getenv("EJES_ARTIFACTS_DIR") return Path(raw).expanduser() if raw else None def _bundled_dir() -> Path: from racing_reports import vendor_env return vendor_env.DATA_DIR / "ejes" def _path(league: str, season: str, fname: str) -> Path: """Orden: dir local de dev (env) → bundle del repo (vendor/data/ejes) → Azure.""" rel = f"ejes/{_slug(league)}/{season}/{fname}" local = _local_dir() if local is not None: p = local / _slug(league) / season / fname if p.exists(): return p raise FileNotFoundError(f"No existe el artefacto local: {p}") bundled = _bundled_dir() / _slug(league) / season / fname if bundled.exists(): return bundled return DataStore().sync_artifact(rel) @lru_cache(maxsize=64) def _scales(league: str, season: str) -> dict: return json.loads(_path(league, season, "scales.json").read_text(encoding="utf-8")) @lru_cache(maxsize=64) def _predictor(league: str, season: str) -> dict: return json.loads(_path(league, season, "predictor.json").read_text(encoding="utf-8")) @lru_cache(maxsize=32) def _match_ejes(league: str, season: str) -> pd.DataFrame: return pd.read_parquet(_path(league, season, "match_team_ejes.parquet")) @lru_cache(maxsize=32) def _match_vars(league: str, season: str) -> pd.DataFrame: return pd.read_parquet(_path(league, season, "match_team_vars.parquet")) @lru_cache(maxsize=32) def _profiles(league: str, season: str) -> pd.DataFrame: return pd.read_parquet(_path(league, season, "team_profiles.parquet")) def available() -> dict[str, list[str]]: """{liga: [temporadas]} con artefactos de ejes. Local si EJES_ARTIFACTS_DIR; si no, Azure.""" def _scan(base: Path) -> dict[str, list[str]]: idx = {} idx_path = base / "_index.json" if idx_path.exists(): idx = json.loads(idx_path.read_text(encoding="utf-8")) out: dict[str, list[str]] = {} for p in sorted(base.glob("*/*/team_profiles.parquet")): slug = p.parent.parent.name out.setdefault(idx.get(slug, slug), []).append(p.parent.name) return out local = _local_dir() if local is not None: return _scan(local) bundled = _bundled_dir() if bundled.is_dir(): out = _scan(bundled) if out: return out now = time.time() if _AVAIL_CACHE["value"] is not None and now - float(_AVAIL_CACHE["ts"]) < _AVAIL_TTL: return _AVAIL_CACHE["value"] # type: ignore[return-value] root = f"{DEFAULT_SETTINGS.azure_report_artifacts_root}/ejes".strip("/") found: dict[str, list[str]] = {} try: fs = DataStore()._filesystem_client() for p in fs.get_paths(path=root, recursive=True): name = getattr(p, "name", "") or "" if not name.endswith("team_profiles.parquet"): continue parts = name[len(root):].strip("/").split("/") if len(parts) >= 3: found.setdefault(parts[0], []).append(parts[1]) except Exception: pass _AVAIL_CACHE["value"] = found _AVAIL_CACHE["ts"] = now return found def teams(league: str, season: str) -> list[str]: return sorted(_profiles(league, season)["equipo"].unique()) def _eje_z(scales: dict, eje_col: str, value: float) -> float: mu = scales["eje_match_mean"].get(eje_col, 0.0) sd = scales["eje_match_std"].get(eje_col, 1.0) or 1.0 return (value - mu) / sd def pre_prediction(league: str, season: str, home: str, away: str) -> dict: """Para cada equipo: por eje, su norma y la predicción del partido (crece/decrece).""" sc = _scales(league, season) pred = _predictor(league, season) out = [] for team, rival in [(home, away), (away, home)]: gen = sc["team_eje_norm_gen"].get(team) con = sc["team_eje_norm_con"].get(rival) if gen is None or con is None: raise ValueError(f"Sin perfil de ejes para {team if gen is None else rival} " f"en {league} {season}") rows = [] for eje in EJES: ec = f"eje_{eje}" c = pred.get(eje) norma = gen.get(ec, 0.0) if c is None: p = norma else: p = c["intercept"] + c["gen"] * norma + c["con"] * con.get(ec, 0.0) sigma = sc["eje_sigma"].get(ec, 1.0) or 1.0 rows.append(dict( eje=eje, label=EJE_LABELS[eje], norma_z=round(_eje_z(sc, ec, norma), 2), pred_z=round(_eje_z(sc, ec, p), 2), delta_sigma=round((p - norma) / sigma, 2), )) out.append(dict(team=team, rival=rival, rows=rows)) return {"league": league, "season": season, "teams": out} def _resolve_match(mt: pd.DataFrame, home: str, away: str, match_id: str | None, match_date: str | None) -> pd.DataFrame: pair = mt[((mt["team"] == home) & (mt["rival"] == away)) | ((mt["team"] == away) & (mt["rival"] == home))] if match_id: got = pair[pair["matchId"] == str(match_id)] if len(got): return got if match_date and "fecha" in pair.columns: got = pair[pair["fecha"].astype(str).str.startswith(str(match_date)[:10])] if len(got): return got if pair.empty: raise ValueError(f"No hay partidos {home} vs {away} en los artefactos.") if "fecha" in pair.columns: last = pair["fecha"].astype(str).max() return pair[pair["fecha"].astype(str) == last] return pair[pair["matchId"] == pair["matchId"].iloc[-1]] def post_signature(league: str, season: str, home: str, away: str, match_id: str | None = None, match_date: str | None = None, top_movers: int = 8) -> dict: """Firma de ataque del partido: ejes vs norma, ejecución esperado-vs-real, variables movidas.""" sc = _scales(league, season) mt = _match_ejes(league, season) mv = _match_vars(league, season) rows_match = _resolve_match(mt, home, away, match_id, match_date) mid = rows_match["matchId"].iloc[0] fecha = str(rows_match["fecha"].iloc[0]) if "fecha" in rows_match.columns else "" teams_out = [] for _, r in rows_match.iterrows(): team = r["team"] norm = sc["team_eje_norm_gen"].get(team, {}) ejes_rows = [] for eje in EJES: ec = f"eje_{eje}" hoy, norma = float(r[ec]), float(norm.get(ec, 0.0)) sigma = sc["eje_sigma"].get(ec, 1.0) or 1.0 ejes_rows.append(dict( eje=eje, label=EJE_LABELS[eje], partido_z=round(_eje_z(sc, ec, hoy), 2), norma_z=round(_eje_z(sc, ec, norma), 2), dev_sigma=round((hoy - norma) / sigma, 2), )) movers = [] vrow = mv[(mv["matchId"] == mid) & (mv["team"] == team)] if len(vrow): vrow = vrow.iloc[0] vnorm = sc["team_var_norm"].get(team, {}) for var, sigma in sc["var_sigma"].items(): if var not in vrow or not sigma: continue hoy_v = float(vrow[var]) norma_v = float(vnorm.get(var, 0.0)) movers.append(dict(variable=var, label=var_label(var), hoy=round(hoy_v, 2), norma=round(norma_v, 2), dev_sigma=round((hoy_v - norma_v) / sigma, 2))) movers.sort(key=lambda d: -abs(d["dev_sigma"])) movers = movers[:top_movers] teams_out.append(dict( team=team, rival=r["rival"], n_seq=int(r["n_seq"]), tiros_real=int(r["tiros_real"]), tiros_esperados=round(float(r["tiros_esp"]), 1), goles=int(r["goles"]), ejes=ejes_rows, movers=movers, )) return {"league": league, "season": season, "match_id": mid, "fecha": fecha, "teams": teams_out} def _payload_rows(league: str, season: str) -> list[dict]: df = _profiles(league, season) tmp: dict[str, dict] = {} for _, r in df.iterrows(): eq = tmp.setdefault(r["equipo"], {"equipo": r["equipo"], "liga": league, "temporada": season, "gen": {}, "con": {}}) dif = (r["real"] - r["liga_real"]) if pd.notna(r["real"]) else None eq[r["lado"]][r["eje"]] = { "z": float(r["mix_z"]), "g": float(r["mix_global"]) if "mix_global" in r and pd.notna(r.get("mix_global")) else None, "dif": round(float(dif), 1) if dif is not None else None, } return list(tmp.values()) def profiles_payload(league: str, season: str) -> dict: """Payload del scatter. ``league='__all__'`` devuelve TODAS las ligas (cada equipo con su liga/temporada; z = vs su liga, g = score en unidades globales para re-estandarizar).""" if league == "__all__": teams: list[dict] = [] for lg, seasons in available().items(): for s in seasons: try: teams.extend(_payload_rows(lg, s)) except FileNotFoundError: continue return {"league": "__all__", "season": None, "ejes": EJE_LABELS, "teams": teams} return {"league": league, "season": season, "ejes": EJE_LABELS, "teams": _payload_rows(league, season)}