Track Line Movements with Python and WebSockets

The most valuable information in betting is not the price. It is the change in the price. A market sitting at $1.95 tells you what the bookmaker thinks. A market that moved from $2.20 to $1.95 in eleven minutes across six bookmakers tells you that somebody with better information than the bookmaker just bet.

This tutorial builds a Python line movement tracker that connects to a live odds stream, persists every price change to a local database, detects steam moves across bookmakers, and alerts on them. It runs on a $5 VPS and it will teach you more about how betting markets work than any amount of reading.

Architecture

Four components, deliberately decoupled:

Keep them separate. The single most common failure in trackers like this is a slow notifier blocking the collector, which drops the connection, which loses the data you were trying to collect.

Step 1: dependencies and schema

pip install httpx python-dotenv

No ORM, no framework. The schema is three tables:

-- schema.sql
CREATE TABLE IF NOT EXISTS price_history (
    id          INTEGER PRIMARY KEY AUTOINCREMENT,
    event_id    TEXT    NOT NULL,
    sport       TEXT    NOT NULL,
    market      TEXT    NOT NULL,
    selection   TEXT    NOT NULL,
    bookmaker   TEXT    NOT NULL,
    price       REAL    NOT NULL,
    captured_at TEXT    NOT NULL      -- ISO8601 UTC
);

CREATE INDEX IF NOT EXISTS ix_ph_lookup
    ON price_history (event_id, market, selection, bookmaker, captured_at);

CREATE INDEX IF NOT EXISTS ix_ph_time
    ON price_history (captured_at);

CREATE TABLE IF NOT EXISTS steam_moves (
    id            INTEGER PRIMARY KEY AUTOINCREMENT,
    event_id      TEXT NOT NULL,
    market        TEXT NOT NULL,
    selection     TEXT NOT NULL,
    direction     TEXT NOT NULL,      -- 'firm' | 'drift'
    book_count    INTEGER NOT NULL,
    avg_move_pct  REAL NOT NULL,
    window_start  TEXT NOT NULL,
    window_end    TEXT NOT NULL,
    detected_at   TEXT NOT NULL
);

CREATE TABLE IF NOT EXISTS alert_cooldown (
    key         TEXT PRIMARY KEY,
    last_fired  TEXT NOT NULL
);

Step 2: the collector

The Krok Odds real-time endpoint speaks server-sent events. The frame format is event: NAME followed by data: JSON, separated by blank lines. Parsing it by hand is about fifteen lines.

# collector.py
import json
import os
import time
import random
from typing import Iterator, Tuple

import httpx

STREAM_URL = "https://krokodds.com.au/api/v1/stream/opportunities"
API_KEY = os.environ["KROK_API_KEY"]


def stream_events(params: dict) -> Iterator[Tuple[str, dict]]:
    """Yield (event_name, payload) forever, reconnecting with backoff."""
    attempt = 0
    while True:
        try:
            headers = {"X-API-Key": API_KEY, "Accept": "text/event-stream"}
            with httpx.stream("GET", STREAM_URL, params=params,
                              headers=headers, timeout=None) as r:
                r.raise_for_status()
                attempt = 0                     # connection succeeded
                event = None
                for line in r.iter_lines():
                    if not line:
                        continue
                    if line.startswith("event: "):
                        event = line[7:].strip()
                    elif line.startswith("data: "):
                        try:
                            yield (event or "message", json.loads(line[6:]))
                        except json.JSONDecodeError:
                            continue
        except Exception as exc:                # noqa: BLE001 — reconnect on anything
            attempt += 1
            # Exponential backoff with jitter, capped at 60s.
            delay = min(60, (2 ** min(attempt, 6))) * (0.5 + random.random())
            print(f"stream error: {exc!r}; reconnecting in {delay:.1f}s")
            time.sleep(delay)

The stream caps at ten minutes of wall clock per connection by design, so the reconnect loop is not an error path — it is the normal path. Resetting attempt on a successful connection is what stops a clean ten-minute cycle from slowly backing off into a sixty-second gap.

Polling collector for price history

The stream carries opportunity deltas. For a complete price series you also want a poller against the odds feed, which is where the actual per-bookmaker prices live.

# poller.py
import httpx, os, time
from datetime import datetime, timezone

BASE = "https://krokodds.com.au/api/v1"
HEADERS = {"X-API-Key": os.environ["KROK_API_KEY"]}


