Store Pinnacle odds in SQLite: a Node.js poller with price history

-- schema.sql
PRAGMA journal_mode = WAL;

CREATE TABLE IF NOT EXISTS events (
  event_id    INTEGER PRIMARY KEY,
  sport_id    INTEGER NOT NULL,
  league_id   INTEGER,
  league_name TEXT,
  home        TEXT NOT NULL,
  away        TEXT NOT NULL,
  starts      TEXT NOT NULL,            -- ISO 8601 UTC from the API
  event_type  TEXT NOT NULL,            -- 'prematch' | 'live'
  last        INTEGER NOT NULL,         -- per-event cursor from the API
  updated_at  INTEGER NOT NULL          -- unix ms, our clock
);
CREATE INDEX IF NOT EXISTS events_starts ON events (starts);

-- current price per selection; one row per (event, period, market, side, points)
CREATE TABLE IF NOT EXISTS prices (
  event_id  INTEGER NOT NULL REFERENCES events (event_id) ON DELETE CASCADE,
  period    INTEGER NOT NULL,           -- 0 = full match
  market    TEXT    NOT NULL,           -- money_line | spreads | totals | team_total
  side      TEXT    NOT NULL,           -- home | away | draw | over | under
  points    REAL,                       -- spread (home side) or total line, NULL for money_line
  price     REAL    NOT NULL,           -- decimal
  seen_at   INTEGER NOT NULL,
  PRIMARY KEY (event_id, period, market, side, points)
);

-- append-only: a row every time a price differs from the stored one
CREATE TABLE IF NOT EXISTS price_history (
  id        INTEGER PRIMARY KEY,
  event_id  INTEGER NOT NULL,
  period    INTEGER NOT NULL,
  market    TEXT    NOT NULL,
  side      TEXT    NOT NULL,
  points    REAL,
  price     REAL    NOT NULL,
  seen_at   INTEGER NOT NULL
);
CREATE INDEX IF NOT EXISTS hist_sel ON price_history (event_id, period, market, side, points, seen_at);

CREATE TABLE IF NOT EXISTS cursors (
  sport_id INTEGER PRIMARY KEY,
  last     INTEGER NOT NULL,
  full_at  INTEGER NOT NULL             -- when we last took a full snapshot
);

// poller.mjs  -  Node >= 22.13 (node:sqlite), pinnodds SDK
import { DatabaseSync } from "node:sqlite";
import { readFileSync } from "node:fs";
import { setTimeout as sleep } from "node:timers/promises";
import { Client, SPORTS, AuthError, RateLimitError, PinnoddsError } from "pinnodds";

const api = new Client(process.env.PINNODDS_KEY);
const db = new DatabaseSync(process.env.DB_PATH ?? "./odds.db");
db.exec(readFileSync(new URL("./schema.sql", import.meta.url), "utf8"));

const SPORT_IDS = (process.env.SPORT_IDS ?? String(SPORTS.soccer)).split(",").map(Number);
const PLAN = process.env.PLAN ?? "trial";                 // trial | pro | scale
const RESYNC_AFTER_MS = 30 * 60 * 1000;                   // full snapshot at least every 30 min

// Budget: stay well under the plan's published limit.
// Trial 20/min & 100/day, Pro 10 req/s, Scale 30 req/s.
const INTERVAL_MS = { trial: 15 * 60 * 1000, pro: 5_000, scale: 2_000 }[PLAN];

const upEvent = db.prepare(`
  INSERT INTO events (event_id, sport_id, league_id, league_name, home, away, starts, event_type, last, updated_at)
  VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
  ON CONFLICT(event_id) DO UPDATE SET league_name=excluded.league_name, starts=excluded.starts,
    event_type=excluded.event_type, last=excluded.last, updated_at=excluded.updated_at`);
const getPrice = db.prepare(`SELECT price FROM prices WHERE event_id=? AND period=? AND market=? AND side=? AND points IS ?`);
const upPrice = db.prepare(`
  INSERT INTO prices (event_id, period, market, side, points, price, seen_at) VALUES (?, ?, ?, ?, ?, ?, ?)
  ON CONFLICT(event_id, period, market, side, points) DO UPDATE SET price=excluded.price, seen_at=excluded.seen_at`);
const addHist = db.prepare(`INSERT INTO price_history (event_id, period, market, side, points, price, seen_at) VALUES (?, ?, ?, ?, ?, ?, ?)`);
const getCursor = db.prepare(`SELECT last, full_at FROM cursors WHERE sport_id=?`);
const setCursor = db.prepare(`INSERT INTO cursors (sport_id, last, full_at) VALUES (?, ?, ?)
  ON CONFLICT(sport_id) DO UPDATE SET last=excluded.last, full_at=excluded.full_at`);

