Spaces:
Sleeping
Sleeping
| """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"] | |
| 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} | |