def poll_sport(sport: str) -> list[dict]:
    """Flatten one sport's feed into (event, market, selection, book, price) rows."""
    r = httpx.get(f"{BASE}/odds-feed/sports/{sport}",
                  params={"limit": 100}, headers=HEADERS, timeout=20)
    r.raise_for_status()
    rows = []
    for ev in r.json()["data"]:
        for market in ev.get("markets", []):
            for sel in market.get("selections", []):
                for p in sel.get("prices", []):
                    rows.append({
                        "event_id": ev["event_id"],
                        "sport": ev.get("sport", sport),
                        "market": market["key"],
                        "selection": sel["name"],
                        "bookmaker": p["bookmaker"],
                        "price": float(p["price"]),
                        "captured_at": p.get("captured_at")
                            or datetime.now(timezone.utc).isoformat(),
                    })
    return rows

Step 3: the store — write on change only

Writing every poll produces a database that is 95% duplicate rows. Keep the last-seen price per tuple in memory and only persist actual changes.

# store.py
import sqlite3
from datetime import datetime, timezone
from pathlib import Path

DB = Path("odds.db")


def connect() -> sqlite3.Connection:
    conn = sqlite3.connect(DB, isolation_level=None)
    conn.execute("PRAGMA journal_mode=WAL")      # concurrent reads while writing
    conn.execute("PRAGMA synchronous=NORMAL")
    conn.executescript(Path("schema.sql").read_text())
    return conn


class PriceStore:
    def __init__(self, conn: sqlite3.Connection):
        self.conn = conn
        self.last: dict[tuple, float] = {}
        self._warm()

    def _warm(self) -> None:
        """Rebuild the last-seen map on startup so a restart doesn't re-log everything."""
        cur = self.conn.execute("""
            SELECT event_id, market, selection, bookmaker, price
            FROM price_history
            WHERE id IN (SELECT MAX(id) FROM price_history
                         GROUP BY event_id, market, selection, bookmaker)
        """)
        for e, m, s, b, p in cur:
            self.last[(e, m, s, b)] = p

    def write(self, rows: list[dict]) -> int:
        changed = []
        for r in rows:
            key = (r["event_id"], r["market"], r["selection"], r["bookmaker"])
            prev = self.last.get(key)
            if prev is not None and abs(prev - r["price"]) < 1e-9:
                continue                          # unchanged, skip
            self.last[key] = r["price"]
            changed.append(r)

        if changed:
            self.conn.executemany("""
                INSERT INTO price_history
                    (event_id, sport, market, selection, bookmaker, price, captured_at)
                VALUES (:event_id, :sport, :market, :selection, :bookmaker, :price, :captured_at)
            """, changed)
        return len(changed)

Step 4: steam detection

The definition matters. A steam move is not "a big move" — it is a consistent move across multiple independent bookmakers inside a short window. Three parameters control it:

# detector.py
from collections import defaultdict
from datetime import datetime, timedelta, timezone

WINDOW = timedelta(minutes=5)
MIN_BOOKS = 4
MIN_MOVE_PCT = 2.5

# Books sharing a pricing feed must count once, not five times.
GROUPS = {
    "ladbrokes": "entain", "neds": "entain", "betstar": "entain",
    "betr": "betr", "betright": "betr", "boombet": "betr",
    "tab": "tabgroup", "unibet": "tabgroup", "tabtouch": "tabgroup",
}


def independent_count(books: set[str]) -> int:
    seen = set()
    for b in books:
        seen.add(GROUPS.get(b.lower(), b.lower()))
    return len(seen)


def detect_steam(conn) -> list[dict]:
    cutoff = (datetime.now(timezone.utc) - WINDOW).isoformat()
    rows = conn.execute("""
        SELECT event_id, market, selection, bookmaker, price, captured_at
        FROM price_history
        WHERE captured_at >= ?
        ORDER BY captured_at ASC
    """, (cutoff,)).fetchall()

    series = defaultdict(lambda: defaultdict(list))
    for event_id, market, selection, book, price, ts in rows:
        series[(event_id, market, selection)][book].append((ts, price))

    moves = []
    for (event_id, market, selection), books in series.items():
        firmed, drifted, pcts = set(), set(), []
        for book, points in books.items():
            if len(points) < 2:
                continue
            first, last = points[0][1], points[-1][1]
            pct = (last / first - 1) * 100
            if pct <= -MIN_MOVE_PCT:
                firmed.add(book); pcts.append(pct)
            elif pct >= MIN_MOVE_PCT:
                drifted.add(book); pcts.append(pct)

        for direction, group in (("firm", firmed), ("drift", drifted)):
            if independent_count(group) >= MIN_BOOKS:
                moves.append({
                    "event_id": event_id,
                    "market": market,
                    "selection": selection,
                    "direction": direction,
                    "book_count": independent_count(group),
                    "avg_move_pct": sum(pcts) / len(pcts),
                    "window_start": cutoff,
                    "window_end": datetime.now(timezone.utc).isoformat(),
                })
    return moves

