""" Layer deterministico dell'agente formazioni: converte FATTI (assenze, dubbi, turnover) in aggiustamenti dei gol attesi e produce le probabilità 1X2 aggiustate. Nessun LLM qui dentro: solo dati giocatore e modello DC. Logica (docs/agent-01-formazioni-design.md): - attaccanti/centrocampisti assenti: quota del contributo offensivo squadra (xG+xA stagionali) -> riduzione dell'attacco: delta = -K_ATT * quota - difensori/portieri assenti: quota minuti del reparto difensivo -> indebolimento della difesa: delta = -K_DEF * quota - dubbio ("doubtful"): pesato per prob_out (default 0.5) - turnover: attacco ridotto di TURNOVER_COEF * rischio - cap di sicurezza: |delta| <= 0.25 per squadra e componente Demo CLI (scenario simulato): python scripts/impact.py --league I1 --home Inter --away Torino \ --absent-home "Lautaro Martinez" --turnover-home 0.5 """ import argparse import sys import unicodedata from datetime import date import numpy as np import pandas as pd import pymysql import config import dixon_coles as dc K_ATT = 0.40 # il sostituto rende ~60% del titolare K_DEF = 0.25 TURNOVER_COEF = 0.10 DELTA_CAP = 0.25 XI = 0.0019 # stesso decadimento temporale del backtest ALPHA_XG = 0.7 # pseudo-gol: il mix con miglior standalone nel backtest STATUS_CERTAINTY = {"confirmed_out": 1.0, "suspended": 1.0, "rested": 1.0} def connect(args): return pymysql.connect(host=args.host, port=args.port, user=args.user, password=args.password, database="tigertips") def load_squad(conn, team: str) -> pd.DataFrame: """Rosa della squadra dalla stagione più recente disponibile in player_stats.""" q = """ SELECT player, position, minutes, goals, assists, xg, xa FROM player_stats WHERE team = %s AND season = (SELECT MAX(season) FROM player_stats WHERE team = %s) """ df = pd.read_sql(q, conn, params=(team, team)) for c in ("xg", "xa"): df[c] = df[c].astype(float) return df def _normalize(s: str) -> str: """minuscole + rimozione accenti, per un matching nomi robusto.""" return "".join(c for c in unicodedata.normalize("NFD", s.casefold()) if unicodedata.category(c) != "Mn") def find_player(squad: pd.DataFrame, name: str) -> pd.Series | None: hits = squad[squad["player"].map(_normalize).str.contains(_normalize(name), regex=False)] if hits.empty: print(f" [ATTENZIONE] '{name}' non trovato nella rosa, ignorato") return None if len(hits) > 1: hits = hits.sort_values("minutes", ascending=False) print(f" [nota] '{name}' ambiguo ({', '.join(hits['player'])}): " f"scelto {hits.iloc[0]['player']}") return hits.iloc[0] def is_defensive(position: str | None) -> bool: pos = (position or "").upper() return pos.startswith("D") or pos.startswith("GK") def side_deltas(squad: pd.DataFrame, facts: dict) -> tuple[float, float, list[str]]: """ Converte i fatti di una squadra in (delta_att, delta_dfn) in log-spazio dei gol attesi. Ritorna anche il dettaglio leggibile per il log. """ off_total = float((squad["xg"] + squad["xa"]).sum()) defenders = squad[squad["position"].map(is_defensive)] def_minutes_total = float(defenders["minutes"].sum()) delta_att, delta_dfn, details = 0.0, 0.0, [] entries = ([(e, STATUS_CERTAINTY.get(e.get("status"), 1.0)) for e in facts.get("absent", [])] + [(e, float(e.get("prob_out", 0.5))) for e in facts.get("doubtful", [])]) for entry, certainty in entries: p = find_player(squad, entry["player"]) if p is None: continue if is_defensive(p["position"]): share = p["minutes"] / def_minutes_total if def_minutes_total else 0.0 d = -K_DEF * share * certainty delta_dfn += d details.append(f"{p['player']} (dif, quota minuti reparto {share:.0%}, " f"certezza {certainty:.0%}): dfn {d:+.3f}") else: share = float(p["xg"] + p["xa"]) / off_total if off_total else 0.0 d = -K_ATT * share * certainty delta_att += d details.append(f"{p['player']} (att, quota xG+xA {share:.0%}, " f"certezza {certainty:.0%}): att {d:+.3f}") for e in facts.get("returning", []): details.append(f"{e.get('player')} (rientro): registrato, non usato in v1") turnover = float(facts.get("turnover_risk", 0.0)) if turnover > 0: d = -TURNOVER_COEF * turnover delta_att += d details.append(f"turnover (rischio {turnover:.0%}): att {d:+.3f}") delta_att = float(np.clip(delta_att, -DELTA_CAP, DELTA_CAP)) delta_dfn = float(np.clip(delta_dfn, -DELTA_CAP, DELTA_CAP)) return delta_att, delta_dfn, details def fit_current(conn, league: str, cutoff: str | None = None) -> dc.DixonColesFit: """ Modello DC per la lega, addestrato sulle partite fino a `cutoff` (default: oggi). Pseudo-gol con xG dove disponibile. """ q = """ SELECT m.match_date, h.name, a.name, m.home_goals, m.away_goals, m.home_xg, m.away_xg FROM matches m JOIN leagues l ON l.id = m.league_id JOIN teams h ON h.id = m.home_team_id JOIN teams a ON a.id = m.away_team_id WHERE l.code = %s AND m.result IS NOT NULL AND m.match_date < COALESCE(%s, DATE_ADD(CURDATE(), INTERVAL 1 DAY)) AND m.match_date >= DATE_SUB(COALESCE(%s, CURDATE()), INTERVAL 5 YEAR) ORDER BY m.match_date """ df = pd.read_sql(q, conn, params=(league, cutoff, cutoff)) df.columns = ["date", "home", "away", "hg", "ag", "xg_h", "xg_a"] for c in ("xg_h", "xg_a"): df[c] = df[c].astype(float) df["hg_t"] = np.where(df["xg_h"].notna(), (1 - ALPHA_XG) * df["hg"] + ALPHA_XG * df["xg_h"], df["hg"]) df["ag_t"] = np.where(df["xg_a"].notna(), (1 - ALPHA_XG) * df["ag"] + ALPHA_XG * df["xg_a"], df["ag"]) teams = sorted(set(df["home"]) | set(df["away"])) team_idx = {t: i for i, t in enumerate(teams)} ref_date = pd.Timestamp(cutoff) if cutoff else pd.Timestamp(date.today()) days_ago = (ref_date - pd.to_datetime(df["date"])).dt.days return dc.fit( home_idx=df["home"].map(team_idx).to_numpy(), away_idx=df["away"].map(team_idx).to_numpy(), home_goals=df["hg_t"].to_numpy(), away_goals=df["ag_t"].to_numpy(), weights=np.exp(-XI * days_ago.to_numpy()), teams=teams, ) def adjusted_prediction(model: dc.DixonColesFit, home: str, away: str, payload: dict) -> dict: """Probabilità baseline e aggiustate per una partita, dato il payload fatti.""" att_h, dfn_h = dc.team_params(model, home) att_a, dfn_a = dc.team_params(model, away) baseline = dc.predict(model, att_h, dfn_h, att_a, dfn_a) da_h, dd_h, det_h = payload.get("_deltas_home", (0, 0, [])) da_a, dd_a, det_a = payload.get("_deltas_away", (0, 0, [])) adjusted = dc.predict(model, att_h + da_h, dfn_h + dd_h, att_a + da_a, dfn_a + dd_a) return {"baseline": baseline, "adjusted": adjusted, "deltas": {"home": (da_h, dd_h), "away": (da_a, dd_a)}, "details": {"home": det_h, "away": det_a}} def main() -> int: parser = argparse.ArgumentParser(description="Demo layer deterministico assenze -> probabilità") parser.add_argument("--league", required=True) parser.add_argument("--home", required=True) parser.add_argument("--away", required=True) parser.add_argument("--absent-home", nargs="*", default=[]) parser.add_argument("--absent-away", nargs="*", default=[]) parser.add_argument("--doubtful-home", nargs="*", default=[]) parser.add_argument("--doubtful-away", nargs="*", default=[]) parser.add_argument("--turnover-home", type=float, default=0.0) parser.add_argument("--turnover-away", type=float, default=0.0) config.add_db_args(parser) args = parser.parse_args() conn = connect(args) print(f"Addestro il modello {args.league} aggiornato a oggi...") model = fit_current(conn, args.league) payload = {} for side, team in (("home", args.home), ("away", args.away)): squad = load_squad(conn, team) if squad.empty: print(f" [ATTENZIONE] nessun dato rosa per {team}") facts = { "absent": [{"player": n, "status": "confirmed_out"} for n in getattr(args, f"absent_{side}")], "doubtful": [{"player": n} for n in getattr(args, f"doubtful_{side}")], "turnover_risk": getattr(args, f"turnover_{side}"), } payload[f"_deltas_{side}"] = side_deltas(squad, facts) out = adjusted_prediction(model, args.home, args.away, payload) conn.close() print(f"\n{args.home} - {args.away}") for side in ("home", "away"): for line in out["details"][side]: print(f" [{side}] {line}") b, a = out["baseline"], out["adjusted"] print(f"\n{'':<12}{'1':>8}{'X':>8}{'2':>8}") print(f"{'baseline':<12}{b[0]:>8.1%}{b[1]:>8.1%}{b[2]:>8.1%}") print(f"{'aggiustata':<12}{a[0]:>8.1%}{a[1]:>8.1%}{a[2]:>8.1%}") return 0 if __name__ == "__main__": sys.exit(main())