Spaces:
Running
Running
| """Data del modelo ofensivo/defensivo 0–10, a nivel PARTIDO. | |
| Lo usan dos apps de Vercel: la de **Estilos de juego** (que es todo este modelo) y la | |
| sección *Ejes ataque/defensa* de **Equipos y variables** (que muestra el mapa global y las | |
| matrices arriba de su scatter de ejes). Por eso el export vive acá y no en el build de una. | |
| El modelo necesita el partido, no el promedio de temporada: los bloques defensivos se miden | |
| sobre **el rival de ese encuentro**, emparejando las dos filas de cada ``matchid``. Y el | |
| score 0–10 se recalcula sobre las ligas que el usuario elige, así que acá solo se serializa | |
| la data cruda y la matemática corre en el navegador. | |
| Salida en ``out_dir``: | |
| index.json ligas, temporadas, equipos, partidos por equipo y catálogo de variables | |
| <liga>.json columnas del MODELO por equipo-partido (una por liga, carga perezosa) | |
| <liga>.vars.json el resto de las variables (opcional: solo lo pide "Variables determinantes") | |
| crests.json escudos de los equipos presentes, con la grafía ya unificada | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import unicodedata | |
| from pathlib import Path | |
| import pandas as pd | |
| from racing_reports.league_names import canonical | |
| from racing_reports.team_names import apply_aliases, canonical_team | |
| from racing_reports.team_vars import TRACKING_ZERO_IS_MISSING | |
| REPO = Path(__file__).resolve().parents[2] | |
| DEFAULT_SRC = REPO / "vendor" / "data" / "match_team_stats.parquet" | |
| DEFAULT_CRESTS = REPO / "webapp-static" / "site" / "crests.json" | |
| DEFAULT_VAR_DICT = REPO / "vendor" / "data" / "var_dict.json" | |
| META = ["competencia", "temporada", "equipo", "rival", "fecha", "matchid", | |
| "formacion", "formacion_rival"] | |
| # Columnas que consume el modelo (las del Streamlit original). El rival sale del self-join | |
| # por matchid, así que no hacen falta las _rival. | |
| MODEL_COLS = [ | |
| "pct_posesion", | |
| "pases_totales", "pases_exitosos", "pases_no_exitosos", "shots", | |
| "pases_endx_1_2_tercio", "pases_endx_totales_1_2_tercio", "pct_pases_exitosos_t1-2", | |
| "pases_progresivos", "pases_progresivos_exitosos", "pases_progresivos_no_exitosos", | |
| "pases_progresivos_sobre_pases", "pct_pases_largos", | |
| "pases_endx_tercio_final", "pases_totales_a_ultimo_tercio", "pct_pases_exitosos_t3", | |
| "pases_totales_ultimo_tercio", "centros_totales_ultimo_tercio", | |
| "avg_end_x_pass", "ancho_promedio_m_por_posesion", "altura_media_recuperacion", | |
| "ritmo_circulacion_pases_por_min", "pases_prom_secuencia", | |
| "acciones_defensivas_en_ofensiva", "acciones_defensivas_totales", "ppda", | |
| "goles_totales", | |
| ] | |
| # Sin umbral: con un partido ya entra. Al arrancar la temporada el score es ruido y se | |
| # acomoda cuando llegan más fechas; el frontend avisa (ver `pj_por_temporada`). | |
| MIN_PARTIDOS_EQUIPO = 1 | |
| def slug(s: str) -> str: | |
| s = unicodedata.normalize("NFKD", str(s)).encode("ascii", "ignore").decode() | |
| return "".join(c.lower() if c.isalnum() else "-" for c in s).strip("-") | |
| def _col(series: pd.Series, nd: int = 4) -> list: | |
| v = pd.to_numeric(series, errors="coerce").round(nd) | |
| return [None if pd.isna(x) else (int(x) if float(x).is_integer() and abs(x) < 1e15 else float(x)) | |
| for x in v] | |
| def _idx(series: pd.Series) -> tuple[list, list]: | |
| vals = series.fillna("").astype(str) | |
| uniq = sorted(vals.unique()) | |
| pos = {v: i for i, v in enumerate(uniq)} | |
| return uniq, [pos[v] for v in vals] | |
| def _dump(path: Path, obj) -> None: | |
| path.write_text(json.dumps(obj, ensure_ascii=False, separators=(",", ":")), encoding="utf-8") | |
| def export(out_dir: Path, src: Path | None = None, with_vars: bool = True, | |
| crests: Path | None = None, var_dict: Path | None = None) -> dict: | |
| src = src or DEFAULT_SRC | |
| if not src.exists(): | |
| raise FileNotFoundError(f"Falta {src}. Corré: python scripts/build_match_team_stats.py") | |
| df = pd.read_parquet(src) | |
| df["competencia"] = df["competencia"].map(canonical).fillna(df["competencia"]) | |
| df["temporada"] = df["temporada"].astype(str) | |
| apply_aliases(df) | |
| for c in TRACKING_ZERO_IS_MISSING: # el 0 de tracking es dato faltante | |
| if c in df.columns: | |
| df[c] = df[c].mask(df[c] == 0) | |
| vd_path = var_dict or DEFAULT_VAR_DICT | |
| vd = json.loads(vd_path.read_text(encoding="utf-8")) if vd_path.exists() else {} | |
| metric_cols = [c for c in df.columns if c not in META] | |
| model = [c for c in MODEL_COLS if c in df.columns] | |
| # Las *_rival no se serializan: se reconstruyen en el navegador con el self-join. | |
| otras = [c for c in metric_cols if c not in model and not c.endswith("_rival")] | |
| out_dir.mkdir(parents=True, exist_ok=True) | |
| for f in out_dir.glob("*.json"): | |
| f.unlink() | |
| ligas = [] | |
| for liga, g in df.groupby("competencia", sort=True): | |
| n_by_team = g.groupby(["temporada", "equipo"])["matchid"].transform("size") | |
| g = g[n_by_team >= MIN_PARTIDOS_EQUIPO] | |
| if g.empty or g["equipo"].nunique() < 4: | |
| continue | |
| g = g.sort_values(["temporada", "fecha", "matchid", "equipo"]).reset_index(drop=True) | |
| sl = slug(liga) | |
| dic, meta = {}, {} | |
| for c in ("equipo", "rival", "temporada", "formacion", "formacion_rival", "matchid"): | |
| dic[c], meta[c] = _idx(g[c]) | |
| meta["fecha"] = g["fecha"].astype(str).str.slice(0, 10).tolist() | |
| _dump(out_dir / f"{sl}.json", { | |
| "liga": liga, "n": int(len(g)), "dic": dic, "meta": meta, | |
| "cols": model, "values": [_col(g[c]) for c in model]}) | |
| kb_vars = 0 | |
| if with_vars: | |
| _dump(out_dir / f"{sl}.vars.json", | |
| {"liga": liga, "n": int(len(g)), "cols": otras, | |
| "values": [_col(g[c]) for c in otras]}) | |
| kb_vars = round((out_dir / f"{sl}.vars.json").stat().st_size / 1024) | |
| ligas.append({ | |
| "liga": liga, "slug": sl, | |
| "temporadas": sorted(g["temporada"].unique().tolist()), | |
| "equipos": sorted(g["equipo"].unique().tolist()), | |
| "pj_por_temporada": {se: int(sg.groupby("equipo")["matchid"].nunique().median()) | |
| for se, sg in g.groupby("temporada")}, | |
| "n_filas": int(len(g)), "n_partidos": int(g["matchid"].nunique()), | |
| "kb": round((out_dir / f"{sl}.json").stat().st_size / 1024), "kb_vars": kb_vars, | |
| }) | |
| cr_path = crests or DEFAULT_CRESTS | |
| if cr_path.exists(): | |
| raw = json.loads(cr_path.read_text(encoding="utf-8")) | |
| equipos = {e for l in ligas for e in l["equipos"]} | |
| cr = {} | |
| for k, v in raw.items(): | |
| cr.setdefault(canonical_team(k), v) | |
| _dump(out_dir / "crests.json", {k: v for k, v in cr.items() if k in equipos}) | |
| index = { | |
| "generado": pd.Timestamp.today().strftime("%Y-%m-%d"), | |
| "ligas": ligas, "model_cols": model, "otras_cols": otras if with_vars else [], | |
| "con_vars": bool(with_vars), "rival_derivadas": True, | |
| "variables": {c: {"label": vd.get(c, {}).get("label", c), | |
| "grupos": vd.get(c, {}).get("grupos", []), | |
| "sub": vd.get(c, {}).get("subgrupo", ""), | |
| "neg": bool(vd.get(c, {}).get("negativa", False))} | |
| for c in metric_cols}, | |
| "total_filas": int(sum(l["n_filas"] for l in ligas)), | |
| "total_partidos": int(df["matchid"].nunique()), | |
| } | |
| _dump(out_dir / "index.json", index) | |
| mb = sum(f.stat().st_size for f in out_dir.glob("*.json")) / 1e6 | |
| print(f" ofdef: {len(ligas)} ligas · {index['total_filas']:,} filas equipo-partido · " | |
| f"{mb:.1f} MB{'' if with_vars else ' (solo modelo)'}", flush=True) | |
| return index | |