Spaces:
Configuration error
Configuration error
File size: 8,219 Bytes
e23172f | 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 | """Run backtesting with full re-analysis.
Instead of using stored predictions, this re-runs the full scoring
pipeline with different weight profiles to find optimal weights.
"""
import json
import logging
from pathlib import Path
from collections import defaultdict
logging.basicConfig(level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s")
import sys
sys.path.insert(0, '/home/brettanthonysjoberg179/trifecta-bro-hf-space')
from trifecta_bro.api.formfav_client import FormFavClient, FormFavAuthError, FormFavNotFoundError
from trifecta_bro.data.normalizer import normalise_meeting, normalise_race
from trifecta_bro.model.analyse_race import analyse_race
from trifecta_bro.model.distance_profiles import classify_distance, get_profile
from trifecta_bro.evaluation.backtest import evaluate_day, load_day_data
DATA_DIR = Path('/home/brettanthonysjoberg179/trifecta-bro-hf-space/data')
RESULTS_DIR = DATA_DIR / 'results'
REPORTS_DIR = DATA_DIR / 'reports'
def run_with_weights(date: str, weights: dict, dry_run: bool = True) -> dict:
"""Run full analysis for a date with custom weights."""
from trifecta_bro.config import settings
from trifecta_bro.data.storage import Storage
from trifecta_bro.model.scoring import load_weights
from trifecta_bro.reporting.dashboard import write_outputs
from trifecta_bro.api.validator import validate_meetings, validate_race_form
client = FormFavClient()
storage = Storage()
meetings_payload = client.get_meetings(date)
meetings = [normalise_meeting(m) for m in meetings_payload
if m.get("country") == settings.country and not m.get("abandoned")]
analyses = []
for m in meetings:
for rn in m.races:
try:
form = client.get_race_form(date, m.slug, rn)
except FormFavNotFoundError:
continue
if form.get("abandoned"):
continue
race = normalise_race(form)
analysis = analyse_race(race, weights=weights)
analysis["date"] = date
analyses.append(analysis)
return analyses
def evaluate_with_weights(date: str, weights: dict) -> dict:
"""Run analysis and evaluate against actual results."""
# Get actual results
pred_path = REPORTS_DIR / f"predictions-{date}.json"
result_path = RESULTS_DIR / f"{date}-ra.json"
if not result_path.exists():
result_path = RESULTS_DIR / f"{date}.json"
if not result_path.exists():
return {"total": 0, "top1": 0, "exact": 0, "error": "no results"}
# Load results
with open(result_path) as f:
rdata = json.load(f)
# Build result lookup
result_lookup = {}
if isinstance(rdata, dict):
if 'tracks' in rdata:
for track, races in rdata['tracks'].items():
if isinstance(races, dict):
for rn, r in races.items():
runners = r.get('runners', [])
winners = []
for runner in runners:
pos = runner.get('position')
if pos and pos <= 3:
winners.append((pos, runner['number']))
winners.sort()
actual = [w[1] for w in winners]
if actual:
result_lookup[(track, int(rn))] = actual
elif isinstance(races, list):
for r in races:
if 'trifecta' in r:
result_lookup[(track, r.get('race', r.get('race_number', 0)))] = r['trifecta']
else:
# Direct track -> races format
for track, races in rdata.items():
if isinstance(races, list):
for r in races:
if 'trifecta' in r:
result_lookup[(track, r['race'])] = r['trifecta']
# Run analysis with new weights
try:
analyses = run_with_weights(date, weights, dry_run=True)
except Exception as e:
return {"total": 0, "top1": 0, "exact": 0, "error": str(e)}
# Evaluate
stats = {"total": 0, "top1": 0, "exact": 0, "box": 0, "top3": 0}
for analysis in analyses:
track = analysis.get("track", "")
race = analysis.get("race", 0)
actual = result_lookup.get((track, race))
if not actual or analysis.get("skipped"):
continue
pred_nums = analysis.get("trifecta", [])
stats["total"] += 1
if pred_nums == actual:
stats["exact"] += 1
if set(pred_nums) == set(actual):
stats["box"] += 1
if pred_nums and pred_nums[0] == actual[0]:
stats["top1"] += 1
stats["top3"] += len(set(pred_nums) & set(actual))
if stats["total"] > 0:
stats["top1_pct"] = stats["top1"] / stats["total"] * 100
stats["top3_pct"] = stats["top3"] / (stats["total"] * 3) * 100
return stats
def grid_search(dates: list[str], param_grid: dict, base_weights: dict, n_top: int = 10):
"""Grid search over weight combinations."""
import itertools
param_names = list(param_grid.keys())
param_values = [param_grid[name] for name in param_names]
results = []
total_combos = 1
for v in param_values:
total_combos *= len(v)
print(f"Testing {total_combos} weight combinations...")
for i, combo in enumerate(itertools.product(*param_values)):
weights = dict(base_weights)
for name, value in zip(param_names, combo):
weights[name] = value
# Normalize
total_w = sum(weights.values())
weights = {k: v / total_w for k, v in weights.items()}
# Evaluate across all dates
agg_stats = {"total": 0, "top1": 0, "exact": 0}
for date in dates:
stats = evaluate_with_weights(date, weights)
if "error" not in stats:
agg_stats["total"] += stats["total"]
agg_stats["top1"] += stats["top1"]
agg_stats["exact"] += stats["exact"]
top1_pct = agg_stats["top1"] / agg_stats["total"] * 100 if agg_stats["total"] > 0 else 0
results.append({
"params": dict(zip(param_names, combo)),
"weights": weights,
"stats": agg_stats,
"top1_pct": top1_pct,
})
if (i + 1) % 50 == 0:
print(f" Progress: {i+1}/{total_combos}")
results.sort(key=lambda x: x["top1_pct"], reverse=True)
return results[:n_top]
def main():
dates = ["2026-08-07", "2026-08-08", "2026-08-09", "2026-08-14"]
# Base v2.0 weights
base_weights = {
"form": 0.18, "class": 0.14, "distance": 0.10, "track": 0.10,
"track_distance": 0.10, "condition": 0.05, "jockey": 0.06,
"fitness": 0.08, "barrier": 0.08, "weight": 0.08, "pace": 0.03,
}
# Define search grid
param_grid = {
"form": [0.15, 0.20, 0.25],
"class": [0.10, 0.15],
"distance": [0.08, 0.12],
"track": [0.08, 0.12],
"track_distance": [0.08, 0.12, 0.15],
"fitness": [0.06, 0.10],
"barrier": [0.06, 0.10, 0.14],
"weight": [0.06, 0.10],
}
print("=" * 60)
print("GRID SEARCH OPTIMIZATION")
print("=" * 60)
top_results = grid_search(dates, param_grid, base_weights, n_top=5)
print("\n" + "=" * 60)
print("TOP 5 WEIGHT COMBINATIONS")
print("=" * 60)
for i, r in enumerate(top_results):
print(f"\n{i+1}. Top1: {r['top1_pct']:.1f}% ({r['stats']['top1']}/{r['stats']['total']})")
p = r['params']
print(f" form={p['form']:.2f}, class={p['class']:.2f}, dist={p['distance']:.2f}, "
f"track={p['track']:.2f}, track_dist={p['track_distance']:.2f}, "
f"fitness={p['fitness']:.2f}, barrier={p['barrier']:.2f}, weight={p['weight']:.2f}")
return top_results
if __name__ == "__main__":
main()
|