2052 lines
82 KiB
Python
2052 lines
82 KiB
Python
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
|