"""Re-score cached form data with different weights to find optimal profile.""" 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.data.models import RaceModel, RunnerModel, MeetingModel from trifecta_bro.model.scoring import score_runner from trifecta_bro.model.pace_analysis import classify_pace from trifecta_bro.model.distance_profiles import classify_distance DATA_DIR = Path('/home/brettanthonysjoberg179/trifecta-bro-hf-space/data') CACHE_DIR = DATA_DIR / 'cache' RESULTS_DIR = DATA_DIR / 'results' REPORTS_DIR = DATA_DIR / 'reports' def load_cached_form(date: str, track: str, race: str) -> dict | None: """Load cached FormFav form data.""" # Try multiple cache file patterns for f in CACHE_DIR.glob('*.json'): try: with open(f) as fp: data = json.load(fp) if data.get('date') == date and data.get('track', '').lower() == track.lower() and str(data.get('raceNumber')) == str(race): return data except: continue return None def re_score_race(weights: dict, race_data: dict) -> dict | None: """Re-score a race with new weights using cached form data.""" try: # Build RunnerModel objects runners = [] for r in race_data.get('runners', []): if r.get('scratched'): continue runner = RunnerModel( number=r['number'], name=r.get('name', ''), jockey=r.get('jockey'), trainer=r.get('trainer'), weight=r.get('weight'), barrier=r.get('barrier'), age=r.get('age'), sex=r.get('sex'), form=r.get('form', ''), last20_starts=r.get('last20Starts', ''), stats=r.get('stats', {}), ) runners.append(runner) if len(runners) < 3: return None # Build RaceModel race = RaceModel( date=race_data.get('date', ''), track=race_data.get('track', ''), track_slug=race_data.get('slug', ''), race_number=race_data.get('raceNumber', 0), race_name=race_data.get('raceName', ''), distance=race_data.get('distance'), condition=race_data.get('condition'), race_class=race_data.get('raceClass'), abandoned=race_data.get('abandoned', False), start_time=race_data.get('startTime'), prize_money=str(race_data.get('prizeMoney', '')), number_of_runners=race_data.get('numberOfRunners', len(runners)), runners=runners, ) # Classify pace pace = classify_pace(race, runners) # Score runners scored = [] for r in runners: sc = score_runner(r, race, pace, weights) scored.append({ 'number': r.number, 'name': r.name, 'score': sc['score'], }) # Sort by score scored.sort(key=lambda x: x['score'], reverse=True) # Generate trifecta trifecta = [s['number'] for s in scored[:3]] return { 'track': race_data.get('track', ''), 'race': race_data.get('raceNumber', 0), 'trifecta': trifecta, 'runners': scored, } except Exception as e: return None def evaluate_date(date: str, weights: dict) -> dict: """Evaluate a single date with given weights.""" # Load results 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"} 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: 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'] # Find all cached form files for this date stats = {"total": 0, "top1": 0, "exact": 0, "box": 0, "top3": 0} for cache_file in CACHE_DIR.glob('*.json'): try: with open(cache_file) as f: data = json.load(f) if data.get('date') != date: continue track = data.get('track', '') race_num = data.get('raceNumber', 0) # Check if we have results for this race actual = result_lookup.get((track, race_num)) if not actual: continue # Re-score with new weights result = re_score_race(weights, data) if not result: continue pred_nums = result['trifecta'] stats["total"] += 1 if pred_nums == actual: stats["exact"] += 1 if set(pred_nums) == set(actual): stats["box"] += 1 if pred_nums[0] == actual[0]: stats["top1"] += 1 stats["top3"] += len(set(pred_nums) & set(actual)) except Exception as e: continue 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): """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_date(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) % 100 == 0: print(f" Progress: {i+1}/{total_combos}") results.sort(key=lambda x: x["top1_pct"], reverse=True) return results 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, 0.30], "class": [0.10, 0.15, 0.20], "distance": [0.06, 0.10, 0.14], "track": [0.06, 0.10, 0.14], "track_distance": [0.06, 0.10, 0.14], "fitness": [0.04, 0.08, 0.12], "barrier": [0.06, 0.10, 0.14], "weight": [0.04, 0.08, 0.12], "pace": [0.02, 0.05, 0.08], } # Fix condition and jockey to reduce search space fixed = {"condition": 0.05, "jockey": 0.06} print("=" * 60) print("GRID SEARCH OPTIMIZATION (v2.0)") print("=" * 60) results = grid_search(dates, param_grid, {**base_weights, **fixed}) print("\n" + "=" * 60) print("TOP 10 WEIGHT COMBINATIONS") print("=" * 60) for i, r in enumerate(results[:10]): 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}, " f"weight={p['weight']:.2f}, pace={p['pace']:.2f}") # Best weights if results: best = results[0] print("\n" + "=" * 60) print("BEST WEIGHTS FOUND") print("=" * 60) print(f"Top 1: {best['top1_pct']:.1f}% ({best['stats']['top1']}/{best['stats']['total']})") # Per-date breakdown print("\nPer-date performance:") for date in dates: stats = evaluate_date(date, best['weights']) if stats['total'] > 0: print(f" {date}: {stats['total']} races, {stats['top1']} top1 ({stats['top1']/stats['total']*100:.0f}%)") return results if __name__ == "__main__": main()