#!/usr/bin/env python3 """ Rainmaker Benchmark Rule Audit Runs every 30 minutes to validate and fix forecast_side / spread_edge / total fields. BENCHMARK RULE: spread_value = market_home_spread + projected_margin If spread_value > 0 → home team is VALUE SIDE If spread_value < 0 → away team is VALUE SIDE spread_edge = |spread_value| total_edge = projected_total_points - market_total Positive → OVER, Negative → UNDER, Zero → NONE """ import json import logging import os import sys from datetime import datetime, date import psycopg2 import psycopg2.extras # ── Config ────────────────────────────────────────────────────────────────── DB_HOST = "127.0.0.1" DB_PORT = 5433 DB_NAME = "eventheodds_sports" DB_USER = "eventheodds" DB_PASS = "eventheodds_dev_password" LOG_DIR = "/var/log/rainmaker" LOG_FILE = os.path.join(LOG_DIR, "forecast-audit.log") # ── Logging ───────────────────────────────────────────────────────────────── os.makedirs(LOG_DIR, exist_ok=True) logging.basicConfig( level=logging.INFO, format="%(asctime)s [%(levelname)s] %(message)s", handlers=[ logging.FileHandler(LOG_FILE), logging.StreamHandler(sys.stdout), ], ) log = logging.getLogger("forecast-audit") def connect(): return psycopg2.connect( host=DB_HOST, port=DB_PORT, dbname=DB_NAME, user=DB_USER, password=DB_PASS, ) def get_market_spread(event_row): """Extract home spread line from rm_events.spread JSONB.""" sp = event_row.get("spread") if not sp: return None home = sp.get("home") if isinstance(home, dict): return home.get("line") if isinstance(home, (int, float)): return float(home) if isinstance(home, str): try: return float(home) except ValueError: return None return None def get_market_total(event_row): """Extract total line from rm_events.total JSONB.""" t = event_row.get("total") if not t: return None over = t.get("over") if isinstance(over, dict): return over.get("line") if isinstance(over, (int, float)): return float(over) if isinstance(over, str): try: return float(over) except ValueError: return None return None def audit_forecasts(cur, fix=True): """ Audit all today's forecasts. Returns counts of checked, issues found, and fixes applied. """ today_et = datetime.now().strftime("%Y-%m-%d") # Pull today's forecasts joined with events for market lines cur.execute(""" SELECT fc.event_id, fc.home_team, fc.away_team, fc.forecast_data, e.spread, e.total FROM rm_forecast_cache fc LEFT JOIN rm_events e ON fc.event_id = e.event_id WHERE (fc.starts_at AT TIME ZONE 'America/New_York')::date = %s::date AND fc.forecast_data IS NOT NULL """, (today_et,)) rows = cur.fetchall() checked = 0 issues = 0 fixed = 0 details = [] for row in rows: event_id = row["event_id"] home = row["home_team"] or "" away = row["away_team"] or "" fd = row["forecast_data"] or {} checked += 1 proj_margin = fd.get("projected_margin") proj_total = fd.get("projected_total_points") market_home_spread = get_market_spread(row) market_total = get_market_total(row) patches = {} # ── Spread benchmark ──────────────────────────────────────────── if proj_margin is not None and market_home_spread is not None: spread_value = market_home_spread + proj_margin expected_edge = round(abs(spread_value), 1) if spread_value > 0: expected_side = home elif spread_value < 0: expected_side = away else: expected_side = fd.get("winner_pick", home) current_side = fd.get("forecast_side") current_edge = fd.get("spread_edge") side_ok = (current_side == expected_side) edge_ok = (current_edge is not None and round(float(current_edge), 1) == expected_edge) if not side_ok or not edge_ok: issues += 1 detail = ( f" {away} @ {home} ({event_id}): " f"margin={proj_margin}, mkt_spread={market_home_spread}, " f"spread_val={round(spread_value,1)}" ) if not side_ok: detail += f" | side: {current_side} → {expected_side}" if not edge_ok: detail += f" | edge: {current_edge} → {expected_edge}" details.append(detail) patches["forecast_side"] = expected_side patches["spread_edge"] = expected_edge # ── Projected winner ──────────────────────────────────────────── winner_pick = fd.get("winner_pick") current_proj_winner = fd.get("projected_winner") if winner_pick and current_proj_winner != winner_pick: patches["projected_winner"] = winner_pick # ── Total benchmark ───────────────────────────────────────────── if proj_total is not None and market_total is not None: total_edge_val = round(proj_total - market_total, 1) if total_edge_val > 0: expected_dir = "OVER" elif total_edge_val < 0: expected_dir = "UNDER" else: expected_dir = "NONE" current_dir = fd.get("total_direction") current_tedge = fd.get("total_edge") dir_ok = (current_dir == expected_dir) tedge_ok = (current_tedge is not None and round(float(current_tedge), 1) == total_edge_val) if not dir_ok or not tedge_ok: if "forecast_side" not in patches: # Only count as new issue if not already counted for spread issues += 1 detail = ( f" {away} @ {home} ({event_id}): " f"proj_total={proj_total}, mkt_total={market_total}" ) if not dir_ok: detail += f" | dir: {current_dir} → {expected_dir}" if not tedge_ok: detail += f" | total_edge: {current_tedge} → {total_edge_val}" details.append(detail) patches["total_direction"] = expected_dir patches["total_edge"] = total_edge_val # ── Missing benchmark fields (no market data available) ───────── if not fd.get("forecast_side") and proj_margin is not None and market_home_spread is None: details.append( f" {away} @ {home} ({event_id}): " f"WARN — no market spread available, cannot compute benchmark" ) # ── Apply patches ─────────────────────────────────────────────── if patches and fix: merged = {**fd, **patches} cur.execute(""" UPDATE rm_forecast_cache SET forecast_data = %s WHERE event_id = %s """, (json.dumps(merged), event_id)) fixed += 1 return checked, issues, fixed, details def main(): log.info("=" * 60) log.info("BENCHMARK RULE AUDIT — %s", datetime.now().strftime("%Y-%m-%d %H:%M:%S ET")) log.info("=" * 60) conn = connect() conn.autocommit = False try: cur = conn.cursor(cursor_factory=psycopg2.extras.RealDictCursor) checked, issues, fixed, details = audit_forecasts(cur, fix=True) if issues > 0: conn.commit() log.warning("Checked %d forecasts — %d issues found, %d fixed", checked, issues, fixed) for d in details: log.warning(d) else: conn.rollback() # nothing to commit log.info("Checked %d forecasts — all benchmark fields correct ✓", checked) cur.close() except Exception: conn.rollback() log.exception("Audit failed with error") sys.exit(1) finally: conn.close() log.info("Audit complete.") if __name__ == "__main__": main()