"""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 from racing_reports.league_names import canonical 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: df = pd.read_parquet(PATH) df["liga"] = df["liga"].astype(str).map(canonical) # normaliza nombres de liga return df 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}