from __future__ import annotations import json import math import os import uuid from pathlib import Path from typing import Any, Dict, Iterable, List, Optional, Tuple from utils.top_leagues import load_top_league_ids class V26ShadowEngine: """ROI-first shadow engine built on top of stable pre-match signals.""" DEFAULT_CONFIG = { "version": "v26.shadow.0", "calibration_version": "v26.shadow.calib.0", "selection_weights": { "edge": 0.44, "confidence": 0.32, "quality": 0.14, "reliability": 0.10, }, "goal_model": { "max_goals": 7, "halftime_ratio": 0.46, "ms_blend_weight": 0.60, "ht_blend_weight": 0.45, "htft_v25_blend_weight": 0.35, "hcap_v25_blend_weight": 0.30, }, "core_markets": ["MS", "DC", "OU15", "OU25", "BTTS", "HT_OU05"], "top_league_market_overrides": { "MS": { "min_confidence": 56.0, "min_edge": 0.03, "min_play_score": 69.0, }, "OU15": { "min_confidence": 74.0, "min_edge": 0.05, "min_play_score": 76.0, "confidence_multiplier": 0.96, }, "OU25": { "min_confidence": 63.0, "min_edge": 0.05, "min_play_score": 74.0, "confidence_multiplier": 0.94, }, "OU35": { "min_confidence": 67.0, "min_edge": 0.05, "min_play_score": 78.0, "confidence_multiplier": 0.9, }, "BTTS": { "min_confidence": 66.0, "min_edge": 0.06, "min_play_score": 76.0, "confidence_multiplier": 0.92, }, "HT_OU05": { "min_confidence": 76.0, "min_edge": 0.08, "min_play_score": 84.0, "confidence_multiplier": 0.88, "weak_market": True, }, "HT_OU15": { "min_confidence": 72.0, "min_edge": 0.08, "min_play_score": 84.0, "confidence_multiplier": 0.84, "weak_market": True, }, }, "top_league_pick_overrides": { "OU15:Over": { "min_confidence": 78.0, "min_edge": 0.09, "min_play_score": 82.0, "min_odds": 1.30, }, "OU25:Over": { "min_confidence": 67.0, "min_edge": 0.08, "min_play_score": 79.0, "min_odds": 1.68, }, "BTTS:Yes": { "min_confidence": 69.0, "min_edge": 0.09, "min_play_score": 80.0, "min_odds": 1.72, }, "HT_OU05:Over": { "disabled": True, "disabled_reason": "top_league_ht_ou05_over_disabled", }, }, "market_profiles": {}, } MARKET_LABELS = { "MS": ("1", "X", "2"), "DC": ("1X", "X2", "12"), "OU15": ("Under", "Over"), "OU25": ("Under", "Over"), "OU35": ("Under", "Over"), "BTTS": ("No", "Yes"), "HT": ("1", "X", "2"), "HT_OU05": ("Under", "Over"), "HT_OU15": ("Under", "Over"), "HTFT": ("1/1", "1/X", "1/2", "X/1", "X/X", "X/2", "2/1", "2/X", "2/2"), "OE": ("Even", "Odd"), "CARDS": ("Under", "Over"), "HCAP": ("1", "X", "2"), } SCORELINE_MARKETS = {"OU15", "OU25", "OU35", "BTTS", "OE", "HT_OU05", "HT_OU15"} WEAK_MARKETS = {"HT", "HT_OU15", "HTFT", "CARDS", "HCAP"} def __init__(self, config_path: Optional[str] = None) -> None: base_dir = Path(__file__).resolve().parents[1] default_path = base_dir / "models" / "v26_shadow" / "market_profiles.json" self.config_path = Path(config_path) if config_path else default_path self.config = self._load_config() self.version = str(self.config.get("version") or "v26.shadow.0") self.calibration_version = str( self.config.get("calibration_version") or f"{self.version}.calib" ) self.market_profiles = self.config.get("market_profiles", {}) self.selection_weights = self.config.get("selection_weights", {}) self.goal_model = self.config.get("goal_model", {}) self.core_markets = set(self.config.get("core_markets", [])) self.top_league_market_overrides = self.config.get( "top_league_market_overrides", {} ) self.top_league_pick_overrides = self.config.get( "top_league_pick_overrides", {} ) self.top_league_ids = load_top_league_ids() self._team_pattern_cache: Dict[Tuple[str, int], Dict[str, float]] = {} self._odds_band_prior_cache: Optional[Dict[str, Dict[str, float]]] = None self._referee_prior_cache: Optional[Dict[str, Dict[str, float]]] = None self._league_prior_cache: Optional[Dict[str, Dict[str, float]]] = None def readiness_summary(self) -> Dict[str, Any]: return { "config_path": str(self.config_path), "config_loaded": bool(self.market_profiles), "version": self.version, "calibration_version": self.calibration_version, "market_count": len(self.market_profiles), } def build_package( self, data: Any, prediction: Any, v25_signal: Optional[Dict[str, Any]], quality: Dict[str, Any], ) -> Dict[str, Any]: decision_trace_id = uuid.uuid4().hex market_reliability = { market: round(float(profile.get("reliability", 0.5)), 4) for market, profile in self.market_profiles.items() } derived = self._derive_market_probabilities(data, prediction, v25_signal) market_rows = [ self._build_market_row( market=market, probs=payload["probs"], data=data, prediction=prediction, quality=quality, source=payload["source"], ) for market, payload in derived.items() ] market_rows = self._apply_scoreline_consistency_controls( market_rows, prediction, ) if self._is_top_league_match(data): market_rows = self._apply_top_league_portfolio_controls(market_rows) row_by_market = { str(row.get("market")): row for row in market_rows } surprise_hunter = self._build_surprise_hunter( data=data, prediction=prediction, quality=quality, derived=derived, row_by_market=row_by_market, ) surprise_pick = surprise_hunter.get("pick") if surprise_pick: surprise_pick = dict(surprise_pick) market_rows.sort( key=lambda row: ( 1 if row.get("playable") else 0, float(row.get("selection_score", 0.0)), float(row.get("play_score", 0.0)), ), reverse=True, ) main_pick = self._select_main_pick(market_rows) supporting_picks = [ row for row in market_rows if not self._same_pick(row, main_pick) and not self._same_pick(row, surprise_pick) ][:6] value_pick = self._select_value_pick(market_rows, main_pick) aggressive_pick = self._select_aggressive_pick(market_rows, main_pick) bet_summary = [self._to_bet_summary_item(row) for row in market_rows] market_board = { market: self._build_market_board_entry( payload["probs"], row_by_market.get(market, {}), ) for market, payload in derived.items() } reasoning_factors = self._build_reasoning_factors( data=data, prediction=prediction, quality=quality, main_pick=main_pick, ) playable_count = sum(1 for row in market_rows if row.get("playable")) shadow_summary = { "model_version": self.version, "calibration_version": self.calibration_version, "decision_trace_id": decision_trace_id, "main_pick": main_pick, "value_pick": value_pick, "surprise_pick": surprise_pick, "playable_count": playable_count, } return { "model_version": self.version, "calibration_version": self.calibration_version, "decision_trace_id": decision_trace_id, "match_info": { "match_id": data.match_id, "match_name": f"{data.home_team_name} vs {data.away_team_name}", "home_team": data.home_team_name, "away_team": data.away_team_name, "league": data.league_name, "match_date_ms": data.match_date_ms, "sport": data.sport, }, "data_quality": { **quality, "flags": self._dedupe( list(quality.get("flags", [])) + self._data_quality_v26_flags(data, quality) ), }, "risk": { "level": prediction.risk_level, "score": round(float(prediction.risk_score), 1), "is_surprise_risk": bool(prediction.is_surprise_risk), "surprise_type": prediction.surprise_type, "surprise_score": round( float(getattr(prediction, "surprise_score", 0.0) or 0.0), 1 ), "surprise_comment": str( getattr(prediction, "surprise_comment", "") or "" ), "surprise_reasons": list( getattr(prediction, "surprise_reasons", []) or [] ), "warnings": list(getattr(prediction, "risk_warnings", []) or []), }, "engine_breakdown": { "team": round(float(prediction.team_confidence), 1), "player": round(float(prediction.player_confidence), 1), "odds": round(float(prediction.odds_confidence), 1), "referee": round(float(prediction.referee_confidence), 1), }, "main_pick": main_pick, "value_pick": value_pick, "surprise_pick": surprise_pick, "bet_advice": { "playable": bool(main_pick and main_pick.get("playable")), "suggested_stake_units": float(main_pick.get("stake_units", 0.0)) if main_pick and main_pick.get("playable") else 0.0, "reason": "playable_pick_found" if main_pick and main_pick.get("playable") else "no_bet_conditions_met", }, "bet_summary": bet_summary, "supporting_picks": supporting_picks, "aggressive_pick": aggressive_pick, "surprise_hunter": surprise_hunter, "scenario_top5": list(prediction.ft_scores_top5 or []), "score_prediction": { "ft": prediction.predicted_ft_score, "ht": prediction.predicted_ht_score, "xg_home": round(float(prediction.home_xg), 2), "xg_away": round(float(prediction.away_xg), 2), "xg_total": round(float(prediction.total_xg), 2), }, "market_board": market_board, "reasoning_factors": reasoning_factors, "market_reliability": market_reliability, "shadow_engine_version": self.version, "shadow_engine": shadow_summary, } def _load_config(self) -> Dict[str, Any]: if not self.config_path.exists(): return dict(self.DEFAULT_CONFIG) try: loaded = json.loads(self.config_path.read_text(encoding="utf-8")) if not isinstance(loaded, dict): return dict(self.DEFAULT_CONFIG) merged = dict(self.DEFAULT_CONFIG) merged.update(loaded) return merged except Exception: return dict(self.DEFAULT_CONFIG) def _derive_market_probabilities( self, data: Any, prediction: Any, v25_signal: Optional[Dict[str, Any]], ) -> Dict[str, Dict[str, Any]]: score_table = self._poisson_score_table( float(prediction.home_xg), float(prediction.away_xg), int(self.goal_model.get("max_goals", 7)), ) ht_ratio = float(self.goal_model.get("halftime_ratio", 0.46)) ht_table = self._poisson_score_table( float(prediction.home_xg) * ht_ratio, float(prediction.away_xg) * ht_ratio, 5, ) v25_signal = v25_signal or {} ms_poisson = self._score_table_to_result_probs(score_table) ms_v25 = self._get_signal_probs(v25_signal, "MS", ("1", "X", "2")) ms_probs = self._blend_probs( ms_poisson, ms_v25, float(self.goal_model.get("ms_blend_weight", 0.60)), ) dc_probs = { "1X": ms_probs["1"] + ms_probs["X"], "X2": ms_probs["X"] + ms_probs["2"], "12": ms_probs["1"] + ms_probs["2"], } ou15_probs = self._score_table_to_total_probs(score_table, 1.5) ou25_probs = self._score_table_to_total_probs(score_table, 2.5) ou35_probs = self._score_table_to_total_probs(score_table, 3.5) btts_probs = self._score_table_to_btts_probs(score_table) oe_probs = self._score_table_to_odd_even_probs(score_table) ht_poisson = self._score_table_to_result_probs(ht_table) ht_v25 = self._get_signal_probs(v25_signal, "HT", ("1", "X", "2")) ht_probs = self._blend_probs( ht_poisson, ht_v25, float(self.goal_model.get("ht_blend_weight", 0.45)), ) ht_ou05 = self._score_table_to_total_probs(ht_table, 0.5) ht_ou15 = self._score_table_to_total_probs(ht_table, 1.5) htft_joint = { f"{ht_pick}/{ft_pick}": ht_probs[ht_pick] * ms_probs[ft_pick] for ht_pick in ("1", "X", "2") for ft_pick in ("1", "X", "2") } htft_v25 = self._get_signal_probs(v25_signal, "HTFT", self.MARKET_LABELS["HTFT"]) htft_probs = self._blend_probs( self._normalize_probs(htft_joint), htft_v25, float(self.goal_model.get("htft_v25_blend_weight", 0.35)), ) hcap_score = self._score_table_to_handicap_probs(score_table) hcap_v25 = self._get_signal_probs(v25_signal, "HCAP", ("1", "X", "2")) hcap_probs = self._blend_probs( hcap_score, hcap_v25, float(self.goal_model.get("hcap_v25_blend_weight", 0.30)), ) cards_v25 = self._get_signal_probs(v25_signal, "CARDS", ("Under", "Over")) cards_default = { "Under": float(cards_v25.get("Under", 0.52)), "Over": float(cards_v25.get("Over", 0.48)), } cards_probs = self._normalize_probs(cards_default) return { "MS": {"probs": ms_probs, "source": "hybrid_result_family"}, "DC": {"probs": self._normalize_probs(dc_probs), "source": "derived_from_ms"}, "OU15": {"probs": ou15_probs, "source": "goal_distribution"}, "OU25": {"probs": ou25_probs, "source": "goal_distribution"}, "OU35": {"probs": ou35_probs, "source": "goal_distribution"}, "BTTS": {"probs": btts_probs, "source": "goal_distribution"}, "HT": {"probs": ht_probs, "source": "hybrid_first_half"}, "HT_OU05": {"probs": ht_ou05, "source": "goal_distribution_ht"}, "HT_OU15": {"probs": ht_ou15, "source": "goal_distribution_ht"}, "HTFT": {"probs": htft_probs, "source": "derived_joint_htft"}, "OE": {"probs": oe_probs, "source": "goal_distribution"}, "CARDS": {"probs": cards_probs, "source": "v25_anchor"}, "HCAP": {"probs": hcap_probs, "source": "derived_handicap"}, } def _build_market_row( self, market: str, probs: Dict[str, float], data: Any, prediction: Any, quality: Dict[str, Any], source: str, ) -> Dict[str, Any]: probs = self._normalize_probs(probs) pick, probability = self._pick_from_probs(probs) odds = self._market_odds(data.odds_data or {}, market, pick) is_top_league = self._is_top_league_match(data) profile = self._market_profile_for_context( market=market, pick=pick, is_top_league=is_top_league, ) reliability = float(profile.get("reliability", 0.5)) confidence_multiplier = float(profile.get("confidence_multiplier", 0.9)) raw_confidence = probability * 100.0 calibrated_confidence = raw_confidence * confidence_multiplier implied_prob = (1.0 / odds) if odds > 1.0 else 0.0 edge = ((probability * odds) - 1.0) if odds > 1.0 else 0.0 quality_score = float(quality.get("score", 0.0)) * 100.0 risk_score = float(getattr(prediction, "risk_score", 50.0) or 50.0) play_score = ( calibrated_confidence * 0.58 + (edge * 100.0 * 2.8) + (quality_score * 0.14) + (reliability * 100.0 * 0.16) - (risk_score * 0.12) ) play_score = max(0.0, min(100.0, play_score)) reasons = [f"source:{source}"] weak_market = bool(profile.get("weak_market", market in self.WEAK_MARKETS)) if weak_market: reasons.append("weak_market_visible_pass_default") if is_top_league: reasons.append("top_league_policy_active") if quality_score < 60.0: reasons.append("limited_data_confidence") if getattr(data, "lineup_source", "none") == "probable_xi" and market in {"MS", "HT", "HTFT", "BTTS"}: reasons.append("lineup_probable_not_confirmed") if not getattr(data, "referee_name", None) and market == "CARDS": reasons.append("missing_referee") ms_gap = abs( float(getattr(prediction, "ms_home_prob", 0.0) or 0.0) - float(getattr(prediction, "ms_away_prob", 0.0) or 0.0) ) if market == "MS": play_score += 4.0 + min(9.0, ms_gap * 16.0) reasons.append("ms_priority_market") if is_top_league and market in {"OU15", "OU25", "OU35", "BTTS", "HT_OU05", "HT_OU15"}: play_score -= 4.0 reasons.append("top_league_goal_market_penalty") if market == "HTFT": htft_support = self._htft_pick_support( data=data, prediction=prediction, pick=pick, probs=probs, ) support_score = float(htft_support.get("score", 0.0)) play_score += min(16.0, support_score * 0.18) reasons.append( f"htft_pick_score_{round(support_score, 1):.1f}" ) reasons.extend(list(htft_support.get("reason_codes", []))) min_confidence = float(profile.get("min_confidence", 55.0)) min_edge = float(profile.get("min_edge", 0.02)) min_play_score = float(profile.get("min_play_score", 68.0)) min_odds = float(profile.get("min_odds", 0.0)) playable = True if odds <= 1.01: playable = False reasons.append("market_odds_missing") if bool(profile.get("disabled")): playable = False reasons.append( str(profile.get("disabled_reason") or "market_disabled_by_policy") ) if calibrated_confidence < min_confidence: playable = False reasons.append("below_calibrated_conf_threshold") if edge < min_edge: playable = False reasons.append(f"below_market_edge_threshold_{edge:+.3f}") if min_odds > 0.0 and odds < min_odds: playable = False reasons.append("below_market_odds_floor") if play_score < min_play_score: playable = False reasons.append("insufficient_play_score") if weak_market and (quality_score < 72.0 or calibrated_confidence < (min_confidence + 4.0)): playable = False reasons.append("weak_market_pass_default") if market == "CARDS" and (not getattr(data, "referee_name", None) or quality_score < 70.0): playable = False reasons.append("cards_market_needs_referee_and_quality") if market == "HTFT": htft_support = self._htft_pick_support( data=data, prediction=prediction, pick=pick, probs=probs, ) support_score = float(htft_support.get("score", 0.0)) profile_type = str(htft_support.get("profile_type", "generic")) min_support = 66.0 if profile_type == "strict_reversal" else 56.0 if support_score < min_support: playable = False reasons.append("htft_pick_support_too_low") if profile_type == "strict_reversal" and float(htft_support.get("reversal_prob", 0.0)) < 0.055: playable = False reasons.append("htft_reversal_prob_too_low") if profile_type == "draw_swing" and float(htft_support.get("swing_prob", 0.0)) < 0.08: playable = False reasons.append("htft_swing_prob_too_low") if playable: if edge >= 0.10: grade = "A" elif edge >= 0.06: grade = "B" else: grade = "C" stake_units = self._kelly_stake(probability, odds) reasons.append("market_passed_all_gates") else: grade = "PASS" stake_units = 0.0 selection_score = self._selection_score( market=market, edge=edge, calibrated_confidence=calibrated_confidence, quality_score=float(quality.get("score", 0.0)), reliability=reliability, ) return { "market": market, "market_type": market, "strategy_channel": "standard", "pick": self._display_pick(market, pick), "raw_pick": pick, "probability": round(probability, 4), "confidence": round(raw_confidence, 1), "odds": round(odds, 2), "raw_confidence": round(raw_confidence, 1), "calibrated_confidence": round(calibrated_confidence, 1), "min_required_confidence": round(min_confidence, 1), "min_required_play_score": round(min_play_score, 1), "min_required_edge": round(min_edge, 4), "edge": round(edge, 4), "ev_edge": round(edge, 4), "implied_prob": round(implied_prob, 4), "play_score": round(play_score, 1), "playable": playable, "bet_grade": grade, "stake_units": round(stake_units, 1) if playable else 0.0, "decision_reasons": self._dedupe(reasons)[:6], "selection_score": round(selection_score, 4), "market_reliability": round(reliability, 4), } def _is_top_league_match(self, data: Any) -> bool: league_id = str(getattr(data, "league_id", "") or "").strip() return bool(league_id and league_id in self.top_league_ids) def _market_profile_for_context( self, market: str, pick: str, is_top_league: bool, ) -> Dict[str, Any]: profile = dict(self.market_profiles.get(market, {})) if not is_top_league: return profile profile.update(self.top_league_market_overrides.get(market, {})) profile.update(self.top_league_pick_overrides.get(f"{market}:{pick}", {})) return profile def _apply_top_league_portfolio_controls( self, rows: List[Dict[str, Any]], ) -> List[Dict[str, Any]]: controlled = [dict(row) for row in rows] goal_cluster = {"OU15", "OU25", "OU35", "BTTS"} early_cluster = {"HT_OU05", "HT_OU15"} ms_row = next( ( row for row in controlled if row.get("market") == "MS" and row.get("playable") ), None, ) ms_score = float(ms_row.get("selection_score", 0.0)) if ms_row else 0.0 def suppress(row: Dict[str, Any], reason: str) -> None: row["playable"] = False row["bet_grade"] = "PASS" row["stake_units"] = 0.0 reasons = list(row.get("decision_reasons", [])) reasons.append(reason) row["decision_reasons"] = self._dedupe(reasons)[:6] for row in controlled: if not row.get("playable"): continue market = str(row.get("market") or "") if market in early_cluster: suppress(row, "top_league_early_market_suppressed") playable_goals = [ row for row in controlled if row.get("playable") and str(row.get("market") or "") in goal_cluster ] playable_goals.sort( key=lambda row: ( float(row.get("selection_score", 0.0)), float(row.get("edge", 0.0)), float(row.get("calibrated_confidence", 0.0)), ), reverse=True, ) keeper: Optional[Dict[str, Any]] = None for row in playable_goals: if keeper is None: keeper = row continue suppress(row, "top_league_goal_cluster_trimmed") if keeper and ms_row: keeper_score = float(keeper.get("selection_score", 0.0)) keeper_edge = float(keeper.get("edge", 0.0)) keeper_odds = float(keeper.get("odds", 0.0)) if keeper_score < (ms_score + 8.0) or keeper_edge < 0.12 or keeper_odds < 1.72: suppress(keeper, "top_league_ms_priority_suppressed_goal_side") return controlled def _apply_scoreline_consistency_controls( self, rows: List[Dict[str, Any]], prediction: Any, ) -> List[Dict[str, Any]]: ft_score = self._parse_scoreline(getattr(prediction, "predicted_ft_score", "")) ht_score = self._parse_scoreline(getattr(prediction, "predicted_ht_score", "")) if not ft_score: return rows expected = self._expected_picks_from_scoreline(ft_score, ht_score) controlled = [dict(row) for row in rows] for row in controlled: market = str(row.get("market") or "") raw_pick = str(row.get("raw_pick") or "") allowed = expected.get(market) reasons = list(row.get("decision_reasons", [])) if allowed: if raw_pick in allowed: reasons.append("scoreline_scenario_aligned") else: if row.get("playable"): row["playable"] = False row["bet_grade"] = "PASS" row["stake_units"] = 0.0 reasons.append("scoreline_scenario_conflict") row["decision_reasons"] = self._dedupe(reasons)[:6] return controlled @staticmethod def _parse_scoreline(score: Any) -> Optional[Tuple[int, int]]: text = str(score or "").strip() if not text or "-" not in text: return None left, right = text.split("-", 1) try: return int(left.strip()), int(right.strip()) except ValueError: return None def _expected_picks_from_scoreline( self, ft_score: Tuple[int, int], ht_score: Optional[Tuple[int, int]], ) -> Dict[str, set[str]]: ft_home, ft_away = ft_score total_goals = ft_home + ft_away ft_result = self._result_pick(ft_home, ft_away) expected: Dict[str, set[str]] = { "MS": {ft_result}, "OU15": {"Over" if total_goals > 1 else "Under"}, "OU25": {"Over" if total_goals > 2 else "Under"}, "OU35": {"Over" if total_goals > 3 else "Under"}, "BTTS": {"Yes" if ft_home > 0 and ft_away > 0 else "No"}, "OE": {"Odd" if total_goals % 2 == 1 else "Even"}, "HCAP": {self._handicap_pick(ft_home, ft_away)}, } if ft_result == "1": expected["DC"] = {"1X", "12"} elif ft_result == "2": expected["DC"] = {"X2", "12"} else: expected["DC"] = {"1X", "X2"} if ht_score: ht_home, ht_away = ht_score ht_goals = ht_home + ht_away ht_result = self._result_pick(ht_home, ht_away) expected["HT"] = {ht_result} expected["HT_OU05"] = {"Over" if ht_goals > 0 else "Under"} expected["HT_OU15"] = {"Over" if ht_goals > 1 else "Under"} expected["HTFT"] = {f"{ht_result}/{ft_result}"} return expected @staticmethod def _result_pick(home_goals: int, away_goals: int) -> str: if home_goals > away_goals: return "1" if home_goals < away_goals: return "2" return "X" @staticmethod def _handicap_pick(home_goals: int, away_goals: int) -> str: diff = home_goals - away_goals if diff >= 2: return "1" if diff == 1: return "X" return "2" def _build_market_board_entry( self, probs: Dict[str, float], row: Dict[str, Any], ) -> Dict[str, Any]: entry = { "pick": row.get("pick"), "confidence": row.get("calibrated_confidence"), "probs": { self._board_key(key): round(float(value), 4) for key, value in probs.items() }, } if row.get("market") == "CARDS": entry["line"] = 4.5 if row.get("market") == "HCAP": entry["line_home"] = -1.0 return entry def _data_quality_v26_flags(self, data: Any, quality: Dict[str, Any]) -> List[str]: flags: List[str] = [] if float(quality.get("score", 0.0)) < 0.60: flags.append("shadow_low_quality_window") if getattr(data, "lineup_source", "none") == "none": flags.append("shadow_missing_confirmed_lineups") if not getattr(data, "referee_name", None): flags.append("shadow_missing_referee") return flags def _build_reasoning_factors( self, data: Any, prediction: Any, quality: Dict[str, Any], main_pick: Optional[Dict[str, Any]], ) -> List[str]: factors = [ "shadow_engine_roi_first", "shadow_goal_distribution_consistent", "shadow_weak_markets_pass_default", "shadow_ms_priority_routing", ] if float(quality.get("score", 0.0)) >= 0.8: factors.append("shadow_high_data_quality") if getattr(data, "lineup_source", "none") == "confirmed_live": factors.append("shadow_confirmed_lineups") if main_pick and main_pick.get("playable"): factors.append("shadow_playable_edge_found") if getattr(prediction, "is_surprise_risk", False): factors.append("upset_risk_detected") factors.append("shadow_surprise_sidecar") if main_pick and str(main_pick.get("market")) == "HTFT": factors.append("shadow_reversal_pick_selected") return self._dedupe(factors) def _selection_score( self, market: str, edge: float, calibrated_confidence: float, quality_score: float, reliability: float, ) -> float: weights = self.selection_weights or {} score = ( (edge * 100.0 * float(weights.get("edge", 0.44))) + (calibrated_confidence * float(weights.get("confidence", 0.32))) + (quality_score * 100.0 * float(weights.get("quality", 0.14))) + (reliability * 100.0 * float(weights.get("reliability", 0.10))) ) if market in self.core_markets: score += 6.0 if market == "MS": score += 8.0 if market == "HTFT": score += 3.0 if market in self.WEAK_MARKETS: score -= 4.0 return score def _select_main_pick( self, rows: List[Dict[str, Any]], ) -> Optional[Dict[str, Any]]: playable = [row for row in rows if row.get("playable")] if not playable: return rows[0] if rows else None ms_playable = [ row for row in playable if row.get("market") == "MS" and float(row.get("calibrated_confidence", 0.0)) >= 58.0 and float(row.get("edge", 0.0)) >= 0.02 ] ms_playable.sort( key=lambda row: ( float(row.get("selection_score", 0.0)), float(row.get("play_score", 0.0)), float(row.get("edge", 0.0)), ), reverse=True, ) best_ms = ms_playable[0] if ms_playable else None if best_ms: main_pick = dict(best_ms) main_pick["is_guaranteed"] = True main_pick["pick_reason"] = "ms_priority_market" return main_pick core_playable = [row for row in playable if row.get("market") in self.core_markets] ranked = core_playable or playable ranked = sorted( ranked, key=lambda row: ( float(row.get("selection_score", 0.0)), float(row.get("play_score", 0.0)), float(row.get("edge", 0.0)), ), reverse=True, ) main_pick = dict(ranked[0]) main_pick["is_guaranteed"] = bool(main_pick.get("market") in self.core_markets) main_pick["pick_reason"] = ( "roi_core_market" if main_pick.get("market") in self.core_markets else "roi_best_available" ) return main_pick def _select_value_pick( self, rows: List[Dict[str, Any]], main_pick: Optional[Dict[str, Any]], ) -> Optional[Dict[str, Any]]: candidates = [ row for row in rows if row.get("playable") and float(row.get("odds", 0.0)) >= 1.6 and not self._same_pick(row, main_pick) ] if not candidates: return None candidates.sort( key=lambda row: ( float(row.get("edge", 0.0)) * float(row.get("odds", 1.0)), float(row.get("selection_score", 0.0)), ), reverse=True, ) return dict(candidates[0]) def _select_aggressive_pick( self, rows: List[Dict[str, Any]], main_pick: Optional[Dict[str, Any]], ) -> Optional[Dict[str, Any]]: aggressive = [ row for row in rows if row.get("market") == "HTFT" and float(row.get("probability", 0.0)) >= 0.07 and float(row.get("odds", 0.0)) > 1.5 and not self._same_pick(row, main_pick) ] if not aggressive: return None aggressive.sort(key=lambda row: float(row.get("probability", 0.0)), reverse=True) return { "market": "HT/FT", "pick": aggressive[0].get("pick"), "probability": aggressive[0].get("probability"), "confidence": aggressive[0].get("calibrated_confidence"), "odds": aggressive[0].get("odds"), } def _build_surprise_hunter( self, data: Any, prediction: Any, quality: Dict[str, Any], derived: Dict[str, Dict[str, Any]], row_by_market: Dict[str, Dict[str, Any]], ) -> Dict[str, Any]: htft_probs = (derived.get("HTFT") or {}).get("probs", {}) or {} ms_probs = (derived.get("MS") or {}).get("probs", {}) or {} context = self._htft_reversal_context(data, prediction, htft_probs) if not context: return { "score": 0.0, "playable": False, "pick": None, "reason_codes": ["surprise_context_unavailable"], } favorite_side = str(context.get("favorite_side") or "") underdog_side = str(context.get("underdog_side") or "") reversal_key = str(context.get("reversal_key") or "") draw_swing_key = str(context.get("draw_swing_key") or "") reversal_prob = float(context.get("reversal_prob", 0.0)) draw_swing_prob = float(context.get("draw_swing_prob", 0.0)) favorite_gap = float(context.get("favorite_gap", 0.0)) favorite_odd = float(context.get("favorite_odd", 0.0)) quality_score = float(quality.get("score", 0.0)) * 100.0 base_surprise = float(getattr(prediction, "surprise_score", 0.0) or 0.0) draw_prob = float(ms_probs.get("X", 0.0)) underdog_ms_prob = float(ms_probs.get(underdog_side, 0.0)) lineups_confirmed = str(getattr(data, "lineup_source", "none")) == "confirmed_live" pattern_support = self._surprise_pattern_support( data=data, prediction=prediction, context=context, ms_probs=ms_probs, ) support_score = float(pattern_support.get("score", 0.0)) reversal_score = min( 100.0, (base_surprise * 0.50) + (reversal_prob * 100.0 * 0.72) + (draw_swing_prob * 100.0 * 0.24) + (max(0.0, favorite_gap - 0.45) * 18.0) + (underdog_ms_prob * 100.0 * 0.12) + (draw_prob * 100.0 * 0.08), ) reversal_score = min(100.0, reversal_score + (support_score * 0.22)) reason_codes: List[str] = [] if favorite_gap >= 0.60: reason_codes.append("favorite_gap_large") if 1.01 < favorite_odd <= 2.35: reason_codes.append("favorite_price_supported") if reversal_prob >= 0.09: reason_codes.append("reversal_prob_hot") elif reversal_prob >= 0.065: reason_codes.append("reversal_prob_warm") if draw_swing_prob >= 0.11: reason_codes.append("draw_swing_support") if getattr(prediction, "is_surprise_risk", False): reason_codes.append("upset_risk_detected") if lineups_confirmed: reason_codes.append("confirmed_lineups") if quality_score >= 78.0: reason_codes.append("quality_supports_reversal") reason_codes.extend(list(pattern_support.get("reason_codes", []))) htft_profile = self.market_profiles.get("HTFT", {}) odds = self._market_odds(data.odds_data or {}, "HTFT", reversal_key) confidence_multiplier = float(htft_profile.get("confidence_multiplier", 0.72)) calibrated_confidence = reversal_prob * 100.0 * confidence_multiplier edge = ((reversal_prob * odds) - 1.0) if odds > 1.0 else 0.0 ms_row = row_by_market.get("MS") or {} selection_score = ( calibrated_confidence * 0.34 + (edge * 100.0 * 2.4) + (reversal_score * 0.26) + (quality_score * 0.08) + (support_score * 0.12) + (float(ms_row.get("selection_score", 0.0)) * 0.06) ) selection_score = max(0.0, min(100.0, selection_score)) candidate_eligible = ( odds > 1.01 and reversal_prob >= 0.045 and base_surprise >= 48.0 and favorite_gap >= 0.45 and quality_score >= 68.0 and support_score >= 42.0 and (favorite_odd == 0.0 or favorite_odd <= 2.35) ) if not lineups_confirmed: reason_codes.append("lineup_not_confirmed_for_surprise") if support_score < 50.0: candidate_eligible = False reason_codes.append("surprise_needs_more_support_without_lineups") if not candidate_eligible: reason_codes.append("surprise_candidate_filtered") return { "strategy_channel": "surprise_sidecar", "score": round(reversal_score, 1), "playable": False, "favorite_side": favorite_side, "underdog_side": underdog_side, "favorite_gap": round(favorite_gap, 3), "favorite_odd": round(favorite_odd, 2), "reversal_pick": reversal_key, "reversal_prob": round(reversal_prob, 4), "draw_swing_pick": draw_swing_key, "draw_swing_prob": round(draw_swing_prob, 4), "support_score": round(support_score, 1), "pattern_break_score": round(float(pattern_support.get("pattern_break_score", 0.0)), 3), "history_support_score": round(float(pattern_support.get("history_support_score", 0.0)), 3), "odds_band_score": round(float(pattern_support.get("odds_band_score", 0.0)), 3), "league_reversal_rate": round(float(pattern_support.get("league_reversal_rate", 0.0)), 4), "reason_codes": self._dedupe(reason_codes)[:6], "pick": None, } playable = ( reversal_prob >= 0.07 and edge >= 0.10 and reversal_score >= 68.0 and quality_score >= 80.0 and support_score >= 45.0 and favorite_gap >= 0.60 and underdog_ms_prob >= 0.14 ) if not lineups_confirmed and playable and support_score < 60.0: playable = False reason_codes.append("surprise_play_needs_more_support_without_lineups") if not playable: reason_codes.append("surprise_pick_not_playable") else: reason_codes.append("surprise_pick_passed") pick = { "market": "HTFT", "market_type": "HTFT", "strategy_channel": "surprise_sidecar", "pick": self._display_pick("HTFT", reversal_key), "raw_pick": reversal_key, "probability": round(reversal_prob, 4), "confidence": round(reversal_prob * 100.0, 1), "raw_confidence": round(reversal_prob * 100.0, 1), "calibrated_confidence": round(calibrated_confidence, 1), "odds": round(odds, 2), "edge": round(edge, 4), "ev_edge": round(edge, 4), "playable": playable, "bet_grade": "A" if playable and edge >= 0.14 else "B" if playable else "PASS", "stake_units": round(self._kelly_stake(reversal_prob, odds), 1) if playable else 0.0, "play_score": round(selection_score, 1), "selection_score": round(selection_score, 4), "market_reliability": round(float(htft_profile.get("reliability", 0.34)), 4), "decision_reasons": self._dedupe(reason_codes)[:6], "surprise_score": round(reversal_score, 1), "support_score": round(support_score, 1), "favorite_side": favorite_side, "underdog_side": underdog_side, "pick_reason": "favorite_reversal_signal", } return { "strategy_channel": "surprise_sidecar", "score": round(reversal_score, 1), "playable": playable, "favorite_side": favorite_side, "underdog_side": underdog_side, "favorite_gap": round(favorite_gap, 3), "favorite_odd": round(float(context.get("favorite_odd", 0.0)), 2), "reversal_pick": reversal_key, "reversal_prob": round(reversal_prob, 4), "draw_swing_pick": draw_swing_key, "draw_swing_prob": round(draw_swing_prob, 4), "support_score": round(support_score, 1), "pattern_break_score": round(float(pattern_support.get("pattern_break_score", 0.0)), 3), "history_support_score": round(float(pattern_support.get("history_support_score", 0.0)), 3), "odds_band_score": round(float(pattern_support.get("odds_band_score", 0.0)), 3), "league_reversal_rate": round(float(pattern_support.get("league_reversal_rate", 0.0)), 4), "reason_codes": self._dedupe(reason_codes)[:6], "pick": pick, } def _surprise_pattern_support( self, data: Any, prediction: Any, context: Dict[str, Any], ms_probs: Dict[str, float], ) -> Dict[str, Any]: favorite_side = str(context.get("favorite_side") or "") underdog_side = str(context.get("underdog_side") or "") favorite_odd = float(context.get("favorite_odd", 0.0)) base_surprise = float(getattr(prediction, "surprise_score", 0.0) or 0.0) odds_band_label = self._favorite_odds_band(favorite_odd) band_prior = self._load_odds_band_priors().get(odds_band_label, {}) referee_prior = self._load_referee_priors().get( str(getattr(data, "referee_name", "") or "").strip(), {}, ) league_prior = self._load_league_priors().get( str(getattr(data, "league_id", "") or ""), {}, ) odds_band_score = min(1.0, float(band_prior.get("strict_rev_rate", 0.0)) * 28.0) referee_score = min(1.0, float(referee_prior.get("strict_rev_rate", 0.0)) * 12.0) league_score = min(1.0, float(league_prior.get("strict_rev_rate", 0.0)) * 20.0) favorite_team_id = ( getattr(data, "home_team_id", "") if favorite_side == "1" else getattr(data, "away_team_id", "") ) underdog_team_id = ( getattr(data, "away_team_id", "") if favorite_side == "1" else getattr(data, "home_team_id", "") ) favorite_form = self._load_recent_team_pattern( str(favorite_team_id or ""), int(getattr(data, "match_date_ms", 0) or 0), ) underdog_form = self._load_recent_team_pattern( str(underdog_team_id or ""), int(getattr(data, "match_date_ms", 0) or 0), ) tendency = self._load_htft_tendency_support(data) underdog_comeback_rate = ( float(tendency.get("htft_home_comeback_rate", 0.0)) if underdog_side == "1" else float(tendency.get("htft_away_comeback_rate", 0.0)) ) underdog_second_half_surge = ( float(tendency.get("htft_home_second_half_surge", 1.0)) if underdog_side == "1" else float(tendency.get("htft_away_second_half_surge", 1.0)) ) favorite_ht_win_rate = ( float(tendency.get("htft_home_ht_win_rate", 0.33)) if favorite_side == "1" else float(tendency.get("htft_away_ht_win_rate", 0.33)) ) league_reversal_rate = float(tendency.get("htft_league_reversal_rate", 0.05)) draw_pressure = min(1.0, float(ms_probs.get("X", 0.0)) * 1.8) pattern_break_score = min( 1.0, (float(favorite_form.get("big_win_streak", 0.0)) * 0.22) + (max(0.0, float(favorite_form.get("winning_streak", 0.0)) - 2.0) * 0.11) + (float(favorite_form.get("clean_sheet_streak", 0.0)) * 0.08) + (float(underdog_form.get("scoring_streak", 0.0)) * 0.13) + (float(underdog_form.get("unbeaten_streak", 0.0)) * 0.08), ) history_support_score = min( 1.0, (league_reversal_rate * 3.0) + (league_score * 0.45) + (referee_score * 0.30) + (underdog_comeback_rate * 0.9) + (max(0.0, underdog_second_half_surge - 1.0) * 0.22) + (favorite_ht_win_rate * 0.32), ) score = min( 100.0, (base_surprise * 0.24) + (odds_band_score * 24.0) + (pattern_break_score * 24.0) + (history_support_score * 20.0) + (referee_score * 10.0) + (league_score * 8.0) + (draw_pressure * 12.0), ) reason_codes: List[str] = [] if odds_band_score >= 0.45: reason_codes.append("favorite_odds_band_reversal_window") if pattern_break_score >= 0.45: reason_codes.append("favorite_streak_break_window") if underdog_comeback_rate >= 0.18: reason_codes.append("underdog_comeback_profile") if underdog_second_half_surge >= 1.35: reason_codes.append("underdog_second_half_surge") if league_reversal_rate >= 0.11: reason_codes.append("league_reversal_hot") if float(league_prior.get("strict_rev_rate", 0.0)) >= 0.03: reason_codes.append("league_strict_reversal_prior") if float(referee_prior.get("strict_rev_rate", 0.0)) >= 0.07: reason_codes.append("referee_reversal_prior") if draw_pressure >= 0.42: reason_codes.append("draw_pressure_supports_swing") return { "score": round(score, 1), "odds_band_score": odds_band_score, "odds_band_label": odds_band_label, "odds_band_rev_rate": float(band_prior.get("strict_rev_rate", 0.0)), "pattern_break_score": pattern_break_score, "history_support_score": history_support_score, "league_reversal_rate": league_reversal_rate, "league_strict_rev_rate": float(league_prior.get("strict_rev_rate", 0.0)), "referee_strict_rev_rate": float(referee_prior.get("strict_rev_rate", 0.0)), "underdog_comeback_rate": underdog_comeback_rate, "underdog_second_half_surge": underdog_second_half_surge, "favorite_ht_win_rate": favorite_ht_win_rate, "reason_codes": reason_codes, } def _load_htft_tendency_support(self, data: Any) -> Dict[str, float]: home_team_id = str(getattr(data, "home_team_id", "") or "") away_team_id = str(getattr(data, "away_team_id", "") or "") match_date_ms = int(getattr(data, "match_date_ms", 0) or 0) if not home_team_id or not away_team_id or match_date_ms <= 0: return {} try: from features.htft_tendency_engine import get_htft_tendency_engine return get_htft_tendency_engine().get_features( home_team_id=home_team_id, away_team_id=away_team_id, league_id=str(getattr(data, "league_id", "") or "") or None, before_date=match_date_ms, ) except Exception: return {} def _load_recent_team_pattern( self, team_id: str, before_date_ms: int, limit: int = 5, ) -> Dict[str, float]: cache_key = (team_id, before_date_ms) if cache_key in self._team_pattern_cache: return self._team_pattern_cache[cache_key] fallback = { "winning_streak": 0.0, "unbeaten_streak": 0.0, "big_win_streak": 0.0, "clean_sheet_streak": 0.0, "scoring_streak": 0.0, } if not team_id or before_date_ms <= 0: self._team_pattern_cache[cache_key] = fallback return fallback try: import psycopg2 from psycopg2.extras import RealDictCursor from data.db import get_clean_dsn with psycopg2.connect(get_clean_dsn()) as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( """ SELECT home_team_id, away_team_id, score_home, score_away FROM matches WHERE (home_team_id = %s OR away_team_id = %s) AND status = 'FT' AND sport = 'football' AND score_home IS NOT NULL AND score_away IS NOT NULL AND mst_utc < %s ORDER BY mst_utc DESC LIMIT %s """, (team_id, team_id, before_date_ms, limit), ) rows = cur.fetchall() except Exception: self._team_pattern_cache[cache_key] = fallback return fallback winning_streak = 0 unbeaten_streak = 0 big_win_streak = 0 clean_sheet_streak = 0 scoring_streak = 0 win_broken = False unbeaten_broken = False big_win_broken = False clean_sheet_broken = False scoring_broken = False for row in rows: is_home = str(row.get("home_team_id")) == team_id goals_for = int((row.get("score_home") if is_home else row.get("score_away")) or 0) goals_against = int((row.get("score_away") if is_home else row.get("score_home")) or 0) won = goals_for > goals_against unbeaten = goals_for >= goals_against big_win = won and (goals_for - goals_against) >= 2 clean_sheet = goals_against == 0 scored = goals_for > 0 if not win_broken: if won: winning_streak += 1 else: win_broken = True if not unbeaten_broken: if unbeaten: unbeaten_streak += 1 else: unbeaten_broken = True if not big_win_broken: if big_win: big_win_streak += 1 else: big_win_broken = True if not clean_sheet_broken: if clean_sheet: clean_sheet_streak += 1 else: clean_sheet_broken = True if not scoring_broken: if scored: scoring_streak += 1 else: scoring_broken = True payload = { "winning_streak": float(winning_streak), "unbeaten_streak": float(unbeaten_streak), "big_win_streak": float(big_win_streak), "clean_sheet_streak": float(clean_sheet_streak), "scoring_streak": float(scoring_streak), } self._team_pattern_cache[cache_key] = payload return payload @staticmethod def _favorite_odds_band(favorite_odd: float) -> str: if favorite_odd <= 0.0: return "unknown" if favorite_odd < 1.30: return "<1.30" if favorite_odd < 1.50: return "1.30-1.49" if favorite_odd < 1.70: return "1.50-1.69" if favorite_odd < 1.90: return "1.70-1.89" if favorite_odd < 2.10: return "1.90-2.09" if favorite_odd < 2.40: return "2.10-2.39" return "2.40+" def _load_odds_band_priors(self) -> Dict[str, Dict[str, float]]: if self._odds_band_prior_cache is not None: return self._odds_band_prior_cache fallback = { "<1.30": {"strict_rev_rate": 0.009, "draw_loss_rate": 0.041}, "1.30-1.49": {"strict_rev_rate": 0.014, "draw_loss_rate": 0.066}, "1.50-1.69": {"strict_rev_rate": 0.017, "draw_loss_rate": 0.085}, "1.70-1.89": {"strict_rev_rate": 0.021, "draw_loss_rate": 0.099}, "1.90-2.09": {"strict_rev_rate": 0.023, "draw_loss_rate": 0.114}, "2.10-2.39": {"strict_rev_rate": 0.022, "draw_loss_rate": 0.126}, "2.40+": {"strict_rev_rate": 0.023, "draw_loss_rate": 0.148}, "unknown": {"strict_rev_rate": 0.020, "draw_loss_rate": 0.100}, } try: import psycopg2 from psycopg2.extras import RealDictCursor from data.db import get_clean_dsn with psycopg2.connect(get_clean_dsn()) as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( """ WITH ms AS ( SELECT oc.match_id, MAX(CASE WHEN os.name = '1' THEN NULLIF(os.odd_value, '')::numeric END) AS ms_h, MAX(CASE WHEN os.name = '2' THEN NULLIF(os.odd_value, '')::numeric END) AS ms_a FROM odd_categories oc JOIN odd_selections os ON os.odd_category_db_id = oc.db_id WHERE oc.name = 'Maç Sonucu' GROUP BY oc.match_id ), base AS ( SELECT CASE WHEN LEAST(ms.ms_h, ms.ms_a) < 1.30 THEN '<1.30' WHEN LEAST(ms.ms_h, ms.ms_a) < 1.50 THEN '1.30-1.49' WHEN LEAST(ms.ms_h, ms.ms_a) < 1.70 THEN '1.50-1.69' WHEN LEAST(ms.ms_h, ms.ms_a) < 1.90 THEN '1.70-1.89' WHEN LEAST(ms.ms_h, ms.ms_a) < 2.10 THEN '1.90-2.09' WHEN LEAST(ms.ms_h, ms.ms_a) < 2.40 THEN '2.10-2.39' ELSE '2.40+' END AS odds_band, CASE WHEN ms.ms_h < ms.ms_a THEN '1' WHEN ms.ms_a < ms.ms_h THEN '2' ELSE NULL END AS favorite_side, CASE WHEN m.ht_score_home > m.ht_score_away THEN '1' WHEN m.ht_score_home < m.ht_score_away THEN '2' ELSE 'X' END || '/' || CASE WHEN m.score_home > m.score_away THEN '1' WHEN m.score_home < m.score_away THEN '2' ELSE 'X' END AS htft FROM matches m JOIN ms ON ms.match_id = m.id WHERE m.sport = 'football' AND m.status = 'FT' AND m.ht_score_home IS NOT NULL AND m.ht_score_away IS NOT NULL AND ms.ms_h > 1.0 AND ms.ms_a > 1.0 AND ms.ms_h <> ms.ms_a ) SELECT odds_band, AVG(CASE WHEN (favorite_side = '1' AND htft = '1/2') OR (favorite_side = '2' AND htft = '2/1') THEN 1.0 ELSE 0.0 END) AS strict_rev_rate, AVG(CASE WHEN (favorite_side = '1' AND htft = 'X/2') OR (favorite_side = '2' AND htft = 'X/1') THEN 1.0 ELSE 0.0 END) AS draw_loss_rate FROM base GROUP BY odds_band """ ) rows = cur.fetchall() cache = dict(fallback) for row in rows: cache[str(row["odds_band"])] = { "strict_rev_rate": float(row["strict_rev_rate"] or 0.0), "draw_loss_rate": float(row["draw_loss_rate"] or 0.0), } self._odds_band_prior_cache = cache return cache except Exception: self._odds_band_prior_cache = fallback return fallback def _load_referee_priors(self) -> Dict[str, Dict[str, float]]: if self._referee_prior_cache is not None: return self._referee_prior_cache fallback: Dict[str, Dict[str, float]] = {} try: import psycopg2 from psycopg2.extras import RealDictCursor from data.db import get_clean_dsn with psycopg2.connect(get_clean_dsn()) as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( """ WITH refs AS ( SELECT mo.name AS referee, CASE WHEN m.ht_score_home > m.ht_score_away THEN '1' WHEN m.ht_score_home < m.ht_score_away THEN '2' ELSE 'X' END || '/' || CASE WHEN m.score_home > m.score_away THEN '1' WHEN m.score_home < m.score_away THEN '2' ELSE 'X' END AS htft FROM matches m JOIN match_officials mo ON mo.match_id = m.id AND mo.role_id = 1 WHERE m.sport = 'football' AND m.status = 'FT' AND m.ht_score_home IS NOT NULL AND m.ht_score_away IS NOT NULL AND mo.name IS NOT NULL ) SELECT referee, COUNT(*) AS matches, AVG(CASE WHEN htft IN ('1/2', '2/1') THEN 1.0 ELSE 0.0 END) AS strict_rev_rate, AVG(CASE WHEN htft IN ('X/1', 'X/2') THEN 1.0 ELSE 0.0 END) AS draw_swing_rate FROM refs GROUP BY referee HAVING COUNT(*) >= 25 """ ) rows = cur.fetchall() cache: Dict[str, Dict[str, float]] = {} for row in rows: cache[str(row["referee"])] = { "matches": float(row["matches"] or 0.0), "strict_rev_rate": float(row["strict_rev_rate"] or 0.0), "draw_swing_rate": float(row["draw_swing_rate"] or 0.0), } self._referee_prior_cache = cache return cache except Exception: self._referee_prior_cache = fallback return fallback def _load_league_priors(self) -> Dict[str, Dict[str, float]]: if self._league_prior_cache is not None: return self._league_prior_cache fallback: Dict[str, Dict[str, float]] = {} try: import psycopg2 from psycopg2.extras import RealDictCursor from data.db import get_clean_dsn with psycopg2.connect(get_clean_dsn()) as conn: with conn.cursor(cursor_factory=RealDictCursor) as cur: cur.execute( """ WITH ms AS ( SELECT oc.match_id, MAX(CASE WHEN os.name = '1' THEN NULLIF(os.odd_value, '')::numeric END) AS ms_h, MAX(CASE WHEN os.name = '2' THEN NULLIF(os.odd_value, '')::numeric END) AS ms_a FROM odd_categories oc JOIN odd_selections os ON os.odd_category_db_id = oc.db_id WHERE oc.name = 'Maç Sonucu' GROUP BY oc.match_id ), base AS ( SELECT m.league_id, CASE WHEN ms.ms_h < ms.ms_a THEN '1' WHEN ms.ms_a < ms.ms_h THEN '2' ELSE NULL END AS favorite_side, CASE WHEN m.ht_score_home > m.ht_score_away THEN '1' WHEN m.ht_score_home < m.ht_score_away THEN '2' ELSE 'X' END || '/' || CASE WHEN m.score_home > m.score_away THEN '1' WHEN m.score_home < m.score_away THEN '2' ELSE 'X' END AS htft FROM matches m JOIN ms ON ms.match_id = m.id WHERE m.sport = 'football' AND m.status = 'FT' AND m.ht_score_home IS NOT NULL AND m.ht_score_away IS NOT NULL AND m.league_id IS NOT NULL AND ms.ms_h > 1.0 AND ms.ms_a > 1.0 AND ms.ms_h <> ms.ms_a ) SELECT league_id, COUNT(*) AS matches, AVG(CASE WHEN (favorite_side = '1' AND htft = '1/2') OR (favorite_side = '2' AND htft = '2/1') THEN 1.0 ELSE 0.0 END) AS strict_rev_rate, AVG(CASE WHEN (favorite_side = '1' AND htft = 'X/2') OR (favorite_side = '2' AND htft = 'X/1') THEN 1.0 ELSE 0.0 END) AS draw_loss_rate FROM base GROUP BY league_id HAVING COUNT(*) >= 40 """ ) rows = cur.fetchall() cache: Dict[str, Dict[str, float]] = {} for row in rows: cache[str(row["league_id"])] = { "matches": float(row["matches"] or 0.0), "strict_rev_rate": float(row["strict_rev_rate"] or 0.0), "draw_loss_rate": float(row["draw_loss_rate"] or 0.0), } self._league_prior_cache = cache return cache except Exception: self._league_prior_cache = fallback return fallback def _htft_pick_support( self, data: Any, prediction: Any, pick: str, probs: Dict[str, float], ) -> Dict[str, Any]: context = self._htft_reversal_context(data, prediction, probs) tendency = self._load_htft_tendency_support(data) ms_h = float(getattr(prediction, "ms_home_prob", 0.0) or 0.0) ms_d = float(getattr(prediction, "ms_draw_prob", 0.0) or 0.0) ms_a = float(getattr(prediction, "ms_away_prob", 0.0) or 0.0) base_prob = float(probs.get(pick, 0.0)) favorite_side = str(context.get("favorite_side") or "") strict_reversal = pick in {"1/2", "2/1"} draw_swing = pick in {"X/1", "X/2"} same_state = pick in {"1/1", "2/2", "X/X"} draw_close = pick in {"1/X", "2/X"} reason_codes: List[str] = [] score = base_prob * 100.0 * 0.85 profile_type = "generic" if strict_reversal: profile_type = "strict_reversal" surprise_support = self._surprise_pattern_support( data=data, prediction=prediction, context=context, ms_probs={"1": ms_h, "X": ms_d, "2": ms_a}, ) score += float(surprise_support.get("score", 0.0)) * 0.42 if float(surprise_support.get("score", 0.0)) >= 48.0: reason_codes.append("htft_strict_reversal_supported") return { "score": min(100.0, score), "profile_type": profile_type, "reversal_prob": float(context.get("reversal_prob", 0.0)), "reason_codes": reason_codes, } if draw_swing: profile_type = "draw_swing" ft_side = pick.split("/", 1)[1] surge = ( float(tendency.get("htft_home_second_half_surge", 1.0)) if ft_side == "1" else float(tendency.get("htft_away_second_half_surge", 1.0)) ) score += float(context.get("draw_swing_prob", 0.0)) * 100.0 * 0.70 score += max(0.0, surge - 1.0) * 18.0 if float(context.get("draw_swing_prob", 0.0)) >= 0.10: reason_codes.append("htft_draw_swing_supported") return { "score": min(100.0, score), "profile_type": profile_type, "swing_prob": float(context.get("draw_swing_prob", 0.0)), "reason_codes": reason_codes, } if same_state: profile_type = "same_state" if pick == "1/1": score += ms_h * 100.0 * 0.24 score += float(tendency.get("htft_home_ht_win_rate", 0.33)) * 22.0 if favorite_side == "1": score += 10.0 reason_codes.append("htft_home_hold_profile") elif pick == "2/2": score += ms_a * 100.0 * 0.24 score += float(tendency.get("htft_away_ht_win_rate", 0.33)) * 22.0 if favorite_side == "2": score += 10.0 reason_codes.append("htft_away_hold_profile") else: score += ms_d * 100.0 * 0.30 score += (1.0 - min(1.0, float(prediction.total_xg) / 4.0)) * 18.0 reason_codes.append("htft_draw_hold_profile") return { "score": min(100.0, score), "profile_type": profile_type, "reason_codes": reason_codes, } if draw_close: profile_type = "draw_close" score += ms_d * 100.0 * 0.26 score += (1.0 - min(1.0, abs(ms_h - ms_a))) * 16.0 reason_codes.append("htft_draw_close_profile") return { "score": min(100.0, score), "profile_type": profile_type, "reason_codes": reason_codes, } return { "score": min(100.0, score), "profile_type": profile_type, "reason_codes": reason_codes, } def _htft_reversal_context( self, data: Any, prediction: Any, htft_probs: Dict[str, float], ) -> Dict[str, Any]: ms_h = self._market_odds(data.odds_data or {}, "MS", "1") ms_a = self._market_odds(data.odds_data or {}, "MS", "2") if ms_h > 1.01 and ms_a > 1.01: favorite_side = "1" if ms_h <= ms_a else "2" underdog_side = "2" if favorite_side == "1" else "1" favorite_odd = ms_h if favorite_side == "1" else ms_a favorite_gap = abs(ms_h - ms_a) else: favorite_side = "1" if float(getattr(prediction, "ms_home_prob", 0.0) or 0.0) >= float(getattr(prediction, "ms_away_prob", 0.0) or 0.0) else "2" underdog_side = "2" if favorite_side == "1" else "1" favorite_odd = 0.0 favorite_gap = abs( float(getattr(prediction, "ms_home_prob", 0.0) or 0.0) - float(getattr(prediction, "ms_away_prob", 0.0) or 0.0) ) * 3.0 reversal_key = "1/2" if favorite_side == "1" else "2/1" draw_swing_key = "X/2" if favorite_side == "1" else "X/1" reversal_prob = float(htft_probs.get(reversal_key, 0.0)) draw_swing_prob = float(htft_probs.get(draw_swing_key, 0.0)) score = min( 100.0, (float(getattr(prediction, "surprise_score", 0.0) or 0.0) * 0.42) + (reversal_prob * 100.0 * 0.88) + (draw_swing_prob * 100.0 * 0.28) + (max(0.0, favorite_gap - 0.40) * 14.0), ) return { "favorite_side": favorite_side, "underdog_side": underdog_side, "favorite_odd": favorite_odd, "favorite_gap": favorite_gap, "reversal_key": reversal_key, "draw_swing_key": draw_swing_key, "reversal_prob": reversal_prob, "draw_swing_prob": draw_swing_prob, "score": score, } def _to_bet_summary_item(self, row: Dict[str, Any]) -> Dict[str, Any]: return { "market": row.get("market"), "pick": row.get("pick"), "raw_confidence": row.get("raw_confidence", row.get("confidence")), "calibrated_confidence": row.get( "calibrated_confidence", row.get("confidence") ), "bet_grade": row.get("bet_grade", "PASS"), "playable": bool(row.get("playable")), "stake_units": float(row.get("stake_units", 0.0)), "play_score": row.get("play_score", 0.0), "ev_edge": row.get("ev_edge", row.get("edge", 0.0)), "implied_prob": row.get("implied_prob", 0.0), "odds_reliability": row.get("market_reliability", 0.5), "odds": row.get("odds", 0.0), "reasons": row.get("decision_reasons", []), } def _market_odds(self, odds: Dict[str, Any], market: str, pick: str) -> float: mapping = { "MS": {"1": "ms_h", "X": "ms_d", "2": "ms_a"}, "DC": {"1X": "dc_1x", "X2": "dc_x2", "12": "dc_12"}, "OU15": {"Over": "ou15_o", "Under": "ou15_u"}, "OU25": {"Over": "ou25_o", "Under": "ou25_u"}, "OU35": {"Over": "ou35_o", "Under": "ou35_u"}, "BTTS": {"Yes": "btts_y", "No": "btts_n"}, "HT": {"1": "ht_h", "X": "ht_d", "2": "ht_a"}, "HT_OU05": {"Over": "ht_ou05_o", "Under": "ht_ou05_u"}, "HT_OU15": {"Over": "ht_ou15_o", "Under": "ht_ou15_u"}, "HTFT": { "1/1": "htft_11", "1/X": "htft_1x", "1/2": "htft_12", "X/1": "htft_x1", "X/X": "htft_xx", "X/2": "htft_x2", "2/1": "htft_21", "2/X": "htft_2x", "2/2": "htft_22" }, "OE": {"Odd": "oe_odd", "Even": "oe_even"}, "CARDS": {"Over": "cards_o", "Under": "cards_u"}, "HCAP": {"1": "hcap_h", "X": "hcap_d", "2": "hcap_a"}, } key = mapping.get(market, {}).get(pick) if not key: return 0.0 try: value = float(odds.get(key, 0.0)) return value if value > 1.0 else 0.0 except Exception: return 0.0 def _get_signal_probs( self, signal: Dict[str, Any], market: str, labels: Iterable[str], ) -> Dict[str, float]: market_payload = signal.get(market, {}) if isinstance(signal, dict) else {} raw_probs = market_payload.get("probs", {}) if isinstance(market_payload, dict) else {} if not isinstance(raw_probs, dict): return {} result = {} for label in labels: for candidate in (label, label.lower(), self._board_key(label)): if candidate in raw_probs: result[label] = float(raw_probs[candidate]) break return self._normalize_probs(result) def _score_table_to_result_probs( self, table: Dict[Tuple[int, int], float], ) -> Dict[str, float]: probs = {"1": 0.0, "X": 0.0, "2": 0.0} for (home_goals, away_goals), prob in table.items(): if home_goals > away_goals: probs["1"] += prob elif home_goals == away_goals: probs["X"] += prob else: probs["2"] += prob return self._normalize_probs(probs) def _score_table_to_total_probs( self, table: Dict[Tuple[int, int], float], line: float, ) -> Dict[str, float]: probs = {"Under": 0.0, "Over": 0.0} for (home_goals, away_goals), prob in table.items(): if (home_goals + away_goals) > line: probs["Over"] += prob else: probs["Under"] += prob return self._normalize_probs(probs) def _score_table_to_btts_probs( self, table: Dict[Tuple[int, int], float], ) -> Dict[str, float]: probs = {"No": 0.0, "Yes": 0.0} for (home_goals, away_goals), prob in table.items(): if home_goals > 0 and away_goals > 0: probs["Yes"] += prob else: probs["No"] += prob return self._normalize_probs(probs) def _score_table_to_odd_even_probs( self, table: Dict[Tuple[int, int], float], ) -> Dict[str, float]: probs = {"Even": 0.0, "Odd": 0.0} for (home_goals, away_goals), prob in table.items(): if (home_goals + away_goals) % 2 == 0: probs["Even"] += prob else: probs["Odd"] += prob return self._normalize_probs(probs) def _score_table_to_handicap_probs( self, table: Dict[Tuple[int, int], float], ) -> Dict[str, float]: probs = {"1": 0.0, "X": 0.0, "2": 0.0} for (home_goals, away_goals), prob in table.items(): diff = home_goals - away_goals if diff >= 2: probs["1"] += prob elif diff == 1: probs["X"] += prob else: probs["2"] += prob return self._normalize_probs(probs) def _poisson_score_table( self, home_xg: float, away_xg: float, max_goals: int, ) -> Dict[Tuple[int, int], float]: table: Dict[Tuple[int, int], float] = {} for home_goals in range(max_goals + 1): for away_goals in range(max_goals + 1): table[(home_goals, away_goals)] = ( self._poisson_prob(home_xg, home_goals) * self._poisson_prob(away_xg, away_goals) ) return self._normalize_table(table) def _normalize_table(self, table: Dict[Tuple[int, int], float]) -> Dict[Tuple[int, int], float]: total = sum(table.values()) if total <= 0: return table return {key: value / total for key, value in table.items()} @staticmethod def _poisson_prob(lmbda: float, k: int) -> float: lmbda = max(0.05, float(lmbda)) return math.exp(-lmbda) * (lmbda ** k) / math.factorial(k) def _blend_probs( self, base: Dict[str, float], anchor: Dict[str, float], base_weight: float, ) -> Dict[str, float]: if not anchor: return self._normalize_probs(base) labels = set(base.keys()) | set(anchor.keys()) blended = { label: (float(base.get(label, 0.0)) * base_weight) + (float(anchor.get(label, 0.0)) * (1.0 - base_weight)) for label in labels } return self._normalize_probs(blended) @staticmethod def _pick_from_probs(probs: Dict[str, float]) -> Tuple[str, float]: if not probs: return "", 0.0 pick = max(probs, key=probs.get) return pick, float(probs[pick]) @staticmethod def _normalize_probs(probs: Dict[str, float]) -> Dict[str, float]: total = sum(max(0.0, float(value)) for value in probs.values()) if total <= 0: if not probs: return {} uniform = 1.0 / len(probs) return {key: uniform for key in probs.keys()} return { key: max(0.0, float(value)) / total for key, value in probs.items() } @staticmethod def _dedupe(items: Iterable[str]) -> List[str]: out: List[str] = [] seen = set() for item in items: if item and item not in seen: out.append(item) seen.add(item) return out @staticmethod def _board_key(label: str) -> str: mapping = { "Over": "over", "Under": "under", "Yes": "yes", "No": "no", "Even": "even", "Odd": "odd", } return mapping.get(label, label) @staticmethod def _display_pick(market: str, pick: str) -> str: mapping = { ("OU15", "Over"): "1.5 Üst", ("OU15", "Under"): "1.5 Alt", ("OU25", "Over"): "2.5 Üst", ("OU25", "Under"): "2.5 Alt", ("OU35", "Over"): "3.5 Üst", ("OU35", "Under"): "3.5 Alt", ("BTTS", "Yes"): "KG Var", ("BTTS", "No"): "KG Yok", ("HT_OU05", "Over"): "İY 0.5 Üst", ("HT_OU05", "Under"): "İY 0.5 Alt", ("HT_OU15", "Over"): "İY 1.5 Üst", ("HT_OU15", "Under"): "İY 1.5 Alt", ("OE", "Odd"): "Tek", ("OE", "Even"): "Çift", ("CARDS", "Over"): "4.5 Üst", ("CARDS", "Under"): "4.5 Alt", } return mapping.get((market, pick), pick) @staticmethod def _same_pick( left: Optional[Dict[str, Any]], right: Optional[Dict[str, Any]], ) -> bool: if not left or not right: return False return ( str(left.get("market")) == str(right.get("market")) and str(left.get("pick")) == str(right.get("pick")) ) @staticmethod def _kelly_stake(true_prob: float, decimal_odds: float) -> float: if decimal_odds <= 1.0 or true_prob <= 0.0 or true_prob >= 1.0: return 0.0 b = decimal_odds - 1.0 q = 1.0 - true_prob f_star = ((b * true_prob) - q) / b if f_star <= 0.0: return 0.0 stake = f_star * 0.25 * 10.0 return min(3.0, max(0.25, stake)) _v26_shadow_engine: Optional[V26ShadowEngine] = None def get_v26_shadow_engine() -> V26ShadowEngine: global _v26_shadow_engine if _v26_shadow_engine is None: config_path = os.getenv("AI_ENGINE_V26_CONFIG_PATH") _v26_shadow_engine = V26ShadowEngine(config_path=config_path) return _v26_shadow_engine