vyndr/scripts/settle-model-snapshots.js

#!/usr/bin/env node
'use strict';

/**
 * settle-model-snapshots — pay the standing debt.
 *
 * 71,192 snapshot rows have never carried an outcome. They are the retention
 * table built for exactly this kind of replay, and until they are settled every
 * measurement in this programme runs on the far smaller ledger slice.
 *
 * ── OUTCOME IS SIDE-ALIGNED, NOT RAW ─────────────────────────────────────
 * The order specifies `outcome = 1[realized > line]`. That is the OVER
 * perspective, and it would be backwards for every under-side prop — `p_win` is
 * side-aligned (verified: TB mean p_win 0.5698 against a 0.5074 side-won rate),
 * so a raw over-indicator would silently invert the target on the under rows and
 * make calibration measure the wrong thing.
 *
 * So: `actual_value` stores the realized stat (raw, unopinionated) and `outcome`
 * stores whether the GRADED SIDE won. Deviation from the literal order, stated
 * because it changes the number.
 *
 * ── INTEGRITY (hard-fail) ────────────────────────────────────────────────
 *   conservation   settled + unresolvable + orphaned == candidates
 *   no dupes       one write per snapshot id
 *   no orphans     a settled row must have matched a real box score
 *   prediction-time logging  captured_at must PRECEDE the game date; a row
 *                  logged after the fact is not a prediction and is refused
 *
 *   node scripts/settle-model-snapshots.js            # dry run, verifies only
 *   SETTLE_WRITE=1 node scripts/settle-model-snapshots.js
 */

require('dotenv').config();
const fs = require('fs');
const path = require('path');
const axios = require('axios');
const { createClient } = require('@supabase/supabase-js');
const { nameKey } = require('../src/utils/playerName');
const { knownNumber } = require('../src/utils/known');

const SB_URL = process.env.SUPABASE_URL;
const SB_KEY = process.env.SUPABASE_SERVICE_ROLE_KEY || process.env.SUPABASE_SERVICE_KEY;
const WRITE = process.env.SETTLE_WRITE === '1';
const BOX_CACHE = path.join(process.cwd(), '.seq-cache', 'batting-lines.json');
const STATS = ['hits', 'total_bases', 'rbi', 'runs'];
const PAGE = 1000;

/** Realized value per stat, from the box-score batting line. */
const FIELD = Object.freeze({
  hits: (b) => knownNumber(b.hits),
  total_bases: (b) => knownNumber(b.totalBases),
  rbi: (b) => knownNumber(b.rbi),
  runs: (b) => knownNumber(b.runs),
});

const get = async (url) => (await axios.get(url, { timeout: 45_000 })).data;

/** Eastern first pitch, conservatively. Anything at or after this is in-game. */
const FIRST_PITCH_ET_HOUR = 19;

/**
 * Was this row logged BEFORE the games it grades?
 *
 * The pipeline runs on UTC cron hours, so a 01:00-UTC cycle is 21:00 the
 * PREVIOUS evening in Eastern -- same game date, three hours into the slate.
 */
function isPreGame(capturedAt, gameDate) {
  if (!capturedAt || !gameDate) return false;
  const cap = new Date(capturedAt);
  if (Number.isNaN(cap.getTime())) return false;
  const et = new Date(cap.getTime() - 4 * 3600 * 1000);        // EDT
  const etDate = et.toISOString().slice(0, 10);
  if (etDate < String(gameDate)) return true;                   // day before, fine
  if (etDate > String(gameDate)) return false;                  // day after, post-game
  return et.getUTCHours() < FIRST_PITCH_ET_HOUR;
}

async function pool(items, fn, n = 6) {
  const out = []; let i = 0;
  await Promise.all(Array.from({ length: n }, async () => {
    while (i < items.length) {
      const idx = i; i += 1;
      try { out[idx] = await fn(items[idx]); } catch { out[idx] = null; }
    }
  }));
  return out.filter(Boolean);
}

async function page(sb, table, select, apply) {
  const out = [];
  for (let from = 0; ; from += PAGE) {
    // STABLE ORDER. model_snapshots is a LIVE table -- the snapshot cron writes
    // to it at 14/19/22/1/3 UTC -- and an unordered .range() walk over a table
    // being appended to returns overlapping pages. The integrity gate caught
    // exactly that on the first run.
    const { data, error } = await apply(sb.from(table).select(select))
      .order('id', { ascending: true })
      .range(from, from + PAGE - 1);
    if (error) throw error;
    if (!data || data.length === 0) break;
    out.push(...data);
    if (data.length < PAGE) break;
  }
  return out;
}

