#!/usr/bin/env python3 """ Detect Sharp Money Movement from Odds Changes Identifies potential sharp action by analyzing: - Speed of line movement (from OddsSnapshot) - Magnitude of changes - Opening vs Closing line movement (historical games) - Reverse line movement (line moves opposite to public %) Creates SharpMove table with synthetic sharp money timing. Run: Every 2-4 hours during betting hours --backfill mode for historical games Usage: python3 detect_sharp_moves.py # Live games only python3 detect_sharp_moves.py --backfill # Also backfill historical python3 detect_sharp_moves.py --days 30 # Backfill last 30 days """ import psycopg2 import os from datetime import datetime, timezone, timedelta from collections import defaultdict import argparse # Books considered "sharp" (move first on sharp money) SHARP_BOOKS = {'pinnacle', 'circa', 'bookmaker', 'betcris', 'lowvig', '5dimes'} # Books considered "square" (follow sharp books) SQUARE_BOOKS = {'draftkings', 'fanduel', 'betmgm', 'caesars', 'pointsbet'} # Thresholds MIN_MOVE_THRESHOLD = 0.5 # Minimum spread move to consider MIN_ML_MOVE = 10 # Minimum moneyline move (points) RAPID_MOVE_WINDOW_MINUTES = 30 # Time window for "rapid" moves def load_db_url(): env_path = '/var/www/html/eventheodds/.env' try: with open(env_path, 'r') as f: for line in f: if line.startswith('SPORTS_DATABASE_URL='): return line.split('=', 1)[1].strip().split('?')[0] except FileNotFoundError: pass return os.environ.get('SPORTS_DATABASE_URL', '').split('?')[0] def ensure_sharp_move_table(cur): """Create/update SharpMove table""" cur.execute(''' CREATE TABLE IF NOT EXISTS "SharpMove" ( id BIGSERIAL PRIMARY KEY, league VARCHAR(50) NOT NULL, "gameId" BIGINT, "externalGameId" VARCHAR(100), "gameDate" TIMESTAMP, "homeTeam" VARCHAR(50), "awayTeam" VARCHAR(50), "detectedAt" TIMESTAMP DEFAULT NOW(), "hoursToGame" FLOAT, market VARCHAR(50), -- 'spread', 'total', 'ml' side VARCHAR(20), -- 'home', 'away', 'over', 'under' "openLine" FLOAT, "currentLine" FLOAT, "lineMove" FLOAT, "moveDirection" VARCHAR(10), -- 'sharp', 'square', 'steam' "moveMagnitude" VARCHAR(20), -- 'small', 'medium', 'large' "moveSpeed" VARCHAR(20), -- 'slow', 'fast', 'steam' "firstMoverBook" VARCHAR(50), "followingBooks" TEXT[], confidence FLOAT, -- 0-1 confidence score covered BOOLEAN, -- Did the sharp side win? (for historical) notes TEXT, "createdAt" TIMESTAMP DEFAULT NOW(), UNIQUE(league, "gameId", market, side, "detectedAt") ) ''') # Add covered column if missing (for existing tables) cur.execute(''' DO $$ BEGIN IF NOT EXISTS (SELECT 1 FROM information_schema.columns WHERE table_name = 'SharpMove' AND column_name = 'covered') THEN ALTER TABLE "SharpMove" ADD COLUMN covered BOOLEAN; END IF; END $$; ''') cur.execute('CREATE INDEX IF NOT EXISTS "SharpMove_league_game_idx" ON "SharpMove" (league, "gameId")') cur.execute('CREATE INDEX IF NOT EXISTS "SharpMove_detected_idx" ON "SharpMove" ("detectedAt")') cur.execute('CREATE INDEX IF NOT EXISTS "SharpMove_confidence_idx" ON "SharpMove" (confidence DESC)') cur.execute('CREATE INDEX IF NOT EXISTS "SharpMove_covered_idx" ON "SharpMove" (covered) WHERE covered IS NOT NULL') def analyze_gameodds_movement(cur, league, days_back=90): """ Analyze GameOdds table for historical line movements. Uses opening vs current lineValue differences. """ sharp_moves = [] # Get games from GameOdds with opening vs current line differences cur.execute(''' SELECT DISTINCT ON (g."gameId", g.market) g."gameId", g."gameDate", g."homeTeam", g."awayTeam", g.market, g."openingLineValue", g."lineValue", g."homeOdds", g."awayOdds", s.id as sports_game_id, s."homeScore", s."awayScore" FROM "GameOdds" g LEFT JOIN "SportsGame" s ON s."externalGameId" = g."gameId" AND s.league = g.league WHERE g.league = %s AND g."openingLineValue" IS NOT NULL AND g."lineValue" IS NOT NULL AND g."openingLineValue" != g."lineValue" AND ABS(g."lineValue" - g."openingLineValue") >= 0.5 AND g."gameDate" > NOW() - INTERVAL '%s days' ORDER BY g."gameId", g.market, g."fetchedAt" DESC ''', (league, days_back)) for row in cur.fetchall(): (ext_id, game_date, home, away, market, open_line, close_line, home_odds, away_odds, game_id, home_score, away_score) = row line_move = close_line - open_line magnitude = 'small' if abs(line_move) < 1 else ('medium' if abs(line_move) < 2 else 'large') speed = 'slow' # Historical, assume slow movement confidence = min(1.0, abs(line_move) / 2.5) if market == 'spreads': side = 'home' if line_move < 0 else 'away' covered = None if home_score is not None and away_score is not None: actual_spread = home_score - away_score if side == 'home': covered = actual_spread > -close_line else: covered = actual_spread < -close_line elif market == 'totals': side = 'over' if line_move > 0 else 'under' covered = None if home_score is not None and away_score is not None: actual_total = home_score + away_score if side == 'over': covered = actual_total > close_line else: covered = actual_total < close_line else: continue # Skip h2h for now sharp_moves.append({ 'game_id': game_id, 'ext_id': ext_id, 'game_date': game_date, 'home': home, 'away': away, 'market': 'spread' if market == 'spreads' else 'total', 'side': side, 'open_line': open_line, 'current_line': close_line, 'line_move': line_move, 'magnitude': magnitude, 'speed': speed, 'confidence': confidence, 'hours_to_game': 0, 'first_mover': 'consensus', 'covered': covered, 'notes': f"{market} moved {line_move:+.1f} from {open_line} to {close_line}" }) return sharp_moves def analyze_historical_games(cur, league, days_back=90): """ Analyze historical completed games using opening vs closing line movement. This catches sharp moves on games that have already been played. """ sharp_moves = [] # Get completed games with both opening and closing lines that differ cur.execute(''' SELECT id, "externalGameId", "gameDate", "homeTeam", "awayTeam", "openingSpreadHome", "closingSpreadHome", "openingTotal", "closingTotal", "openingMoneylineHome", "closingMoneylineHome", "openingMoneylineAway", "closingMoneylineAway", "homeScore", "awayScore" FROM "SportsGame" WHERE league = %s AND "homeScore" IS NOT NULL AND "openingSpreadHome" IS NOT NULL AND "closingSpreadHome" IS NOT NULL AND "openingSpreadHome" != "closingSpreadHome" AND "gameDate" > NOW() - INTERVAL '%s days' AND id NOT IN (SELECT DISTINCT "gameId" FROM "SharpMove" WHERE "gameId" IS NOT NULL) ORDER BY "gameDate" DESC LIMIT 500 ''', (league, days_back)) games = cur.fetchall() for row in games: (game_id, ext_id, game_date, home, away, open_spread, close_spread, open_total, close_total, open_ml_home, close_ml_home, open_ml_away, close_ml_away, home_score, away_score) = row # Spread movement if open_spread is not None and close_spread is not None: spread_move = close_spread - open_spread if abs(spread_move) >= MIN_MOVE_THRESHOLD: magnitude = 'small' if abs(spread_move) < 1 else ('medium' if abs(spread_move) < 2 else 'large') # For historical, we assume "slow" movement (happened over betting period) speed = 'slow' if abs(spread_move) < 1.5 else 'fast' # Higher confidence for larger moves confidence = min(1.0, abs(spread_move) / 2.5) # Sharp side is opposite of line movement direction side = 'home' if spread_move < 0 else 'away' # Check if sharp side won (ATS) actual_spread = home_score - away_score covered = None if actual_spread is not None and close_spread is not None: if side == 'home': covered = actual_spread > -close_spread else: covered = actual_spread < -close_spread sharp_moves.append({ 'game_id': game_id, 'ext_id': ext_id, 'game_date': game_date, 'home': home, 'away': away, 'market': 'spread', 'side': side, 'open_line': open_spread, 'current_line': close_spread, 'line_move': spread_move, 'magnitude': magnitude, 'speed': speed, 'confidence': confidence, 'hours_to_game': 0, 'first_mover': 'consensus', 'covered': covered, 'notes': f"Spread moved {spread_move:+.1f} from {open_spread} to {close_spread}" }) # Total movement if open_total is not None and close_total is not None: total_move = close_total - open_total if abs(total_move) >= MIN_MOVE_THRESHOLD: magnitude = 'small' if abs(total_move) < 1.5 else ('medium' if abs(total_move) < 3 else 'large') speed = 'slow' if abs(total_move) < 2 else 'fast' confidence = min(1.0, abs(total_move) / 3) side = 'over' if total_move > 0 else 'under' # Check if sharp side hit actual_total = (home_score or 0) + (away_score or 0) covered = None if actual_total and close_total: if side == 'over': covered = actual_total > close_total else: covered = actual_total < close_total sharp_moves.append({ 'game_id': game_id, 'ext_id': ext_id, 'game_date': game_date, 'home': home, 'away': away, 'market': 'total', 'side': side, 'open_line': open_total, 'current_line': close_total, 'line_move': total_move, 'magnitude': magnitude, 'speed': speed, 'confidence': confidence, 'hours_to_game': 0, 'first_mover': 'consensus', 'covered': covered, 'notes': f"Total moved {total_move:+.1f} from {open_total} to {close_total}" }) # Moneyline movement if open_ml_home is not None and close_ml_home is not None: ml_move = close_ml_home - open_ml_home if abs(ml_move) >= MIN_ML_MOVE: magnitude = 'small' if abs(ml_move) < 20 else ('medium' if abs(ml_move) < 50 else 'large') speed = 'slow' if abs(ml_move) < 30 else 'fast' confidence = min(1.0, abs(ml_move) / 80) side = 'home' if ml_move < 0 else 'away' # Check if sharp side won outright covered = None if home_score is not None and away_score is not None: if side == 'home': covered = home_score > away_score else: covered = away_score > home_score sharp_moves.append({ 'game_id': game_id, 'ext_id': ext_id, 'game_date': game_date, 'home': home, 'away': away, 'market': 'ml', 'side': side, 'open_line': open_ml_home, 'current_line': close_ml_home, 'line_move': ml_move, 'magnitude': magnitude, 'speed': speed, 'confidence': confidence, 'hours_to_game': 0, 'first_mover': 'consensus', 'covered': covered, 'notes': f"ML moved {ml_move:+.0f} from {open_ml_home} to {close_ml_home}" }) return sharp_moves def analyze_line_movement(cur, league): """Analyze line movements to detect sharp action on upcoming games""" sharp_moves = [] # Get games with multiple snapshots (upcoming games) cur.execute(''' SELECT DISTINCT "externalGameId", "gameDate", "homeTeam", "awayTeam" FROM "OddsSnapshot" WHERE league = %s AND "gameDate" > NOW() AND "gameDate" < NOW() + INTERVAL '7 days' ''', (league,)) games = cur.fetchall() for ext_id, game_date, home, away in games: # Get snapshots for this game cur.execute(''' SELECT "snapshotAt", "spreadHome", "total", "moneylineHome", "moneylineAway", bookmaker, "hoursToGame" FROM "OddsSnapshot" WHERE league = %s AND "externalGameId" = %s ORDER BY "snapshotAt" ''', (league, ext_id)) snapshots = cur.fetchall() if len(snapshots) < 2: continue # Track line movements first_snap = snapshots[0] last_snap = snapshots[-1] # Spread movement if first_snap[1] is not None and last_snap[1] is not None: spread_move = last_snap[1] - first_snap[1] if abs(spread_move) >= MIN_MOVE_THRESHOLD: # Determine move characteristics magnitude = 'small' if abs(spread_move) < 1 else ('medium' if abs(spread_move) < 2 else 'large') # Time between first and last snapshot time_diff = (last_snap[0] - first_snap[0]).total_seconds() / 60 speed = 'slow' if time_diff > 360 else ('fast' if time_diff > 60 else 'steam') # Confidence based on magnitude and speed confidence = min(1.0, (abs(spread_move) / 3) * (1 + (1 / max(time_diff / 60, 1)))) side = 'home' if spread_move < 0 else 'away' # Line moving toward = sharp on opposite sharp_moves.append({ 'ext_id': ext_id, 'game_date': game_date, 'home': home, 'away': away, 'market': 'spread', 'side': side, 'open_line': first_snap[1], 'current_line': last_snap[1], 'line_move': spread_move, 'magnitude': magnitude, 'speed': speed, 'confidence': confidence, 'hours_to_game': last_snap[6], 'first_mover': last_snap[5], 'notes': f"Spread moved {spread_move:+.1f} from {first_snap[1]} to {last_snap[1]}" }) # Total movement if first_snap[2] is not None and last_snap[2] is not None: total_move = last_snap[2] - first_snap[2] if abs(total_move) >= MIN_MOVE_THRESHOLD: magnitude = 'small' if abs(total_move) < 1.5 else ('medium' if abs(total_move) < 3 else 'large') time_diff = (last_snap[0] - first_snap[0]).total_seconds() / 60 speed = 'slow' if time_diff > 360 else ('fast' if time_diff > 60 else 'steam') confidence = min(1.0, (abs(total_move) / 4) * (1 + (1 / max(time_diff / 60, 1)))) side = 'over' if total_move > 0 else 'under' sharp_moves.append({ 'ext_id': ext_id, 'game_date': game_date, 'home': home, 'away': away, 'market': 'total', 'side': side, 'open_line': first_snap[2], 'current_line': last_snap[2], 'line_move': total_move, 'magnitude': magnitude, 'speed': speed, 'confidence': confidence, 'hours_to_game': last_snap[6], 'first_mover': last_snap[5], 'notes': f"Total moved {total_move:+.1f} from {first_snap[2]} to {last_snap[2]}" }) # Moneyline movement if first_snap[3] is not None and last_snap[3] is not None: ml_move = last_snap[3] - first_snap[3] if abs(ml_move) >= MIN_ML_MOVE: magnitude = 'small' if abs(ml_move) < 20 else ('medium' if abs(ml_move) < 50 else 'large') time_diff = (last_snap[0] - first_snap[0]).total_seconds() / 60 speed = 'slow' if time_diff > 360 else ('fast' if time_diff > 60 else 'steam') confidence = min(1.0, (abs(ml_move) / 100) * (1 + (1 / max(time_diff / 60, 1)))) # Negative ML becoming more negative = sharp on that side side = 'home' if ml_move < 0 else 'away' sharp_moves.append({ 'ext_id': ext_id, 'game_date': game_date, 'home': home, 'away': away, 'market': 'ml', 'side': side, 'open_line': first_snap[3], 'current_line': last_snap[3], 'line_move': ml_move, 'magnitude': magnitude, 'speed': speed, 'confidence': confidence, 'hours_to_game': last_snap[6], 'first_mover': last_snap[5], 'notes': f"ML moved {ml_move:+d} from {first_snap[3]} to {last_snap[3]}" }) return sharp_moves def insert_sharp_move(cur, league, move, game_id=None): """Insert a sharp move record""" try: cur.execute(''' INSERT INTO "SharpMove" ( league, "gameId", "externalGameId", "gameDate", "homeTeam", "awayTeam", "hoursToGame", market, side, "openLine", "currentLine", "lineMove", "moveMagnitude", "moveSpeed", "firstMoverBook", confidence, covered, notes ) VALUES (%s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s, %s) ON CONFLICT (league, "gameId", market, side, "detectedAt") DO NOTHING RETURNING id ''', ( league, move.get('game_id') or game_id, move['ext_id'], move['game_date'], move['home'], move['away'], move.get('hours_to_game'), move['market'], move['side'], move['open_line'], move['current_line'], move['line_move'], move['magnitude'], move['speed'], move.get('first_mover'), move['confidence'], move.get('covered'), move['notes'] )) return cur.fetchone() is not None except Exception as e: return False def detect_sharp_moves(backfill=False, days_back=90): """Main detection function""" db_url = load_db_url() if not db_url: raise RuntimeError('SPORTS_DATABASE_URL not set') conn = psycopg2.connect(db_url) cur = conn.cursor() print("=" * 60) print("DETECT SHARP LINE MOVEMENTS") print(f"Time: {datetime.now(timezone.utc).isoformat()}") print(f"Mode: {'Backfill + Live' if backfill else 'Live only'}") if backfill: print(f"Backfill period: {days_back} days") print("=" * 60) ensure_sharp_move_table(cur) conn.commit() # Get leagues with odds data cur.execute('SELECT DISTINCT league FROM "OddsSnapshot"') snapshot_leagues = set(r[0] for r in cur.fetchall()) # Also get leagues from SportsGame for backfill if backfill: cur.execute('''SELECT DISTINCT league FROM "SportsGame" WHERE league IN ('nba', 'nfl', 'nhl', 'mlb', 'ncaab', 'ncaaf')''') game_leagues = set(r[0] for r in cur.fetchall()) leagues = snapshot_leagues | game_leagues else: leagues = snapshot_leagues print(f"Analyzing {len(leagues)} leagues...") total_moves = 0 high_conf = 0 historical_moves = 0 for league in leagues: # Analyze live games from snapshots if league in snapshot_leagues: moves = analyze_line_movement(cur, league) if moves: print(f"\n{league.upper()}: Found {len(moves)} live sharp moves") for move in moves: cur.execute(''' SELECT id FROM "SportsGame" WHERE league = %s AND "externalGameId" = %s LIMIT 1 ''', (league, move['ext_id'])) game_row = cur.fetchone() game_id = game_row[0] if game_row else None if insert_sharp_move(cur, league, move, game_id): total_moves += 1 if move['confidence'] >= 0.6: high_conf += 1 # Backfill historical games if backfill: # From SportsGame opening/closing hist_moves = analyze_historical_games(cur, league, days_back) if hist_moves: print(f" {league.upper()}: Found {len(hist_moves)} SportsGame historical moves") for move in hist_moves: if insert_sharp_move(cur, league, move): total_moves += 1 historical_moves += 1 if move['confidence'] >= 0.6: high_conf += 1 # From GameOdds opening/current gameodds_moves = analyze_gameodds_movement(cur, league, days_back) if gameodds_moves: print(f" {league.upper()}: Found {len(gameodds_moves)} GameOdds historical moves") for move in gameodds_moves: if insert_sharp_move(cur, league, move): total_moves += 1 historical_moves += 1 if move['confidence'] >= 0.6: high_conf += 1 conn.commit() # Summary cur.execute(''' SELECT market, "moveMagnitude", COUNT(*) FROM "SharpMove" WHERE "detectedAt" > NOW() - INTERVAL '24 hours' GROUP BY market, "moveMagnitude" ORDER BY market, 3 DESC ''') print(f"\nLast 24h Sharp Moves by Type:") for row in cur.fetchall(): print(f" {row[0]} - {row[1]}: {row[2]}") # Sharp side performance (if we have covered data) cur.execute(''' SELECT market, COUNT(*) FILTER (WHERE covered = true) as wins, COUNT(*) FILTER (WHERE covered = false) as losses, COUNT(*) FILTER (WHERE covered IS NOT NULL) as total FROM "SharpMove" GROUP BY market HAVING COUNT(*) FILTER (WHERE covered IS NOT NULL) > 0 ''') perf = cur.fetchall() if perf: print(f"\nSharp Side Performance:") for row in perf: if row[3] > 0: pct = 100.0 * row[1] / row[3] print(f" {row[0]}: {row[1]}-{row[2]} ({pct:.1f}%)") cur.close() conn.close() print(f"\n{'='*60}") print(f"TOTAL: {total_moves} moves detected, {high_conf} high confidence (>=0.6)") if backfill: print(f" Historical: {historical_moves}") print("=" * 60) return {'total': total_moves, 'high_conf': high_conf, 'historical': historical_moves} def main(): parser = argparse.ArgumentParser(description='Detect sharp money movements') parser.add_argument('--backfill', action='store_true', help='Also backfill historical games using opening/closing lines') parser.add_argument('--days', type=int, default=90, help='Days to look back for backfill (default: 90)') args = parser.parse_args() try: result = detect_sharp_moves(backfill=args.backfill, days_back=args.days) print(f"\nDetection complete: {result['total']} moves, {result['high_conf']} high confidence") except Exception as e: print(f"ERROR: {e}") import traceback traceback.print_exc() if __name__ == '__main__': main()