""" compare_block_methods_seq.py ----------------------------- Igual que compare_block_methods.py pero usando sequenceId (Opta) como unidad base en lugar de possessionId. Una sequence no cruza interrupciones (faltas, saques, etc.), por lo que es una unidad más limpia para clasificar el bloque defensivo. Todo se computa desde el preprocessed — no se necesitan los xlsx de 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") 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", } SHOT_EVENTS = {"SavedShot","MissedShots","Goal","ShotOnPost","KeeperSweeper"} 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) 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 # --------------------------------------------------------------------------- # 1. Cargar preprocessed # --------------------------------------------------------------------------- print("Cargando preprocessed...", flush=True) load_cols = ["matchId", "sequenceId", "phaseLabel", "TeamName", "TeamRival", "fecha", PERIOD_COL, "minute", "second", "x", "pressure", "passOption", "event_name"] df_pre = pd.read_csv(PREPROCESSED, usecols=load_cols, low_memory=False) df_pre["matchId"] = df_pre["matchId"].astype(str) df_pre["sequenceId"] = df_pre["sequenceId"].astype(float).astype(str) df_pre["block_raw"] = df_pre["phaseLabel"].map(PHASE_TO_BLOCK) df_pre[PERIOD_COL] = pd.to_numeric(df_pre[PERIOD_COL], errors="coerce") df_pre = df_pre[df_pre[PERIOD_COL].isin([1, 2])] df_pre["x_val"] = pd.to_numeric(df_pre["x"], errors="coerce") df_pre["is_pass"] = df_pre["event_name"] == "Pass" df_pre["is_key"] = df_pre["event_name"].isin({"Pass"} | SHOT_EVENTS) df_pre["has_press"] = df_pre["pressure"].notna() df_pre["n_opt"] = df_pre["passOption"].apply(n_opts) df_pre["fecha_str"] = df_pre["fecha"].str[:10] print(f" {df_pre['matchId'].nunique()} partidos, " f"{df_pre.groupby(['matchId','sequenceId']).ngroups:,} sequences totales") # Subconjunto Racing 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"] is_racing_home = "Racing" in home rival = away if is_racing_home else home match_info[mid] = { "fecha": r["fecha_str"], "home": home, "away": away, "racing_local": is_racing_home, "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() # Equipo y minuto de inicio de cada sequence seq_info = (df_ar.dropna(subset=["sequenceId"]) .sort_values(["matchId","sequenceId","minute","second"]) .groupby(["matchId","sequenceId"])[["TeamName", PERIOD_COL,"minute","second"]] .first().reset_index()) seq_info["is_racing_att"] = seq_info["TeamName"].str.contains("Racing de Santander", na=False) # --------------------------------------------------------------------------- # 2. Raw stats por sequence — desde preprocessed # --------------------------------------------------------------------------- print("\nComputando stats por sequence...", flush=True) raw_stats = {} for (mid, sid), grp in df_ar.dropna(subset=["sequenceId"]).groupby(["matchId","sequenceId"]): 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[(mid, sid)] = { "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, s) for (m, s), v in raw_stats.items() if v.get("n_key_events", 0) >= 3} print(f" {len(raw_stats):,} sequences Racing con stats | {len(key_ok):,} con ≥3 eventos clave") # --------------------------------------------------------------------------- # 3. Eventos de bloque por sequence # --------------------------------------------------------------------------- block_ev_racing = (df_ar.dropna(subset=["block_raw","sequenceId"]) .groupby(["matchId","sequenceId","block_raw"]) .size().reset_index(name="n_ev")) # SIMPLE global (toda la liga, ≥3 eventos de bloque totales) block_ev_global = (df_pre.dropna(subset=["block_raw","sequenceId"]) .groupby(["matchId","sequenceId","block_raw"]) .size().reset_index(name="n_ev")) # --------------------------------------------------------------------------- # 4. MÉTODO SIMPLE — toda la liga # --------------------------------------------------------------------------- print("\nMétodo SIMPLE (toda la liga)...", flush=True) simple_global_rows = [] for (mid, sid), grp in block_ev_global.groupby(["matchId","sequenceId"]): if grp["n_ev"].sum() < 3: continue dominant = grp.loc[grp["n_ev"].idxmax(), "block_raw"] simple_global_rows.append({"matchId": mid, "sequenceId": sid, "block": dominant, "weight": 1.0}) df_simple_global = pd.DataFrame(simple_global_rows) if simple_global_rows else pd.DataFrame( columns=["matchId","sequenceId","block","weight"]) print(f" {len(df_simple_global):,} sequences (liga completa)") # SIMPLE Racing simple_rows = [] for (mid, sid), grp in block_ev_racing.groupby(["matchId","sequenceId"]): if (mid, sid) not in key_ok: continue dominant = grp.loc[grp["n_ev"].idxmax(), "block_raw"] simple_rows.append({"matchId": mid, "sequenceId": sid, "block": dominant, "weight": 1.0}) df_simple = pd.DataFrame(simple_rows) if simple_rows else pd.DataFrame( columns=["matchId","sequenceId","block","weight"]) df_simple = df_simple.merge(seq_info, on=["matchId","sequenceId"], how="left") print(f" {len(df_simple):,} sequences Racing") # --------------------------------------------------------------------------- # 5. MÉTODO FILTRADO — Racing, con multi-bloque para sequences largas (≥7 pases) # --------------------------------------------------------------------------- print("\nMétodo FILTRADO (Racing)...", flush=True) def resolve_blocks(grp): max_n = grp["n_ev"].max() tied = set(grp[grp["n_ev"] == max_n]["block_raw"]) if len(tied) == 1: return list(tied) valid = set(grp[(grp["n_ev"] >= 3) & grp["block_raw"].isin(tied)]["block_raw"]) blocks = list(valid) if valid else list(tied) if len(blocks) > 1 and "Medium" in blocks: nm = [b for b in blocks if b != "Medium"] if nm: blocks = nm return blocks filt_rows = [] labeled_seqs = set() for (mid, sid), grp in block_ev_racing.groupby(["matchId","sequenceId"]): if (mid, sid) not in key_ok: continue labeled_seqs.add((mid, sid)) n_passes = raw_stats.get((mid, sid), {}).get("n_passes", 0) total_block_ev = grp["n_ev"].sum() # Multi-bloque: ≥7 pases, y cada bloque debe tener ≥3 eventos grp_multi = grp[grp["n_ev"] >= 3] if n_passes >= 7 and len(grp_multi) > 1: total_multi_ev = grp_multi["n_ev"].sum() for _, row in grp_multi.iterrows(): filt_rows.append({"matchId": mid, "sequenceId": sid, "block": row["block_raw"], "weight": row["n_ev"] / total_multi_ev}) else: blocks = resolve_blocks(grp) w = 1.0 / len(blocks) for b in blocks: filt_rows.append({"matchId": mid, "sequenceId": sid, "block": b, "weight": w}) # Heurística para sequences sin etiqueta phaseLabel for (mid, sid), s in raw_stats.items(): if (mid, sid) in labeled_seqs: continue if s.get("n_passes", 0) <= 3: continue if (mid, sid) not in key_ok: continue 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") filt_rows.append({"matchId": mid, "sequenceId": sid, "block": b, "weight": 1.0}) df_fx_all = pd.DataFrame(filt_rows) if filt_rows else pd.DataFrame( columns=["matchId","sequenceId","block","weight"]) df_fx_all = df_fx_all.merge(seq_info, on=["matchId","sequenceId"], how="left") # Filtro ≥3 eventos clave df_fx_all = df_fx_all[df_fx_all.apply( lambda r: (r["matchId"], r["sequenceId"]) in key_ok, axis=1)].copy() # Reclasificación High→Med df_fx_all["n_press_own"] = df_fx_all.apply( lambda r: raw_stats.get((r["matchId"],r["sequenceId"]),{}).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["sequenceId"]),{}).get("max_opts_own",0), axis=1) matches_with_press = {m for (m,s), v in raw_stats.items() if v.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_seq_fx = df_fx_all[["matchId","sequenceId"]].drop_duplicates().shape[0] print(f" Reclasificados High→Med: {reclassif_am}") print(f" {n_seq_fx:,} sequences únicas Racing clasificadas ({len(df_fx_all):,} filas)") # --------------------------------------------------------------------------- # 6. Distribuciones # --------------------------------------------------------------------------- def weighted_dist(df): 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","sequenceId"]].drop_duplicates().shape[0] return dist, n def global_dist(df, role_val): return weighted_dist(df[df["is_racing_att"] == role_val]) def global_dist_all(df): 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","sequenceId"]].drop_duplicates().shape[0], **{b: 100 * m_sub[m_sub["block"]==b]["weight"].sum() / total_w for b in BLOCKS} }) return pd.DataFrame(rows) 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) def dominant_block(grp): return grp.loc[grp["weight"].idxmax(), "block"] simple_dom = (df_simple.groupby(["matchId","sequenceId"]) .apply(dominant_block).reset_index(name="block_simple")) fx_dom = (df_fx_all.groupby(["matchId","sequenceId"]) .apply(dominant_block).reset_index(name="block_fx")) merged = simple_dom.merge(fx_dom, on=["matchId","sequenceId"], how="inner") differ = (merged["block_simple"] != merged["block_fx"]).sum() total_shared = len(merged) print(f"\n Sequences en común: {total_shared:,} | Difieren: {differ:,} ({100*differ/max(total_shared,1):.1f}%)") # --------------------------------------------------------------------------- # 7. Gráficos # --------------------------------------------------------------------------- print("\nGenerando gráficos...", flush=True) # ── Fig 1: Donuts globales ─────────────────────────────────────────────────── 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}%", name=f"n={n:,}", ), row=1, col=col) col_sfx = "" if col == 1 else str(col) fig1.add_annotation( text=f"n={n:,}", showarrow=False, font=dict(size=8, color=TEXT_COLOR), xref=f"x{col_sfx} domain", yref=f"y{col_sfx} domain", x=0.5, y=0.5, ) fig1.update_layout(**base_layout( title=dict(text="Distribución global por SEQUENCE — SIMPLE (toda la liga) vs FILTRADO (Racing)", font=dict(color=TEXT_COLOR, size=13), x=0.01), height=380, )) # ── Fig 2 & 3: Por partido ─────────────────────────────────────────────────── def all_match_bar_fig(df_s, df_f, role_val, title): labels = []; vals_s = {b: [] for b in BLOCKS}; vals_f = {b: [] for b in BLOCKS} for mid in all_racing_ids: short = match_info[mid]["label"] labels.append(f"{short[:5]} {short[11:]}") for df_, vals in [(df_s, vals_s), (df_f, vals_f)]: m_sub = df_[(df_["matchId"] == mid) & (df_["is_racing_att"] == role_val)] total_w = max(m_sub["weight"].sum(), 1e-9) for b in BLOCKS: vals[b].append(100 * m_sub[m_sub["block"]==b]["weight"].sum() / total_w) fig = make_subplots(rows=2, cols=1, shared_xaxes=True, subplot_titles=["SIMPLE","FILTRADO"], vertical_spacing=0.06) for b in BLOCKS: for row, vals in [(1, vals_s), (2, vals_f)]: texts = [f"{v:.0f}%" if v >= 8 else "" for v in vals[b]] fig.add_trace(go.Bar( name=BLOCK_LBL[b], x=labels, y=vals[b], marker_color=COLORS[b], text=texts, textposition="inside", textfont=dict(size=8, color="white"), showlegend=(row == 1), legendgroup=b, ), row=row, col=1) fig.update_layout(**base_layout( title=dict(text=title, font=dict(color=TEXT_COLOR, size=13), x=0.01), barmode="stack", height=700, xaxis2=dict(**axis_style(tickangle=-45, tickfont=dict(size=8))), yaxis=dict(**axis_style(title="% sequences", range=[0,100])), yaxis2=dict(**axis_style(title="% sequences", range=[0,100])), )) return fig fig2 = all_match_bar_fig(df_simple, df_fx_all, True, "Por partido — Racing ATACA") fig3 = all_match_bar_fig(df_simple, df_fx_all, False, "Por partido — Racing DEFIENDE") # ── Fig 4: % que difieren por partido ──────────────────────────────────────── merged_m = merged.merge(seq_info[["matchId","sequenceId","TeamName"]], on=["matchId","sequenceId"], how="left") diff_pm = (merged_m.groupby("matchId") .apply(lambda g: pd.Series({ "total": len(g), "difieren": (g["block_simple"] != g["block_fx"]).sum(), })).reset_index()) diff_pm["pct"] = 100 * diff_pm["difieren"] / diff_pm["total"].clip(lower=1) diff_pm["label"] = diff_pm["matchId"].map(lambda m: match_info.get(m,{}).get("label","?")) diff_pm = diff_pm.sort_values("matchId") fig4 = go.Figure() fig4.add_trace(go.Bar( x=diff_pm["label"], y=diff_pm["pct"], marker_color=COLORS["Medium"], text=[f"{v:.0f}%" for v in diff_pm["pct"]], textposition="outside", textfont=dict(color=TEXT_COLOR, size=9), )) fig4.update_layout(**base_layout( title=dict(text="% sequences con distinta clasificación entre Simple y Filtrado (Racing)", font=dict(color=TEXT_COLOR, size=13), x=0.01), height=380, xaxis=dict(**axis_style(tickangle=-45, tickfont=dict(size=8))), yaxis=dict(**axis_style(title="% difieren", range=[0,100])), )) # ── Fig 5: Matriz de transición ────────────────────────────────────────────── trans = merged.groupby(["block_simple","block_fx"]).size().reset_index(name="n") fig5 = go.Figure(go.Heatmap( x=[BLOCK_LBL[b] for b in BLOCKS], y=[BLOCK_LBL[b] for b in BLOCKS], z=[[trans[(trans["block_simple"]==bs)&(trans["block_fx"]==bc)]["n"].sum() for bc in BLOCKS] for bs in BLOCKS], colorscale="Blues", text=[[str(int(trans[(trans["block_simple"]==bs)&(trans["block_fx"]==bc)]["n"].sum())) for bc in BLOCKS] for bs in BLOCKS], texttemplate="%{text}", textfont=dict(size=12, color=TEXT_COLOR), hovertemplate="Simple=%{y}, Filtrado=%{x}: %{text}", xgap=2, ygap=2, )) fig5.update_layout(**base_layout( title=dict(text="Transición Simple → Filtrado (Racing)", font=dict(color=TEXT_COLOR, size=13), x=0.01), height=320, xaxis=dict(**axis_style(title="Filtrado")), yaxis=dict(**axis_style(title="Simple")), )) # --------------------------------------------------------------------------- # 8. HTML # --------------------------------------------------------------------------- print("Generando HTML...", flush=True) def stat_box(label, val, color="#F0A500"): return (f'
{val}' f'{label}
') html = f""" Comparación métodos bloque (sequenceId) — Racing

