#!/usr/bin/env python3 """ Prop Line Snapshot Worker V2 — Dynamic Frequency Based on Hours-to-Game ======================================================================== Upgrade from flat 2-hour intervals to market-grade snapshot frequency: | Hours to Game | Snapshot Interval | |---------------|-------------------| | >12h | Every 60 min | | 3h–12h | Every 15 min | | 30min–3h | Every 5 min | | <30min | Every 2 min | The worker runs a tight main loop (every 2 minutes) and decides per-league whether enough time has elapsed since the last fetch, based on the closest game's hoursToGame. This captures the critical final-hour line movements that drive real CLV measurement. RUN: PM2 process (replaces prop-snapshot-worker) """ import os import sys import time import psycopg2 import requests import traceback from datetime import datetime, timezone, timedelta from collections import defaultdict sys.path.insert(0, os.path.dirname(os.path.abspath(__file__))) from dotenv import load_dotenv load_dotenv('/var/www/html/eventheodds/.env') # SGO API config API_KEY = os.getenv('SPORTSGAMEODDS_API_KEY', '47d6ce020d896ece307a284e8c78ff7f') BASE_URL = os.getenv('SPORTSGAMEODDS_BASE_URL', 'https://api.sportsgameodds.com/v2') HEADER_NAME = os.getenv('SPORTSGAMEODDS_HEADER', 'x-api-key') DB_URL = os.getenv('SPORTS_DATABASE_URL', 'postgresql://eventheodds:eventheodds_dev_password@127.0.0.1:5433/eventheodds_sports').split('?')[0] # Leagues to snapshot LEAGUES = { 'nba': 'NBA', 'nhl': 'NHL', 'mlb': 'MLB', 'ncaab': 'NCAAB', 'epl': 'EPL', 'bundesliga': 'BUNDESLIGA', 'la_liga': 'LA_LIGA', 'serie_a': 'IT_SERIE_A', 'ligue_1': 'FR_LIGUE_1', 'champions_league': 'UEFA_CHAMPIONS_LEAGUE', 'mls': 'MLS', } # Prop stat types we care about PROP_STATS = { 'points': 'points', 'rebounds': 'rebounds', 'assists': 'assists', 'threePointersMade': 'threePointersMade', 'steals': 'steals', 'blocks': 'blocks', 'turnovers': 'turnovers', 'points+rebounds+assists': 'points+rebounds+assists', 'fantasyScore': 'fantasyScore', 'goals': 'goals', 'shots_onGoal': 'shots_onGoal', 'shots_on_goal': 'shots_onGoal', 'shots': 'shots', 'saves': 'saves', } # Main loop tick interval (seconds) — how often we check if any league needs a snapshot TICK_INTERVAL = 120 # 2 minutes # Dynamic frequency tiers: (max_hours_to_game, snapshot_interval_seconds) # Ordered from tightest to widest — first match wins FREQUENCY_TIERS = [ (0.5, 120), # <30 min: every 2 min (3.0, 300), # 30min-3h: every 5 min (12.0, 900), # 3h-12h: every 15 min (float('inf'), 3600), # >12h: every 60 min ] # Per-league tracking league_last_fetch = {} # {league: datetime} — when we last fetched from SGO API league_game_starts = {} # {league: [datetime, ...]} — cached game start times league_event_cache = {} # {league: [(event_id, start_dt), ...]} — for quick htg calc def get_required_interval(hours_to_game): """Determine snapshot interval (seconds) based on hours to closest game.""" for max_htg, interval in FREQUENCY_TIERS: if hours_to_game <= max_htg: return interval return 3600 def should_fetch_league(league, now): """Check if this league needs a fresh SGO API fetch based on its closest game.""" last_fetch = league_last_fetch.get(league) # Never fetched — always fetch if last_fetch is None: return True # Check cached game starts to determine closest game starts = league_game_starts.get(league, []) if not starts: # No cached games — use longest interval (60 min) elapsed = (now - last_fetch).total_seconds() return elapsed >= 3600 # Find minimum hoursToGame across all cached upcoming games min_htg = float('inf') for start_dt in starts: htg = max(0, (start_dt - now).total_seconds() / 3600) if htg < min_htg: min_htg = htg # If all games have started (htg <= 0), skip this league if min_htg <= 0: # But still fetch occasionally to catch late-added games elapsed = (now - last_fetch).total_seconds() return elapsed >= 3600 required_interval = get_required_interval(min_htg) elapsed = (now - last_fetch).total_seconds() return elapsed >= required_interval def update_game_starts_cache(league, events): """Cache game start times from SGO API response for hoursToGame calculations.""" starts = [] for event in events: starts_at = event.get('status', {}).get('startsAt', '') if starts_at: try: dt = datetime.fromisoformat(starts_at.replace('Z', '+00:00')) starts.append(dt) except (ValueError, TypeError): pass league_game_starts[league] = starts class SGOClient: """Lightweight SGO API client for prop snapshots.""" def __init__(self): self.session = requests.Session() self.session.headers.update({HEADER_NAME: API_KEY}) def get_upcoming_events(self, league_id, limit=50): """Get upcoming events with odds for a league.""" url = f"{BASE_URL}/events" params = { 'leagueID': league_id, 'limit': limit, 'oddsAvailable': 'true', 'started': 'false', } try: resp = self.session.get(url, params=params, timeout=30) if resp.status_code == 400: return [] resp.raise_for_status() data = resp.json() return data.get('data', []) except Exception as e: print(f" SGO API error for {league_id}: {e}", flush=True) return [] def extract_player_props(event): """Extract player prop odds from an SGO event.""" odds = event.get('odds', {}) if not odds: return [] props = [] for odd_id, odd_data in odds.items(): stat_id = odd_data.get('statID', '') entity_id = odd_data.get('statEntityID', '') period_id = odd_data.get('periodID', '') bet_type = odd_data.get('betTypeID', '') side_id = odd_data.get('sideID', '') if period_id != 'game' or bet_type != 'ou' or side_id != 'over': continue if entity_id in ('home', 'away', 'all', ''): continue prop_type = PROP_STATS.get(stat_id) if not prop_type: continue book_odds_str = odd_data.get('bookOdds') fair_odds_str = odd_data.get('fairOdds') line_value_str = odd_data.get('bookOverUnder') or odd_data.get('fairOverUnder') odds_american = None if book_odds_str: try: odds_american = int(float(book_odds_str)) except (ValueError, TypeError): pass if odds_american is None and fair_odds_str: try: odds_american = int(float(fair_odds_str)) except (ValueError, TypeError): pass line_value = None if line_value_str: try: line_value = float(line_value_str) except (ValueError, TypeError): pass if odds_american is None or line_value is None: continue # Also capture fair odds for cross-book analysis fair_odds_val = None if fair_odds_str: try: fair_odds_val = int(float(fair_odds_str)) except (ValueError, TypeError): pass props.append({ 'playerExternalId': entity_id, 'propType': prop_type, 'lineValue': line_value, 'oddsAmerican': odds_american, 'fairOdds': fair_odds_val, 'oddId': odd_id, }) return props def build_game_map(cur, events, canon_league): """Map SGO eventIDs to SportsGame rows.""" sg_map = {} sgo_event_ids = [evt['eventID'] for evt in events] if sgo_event_ids: placeholders = ','.join(['%s'] * len(sgo_event_ids)) cur.execute(f''' SELECT "externalGameId", id, "gameDate" FROM "SportsGame" WHERE "externalGameId" IN ({placeholders}) ''', [f'sgo:{eid}' for eid in sgo_event_ids]) for ext_id, sg_id, game_date in cur.fetchall(): sgo_eid = ext_id.replace('sgo:', '') sg_map[sgo_eid] = (sg_id, game_date) unmatched = [evt for evt in events if evt['eventID'] not in sg_map] if unmatched: for event in unmatched: eid = event['eventID'] starts_at = event.get('status', {}).get('startsAt', '') teams = event.get('teams', {}) home_names = teams.get('home', {}).get('names', {}) home_team = home_names.get('long', '') or home_names.get('medium', '') or home_names.get('short', '') if not home_team or not starts_at: continue try: game_dt = datetime.fromisoformat(starts_at.replace('Z', '+00:00')) game_date_str = game_dt.strftime('%Y-%m-%d') except (ValueError, TypeError): continue cur.execute(''' SELECT id, "gameDate" FROM "SportsGame" WHERE league = %s AND LOWER("homeTeam") = LOWER(%s) AND "gameDate"::date BETWEEN %s::date - 1 AND %s::date + 1 LIMIT 1 ''', (canon_league, home_team, game_date_str, game_date_str)) row = cur.fetchone() if row: sg_map[eid] = (row[0], row[1]) return sg_map def snapshot_league(client, cur, conn, canon_league, sgo_league_id, now): """Snapshot a single league. Returns (new, changed, skipped) counts.""" events = client.get_upcoming_events(sgo_league_id) if not events: return 0, 0, 0, 0 # Update cached game start times update_game_starts_cache(canon_league, events) sg_map = build_game_map(cur, events, canon_league) matched = len(sg_map) league_new = 0 league_changed = 0 league_skipped = 0 league_total = 0 for event in events: eid = event['eventID'] mapping = sg_map.get(eid) if not mapping: continue sg_id, game_date = mapping hours_to_game = None if game_date: game_dt = game_date.replace(tzinfo=timezone.utc) if game_date.tzinfo is None else game_date hours_to_game = max(0, round((game_dt - now).total_seconds() / 3600, 2)) props = extract_player_props(event) if not props: continue for prop in props: pid = prop['playerExternalId'] pt = prop['propType'] lv = prop['lineValue'] odds = prop['oddsAmerican'] # Dedup: check if most recent snapshot has same odds cur.execute(''' SELECT "oddsAmerican" FROM "PropLineHistory" WHERE "playerExternalId" = %s AND "gameId" = %s AND "propType" = %s AND "lineValue" = %s ORDER BY "snapshotAt" DESC LIMIT 1 ''', (pid, str(sg_id), pt, lv)) last = cur.fetchone() if last and last[0] == odds: league_skipped += 1 continue is_new = last is None is_changed = last is not None and last[0] != odds previous_odds = last[0] if last else None cur.execute(''' INSERT INTO "PropLineHistory" (league, "gameId", "playerExternalId", "propType", "snapshotAt", "hoursToGame", "lineValue", "oddsAmerican", "previousLine", "lineChange", "isOpening", "createdAt") VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, NOW()) ''', ( canon_league, str(sg_id), pid, pt, now, hours_to_game, lv, odds, previous_odds, (odds - previous_odds) if previous_odds else None, is_new, )) league_total += 1 if is_new: league_new += 1 if is_changed: league_changed += 1 conn.commit() return league_total, league_new, league_changed, league_skipped def mark_closing_lines(cur, conn): """Mark isClosingLine on PPL for recently started games.""" cur.execute(''' UPDATE "PlayerPropLine" ppl SET "isClosingLine" = TRUE FROM ( SELECT DISTINCT ON (plh."playerExternalId", plh."gameId", plh."propType", plh."lineValue") plh."playerExternalId", plh."gameId"::bigint AS game_id, plh."propType", plh."lineValue" FROM "PropLineHistory" plh JOIN "SportsGame" sg ON plh."gameId" = sg.id::text WHERE sg."gameDate" BETWEEN NOW() - INTERVAL '3 hours' AND NOW() ORDER BY plh."playerExternalId", plh."gameId", plh."propType", plh."lineValue", plh."hoursToGame" ASC NULLS LAST ) closing WHERE ppl."playerExternalId" = closing."playerExternalId" AND ppl."gameId" = closing.game_id AND ppl."propType" = closing."propType" AND ppl."lineValue" = closing."lineValue" AND ppl."isClosingLine" IS NOT TRUE ''') marked = cur.rowcount conn.commit() return marked def _connect_db(max_retries=3, retry_delay=5): """Create a DB connection with retry logic for transient failures.""" for attempt in range(1, max_retries + 1): try: conn = psycopg2.connect(DB_URL) conn.autocommit = False return conn except psycopg2.OperationalError as e: if attempt < max_retries: print(f" DB connect failed (attempt {attempt}/{max_retries}), retrying in {retry_delay}s: {e}", flush=True) time.sleep(retry_delay) else: raise def _is_connection_dead(conn): """Check if a psycopg2 connection is still alive.""" try: conn.isolation_level # triggers check cur = conn.cursor() cur.execute("SELECT 1") cur.fetchone() cur.close() return False except Exception: return True def tick_cycle(client, cycle_num): """One tick of the main loop — check which leagues need snapshots and fetch them.""" now = datetime.now(timezone.utc) # Determine which leagues need a fetch this tick leagues_to_fetch = [] for canon_league, sgo_league_id in LEAGUES.items(): if should_fetch_league(canon_league, now): leagues_to_fetch.append((canon_league, sgo_league_id)) if not leagues_to_fetch: return 0 # Nothing to do this tick print(f"\n{'='*65}", flush=True) print(f"TICK {cycle_num} — {now.strftime('%Y-%m-%d %H:%M:%S UTC')} " f"— {len(leagues_to_fetch)}/{len(LEAGUES)} leagues due", flush=True) print(f"{'='*65}", flush=True) conn = _connect_db() total_snapshots = 0 total_new = 0 total_changed = 0 total_skipped = 0 try: cur = conn.cursor() for canon_league, sgo_league_id in leagues_to_fetch: # Compute min hoursToGame for logging cached_starts = league_game_starts.get(canon_league, []) min_htg = None if cached_starts: min_htg = min(max(0, (s - now).total_seconds() / 3600) for s in cached_starts) try: lt, ln, lc, ls = snapshot_league(client, cur, conn, canon_league, sgo_league_id, now) except psycopg2.OperationalError as e: # Connection dropped mid-league — reconnect and retry this league print(f" DB connection lost during [{canon_league}], reconnecting: {e}", flush=True) try: conn.close() except Exception: pass conn = _connect_db() cur = conn.cursor() try: lt, ln, lc, ls = snapshot_league(client, cur, conn, canon_league, sgo_league_id, now) except Exception as retry_e: print(f" Retry failed for [{canon_league}]: {retry_e}", flush=True) traceback.print_exc() lt, ln, lc, ls = 0, 0, 0, 0 # Update last fetch time league_last_fetch[canon_league] = now total_snapshots += lt total_new += ln total_changed += lc total_skipped += ls # Determine current interval for this league interval = get_required_interval(min_htg) if min_htg is not None else 3600 htg_str = f"{min_htg:.1f}h" if min_htg is not None else "?" print(f" [{canon_league}] closest={htg_str} interval={interval}s " f"→ {lt} snaps ({ln} new, {lc} chg, {ls} skip)", flush=True) # Mark closing lines try: if _is_connection_dead(conn): conn = _connect_db() cur = conn.cursor() closing_marked = mark_closing_lines(cur, conn) if closing_marked > 0: print(f" Marked {closing_marked} PPL rows as isClosingLine", flush=True) except psycopg2.OperationalError as e: print(f" DB error during closing line marking (skipping): {e}", flush=True) cur.close() except Exception as e: try: conn.rollback() except Exception: pass print(f" ERROR in tick: {e}", flush=True) traceback.print_exc() finally: try: conn.close() except Exception: pass print(f" TOTAL: {total_snapshots} snaps ({total_new} new, " f"{total_changed} chg, {total_skipped} skip)", flush=True) return total_snapshots def main(): """Main loop — runs forever as a PM2 worker with dynamic frequency.""" print("=" * 65, flush=True) print("PROP SNAPSHOT WORKER V2 — DYNAMIC FREQUENCY", flush=True) print(f"Tick interval: {TICK_INTERVAL}s ({TICK_INTERVAL/60:.0f}min)", flush=True) print(f"Frequency tiers:", flush=True) for max_htg, interval in FREQUENCY_TIERS: htg_label = f"<{max_htg}h" if max_htg < 999 else ">12h" print(f" {htg_label}: every {interval}s ({interval/60:.0f}min)", flush=True) print(f"Leagues: {', '.join(LEAGUES.keys())}", flush=True) print(f"DB: {DB_URL[:50]}...", flush=True) print("=" * 65, flush=True) client = SGOClient() # Check API connectivity try: resp = client.session.get(f"{BASE_URL}/account/usage", timeout=10) usage = resp.json() print(f"SGO API: credits remaining = {usage.get('creditsRemaining', '?')}", flush=True) except Exception as e: print(f"WARNING: SGO API check failed: {e}", flush=True) # Verify DB connectivity and indexes try: conn = _connect_db() cur = conn.cursor() cur.execute('SELECT COUNT(*) FROM "PropLineHistory"') count = cur.fetchone()[0] print(f"PropLineHistory: {count:,} existing rows", flush=True) cur.execute('''CREATE INDEX IF NOT EXISTS "PropLineHistory_player_game" ON "PropLineHistory" ("playerExternalId", "gameId")''') cur.execute('''CREATE INDEX IF NOT EXISTS "PropLineHistory_snapshot" ON "PropLineHistory" ("snapshotAt")''') cur.execute('''CREATE INDEX IF NOT EXISTS "PropLineHistory_league_game" ON "PropLineHistory" (league, "gameId")''') cur.execute('''CREATE INDEX IF NOT EXISTS "PropLineHistory_clv_lookup" ON "PropLineHistory" ("playerExternalId", "gameId", "propType", "lineValue", "snapshotAt" DESC)''') # New index for velocity queries cur.execute('''CREATE INDEX IF NOT EXISTS "PropLineHistory_velocity" ON "PropLineHistory" ("gameId", "propType", "playerExternalId", "snapshotAt")''') conn.commit() conn.close() print("DB: OK, indexes verified", flush=True) except Exception as e: print(f"DB connection failed: {e}", flush=True) sys.exit(1) cycle = 0 while True: cycle += 1 try: tick_cycle(client, cycle) except Exception as e: print(f"\nTick {cycle} FAILED: {e}", flush=True) traceback.print_exc() time.sleep(TICK_INTERVAL) if __name__ == '__main__': main()