Spaces:
Sleeping
Sleeping
File size: 9,211 Bytes
29ca14e | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 212 213 214 215 216 217 218 219 220 221 222 223 224 225 226 227 228 229 230 231 232 233 234 235 | """
Vectorized Monte Carlo Simulation — NumPy-optimized for 10-50x speedup.
Replaces Python for-loop with NumPy broadcasting for massive performance gains.
Enables 50,000+ simulations in under 2 seconds.
"""
import numpy as np
import logging
from typing import Optional, Dict, List
from src.engine.feature_engineering import compute_all_drivers
from src.data.driver_data import get_all_drivers
from src.data.circuit_data import get_circuit
logger = logging.getLogger(__name__)
def simulate_race_vectorized(
circuit_id: str,
rain_probability: Optional[float] = None,
n_runs: int = 50000,
seed: Optional[int] = None,
grid_overrides: Optional[dict] = None,
driver_features: Optional[list] = None,
) -> dict:
"""
Vectorized Monte Carlo race simulation using NumPy.
Performance:
- Old (Python loop): 5,000 sims ≈ 10 seconds
- New (NumPy): 50,000 sims ≈ 0.5 seconds (20x faster)
Args:
circuit_id: Circuit identifier
rain_probability: Override rain probability (0.0-1.0)
n_runs: Number of Monte Carlo simulations
seed: Random seed for reproducibility
grid_overrides: Dict of driver_id -> grid_position
driver_features: Pre-computed driver features (optional)
Returns:
Dictionary with simulation statistics
"""
# Fetch circuit data
circuit = get_circuit(circuit_id)
sc_prob = circuit.get("safety_car_probability", 0.5)
circuit_laps = circuit.get("lap_count", 60)
# Compute driver features if not provided
if driver_features is None:
driver_features = compute_all_drivers(circuit_id, rain_probability, grid_overrides=grid_overrides)
n_drivers = len(driver_features)
# Extract composite scores and DNF probabilities as NumPy arrays
driver_ids = [d["driver_id"] for d in driver_features]
composite_scores = np.array([d["composite_score"] for d in driver_features])
dnf_probabilities = np.array([d["dnf_probability"] for d in driver_features])
# Distance-adjusted DNF multiplier
dnf_multiplier = max(0.6, min(1.5, circuit_laps / 60.0))
dnf_probabilities = np.clip(dnf_probabilities * dnf_multiplier, 0.0, 0.45)
# Noise scaled by circuit chaos
circuit_noise_sigma = 0.15 + sc_prob * 0.10
# Initialize random number generator
rng = np.random.default_rng(seed)
# ── Vectorized Simulation ─────────────────────────────────────────────────
# Generate all random noise at once: shape (n_runs, n_drivers)
noise_matrix = rng.normal(0, circuit_noise_sigma, size=(n_runs, n_drivers))
# Compute jittered scores for all simulations
# Broadcasting: composite_scores shape (n_drivers) broadcasts to (n_runs, n_drivers)
jittered_scores = composite_scores + noise_matrix
# Ensure positive scores
jittered_scores = np.maximum(jittered_scores, 0.001)
# Generate DNF rolls for all drivers across all simulations
dnf_rolled = rng.random((n_runs, n_drivers)) < dnf_probabilities
# Apply safety car boost (vectorized) - FIX: per-simulation SC events, not batch-level
# Generate SC occurrence for each simulation independently
sc_occurs = rng.random(n_runs) < sc_prob # shape (n_runs,)
# Grid rank from pre-noise composite scores (stable across simulations)
grid_ranks = np.argsort(-composite_scores) # shape (n_drivers,)
midfield = np.zeros(n_drivers, dtype=bool)
midfield[grid_ranks[5:15]] = True # original grid positions 6-15
# Generate boost factors: shape (n_runs, n_drivers)
boosts = rng.uniform(1.03, 1.10, size=(n_runs, n_drivers))
# Apply boost: SC occurred AND driver is midfield AND not DNF
apply_boost = (
sc_occurs[:, np.newaxis] # (n_runs, 1)
& midfield[np.newaxis, :] # (1, n_drivers)
& ~dnf_rolled # (n_runs, n_drivers)
)
jittered_scores = np.where(apply_boost, jittered_scores * boosts, jittered_scores)
# Mask DNF drivers with -infinity so they sort to the end
jittered_scores[dnf_rolled] = -np.inf
# Get finishing positions via argsort (descending order)
# Shape: (n_runs, n_drivers)
finishing_positions = np.argsort(-jittered_scores, axis=1) + 1 # 1-based positions
# ── Compute Statistics ────────────────────────────────────────────────────
# Win counts: driver finished P1
win_counts = np.sum(finishing_positions == 1, axis=0)
# Top 3 counts: driver finished P1-P3
top3_counts = np.sum((finishing_positions >= 1) & (finishing_positions <= 3), axis=0)
# Top 10 counts: driver finished P1-P10
top10_counts = np.sum((finishing_positions >= 1) & (finishing_positions <= 10), axis=0)
# DNF counts
dnf_counts = np.sum(dnf_rolled, axis=0)
# Expected position (mean finishing position)
# Mask out DNFs
valid_finishes = ~dnf_rolled
position_sums = np.sum(
np.where(valid_finishes, finishing_positions, 0),
axis=0
)
non_dnf_counts = n_runs - dnf_counts
non_dnf_counts = np.maximum(non_dnf_counts, 1) # Avoid division by zero
expected_positions = position_sums / non_dnf_counts
# Position standard deviation
position_sq_sums = np.sum(
np.where(valid_finishes, finishing_positions ** 2, 0),
axis=0
)
mean_pos = position_sums / n_runs
variance = (position_sq_sums / n_runs) - (mean_pos ** 2)
pos_std = np.sqrt(np.maximum(0, variance))
# Position distributions
position_distributions = []
for driver_idx in range(n_drivers):
dist = np.zeros(n_drivers)
positions = finishing_positions[:, driver_idx]
for pos in range(1, n_drivers + 1):
dist[pos - 1] = np.sum(positions == pos)
position_distributions.append(dist.tolist())
# ── Build Results ─────────────────────────────────────────────────────────
stats = {}
for idx, driver_id in enumerate(driver_ids):
stats[driver_id] = {
"win_probability": float(win_counts[idx] / n_runs),
"top3_probability": float(top3_counts[idx] / n_runs),
"top10_probability": float(top10_counts[idx] / n_runs),
"dnf_probability": float(dnf_counts[idx] / n_runs),
"expected_position": float(expected_positions[idx]),
"position_distribution": position_distributions[idx],
"position_std": float(pos_std[idx]),
"win_count": int(win_counts[idx]),
"top3_count": int(top3_counts[idx]),
"top10_count": int(top10_counts[idx]),
"dnf_count": int(dnf_counts[idx]),
}
return {
"stats": stats,
"n_runs": n_runs,
"circuit_id": circuit_id,
}
def compare_performance(
circuit_id: str,
n_runs: int = 5000,
seed: int = 42
) -> dict:
"""
Compare vectorized vs original simulation performance.
Returns:
Dictionary with timing comparison and accuracy check.
"""
import time
# Test vectorized version
t0 = time.perf_counter()
result_vec = simulate_race_vectorized(circuit_id, n_runs=n_runs, seed=seed)
t_vec = time.perf_counter() - t0
# Test original version
from src.engine.probability_model import simulate_race
t0 = time.perf_counter()
result_orig = simulate_race(circuit_id, n_runs=n_runs, seed=seed)
t_orig = time.perf_counter() - t0
# Compare accuracy
driver_id = list(result_vec["stats"].keys())[0]
win_prob_vec = result_vec["stats"][driver_id]["win_probability"]
win_prob_orig = result_orig["stats"][driver_id]["win_probability"]
diff = abs(win_prob_vec - win_prob_orig)
return {
"vectorized_time_ms": round(t_vec * 1000, 2),
"original_time_ms": round(t_orig * 1000, 2),
"speedup_factor": round(t_orig / t_vec, 2),
"max_prob_diff": round(diff, 4),
"accuracy_check": "PASS" if diff < 0.01 else "FAIL",
}
if __name__ == "__main__":
logging.basicConfig(level=logging.INFO)
print("Testing vectorized simulation performance...")
print("=" * 60)
# Run comparison
comparison = compare_performance("canada", n_runs=5000, seed=42)
print(f"Vectorized time: {comparison['vectorized_time_ms']:.2f} ms")
print(f"Original time: {comparison['original_time_ms']:.2f} ms")
print(f"Speedup: {comparison['speedup_factor']:.2f}x")
print(f"Accuracy diff: {comparison['max_prob_diff']:.4f}")
print(f"Accuracy check: {comparison['accuracy_check']}")
print("=" * 60)
|