// Flatten one period object into (market, side, points, price) rows.
// money_line: { home, away, draw? }. spreads: { "<hdp>": { hdp, home, away, max } } (home side).
// totals: { "<points>": { points, over, under, max } }.
function* selections(period) {
  const ml = period.money_line ?? {};
  for (const side of ["home", "draw", "away"]) if (ml[side] != null) yield ["money_line", side, null, ml[side]];
  for (const s of Object.values(period.spreads ?? {})) {
    if (s.home != null) yield ["spreads", "home", s.hdp, s.home];
    if (s.away != null) yield ["spreads", "away", s.hdp, s.away];
  }
  for (const t of Object.values(period.totals ?? {})) {
    if (t.over  != null) yield ["totals", "over",  t.points, t.over];
    if (t.under != null) yield ["totals", "under", t.points, t.under];
  }
}

// node:sqlite has no transaction helper, so BEGIN/COMMIT are explicit.
function apply(board, sportId, now) {
  db.exec("BEGIN");
  try {
    let changed = 0;
    for (const ev of board.events) {
      upEvent.run(ev.event_id, ev.sport_id, ev.league_id ?? null, ev.league_name ?? null, ev.home, ev.away,
                  ev.starts, ev.event_type, ev.last, now);
      for (const [key, period] of Object.entries(ev.periods ?? {})) {
        const pnum = period.number ?? Number(key.replace("num_", ""));
        for (const [market, side, points, price] of selections(period)) {
          const cur = getPrice.get(ev.event_id, pnum, market, side, points);
          if (cur && cur.price === price) continue;
          upPrice.run(ev.event_id, pnum, market, side, points, price, now);
          addHist.run(ev.event_id, pnum, market, side, points, price, now);
          changed++;
        }
      }
    }
    db.exec("COMMIT");
    return changed;
  } catch (e) { db.exec("ROLLBACK"); throw e; }
}

async function pollSport(sportId) {
  const now = Date.now();
  const c = getCursor.get(sportId);
  const full = !c || now - c.full_at > RESYNC_AFTER_MS;      // cursors age fast: resync periodically
  const board = await api.markets({ sportId, eventType: "prematch", ...(full ? {} : { since: c.last }) });
  const changed = apply(board, sportId, now);
  setCursor.run(sportId, board.last, full ? now : c.full_at);
  console.log(new Date(now).toISOString(), `sport ${sportId} ${full ? "FULL" : "delta"}: ${board.events.length} events, ${changed} price changes, cursor ${board.last}`);
}

for (;;) {
  for (const sportId of SPORT_IDS) {
    try {
      await pollSport(sportId);
    } catch (err) {
      if (err instanceof AuthError) { console.error("bad key"); process.exit(1); }
      if (err instanceof RateLimitError) { console.warn("429, sleeping", err.retryAfter, "s"); await sleep(err.retryAfter * 1000); continue; }
      if (err instanceof PinnoddsError) { console.warn("api error:", err.message); continue; }
      throw err;
    }
  }
  await sleep(INTERVAL_MS);
}

-- Opening price, price at a chosen time, and closing price for one selection.
-- "Closing" = last price we saw before kickoff. Requires the poller to run up to kickoff.
WITH sel AS (
  SELECT h.*, e.starts
  FROM price_history h JOIN events e USING (event_id)
  WHERE h.event_id = :event_id AND h.period = 0 AND h.market = 'money_line' AND h.side = 'home'
    AND h.seen_at <= strftime('%s', e.starts) * 1000
)
SELECT
  (SELECT price FROM sel ORDER BY seen_at ASC  LIMIT 1) AS opening,
  (SELECT price FROM sel WHERE seen_at <= :bet_time ORDER BY seen_at DESC LIMIT 1) AS at_bet,
  (SELECT price FROM sel ORDER BY seen_at DESC LIMIT 1) AS closing;

-- CLV in implied-probability points for a bet taken at price p_bet:
-- clv_pp = (1/closing - 1/p_bet) * 100   -> positive means you beat the close

npm init -y && npm pkg set type=module && npm install pinnodds
export PINNODDS_KEY=...  PLAN=trial  SPORT_IDS=1
node poller.mjs

# inspect
sqlite3 odds.db "SELECT count(*) FROM events; SELECT count(*) FROM price_history;"