src/racing_reports/lowblock_data.py

"""Datos de la sección "Recursos bloque bajo" (scatter + tablas por equipo).

Lee el artefacto bundleado ``vendor/data/lowblock/team_resources.parquet``:
una fila por (liga, equipo, lado of/def, recurso) con las métricas de intento
(n, % éxito, pv por jugada) y de resultado por secuencia (tiros/jugada, xG, goles).
"""

from __future__ import annotations

from functools import lru_cache

import pandas as pd

from racing_reports import vendor_env

PATH = vendor_env.DATA_DIR / "lowblock" / "team_resources.parquet"

RECURSO_LABELS = {
    "todos": "Todos los recursos (construcción)",
    "pase_entre_lineas": "Pase al espacio (throughball)",
    "cutback": "Cutback",
    "centro_colgado": "Centro colgado",
    "centro_inswinger": "Centro inswinger",
    "centro_outswinger": "Centro outswinger",
    "centro_raso": "Centro raso",
    "rompe_lineas_ok": "Rompe-líneas (completado)",
    "gambeta": "Gambeta (1v1)",
    "tiro_lejano": "Tiro desde fuera del área",
}
METRICA_LABELS = {
    "pv_jugada": "Peligro por jugada (pv ‰)",
    "pv_jugada_exitosa": "Peligro por jugada exitosa (pv ‰)",
    "evento_clave": "Veces evento clave del ataque",
    "tiros_jugada": "Tiros por jugada",
    "xg_acum": "xG acumulado",
    "xg_jugada": "xG por jugada",
    "xg_jugada_exitosa": "xG por jugada exitosa",
    "goles": "Goles",
    "pct_goles": "% de goles",
    "pct_exito": "% de éxito",
    "n_intentos": "Cantidad de intentos",
}


@lru_cache(maxsize=1)
def _df() -> pd.DataFrame:
    df = pd.read_parquet(PATH)
    # pv de un tiro errado es contablemente negativo (muere la posesión) → engañoso como
    # métrica del recurso; y rompe-líneas solo existe completado (Opta no marca intentos).
    df.loc[df["recurso"] == "tiro_lejano", "pv_jugada"] = pd.NA
    df.loc[df["recurso"] == "tiro_lejano", "pv_jugada_exitosa"] = pd.NA
    df.loc[df["recurso"] == "tiro_lejano", "evento_clave"] = pd.NA
    df.loc[df["recurso"] == "rompe_lineas_ok", "pct_exito"] = pd.NA
    return df


def options() -> dict:
    df = _df()
    seasons = {lg: sorted(g["temporada"].astype(str).unique(), reverse=True)
               for lg, g in df.groupby("liga")}
    return {
        "leagues": sorted(df["liga"].unique().tolist()),
        "seasons": seasons,
        "recursos": RECURSO_LABELS,
        "metricas": METRICA_LABELS,
        "lados": {"of": "Ofensivo", "def": "Defensivo"},
    }


def scatter_rows(league: str) -> list[dict]:
    df = _df()
    if league and league != "__all__":
        df = df[df["liga"] == league]
    cols = ["liga", "temporada", "equipo", "lado", "recurso", "n_intentos", "pct_exito", "pv_jugada",
            "pv_jugada_exitosa", "evento_clave",
            "tiros_jugada", "xg_acum", "xg_jugada", "xg_jugada_exitosa", "goles", "pct_goles"]
    out = df[cols].where(pd.notna(df[cols]), None)
    return out.to_dict("records")


DIF_METRICS = {"pct_exito": 1, "pv_jugada": 2, "pv_jugada_exitosa": 2,
               "tiros_jugada": 3, "xg_jugada": 4, "pct_goles": 2}
COUNT_METRICS = {"n_intentos": 1, "evento_clave": 1, "goles": 1}  # media simple entre equipos


def team_tables(league: str, equipo: str, season: str | None = None) -> dict:
    df = _df()
    liga_df = df[df["liga"] == league]
    if season:
        liga_df = liga_df[liga_df["temporada"].astype(str) == str(season)]
    if liga_df.empty:
        raise ValueError(f"Liga/temporada sin datos: {league} {season or ''}")
    out = {}
    for lado in ("of", "def"):
        sub = liga_df[liga_df["lado"] == lado]
        # media de la liga ponderada por intentos, por recurso y métrica
        medias: dict[str, dict] = {}
        for rec, g in sub.groupby("recurso"):
            w = g["n_intentos"].clip(lower=1)
            medias[rec] = {}
            for m, nd in DIF_METRICS.items():
                v = (g[m] * w).sum() / w.sum()
                medias[rec][m] = round(float(v), nd) if pd.notna(v) else None
            for m, nd in COUNT_METRICS.items():
                v = g[m].mean()
                medias[rec][m] = round(float(v), nd) if pd.notna(v) else None
        rows = sub[sub["equipo"] == equipo].sort_values("pv_jugada", ascending=False)
        recs = rows.where(pd.notna(rows), None).to_dict("records")
        for r in recs:
            mu = medias.get(r["recurso"], {})
            r["dif"] = {}
            for m, nd in {**DIF_METRICS, **COUNT_METRICS}.items():
                if r.get(m) is not None and mu.get(m) is not None:
                    r["dif"][m] = round(float(r[m]) - mu[m], nd)
        out[lado] = {"rows": recs, "liga_media": medias}
    return {"league": league, "season": season, "equipo": equipo,
            "recursos": RECURSO_LABELS, **out}