Spaces:
Configuration error
Configuration error
File size: 3,713 Bytes
e23172f | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 | """Normalise raw FormFav payloads into internal models.
Key responsibilities:
- drop scratched runners from the active set (but keep record)
- mark abandoned races
- never invent missing values; track data_completeness instead
- tolerate missing `form` (fall back to `last20Starts`)
"""
from __future__ import annotations
import logging
from typing import Any
from .models import RaceModel, RunnerModel, MeetingModel
logger = logging.getLogger("trifecta_bro.normalizer")
EXPECTED_RUNNER_FIELDS = [
"name", "jockey", "trainer", "weight", "barrier", "age",
"careerPrizeMoney", "form", "stats",
]
def _to_float(value: Any) -> float | None:
if value is None:
return None
try:
return float(str(value).replace("$", "").replace(",", ""))
except (ValueError, TypeError):
return None
def _normalise_runner(raw: dict[str, Any]) -> RunnerModel:
present = sum(1 for f in EXPECTED_RUNNER_FIELDS if raw.get(f) not in (None, "", {}))
completeness = present / len(EXPECTED_RUNNER_FIELDS)
return RunnerModel(
number=int(raw.get("number", 0)),
name=raw.get("name", "") or "",
jockey=raw.get("jockey"),
trainer=raw.get("trainer"),
weight=_to_float(raw.get("weight")),
barrier=int(raw["barrier"]) if raw.get("barrier") not in (None, "") else None,
age=int(raw["age"]) if raw.get("age") not in (None, "") else None,
sex=raw.get("sex"),
career_prize_money=_to_float(raw.get("careerPrizeMoney")),
form=raw.get("form") or raw.get("last20Starts") or "",
last20_starts=raw.get("last20Starts") or raw.get("form") or "",
scratched=bool(raw.get("scratched", False)),
stats=raw.get("stats", {}) or {},
data_completeness=round(completeness, 3),
)
def normalise_race(payload: dict[str, Any]) -> RaceModel:
warnings: list[str] = []
raw_runners = payload.get("runners", []) or []
runners = [_normalise_runner(r) for r in raw_runners]
active = [r for r in runners if not r.scratched]
if len(active) < 3:
warnings.append(f"only {len(active)} active runners after scratches")
# distance may be like "1200m"
dist = payload.get("distance")
nrun = payload.get("numberOfRunners")
if nrun is None:
nrun = len(active)
return RaceModel(
date=payload.get("date", ""),
track=payload.get("track", ""),
track_slug=payload.get("trackSlug") or payload.get("slug") or payload.get("track", "").lower(),
race_number=int(payload.get("raceNumber", payload.get("race", 0))),
race_name=payload.get("raceName", "") or "",
distance=str(dist) if dist is not None else None,
condition=payload.get("condition"),
weather=payload.get("weather"),
race_class=payload.get("raceClass"),
abandoned=bool(payload.get("abandoned", False)),
start_time=payload.get("startTime"),
timezone=payload.get("timezone"),
prize_money=str(payload.get("prizeMoney")) if payload.get("prizeMoney") is not None else None,
number_of_runners=int(nrun) if nrun is not None else 0,
runners=runners,
source_warnings=warnings,
)
def normalise_meeting(raw: dict[str, Any]) -> MeetingModel:
races = []
for r in raw.get("races", []) or []:
rn = r.get("raceNumber") if isinstance(r, dict) else None
if rn is not None:
races.append(int(rn))
return MeetingModel(
track=raw.get("track", ""),
slug=raw.get("slug", "") or raw.get("track", "").lower(),
country=raw.get("country", ""),
abandoned=bool(raw.get("abandoned", False)),
races=sorted(set(races)),
)
|