Arm harness on our scheduler + start closing-line capture (capture only)

PART A — HARNESS ARMED ON OUR OWN INFRA. snapshotScheduler now runs the nightly backtest at HARNESS_HOUR_UTC (default 14), appends to harness_results, and pages via opsWatch.harnessStaleAlarm — a validator that stops running looks exactly like one that keeps passing. No external dependency: the join is plain SQL through the service client and the harness is a pure function. POST /api/internal/harness/run induces the same code path on demand, because a scheduled mechanism is verified by inducing it, never by waiting for a slot. PART B PHASE 0 — GATE PASSED for what is capturable: - C4 diagnosed: closing_line is ONE overwritable field with no timestamp and no provenance. captureClosing writes the current line and, when a prop fails to match, silently leaves the earlier value (= the lock) in place — so "captured a real close" is indistinguishable from "never updated". It is 92% equal, not 100%: 56 rows DID record movement, so the defect is provenance, not the value. - Feeds: normalized props already carry BOTH raw side prices per book, with game_time, and the intraday refresh polls every ~20 min during slate hours — so the last observable pre-lock line is available. - SHARP close: pinnacle is in ALLOWED_BOOKS -> a no-vig reference is capturable ("beat the market"). - ODAWA: NOT capturable. 'odawa' exists only as a UI preference option in onboarding/settings; it is in no adapter, no ALLOWED_BOOKS, no feed. An un-capturable source is a finding, not a gap to paper over. - JOIN: must drop `line` from the natural key, because a close that MOVED off the graded line is the entire point of CLV. Verified safe — all 164 current identity groups have exactly ONE line per (sport, player_key, stat, side, game_date). Zero ambiguity. PART B PHASE 1 — CAPTURE ONLY, built test-first. The refusal was proven before the capture logic existed: unbound game_time, doubleheader ambiguity, a missed pre-lock window, or a one-sided price all record missed_reason with NO price. A stale or mid-day line substituted for a close would manufacture a CLV proof from a number that was never the close. migration 029 closing_captures: append-only, never overwritten (that is the provenance C4 lacked), BOTH raw side prices so the existing de-vig engine can compute a fair closing probability later, sharp vs book line types kept distinct. Wired into the intraday refresh with a capture-rate alarm — a missed close is unrecoverable. NO CLV metric built, as ordered. This starts the clock. Suite 284/3417 green, build exit 0. Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01SmNjJAwEnqHPtXbvSZR8kA
2026-07-20 12:05:16 -04:00
parent e809a0eb3c
commit 77a58e4113
6 changed files with 499 additions and 0 deletions
@@ -0,0 +1,118 @@
+'use strict';
+
+/**
+ * HARNESS RUNNER (Session 64) — joins, runs the backtest, appends the result.
+ *
+ * Runs on OUR scheduler, in our container. No external dependency: the join is
+ * plain SQL through the service client, the harness is a pure function, and the
+ * result lands in `harness_results`.
+ *
+ * THE JOIN: model_snapshots (the model's INPUTS + prediction) → ledger_entries
+ * (the single source of truth for OUTCOMES) on the natural key
+ * (sport, player_key, stat, line, side, game_date). Outcomes are never
+ * denormalized onto snapshots.
+ *
+ * Rows that don't join are EXPECTED: retention stores both sides of every prop
+ * plus refusals, while the ledger keeps only the graded side of graded props.
+ */
+
+const harness = require('./backtestHarness');
+
+const HARNESS_VERSION = 'harness@2026-07-20';
+
+/** Pull joined rows. Read-only. */
+async function fetchJoinedRows(deps = {}) {
+  const getClient = deps.getClient || require('../utils/supabase').getSupabaseServiceClient;
+  const sb = getClient();
+  if (!sb) return { rows: [], skipped: true };
+
+  // Supabase's client cannot express this join, so it runs as two reads and is
+  // joined in memory — the volumes here are hundreds of rows, not millions.
+  const [{ data: snaps }, { data: ledger }] = await Promise.all([
+    sb.from('model_snapshots')
+      .select('sport, model_version, grade, grade_11, p_win, player_key, stat, line, side, game_date, quarantine_reason, refused')
+      .eq('refused', false)
+      .limit(20000),
+    sb.from('ledger_entries')
+      .select('sport, player_key, stat, line, side, game_date, outcome, quarantine_reason')
+      .is('user_id', null)
+      .limit(20000),
+  ]);
+
+  const key = (r) => `${r.sport}|${r.player_key}|${r.stat}|${Number(r.line)}|${r.side}|${r.game_date}`;
+  const byKey = new Map();
+  for (const l of ledger || []) byKey.set(key(l), l);
+
+  const rows = [];
+  for (const s of snaps || []) {
+    const l = byKey.get(key(s));
+    if (!l) continue;                 // expected: unselected side / refusal
+    rows.push({
+      sport: s.sport,
+      model_version: s.model_version,
+      grade: s.grade,
+      grade_11: s.grade_11,
+      p_win: s.p_win,
+      outcome: l.outcome,
+      quarantine_reason: l.quarantine_reason,
+      snap_quarantine: s.quarantine_reason,
+    });
+  }
+  return { rows, skipped: false, joined: rows.length, snapshots: (snaps || []).length };
+}
+
+/**
+ * Run the harness and append the result. Never throws — a failed harness run
+ * must not break the scheduler tick, and the staleness alarm is what surfaces
+ * a run that stops happening.
+ */
+async function runAndRecord(deps = {}) {
+  try {
+    const { rows, skipped, joined, snapshots } = await fetchJoinedRows(deps);
+    if (skipped) return { skipped: true, reason: 'no supabase client' };
+
+    const report = harness.runBacktest(rows, {});
+    const getClient = deps.getClient || require('../utils/supabase').getSupabaseServiceClient;
+    const sb = getClient();
+
+    const record = {
+      verdict: report.verdict,
+      can_validate: report.can_validate,
+      min_sample: report.min_sample,
+      scored: report.counts.scored,
+      excluded: {
+        quarantine: report.counts.excluded_quarantine,
+        terminal: report.counts.excluded_terminal,
+        pending: report.counts.excluded_pending,
+        push: report.counts.excluded_push,
+      },
+      report,
+      harness_version: HARNESS_VERSION,
+    };
+    const { error } = await sb.from('harness_results').insert([record]);
+    return {
+      ok: !error,
+      error: error ? error.message : null,
+      verdict: report.verdict,
+      scored: report.counts.scored,
+      joined,
+      snapshots,
+    };
+  } catch (e) {
+    return { ok: false, error: e && e.message ? e.message : String(e) };
+  }
+}
+
+/** Newest run timestamp, for the staleness alarm. */
+async function lastRunAt(deps = {}) {
+  try {
+    const getClient = deps.getClient || require('../utils/supabase').getSupabaseServiceClient;
+    const sb = getClient();
+    if (!sb) return null;
+    const { data } = await sb.from('harness_results')
+      .select('ran_at').order('ran_at', { ascending: false }).limit(1);
+    return data && data[0] ? data[0].ran_at : null;
+  } catch { return null; }
+}
+
+module.exports = { runAndRecord, fetchJoinedRows, lastRunAt, HARNESS_VERSION };