"""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()