src/racing_reports/lowblock_data.py

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

Lee el artefacto bundleado ``vendor/data/lowblock/team_resources.parquet``: una fila por
(liga, temporada, equipo, lado of/def, recurso, bloque bajo/medio/alto) con **sumas crudas**.
El bloque se elige en el front (uno o varios); acá se **suman** las sumas de los bloques
elegidos y se recalculan las tasas, para que cualquier combinación de bloques sea consistente.
``n_matches`` (partidos jugados por equipo-temporada) da "intentos por 90'".
"""

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": "1 vs 1",
    "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",
    "intentos_90": "Intentos por 90'",
}
BLOQUE_LABELS = {"bajo": "Bloque bajo", "medio": "Bloque medio", "alto": "Bloque alto"}
BLOQUES = ("bajo", "medio", "alto")

# sumas crudas del artefacto que se agregan al combinar bloques
RAW = ["n_intentos", "n_exitos", "pv_sum", "pv_n", "pv_ok_sum", "pv_ok_n", "evento_clave",
       "n_seqs", "tiros", "goles", "xg_acum", "n_ok_seqs", "xg_ok"]
KEYS = ["liga", "temporada", "equipo", "lado", "recurso"]


@lru_cache(maxsize=1)
def _df() -> pd.DataFrame:
    return pd.read_parquet(PATH)


def _parse_bloques(bloques: str | None) -> list[str]:
    if not bloques:
        return list(BLOQUES)
    sel = [b.strip() for b in bloques.split(",") if b.strip() in BLOQUES]
    return sel or list(BLOQUES)


def _combine(df: pd.DataFrame) -> pd.DataFrame:
    """Suma las sumas crudas de los bloques ya filtrados y recalcula tasas por fila."""
    g = df.groupby(KEYS, as_index=False)
    out = g[RAW].sum()
    out = out.merge(g["n_matches"].max(), on=KEYS)

    def rate(num, den, factor=1.0, nd=2):
        return (out[num] / out[den].where(out[den] > 0) * factor).round(nd)

    out["pct_exito"] = rate("n_exitos", "n_intentos", 100, 1)
    out["pv_jugada"] = rate("pv_sum", "pv_n", 1000, 2)
    out["pv_jugada_exitosa"] = rate("pv_ok_sum", "pv_ok_n", 1000, 2)
    out["tiros_jugada"] = rate("tiros", "n_seqs", 1, 3)
    out["xg_jugada"] = rate("xg_acum", "n_seqs", 1, 4)
    out["xg_jugada_exitosa"] = rate("xg_ok", "n_ok_seqs", 1, 4)
    out["pct_goles"] = rate("goles", "n_seqs", 100, 2)
    out["intentos_90"] = rate("n_intentos", "n_matches", 1, 1)
    out["xg_acum"] = out["xg_acum"].round(2)
    # recursos donde algunas métricas no aplican (ver build): tiro_lejano REALIZA el peligro,
    # rompe-líneas solo existe completado.
    tiro = out["recurso"] == "tiro_lejano"
    out.loc[tiro, ["pv_jugada", "pv_jugada_exitosa", "evento_clave"]] = pd.NA
    out.loc[out["recurso"] == "rompe_lineas_ok", "pct_exito"] = pd.NA
    return out


DISPLAY = ["liga", "temporada", "equipo", "lado", "recurso", "n_intentos", "intentos_90",
           "n_matches", "pct_exito", "pv_jugada", "pv_jugada_exitosa", "evento_clave",
           "tiros_jugada", "xg_acum", "xg_jugada", "xg_jugada_exitosa", "goles", "pct_goles"]


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"},
        "bloques": BLOQUE_LABELS,
    }


def scatter_rows(league: str, bloques: str | None = None) -> list[dict]:
    df = _df()
    df = df[df["bloque"].isin(_parse_bloques(bloques))]
    if league and league != "__all__":
        df = df[df["liga"] == league]
    out = _combine(df)[DISPLAY]
    return out.where(pd.notna(out), None).to_dict("records")


# métricas de tasa (media de liga ponderada por intentos) vs de conteo (media simple)
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, "intentos_90": 1, "evento_clave": 1, "goles": 1}


def team_tables(league: str, equipo: str, season: str | None = None,
                bloques: str | None = None) -> dict:
    df = _df()
    df = df[df["bloque"].isin(_parse_bloques(bloques))]
    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 ''}")
    comb = _combine(liga_df)
    out = {}
    for lado in ("of", "def"):
        sub = comb[comb["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[DISPLAY].where(pd.notna(rows[DISPLAY]), 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}