Real-time PrizePicks line alerts, in one file.

#!/usr/bin/env python3
"""Line-move alerts from the prizepicksapi.com change stream (SSE).

Standard library only. Resumes from the last applied event with Last-Event-ID,
so a restart or a dropped connection never skips or repeats an alert.
Run:  PRIZEPICKS_API_KEY=pp_... python3 line_alerts.py
"""
import json
import os
import sys
import time
from urllib.error import HTTPError, URLError
from urllib.parse import urlencode
from urllib.request import Request, urlopen

BASE = os.environ.get("PRIZEPICKS_BASE", "https://prizepicksapi.com")
KEY = os.environ["PRIZEPICKS_API_KEY"]
STATE_FILE = os.environ.get("STREAM_STATE", "last_event_id.txt")
FILTERS = {"league": "NBA", "type": "offer.changed"}
MIN_MOVE = 0.5  # alert only when the line moves at least this much


def load_cursor():
    try:
        with open(STATE_FILE) as f:
            return f.read().strip() or None
    except FileNotFoundError:
        return None


def save_cursor(cursor):
    tmp = STATE_FILE + ".tmp"
    with open(tmp, "w") as f:
        f.write(cursor)
    os.replace(tmp, STATE_FILE)  # atomic: never a half-written cursor


def on_event(kind, event):
    prev, line = event.get("previous_line"), event.get("line")
    if kind != "offer.changed" or not isinstance(prev, (int, float)) or not isinstance(line, (int, float)):
        return
    if abs(line - prev) >= MIN_MOVE:
        print(f"{event['observed_at']}  {event['league']}  {event['player_name']}  "
              f"{event['stat']}: {prev} -> {line}", flush=True)


def stream_once(cursor):
    """Read one connection until the server asks for a reconnect. Returns (cursor, retry_seconds)."""
    headers = {"Authorization": "Bearer " + KEY, "Accept": "text/event-stream"}
    if cursor:
        headers["Last-Event-ID"] = cursor
    request = Request(BASE + "/api/v1/stream?" + urlencode(FILTERS), headers=headers)
    retry = 2.0
    # Heartbeats arrive every 15 s, so a 60 s read timeout only fires on a dead connection.
    with urlopen(request, timeout=60) as response:
        event_id, kind, data = None, None, []
        for raw in response:
            text = raw.decode("utf-8").rstrip("\r\n")
            if text == "":  # blank line: dispatch the event
                if data:
                    payload = "\n".join(data)
                    if kind == "reconnect":
                        return cursor, retry
                    if kind == "error":
                        raise ConnectionError(payload)
                    on_event(kind, json.loads(payload))
                    if event_id:
                        cursor = event_id
                        save_cursor(cursor)  # after handling: at-least-once delivery
                event_id, kind, data = None, None, []
                continue
            if text.startswith(":"):  # comment: connected banner or heartbeat
                continue
            field, _, value = text.partition(":")
            value = value[1:] if value.startswith(" ") else value
            if field == "id":
                event_id = value
            elif field == "event":
                kind = value
            elif field == "data":
                data.append(value)
            elif field == "retry" and value.isdigit():
                retry = int(value) / 1000
    return cursor, retry


def main():
    cursor, backoff = load_cursor(), 2
    print("Resuming after", cursor or "the newest event", file=sys.stderr)
    while True:
        try:
            cursor, retry = stream_once(cursor)
            backoff = 2
            time.sleep(retry)
        except HTTPError as error:
            body = error.read().decode("utf-8", errors="replace")
            if error.code in (400, 401, 403):
                sys.exit(f"HTTP {error.code}: {body}")  # bad key or query: retrying will not help
            wait = int(error.headers.get("Retry-After") or backoff)
            print(f"HTTP {error.code}, retrying in {wait}s", file=sys.stderr)
            time.sleep(wait)
            backoff = min(backoff * 2, 60)
        except (URLError, ConnectionError, TimeoutError, OSError) as error:
            print(f"Connection lost ({error}), retrying in {backoff}s", file=sys.stderr)
            time.sleep(backoff)
            backoff = min(backoff * 2, 60)


if __name__ == "__main__":
    try:
        main()
    except KeyboardInterrupt:
        pass

