Spaces:
Configuration error
Configuration error
| """Configurable runner scoring engine. | |
| Each component yields a 0..100 sub-score. The final score is a weighted blend | |
| of the configured weights. When a component has insufficient data for a runner, | |
| its weight is redistributed across the components that DO have data (so a | |
| runner missing jockey/trainer stats isn't unfairly zeroed). This keeps the model | |
| transparent and degrades gracefully on incomplete FormFav data. | |
| v2.0 changes: | |
| - Distance-specific weight profiles (sprint/middle/staying) | |
| - Market component placeholder (SP odds when available via Betfair API) | |
| - track_distance component | |
| - fitness component (replaces trainer proxy) | |
| """ | |
| from __future__ import annotations | |
| import logging | |
| from pathlib import Path | |
| from typing import Optional | |
| import yaml | |
| from ..config import settings | |
| from ..data.models import RaceModel, RunnerModel | |
| from .feature_engine import form_quality_score | |
| from .pace_analysis import classify_pace, pace_adjustment | |
| from .distance_profiles import get_profile, parse_distance, classify_distance | |
| logger = logging.getLogger("trifecta_bro.scoring") | |
| # Base weights used as fallback | |
| DEFAULT_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, | |
| } | |
| def load_weights(path: Optional[Path] = None) -> dict: | |
| p = Path(path or settings.weights_path) | |
| if not p.exists(): | |
| return dict(DEFAULT_WEIGHTS) | |
| data = yaml.safe_load(p.read_text()) or {} | |
| w = data.get("weights", {}) | |
| merged = {**DEFAULT_WEIGHTS, **{k: float(v) for k, v in w.items()}} | |
| return merged | |
| def _safe_ratio(places: float, starts: float, cap: float, weight: float) -> float: | |
| if not starts: | |
| return 0.0 | |
| return min((places / starts) * 100 * (weight / cap), cap) | |
| def _form_component(r: RunnerModel) -> float: | |
| score, _ = form_quality_score(r.form) | |
| return score | |
| def _class_component(r: RunnerModel, race: RaceModel) -> float: | |
| wp = r.win_percent | |
| pp = r.place_percent | |
| return min(100.0, wp * 1.5 + pp * 0.5) | |
| def _distance_component(r: RunnerModel) -> float: | |
| d = r.stats.get("distance", {}) | |
| return _safe_ratio(d.get("places", 0), d.get("starts", 0), 100.0, 100.0) | |
| def _track_component(r: RunnerModel) -> float: | |
| t = r.stats.get("track", {}) | |
| return _safe_ratio(t.get("places", 0), t.get("starts", 0), 100.0, 100.0) | |
| def _track_distance_component(r: RunnerModel) -> float: | |
| """Use FormFav trackDistance stats — the most specific signal.""" | |
| td = r.stats.get("trackDistance", {}) | |
| if not td or not td.get("starts"): | |
| return 0.0 | |
| return _safe_ratio(td.get("places", 0), td.get("starts", 0), 100.0, 100.0) | |
| def _condition_component(r: RunnerModel, race: RaceModel) -> float: | |
| cond = (race.condition or "").lower() | |
| cond_stats = r.stats.get("conditions", {}) | |
| if not cond_stats: | |
| return 0.0 | |
| best = 0.0 | |
| for key, data in cond_stats.items(): | |
| if cond and cond[:4] in key.lower(): | |
| best = max(best, _safe_ratio(data.get("places", 0), data.get("starts", 0), 100.0, 100.0)) | |
| if best == 0.0: | |
| vals = [_safe_ratio(v.get("places", 0), v.get("starts", 0), 100.0, 100.0) for v in cond_stats.values()] | |
| best = sum(vals) / len(vals) if vals else 0.0 | |
| return best | |
| def _jockey_component(r: RunnerModel) -> float: | |
| # Form consistency (low variance = reliable) | |
| if not r.form: | |
| return 0.0 | |
| positions = [] | |
| for ch in r.form: | |
| if ch.isdigit(): | |
| val = int(ch) | |
| if val <= 9: | |
| positions.append(val) | |
| if len(positions) < 2: | |
| return 0.0 | |
| mean = sum(positions) / len(positions) | |
| variance = sum((p - mean) ** 2 for p in positions) / len(positions) | |
| stddev = variance ** 0.5 | |
| score = max(0.0, 100 - stddev * 30) | |
| return score | |
| def _fitness_component(r: RunnerModel) -> float: | |
| """Combines first-up and second-up strike rates as fitness proxy.""" | |
| first_up = r.stats.get("firstUp", {}) | |
| second_up = r.stats.get("secondUp", {}) | |
| fu_starts = first_up.get("starts", 0) or 0 | |
| fu_wins = first_up.get("wins", 0) or 0 | |
| su_starts = second_up.get("starts", 0) or 0 | |
| su_wins = second_up.get("wins", 0) or 0 | |
| total_starts = fu_starts + su_starts | |
| if total_starts == 0: | |
| return 0.0 | |
| fu_rate = (fu_wins / fu_starts) if fu_starts > 0 else 0 | |
| su_rate = (su_wins / su_starts) if su_starts > 0 else 0 | |
| if fu_starts > 0 and su_starts > 0: | |
| score = (fu_rate * 0.6 + su_rate * 0.4) * 100 | |
| elif fu_starts > 0: | |
| score = fu_rate * 100 | |
| else: | |
| score = su_rate * 100 | |
| return min(100.0, score) | |
| def _barrier_component(r: RunnerModel, race: RaceModel, pace: dict) -> float: | |
| barrier = r.barrier | |
| if barrier is None: | |
| return 0.0 | |
| n = max(race.number_of_runners, 8) | |
| distance = race.distance or "" | |
| dist_num = 1200 | |
| try: | |
| dist_num = int(''.join(filter(str.isdigit, distance))) if distance else 1200 | |
| except: | |
| pass | |
| # Distance-aware barrier logic | |
| if dist_num <= 1200: | |
| spread = abs(barrier - (n / 2.0 + 0.5)) | |
| base = max(0.0, 100.0 - spread * 12.0) | |
| elif dist_num <= 1600: | |
| spread = abs(barrier - (n / 2.0 + 0.5)) | |
| base = max(0.0, 100.0 - spread * 8.0) | |
| else: | |
| spread = abs(barrier - (n / 2.0 + 0.5)) | |
| base = max(0.0, 100.0 - spread * 5.0) | |
| role = pace["roles"].get(r.number, "midfield") | |
| if pace["label"] == "SLOW": | |
| base += (5 - barrier) * 1.2 if barrier <= 5 else 0.0 | |
| elif pace["label"] in ("VERY FAST", "FAST"): | |
| base += (barrier - 5) * 0.8 if barrier > 5 else 0.0 | |
| return max(0.0, min(100.0, base)) | |
| def _weight_component(r: RunnerModel, race: RaceModel) -> float: | |
| if r.weight is None: | |
| return 0.0 | |
| w = r.weight | |
| score = 40 + (w - 52) * 4.0 | |
| return max(0.0, min(100.0, score)) | |
| def _market_component(r: RunnerModel) -> float: | |
| """Market component — placeholder for SP odds. | |
| Returns 0 until Betfair API key is configured. | |
| When available, this will fetch SP odds and convert to implied probability. | |
| """ | |
| return 0.0 | |
| def score_runner(r: RunnerModel, race: RaceModel, pace: dict, weights: dict | None = None) -> dict: | |
| """Score a runner using distance-specific weights if no weights provided.""" | |
| if weights is None: | |
| weights = get_profile(race) | |
| comp = { | |
| "form": _form_component(r), | |
| "class": _class_component(r, race), | |
| "distance": _distance_component(r), | |
| "track": _track_component(r), | |
| "track_distance": _track_distance_component(r), | |
| "condition": _condition_component(r, race), | |
| "jockey": _jockey_component(r), | |
| "fitness": _fitness_component(r), | |
| "barrier": _barrier_component(r, race, pace), | |
| "weight": _weight_component(r, race), | |
| "pace": 50.0 + pace_adjustment(r, pace), | |
| "market": _market_component(r), | |
| } | |
| available = {} | |
| total_w = 0.0 | |
| for k, w in weights.items(): | |
| if comp[k] > 0 or k in ("form", "pace"): | |
| available[k] = w | |
| total_w += w | |
| if total_w == 0: | |
| return {"sub_scores": comp, "score": 0.0, "available": {}, "completeness": r.data_completeness} | |
| norm = {k: w / total_w for k, w in available.items()} | |
| score = sum(comp[k] * norm[k] for k in norm) | |
| return { | |
| "sub_scores": comp, | |
| "score": round(score, 1), | |
| "available": {k: round(v, 3) for k, v in norm.items()}, | |
| "completeness": r.data_completeness, | |
| } | |