scripts/build_team_stats_agg.py

"""Construye el agregado por equipo de team_stats para la sección "Equipos y variables".

Baja ``raw/eventing/team_stats/team_stats_all_leagues.csv`` (stats POR PARTIDO y por
equipo, ~870 MB) de Azure y lo agrega a un **perfil por (competencia, temporada,
equipo)**: promedio por partido de cada métrica + nº de partidos. Sale a
``vendor/data/team_stats_agg.parquet`` (chico), que la app carga para el scoring
z-score — así no hay que bajar el CSV gigante en el Space.

Uso: python scripts/build_team_stats_agg.py
"""

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

from racing_reports.datastore import DataStore

REMOTE = "raw/eventing/team_stats/team_stats_all_leagues.csv"
OUT = Path(__file__).resolve().parents[1] / "vendor" / "data" / "team_stats_agg.parquet"
# Columnas de identificación / no-métricas (no entran como variables).
META = ["equipo", "rival", "competencia", "fecha", "temporada", "matchid",
        "estado", "formacion", "formacion_rival"]
KEYS = ["competencia", "temporada", "equipo"]
CHUNK = 200_000


def _download(local: Path) -> None:
    fs = DataStore()._filesystem_client()
    fc = fs.get_file_client(REMOTE)
    print(f"Descargando {REMOTE} → {local} …", flush=True)
    with open(local, "wb") as fh:
        fc.download_file().readinto(fh)
    print(f"  {local.stat().st_size / 1e6:.0f} MB", flush=True)


def _pick_consistent_version(full: pd.DataFrame) -> pd.DataFrame:
    """Cada equipo-partido aparece con 2 versiones (conteos idénticos pero posesión/
    derivadas distintas; una está mal). Por (equipo, partido) nos quedamos con la
    versión cuya posesión es COHERENTE con el dominio de pases: el equipo que metió
    más pases que el rival debe tener >50% de posesión. Así descartamos la versión
    espuria (p.ej. 73% con menos pases que el rival)."""
    mid = KEYS + ["matchid"]
    match_keys = ["competencia", "temporada", "matchid"]
    # pases del equipo por partido (idénticos entre versiones) y dominio vs rival:
    # el total del partido suma los DOS equipos (agrupar por partido, sin 'equipo').
    tp = full.groupby(mid, dropna=False)["pases_totales"].first().reset_index()
    tot = tp.groupby(match_keys)["pases_totales"].transform("sum")
    tp["dominant"] = tp["pases_totales"] > (tot - tp["pases_totales"])
    full = full.merge(tp[mid + ["dominant"]], on=mid, how="left")
    # rank: dominante → mayor posesión; no dominante → menor posesión.
    pos = pd.to_numeric(full["pct_posesion"], errors="coerce")
    full["__poskey"] = pos.where(full["dominant"], -pos).fillna(-1e9)
    keep = full.groupby(mid, dropna=False)["__poskey"].idxmax()
    return full.loc[keep].drop(columns=["dominant", "__poskey"]).reset_index(drop=True)


def _aggregate(csv_path: Path) -> pd.DataFrame:
    # Detectar columnas métricas (numéricas) desde el header.
    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()]
    # Quedarnos solo con las filas "90 mins" (total del partido) — son pocas (~200k),
    # así que las juntamos para deduplicar por matchid de forma global y agregar.
    parts = []
    # matchid y temporada SIEMPRE como str: hay matchids numéricos y alfanuméricos,
    # y leer por chunks infiere el tipo por chunk → el mismo matchid numérico queda
    # int en un chunk y str en otro y NO agrupa → versiones duplicadas sin colapsar.
    for chunk in pd.read_csv(csv_path, usecols=KEYS + ["estado", "matchid"] + metric_cols,
                             low_memory=False, chunksize=CHUNK,
                             dtype={"matchid": str, "temporada": str}):
        parts.append(chunk[chunk["estado"] == "90 mins"])
    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)
    full = full.drop(columns=["matchid"])
    g = full.groupby(KEYS, dropna=False)
    mean = g[metric_cols].mean()
    out = mean.reset_index()
    out["n_partidos"] = g.size().values
    print(f"Partidos (versión consistente): {len(full):,} | equipos-temporada: {len(out):,} | "
          f"métricas: {len(metric_cols)}", flush=True)
    return out


def main() -> None:
    OUT.parent.mkdir(parents=True, exist_ok=True)
    with tempfile.TemporaryDirectory() as td:
        csv_path = Path(td) / "team_stats_all_leagues.csv"
        _download(csv_path)
        agg = _aggregate(csv_path)
    agg.to_parquet(OUT, index=False)
    print(f"Guardado: {OUT}  ({OUT.stat().st_size / 1e6:.1f} MB)")


if __name__ == "__main__":
    main()