from __future__ import annotations from datetime import datetime, timezone from statistics import pstdev from app.config import MODEL_VERSION from app.core.calibration import calibrate_probability from app.core.competitions import MIN_COMPETITION_HISTORY, competition_for_sport_key from app.core.market import market_consensus from app.core.names import build_team_catalog, resolve_event_pair from app.core.stats import build_elo, predictive_models, tune_ensemble_weights from app.models import FinishedMatch, Pick def _clamp(x: float, lo: float = 0.0, hi: float = 1.0) -> float: return max(lo, min(hi, x)) def _renormalize(values: tuple[float, float, float]) -> tuple[float, float, float]: total = sum(values) if total <= 0: return 1 / 3, 1 / 3, 1 / 3 return tuple(v / total for v in values) # type: ignore[return-value] def _market_quality(bookmakers: int, dispersion: float, stale: int) -> float: depth = _clamp((bookmakers - 1) / 5.0) stability = _clamp(1.0 - dispersion / 0.075) freshness = _clamp(1.0 - stale / max(1.0, bookmakers + stale)) return 0.48 * depth + 0.38 * stability + 0.14 * freshness def _safe_score( probability: float, conservative: float, data_quality: float, reliability: float, agreement: float, market_quality: float, edge: float, ) -> float: probability_component = _clamp((probability - 0.58) / 0.25) conservative_component = _clamp((conservative - 0.53) / 0.20) value_component = _clamp((edge + 0.015) / 0.075) return 100.0 * ( 0.31 * probability_component + 0.25 * conservative_component + 0.13 * data_quality + 0.12 * reliability + 0.08 * agreement + 0.07 * market_quality + 0.04 * value_component ) def _kickoff(event: dict) -> datetime | None: raw = event.get("commence_time") if not raw: return None try: dt = datetime.fromisoformat(str(raw).replace("Z", "+00:00")) return dt if dt.tzinfo else dt.replace(tzinfo=timezone.utc) except Exception: return None def analyze_events( events: list[dict], matches: list[FinishedMatch], min_safe_score: float, limit: int, *, calibration_history: list[dict] | None = None, previous_picks: list[dict] | None = None, min_probability: float = 0.64, min_conservative_probability: float = 0.57, min_bookmakers: int = 3, min_name_score: float = 82.0, ) -> tuple[list[Pick], list[dict]]: if not matches: return [], [{"reason": "sem histórico"}] calibration_history = calibration_history or [] previous_by_event = { str(p.get("event_id")): p for p in (previous_picks or []) if p.get("event_id") } competitions = sorted({m.competition for m in matches}) matches_by_comp = { code: [m for m in matches if m.competition == code] for code in competitions } catalogs = { code: build_team_catalog(comp_matches, code) for code, comp_matches in matches_by_comp.items() } elo_by_comp = { code: build_elo(comp_matches) for code, comp_matches in matches_by_comp.items() } tuning_by_comp = { code: tune_ensemble_weights(comp_matches, code) for code, comp_matches in matches_by_comp.items() } picks: list[Pick] = [] rejected: list[dict] = [] for event in events: home_api = str(event.get("home_team") or "") away_api = str(event.get("away_team") or "") sport_key = str(event.get("_sport_key") or "") spec = competition_for_sport_key(sport_key) event_name = f"{home_api} x {away_api}" if not spec: rejected.append({"event": event_name, "reason": "competição sem mapeamento seguro"}) continue comp_code = spec.football_data_code comp_matches = matches_by_comp.get(comp_code, []) if len(comp_matches) < MIN_COMPETITION_HISTORY: rejected.append({ "event": event_name, "reason": f"histórico insuficiente em {comp_code} ({len(comp_matches)} jogos)", }) continue kickoff = _kickoff(event) if kickoff is None: rejected.append({"event": event_name, "reason": "horário inválido"}) continue if kickoff <= datetime.now(timezone.utc): rejected.append({"event": event_name, "reason": "evento já iniciado"}) continue market = market_consensus(event) if market.bookmakers < 1 or market.home_prob is None or market.away_prob is None or market.draw_prob is None: rejected.append({"event": event_name, "reason": "sem consenso H2H utilizável"}) continue catalog = catalogs.get(comp_code, []) home_identity, away_identity, name_confidence, name_detail = resolve_event_pair( home_api, away_api, catalog, minimum=min_name_score, ) if not home_identity or not away_identity: rejected.append({ "event": event_name, "reason": ( "matching de times ambíguo " f"(casa {name_detail['home_score']:.0f}, fora {name_detail['away_score']:.0f})" ), }) continue model = predictive_models( home_identity.key, away_identity.key, comp_matches, elo_by_comp.get(comp_code, {}), competition=comp_code, as_of=kickoff, ensemble_weights=tuple(tuning_by_comp[comp_code]["weights"]), ) poisson = tuple(float(v) for v in model["poisson"]) elo_p = tuple(float(v) for v in model["elo"]) form = tuple(float(v) for v in model["form"]) internal = tuple(float(v) for v in model["ensemble"]) data_quality = float(model["quality"]) market_vector = ( float(market.home_prob), float(market.draw_prob), float(market.away_prob), ) market_q = _market_quality( market.bookmakers, market.dispersion, market.stale_bookmakers, ) tuning = tuning_by_comp[comp_code] tuning_total_samples = float(tuning["samples"]) tuning_samples = float(tuning.get("validation_samples", tuning_total_samples)) tuning_skill = float(tuning.get("brier_skill", 0.0)) if tuning_samples >= 10: sample_validation = _clamp((tuning_samples - 10.0) / 12.0) skill_validation = _clamp((tuning_skill + 0.03) / 0.12) model_validation = 0.35 * sample_validation + 0.65 * skill_validation else: # Unknown is not the same as bad. Keep the model usable, but make the # current market prior more influential until walk-forward evidence grows. model_validation = 0.45 overall_disagreement = max( pstdev([poisson[i], elo_p[i], form[i]]) for i in range(3) ) agreement = _clamp(1.0 - overall_disagreement / 0.11) # The betting market is treated as a strong prior, not as a model feature. # Good internal data earns more weight; weak/unstable data is shrunk harder # toward the de-vig market consensus. base_internal_weight = _clamp( 0.36 + 0.20 * data_quality + 0.08 * agreement + 0.05 * (1.0 - market_q), 0.36, 0.67, ) # Out-of-sample validation acts as a trust regulator. A model that has not # demonstrated skill does not get to overpower a deep current market simply # because its internal components happen to agree. internal_weight = _clamp( base_internal_weight * (0.82 + 0.18 * model_validation), 0.32, 0.65, ) posterior_vector = _renormalize(tuple( internal_weight * internal[i] + (1.0 - internal_weight) * market_vector[i] for i in range(3) )) candidate_rows = [ ("home", home_api, 0, market.home_odd, market.home_prob), ("away", away_api, 2, market.away_odd, market.away_prob), ] best = None best_rejected = None previous = previous_by_event.get(str(event.get("id") or "")) for side, selection, idx, odd, mprob in candidate_rows: if odd is None or mprob is None: continue side_market_dispersion = ( market.home_dispersion if side == "home" else market.away_dispersion ) side_disagreement = pstdev([poisson[idx], elo_p[idx], form[idx]]) side_agreement = _clamp(1.0 - side_disagreement / 0.11) raw_p = float(internal[idx]) anchored_p = float(posterior_vector[idx]) core_model_floor = min(float(poisson[idx]), float(elo_p[idx])) calibrated_p, calibration_meta = calibrate_probability( anchored_p, calibration_history, model_version=MODEL_VERSION, competition_code=comp_code, ) reliability = _clamp( 0.30 * data_quality + 0.22 * side_agreement + 0.18 * market_q + 0.20 * name_confidence + 0.10 * model_validation ) # This is deliberately a reliability shrinkage, not a claimed # frequentist confidence interval. conservative = 0.5 + max(0.0, calibrated_p - 0.5) * reliability edge = calibrated_p - float(mprob) ev = calibrated_p * float(odd) - 1.0 score = _safe_score( calibrated_p, conservative, data_quality, reliability, side_agreement, market_q, edge, ) market_move = 0.0 selection_changed = False version_conflict = bool( previous and previous.get("model_version") != MODEL_VERSION ) if previous and previous.get("model_version") == MODEL_VERSION: if previous.get("side") == side and isinstance(previous.get("market_probability"), (int, float)): market_move = float(mprob) - float(previous["market_probability"]) elif previous.get("side") and previous.get("side") != side: selection_changed = True reasons: list[str] = [] if name_confidence < min_name_score / 100.0: reasons.append("matching de time abaixo do mínimo") if market.bookmakers < min_bookmakers: reasons.append(f"poucas casas no consenso ({market.bookmakers})") if side_market_dispersion > 0.060: reasons.append("mercado muito disperso para a seleção") if data_quality < 0.52: reasons.append("qualidade de dados insuficiente") if reliability < 0.60: reasons.append("confiabilidade combinada abaixo do mínimo") if calibrated_p < min_probability: reasons.append("probabilidade abaixo do filtro") if conservative < min_conservative_probability: reasons.append("probabilidade conservadora baixa") if side_disagreement > 0.095: reasons.append("modelos divergentes") if core_model_floor < 0.50 and calibrated_p < 0.74: reasons.append("Poisson/Elo não sustentam o favorito com segurança") if abs(raw_p - float(mprob)) > 0.17: reasons.append("modelo interno diverge demais do mercado") if not 1.15 <= float(odd) <= 2.15: reasons.append("odd de referência fora da faixa SAFE") if ev < 0.0: reasons.append("retorno esperado negativo no preço de referência") if market_move < -0.04: reasons.append("movimento de mercado relevante contra a seleção") if selection_changed: reasons.append("seleção mudou desde o último scan") if version_conflict: reasons.append("evento já rastreado por uma versão anterior") # SafeScore is a transparent ranking/label. Approval is controlled by # the explicit safety gates above, avoiding a second, opaque veto over # candidates that already satisfy every measurable requirement. row = { "side": side, "selection": selection, "idx": idx, "odd": float(odd), "mprob": float(mprob), "raw_p": raw_p, "p": calibrated_p, "conservative": conservative, "edge": edge, "ev": ev, "score": score, "reliability": reliability, "disagreement": side_disagreement, "market_move": market_move, "market_dispersion": side_market_dispersion, "core_model_floor": core_model_floor, "model_validation": model_validation, "calibration_delta": float(calibration_meta["delta"]), "calibration_samples": float(calibration_meta["effective_samples"]), "reasons": reasons, "models": { "poisson": poisson[idx], "elo": elo_p[idx], "form": form[idx], "internal": raw_p, "market": float(mprob), "posterior_pre_calibration": anchored_p, "lambda_home": float(model["lambda_home"]), "lambda_away": float(model["lambda_away"]), "rho": float(model["rho"]), "league_draw_rate": float(model["league_draw_rate"]), "league_sample": float(model["league_sample"]), "internal_weight": internal_weight, "base_internal_weight": base_internal_weight, "model_validation": model_validation, "core_model_floor": core_model_floor, "calibration_samples": float(calibration_meta["effective_samples"]), "weight_poisson": float(model["weight_poisson"]), "weight_elo": float(model["weight_elo"]), "weight_form": float(model["weight_form"]), "tuning_samples": tuning_samples, "tuning_total_samples": tuning_total_samples, "tuning_brier": float(tuning["brier"]), "tuning_climatology_brier": float(tuning.get("climatology_brier", 0.0)), "tuning_brier_skill": tuning_skill, "tuning_gain": float(tuning["gain"]), }, } rank = (row["conservative"], row["score"], row["ev"]) if row["reasons"]: if best_rejected is None or rank > ( best_rejected["conservative"], best_rejected["score"], best_rejected["ev"], ): best_rejected = row elif best is None or rank > ( best["conservative"], best["score"], best["ev"], ): best = row if best is None and best_rejected is None: rejected.append({"event": event_name, "reason": "mercado incompleto"}) continue if best is None: assert best_rejected is not None blockers = list(best_rejected["reasons"]) rejected.append({ "approved": False, "event_id": str(event.get("id") or f"{home_api}-{away_api}-{kickoff.isoformat()}"), "event": event_name, "kickoff": kickoff.isoformat(), "competition": spec.label, "competition_code": comp_code, "home": home_api, "away": away_api, "selection": best_rejected["selection"], "side": best_rejected["side"], "odd": round(best_rejected["odd"], 3), "fair_odd": round(1.0 / max(best_rejected["p"], 1e-9), 3), "market_probability": round(best_rejected["mprob"], 4), "model_ev": round(best_rejected["ev"], 4), "edge": round(best_rejected["edge"], 4), "quality": round(data_quality, 4), "reliability": round(best_rejected["reliability"], 4), "market_bookmakers": market.bookmakers, "label": "EM OBSERVAÇÃO", "blockers": blockers, "reason": "; ".join(blockers), "safe_score": round(best_rejected["score"], 1), "probability": round(best_rejected["p"], 4), "conservative_probability": round(best_rejected["conservative"], 4), }) continue score = float(best["score"]) label = "ULTRA SELECTIVO" if score >= 89 else "SAFE" if score >= 82 else "SELECTIVO" why: list[str] = [] if data_quality >= 0.82: why.append("amostra forte") else: why.append("amostra aprovada") if best["disagreement"] <= 0.035: why.append("modelos muito alinhados") elif best["disagreement"] <= 0.065: why.append("modelos alinhados") if market.bookmakers >= 5: why.append(f"consenso de {market.bookmakers} casas") else: why.append(f"consenso de {market.bookmakers} casas") if best["conservative"] >= 0.65: why.append("forte margem conservadora") if best["market_move"] > 0.025: why.append("mercado moveu a favor") if best["calibration_samples"] >= 12: why.append("calibração forward ativa") why.append("Risk Gate aprovado") picks.append(Pick( event_id=str(event.get("id") or f"{home_api}-{away_api}-{kickoff.isoformat()}"), kickoff=kickoff.isoformat(), competition=spec.label, competition_code=comp_code, home=home_api, away=away_api, resolved_home_key=home_identity.key, resolved_away_key=away_identity.key, selection=best["selection"], side=best["side"], odd=round(best["odd"], 3), probability=round(best["p"], 4), raw_model_probability=round(best["raw_p"], 4), conservative_probability=round(best["conservative"], 4), market_probability=round(best["mprob"], 4), fair_odd=round(1.0 / max(best["p"], 1e-9), 3), model_ev=round(best["ev"], 4), edge=round(best["edge"], 4), safe_score=round(score, 1), quality=round(data_quality, 4), reliability=round(best["reliability"], 4), disagreement=round(best["disagreement"], 4), market_dispersion=round(best["market_dispersion"], 4), market_bookmakers=market.bookmakers, name_confidence=round(name_confidence, 4), calibration_delta=round(best["calibration_delta"], 4), market_move=round(best["market_move"], 4), label=label, reasons=why, model_detail={k: round(v, 4) for k, v in best["models"].items()}, model_version=MODEL_VERSION, )) picks.sort( key=lambda p: ( p.conservative_probability, p.safe_score, p.reliability, p.model_ev, ), reverse=True, ) return picks[:limit], rejected