"""
compare_block_methods.py
------------------------
Compara dos métodos de clasificación de bloque defensivo:
SIMPLE: phaseLabel dominante por posesión, filtro ≥3 eventos de bloque totales.
Sin reglas extra. Global = toda la liga.
FILTRADO: phaseLabel con desempate ≥3 eventos, filtro ≥3 eventos clave (pase/tiro),
multi-bloque con pesos para posesiones largas (≥7 pases),
reclasificación High→Med, y regla heurística para posesiones sin etiqueta.
Solo partidos de Racing (requiere raw events).
"""
import ast
import warnings
from pathlib import Path
import numpy as np
import pandas as pd
import plotly.graph_objects as go
from plotly.subplots import make_subplots
warnings.filterwarnings("ignore")
PREPROCESSED = Path("/Users/pagrois/Documents/Racing/preprocessed_SSD_25-26.csv")
RAW_DIR = Path("/Users/pagrois/Documents/Racing/raw_events")
REPORTS_DIR = Path("/Users/pagrois/Racing/reports")
REPORTS_DIR.mkdir(parents=True, exist_ok=True)
DARK_BG = "#1A1F2E"
CARD_BG = "#232A3B"
TEXT_COLOR = "#E8EDF5"
GRID_COLOR = "#2E3650"
COLORS = {"High": "#E06C5F", "Medium": "#F0A500", "Low": "#00A86B"}
BLOCK_LBL = {"High": "Bloque Alto", "Medium": "Bloque Medio", "Low": "Bloque Bajo"}
BLOCKS = ["High", "Medium", "Low"]
PHASE_TO_BLOCK = {
"Build Up against High Block": "High",
"Build Up against Medium Block": "Medium",
"Build Up against Low Block": "Low",
}
PERIOD_COL = "period_id"
def base_layout(**extra):
d = dict(paper_bgcolor=CARD_BG, plot_bgcolor=DARK_BG,
font=dict(color=TEXT_COLOR, size=11),
margin=dict(l=60, r=20, t=55, b=50),
legend=dict(bgcolor=CARD_BG, bordercolor=GRID_COLOR, borderwidth=1))
d.update(extra); return d
def axis_style(**kw):
d = dict(gridcolor=GRID_COLOR, zerolinecolor=GRID_COLOR, linecolor=GRID_COLOR)
d.update(kw); return d
def fig2html(fig):
return fig.to_html(full_html=False, include_plotlyjs=False)
# ---------------------------------------------------------------------------
# 1. Cargar preprocessed — COMPLETO (toda la liga) para SIMPLE global
# ---------------------------------------------------------------------------
print("Cargando preprocessed (toda la liga)...", flush=True)
load_cols = ["matchId", "possessionId", "phaseLabel", "TeamName", "TeamRival",
"fecha", PERIOD_COL]
df_pre = pd.read_csv(PREPROCESSED, usecols=load_cols, low_memory=False)
df_pre["fecha_str"] = df_pre["fecha"].str[:10]
df_pre["block_raw"] = df_pre["phaseLabel"].map(PHASE_TO_BLOCK)
df_pre["matchId"] = df_pre["matchId"].astype(str)
df_pre["possessionId"] = df_pre["possessionId"].astype(float).astype(str)
df_pre[PERIOD_COL] = pd.to_numeric(df_pre[PERIOD_COL], errors="coerce")
df_pre = df_pre[df_pre[PERIOD_COL].isin([1, 2])]
print(f" {df_pre['matchId'].nunique()} partidos totales en preprocessed")
# Subconjunto Racing (para FILTRADO y gráficos por partido)
racing_mask = (df_pre["TeamName"].str.contains("Racing de Santander", na=False) |
df_pre["TeamRival"].str.contains("Racing de Santander", na=False))
df_racing = df_pre[racing_mask].copy()
has_label_mids = set(df_racing.dropna(subset=["block_raw"])["matchId"])
match_dates = (df_racing[df_racing["matchId"].isin(has_label_mids)]
.groupby("matchId")["fecha_str"].first().sort_values())
all_racing_ids = match_dates.index.tolist()
match_info = {}
for mid in all_racing_ids:
r = df_racing[df_racing["matchId"] == mid].iloc[0]
home = r["TeamName"]; away = r["TeamRival"]
rival = away if "Racing" in home else home
match_info[mid] = {
"home": home, "away": away, "fecha": r["fecha_str"],
"rival": rival.replace("Racing de Santander", "Racing"),
"label": r["fecha_str"] + " " + rival.replace("Racing de Santander","Racing")[:16],
}
print(f" {len(all_racing_ids)} partidos Racing con etiquetas phaseLabel")
df_ar = df_racing[df_racing["matchId"].isin(all_racing_ids)].copy()
all_poss = (df_ar.dropna(subset=["possessionId"])
.groupby(["matchId","possessionId"])[["TeamName", PERIOD_COL]]
.first().reset_index())
# ---------------------------------------------------------------------------
# 2. Raw events Racing — para FILTRADO y filtro key_ok
# ---------------------------------------------------------------------------
print("\nCargando raw events Racing...", flush=True)
SHOT_EVENTS = {"SavedShot","MissedShots","Goal","ShotOnPost","KeeperSweeper"}
def n_opts(v):
if pd.isna(v): return 0
try:
d = ast.literal_eval(str(v))
return len(d.get("player",[])) if isinstance(d, dict) else 0
except: return 0
# Lookup por nombre completo del archivo (maneja múltiples partidos por fecha)
xlsx_all_files = list(RAW_DIR.glob("*.xlsx"))
def find_xlsx(fecha, home, away):
"""Busca el xlsx del partido por fecha y equipos."""
h_sub = home.lower()[:10]
a_sub = away.lower()[:10]
for f in xlsx_all_files:
if not f.stem.startswith(fecha): continue
rest = f.stem[13:].lower() # "HomeTeam vs AwayTeam"
if h_sub in rest or a_sub in rest:
return f
return None
raw_stats = {}
for i, mid in enumerate(all_racing_ids, 1):
fpath = find_xlsx(match_info[mid]["fecha"], match_info[mid]["home"], match_info[mid]["away"])
if not fpath: continue
if i % 8 == 0 or i == len(all_racing_ids):
print(f" {i}/{len(all_racing_ids)} {fpath.name}", flush=True)
df_ev = pd.read_excel(fpath, sheet_name="Eventos", engine="openpyxl")
df_ev = df_ev[df_ev["possessionId"].notna()].copy()
df_ev["matchId"] = str(mid)
df_ev["possessionId"] = df_ev["possessionId"].astype(float).astype(str)
df_ev["x_val"] = pd.to_numeric(df_ev.get("x", pd.Series(dtype=float)), errors="coerce")
df_ev["is_key"] = df_ev["event_name"].isin({"Pass"} | SHOT_EVENTS)
df_ev["is_pass"] = df_ev["event_name"] == "Pass"
df_ev["has_press"] = df_ev["pressure"].notna() if "pressure" in df_ev.columns else False
df_ev["n_opt"] = df_ev["passOption"].apply(n_opts) if "passOption" in df_ev.columns else 0
for (m, p), grp in df_ev.groupby(["matchId","possessionId"]):
press_grp = grp[grp["has_press"] & grp["x_val"].notna()]
n_press_own = int((press_grp["x_val"] < 50).sum())
pass_rival = grp[grp["is_pass"] & grp["x_val"].notna() & (grp["x_val"] > 50)]
pass_7opt = int((pass_rival["n_opt"] >= 7).sum())
pass_own = grp[grp["is_pass"] & grp["x_val"].notna() & (grp["x_val"] < 50)]
max_opts_own = int(pass_own["n_opt"].max()) if len(pass_own) > 0 else 0
raw_stats[(m, p)] = {
"n_passes": int(grp["is_pass"].sum()),
"pass_7opt_rival": pass_7opt,
"max_opts_own": max_opts_own,
"n_press_own": n_press_own,
"n_key_events": int(grp["is_key"].sum()),
}
key_ok = {(m, p) for (m, p), s in raw_stats.items() if s.get("n_key_events", 0) >= 3}
# ---------------------------------------------------------------------------
# 3. MÉTODO SIMPLE — toda la liga, filtro ≥3 eventos de bloque totales
# ---------------------------------------------------------------------------
print("\nMétodo SIMPLE (toda la liga)...", flush=True)
block_ev_global = (df_pre.dropna(subset=["block_raw","possessionId"])
.groupby(["matchId","possessionId","block_raw"])
.size().reset_index(name="n_ev"))
simple_global_rows = []
for (mid, pid), grp in block_ev_global.groupby(["matchId","possessionId"]):
if grp["n_ev"].sum() < 3:
continue
dominant = grp.loc[grp["n_ev"].idxmax(), "block_raw"]
simple_global_rows.append({"matchId": mid, "possessionId": pid,
"block": dominant, "weight": 1.0})
df_simple_global = pd.DataFrame(simple_global_rows) if simple_global_rows else pd.DataFrame(
columns=["matchId","possessionId","block","weight"])
print(f" {len(df_simple_global):,} posesiones clasificadas (liga completa)")
# SIMPLE Racing (mismo filtro, para gráficos por partido)
block_ev_racing = (df_ar.dropna(subset=["block_raw","possessionId"])
.groupby(["matchId","possessionId","block_raw"])
.size().reset_index(name="n_ev"))
simple_rows = []
for (mid, pid), grp in block_ev_racing.groupby(["matchId","possessionId"]):
if (mid, pid) not in key_ok:
continue
dominant = grp.loc[grp["n_ev"].idxmax(), "block_raw"]
simple_rows.append({"matchId": mid, "possessionId": pid, "block": dominant, "weight": 1.0})
df_simple = pd.DataFrame(simple_rows) if simple_rows else pd.DataFrame(
columns=["matchId","possessionId","block","weight"])
df_simple = df_simple.merge(all_poss, on=["matchId","possessionId"], how="left")
df_simple["is_racing_att"] = df_simple["TeamName"].str.contains("Racing de Santander", na=False)
print(f" {len(df_simple):,} posesiones Racing clasificadas")
# ---------------------------------------------------------------------------
# 4. MÉTODO FILTRADO — Racing, con pesos para posesiones largas (≥7 pases)
# ---------------------------------------------------------------------------
print("\nMétodo FILTRADO (Racing)...", flush=True)
comp_ev = block_ev_racing.rename(columns={"block_raw":"block_type"})
def resolve_blocks(grp):
max_n = grp["n_ev"].max()
tied = set(grp[grp["n_ev"] == max_n]["block_type"])
if len(tied) == 1:
return list(tied)
valid = set(grp[(grp["n_ev"] >= 3) & grp["block_type"].isin(tied)]["block_type"])
blocks = list(valid) if valid else list(tied)
if len(blocks) > 1 and "Medium" in blocks:
non_med = [b for b in blocks if b != "Medium"]
if non_med: blocks = non_med
return blocks
filt_rows = []
for (mid, pid), grp in comp_ev.groupby(["matchId","possessionId"]):
n_passes = raw_stats.get((mid, pid), {}).get("n_passes", 0)
total_block_ev = grp["n_ev"].sum()
if n_passes >= 7 and len(grp) > 1:
# Posesión larga: permite múltiples bloques con peso proporcional
for _, row in grp.iterrows():
filt_rows.append({
"matchId": mid, "possessionId": pid,
"block": row["block_type"],
"weight": row["n_ev"] / total_block_ev,
})
else:
blocks = resolve_blocks(grp)
w = 1.0 / len(blocks)
for b in blocks:
filt_rows.append({"matchId": mid, "possessionId": pid, "block": b, "weight": w})
df_fx_labeled = pd.DataFrame(filt_rows) if filt_rows else pd.DataFrame(
columns=["matchId","possessionId","block","weight"])
labeled_poss = set(zip(df_fx_labeled["matchId"], df_fx_labeled["possessionId"]))
# Regla heurística para posesiones sin etiqueta
rule_rows = []
for _, row in all_poss.iterrows():
mid, pid = row["matchId"], row["possessionId"]
if (mid, pid) in labeled_poss: continue
s = raw_stats.get((mid, pid), {})
if s.get("n_passes", 0) <= 3: continue
# High: presiona arriba O defensor empujó ≥5 jugadores al campo de Racing
is_high = s["n_press_own"] > 3 or s.get("max_opts_own", 0) >= 5
b = "Low" if s["pass_7opt_rival"] >= 2 else ("High" if is_high else "Medium")
rule_rows.append({"matchId": mid, "possessionId": pid, "block": b, "weight": 1.0})
df_fx_all = pd.concat([df_fx_labeled,
pd.DataFrame(rule_rows) if rule_rows else pd.DataFrame(
columns=["matchId","possessionId","block","weight"])],
ignore_index=True)
df_fx_all = df_fx_all.merge(all_poss, on=["matchId","possessionId"], how="left")
df_fx_all["is_racing_att"] = df_fx_all["TeamName"].str.contains("Racing de Santander", na=False)
# Filtro ≥3 eventos clave
df_fx_all = df_fx_all[df_fx_all.apply(
lambda r: (r["matchId"], r["possessionId"]) in key_ok, axis=1)].copy()
# Reclasificación High→Med: 0 presiones propias Y defensor sin ≥5 jugadores en campo Racing
df_fx_all["n_press_own"] = df_fx_all.apply(
lambda r: raw_stats.get((r["matchId"],r["possessionId"]),{}).get("n_press_own",0), axis=1)
df_fx_all["max_opts_own"] = df_fx_all.apply(
lambda r: raw_stats.get((r["matchId"],r["possessionId"]),{}).get("max_opts_own",0), axis=1)
matches_with_press = {m for (m,p),s in raw_stats.items() if s.get("n_press_own",0) > 0}
mask_am = (df_fx_all["block"] == "High") & \
(df_fx_all["n_press_own"] < 1) & \
(df_fx_all["max_opts_own"] < 5) & \
(df_fx_all["matchId"].isin(matches_with_press))
df_fx_all.loc[mask_am, "block"] = "Medium"
reclassif_am = int(mask_am.sum())
n_poss_fx = df_fx_all[["matchId","possessionId"]].drop_duplicates().shape[0]
print(f" Reclasificados High→Med: {reclassif_am}")
print(f" {n_poss_fx:,} posesiones únicas Racing clasificadas (con multi-bloque: {len(df_fx_all):,} filas)")
# ---------------------------------------------------------------------------
# 5. Funciones de distribución con pesos
# ---------------------------------------------------------------------------
def weighted_dist(df):
"""Distribución porcentual usando weights. Retorna dict {block: pct} y n únicas posesiones."""
total_w = df["weight"].sum()
if total_w == 0:
return {b: 0.0 for b in BLOCKS}, 0
dist = {b: 100 * df[df["block"] == b]["weight"].sum() / total_w for b in BLOCKS}
n = df[["matchId","possessionId"]].drop_duplicates().shape[0]
return dist, n
def global_dist(df, role_val):
sub = df[df["is_racing_att"] == role_val]
return weighted_dist(sub)
def global_dist_all(df):
"""Distribución sin filtro por rol (para toda la liga)."""
return weighted_dist(df)
def dist_by_match(df, role_val):
sub = df[df["is_racing_att"] == role_val].copy()
rows = []
for mid in all_racing_ids:
m_sub = sub[sub["matchId"] == mid]
total_w = max(m_sub["weight"].sum(), 1e-9)
rows.append({
"matchId": mid,
"label": match_info[mid]["label"],
"total": m_sub[["matchId","possessionId"]].drop_duplicates().shape[0],
**{b: 100 * m_sub[m_sub["block"] == b]["weight"].sum() / total_w for b in BLOCKS}
})
return pd.DataFrame(rows)
# Distribuciones globales
simple_global_dist, simple_global_n = global_dist_all(df_simple_global)
simple_att_dist, simple_att_n = global_dist(df_simple, True)
simple_def_dist, simple_def_n = global_dist(df_simple, False)
fx_att_dist, fx_att_n = global_dist(df_fx_all, True)
fx_def_dist, fx_def_n = global_dist(df_fx_all, False)
# % que difiere (por possessionId único, tomando el bloque mayoritario)
def dominant_block(grp):
return grp.loc[grp["weight"].idxmax(), "block"]
simple_dom = (df_simple.groupby(["matchId","possessionId"])
.apply(dominant_block).reset_index(name="block_simple"))
fx_dom = (df_fx_all.groupby(["matchId","possessionId"])
.apply(dominant_block).reset_index(name="block_fx"))
merged = simple_dom.merge(fx_dom, on=["matchId","possessionId"], how="inner")
differ = (merged["block_simple"] != merged["block_fx"]).sum()
total_shared = len(merged)
print(f"\n Posesiones en común: {total_shared:,} | Difieren: {differ:,} ({100*differ/max(total_shared,1):.1f}%)")
# ---------------------------------------------------------------------------
# 6. Gráficos
# ---------------------------------------------------------------------------
print("\nGenerando gráficos...", flush=True)
# ── Fig 1: Distribución global — donuts ─────────────────────────────────────
# Fila 1: toda la liga (SIMPLE) + Racing Racing ataca/defiende (SIMPLE y FILTRADO)
fig1 = make_subplots(
rows=1, cols=5, specs=[[{"type":"domain"}]*5],
subplot_titles=[
"SIMPLE — Toda la liga",
"SIMPLE — Racing ataca", "SIMPLE — Racing defiende",
"FILTRADO — Racing ataca", "FILTRADO — Racing defiende",
]
)
for col, (dist, n) in enumerate([
(simple_global_dist, simple_global_n),
(simple_att_dist, simple_att_n),
(simple_def_dist, simple_def_n),
(fx_att_dist, fx_att_n),
(fx_def_dist, fx_def_n),
], 1):
fig1.add_trace(go.Pie(
labels=[BLOCK_LBL[b] for b in BLOCKS],
values=[round(dist[b], 1) for b in BLOCKS],
marker_colors=[COLORS[b] for b in BLOCKS],
hole=0.55,
textinfo="label+percent",
textfont=dict(size=9, color=TEXT_COLOR),
hovertemplate="%{label}: %{value:.1f}%
Primer donut: SIMPLE aplicado a toda la liga (filtro ≥3 eventos de bloque por phaseLabel, sin raw events).
Resto: comparación SIMPLE vs FILTRADO en partidos de Racing.
FILTRADO permite múltiples bloques ponderados en posesiones largas.
Calculado sobre posesiones en común usando el bloque de mayor peso.
Fila = clasificación Simple · Columna = clasificación Filtrado · valores = n posesiones