""" Agente 01 "Formazioni & Assenze" — raccolta fatti via LLM + web search. L'LLM (API Claude con ricerca web) raccoglie SOLO FATTI con fonte: assenze, dubbi, rientri, turnover. Nessun numero di probabilità esce dall'LLM: la conversione in probabilità è del layer deterministico (impact.py). Il segnale viene salvato in agent_signals + predictions se la partita esiste nel DB (shadow mode: si logga, non si pubblica). Se la partita non è a DB, l'esito viene solo stampato. Uso: python scripts/agent_formazioni.py --league I1 --home Inter --away Torino \ --kickoff "2026-08-23 20:45" --stage S1 """ import argparse import json import re import sys from datetime import datetime from pathlib import Path import anthropic import pymysql import config import dixon_coles as dc import impact PROJECT_ROOT = Path(__file__).resolve().parent.parent MODEL_VERSION = "formazioni-v1" DEFAULT_MODEL = "claude-opus-4-8" ALLOWED_STATUS = {"confirmed_out", "doubtful", "rested", "suspended"} SYSTEM_PROMPT = """Sei l'agente "Formazioni & Assenze" di un sistema di previsione calcio. Il tuo UNICO compito è raccogliere FATTI verificabili sulle disponibilità dei giocatori per una specifica partita, tramite ricerche web su fonti affidabili (Sky Sport, Gazzetta, Corriere dello Sport, Tuttosport, siti/canali ufficiali dei club, Sos Fanta, Fantacalcio.it). REGOLE FERREE: 1. Solo fatti, mai valutazioni numeriche di probabilità dell'esito. 2. Ogni giocatore citato deve avere una fonte (URL reale trovato in ricerca). 3. Se non trovi notizie rilevanti, restituisci liste vuote: è un esito perfettamente valido. NON inventare mai nulla. 4. status ammessi: "confirmed_out" (infortunio/assenza certa), "suspended" (squalificato), "rested" (riposo/turnover annunciato), "doubtful" (in dubbio, con prob_out tra 0 e 1). 5. turnover_risk (0-1) per squadra: alto solo con segnali concreti (coppa imminente, dichiarazioni dell'allenatore). 6. Considera solo notizie recenti e riferite a QUESTA partita o al periodo immediatamente precedente. Ignora infortuni già rientrati. La tua risposta finale deve essere SOLO un blocco JSON (```json ... ```), con questo schema esatto: { "home": {"absent": [{"player": "...", "status": "confirmed_out", "reason": "...", "source_url": "..."}], "doubtful": [{"player": "...", "prob_out": 0.5, "reason": "...", "source_url": "..."}], "returning": [{"player": "...", "source_url": "..."}]}, "away": { come home }, "turnover_risk": {"home": 0.0, "away": 0.0, "reason": "..."}, "confidence": 0.8, "notes": "sintesi in una riga per il log umano" }""" def load_api_key() -> str | None: import os if os.environ.get("ANTHROPIC_API_KEY"): return os.environ["ANTHROPIC_API_KEY"] env_file = PROJECT_ROOT / ".env" if env_file.exists(): for line in env_file.read_text().splitlines(): if line.startswith("ANTHROPIC_API_KEY="): return line.split("=", 1)[1].strip() return None def gather_facts(client: anthropic.Anthropic, model: str, league: str, home: str, away: str, kickoff: str, stage: str) -> tuple[dict, dict]: """Interroga l'LLM con web search e restituisce (payload, usage_info).""" stage_hint = ( "Siamo a circa 24 ore dal calcio d'inizio: cerca formazioni probabili, " "infortunati, squalificati, dichiarazioni dell'allenatore, turnover." if stage == "S1" else "Siamo a meno di 75 minuti dal calcio d'inizio: cerca la FORMAZIONE " "UFFICIALE appena pubblicata e confrontala con le attese; segnala " "SOLO le sorprese (titolari attesi che mancano, rientri imprevisti)." ) user_msg = (f"Partita: {home} - {away} ({league}), calcio d'inizio {kickoff}. " f"Stage {stage}. {stage_hint}") web_search = {"type": "web_search_20260209", "name": "web_search", "max_uses": 8} if "haiku" in model: web_search["type"] = "web_search_20250305" kwargs = dict(model=model, max_tokens=16000, system=SYSTEM_PROMPT, tools=[web_search], messages=[{"role": "user", "content": user_msg}]) if "opus" in model or "sonnet" in model: kwargs["thinking"] = {"type": "adaptive"} messages = kwargs["messages"] for _ in range(4): # gestisce pause_turn del loop server-side response = client.messages.create(**kwargs) if response.stop_reason == "pause_turn": messages = [{"role": "user", "content": user_msg}, {"role": "assistant", "content": response.content}] kwargs["messages"] = messages continue break if response.stop_reason == "refusal": raise RuntimeError("La richiesta è stata rifiutata dal modello") text = "".join(b.text for b in response.content if b.type == "text") usage = {"input_tokens": response.usage.input_tokens, "output_tokens": response.usage.output_tokens, "stop_reason": response.stop_reason} m = re.search(r"```json\s*(\{.*?\})\s*```", text, re.DOTALL) raw = m.group(1) if m else text[text.find("{"):text.rfind("}") + 1] payload = json.loads(raw) return payload, usage def validate(payload: dict) -> dict: """Normalizza e sanifica il payload dell'LLM.""" out = {"home": {}, "away": {}, "turnover_risk": payload.get("turnover_risk") or {}, "confidence": float(payload.get("confidence") or 0.5), "notes": str(payload.get("notes") or "")[:500]} for side in ("home", "away"): src = payload.get(side) or {} absent = [e for e in (src.get("absent") or []) if e.get("player") and e.get("status") in ALLOWED_STATUS] doubtful = [e for e in (src.get("doubtful") or []) if e.get("player")] for e in doubtful: e["prob_out"] = min(max(float(e.get("prob_out") or 0.5), 0.0), 1.0) out[side] = {"absent": absent, "doubtful": doubtful, "returning": src.get("returning") or []} for side in ("home", "away"): risk = out["turnover_risk"].get(side) or 0.0 out["turnover_risk"][side] = min(max(float(risk), 0.0), 1.0) return out def main() -> int: parser = argparse.ArgumentParser(description="Agente formazioni: fatti via LLM -> probabilità") parser.add_argument("--league", required=True) parser.add_argument("--home", required=True) parser.add_argument("--away", required=True) parser.add_argument("--kickoff", required=True, help='es. "2026-08-23 20:45"') parser.add_argument("--stage", choices=["S1", "S2"], default="S1") parser.add_argument("--model", default=DEFAULT_MODEL) config.add_db_args(parser) args = parser.parse_args() api_key = load_api_key() if not api_key: print("Manca ANTHROPIC_API_KEY (env o file .env)") return 1 client = anthropic.Anthropic(api_key=api_key) print(f"[1/4] Raccolta fatti via {args.model} + web search...") payload, usage = gather_facts(client, args.model, args.league, args.home, args.away, args.kickoff, args.stage) payload = validate(payload) print(f" tokens: {usage['input_tokens']} in / {usage['output_tokens']} out") print(f" note agente: {payload['notes'] or '(nessuna)'}") print("[2/4] Conversione deterministica fatti -> delta...") conn = pymysql.connect(host=args.host, port=args.port, user=args.user, password=args.password, database="tigertips") deltas = {} for side, team in (("home", args.home), ("away", args.away)): squad = impact.load_squad(conn, team) facts = dict(payload[side]) facts["turnover_risk"] = payload["turnover_risk"][side] deltas[side] = impact.side_deltas(squad, facts) for line in deltas[side][2]: print(f" [{side}] {line}") if not deltas["home"][2] and not deltas["away"][2]: print(" nessun impatto rilevato (esito valido)") print("[3/4] Probabilità col modello DC aggiornato...") model = impact.fit_current(conn, args.league) prediction = impact.adjusted_prediction( model, args.home, args.away, {"_deltas_home": deltas["home"], "_deltas_away": deltas["away"]}) b, a = prediction["baseline"], prediction["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%}") print("\n[4/4] Salvataggio nel DB...") kickoff_date = datetime.strptime(args.kickoff, "%Y-%m-%d %H:%M").date() cur = conn.cursor() cur.execute(""" SELECT m.id FROM matches m JOIN teams h ON h.id = m.home_team_id JOIN teams a ON a.id = m.away_team_id WHERE m.match_date = %s AND h.name = %s AND a.name = %s """, (kickoff_date, args.home, args.away)) row = cur.fetchone() if row is None: print(" partita non presente in matches: segnale NON salvato " "(esegui prima sync_fixtures.py). Output solo a video.") else: match_id = row[0] cur.execute(""" INSERT INTO agent_signals (match_id, agent, stage, checked_at, payload, delta_lambda_home, delta_lambda_away, confidence, model_version) VALUES (%s, 'formazioni', %s, NOW(), %s, %s, %s, %s, %s) """, (match_id, args.stage, json.dumps(payload, ensure_ascii=False), round(deltas["home"][0] + deltas["home"][1], 4), round(deltas["away"][0] + deltas["away"][1], 4), payload["confidence"], f"{MODEL_VERSION}-{args.model}")) for source, probs in (("model", b), ("agent_adj", a)): cur.execute(""" INSERT INTO predictions (match_id, made_at, stage, source, prob_home, prob_draw, prob_away, details) VALUES (%s, NOW(), %s, %s, %s, %s, %s, %s) """, (match_id, args.stage, source, round(probs[0], 4), round(probs[1], 4), round(probs[2], 4), json.dumps({"deltas": {k: v[:2] for k, v in deltas.items()}}))) conn.commit() print(f" salvato: match_id={match_id}, segnale + 2 predizioni") conn.close() return 0 if __name__ == "__main__": sys.exit(main())