Spaces:
Configuration error
Configuration error
| """Optimize with distance-specific profiles.""" | |
| import json | |
| from pathlib import Path | |
| import sys | |
| sys.path.insert(0, '/home/brettanthonysjoberg179/trifecta-bro-hf-space') | |
| from trifecta_bro.data.models import RaceModel, RunnerModel | |
| 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, parse_distance | |
| DATA_DIR = Path('/home/brettanthonysjoberg179/trifecta-bro-hf-space/data') | |
| CACHE_DIR = DATA_DIR / 'cache' | |
| RESULTS_DIR = DATA_DIR / 'results' | |
| def load_cached_races(date: str) -> list[dict]: | |
| races = [] | |
| for f in CACHE_DIR.glob('*.json'): | |
| try: | |
| with open(f) as fp: | |
| data = json.load(fp) | |
| if data.get('date') == date: | |
| races.append(data) | |
| except: | |
| continue | |
| return races | |
| def load_results(date: str) -> dict: | |
| result_lookup = {} | |
| 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 result_lookup | |
| with open(result_path) as f: | |
| rdata = json.load(f) | |
| 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', 0))] = [int(str(x).replace('e','')) for x in 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'])] = [int(str(x).replace('e','')) for x in r['trifecta']] | |
| return result_lookup | |
| def score_race(race_data: dict, weights_by_dist: dict) -> list[int] | None: | |
| try: | |
| 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 | |
| distance = race_data.get('distance', '1200m') | |
| dist_class = classify_distance(distance) | |
| weights = weights_by_dist.get(dist_class, weights_by_dist['middle']) | |
| race = RaceModel( | |
| date=race_data.get('date', ''), track=race_data.get('track', ''), | |
| track_slug=race_data.get('slug', race_data.get('track', '').lower()), | |
| race_number=race_data.get('raceNumber', 0), | |
| race_name=race_data.get('raceName', ''), | |
| distance=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, | |
| ) | |
| pace = classify_pace(race, runners) | |
| scored = [] | |
| for r in runners: | |
| sc = score_runner(r, race, pace, weights) | |
| scored.append((r.number, sc['score'])) | |
| scored.sort(key=lambda x: x[1], reverse=True) | |
| return [s[0] for s in scored[:3]] | |
| except: | |
| return None | |
| def evaluate_date(date: str, weights_by_dist: dict) -> dict: | |
| result_lookup = load_results(date) | |
| races = load_cached_races(date) | |
| stats = {"total": 0, "top1": 0, "exact": 0, "box": 0} | |
| for race_data in races: | |
| track = race_data.get('track', '') | |
| race_num = race_data.get('raceNumber', 0) | |
| actual = result_lookup.get((track, race_num)) | |
| if not actual: | |
| continue | |
| pred = score_race(race_data, weights_by_dist) | |
| if not pred: | |
| continue | |
| stats["total"] += 1 | |
| if pred == actual: | |
| stats["exact"] += 1 | |
| if set(pred) == set(actual): | |
| stats["box"] += 1 | |
| if pred[0] == actual[0]: | |
| stats["top1"] += 1 | |
| if stats["total"] > 0: | |
| stats["top1_pct"] = stats["top1"] / stats["total"] * 100 | |
| return stats | |
| def evaluate_all(dates: list[str], weights_by_dist: dict) -> dict: | |
| agg = {"total": 0, "top1": 0, "exact": 0, "box": 0} | |
| for date in dates: | |
| stats = evaluate_date(date, weights_by_dist) | |
| agg["total"] += stats["total"] | |
| agg["top1"] += stats["top1"] | |
| agg["exact"] += stats["exact"] | |
| agg["box"] += stats["box"] | |
| agg["top1_pct"] = agg["top1"] / agg["total"] * 100 if agg["total"] > 0 else 0 | |
| return agg | |
| def main(): | |
| dates = ["2026-08-07", "2026-08-08", "2026-08-09", "2026-08-14"] | |
| # Distance-specific profiles | |
| profiles = { | |
| "sprint": { | |
| "form": 0.15, "class": 0.10, "distance": 0.06, "track": 0.06, | |
| "track_distance": 0.08, "condition": 0.04, "jockey": 0.04, | |
| "fitness": 0.04, "barrier": 0.14, "weight": 0.08, "pace": 0.10, | |
| "market": 0.11, | |
| }, | |
| "middle": { | |
| "form": 0.20, "class": 0.13, "distance": 0.08, "track": 0.08, | |
| "track_distance": 0.08, "condition": 0.05, "jockey": 0.06, | |
| "fitness": 0.05, "barrier": 0.13, "weight": 0.05, "pace": 0.04, | |
| "market": 0.05, | |
| }, | |
| "staying": { | |
| "form": 0.14, "class": 0.14, "distance": 0.10, "track": 0.06, | |
| "track_distance": 0.08, "condition": 0.06, "jockey": 0.08, | |
| "fitness": 0.10, "barrier": 0.04, "weight": 0.06, "pace": 0.04, | |
| "market": 0.10, | |
| }, | |
| } | |
| print("=" * 60) | |
| print("DISTANCE-SPECIFIC PROFILES") | |
| print("=" * 60) | |
| # Evaluate with distance-specific profiles | |
| agg = evaluate_all(dates, profiles) | |
| print(f"\nDistance-specific: {agg['top1_pct']:.1f}% ({agg['top1']}/{agg['total']})") | |
| # Per-date breakdown | |
| for date in dates: | |
| stats = evaluate_date(date, profiles) | |
| if stats['total'] > 0: | |
| print(f" {date}: {stats['top1']}/{stats['total']} ({stats['top1_pct']:.1f}%)") | |
| # Compare with flat weights | |
| flat_weights = { | |
| "form": 0.20, "class": 0.13, "distance": 0.08, "track": 0.08, | |
| "track_distance": 0.08, "condition": 0.07, "jockey": 0.08, | |
| "fitness": 0.05, "barrier": 0.13, "weight": 0.05, "pace": 0.04, | |
| } | |
| flat_profiles = {"sprint": flat_weights, "middle": flat_weights, "staying": flat_weights} | |
| flat_agg = evaluate_all(dates, flat_profiles) | |
| print(f"\nFlat weights: {flat_agg['top1_pct']:.1f}% ({flat_agg['top1']}/{flat_agg['total']})") | |
| # Per-distance breakdown | |
| print("\nPer-distance breakdown:") | |
| for dist_class in ["sprint", "middle", "staying"]: | |
| dist_stats = {"total": 0, "top1": 0} | |
| for date in dates: | |
| result_lookup = load_results(date) | |
| races = load_cached_races(date) | |
| for race_data in races: | |
| distance = race_data.get('distance', '1200m') | |
| if classify_distance(distance) == dist_class: | |
| track = race_data.get('track', '') | |
| race_num = race_data.get('raceNumber', 0) | |
| actual = result_lookup.get((track, race_num)) | |
| if not actual: | |
| continue | |
| pred = score_race(race_data, profiles) | |
| if not pred: | |
| continue | |
| dist_stats["total"] += 1 | |
| if pred[0] == actual[0]: | |
| dist_stats["top1"] += 1 | |
| if dist_stats["total"] > 0: | |
| print(f" {dist_class}: {dist_stats['top1']}/{dist_stats['total']} ({dist_stats['top1']/dist_stats['total']*100:.1f}%)") | |
| return profiles | |
| if __name__ == "__main__": | |
| main() | |