303 lines
12 KiB
Python
303 lines
12 KiB
Python
"""
|
||
Quantitative Finance Module — V2 Betting Engine
|
||
Edge calculation, Fractional Kelly Criterion staking, bet grading, and risk assessment.
|
||
"""
|
||
|
||
from __future__ import annotations
|
||
|
||
import math
|
||
from dataclasses import dataclass
|
||
from typing import Any
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Constants
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
BANKROLL_UNITS: float = 10.0 # Total bankroll in abstract units
|
||
KELLY_FRACTION: float = 0.25 # Quarter-Kelly (conservative, anti-ruin)
|
||
MIN_EDGE_PLAYABLE: float = 0.05 # 5% edge minimum to mark as playable
|
||
MIN_ODDS_PLAYABLE: float = 1.30 # Skip extreme chalk below 1.30
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Edge Calculation
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def calculate_edge(true_prob: float, decimal_odds: float) -> float:
|
||
"""
|
||
Edge = (True_Probability × Decimal_Odds) - 1.0
|
||
Positive edge → the model says we have an advantage over the bookmaker.
|
||
"""
|
||
if decimal_odds <= 1.0 or true_prob <= 0.0:
|
||
return -1.0
|
||
return round((true_prob * decimal_odds) - 1.0, 4)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Kelly Criterion Staking
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def kelly_stake(true_prob: float, decimal_odds: float) -> float:
|
||
"""
|
||
Fractional Kelly Criterion for a bankroll of BANKROLL_UNITS.
|
||
|
||
Full Kelly: f* = ((b × p) - q) / b
|
||
where b = decimal_odds - 1, p = true_prob, q = 1 - true_prob
|
||
|
||
We use KELLY_FRACTION (25%) to reduce variance and avoid ruin.
|
||
Returns stake in units, rounded to 0.1.
|
||
"""
|
||
if decimal_odds <= 1.0 or true_prob <= 0.0 or true_prob >= 1.0:
|
||
return 0.0
|
||
|
||
b = decimal_odds - 1.0
|
||
p = true_prob
|
||
q = 1.0 - p
|
||
|
||
f_star = ((b * p) - q) / b
|
||
|
||
if f_star <= 0.0:
|
||
return 0.0
|
||
|
||
# Scale by fraction and bankroll
|
||
stake = f_star * KELLY_FRACTION * BANKROLL_UNITS
|
||
|
||
# Cap at a sensible maximum (3 units on a 10-unit bankroll)
|
||
stake = min(stake, 3.0)
|
||
|
||
return round(max(0.0, stake), 1)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Bet Grading
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def grade_bet(edge: float, playable: bool) -> str:
|
||
"""
|
||
Assign a letter grade based on edge magnitude.
|
||
A: Edge > 10% — Elite value, rare
|
||
B: Edge > 5% — Strong value, core bets
|
||
C: Edge > 2% — Marginal value, supporting picks only
|
||
PASS: Below threshold — Do not bet
|
||
"""
|
||
if not playable or edge < 0.02:
|
||
return "PASS"
|
||
if edge > 0.10:
|
||
return "A"
|
||
if edge > 0.05:
|
||
return "B"
|
||
return "C"
|
||
|
||
|
||
def is_playable(edge: float, decimal_odds: float) -> bool:
|
||
"""A pick is playable if it has sufficient edge AND reasonable odds."""
|
||
return edge >= MIN_EDGE_PLAYABLE and decimal_odds >= MIN_ODDS_PLAYABLE
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Play Score (0-100 composite)
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
def calculate_play_score(
|
||
edge: float,
|
||
true_prob: float,
|
||
data_quality: float,
|
||
) -> float:
|
||
"""
|
||
Composite score combining edge strength, probability confidence,
|
||
and data quality. Used for ranking picks and filtering.
|
||
|
||
Components:
|
||
- Edge contribution (0-50): edge * 250, capped at 50
|
||
- Prob contribution (0-30): probability * 30
|
||
- DQ contribution (0-20): data_quality * 20
|
||
"""
|
||
edge_score = min(50.0, max(0.0, edge * 250.0))
|
||
prob_score = min(30.0, max(0.0, true_prob * 30.0))
|
||
dq_score = min(20.0, max(0.0, data_quality * 20.0))
|
||
return round(edge_score + prob_score + dq_score, 1)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Risk Assessment
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
@dataclass
|
||
class RiskResult:
|
||
level: str # LOW, MEDIUM, HIGH, EXTREME
|
||
score: float # 0.0 - 1.0
|
||
is_surprise_risk: bool
|
||
surprise_type: str | None
|
||
warnings: list[str]
|
||
|
||
|
||
def assess_risk(
|
||
missing_players_impact: float,
|
||
data_quality_score: float,
|
||
elo_diff: float,
|
||
implied_prob_fav: float,
|
||
) -> RiskResult:
|
||
"""
|
||
Multi-factor risk assessment.
|
||
|
||
Factors:
|
||
1. Missing key players (injuries/suspensions)
|
||
2. Data quality (missing stats, odds)
|
||
3. ELO closeness (tight matches are riskier)
|
||
4. Surprise potential (heavy favorite vulnerable)
|
||
"""
|
||
warnings: list[str] = []
|
||
risk_score = 0.0
|
||
|
||
# ─── Factor 1: Missing players ────────────────────────────────────
|
||
if missing_players_impact > 0.3:
|
||
risk_score += 0.35
|
||
warnings.append(
|
||
f"High missing-player impact: {missing_players_impact:.2f}"
|
||
)
|
||
elif missing_players_impact > 0.15:
|
||
risk_score += 0.15
|
||
warnings.append(
|
||
f"Moderate missing-player impact: {missing_players_impact:.2f}"
|
||
)
|
||
|
||
# ─── Factor 2: Data quality ───────────────────────────────────────
|
||
if data_quality_score < 0.5:
|
||
risk_score += 0.25
|
||
warnings.append(
|
||
f"Low data quality: {data_quality_score:.2f}"
|
||
)
|
||
elif data_quality_score < 0.75:
|
||
risk_score += 0.10
|
||
|
||
# ─── Factor 3: ELO closeness ──────────────────────────────────────
|
||
abs_elo_diff = abs(elo_diff)
|
||
if abs_elo_diff < 50:
|
||
risk_score += 0.15
|
||
warnings.append("Very tight ELO difference — coin-flip territory")
|
||
elif abs_elo_diff < 100:
|
||
risk_score += 0.05
|
||
|
||
# ─── Factor 4: Surprise detection ─────────────────────────────────
|
||
is_surprise = False
|
||
surprise_type: str | None = None
|
||
|
||
if implied_prob_fav > 0.65 and abs_elo_diff < 80:
|
||
# Heavy favorite by odds but ELO says match is closer
|
||
is_surprise = True
|
||
surprise_type = "odds_elo_divergence"
|
||
risk_score += 0.15
|
||
warnings.append(
|
||
"Upset potential: bookmaker odds suggest heavy favorite "
|
||
"but ELO says the match is closer than the market thinks"
|
||
)
|
||
|
||
# ─── Classify ─────────────────────────────────────────────────────
|
||
risk_score = min(1.0, risk_score)
|
||
if risk_score >= 0.7:
|
||
level = "EXTREME"
|
||
elif risk_score >= 0.45:
|
||
level = "HIGH"
|
||
elif risk_score >= 0.2:
|
||
level = "MEDIUM"
|
||
else:
|
||
level = "LOW"
|
||
|
||
return RiskResult(
|
||
level=level,
|
||
score=round(risk_score, 3),
|
||
is_surprise_risk=is_surprise,
|
||
surprise_type=surprise_type,
|
||
warnings=warnings,
|
||
)
|
||
|
||
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
# Market Analysis (orchestrates edge/kelly/grade per market)
|
||
# ═══════════════════════════════════════════════════════════════════════════
|
||
|
||
@dataclass
|
||
class MarketPick:
|
||
market: str
|
||
pick: str
|
||
probability: float
|
||
odds: float
|
||
edge: float
|
||
playable: bool
|
||
bet_grade: str
|
||
stake_units: float
|
||
play_score: float
|
||
decision_reasons: list[str]
|
||
|
||
|
||
def analyze_market(
|
||
market: str,
|
||
probs: dict[str, float],
|
||
odds_map: dict[str, float],
|
||
data_quality_score: float,
|
||
) -> MarketPick:
|
||
"""
|
||
For a given market (MS, OU25, BTTS), find the best pick,
|
||
calculate edge, kelly stake, and grade it.
|
||
|
||
Parameters:
|
||
market: "MS", "OU25", "BTTS"
|
||
probs: {"1": 0.55, "X": 0.25, "2": 0.20} — calibrated model probs
|
||
odds_map: {"1": 2.10, "X": 3.40, "2": 3.50} — decimal odds
|
||
data_quality_score: 0.0-1.0
|
||
"""
|
||
best_pick: str = ""
|
||
best_edge: float = -99.0
|
||
best_prob: float = 0.0
|
||
best_odds: float = 0.0
|
||
reasons: list[str] = []
|
||
|
||
for pick_name, prob in probs.items():
|
||
odd = odds_map.get(pick_name, 0.0)
|
||
if odd <= 1.0:
|
||
continue
|
||
|
||
edge = calculate_edge(prob, odd)
|
||
if edge > best_edge:
|
||
best_edge = edge
|
||
best_pick = pick_name
|
||
best_prob = prob
|
||
best_odds = odd
|
||
|
||
if not best_pick:
|
||
return MarketPick(
|
||
market=market, pick="", probability=0.0, odds=0.0,
|
||
edge=0.0, playable=False, bet_grade="PASS",
|
||
stake_units=0.0, play_score=0.0,
|
||
decision_reasons=["no_valid_odds_found"],
|
||
)
|
||
|
||
playable = is_playable(best_edge, best_odds)
|
||
grade = grade_bet(best_edge, playable)
|
||
stake = kelly_stake(best_prob, best_odds) if playable else 0.0
|
||
play_score = calculate_play_score(best_edge, best_prob, data_quality_score)
|
||
|
||
# Build decision reasons
|
||
if playable:
|
||
reasons.append(f"edge_{best_edge:.1%}_above_threshold")
|
||
reasons.append(f"kelly_stake_{stake:.1f}_units")
|
||
else:
|
||
if best_edge < MIN_EDGE_PLAYABLE:
|
||
reasons.append(f"edge_{best_edge:.1%}_below_{MIN_EDGE_PLAYABLE:.0%}_threshold")
|
||
if best_odds < MIN_ODDS_PLAYABLE:
|
||
reasons.append(f"odds_{best_odds:.2f}_below_{MIN_ODDS_PLAYABLE:.2f}_minimum")
|
||
|
||
return MarketPick(
|
||
market=market,
|
||
pick=best_pick,
|
||
probability=round(best_prob, 4),
|
||
odds=round(best_odds, 2),
|
||
edge=round(best_edge, 4),
|
||
playable=playable,
|
||
bet_grade=grade,
|
||
stake_units=stake,
|
||
play_score=play_score,
|
||
decision_reasons=reasons,
|
||
)
|