"""Repository layer for database operations with idempotent upserts. Supports TAB Affiliates API data structures. """ import copy import hashlib from datetime import UTC, date, datetime from typing import Any from sqlalchemy import and_, desc, func, or_ from sqlalchemy.dialects.postgresql import insert from sqlalchemy.orm import Session, joinedload from packages.core.common.cache import cache_get, cache_set from packages.core.common.logging import get_logger from packages.core.storage.audit import AuditLogger from packages.core.storage.models import ( BarrierAdjustment, Driver, EntityType, HandicapAdjustment, Horse, Meeting, Race, RatingSnapshot, Starter, Trainer, ) logger = get_logger(__name__) _HRNZ_ID_MODULUS = 2147483647 def normalize_entity_id(value: Any, fallback_name: str | None = None) -> int | None: """Normalize entity identifiers across TAB/HRNZ sources.""" if value is None and not fallback_name: return None if isinstance(value, int): return value if isinstance(value, str): stripped = value.strip() if stripped.isdigit(): return int(stripped) if stripped: return ( int(hashlib.md5(stripped.encode()).hexdigest()[:8], 16) % _HRNZ_ID_MODULUS ) if fallback_name: return ( int(hashlib.md5(fallback_name.encode()).hexdigest()[:8], 16) % _HRNZ_ID_MODULUS ) return None def normalize_runner_data(runner_data: dict[str, Any]) -> dict[str, Any]: """Normalize runner payloads from TAB/HRNZ into a consistent shape.""" data = dict(runner_data or {}) horse_name = ( data.get("name") or data.get("horse_name") or data.get("horse") or "Unknown" ) data["name"] = horse_name data["horse_name"] = horse_name horse_id = data.get("horse_id") or data.get("horseId") or data.get("horse_uuid") data["horse_id"] = normalize_entity_id(horse_id, fallback_name=horse_name) driver_name = data.get("jockey") or data.get("driver_name") or data.get("driver") if driver_name: data["driver_name"] = driver_name data["jockey"] = driver_name trainer_name = data.get("trainer_name") or data.get("trainer") if trainer_name: data["trainer_name"] = trainer_name driver_id = data.get("driver_id") or data.get("driverId") or data.get("driver_uuid") data["driver_id"] = normalize_entity_id(driver_id, fallback_name=driver_name) trainer_id = ( data.get("trainer_id") or data.get("trainerId") or data.get("trainer_uuid") ) data["trainer_id"] = normalize_entity_id(trainer_id, fallback_name=trainer_name) if "runner_number" not in data: data["runner_number"] = data.get("number") or data.get("runnerNo") if "barrier" not in data: data["barrier"] = data.get("barrier_draw") if "handicap_m" not in data: data["handicap_m"] = data.get("handicap") if "handicap" not in data: data["handicap"] = data.get("handicap_m") placing_value = data.get("placing") if isinstance(placing_value, str) and placing_value.strip().isdigit(): data["placing"] = int(placing_value.strip()) for key in ( "barrier_position", "margin", "race_time", "placing", "barrier", "handicap_m", "runner_number", "driver_name", "trainer_name", ): data.setdefault(key, None) return data def normalize_race_data(race_data: dict[str, Any]) -> dict[str, Any]: """Normalize race payloads to ensure consistent nested structure.""" data = copy.deepcopy(race_data or {}) if "distance_m" not in data and "distance" in data: data["distance_m"] = data.get("distance") nested = data.get("raw_json") if isinstance(nested, dict): if "distance_m" not in nested and data.get("distance_m") is not None: nested["distance_m"] = data.get("distance_m") if "race_number" not in nested and data.get("race_number") is not None: nested["race_number"] = data.get("race_number") starters = nested.get("starters") if isinstance(starters, list): nested["starters"] = [ normalize_runner_data(starter) if isinstance(starter, dict) else starter for starter in starters ] data["raw_json"] = nested return data def generate_driver_id(name: str) -> int: """Generate deterministic integer ID from driver name. Uses SHA256 hash truncated to 31 bits (positive integer). Collision risk: ~0.0001% at 100,000 drivers. Args: name: Driver name Returns: Positive integer ID """ if not name: raise ValueError("Cannot generate ID from empty name") normalized = name.strip().upper() hash_bytes = hashlib.sha256(normalized.encode("utf-8")).digest() return int.from_bytes(hash_bytes[:4], "big") & 0x7FFFFFFF def generate_trainer_id(name: str) -> int: """Generate deterministic integer ID from trainer name. Uses SHA256 hash truncated to 31 bits (positive integer). Args: name: Trainer name Returns: Positive integer ID """ if not name: raise ValueError("Cannot generate ID from empty name") normalized = name.strip().upper() hash_bytes = hashlib.sha256(normalized.encode("utf-8")).digest() return int.from_bytes(hash_bytes[:4], "big") & 0x7FFFFFFF def parse_tab_datetime(dt_string: str | None) -> datetime | None: """Parse TAB API datetime string. Args: dt_string: ISO datetime string (e.g., "2024-01-15T14:30:00+13:00") Returns: datetime object or None if parsing fails """ if not dt_string: return None try: # Try ISO format with timezone return datetime.fromisoformat(dt_string.replace("Z", "+00:00")) except ValueError: try: # Try without timezone return datetime.fromisoformat(dt_string[:19]) except ValueError: logger.warning(f"Failed to parse datetime: {dt_string}") return None def parse_tab_date(date_string: str | None) -> date | None: """Parse TAB API date from datetime string or date string. Args: date_string: Date string (e.g., "2024-01-15" or "20240115" or full datetime) Returns: date object or None if parsing fails """ if not date_string: return None try: # Try ISO date format YYYY-MM-DD if len(date_string) >= 10 and "-" in date_string: return date.fromisoformat(date_string[:10]) # Try YYYYMMDD format elif len(date_string) >= 8 and date_string[:8].isdigit(): return datetime.strptime(date_string[:8], "%Y%m%d").date() else: return None except ValueError: logger.warning(f"Failed to parse date: {date_string}") return None class MeetingRepository: """Repository for meeting operations.""" @staticmethod def upsert(session: Session, meeting_data: dict[str, Any]) -> Meeting: """Upsert meeting from TAB API data. Args: session: Database session meeting_data: Raw meeting data from TAB API Returns: Meeting instance """ # TAB API uses "meeting" as the ID field (string) meeting_id = meeting_data.get("meeting") if not meeting_id: raise ValueError("Meeting data missing 'meeting' ID") # Parse meeting date from various formats meeting_date_raw = meeting_data.get("date") or meeting_data.get("meeting_date") meeting_date = parse_tab_date(meeting_date_raw) if not meeting_date: raise ValueError(f"Could not parse meeting date: {meeting_date_raw}") # Use meeting name as venue venue = meeting_data.get("name", "Unknown") # Category: T (Thoroughbred), H (Harness), G (Greyhound) category = meeting_data.get("category", "H") # Upsert using PostgreSQL INSERT ... ON CONFLICT stmt = insert(Meeting).values( id=meeting_id, meeting_date=meeting_date, venue=venue, category=category, raw_json=meeting_data, ) stmt = stmt.on_conflict_do_update( index_elements=["id"], set_={ "meeting_date": stmt.excluded.meeting_date, "venue": stmt.excluded.venue, "category": stmt.excluded.category, "raw_json": stmt.excluded.raw_json, "updated_at": stmt.excluded.updated_at, }, ) session.execute(stmt) session.flush() meeting = session.query(Meeting).filter(Meeting.id == meeting_id).one() logger.debug( f"Upserted meeting {meeting_id} ({venue}, {meeting_date}, {category})" ) return meeting @staticmethod def get_by_id(session: Session, meeting_id: str) -> Meeting | None: """Get meeting by ID. Args: session: Database session meeting_id: Meeting ID (string) Returns: Meeting or None """ return session.query(Meeting).filter(Meeting.id == meeting_id).first() @staticmethod def get_by_date_range( session: Session, date_from: date, date_to: date ) -> list[Meeting]: """Get meetings in date range. Args: session: Database session date_from: Start date (inclusive) date_to: End date (inclusive) Returns: List of meetings """ return ( session.query(Meeting) .filter(Meeting.meeting_date >= date_from, Meeting.meeting_date <= date_to) .order_by(Meeting.meeting_date) .all() ) class RaceRepository: """Repository for race operations.""" @staticmethod def upsert(session: Session, meeting_id: str, race_data: dict[str, Any]) -> Race: """Upsert race from TAB API data. Args: session: Database session meeting_id: Parent meeting ID (string) race_data: Raw race data from TAB API (from EventRaceDetails.race) Returns: Race instance """ race_data = normalize_race_data(race_data) race_number = race_data.get("race_number") if not race_number: raise ValueError("Race data missing race_number") # TAB event ID (string) tab_event_id = race_data.get("event_id") # TAB API returns distance as integer (metres) distance_m = race_data.get("distance") # Start type (Mobile, Standing for harness) start_type = race_data.get("start_type") # Gait (Pace, Trot for harness) gait = race_data.get("gait") weather = race_data.get("weather") track_condition = race_data.get("track_condition") # Parse race datetime from available fields race_datetime = RaceRepository._parse_race_datetime(race_data) # Find existing race by composite key existing = ( session.query(Race) .filter(Race.meeting_id == meeting_id, Race.race_number == race_number) .first() ) if existing: # Update existing existing.tab_event_id = tab_event_id existing.distance_m = distance_m existing.start_type = start_type existing.gait = gait existing.weather = weather existing.track_condition = track_condition existing.race_datetime = race_datetime existing.raw_json = race_data session.flush() logger.debug(f"Updated race {meeting_id}/{race_number}") return existing else: # Insert new race = Race( meeting_id=meeting_id, tab_event_id=tab_event_id, race_number=race_number, distance_m=distance_m, start_type=start_type, gait=gait, weather=weather, track_condition=track_condition, race_datetime=race_datetime, raw_json=race_data, ) session.add(race) session.flush() logger.debug( f"Inserted race {meeting_id}/{race_number} (event_id={tab_event_id})" ) return race @staticmethod def _parse_race_datetime(race_data: dict[str, Any]) -> datetime | None: """Parse race datetime from known TAB/HRNZ fields.""" candidates = [ race_data.get("advertised_start_string"), race_data.get("race_datetime"), race_data.get("start_time"), race_data.get("start_datetime"), race_data.get("advertised_start_time"), ] raw_json = race_data.get("raw_json") if isinstance(raw_json, dict): candidates.append(raw_json.get("race_datetime")) for value in candidates: if isinstance(value, datetime): return value if isinstance(value, str) and value.strip(): parsed = parse_tab_datetime(value) if parsed: return parsed advertised_start = race_data.get("advertised_start") if isinstance(advertised_start, dict): for key in ("seconds", "secs", "time"): if key in advertised_start: try: timestamp = int(advertised_start[key]) except (TypeError, ValueError): continue return datetime.fromtimestamp(timestamp, tz=UTC) return None @staticmethod def get_by_meeting(session: Session, meeting_id: str) -> list[Race]: """Get all races for a meeting. Args: session: Database session meeting_id: Meeting ID (string) Returns: List of races """ return ( session.query(Race) .filter(Race.meeting_id == meeting_id) .order_by(Race.race_number) .all() ) @staticmethod def get_races_for_recompute( session: Session, date_from: date, date_to: date ) -> list[Race]: """Get races in date range, ordered chronologically. Args: session: Database session date_from: Start date date_to: End date Returns: List of races ordered by datetime/meeting date """ return ( session.query(Race) .join(Meeting) .filter( Meeting.meeting_date >= date_from, Meeting.meeting_date <= date_to, ) .order_by(Meeting.meeting_date, Race.race_number) .all() ) class HorseRepository: """Repository for horse operations.""" @staticmethod def upsert( session: Session, horse_id: int, name: str, raw_data: dict[str, Any] | None = None, ) -> Horse: """Upsert horse from TAB runner data. Args: session: Database session horse_id: TAB horse_id (integer) name: Horse name raw_data: Optional raw runner data for metadata Returns: Horse instance """ if not horse_id: raise ValueError("Horse data missing ID") stmt = insert(Horse).values( id=horse_id, name=name, raw_json=raw_data, ) stmt = stmt.on_conflict_do_update( index_elements=["id"], set_={ "name": stmt.excluded.name, "raw_json": stmt.excluded.raw_json, "updated_at": stmt.excluded.updated_at, }, ) session.execute(stmt) session.flush() return session.query(Horse).filter(Horse.id == horse_id).one() class DriverRepository: """Repository for driver operations. TAB API only provides driver names, not IDs. IDs are generated from name hash when no ID is provided. """ @staticmethod def upsert(session: Session, name: str, driver_id: int | None = None) -> Driver: """Upsert driver by name. Driver ID is generated from name hash when not provided since TAB API only provides names (in the "jockey" field for harness). Args: session: Database session name: Driver name driver_id: Optional explicit ID from upstream sources Returns: Driver instance """ if not name or not name.strip(): raise ValueError("Driver name is required") if driver_id is None: driver_id = generate_driver_id(name) stmt = insert(Driver).values( id=driver_id, name=name.strip(), raw_json={"name": name}, ) stmt = stmt.on_conflict_do_update( index_elements=["id"], set_={ "name": stmt.excluded.name, "raw_json": stmt.excluded.raw_json, "updated_at": stmt.excluded.updated_at, }, ) session.execute(stmt) session.flush() return session.query(Driver).filter(Driver.id == driver_id).one() class TrainerRepository: """Repository for trainer operations. TAB API only provides trainer names, not IDs. IDs are generated from name hash when no ID is provided. """ @staticmethod def upsert(session: Session, name: str, trainer_id: int | None = None) -> Trainer: """Upsert trainer by name. Trainer ID is generated from name hash when not provided since TAB API only provides names (in the "trainer_name" field). Args: session: Database session name: Trainer name trainer_id: Optional explicit ID from upstream sources Returns: Trainer instance """ if not name or not name.strip(): raise ValueError("Trainer name is required") if trainer_id is None: trainer_id = generate_trainer_id(name) stmt = insert(Trainer).values( id=trainer_id, name=name.strip(), raw_json={"name": name}, ) stmt = stmt.on_conflict_do_update( index_elements=["id"], set_={ "name": stmt.excluded.name, "raw_json": stmt.excluded.raw_json, "updated_at": stmt.excluded.updated_at, }, ) session.execute(stmt) session.flush() return session.query(Trainer).filter(Trainer.id == trainer_id).one() class StarterRepository: """Repository for starter (runner) operations.""" @staticmethod def upsert( session: Session, race_id: int, runner_data: dict[str, Any], placing: int | None = None, ) -> Starter | None: """Upsert starter/runner from TAB API data. Args: session: Database session race_id: Parent race ID runner_data: Runner data from TAB API (EventRunner) placing: Optional placing from results (matched by entrant_id) Returns: Starter instance or None if scratched """ runner_data = normalize_runner_data(runner_data) # Skip scratched runners is_scratched = runner_data.get("is_scratched", False) if is_scratched: logger.debug(f"Skipping scratched runner: {runner_data.get('name')}") return None # Extract horse data - TAB provides horse_id as integer horse_id = runner_data.get("horse_id") horse_name = runner_data.get("name", "Unknown") if horse_id: HorseRepository.upsert(session, horse_id, horse_name, runner_data) # Extract driver from "jockey" field (TAB uses jockey for all racing types) # For harness, this is actually the driver jockey_name = runner_data.get("jockey") or runner_data.get("driver_name") driver_id = None if jockey_name and jockey_name.strip(): driver = DriverRepository.upsert( session, jockey_name, driver_id=runner_data.get("driver_id") ) driver_id = driver.id elif runner_data.get("driver_id") is not None: driver_id = runner_data.get("driver_id") # Extract trainer from "trainer_name" field trainer_name = runner_data.get("trainer_name") trainer_id = None if trainer_name and trainer_name.strip(): trainer = TrainerRepository.upsert( session, trainer_name, trainer_id=runner_data.get("trainer_id") ) trainer_id = trainer.id elif runner_data.get("trainer_id") is not None: trainer_id = runner_data.get("trainer_id") # TAB provides barrier as integer directly barrier = runner_data.get("barrier") # Harness-specific barrier position (e.g., "1F" for front row, "2B" for back) barrier_position = runner_data.get("barrier_position") # Runner/saddlecloth number runner_number = runner_data.get("runner_number") # Handicap in meters handicap_m = runner_data.get("handicap") if handicap_m is None: handicap_m = runner_data.get("handicap_m") if handicap_m: handicap_m = int(handicap_m) # Placing is provided separately from results array did_not_finish = False if placing is None: # No result yet (upcoming race) or check if it's a DNF # TAB results have position as integer, non-finishers typically not in results placing = runner_data.get("placing") if isinstance(placing, str) and placing.strip().isdigit(): placing = int(placing.strip()) # Check for existing starter (by race + horse) existing = None if horse_id: existing = ( session.query(Starter) .filter(Starter.race_id == race_id, Starter.horse_id == horse_id) .first() ) if existing is None: runner_number = runner_data.get("runner_number") if runner_number is not None: existing = ( session.query(Starter) .filter( Starter.race_id == race_id, Starter.runner_number == runner_number, ) .first() ) if existing: # Track placing changes for audit logging old_placing = existing.placing placing_changed = ( placing is not None and old_placing is not None and placing != old_placing ) # Update existing existing.driver_id = driver_id existing.trainer_id = trainer_id existing.runner_number = runner_number existing.barrier = barrier existing.barrier_position = barrier_position existing.handicap_m = handicap_m existing.placing = placing existing.did_not_finish = did_not_finish existing.raw_json = runner_data session.flush() # Audit log placing correction if it changed if placing_changed: AuditLogger.log_change( session=session, table_name="starters", record_id=str(existing.id), action="CORRECT", old_values={"placing": old_placing}, new_values={"placing": placing}, changed_by="system", change_reason=( f"Placing corrected from {old_placing} to {placing} " f"for starter {existing.id} in race {race_id}" ), ) return existing else: # Insert new starter = Starter( race_id=race_id, horse_id=horse_id, driver_id=driver_id, trainer_id=trainer_id, runner_number=runner_number, barrier=barrier, barrier_position=barrier_position, handicap_m=handicap_m, placing=placing, did_not_finish=did_not_finish, raw_json=runner_data, ) session.add(starter) session.flush() return starter @staticmethod def get_by_race(session: Session, race_id: int) -> list[Starter]: """Get all starters for a race. Args: session: Database session race_id: Race ID Returns: List of starters """ return session.query(Starter).filter(Starter.race_id == race_id).all() class RatingSnapshotRepository: """Repository for rating snapshot operations.""" @staticmethod def upsert( session: Session, entity_type: EntityType, entity_id: int, as_of_race_id: int, rating: float, rd: float | None = None, meta: dict[str, Any] | None = None, ) -> RatingSnapshot: """Upsert rating snapshot. Args: session: Database session entity_type: Type of entity (horse/driver/trainer) entity_id: Entity ID as_of_race_id: Race ID this rating is computed after rating: Elo rating value rd: Rating deviation (optional) meta: Additional metadata (optional) Returns: RatingSnapshot instance """ stmt = insert(RatingSnapshot).values( entity_type=entity_type, entity_id=entity_id, as_of_race_id=as_of_race_id, rating=rating, rd=rd, meta=meta, ) stmt = stmt.on_conflict_do_update( index_elements=["entity_type", "entity_id", "as_of_race_id"], set_={ "rating": stmt.excluded.rating, "rd": stmt.excluded.rd, "meta": stmt.excluded.meta, }, ) session.execute(stmt) session.flush() return ( session.query(RatingSnapshot) .filter( RatingSnapshot.entity_type == entity_type, RatingSnapshot.entity_id == entity_id, RatingSnapshot.as_of_race_id == as_of_race_id, ) .one() ) @staticmethod def get_latest_rating( session: Session, entity_type: EntityType, entity_id: int, before_race_id: int | None = None, ) -> RatingSnapshot | None: """Get latest rating for an entity. Args: session: Database session entity_type: Type of entity entity_id: Entity ID before_race_id: Only consider ratings before this race (exclusive) Returns: Latest RatingSnapshot or None """ query = ( session.query(RatingSnapshot) .join(Race, RatingSnapshot.as_of_race_id == Race.id) .join(Meeting, Race.meeting_id == Meeting.id) .filter( RatingSnapshot.entity_type == entity_type, RatingSnapshot.entity_id == entity_id, ) ) if before_race_id is not None: target_race = ( session.query(Race) .options(joinedload(Race.meeting)) .filter(Race.id == before_race_id) .first() ) if target_race is None or target_race.meeting is None: return None if target_race.race_datetime: query = query.filter(Race.race_datetime < target_race.race_datetime) else: query = query.filter( or_( Meeting.meeting_date < target_race.meeting.meeting_date, and_( Meeting.meeting_date == target_race.meeting.meeting_date, Race.meeting_id == target_race.meeting_id, Race.race_number < target_race.race_number, ), ) ) return query.order_by( Meeting.meeting_date.desc(), Race.race_datetime.desc().nulls_last(), Race.race_number.desc(), Race.id.desc(), ).first() @staticmethod def get_top_ratings( session: Session, entity_type: EntityType, limit: int = 100, as_of_date: date | None = None, ) -> list[RatingSnapshot]: """Get top-rated entities. Args: session: Database session entity_type: Type of entity limit: Maximum results as_of_date: Only include ratings up to this date Returns: List of latest rating snapshots, sorted by rating descending """ # Use cache when querying live data (no as_of_date filter) if as_of_date is None: cache_key = f"top_ratings:{entity_type.value}:{limit}:live" cached = cache_get(cache_key) if cached is not None: # Reconstruct RatingSnapshot objects from cached IDs ids = cached.get("snapshot_ids", []) if ids: # Fetch full objects from DB by ID (preserves ORM state) objects = ( session.query(RatingSnapshot) .filter(RatingSnapshot.id.in_(ids)) .order_by(desc(RatingSnapshot.rating)) .all() ) return objects return [] sort_order = ( Meeting.meeting_date.desc(), Race.race_datetime.desc().nulls_last(), Race.race_number.desc(), Race.id.desc(), ) ranked = ( session.query( RatingSnapshot.id.label("snapshot_id"), RatingSnapshot.entity_id, func.row_number() .over(partition_by=RatingSnapshot.entity_id, order_by=sort_order) .label("rn"), ) .join(Race, RatingSnapshot.as_of_race_id == Race.id) .join(Meeting, Race.meeting_id == Meeting.id) .filter(RatingSnapshot.entity_type == entity_type) ) if as_of_date is not None: ranked = ranked.filter(Meeting.meeting_date <= as_of_date) ranked = ranked.subquery() results = ( session.query(RatingSnapshot) .join(ranked, RatingSnapshot.id == ranked.c.snapshot_id) .filter(ranked.c.rn == 1) .order_by(desc(RatingSnapshot.rating)) .limit(limit) .all() ) # Cache the snapshot IDs for live queries (TTL: 60 seconds) if as_of_date is None and results: cache_key = f"top_ratings:{entity_type.value}:{limit}:live" cache_set(cache_key, {"snapshot_ids": [r.id for r in results]}, ttl=60) return results class BarrierAdjustmentRepository: """Repository for barrier adjustment operations.""" @staticmethod def upsert( session: Session, venue: str | None, start_type: str | None, distance_bucket: str, barrier: int, adjustment: float, sample_count: int = 1, ) -> BarrierAdjustment: """Upsert barrier adjustment. Args: session: Database session venue: Venue name or None for global start_type: Start type or None for any distance_bucket: Distance bucket string barrier: Barrier number adjustment: Adjustment value sample_count: Number of observations Returns: BarrierAdjustment instance """ stmt = insert(BarrierAdjustment).values( venue=venue, start_type=start_type, distance_bucket=distance_bucket, barrier=barrier, adjustment=adjustment, sample_count=sample_count, ) stmt = stmt.on_conflict_do_update( index_elements=["venue", "start_type", "distance_bucket", "barrier"], set_={ "adjustment": stmt.excluded.adjustment, "sample_count": stmt.excluded.sample_count, "updated_at": stmt.excluded.updated_at, }, ) session.execute(stmt) session.flush() return ( session.query(BarrierAdjustment) .filter( BarrierAdjustment.venue == venue, BarrierAdjustment.start_type == start_type, BarrierAdjustment.distance_bucket == distance_bucket, BarrierAdjustment.barrier == barrier, ) .one() ) @staticmethod def get_all(session: Session) -> list[BarrierAdjustment]: """Get all barrier adjustments. Args: session: Database session Returns: List of all barrier adjustments """ return session.query(BarrierAdjustment).all() @staticmethod def increment_sample( session: Session, venue: str | None, start_type: str | None, distance_bucket: str, barrier: int, delta: float, learning_rate: float, ) -> None: """Incrementally update barrier adjustment. Args: session: Database session venue: Venue name or None start_type: Start type or None distance_bucket: Distance bucket barrier: Barrier number delta: Performance delta to incorporate learning_rate: Learning rate for update """ existing = ( session.query(BarrierAdjustment) .filter( BarrierAdjustment.venue == venue, BarrierAdjustment.start_type == start_type, BarrierAdjustment.distance_bucket == distance_bucket, BarrierAdjustment.barrier == barrier, ) .first() ) if existing: # Incremental update: adj_new = adj_old + lr * delta new_adjustment = existing.adjustment + learning_rate * delta new_count = existing.sample_count + 1 existing.adjustment = new_adjustment existing.sample_count = new_count else: # Initialize with first observation BarrierAdjustmentRepository.upsert( session, venue=venue, start_type=start_type, distance_bucket=distance_bucket, barrier=barrier, adjustment=learning_rate * delta, sample_count=1, ) class HandicapAdjustmentRepository: """Repository for handicap adjustment operations.""" @staticmethod def upsert( session: Session, venue: str | None, start_type: str | None, distance_bucket: str, handicap_m: int, adjustment: float, sample_count: int = 1, ) -> HandicapAdjustment: """Upsert handicap adjustment. Args: session: Database session venue: Venue name or None for global start_type: Start type or None for any distance_bucket: Distance bucket string handicap_m: Handicap in meters adjustment: Adjustment value sample_count: Number of observations Returns: HandicapAdjustment instance """ stmt = insert(HandicapAdjustment).values( venue=venue, start_type=start_type, distance_bucket=distance_bucket, handicap_m=handicap_m, adjustment=adjustment, sample_count=sample_count, ) stmt = stmt.on_conflict_do_update( index_elements=["venue", "start_type", "distance_bucket", "handicap_m"], set_={ "adjustment": stmt.excluded.adjustment, "sample_count": stmt.excluded.sample_count, "updated_at": stmt.excluded.updated_at, }, ) session.execute(stmt) session.flush() return ( session.query(HandicapAdjustment) .filter( HandicapAdjustment.venue == venue, HandicapAdjustment.start_type == start_type, HandicapAdjustment.distance_bucket == distance_bucket, HandicapAdjustment.handicap_m == handicap_m, ) .one() ) @staticmethod def get_all(session: Session) -> list[HandicapAdjustment]: """Get all handicap adjustments. Args: session: Database session Returns: List of all handicap adjustments """ return session.query(HandicapAdjustment).all() @staticmethod def increment_sample( session: Session, venue: str | None, start_type: str | None, distance_bucket: str, handicap_m: int, delta: float, learning_rate: float, ) -> None: """Incrementally update handicap adjustment. Args: session: Database session venue: Venue name or None start_type: Start type or None distance_bucket: Distance bucket handicap_m: Handicap in meters delta: Performance delta to incorporate learning_rate: Learning rate for update """ existing = ( session.query(HandicapAdjustment) .filter( HandicapAdjustment.venue == venue, HandicapAdjustment.start_type == start_type, HandicapAdjustment.distance_bucket == distance_bucket, HandicapAdjustment.handicap_m == handicap_m, ) .first() ) if existing: # Incremental update: adj_new = adj_old + lr * delta new_adjustment = existing.adjustment + learning_rate * delta new_count = existing.sample_count + 1 existing.adjustment = new_adjustment existing.sample_count = new_count else: # Initialize with first observation HandicapAdjustmentRepository.upsert( session, venue=venue, start_type=start_type, distance_bucket=distance_bucket, handicap_m=handicap_m, adjustment=learning_rate * delta, sample_count=1, )