Spaces:
Running
Running
| """team_stats a nivel PARTIDO (una fila por equipo-partido), sin agregar. | |
| `build_team_stats_agg.py` promedia por (competencia, temporada, equipo) y con eso alcanza | |
| para la app de Equipos y variables. El modelo ofensivo/defensivo de la app de Estilos, en | |
| cambio, necesita el partido: los ejes defensivos se miden sobre **el rival de ese partido** | |
| (self-join por `matchid`), y la vista de partido a partido grafica cada encuentro. | |
| Baja el mismo CSV de Azure, se queda con las filas "90 mins", deduplica a la versión | |
| coherente (misma lógica que el agregado) y guarda todo sin promediar. | |
| Salida: vendor/data/match_team_stats.parquet | |
| Uso: python scripts/build_match_team_stats.py [--csv /ruta/local.csv] | |
| """ | |
| from __future__ import annotations | |
| import sys | |
| import tempfile | |
| from pathlib import Path | |
| sys.path.insert(0, str(Path(__file__).resolve().parents[1] / "src")) | |
| import pandas as pd | |
| sys.path.insert(0, str(Path(__file__).resolve().parent)) | |
| from build_team_stats_agg import CHUNK, META, REMOTE, _download, _pick_consistent_version # noqa: E402 | |
| OUT = Path(__file__).resolve().parents[1] / "vendor" / "data" / "match_team_stats.parquet" | |
| KEEP_META = ["competencia", "temporada", "equipo", "rival", "fecha", "matchid", | |
| "formacion", "formacion_rival"] | |
| def extract(csv_path: Path) -> pd.DataFrame: | |
| head = pd.read_csv(csv_path, nrows=2000) | |
| metric_cols = [c for c in head.columns if c not in META | |
| and pd.to_numeric(head[c], errors="coerce").notna().any()] | |
| have_meta = [c for c in KEEP_META if c in head.columns] | |
| parts = [] | |
| for i, chunk in enumerate(pd.read_csv( | |
| csv_path, usecols=have_meta + ["estado"] + metric_cols, low_memory=False, | |
| chunksize=CHUNK, dtype={"matchid": str, "temporada": str}), 1): | |
| parts.append(chunk[chunk["estado"] == "90 mins"]) | |
| if i % 20 == 0: | |
| print(f" chunk {i}…", flush=True) | |
| full = pd.concat(parts, ignore_index=True).drop(columns=["estado"]) | |
| for c in metric_cols: | |
| full[c] = pd.to_numeric(full[c], errors="coerce") | |
| full = _pick_consistent_version(full) | |
| # El parquet queda FIEL al proveedor, igual que team_stats_agg.parquet. El saneo del | |
| # 0 de tracking (= dato faltante) se aplica aguas abajo, en el build de cada app, para | |
| # que las tres compartan exactamente la misma limpieza. | |
| # Misma derivada que el agregado, pero por partido. | |
| if {"cross_completados", "cross_total"} <= set(full.columns): | |
| tot = full["cross_total"].where(full["cross_total"] > 0) | |
| full["pct_exito_centros"] = (full["cross_completados"] / tot).round(4) | |
| print(f" filas equipo-partido: {len(full):,} · partidos: {full['matchid'].nunique():,} · " | |
| f"ligas: {full['competencia'].nunique()} · métricas: {len(metric_cols)}", flush=True) | |
| return full | |
| def main() -> None: | |
| local = None | |
| for i, a in enumerate(sys.argv): | |
| if a == "--csv" and i + 1 < len(sys.argv): | |
| local = Path(sys.argv[i + 1]) | |
| OUT.parent.mkdir(parents=True, exist_ok=True) | |
| if local and local.exists(): | |
| print(f"[match_stats] usando CSV local {local}", flush=True) | |
| df = extract(local) | |
| else: | |
| with tempfile.TemporaryDirectory() as td: | |
| tmp = Path(td) / "team_stats.csv" | |
| _download(tmp) | |
| df = extract(tmp) | |
| df.to_parquet(OUT, index=False) | |
| print(f"[match_stats] OK → {OUT} ({OUT.stat().st_size/1e6:.1f} MB)", flush=True) | |
| if __name__ == "__main__": | |
| main() | |