Spaces:
Configuration error
Configuration error
| """Backtesting harness for trifecta model optimization. | |
| Evaluates any weight profile against historical results with: | |
| - Leave-one-day-out cross-validation | |
| - Grid search over weight space | |
| - Statistical significance testing | |
| """ | |
| from __future__ import annotations | |
| import itertools | |
| import json | |
| import logging | |
| from pathlib import Path | |
| from typing import Optional | |
| logger = logging.getLogger("trifecta_bro.backtest") | |
| def evaluate_prediction(pred_nums: list[int], actual_nums: list[int]) -> dict: | |
| """Compare a prediction against actual results.""" | |
| if not pred_nums or not actual_nums: | |
| return {"exact": False, "box": False, "top1": False, "top3": 0} | |
| return { | |
| "exact": pred_nums == actual_nums, | |
| "box": set(pred_nums) == set(actual_nums), | |
| "top1": pred_nums[0] == actual_nums[0], | |
| "top3": len(set(pred_nums) & set(actual_nums)), | |
| } | |
| def evaluate_day( | |
| predictions: list[dict], | |
| results: list[dict], | |
| ) -> dict: | |
| """Evaluate all predictions for a single day.""" | |
| # Build result lookup | |
| result_lookup = {} | |
| for r in results: | |
| key = (r.get("track", ""), r.get("race", 0)) | |
| actual = r.get("trifecta", []) | |
| if actual: | |
| result_lookup[key] = actual | |
| stats = { | |
| "total": 0, | |
| "exact": 0, | |
| "box": 0, | |
| "top1": 0, | |
| "top3_total": 0, | |
| } | |
| for pred in predictions: | |
| track = pred.get("track", "") | |
| race = pred.get("race", 0) | |
| pred_nums = pred.get("trifecta", []) | |
| actual = result_lookup.get((track, race)) | |
| if not actual or pred.get("skipped"): | |
| continue | |
| stats["total"] += 1 | |
| result = evaluate_prediction(pred_nums, actual) | |
| stats["exact"] += result["exact"] | |
| stats["box"] += result["box"] | |
| stats["top1"] += result["top1"] | |
| stats["top3_total"] += result["top3"] | |
| if stats["total"] > 0: | |
| stats["top1_pct"] = stats["top1"] / stats["total"] * 100 | |
| stats["top3_pct"] = stats["top3_total"] / (stats["total"] * 3) * 100 | |
| return stats | |
| def load_day_data(date: str, data_dir: str = "/home/brettanthonysjoberg179/trifecta-bro-hf-space/data") -> tuple: | |
| """Load predictions and results for a single day.""" | |
| # Load predictions | |
| pred_path = Path(data_dir) / "reports" / f"predictions-{date}.json" | |
| if not pred_path.exists(): | |
| return [], [] | |
| with open(pred_path) as f: | |
| pred_data = json.load(f) | |
| predictions = pred_data.get("predictions", []) | |
| # Load results | |
| results = [] | |
| # Helper to parse runner numbers (handle "13e" emergency runners) | |
| def parse_num(s): | |
| return int(str(s).replace("e", "").replace("E", "")) | |
| # Try multiple result file formats | |
| result_path = Path(data_dir) / "results" / f"{date}.json" | |
| if result_path.exists(): | |
| with open(result_path) as f: | |
| rdata = json.load(f) | |
| if "tracks" in rdata: | |
| for track, races in rdata["tracks"].items(): | |
| if isinstance(races, list): | |
| for r in races: | |
| if "trifecta" in r: | |
| results.append({ | |
| "track": track, | |
| "race": r.get("race", r.get("race_number", 0)), | |
| "trifecta": [parse_num(x) for x in r["trifecta"]], | |
| }) | |
| elif isinstance(races, dict): | |
| for rn, r in races.items(): | |
| if "runners" in r: | |
| 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: | |
| results.append({ | |
| "track": track, | |
| "race": int(rn), | |
| "trifecta": actual, | |
| }) | |
| # Try alternate result file | |
| alt_path = Path(data_dir) / "results" / f"{date}-ra.json" | |
| if alt_path.exists() and not results: | |
| with open(alt_path) as f: | |
| rdata = json.load(f) | |
| for track, races in rdata.items(): | |
| for r in races: | |
| if "trifecta" in r: | |
| results.append({ | |
| "track": track, | |
| "race": r["race"], | |
| "trifecta": [parse_num(x) for x in r["trifecta"]], | |
| }) | |
| return predictions, results | |
| def cross_validate( | |
| dates: list[str], | |
| weight_profile: dict, | |
| ) -> dict: | |
| """Run leave-one-day-out cross-validation.""" | |
| all_predictions = {} | |
| all_results = {} | |
| for date in dates: | |
| predictions, results = load_day_data(date) | |
| if predictions and results: | |
| all_predictions[date] = predictions | |
| all_results[date] = results | |
| if not all_predictions: | |
| return {"error": "No data loaded"} | |
| # Evaluate each day | |
| day_stats = {} | |
| for date in sorted(all_predictions.keys()): | |
| stats = evaluate_day(all_predictions[date], all_results[date]) | |
| day_stats[date] = stats | |
| # Aggregate | |
| total_races = sum(s["total"] for s in day_stats.values()) | |
| total_top1 = sum(s["top1"] for s in day_stats.values()) | |
| total_exact = sum(s["exact"] for s in day_stats.values()) | |
| return { | |
| "dates": list(day_stats.keys()), | |
| "day_stats": day_stats, | |
| "total_races": total_races, | |
| "total_top1": total_top1, | |
| "total_exact": total_exact, | |
| "top1_pct": total_top1 / total_races * 100 if total_races > 0 else 0, | |
| "profile": weight_profile, | |
| } | |
| def grid_search( | |
| dates: list[str], | |
| param_grid: dict, | |
| base_profile: dict, | |
| ) -> list[dict]: | |
| """Grid search over weight combinations.""" | |
| results = [] | |
| # Generate all combinations | |
| param_names = list(param_grid.keys()) | |
| param_values = [param_grid[name] for name in param_names] | |
| for combo in itertools.product(*param_values): | |
| profile = dict(base_profile) | |
| for name, value in zip(param_names, combo): | |
| profile[name] = value | |
| # Normalize weights to sum to 1.0 | |
| total_w = sum(profile.values()) | |
| if total_w > 0: | |
| profile = {k: v / total_w for k, v in profile.items()} | |
| # Evaluate | |
| cv_result = cross_validate(dates, profile) | |
| cv_result["params"] = dict(zip(param_names, combo)) | |
| results.append(cv_result) | |
| # Sort by top1_pct | |
| results.sort(key=lambda x: x.get("top1_pct", 0), reverse=True) | |
| return results | |