// Line-move alerts from the prizepicksapi.com change stream (SSE). Node 18+, no dependencies.
// Resumes with Last-Event-ID, so a restart or dropped connection never skips or repeats an alert.
// Run:  PRIZEPICKS_API_KEY=pp_... node line-alerts.mjs
import { readFile, rename, writeFile } from "node:fs/promises";

const BASE = process.env.PRIZEPICKS_BASE ?? "https://prizepicksapi.com";
const KEY = process.env.PRIZEPICKS_API_KEY;
const STATE_FILE = process.env.STREAM_STATE ?? "last_event_id.txt";
const FILTERS = new URLSearchParams({ league: "NBA", type: "offer.changed" });
const MIN_MOVE = 0.5;
const sleep = (s) => new Promise((r) => setTimeout(r, s * 1000));

async function loadCursor() {
  try { return (await readFile(STATE_FILE, "utf8")).trim() || null; } catch { return null; }
}
async function saveCursor(cursor) {
  await writeFile(STATE_FILE + ".tmp", cursor);
  await rename(STATE_FILE + ".tmp", STATE_FILE); // atomic
}

function onEvent(kind, e) {
  if (kind !== "offer.changed" || typeof e.previous_line !== "number" || typeof e.line !== "number") return;
  if (Math.abs(e.line - e.previous_line) >= MIN_MOVE) {
    console.log(`${e.observed_at}  ${e.league}  ${e.player_name}  ${e.stat}: ${e.previous_line} -> ${e.line}`);
  }
}

class HttpError extends Error {
  constructor(status, body, retryAfter) { super(`HTTP ${status}: ${body}`); Object.assign(this, { status, retryAfter }); }
}

// Reads one connection until the server sends "reconnect". Returns { cursor, retry }.
async function streamOnce(cursor) {
  const headers = { Authorization: `Bearer ${KEY}`, Accept: "text/event-stream" };
  if (cursor) headers["Last-Event-ID"] = cursor;
  const res = await fetch(`${BASE}/api/v1/stream?${FILTERS}`, { headers });
  if (!res.ok) throw new HttpError(res.status, await res.text(), Number(res.headers.get("retry-after")) || null);

  const decoder = new TextDecoder();
  let buffer = "", retry = 2, id = null, kind = null, data = [];
  for await (const chunk of res.body) {
    buffer += decoder.decode(chunk, { stream: true });
    let newline;
    while ((newline = buffer.indexOf("\n")) >= 0) {
      const line = buffer.slice(0, newline).replace(/\r$/, "");
      buffer = buffer.slice(newline + 1);
      if (line === "") { // blank line: dispatch
        if (data.length) {
          const payload = data.join("\n");
          if (kind === "reconnect") return { cursor, retry };
          if (kind === "error") throw new Error(`stream error: ${payload}`);
          onEvent(kind, JSON.parse(payload));
          if (id) { cursor = id; await saveCursor(cursor); } // after handling: at-least-once
        }
        id = null; kind = null; data = [];
        continue;
      }
      if (line.startsWith(":")) continue; // connected banner or heartbeat
      const colon = line.indexOf(":");
      const field = colon < 0 ? line : line.slice(0, colon);
      let value = colon < 0 ? "" : line.slice(colon + 1);
      if (value.startsWith(" ")) value = value.slice(1);
      if (field === "id") id = value;
      else if (field === "event") kind = value;
      else if (field === "data") data.push(value);
      else if (field === "retry" && /^\d+$/.test(value)) retry = Number(value) / 1000;
    }
  }
  return { cursor, retry }; // connection closed without a reconnect event
}

if (!KEY) throw new Error("Set PRIZEPICKS_API_KEY");
let cursor = await loadCursor(), backoff = 2;
console.error("Resuming after", cursor ?? "the newest event");
for (;;) {
  try {
    const next = await streamOnce(cursor);
    cursor = next.cursor; backoff = 2;
    await sleep(next.retry);
  } catch (error) {
    if (error instanceof HttpError && [400, 401, 403].includes(error.status)) throw error; // bad key or query
    const wait = error.retryAfter ?? backoff;
    console.error(`${error.message}, retrying in ${wait}s`);
    await sleep(wait);
    backoff = Math.min(backoff * 2, 60);
  }
}