351 lines
14 KiB
Python
351 lines
14 KiB
Python
"""Upper Brain Mixin — V27 cross-check guards and assessments.
|
|
|
|
Auto-extracted mixin module — split from services/single_match_orchestrator.py.
|
|
All methods here are composed into SingleMatchOrchestrator via inheritance.
|
|
`self` attributes (self.dsn, self.enrichment, self.v25_predictor, etc.) are
|
|
initialised in the main __init__.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import json
|
|
import re
|
|
import time
|
|
import math
|
|
import os
|
|
import pickle
|
|
from collections import defaultdict
|
|
from typing import Any, Dict, List, Optional, Set, Tuple, overload
|
|
|
|
import pandas as pd
|
|
import numpy as np
|
|
|
|
import psycopg2
|
|
from psycopg2.extras import RealDictCursor
|
|
|
|
from data.db import get_clean_dsn
|
|
from schemas.prediction import FullMatchPrediction
|
|
from schemas.match_data import MatchData
|
|
from models.v25_ensemble import V25Predictor, get_v25_predictor
|
|
try:
|
|
from models.v27_predictor import V27Predictor, compute_divergence, compute_value_edge
|
|
except ImportError:
|
|
class V27Predictor: # type: ignore[no-redef]
|
|
def __init__(self): self.models = {}
|
|
def load_models(self): return False
|
|
def predict_all(self, features): return {}
|
|
def compute_divergence(*args, **kwargs):
|
|
return {}
|
|
def compute_value_edge(*args, **kwargs):
|
|
return {}
|
|
from features.odds_band_analyzer import OddsBandAnalyzer
|
|
try:
|
|
from models.basketball_v25 import (
|
|
BasketballMatchPrediction,
|
|
get_basketball_v25_predictor,
|
|
)
|
|
except ImportError:
|
|
BasketballMatchPrediction = Any # type: ignore[misc]
|
|
def get_basketball_v25_predictor() -> Any:
|
|
raise ImportError("Basketball predictor is not available")
|
|
from core.engines.player_predictor import PlayerPrediction, get_player_predictor
|
|
from services.feature_enrichment import FeatureEnrichmentService
|
|
from services.betting_brain import BettingBrain
|
|
from services.v26_shadow_engine import V26ShadowEngine, get_v26_shadow_engine
|
|
from services.match_commentary import generate_match_commentary
|
|
from utils.top_leagues import load_top_league_ids
|
|
from utils.league_reliability import load_league_reliability
|
|
from config.config_loader import build_threshold_dict, get_threshold_default
|
|
from models.calibration import get_calibrator
|
|
|
|
|
|
class UpperBrainMixin:
|
|
def _apply_upper_brain_guards(self, package: Dict[str, Any]) -> Dict[str, Any]:
|
|
return BettingBrain().judge(package)
|
|
|
|
v27_engine = package.get("v27_engine")
|
|
if not isinstance(v27_engine, dict) or not v27_engine.get("triple_value"):
|
|
return package
|
|
|
|
guarded = dict(package)
|
|
vetoed_keys = set()
|
|
guarded_keys = set()
|
|
|
|
def mark_guard(item: Dict[str, Any]) -> Dict[str, Any]:
|
|
if not isinstance(item, dict):
|
|
return item
|
|
|
|
out = dict(item)
|
|
assessment = self._upper_brain_assessment(out, guarded)
|
|
if not assessment.get("applies"):
|
|
return out
|
|
|
|
key = f"{out.get('market')}:{out.get('pick')}"
|
|
guarded_keys.add(key)
|
|
out["upper_brain"] = assessment
|
|
|
|
reason_key = "decision_reasons" if "decision_reasons" in out else "reasons"
|
|
reasons = list(out.get(reason_key) or [])
|
|
for reason in assessment.get("reason_codes", []):
|
|
if reason not in reasons:
|
|
reasons.append(reason)
|
|
out[reason_key] = reasons[:6]
|
|
|
|
if assessment.get("veto"):
|
|
vetoed_keys.add(key)
|
|
out["playable"] = False
|
|
out["stake_units"] = 0.0
|
|
out["bet_grade"] = "PASS"
|
|
out["is_guaranteed"] = False
|
|
out["pick_reason"] = "upper_brain_veto"
|
|
if "signal_tier" in out:
|
|
out["signal_tier"] = "PASS"
|
|
elif assessment.get("downgrade"):
|
|
out["is_guaranteed"] = False
|
|
if out.get("signal_tier") == "CORE":
|
|
out["signal_tier"] = "LEAN"
|
|
if out.get("pick_reason") == "high_accuracy_market":
|
|
out["pick_reason"] = "upper_brain_downgraded"
|
|
|
|
return out
|
|
|
|
main_pick = mark_guard(guarded.get("main_pick") or {})
|
|
value_pick = mark_guard(guarded.get("value_pick") or {}) if guarded.get("value_pick") else None
|
|
supporting = [
|
|
mark_guard(row)
|
|
for row in list(guarded.get("supporting_picks") or [])
|
|
if isinstance(row, dict)
|
|
]
|
|
bet_summary = [
|
|
mark_guard(row)
|
|
for row in list(guarded.get("bet_summary") or [])
|
|
if isinstance(row, dict)
|
|
]
|
|
|
|
main_safe = bool(main_pick and main_pick.get("playable") and not main_pick.get("upper_brain", {}).get("veto"))
|
|
if not main_safe:
|
|
candidates = [
|
|
row for row in supporting
|
|
if row.get("playable")
|
|
and not row.get("upper_brain", {}).get("veto")
|
|
and float(row.get("odds", 0.0) or 0.0) >= 1.30
|
|
]
|
|
candidates.sort(key=lambda row: float(row.get("play_score", 0.0) or 0.0), reverse=True)
|
|
if candidates:
|
|
main_pick = dict(candidates[0])
|
|
main_pick["is_guaranteed"] = False
|
|
main_pick["pick_reason"] = "upper_brain_reselected"
|
|
reasons = list(main_pick.get("decision_reasons") or [])
|
|
if "upper_brain_reselected_after_veto" not in reasons:
|
|
reasons.append("upper_brain_reselected_after_veto")
|
|
main_pick["decision_reasons"] = reasons[:6]
|
|
elif main_pick:
|
|
main_pick["is_guaranteed"] = False
|
|
main_pick["pick_reason"] = "upper_brain_no_safe_pick"
|
|
|
|
if main_pick:
|
|
supporting = [
|
|
row for row in supporting
|
|
if not (
|
|
row.get("market") == main_pick.get("market")
|
|
and row.get("pick") == main_pick.get("pick")
|
|
)
|
|
][:6]
|
|
|
|
guarded["main_pick"] = main_pick if main_pick else None
|
|
guarded["value_pick"] = value_pick
|
|
guarded["supporting_picks"] = supporting
|
|
guarded["bet_summary"] = bet_summary
|
|
|
|
playable = bool(main_pick and main_pick.get("playable") and not main_pick.get("upper_brain", {}).get("veto"))
|
|
advice = dict(guarded.get("bet_advice") or {})
|
|
advice["playable"] = playable
|
|
advice["suggested_stake_units"] = float(main_pick.get("stake_units", 0.0)) if playable else 0.0
|
|
if playable:
|
|
advice["reason"] = "playable_pick_found"
|
|
elif vetoed_keys:
|
|
advice["reason"] = "upper_brain_no_safe_pick"
|
|
else:
|
|
advice["reason"] = "no_bet_conditions_met"
|
|
guarded["bet_advice"] = advice
|
|
|
|
guarded["upper_brain"] = {
|
|
"applied": True,
|
|
"guarded_count": len(guarded_keys),
|
|
"vetoed_count": len(vetoed_keys),
|
|
"vetoed": sorted(vetoed_keys)[:8],
|
|
"rules": {
|
|
"min_band_sample": 8,
|
|
"max_v25_v27_divergence": 0.18,
|
|
"dc_requires_triple_value": True,
|
|
},
|
|
}
|
|
guarded.setdefault("analysis_details", {})
|
|
guarded["analysis_details"]["upper_brain_guards_applied"] = True
|
|
guarded["analysis_details"]["upper_brain_vetoed_count"] = len(vetoed_keys)
|
|
return guarded
|
|
|
|
def _upper_brain_assessment(
|
|
self,
|
|
item: Dict[str, Any],
|
|
package: Dict[str, Any],
|
|
) -> Dict[str, Any]:
|
|
market = str(item.get("market") or "")
|
|
pick = str(item.get("pick") or "")
|
|
if not market or not pick:
|
|
return {"applies": False}
|
|
|
|
v27_engine = package.get("v27_engine") or {}
|
|
triple_value = v27_engine.get("triple_value") or {}
|
|
model_prob = self._upper_brain_market_probability(item, package)
|
|
v27_prob = self._upper_brain_v27_probability(market, pick, v27_engine)
|
|
triple_key = self._upper_brain_triple_key(market, pick)
|
|
triple = triple_value.get(triple_key) if triple_key else None
|
|
|
|
veto = False
|
|
downgrade = False
|
|
reasons: List[str] = []
|
|
divergence = None
|
|
|
|
if model_prob is not None and v27_prob is not None:
|
|
divergence = abs(float(model_prob) - float(v27_prob))
|
|
if divergence >= 0.18:
|
|
veto = True
|
|
reasons.append("upper_brain_v25_v27_divergence")
|
|
elif divergence >= 0.12:
|
|
downgrade = True
|
|
reasons.append("upper_brain_v25_v27_warning")
|
|
|
|
if isinstance(triple, dict):
|
|
band_sample = int(float(triple.get("band_sample", 0) or 0))
|
|
is_value = bool(triple.get("is_value"))
|
|
if market == "DC":
|
|
if band_sample < 8:
|
|
veto = True
|
|
reasons.append("upper_brain_band_sample_too_low")
|
|
elif not is_value:
|
|
veto = True
|
|
reasons.append("upper_brain_triple_value_rejected")
|
|
elif market in {"MS", "OU25"} and band_sample > 0 and band_sample < 8:
|
|
downgrade = True
|
|
reasons.append("upper_brain_band_sample_thin")
|
|
elif market in {"OU15", "HT_OU05"} and band_sample < 8:
|
|
downgrade = True
|
|
reasons.append("upper_brain_band_sample_thin")
|
|
|
|
consensus = str(v27_engine.get("consensus") or "").upper()
|
|
if consensus == "DISAGREE" and market in {"MS", "DC"} and not veto:
|
|
downgrade = True
|
|
reasons.append("upper_brain_consensus_disagree")
|
|
|
|
applies = bool(reasons or triple is not None or v27_prob is not None)
|
|
return {
|
|
"applies": applies,
|
|
"veto": veto,
|
|
"downgrade": downgrade,
|
|
"reason_codes": reasons,
|
|
"model_prob": round(float(model_prob), 4) if model_prob is not None else None,
|
|
"v27_prob": round(float(v27_prob), 4) if v27_prob is not None else None,
|
|
"divergence": round(float(divergence), 4) if divergence is not None else None,
|
|
"triple_key": triple_key,
|
|
"triple_value": triple,
|
|
}
|
|
|
|
def _upper_brain_market_probability(
|
|
self,
|
|
item: Dict[str, Any],
|
|
package: Dict[str, Any],
|
|
) -> Optional[float]:
|
|
raw_prob = item.get("probability")
|
|
if raw_prob is not None:
|
|
try:
|
|
return float(raw_prob)
|
|
except (TypeError, ValueError):
|
|
pass
|
|
|
|
market = str(item.get("market") or "")
|
|
pick = str(item.get("pick") or "")
|
|
board = package.get("market_board") or {}
|
|
payload = board.get(market) if isinstance(board, dict) else None
|
|
probs = payload.get("probs") if isinstance(payload, dict) else None
|
|
if not isinstance(probs, dict):
|
|
return None
|
|
|
|
prob_key = self._upper_brain_prob_key(market, pick)
|
|
if prob_key is None:
|
|
return None
|
|
return self._safe_float(probs.get(prob_key))
|
|
|
|
def _upper_brain_v27_probability(
|
|
self,
|
|
market: str,
|
|
pick: str,
|
|
v27_engine: Dict[str, Any],
|
|
) -> Optional[float]:
|
|
predictions = v27_engine.get("predictions") or {}
|
|
ms = predictions.get("ms") or {}
|
|
ou25 = predictions.get("ou25") or {}
|
|
|
|
if market == "MS":
|
|
ms_key = {"1": "home", "X": "draw", "2": "away"}.get(pick or "")
|
|
return self._safe_float(ms.get(ms_key), 0.0) if ms_key else 0.0
|
|
if market == "DC":
|
|
if pick == "1X":
|
|
return self._safe_float(ms.get("home"), 0.0) + self._safe_float(ms.get("draw"), 0.0)
|
|
if pick == "X2":
|
|
return self._safe_float(ms.get("draw"), 0.0) + self._safe_float(ms.get("away"), 0.0)
|
|
if pick == "12":
|
|
return self._safe_float(ms.get("home"), 0.0) + self._safe_float(ms.get("away"), 0.0)
|
|
if market == "OU25":
|
|
prob_key = self._upper_brain_prob_key(market, pick)
|
|
return self._safe_float(ou25.get(prob_key), 0.0) if prob_key else 0.0
|
|
return 0.0
|
|
|
|
@staticmethod
|
|
def _upper_brain_prob_key(market: str, pick: str) -> Optional[str]:
|
|
pick_norm = str(pick or "").strip().casefold()
|
|
if market in {"MS", "HT", "HCAP"}:
|
|
return pick if pick in {"1", "X", "2"} else None
|
|
if market == "DC":
|
|
return pick.upper() if pick.upper() in {"1X", "X2", "12"} else None
|
|
if market in {"OU15", "OU25", "OU35", "HT_OU05", "HT_OU15", "CARDS"}:
|
|
if "over" in pick_norm or "st" in pick_norm:
|
|
return "over"
|
|
if "under" in pick_norm or "alt" in pick_norm:
|
|
return "under"
|
|
if market == "BTTS":
|
|
if "yes" in pick_norm or "var" in pick_norm:
|
|
return "yes"
|
|
if "no" in pick_norm or "yok" in pick_norm:
|
|
return "no"
|
|
if market == "OE":
|
|
if "odd" in pick_norm or "tek" in pick_norm:
|
|
return "odd"
|
|
if "even" in pick_norm or "ift" in pick_norm:
|
|
return "even"
|
|
if market == "HTFT" and "/" in pick:
|
|
return pick
|
|
return None
|
|
|
|
def _upper_brain_triple_key(self, market: str, pick: str) -> Optional[str]:
|
|
prob_key = self._upper_brain_prob_key(market, pick)
|
|
if market == "MS":
|
|
return {"1": "home", "2": "away"}.get(pick)
|
|
if market == "DC":
|
|
return f"dc_{pick.lower()}" if pick.upper() in {"1X", "X2", "12"} else None
|
|
if market in {"OU15", "OU25", "OU35"} and prob_key == "over":
|
|
return f"{market.lower()}_over"
|
|
if market == "BTTS" and prob_key == "yes":
|
|
return "btts_yes"
|
|
if market == "HT":
|
|
return {"1": "ht_home", "2": "ht_away"}.get(pick)
|
|
if market in {"HT_OU05", "HT_OU15"} and prob_key == "over":
|
|
return f"{market.lower()}_over"
|
|
if market == "OE" and prob_key == "odd":
|
|
return "oe_odd"
|
|
if market == "CARDS" and prob_key == "over":
|
|
return "cards_over"
|
|
if market == "HTFT" and "/" in pick:
|
|
return f"htft_{pick.replace('/', '').lower()}"
|
|
return None
|