trifecta_bro_v1/predictor.py

"""Trifecta-Bro v1 — open-source multi-factor trifecta predictor.

Model id: Brettapps/trifecta-bro/v1  (HF repo: Brettapps/trifecta-bro-v1)

This is the standalone inference module for the HuggingFace model repo. It is a
self-contained, dependency-light implementation of the trifecta prediction logic
(rule-based multi-factor scoring). It mirrors `src/prediction_model.py` in the
Trifecta-Bro Space so the published model is directly usable without cloning the
whole Space.

Load it:
    from trifecta_bro_v1.predictor import TrifectaPredictor, Race, Runner
    pred = TrifectaPredictor().predict(race)
"""

from __future__ import annotations

from dataclasses import dataclass, field
from typing import Any


@dataclass
class Runner:
    number: int
    name: str
    jockey: str = ""
    trainer: str = ""
    weight: float | None = None
    barrier: int | None = None
    form: str = ""
    last20Starts: str = ""
    careerPrizeMoney: str = "$0"
    scratched: bool = False
    stats: dict[str, Any] = field(default_factory=dict)


@dataclass
class Race:
    date: str
    track: str
    track_slug: str
    race_number: str
    race_name: str
    distance: str
    condition: str
    weather: str
    race_class: str
    start_time: str
    prize_money: str
    number_of_runners: int
    runners: list[Runner] = field(default_factory=list)


class TrifectaPredictor:
    """Open-source trifecta prediction model.

    Scores each runner on a 0-100 scale using recent form, career/overall
    win & place percentages, track/distance/condition strike rates, barrier
    draw, and career prize money. The top three by score form the primary
    trifecta; secondary and value bets are derived from the next-best runners.
    """

    MODEL_ID = "Brettapps/trifecta-bro/v1"

    def predict(self, race: Race) -> dict[str, Any]:
        runners = [r for r in race.runners if not r.scratched]
        if len(runners) < 3:
            return {"error": "Insufficient runners"}

        scored: list[dict[str, Any]] = []
        for runner in runners:
            scored.append({
                "number": runner.number,
                "name": runner.name,
                "score": self._score_runner(runner),
                "win_prob": self._win_probability(runner),
                "place_prob": self._place_probability(runner),
            })

        scored.sort(key=lambda x: x["score"], reverse=True)
        top = scored[:3]
        primary = f"{top[0]['number']}-{top[1]['number']}-{top[2]['number']}"

        secondary = None
        value = None
        if len(scored) > 3:
            secondary = f"{top[0]['number']}-{top[2]['number']}-{scored[3]['number']}"
            outsiders = [s for s in scored[3:] if s["score"] > 30]
            if outsiders:
                value = f"{scored[1]['number']}-{top[0]['number']}-{outsiders[0]['number']}"
            else:
                value = f"{scored[1]['number']}-{top[0]['number']}-{top[2]['number']}"

        return {
            "model_id": self.MODEL_ID,
            "date": race.date,
            "track": race.track,
            "race_number": race.race_number,
            "race_name": race.race_name,
            "primary": primary,
            "secondary": secondary,
            "value": value,
            "top3": top,
            "confidence": "MEDIUM",
        }

    def _score_runner(self, runner: Runner) -> float:
        score = 0.0
        form = str(runner.form or runner.last20Starts or "")
        recent = form[-5:] if len(form) > 5 else form
        score += min((recent.count("1") * 8 + recent.count("2") * 4 + recent.count("3") * 4), 25)

        overall = runner.stats.get("overall", {})
        win_pct = overall.get("winPercent", 0) or 0
        place_pct = overall.get("placePercent", 0) or 0
        score += win_pct * 20
        score += place_pct * 10

        track_stats = runner.stats.get("track", {})
        track_starts = track_stats.get("starts", 0) or 0
        track_places = track_stats.get("places", 0) or 0
        score += min((track_places / max(track_starts, 1)) * 10, 10)

        dist_stats = runner.stats.get("distance", {})
        dist_starts = dist_stats.get("starts", 0) or 0
        dist_places = dist_stats.get("places", 0) or 0
        score += min((dist_places / max(dist_starts, 1)) * 8, 8)

        cond_stats = runner.stats.get("conditions", {})
        for _key, data in cond_stats.items():
            c_starts = data.get("starts", 0) or 0
            c_places = data.get("places", 0) or 0
            score += min((c_places / max(c_starts, 1)) * 8, 8)

        try:
            barrier = int(runner.barrier) if runner.barrier else 5
            score += max(0, 5 - abs(barrier - 5))
        except Exception:
            score += 3

        try:
            prize = float(str(runner.careerPrizeMoney).replace("$", "").replace(",", ""))
            score += min(prize / 20000, 5)
        except Exception:
            pass

        return min(round(score, 1), 100)

    def _win_probability(self, runner: Runner) -> float:
        overall = runner.stats.get("overall", {})
        win_pct = overall.get("winPercent", 0) or 0
        return min(max(round(win_pct * 100, 1), 0), 100)

    def _place_probability(self, runner: Runner) -> float:
        overall = runner.stats.get("overall", {})
        place_pct = overall.get("placePercent", 0) or 0
        return min(max(round(place_pct * 100, 1), 0), 100)