/** Box-score batting lines for a date range, cached. */
async function battingLines(dates) {
  if (fs.existsSync(BOX_CACHE)) {
    const c = JSON.parse(fs.readFileSync(BOX_CACHE, 'utf8'));
    if (dates.every((d) => c.dates.includes(d))) return c.lines;
  }
  const games = [];
  for (const d of dates) {
    try {
      const s = await get(`https://statsapi.mlb.com/api/v1/schedule?sportId=1&date=${d}`);
      for (const day of s.dates || []) {
        for (const g of day.games || []) {
          if (String(g.status && g.status.detailedState) === 'Final') {
            games.push({ pk: g.gamePk, date: g.officialDate || d });
          }
        }
      }
    } catch { /* absent day */ }
  }
  console.error(`[settle] ${games.length} final games across ${dates.length} dates`);

  const lines = {};
  const loaded = await pool(games, async (g) => {
    const box = await get(`https://statsapi.mlb.com/api/v1/game/${g.pk}/boxscore`);
    const out = [];
    for (const side of ['home', 'away']) {
      const t = box.teams[side];
      if (!t) continue;
      for (const id of t.batters || []) {
        const pl = t.players[`ID${id}`];
        const b = pl && pl.stats && pl.stats.batting;
        if (!b || b.atBats == null) continue;      // did not bat -> absent, not zero
        out.push({
          date: g.date,
          key: nameKey(pl.person && pl.person.fullName),
          name: pl.person && pl.person.fullName,
          gamePk: g.pk,
          hits: b.hits, totalBases: b.totalBases, rbi: b.rbi, runs: b.runs, atBats: b.atBats,
        });
      }
    }
    return out;
  });
  for (const arr of loaded) for (const r of arr) {
    const k = `${r.date}|${r.key}`;
    // A doubleheader gives two lines; sum them — the prop covers the day.
    if (!lines[k]) lines[k] = { ...r, games: 1 };
    else {
      lines[k].hits += r.hits; lines[k].totalBases += r.totalBases;
      lines[k].rbi += r.rbi; lines[k].runs += r.runs; lines[k].atBats += r.atBats;
      lines[k].games += 1;
    }
  }
  fs.mkdirSync(path.dirname(BOX_CACHE), { recursive: true });
  fs.writeFileSync(BOX_CACHE, JSON.stringify({ dates, lines }));
  return lines;
}

async function main() {
  const sb = createClient(SB_URL, SB_KEY, { auth: { persistSession: false } });

  const snaps = await page(sb, 'model_snapshots',
    'id, game_date, captured_at, stat, player_key, player_name, line, side, p_win, refused, outcome',
    (q) => q.eq('sport', 'mlb').in('stat', STATS).is('outcome', null));
  console.error(`[settle] ${snaps.length} unsettled snapshot rows`);

  const dates = [...new Set(snaps.map((r) => r.game_date))].sort();
  const lines = await battingLines(dates);

  const counts = { candidates: snaps.length, settled: 0, unresolvable: 0, orphaned: 0, post_hoc_logged: 0 };
  const updates = [];
  const seenIds = new Set();

  for (const s of snaps) {
    // Belt and braces: ordered pagination should make this impossible, and a
    // duplicate would double-count a prediction in every downstream measurement.
    if (seenIds.has(s.id)) throw new Error(`INTEGRITY: duplicate snapshot id ${s.id}`);
    seenIds.add(s.id);

    // A row logged after first pitch is not a prediction.
    //
    // Measured: cycles at ET 21:00/22:00/23:00 on the game date (10,738 rows)
    // were captured DURING or AFTER the games they grade, and a further 664 the
    // following morning. Games start ~19:05 ET, so the honest cutoff is ET
    // first pitch on the game date -- not a UTC date compare, which both keeps
    // post-game 01:00-UTC rows and discards legitimate pre-dawn ones.
    if (!isPreGame(s.captured_at, s.game_date)) {
      counts.post_hoc_logged += 1; counts.unresolvable += 1; continue;
    }
    const line = knownNumber(s.line);
    if (line === null || !s.side) { counts.unresolvable += 1; continue; }

    const b = lines[`${s.game_date}|${s.player_key}`];
    if (!b) { counts.orphaned += 1; continue; }

    const realized = FIELD[s.stat](b);
    if (realized === null) { counts.unresolvable += 1; continue; }

    // SIDE-ALIGNED, so it matches how p_win is expressed.
    const over = realized > line;
    const won = String(s.side).toLowerCase() === 'under' ? !over : over;
    updates.push({ id: s.id, outcome: won ? 'hit' : 'miss', actual_value: realized });
    counts.settled += 1;
  }

  // CONSERVATION — hard fail.
  const acc = counts.settled + counts.unresolvable + counts.orphaned;
  if (acc !== counts.candidates) {
    throw new Error(`INTEGRITY: conservation violated ${acc} != ${counts.candidates}`);
  }

  // Hand-verifiable sample.
  const sample = updates.slice(0, 12).map((u) => {
    const s = snaps.find((x) => x.id === u.id);
    return { player: s.player_name, date: s.game_date, stat: s.stat, line: s.line, side: s.side,
      realized: u.actual_value, outcome: u.outcome };
  });

  if (WRITE) {
    let written = 0;
    for (let i = 0; i < updates.length; i += 500) {
      const batch = updates.slice(i, i + 500);
      const results = await Promise.all(batch.map((u) => sb.from('model_snapshots')
        .update({ outcome: u.outcome, actual_value: u.actual_value, settled_at: new Date().toISOString(), settlement_source: 'statsapi_boxscore' })
        .eq('id', u.id).is('outcome', null)));
      written += results.filter((r) => !r.error).length;
    }
    counts.written = written;
  }

  console.log(JSON.stringify({
    mode: WRITE ? 'WRITE' : 'DRY RUN',
    counts,
    dates_before: 'ledger-only slice',
    snapshot_dates: dates.length,
    date_span: [dates[0], dates[dates.length - 1]],
    hand_verify_sample: sample,
    note: 'outcome is SIDE-ALIGNED (matches p_win); actual_value holds the raw realized stat',
  }, null, 2));
  process.exit(0);
}

main().catch((e) => { console.error(e); process.exit(1); });