How to Track Line Movement in Sports Betting

import time
import sqlite3
from datetime import datetime
from sharpapi import SharpAPI

client = SharpAPI(api_key="YOUR_KEY")

conn = sqlite3.connect("line_history.db")
conn.execute("""
    CREATE TABLE IF NOT EXISTS snapshots (
        id INTEGER PRIMARY KEY,
        event_id TEXT,
        home_team TEXT,
        away_team TEXT,
        sport TEXT,
        sportsbook TEXT,
        market TEXT,
        outcome TEXT,
        price REAL,
        no_vig_price REAL,
        ts TEXT
    )
""")

def snapshot(sport: str):
    odds = client.odds.list(sport=sport, markets=["h2h", "spreads"])
    now = datetime.utcnow().isoformat()
    rows = []
    for event in odds.data:
        for book in event.bookmakers:
            for market in book.markets:
                for outcome in market.outcomes:
                    rows.append((
                        event.id, event.home_team, event.away_team, sport,
                        book.title, market.key, outcome.name,
                        outcome.price,
                        getattr(outcome, 'no_vig_price', None),
                        now,
                    ))
    conn.executemany(
        "INSERT INTO snapshots VALUES (NULL,?,?,?,?,?,?,?,?,?,?)", rows,
    )
    conn.commit()
    return len(rows)

# Poll every 2 minutes
while True:
    n = snapshot("basketball_nba")
    print(f"[{datetime.utcnow().isoformat()}] Snapped {n} prices")
    time.sleep(120)

def detect_moves(event_id: str, sportsbook: str, market: str, outcome: str):
    rows = conn.execute("""
        SELECT price, ts FROM snapshots
        WHERE event_id=? AND sportsbook=? AND market=? AND outcome=?
        ORDER BY ts ASC
    """, (event_id, sportsbook, market, outcome)).fetchall()

    moves = []
    for i in range(1, len(rows)):
        prev_price, prev_ts = rows[i - 1]
        curr_price, curr_ts = rows[i]
        if curr_price != prev_price:
            moves.append({
                "from": prev_price,
                "to": curr_price,
                "change": curr_price - prev_price,
                "ts": curr_ts,
            })
    return moves

from collections import defaultdict

def detect_steam(snapshots_before, snapshots_after, threshold: float = 3.0):
    """
    Return events where 3+ books moved the same direction by >= threshold
    points within the same snapshot window.
    """
    steam = []
    event_moves = defaultdict(list)

    for row in snapshots_after:
        event_id, book, market, outcome, price_after = row
        price_before = next(
            (r[4] for r in snapshots_before
             if r[0] == event_id and r[1] == book
             and r[2] == market and r[3] == outcome),
            None,
        )
        if price_before and abs(price_after - price_before) >= threshold:
            event_moves[event_id].append({
                "book": book,
                "market": market,
                "outcome": outcome,
                "change": price_after - price_before,
            })

    for event_id, moves in event_moves.items():
        directions = [m["change"] > 0 for m in moves]
        if len(moves) >= 3 and (all(directions) or not any(directions)):
            steam.append({
                "event_id": event_id,
                "books": len(moves),
                "moves": moves,
            })
    return steam