The independent_count helper is the part most implementations miss. Ladbrokes, Neds and Betstar all move together because they are the same book — counting them as three confirmations turns a single trader's adjustment into a false steam alert.

Step 5: alert cooldown

Steam propagates over minutes, so the same underlying move re-qualifies on every detection cycle. Debounce per selection.

# notify.py
from datetime import datetime, timedelta, timezone

COOLDOWN = timedelta(minutes=20)


def should_alert(conn, move: dict) -> bool:
    key = f"{move['event_id']}|{move['market']}|{move['selection']}|{move['direction']}"
    row = conn.execute(
        "SELECT last_fired FROM alert_cooldown WHERE key = ?", (key,)
    ).fetchone()

    now = datetime.now(timezone.utc)
    if row:
        last = datetime.fromisoformat(row[0])
        if now - last < COOLDOWN:
            return False

    conn.execute(
        "INSERT INTO alert_cooldown (key, last_fired) VALUES (?, ?) "
        "ON CONFLICT(key) DO UPDATE SET last_fired = excluded.last_fired",
        (key, now.isoformat()),
    )
    return True

Step 6: tying it together

# main.py
import time
from store import connect, PriceStore
from poller import poll_sport
from detector import detect_steam
from notify import should_alert

SPORTS = ["afl", "nrl", "epl", "nba"]
POLL_SECONDS = 45


def main() -> None:
    conn = connect()
    store = PriceStore(conn)

    while True:
        started = time.monotonic()

        for sport in SPORTS:
            try:
                n = store.write(poll_sport(sport))
                if n:
                    print(f"{sport}: {n} price changes")
            except Exception as exc:               # noqa: BLE001
                print(f"{sport} poll failed: {exc!r}")

        for move in detect_steam(conn):
            if should_alert(conn, move):
                arrow = "↓" if move["direction"] == "firm" else "↑"
                print(f"STEAM {arrow} {move['selection']} "
                      f"({move['market']}) {move['avg_move_pct']:+.1f}% "
                      f"across {move['book_count']} independent books")

        elapsed = time.monotonic() - started
        time.sleep(max(0, POLL_SECONDS - elapsed))


if __name__ == "__main__":
    main()

What the data tells you once you have it

A month of price history unlocks analysis you cannot buy:

• Which books move first. Rank bookmakers by how often they lead a steam move. The leaders are your sharp reference; the laggards are where you get your bets on.
• How long opportunities live. Measure the gap between a book's move and the last book to follow. That number is your realistic execution window.
• Whether steam predicts results. Join your steam table against results and check whether steamed selections outperform their pre-move price.
• Your own CLV. With a full price series you can compute closing line value on every bet you place, without waiting for anyone else to compute it for you.

The query for the first of those:

SELECT bookmaker,
       COUNT(*) AS leads
FROM (
  SELECT ph.bookmaker,
         ph.selection,
         ph.event_id,
         ROW_NUMBER() OVER (
           PARTITION BY ph.event_id, ph.market, ph.selection
           ORDER BY ph.captured_at
         ) AS rn
  FROM price_history ph
  JOIN steam_moves sm
    ON sm.event_id = ph.event_id
   AND sm.market   = ph.market
   AND sm.selection = ph.selection
  WHERE ph.captured_at BETWEEN sm.window_start AND sm.window_end
)
WHERE rn = 1
GROUP BY bookmaker
ORDER BY leads DESC;

Operational notes

• Disk growth. Change-only writes keep it modest, but budget roughly 50-150MB per month per sport at 45-second polling. Archive by month.
• Clock skew. Use the captured_at from the API, not your local clock, or your movement windows will be wrong by whatever your NTP drift is.
• Do not retry 4xx. Only 429 and 5xx deserve a retry. Retrying a 401 in a tight loop is how you get rate limited.
• Run it under a supervisor. systemd with Restart=always covers the failure modes your backoff loop does not.

If you would rather consume steam moves than compute them, Krok Odds exposes detected moves directly at /v1/steam-moves and renders them on the odds movement board with the same independent-book counting applied. The full endpoint reference is at /api-docs.

The Complete Guide to AU Racing Arbitrage

Racing arbitrage between Betfair and fixed-odds bookmakers is real but underused. Here's how it works, the measured data, and how to do it.