Spaces:
Running
Running
| """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/<slug>/<season>/``: | |
| 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_profundidad": "Rotura de líneas / profundidad", | |
| "amplitud_centro": "Amplitud / centros", | |
| "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)", | |
| "indice_verticalidad": "Índice de verticalidad de la posesión", | |
| "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_prog_rival": "% pases progresivos en campo rival", | |
| "pct_switch": "% cambios de orientación", | |
| "pct_through": "% pases entre líneas", | |
| "pct_verticales": "% pases verticales", | |
| "pelotazos_espalda": "Pelotazos a la espalda", | |
| "ppda": "PPDA (pases rival por acción defensiva; menos = más presión)", | |
| "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) | |
| def _scales(league: str, season: str) -> dict: | |
| return json.loads(_path(league, season, "scales.json").read_text(encoding="utf-8")) | |
| def _predictor(league: str, season: str) -> dict: | |
| return json.loads(_path(league, season, "predictor.json").read_text(encoding="utf-8")) | |
| def _match_ejes(league: str, season: str) -> pd.DataFrame: | |
| return pd.read_parquet(_path(league, season, "match_team_ejes.parquet")) | |
| def _match_vars(league: str, season: str) -> pd.DataFrame: | |
| return pd.read_parquet(_path(league, season, "match_team_vars.parquet")) | |
| 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 _norm_for(scales: dict, team: str, venue: str, kind: str): | |
| """Norma (gen/con) de un equipo con split de localía y caída a la combinada.""" | |
| combined = scales.get(f"team_eje_norm_{kind}", {}) | |
| split = scales.get(f"team_eje_norm_{kind}_{venue}", {}) | |
| return split.get(team) or combined.get(team) | |
| def pre_prediction(league: str, season: str, home: str, away: str, | |
| subs: dict[str, tuple[str, str]] | None = None) -> dict: | |
| """Para cada equipo: por eje, su norma y la predicción del partido (crece/decrece). | |
| Usa normas separadas por localía: el LOCAL ataca con su norma de local contra la | |
| concesión de VISITANTE del rival, y viceversa. Cae a la norma combinada si falta el | |
| split (artefactos viejos o equipo sin partidos en esa condición). | |
| ``subs`` (opcional): {equipo: (liga, temporada)} — para un equipo con muestra chica | |
| en la temporada pedida, su NORMA sale de esa (liga, temporada) anterior; el predictor | |
| y la escala z siguen siendo los de la liga mostrada (mismas unidades de eje).""" | |
| subs = subs or {} | |
| sc = _scales(league, season) # predictor / z-score / sigma: contexto de la liga mostrada | |
| pred = _predictor(league, season) | |
| out = [] | |
| # (equipo, rival, equipo_es_local) | |
| for team, rival, team_home in [(home, away, True), (away, home, False)]: | |
| gvenue = "home" if team_home else "away" # localía del equipo que ataca | |
| cvenue = "away" if team_home else "home" # el rival defiende en la condición opuesta | |
| gsc = _scales(*subs[team]) if team in subs else sc | |
| csc = _scales(*subs[rival]) if rival in subs else sc | |
| gen = _norm_for(gsc, team, gvenue, "gen") | |
| con = _norm_for(csc, rival, cvenue, "con") | |
| 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), | |
| )) | |
| item = dict(team=team, rival=rival, | |
| venue="local" if team_home else "visitante", rows=rows) | |
| if team in subs: | |
| item["norma_source"] = f"{subs[team][1]}" # temporada de la que sale la norma | |
| out.append(item) | |
| 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, | |
| "n": int(r["n_alto"]) if "n_alto" in r and pd.notna(r.get("n_alto")) else 0, | |
| } | |
| return list(tmp.values()) | |
| # Un eje se calcula sobre "secuencias-alto" (n_alto). Con MUY POCAS secuencias (partido suelto de | |
| # inicio de temporada, o datos que no entraron al pipeline) el valor sale basura de dos formas: | |
| # explota (|g| = 8..70; visto elaboración = 41.7 contra ±0.1 del resto de su liga) o queda acotado | |
| # pero sin sentido (n_alto = 0). Además ese valor infla el desvío de la liga y APLASTA el z de los | |
| # demás contra el origen. Por eso se anulan esas celdas y se recomputa el z dentro de cada | |
| # (liga, temporada, lado, eje) con los valores limpios. MIN_N=3 no toca a equipos legítimos con | |
| # pocas secuencias (p.ej. robo_alto con n_alto=4). El z guardado ES el z-score de g dentro de la | |
| # liga (corr 0.997), así que recomputarlo es consistente con el original. | |
| EJE_MIN_N = 3 | |
| EJE_MAX_G = 3.0 | |
| def _clean_payload(teams: list[dict]) -> list[dict]: | |
| """Anula ejes con muestra insuficiente/reventada y recomputa el z dentro de liga-temporada.""" | |
| for t in teams: # 1) anular celdas con muestra insuficiente | |
| for lado in ("gen", "con"): | |
| for c in (t.get(lado) or {}).values(): | |
| g = c.get("g") | |
| if (c.get("n") or 0) < EJE_MIN_N or (g is not None and abs(g) > EJE_MAX_G): | |
| c["z"] = c["g"] = c["dif"] = None | |
| buckets: dict = {} # 2) recomputar z por (liga,temp,lado,eje) | |
| for t in teams: | |
| for lado in ("gen", "con"): | |
| for eje, c in (t.get(lado) or {}).items(): | |
| if c.get("g") is not None: | |
| buckets.setdefault((t["liga"], str(t["temporada"]), lado, eje), []).append(c) | |
| for cells in buckets.values(): | |
| gs = [c["g"] for c in cells] | |
| if len(gs) < 2: | |
| continue | |
| mu = sum(gs) / len(gs) | |
| sd = (sum((x - mu) ** 2 for x in gs) / len(gs)) ** 0.5 | |
| if sd < 1e-9: | |
| continue | |
| for c in cells: | |
| c["z"] = round((c["g"] - mu) / sd, 2) | |
| for t in teams: # "n" es interno: no sale en el payload | |
| for lado in ("gen", "con"): | |
| for c in (t.get(lado) or {}).values(): | |
| c.pop("n", None) | |
| return teams | |
| 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": _clean_payload(teams)} | |
| return {"league": league, "season": season, "ejes": EJE_LABELS, | |
| "teams": _clean_payload(_payload_rows(league, season))} | |