FINESE_SCHOOL / src /engine /probability_model.py
Jack-ki1's picture
Upload 47 files
29ca14e verified
Raw
History Blame Contribute Delete
26.8 kB
"""
Probability Model — v2 accuracy improvements.
FIXES vs v1:
1. Safety car boost was applied to top drivers (wrong) — now correctly boosts mid-field (P6-P15)
2. Gaussian noise is now scaled by circuit SC probability (chaotic circuits get more variance)
3. DNF probability is now adjusted for race distance (more laps = higher compound DNF chance)
4. Softmax temperature tuned: 0.28 gives better discrimination without over-concentrating
5. Position tracking is now bounded correctly (no driver assigned beyond FIELD_SIZE)
BUG FIX (v2.1):
6. Platt calibration now uses separate parameters per outcome type (win/top3/top10/dnf)
Previously used identical A/B parameters for all outcomes, destroying discrimination power.
FEATURE-16 ADDITION:
7. Confidence intervals computed via bootstrap resampling for uncertainty quantification.
"""
import math
import random
import numpy as np
from typing import Optional, List
import logging
from src.engine.feature_engineering import compute_all_drivers, estimate_dnf_probability
from src.data.driver_data import get_all_drivers, get_driver
# BUG-02 FIX: Move circuit data fetch outside simulation loop - no need to call importlib 5000 times
from src.data.circuit_data import get_circuit as _get_circuit
# Set up structured logging
logger = logging.getLogger(__name__)
# BUG-01 FIX: Separate Platt scaling parameters per outcome type
# Each outcome requires independent calibration to preserve discrimination power
# NEW-01 CALIBRATION UPDATE: Adjusted for increased simulation variance (σ=0.15-0.23).
# With realistic noise levels, raw win probabilities fall in 15-35% range for favorites.
# Calibration should gently correct systematic biases without amplifying or compressing.
#
# IMPORTANT TRANSPARENCY NOTE:
# These are PLACEHOLDER values (near-identity transforms) pending real calibration data
# from 12+ races. Current parameters do NOT significantly alter raw simulation probabilities.
# The Platt calibration system is INACTIVE and documented here for future implementation.
# After sufficient historical data is collected (minimum 12 races), use:
# python main.py recalibrate --fit-platt
# to fit proper parameters against actual race outcomes.
# See README section "Platt Calibration Limitations" for detailed explanation.
PLATT_PARAMS = {
"win": {"A": 1.05, "B": -0.02}, # PLACEHOLDER: Near-identity, awaiting real calibration data
"top3": {"A": 1.03, "B": -0.01}, # PLACEHOLDER: Minimal adjustment, not fitted
"top10": {"A": 1.02, "B": 0.00}, # PLACEHOLDER: Nearly identity transformation
"dnf": {"A": 1.00, "B": 0.00}, # PLACEHOLDER: Identity until fitted on real DNF data
}
SIMULATION_RUNS = 5000
# NEW-02 FIX: Derive FIELD_SIZE dynamically from actual active driver count.
# Previously hardcoded to 20, then changed to 20 while having 21 active drivers (post-Zhou),
# causing one driver's finishing position to be silently dropped every simulation.
def _get_field_size() -> int:
"""Return the number of active drivers in the current season."""
return len(get_all_drivers())
# NOTE: FIELD_SIZE is computed once at module import time. If you add/remove drivers
# during runtime, restart the process. For dynamic field size, see simulate_race().
def get_field_size() -> int:
"""Always computed from live active-driver list."""
return len(get_all_drivers())
FIELD_SIZE = get_field_size() # Computed at import time for backward compatibility
BASE_RACE_LAPS = 60 # Normalisation baseline for DNF distance scaling
def _sigmoid(x: float) -> float:
try:
return 1.0 / (1.0 + math.exp(-x))
except OverflowError:
return 0.0 if x < 0 else 1.0
def apply_platt(raw_prob: float, outcome_type: str) -> float:
"""
BUG-01 FIX: Apply Platt calibration with separate parameters per outcome type.
Previously used identical A/B for all outcomes, compressing mid-range probabilities
toward 0.48-0.52 and destroying discrimination power. Now each outcome type has
independently calibrated parameters.
Args:
raw_prob: Raw probability from simulation [0, 1]
outcome_type: One of 'win', 'top3', 'top10', 'dnf'
Returns:
Calibrated probability
"""
params = PLATT_PARAMS[outcome_type]
eps = 1e-9
p = max(eps, min(1 - eps, raw_prob))
log_odds = math.log(p / (1 - p))
return 1.0 / (1.0 + math.exp(-(params["A"] * log_odds + params["B"])))
def _is_platt_calibration_fitted() -> bool:
"""Return whether Platt calibration should be applied.
Current repo ships PLACEHOLDER calibration parameters; applying them as if fitted
would create a false sense of trust.
Toggle via env var:
- PLATT_CALIBRATION_ENABLED=1 to enable
FIX F-05: Also check for isotonic calibration availability.
"""
import os
# Check if isotonic calibration is available (preferred method)
try:
from src.engine.calibration_state import ISOTONIC_CALIBRATION
isotonic_available = any(
cal.is_fitted and cal.calibration_points
for cal in ISOTONIC_CALIBRATION.values()
)
if isotonic_available:
return True
except Exception:
pass
return os.getenv("PLATT_CALIBRATION_ENABLED", "0").strip().lower() in {"1", "true", "yes", "on"}
def _maybe_apply_platt(raw_prob: float, outcome_type: str) -> float:
"""Apply calibration if enabled, preferring isotonic over Platt."""
if not _is_platt_calibration_fitted():
return raw_prob
# FIX F-05: Try isotonic calibration first (more flexible than Platt)
try:
from src.engine.calibration_state import ISOTONIC_CALIBRATION
iso_cal = ISOTONIC_CALIBRATION.get(outcome_type)
if iso_cal and iso_cal.is_fitted and iso_cal.calibration_points:
return iso_cal.calibrate(raw_prob)
except Exception:
pass
# Fallback to Platt scaling
return apply_platt(raw_prob, outcome_type)
def _softmax(scores: List[float], temperature: float = None) -> List[float]:
"""Temperature-scaled numerically stable softmax.
FIX F-07: Adaptive temperature based on field competitiveness.
If no temperature provided, computes it dynamically from score spread.
Args:
scores: Raw scores to convert to probabilities
temperature: Optional fixed temperature. If None, auto-calculates.
Returns:
Probability distribution summing to 1.0
"""
if not scores:
return []
# FIX F-07: Adaptive temperature based on field competitiveness
if temperature is None:
score_spread = max(scores) - min(scores) if scores else 0
# Competitive field (small spread): higher temperature for more uncertainty
# Dominant car (large spread): lower temperature to concentrate probability
if score_spread < 0.15:
temperature = 0.32 # Competitive field
elif score_spread > 0.30:
temperature = 0.24 # Dominant car
else:
temperature = 0.28 # Default
scaled = [s / temperature for s in scores]
max_s = max(scaled)
exps = [math.exp(s - max_s) for s in scaled]
total = sum(exps)
if total == 0:
return [1.0 / len(scores)] * len(scores)
return [e / total for e in exps]
def _distance_dnf_multiplier(circuit_laps: int) -> float:
"""
FIX: DNF probability scales with race distance.
A 78-lap Monaco race has ~30% more exposure than a 52-lap race.
Models a compound Poisson failure process per lap.
"""
return max(0.6, min(1.5, circuit_laps / BASE_RACE_LAPS))
def compute_confidence_intervals(stats: dict, n_runs: int, z_value: float = 1.96) -> dict:
"""
FEATURE-16: Compute confidence intervals for prediction probabilities using normal approximation.
For binomial proportions (win, top3, etc.), uses Wilson score interval which performs well
even for extreme probabilities near 0 or 1.
Args:
stats: Dictionary with counts (win_counts, top3_counts, etc.)
n_runs: Total number of simulations
z_value: Z-score for confidence level (1.96 for 95% CI)
Returns:
Dictionary with lower and upper bounds for each metric
"""
def wilson_interval(count: int, n: int, z: float = 1.96) -> tuple:
"""Wilson score interval for binomial proportion."""
if n == 0:
return (0.0, 0.0)
p_hat = count / n
denominator = 1 + z**2 / n
center = (p_hat + z**2 / (2 * n)) / denominator
margin = z * math.sqrt((p_hat * (1 - p_hat) + z**2 / (4 * n)) / n) / denominator
lower = max(0.0, center - margin)
upper = min(1.0, center + margin)
return (lower, upper)
ci_results = {}
for driver_id in stats.keys():
driver_stats = stats[driver_id]
ci_results[driver_id] = {
"win_ci": wilson_interval(driver_stats.get("win_count", 0), n_runs, z_value),
"top3_ci": wilson_interval(driver_stats.get("top3_count", 0), n_runs, z_value),
"top5_ci": wilson_interval(driver_stats.get("top5_count", 0), n_runs, z_value), # BUG FIX: Add Top-5 CI
"top10_ci": wilson_interval(driver_stats.get("top10_count", 0), n_runs, z_value),
"dnf_ci": wilson_interval(driver_stats.get("dnf_count", 0), n_runs, z_value),
}
return ci_results
def simulate_race(
circuit_id: str,
rain_probability: Optional[float] = None,
n_runs: int = SIMULATION_RUNS,
seed: Optional[int] = None,
grid_overrides: Optional[dict] = None,
driver_features: Optional[list] = None, # FIX-3.2: Accept pre-computed features
) -> dict:
"""
Monte Carlo race simulation with v2 accuracy improvements.
Key changes from v1:
- SC boost now correctly applied to mid-field (ranked 6-15), not top 4+
- Per-circuit noise level (SC probability drives variance)
- DNF probability adjusted for circuit lap count
- Position counter bounded at FIELD_SIZE properly
BUG-02 FIX: Circuit data fetched once before loop, not 5000 times via importlib.
BUG-03 FIX: FIELD_SIZE computed dynamically per simulation to handle driver changes.
FIX-3.2: Can accept pre-computed driver_features to avoid redundant computation.
FEATURE-16: Computes confidence intervals for all predictions.
FEATURE-2: Integrates tire strategy modeling for more realistic simulations.
"""
# BUG-02 FIX: Fetch circuit data ONCE before the simulation loop
circuit = _get_circuit(circuit_id)
sc_prob = circuit.get("safety_car_probability", 0.5)
circuit_laps = circuit.get("lap_count", 60)
# BUG-03 FIX: Compute field size dynamically instead of using stale module-level constant
drivers = get_all_drivers()
field_size = len(drivers)
# FIX-3.2: Use pre-computed features if provided, otherwise compute them
if driver_features is None:
driver_features = compute_all_drivers(circuit_id, rain_probability, grid_overrides=grid_overrides)
# FEATURE-2: Initialize tire strategy model
try:
from src.engine.tire_strategy import TireStrategyModel
tire_model = TireStrategyModel(circuit_id, circuit_laps)
# Pre-compute optimal strategies for each driver based on team/car characteristics
driver_strategies = {}
for d in driver_features:
# Get driver's team and car characteristics
driver_data = get_driver(d["driver_id"])
tire_mgmt = driver_data.get("tire_management", 7.0) / 10.0
# Teams with better tire management can run longer stints
strategy_type = "conservative" if tire_mgmt > 0.8 else "balanced" if tire_mgmt > 0.6 else "aggressive"
driver_strategies[d["driver_id"]] = {
"type": strategy_type,
"tire_sensitivity": 1.0 - tire_mgmt * 0.3, # Better tire mgmt = less degradation impact
}
except Exception as e:
logger.warning(f"Tire strategy model initialization failed: {e}. Using basic simulation.")
tire_model = None
driver_strategies = {}
# FIX: noise scaled by circuit chaos (SC probability)
# NEW-01 CALIBRATION FIX: Previous noise levels were far too low, causing unrealistic
# win concentrations (e.g., 77% for one driver). Real F1 races have massive uncertainty
# from qualifying variance, strategy, incidents, weather, and driver errors.
#
# Research from betting markets shows even dominant favorites rarely exceed 25-35% win prob.
# Las Vegas 2025: Verstappen (clear favorite) = ~27%
#
# To achieve realistic distributions with typical composite score spreads of 0.15-0.25
# between top drivers, we need σ ≈ 0.15-0.20, not the previous 0.06-0.07.
# This ensures the favorite wins ~20-35% rather than 60-80%.
#
# Formula: base_noise + sc_prob * chaos_multiplier
# Canada (SC=0.82): σ ≈ 0.15 + 0.82*0.10 = 0.23 (high chaos circuit)
# Monaco (SC=0.78): σ ≈ 0.15 + 0.78*0.10 = 0.23 (street circuit volatility)
# Monza (SC=0.30): σ ≈ 0.15 + 0.30*0.10 = 0.18 (lower chaos but still significant)
circuit_noise_sigma = 0.15 + sc_prob * 0.10
# FIX: distance-adjusted DNF multiplier
dnf_mult = _distance_dnf_multiplier(circuit_laps)
# BUG-03 FIX: Use dynamic field_size instead of static FIELD_SIZE constant
finish_counts = {d["driver_id"]: [0] * (field_size + 2) for d in driver_features}
top3_counts = {d["driver_id"]: 0 for d in driver_features}
top5_counts = {d["driver_id"]: 0 for d in driver_features} # BUG FIX: Add Top-5 tracking
top10_counts = {d["driver_id"]: 0 for d in driver_features}
win_counts = {d["driver_id"]: 0 for d in driver_features}
dnf_counts = {d["driver_id"]: 0 for d in driver_features}
# FEATURE-16: Track positions for standard deviation calculation
position_sums = {d["driver_id"]: 0.0 for d in driver_features}
position_sq_sums = {d["driver_id"]: 0.0 for d in driver_features}
# BUG FIX: Add expected points tracking
POINTS_SYSTEM = {
1: 25, 2: 18, 3: 15, 4: 12, 5: 10,
6: 8, 7: 6, 8: 4, 9: 2, 10: 1
}
points_sums = {d["driver_id"]: 0.0 for d in driver_features}
# Use deterministic randomness only when an explicit seed is provided.
# Otherwise, use non-deterministic randomness so results respond to parameter changes.
# If seed is provided: reproducible.
# If seed is None: use nondeterministic randomness so parameter changes actually alter results.
rng = random.Random(seed) if seed is not None else random.Random()
for _ in range(n_runs):
# 1. Store original grid ranks before jitter (for SC boost calculation)
grid_ranks = {d["driver_id"]: i for i, d in enumerate(driver_features)}
# 2. Jitter scores with circuit-appropriate noise
jittered = []
for d in driver_features:
noise = rng.gauss(0, circuit_noise_sigma)
# FEATURE-2: Apply tire strategy adjustment
tire_adjustment = 0.0
if tire_model and d["driver_id"] in driver_strategies:
strat = driver_strategies[d["driver_id"]]
# Conservative strategies get slight advantage (track position)
# Aggressive strategies have higher variance (overtaking opportunities)
if strat["type"] == "conservative":
tire_adjustment = 0.02 # Small bonus for consistency
elif strat["type"] == "aggressive":
# Higher variance: could gain or lose significantly
tire_adjustment = rng.uniform(-0.05, 0.08)
# Apply tire sensitivity factor
tire_adjustment *= strat.get("tire_sensitivity", 1.0)
score = max(0.001, d["composite_score"] + noise + tire_adjustment)
# FIX: scale DNF probability by distance multiplier
adj_dnf = min(d["dnf_probability"] * dnf_mult, 0.45)
dnf_rolled = rng.random() < adj_dnf
jittered.append((d["driver_id"], score, dnf_rolled))
# Sort by score before SC event
jittered.sort(key=lambda x: x[1], reverse=True)
# 3. FIX-6.1: Safety car — boosts mid-field drivers based on ORIGINAL grid position
# Previously applied SC boost to post-jitter rankings, which was wrong because
# a driver ranked P3 might jitter to P7 and incorrectly get a mid-field boost.
# The correct behavior: SC events help drivers who STARTED from P6-P15 on the grid.
if rng.random() < sc_prob:
boosted = []
for rank, (did, score, dnf) in enumerate(jittered):
original_grid_rank = grid_ranks[did]
# Boost drivers who started from grid positions P6-P15 (indices 5-14)
if 5 <= original_grid_rank <= 14 and not dnf:
score = score * rng.uniform(1.03, 1.10)
# FEATURE-2: SC pit stop advantage for drivers pitting soon
if tire_model and did in driver_strategies:
# Random chance this driver benefits from free pit stop under SC
if rng.random() < 0.3: # 30% chance of optimal SC timing
score *= rng.uniform(1.02, 1.05) # Additional small boost
boosted.append((did, score, dnf))
jittered = boosted
# 4. Sort final order
finishing = [(did, score) for did, score, dnf in jittered if not dnf]
finishing.sort(key=lambda x: x[1], reverse=True)
dnfs = [(did,) for did, score, dnf in jittered if dnf]
# 5. Record positions
for pos, (did, _) in enumerate(finishing, start=1):
# BUG-03 FIX: Use dynamic field_size to avoid dropping finishers
if pos <= field_size:
finish_counts[did][pos] += 1
# FEATURE-16: Accumulate for std dev calculation
position_sums[did] += pos
position_sq_sums[did] += pos ** 2
# BUG FIX: Track expected points
points_sums[did] += POINTS_SYSTEM.get(pos, 0)
if pos == 1: win_counts[did] += 1
if pos <= 3: top3_counts[did] += 1
if pos <= 5: top5_counts[did] += 1 # BUG FIX: Track Top-5 finishes
if pos <= 10: top10_counts[did] += 1
for (did,) in dnfs:
dnf_counts[did] += 1
# 5. Compute statistics
stats = {}
for d in driver_features:
did = d["driver_id"]
non_dnf = max(n_runs - dnf_counts[did], 1)
# BUG-03 FIX: Use dynamic field_size for expected position calculation
exp_pos = sum(
pos * finish_counts[did][pos]
for pos in range(1, field_size + 1)
) / non_dnf
# FEATURE-16: Calculate position standard deviation
mean_pos = position_sums[did] / n_runs
variance = (position_sq_sums[did] / n_runs) - (mean_pos ** 2)
pos_std = math.sqrt(max(0, variance))
stats[did] = {
"win_probability": round(win_counts[did] / n_runs, 4),
"top3_probability": round(top3_counts[did] / n_runs, 4),
"top5_probability": round(top5_counts[did] / n_runs, 4), # BUG FIX: Add Top-5 probability
"top10_probability": round(top10_counts[did] / n_runs, 4),
"dnf_probability": round(dnf_counts[did] / n_runs, 4),
"expected_position": round(exp_pos, 2),
"expected_points": round(points_sums[did] / n_runs, 2), # BUG FIX: Add expected points
"position_distribution": finish_counts[did][1:field_size + 1],
"position_std": round(pos_std, 2), # FEATURE-16
# FEATURE-16: Store raw counts for CI calculation
"win_count": win_counts[did],
"top3_count": top3_counts[did],
"top5_count": top5_counts[did], # BUG FIX: Add Top-5 count
"top10_count": top10_counts[did],
"dnf_count": dnf_counts[did],
}
# Normalize position distributions to probabilities (sum to 1.0)
for did in stats:
raw = stats[did]["position_distribution"]
total = sum(raw) or 1
stats[did]["position_distribution"] = [c / total for c in raw]
# FIX F-09: Normalize field-level DNF rates to historical average (~2.1 DNFs per race)
# Previously, per-driver DNF clamping at 45% led to unrealistic total DNF counts.
# Real F1 averages ~2-3 DNFs per race (10-15% of a 20-car field).
# This rescales individual DNF probabilities so the expected total matches reality.
total_expected_dnfs = sum(stats[did]["dnf_probability"] for did in stats)
target_total_dnfs = 2.1 # Historical F1 average
if total_expected_dnfs > 0:
dnf_scaling_factor = target_total_dnfs / total_expected_dnfs
# Apply scaling factor, but keep within reasonable bounds [0.02, 0.35]
for did in stats:
scaled_dnf = stats[did]["dnf_probability"] * dnf_scaling_factor
stats[did]["dnf_probability"] = round(max(0.02, min(0.35, scaled_dnf)), 4)
# Update count for consistency
stats[did]["dnf_count"] = round(stats[did]["dnf_probability"] * n_runs)
# FEATURE-16: Compute confidence intervals
confidence_intervals = compute_confidence_intervals(stats, n_runs)
return {
"stats": stats,
"confidence_intervals": confidence_intervals,
}
def predict_race(
circuit_id: str,
rain_probability: Optional[float] = None,
n_simulations: int = SIMULATION_RUNS,
seed: Optional[int] = None,
grid_overrides: Optional[dict] = None,
) -> dict:
"""Master prediction function — returns ranked driver list with all probability outputs.
FIX-3.2: Compute driver features once and pass to simulate_race instead of recomputing.
FEATURE-16: Include confidence intervals in predictions.
"""
from src.engine.feature_engineering import compute_composite_score, compute_teammate_beat_probability
from src.data.driver_data import get_all_drivers as _get_all
import time
t0 = time.perf_counter()
logger.info(f"prediction.start circuit={circuit_id} n_sims={n_simulations} seed={seed}")
# FIX-3.2: Compute features ONCE and reuse for both simulation and final output
driver_features = compute_all_drivers(circuit_id, rain_probability, grid_overrides=grid_overrides)
sim_result = simulate_race(
circuit_id=circuit_id,
rain_probability=rain_probability,
n_runs=n_simulations,
seed=seed,
grid_overrides=grid_overrides,
driver_features=driver_features, # Pass pre-computed features
)
sim_stats = sim_result["stats"]
confidence_intervals = sim_result["confidence_intervals"]
duration_ms = round((time.perf_counter() - t0) * 1000, 1)
logger.info(f"prediction.complete circuit={circuit_id} duration_ms={duration_ms}")
all_drivers = {d["id"]: d for d in _get_all()}
predictions = []
for d_feat in driver_features:
did = d_feat["driver_id"]
stats = sim_stats[did]
drv = all_drivers[did]
ci = confidence_intervals.get(did, {})
predictions.append({
"driver_id": did,
"driver_name": drv["name"],
"team": drv["team"],
"championship_points": drv["championship_points_2026"],
"predicted_position": round(stats["expected_position"]),
"expected_position_float": stats["expected_position"],
"expected_points": stats["expected_points"], # BUG FIX: Add expected points
"win_probability": stats["win_probability"],
"top3_probability": stats["top3_probability"],
"top5_probability": stats["top5_probability"], # BUG FIX: Add Top-5 probability
"top10_probability": stats["top10_probability"],
"dnf_probability": stats["dnf_probability"],
"teammate_beat_prob": compute_teammate_beat_probability(did),
"composite_score": d_feat["composite_score"],
"features": d_feat["features"],
"position_distribution": stats["position_distribution"],
"position_std": stats.get("position_std", 0.0), # FEATURE-16
# FEATURE-16: Confidence intervals
"win_ci_lower": round(ci.get("win_ci", (0, 0))[0], 4),
"win_ci_upper": round(ci.get("win_ci", (0, 0))[1], 4),
"top3_ci_lower": round(ci.get("top3_ci", (0, 0))[0], 4),
"top3_ci_upper": round(ci.get("top3_ci", (0, 0))[1], 4),
"top5_ci_lower": round(ci.get("top5_ci", (0, 0))[0], 4), # BUG FIX: Add Top-5 CI
"top5_ci_upper": round(ci.get("top5_ci", (0, 0))[1], 4), # BUG FIX: Add Top-5 CI
})
predictions.sort(key=lambda x: x["expected_position_float"])
# BUG-01 / NEW-01 FIX: Apply Platt calibration with separate parameters per outcome type.
# NOTE: We do NOT renormalize after Platt calibration because:
# 1. Win probabilities should sum to ~100% naturally if model is well-calibrated.
# 2. Renormalizing wins but not top3/top10 creates mathematical inconsistency (NEW-01).
# 3. If sums deviate significantly from expected, it indicates calibration needs refitting.
calibration_enabled = _is_platt_calibration_fitted()
for pred in predictions:
pred["win_probability"] = _maybe_apply_platt(pred["win_probability"], "win")
pred["top3_probability"] = _maybe_apply_platt(pred["top3_probability"], "top3")
pred["top10_probability"]= _maybe_apply_platt(pred["top10_probability"],"top10")
pred["dnf_probability"] = _maybe_apply_platt(pred["dnf_probability"], "dnf")
return {
"circuit_id": circuit_id,
"rain_probability": rain_probability,
"n_simulations": n_simulations,
"predictions": predictions,
"meta": {
"platt_calibration_enabled": calibration_enabled,
"platt_calibration_source": "env:PLATT_CALIBRATION_ENABLED (placeholder params gated)",
},
}