Comparación de métodos de bloque — unidad: sequenceId — Racing de Santander

Unidad de análisis: sequenceId de Opta (más granular que possessionId — no cruza interrupciones como faltas o saques de banda).
SIMPLE: phaseLabel dominante · filtro ≥3 eventos de bloque · sin reclasificaciones · Global = toda la liga ({df_pre['matchId'].nunique()} partidos)
FILTRADO: phaseLabel con desempate · filtro ≥3 eventos clave · multi-bloque ponderado para sequences largas (≥7 pases) · High→Med si 0 presiones Y defensor sin ≥5 jugadores en campo Racing · heurística para sequences sin etiqueta · Solo Racing ({len(all_racing_ids)} partidos)
{stat_box("SIMPLE — toda la liga", f"{simple_global_n:,}")} {stat_box("SIMPLE — Racing ataca", f"{simple_att_n:,}")} {stat_box("SIMPLE — Racing defiende", f"{simple_def_n:,}")} {stat_box("FILTRADO — Racing ataca", f"{fx_att_n:,}")} {stat_box("FILTRADO — Racing defiende", f"{fx_def_n:,}")} {stat_box("En común (Racing)", f"{total_shared:,}")} {stat_box("Difieren", f"{differ:,} ({100*differ/max(total_shared,1):.1f}%)", "#E06C5F")} {stat_box("Reclasif. High→Med", f"{reclassif_am:,}")}

1. Distribución global — donuts

{fig2html(fig1)}

2. Por partido — Racing ataca (Simple arriba · Filtrado abajo)

{fig2html(fig2)}

3. Por partido — Racing defiende (Simple arriba · Filtrado abajo)

{fig2html(fig3)}

4. % sequences con clasificación distinta entre métodos (Racing)

Calculado sobre sequences presentes en ambos métodos.

{fig2html(fig4)}

5. Matriz de transición Simple → Filtrado (Racing)

Fila = Simple · Columna = Filtrado · valores = n sequences

{fig2html(fig5)}
""" out = REPORTS_DIR / "comparacion_metodos_bloque_seq.html" out.write_text(html, encoding="utf-8") print(f"\nReporte: {out}")