Spaces:
Configuration error
Configuration error
| """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)), | |
| ) | |