/** * snapshotService — scheduled grade pipeline (Session 45). * * Orchestrates ONE snapshot cycle for one sport. The on-demand "Read" model is * retired: a snapshot pre-grades the full slate, LOCKS each grade to the line at * snapshot time (`gradedAt`), classifies each player's archetype, computes line * deltas vs the previous snapshot, and emits ticker events. * * Everything is orchestration of EXISTING services (oddsService, * gradeSlateService, archetypeService, playerIntelService). All I/O is injectable * so the whole cycle is unit-testable with zero network. * * Redis keys written: * snapshot:{sport}:latest — current locked snapshot { sport, updated_at, grades, deltas } * snapshot:{sport}:previous — prior snapshot (for the next delta computation) * grades:{sport} — { grades, updated_at, source } (GameCard / Explore / leaders) * ticker:items — capped array of ticker events (newest first) */ // Session 60 (AUTONOMY P0) — 24h, NOT 6h. The overnight cron gap is 11 hours // (03:00 UTC → 14:00 UTC); with a 6h TTL the morning settle pass read an // EXPIRED snapshot and the accuracy loop silently settled nothing. The // snapshot must survive until the morning-after settle reads it. const SNAP_TTL = 24 * 3600; const TICKER_TTL = 24 * 3600; const TICKER_CAP = 50; const DELTA_NOISE = 0.5; // ignore movements smaller than this const DELTA_MOVE = 1.0; // ticker MOVE threshold const STATS_CONCURRENCY = 5; const { nameKey, normalizeName } = require('../utils/playerName'); // Session 46 — group/dedupe by the normalized name key so "A.J. Ewing" and // "AJ Ewing" (or "Jazz Chisholm" / "Jazz Chisholm Jr.") collapse to one player. const norm = (s) => nameKey(s); const lastName = (full) => { const parts = String(full || '').trim().split(/\s+/); return parts.length > 1 ? parts[parts.length - 1] : (parts[0] || ''); }; const sideChar = (dir) => (String(dir || 'over').toLowerCase() === 'under' ? 'U' : 'O'); const propKey = (g) => `${norm(g.player || g.player_name)}|${String(g.stat_type || g.stat || '').toLowerCase()}|${String(g.direction || '').toLowerCase()}`; async function mapLimit(items, concurrency, fn) { const out = new Array(items.length); let i = 0; async function worker() { while (i < items.length) { const idx = i++; out[idx] = await fn(items[idx], idx); } } await Promise.all(Array.from({ length: Math.min(concurrency, items.length || 1) }, worker)); return out; } /** Index original odds props by propKey-ish (player|stat) for odds lookup. * Session 61 — prefer a book row with BOTH sides priced: the first-seen row * sometimes carried only one side (e.g. betmgm SB unders with no juice), * which locked a NULL odds even though another book priced it. Still a * REAL book row, never synthesized. */ /** * Index the odds rows the LOCKED PRICE is read from. * * TAKEABLE-ONLY (2026-08-02). This used to index the FULL props list, which * became the display-widened list on 2026-08-01 — so `gradedAt.odds`, the price * a grade is locked at, could be a DFS pick'em or exchange price. Measured: the * ledger's non-takeable share went 0% -> 47.9% overnight. * * The lock price must be one a bettor could ACTUALLY have taken, so this gates * on TAKEABLE_BOOKS — not MODEL_BOOKS. `pinnacle` is model-eligible and * deliberately NOT takeable, so gating on MODEL would re-break this the moment * pinnacle's feed recovers. * * No takeable quote → the key is ABSENT and the price stays null. An honest * missing price beats a price from a book you cannot bet. */ function indexOdds(props) { const { isTakeableBook } = require('../config/bookRoles'); const map = {}; const bothSides = (p) => p && p.over_odds != null && p.under_odds != null; for (const p of props || []) { if (!isTakeableBook(p && p.book)) continue; const k = `${norm(p.player)}|${String(p.stat_type || '').toLowerCase()}`; if (!map[k] || (!bothSides(map[k]) && bothSides(p))) map[k] = p; } return map; } function gradedAtFor(g, oddsByKey, ts) { const k = `${norm(g.player || g.player_name)}|${String(g.stat_type || g.stat || '').toLowerCase()}`; const o = oddsByKey[k]; let odds = null; if (o) { odds = String(g.direction || '').toLowerCase() === 'under' ? (o.under_odds ?? o.under ?? o.odds ?? null) : (o.over_odds ?? o.over ?? o.odds ?? null); } return { line: g.line, odds, timestamp: ts }; } /** * Compare current grades to the previous snapshot's locked lines. A delta is * emitted only when |movement| >= DELTA_NOISE. `direction`: * 'toward' = market moving in the direction of our graded side (confirming) * 'away' = market moving against it. * For an OVER, a rising line confirms (toward); for an UNDER, a falling line. */ function computeLineDeltas(current, previous) { const prevMap = {}; for (const p of previous || []) prevMap[propKey(p)] = p; // Session 52 — opt-in verification log (SNAPSHOT_DEBUG=1). Confirms the delta // pipeline has a previous snapshot to diff against; off by default (hot path). if (process.env.SNAPSHOT_DEBUG === '1') { console.log(`[deltas] diffing ${(current || []).length} current vs ${(previous || []).length} previous locked lines`); } const out = []; for (const c of current || []) { const prev = prevMap[propKey(c)]; if (!prev) continue; const gradedLine = prev.gradedAt ? prev.gradedAt.line : prev.line; const currentLine = c.line; if (gradedLine == null || currentLine == null) continue; const delta = +(Number(currentLine) - Number(gradedLine)).toFixed(2); if (Math.abs(delta) < DELTA_NOISE) continue; const side = String(c.direction || 'over').toLowerCase(); const toward = side === 'over' ? delta > 0 : delta < 0; out.push({ player: c.player || c.player_name, stat: c.stat_type || c.stat, side: sideChar(side), gradedLine: Number(gradedLine), currentLine: Number(currentLine), delta, direction: toward ? 'toward' : 'away', grade: c.grade, }); } return out; } const isTopGrade = (g) => g === 'A+' || g === 'A'; /** * Build ticker events from a snapshot: a SCAN summary, GRADE events for the top * grades, and MOVE events for significant deltas. Newest-relevant first. */ function generateTickerEvents(sport, grades, deltas, ts) { const events = []; events.push({ // VERB LAW — the verb is READ, never SCAN. One READ event per sport (deduped). tag: 'READ', color: 'var(--g-a)', ts, sport, text: `${sport.toUpperCase()} slate read · ${grades.length} props graded`, }); for (const g of grades.filter((x) => isTopGrade(x.grade)).slice(0, 6)) { const arch = g.archetype ? `${g.archetype} ` : ''; events.push({ tag: g.grade, color: g.grade === 'A+' ? 'var(--g-ap)' : 'var(--g-a)', ts, text: `${arch}${lastName(g.player || g.player_name)} ${g.stat_type || g.stat} ${sideChar(g.direction)}${g.line} graded ${g.grade}`, }); } for (const d of deltas.filter((x) => Math.abs(x.delta) >= DELTA_MOVE).slice(0, 6)) { const arrow = d.delta > 0 ? '▲' : '▼'; const s = d.side === 'U' ? 'u' : 'o'; events.push({ tag: 'MOVE', color: 'var(--amber)', ts, text: `${lastName(d.player)} ${s}${d.gradedLine} → ${s}${d.currentLine} ${arrow}${d.delta > 0 ? '+' : ''}${d.delta}`, }); } return events; } // Session 47 — a READ event's sport, from the event field or its text prefix // (defends ticker items written before the `sport` field existed). Accepts the // legacy 'SCAN'/'slate scanned' shape too so cached items dedupe cleanly through // the verb-law rollover (the ticker regenerates as READ on the next snapshot). function readSportOf(e) { if (e.tag !== 'READ' && e.tag !== 'SCAN') return null; if (e.sport) return String(e.sport).toLowerCase(); const m = String(e.text || '').match(/^([a-z]+)\s+slate (?:read|scanned)/i); return m ? m[1].toLowerCase() : null; } async function pushTickerItems(events, deps) { if (!events || events.length === 0) return; const existing = await deps.cacheGet('ticker:items'); const arr = Array.isArray(existing) ? existing : []; // Keep only the LATEST READ per sport: drop existing READ events for any sport // that has a fresh READ in this batch. MOVE/GRADE events are time-specific and // preserved. const freshReadSports = new Set(events.map(readSportOf).filter(Boolean)); const pruned = arr.filter((e) => { const sp = readSportOf(e); return !(sp && freshReadSports.has(sp)); }); const merged = [...events, ...pruned].slice(0, TICKER_CAP); await deps.cacheSet('ticker:items', merged, TICKER_TTL); } // Session 60 (night2/B) — accumulate slate players' game logs into the // roster blob the streaks/hot-list engines read. Merge by nameKey (newer // entry wins), cap the blob, 72h TTL (a player off the slate for 3 days // ages out — honest churn, not a leak). const ROSTERLOGS_TTL = 72 * 3600; const ROSTERLOGS_CAP = 300; async function mergeRosterLogs(sport, entries, deps) { if (!entries || entries.length === 0) return; try { const key = `rosterlogs:${sport}`; const existing = await deps.cacheGet(key); const byKey = new Map(); for (const e of Array.isArray(existing) ? existing : []) byKey.set(nameKey(e.name), e); for (const e of entries) byKey.set(nameKey(e.name), e); // fresh resolve wins const merged = [...byKey.values()].slice(-ROSTERLOGS_CAP); await deps.cacheSet(key, merged, ROSTERLOGS_TTL); } catch (e) { console.warn(`[snapshot] rosterlogs merge failed for ${sport}:`, e.message); } } const ACTIVE_SPORTS = ['mlb', 'nba', 'wnba', 'soccer']; /** * Run one snapshot cycle for `sport`. Returns a summary; never throws. * opts (all injectable): getOdds, gradeAndCacheSlate, resolveStats, classify, * cacheGet, cacheSet, now, nowMs. */ /** * Statcast aggregates for the season, indexed by our player key. One read per * snapshot run (~1,350 rows / 5 MB), reused for every grade — the alternative * is a per-prop lookup inside a tight grading loop. */ async function loadStatcastRows(sport) { try { const sb = require('../utils/supabase').getSupabaseServiceClient(); if (!sb) return null; const { data, error } = await sb.from('statcast_aggregates') .select('*').eq('sport', sport).limit(5000); if (error || !data) return null; const map = new Map(); for (const r of data) { if (!r.player_key) continue; // A two-way player has two rows; the one with the larger sample is the // profile his props are about far more often than not. const prev = map.get(r.player_key); const size = Number(r.sample_pa || r.sample_ip || 0); const prevSize = prev ? Number(prev.sample_pa || prev.sample_ip || 0) : -1; if (!prev || size > prevSize) map.set(r.player_key, r); } return map; } catch { return null; } } /** * proj-v1 — opposing-pitcher arsenals by source_id (MLBAM id), classified once * per snapshot. Read-only; a failure returns an empty map (proj-v1 then has no * matchup contribution, honest-absent). Separate from loadStatcastRows (which * keys by player name) because the opposing pitcher is resolved by ID. */ async function loadPitcherArsenals(sport) { const out = new Map(); try { if (String(sport).toLowerCase() !== 'mlb') return out; const sb = require('../utils/supabase').getSupabaseServiceClient(); if (!sb) return out; const matchupRead = require('./projection/matchupRead'); const { data, error } = await sb.from('statcast_aggregates') .select('source_id,pitch_mix').eq('sport', sport).eq('role', 'pitcher').limit(5000); if (error || !data) return out; for (const r of data) { if (r.source_id == null || !r.pitch_mix) continue; const arsenal = matchupRead.classifyArsenal(r.pitch_mix); if (arsenal) out.set(Number(r.source_id), arsenal); } return out; } catch { return out; } } /** * Stats that may serve a calibrated number, and on what basis. * * ── THE BIAS IS ROBUST; THE MAP WAS NOT CERTIFIABLE ────────────────────── * Tested model-free and map-free on the picked-side population: the model * over-predicts its own favourites, and the sign survives 99.5% of date-block * resamples pooled and replicates in 4 of 4 stats. Over-prediction rises * monotonically from -0.008 near p=0.55 to +0.245 above 0.9. * * What could NOT be certified on 19 dates is the stability of a specific * isotonic MAP — LODO has 1.4-9.3% power there. So isotonic is retired and the * correction is a TWO-PARAMETER Platt curve, which has no capacity to encode a * single odd day, shrunk toward the raw forecast by fit-date count. * * Validated as a NEW estimator against RAW, date-block bootstrap: * hits a=0.406 shrink 0.565 0.2626 -> 0.2540 CI [-0.0112,-0.0069] * total_bases a=0.472 shrink 0.333 0.2490 -> 0.2429 CI [-0.0062,-0.0059] * * rbi is WITHDRAWN (deployed last order on isotonic): the low-parameter fit does * not beat raw, CI [-0.0007, 0] touching zero. runs is refused by the slope * guard — it fitted a = -0.032, which would invert the forecast rather than * flatten it. Both now serve RAW. */ const CALIBRATION_DEPLOYED = Object.freeze([]); /** * NOTHING IS SERVED CALIBRATED, AND THIS IS DELIBERATE. * * The low-parameter maps were fitted on the OLD forecast — the one whose base * rate was a ten-game frequency. That distribution no longer exists: the * champion now reads the full season log at a 0.20 recency weight, which tripled * its resolution on hits (0.00251 -> 0.00817) and roughly doubled it on * total_bases and runs. * * A calibration map applied to a forecast it was not fitted on is the stale-fit * trap this session has already been caught by once, and it corrects toward a * bias the new forecast may not have. `fromLedger` cannot rescue it either: the * settled ledger rows still carry OLD p_win values, so refitting today would fit * the retired forecast again. * * So calibration is OFF until enough dates settle under the repaired champion to * refit honestly, and the favourite-longshot bias must be re-measured on the new * forecast rather than assumed to have survived. Serving the raw repaired number * is the honest state, not a regression. * * The shadow duel is likewise void — it accumulated against the old forecast. */ const CALIBRATION_BASIS = Object.freeze({}); async function runSnapshot(sport, opts = {}) { const sp = String(sport || '').toLowerCase(); const deps = { getOdds: opts.getOdds || require('./oddsService').getOdds, gradeAndCacheSlate: opts.gradeAndCacheSlate || require('./gradeSlateService').gradeAndCacheSlate, resolveStats: opts.resolveStats || require('./playerIntelService').resolvePlayerStats, classify: opts.classify || require('./archetypeService').classify, cacheGet: opts.cacheGet || require('../utils/redis').cacheGet, cacheSet: opts.cacheSet || require('../utils/redis').cacheSet, now: opts.now || (() => new Date().toISOString()), nowMs: opts.nowMs || (() => Date.now()), // Session 56 — ops alerting (ntfy) + retry-once on a hard odds failure. notify: opts.notify || require('../utils/opsNotify').notify, sleep: opts.sleep || ((ms) => new Promise((r) => setTimeout(r, ms))), retryDelayMs: opts.retryDelayMs != null ? opts.retryDelayMs : 60_000, // Session 58 — Phase 1 truth infrastructure. ledger no-ops without // SUPABASE env, so tests / local dev never touch a database. ledger: opts.ledger || require('./ledgerService'), // Wave 2B — reliable ESPN athlete id + DIRECT headshot href from feeds the // pipeline already calls (schedule + summary). Fills the NBA/WNBA espnId gap // when the stats-resolve fallback misses. Returns {} for MLB / errors. buildEspnIndex: opts.buildEspnIndex || require('./espnAthleteIndex').buildEspnAthleteIndex, // Session 63 — the opponent-rank feed. Injectable so tests never hit ESPN; // under NODE_ENV=test it defaults to a no-op (the opsNotify precedent) so a // suite that doesn't know about this dep can never make a live ESPN call. // Session 64 — model-snapshot retention. Injectable; null disables it. retention: opts.retention !== undefined ? opts.retention : require('./retentionService'), // Book Comparison order (Phase 1) — DISPLAY-ONLY per-book price capture. // Fenced: written to its own `bookprices:{sport}` key, read by nothing on // the grade path. Injectable; a failure never touches grading. captureBookPrices: opts.captureBookPrices || require('./bookPriceStore').captureBookPrices, // Lock-line persistence (measurement-only). Persists multi-book lines to the DB at // the lock moment so a future audit can run the currently-BLOCKED staleness check. // Fenced: writes its own `lock_lines` table, read by nothing on the grade path. lockLineCapture: opts.lockLineCapture || require('./lockLineCapture'), refreshTeamStats: opts.refreshTeamStats || (process.env.NODE_ENV === 'test' ? async () => null : require('./intelligence/teamStatsCache').refreshTeamStats), }; const start = deps.nowMs(); const ts = deps.now(); // Session 56 — retry ONCE on a hard failure (thrown error / null response = a // transient PropLine/network blip). A successful-but-empty slate is NOT a // failure (off-hours), so it is not retried — that would waste quota + latency. let odds; try { odds = await deps.getOdds(sp); if (odds == null) throw new Error('null odds response'); } catch (e) { try { await deps.sleep(deps.retryDelayMs); odds = await deps.getOdds(sp); if (odds == null) throw new Error('null odds response (retry)'); } catch (e2) { await deps.notify(`❌ ${sp.toUpperCase()} snapshot FAILED: ${e2.message}`, { title: 'VYNDR pipeline', priority: 'high', tags: ['x'] }); return { sport: sp, status: 'error', reason: e2.message, gradeCount: 0 }; } } const props = (odds && Array.isArray(odds.props)) ? odds.props : []; if (props.length === 0) { await deps.notify(`⚠️ ${sp.toUpperCase()} snapshot: 0 props (odds unavailable)`, { title: 'VYNDR pipeline', priority: 'low', tags: ['warning'] }); return { sport: sp, status: 'skipped', reason: 'no props', gradeCount: 0 }; } // Session 64 (Order 1.5) — BIND EVERY PROP TO ITS REAL GAME before anything // downstream dates it. PropLine emits no commence_time, so ledgerService's // `dateET(prop.game_time) || dateET(gradedTs)` always fell through to the // GRADE timestamp — and a 01:00/03:00 UTC snapshot is the previous ET day, // so props for tonight were filed under yesterday. Everything downstream // (ledger, retention, settlement) reads prop.game_time, so fixing it here // fixes all of them at once. try { const binder = deps.gameBinder || require('./gameBinder'); const b = await binder.attachGameTimes(sp, props, { gradedAt: ts }); console.log(`[snapshot] game binding ${sp}: ${b.bound} bound, ${b.alreadyHad} already had times, ${b.unresolved} UNRESOLVED, ${b.ambiguous} ambiguous(doubleheader)`); if (b.unresolved > 0 && b.bound === 0 && b.alreadyHad === 0) { await deps.notify(`Game binding produced NOTHING for ${sp.toUpperCase()} — ${b.unresolved} props could not be tied to a scheduled game. Their rows will be skipped rather than mis-dated.`, { title: 'VYNDR pipeline', priority: 'high', tags: ['rotating_light'], }); } } catch (e) { console.warn(`[snapshot] game binding failed for ${sp} (rows without a real game time will be skipped):`, e.message); } // Book Comparison order (Phase 1) — CAPTURE PER-BOOK PRICES BEFORE DEDUP. // `props` still carries one row per book here (dedupeProps runs downstream // inside gradeAndCacheSlate). The capture only READS props and writes its own // `bookprices:{sport}` key at SNAP_TTL — long enough to outlive the cron gap // (the 1h odds cache would blank between runs). Best-effort and structurally // fenced: nothing on the grade path reads this key, and the graded slate is // byte-identical whether or not this runs (bookPriceStore.test.js locks it). try { const bookStore = deps.captureBookPrices(props, { now: deps.now }); await deps.cacheSet(`bookprices:${sp}`, bookStore, SNAP_TTL); console.log(`[snapshot] book prices ${sp}: ${bookStore.count} props with book rows captured`); } catch (e) { console.warn(`[snapshot] book-price capture failed for ${sp} (display-only, grading continues):`, e.message); } // Session 63 — REFRESH TEAM STATS BEFORE GRADING. // `refreshTeamStats` is the ONLY writer of `team_stats:{sport}:{abbr}`, which // is the ONLY source of `opp_rank_stat` — and it had zero production callers, // so that feature was permanently null and engine1's ±1.0 opponent-defense // factor could never fire. It is 24h-cached and rate-limited, so this is one // cheap ESPN pass per snapshot. Best-effort: a failure here must never break // the snapshot — the features simply stay absent, as before. try { const summary = await deps.refreshTeamStats(sp); if (summary && summary.captured != null) { console.log(`[snapshot] team stats refreshed for ${sp}: ${summary.captured} captured, ${summary.errored ?? 0} errored`); } } catch (e) { console.warn(`[snapshot] team stats refresh failed for ${sp} (grading continues):`, e.message); } // Session 64 — RETENTION (Phase 2, priority zero). Collect one row per graded // prop per SIDE — graded AND refused — with the feature vector that produced // it, so history compounds from tonight and a future model can be replayed // against the exact conditions this one faced. Refusals are included on // purpose: the ledger drops them, so a gate refusing props that would have // won is otherwise invisible. const retention = deps.retention; // Reuse the LEDGER's date + game-id helpers so retention rows share the // ledger's natural key exactly — otherwise the settle pass could never join // outcomes onto them. const ledgerInternals = (deps.ledger && deps.ledger.__internals) || require('./ledgerService').__internals; const retentionGameDate = ledgerInternals.dateET(ts) || ledgerInternals.dateET(new Date().toISOString()); const retentionCtx = { snapshotId: retention ? retention.newSnapshotId() : null, capturedAt: ts, cycleHourUtc: new Date(ts).getUTCHours(), sport: sp, gameDate: retentionGameDate, gameIdFor: (base) => ledgerInternals.gameIdFor(sp, base, retentionGameDate), }; const collector = retention ? retention.createCollector(retentionCtx) : null; // Grade the slate via the existing service; capture the envelope instead of // letting it write (we re-write an ENRICHED version below). let envelope = null; // ── PROVEN FACTORS, LOADED BEFORE THE GRADE ───────────────────────────── // This ordering IS the fix. The same inputs were previously read at line 640+, // downstream of the grade they should inform, so three proven factors never // once moved a served number. Best-effort: a failed load means the slate is // graded unadjusted, exactly as before. let factorContext = null; if (sp === 'mlb') { try { const ctxSvc = deps.hitsFactorContext || require('./model/hitsFactorContext'); const sbc = require('../utils/supabase').getSupabaseServiceClient(); factorContext = sbc ? await ctxSvc.build(sbc) : null; if (factorContext) console.log(`[factors] ${sp} hits context loaded — ${JSON.stringify(factorContext.__stats)}`); else console.log(`[factors] ${sp} — no factor context; grading unadjusted`); } catch (e) { console.warn('[factors] context skipped:', e.message); } } await deps.gradeAndCacheSlate(sp, props, { factorContext, // Bisect hook (2026-08-01): lets the internal trigger run a bounded slate // without a prod env change, so a cap regression can be isolated by // measurement instead of guessed at. Omitted => gradeSlateService's own // DEFAULT_LIMIT (the real production value). ...(Number(opts.limit) > 0 ? { limit: Number(opts.limit) } : {}), source: (odds && odds.provider) || 'odds-api', now: deps.now, cacheSet: async (_k, v) => { envelope = v; }, onGraded: collector ? collector.onGraded : undefined, }); // Retention is COLLECTED here (grade time — features must be exactly what the // model saw) but PERSISTED after enrichment below, so archetype/team/opponent // are filled in. `persistRetention` is called on BOTH exits, including the // empty-slate early return: a slate that graded nothing but refused // everything is exactly the case worth recording. let retentionRows = 0; const persistRetention = async (enrichedGrades) => { if (!retention || !collector || !collector.rows.length) return; try { const rows = retention.mergeEnrichment ? retention.mergeEnrichment(collector.rows, enrichedGrades || []) : collector.rows; const r = await retention.persist(rows); retentionRows = r.written || 0; console.log(`[snapshot] retention ${sp}: ${r.written}/${r.attempted} rows${r.skipped ? ' (skipped — no supabase env)' : ''}${r.error ? ` ERROR: ${r.error}` : ''}`); } catch (e) { console.warn(`[snapshot] retention write failed for ${sp} (snapshot continues):`, e.message); } }; const rawGraded = (envelope && Array.isArray(envelope.grades)) ? envelope.grades : []; if (rawGraded.length === 0) { await persistRetention([]); // refusal-only slate is still history return { sport: sp, status: 'skipped', reason: 'no grades', gradeCount: 0 }; } // Session 48 — normalize player display names + dedupe variant grades at the // SOURCE so every consumer (GameCard, Explore, leaders, profile) gets clean, // merged names. PropLine sends "Matt"/"Matthew", "A.J."/"AJ", "(STL)" tags as // separate players; collapse to ONE grade per normalized player + stat (keep // the highest-confidence; rawGraded is already confidence-desc). // Session 54 — also keep the RICHEST display per player (prefer the accented // variant: "José" over "Jose", then the longer string) so the prop rows match // the accented pitcher line. The GRADE picked is still the highest-confidence. const hasAccent = (s) => [...String(s)].some((c) => c.charCodeAt(0) > 127); const richerDisplay = (a, b) => { if (!b) return a; if (hasAccent(a) !== hasAccent(b)) return hasAccent(a) ? a : b; return a.length >= b.length ? a : b; }; const dedup = new Map(); const bestDisplay = new Map(); for (const g of rawGraded) { const disp = normalizeName(g.player || g.player_name).display || g.player || g.player_name || ''; const pk = nameKey(disp); bestDisplay.set(pk, richerDisplay(disp, bestDisplay.get(pk))); const k = `${pk}|${String(g.stat_type || g.stat || '').toLowerCase()}`; const cur = { ...g, player: disp, player_name: disp }; const prev = dedup.get(k); if (!prev || (Number(g.confidence) || 0) > (Number(prev.confidence) || 0)) dedup.set(k, cur); } const graded = [...dedup.values()].map((g) => { const disp = bestDisplay.get(nameKey(g.player)) || g.player; return { ...g, player: disp, player_name: disp }; }); // Archetype per unique player (pure math once we have stats). Best-effort — // a missing stat line → no badge (not a fallback archetype). const oddsByKey = indexOdds(props); const players = [...new Set(graded.map((g) => g.player || g.player_name).filter(Boolean))]; const archByPlayer = {}; // Session 59 — capture the player's REAL team from the same stats resolve // (statsapi/ESPN), never guessed. Feeds the ledger team/opponent columns // and the slate join guard (a prop only attaches to its own game). const teamByPlayer = {}; // Wave 2A — the REAL athlete id from the SAME stats resolve, keyed by player. // MLB → MLBAM id (mlbstatic headshot CDN); NBA/WNBA → ESPN athlete id // (a.espncdn headshot CDN). Stored on the enriched grade so it flows free to // grades:{sport} → slate strips → PlayerAvatar. Zero new I/O. Absent → the // component falls to a team-colored monogram (never a fabricated face). const playerIdByPlayer = {}; const espnIdByPlayer = {}; // Wave 1 (trust bug) — the prop's game participants become the resolve's // teamHint: it disambiguates namesake collisions (two "James Wood") and, when // the resolved player's real team isn't in the prop's game, the resolver drops // the team rather than tag a foreign one (the streaks/rosterlogs JOIN // INVARIANT, mirroring the S59 slate guard). Keyed by the normalized name. const teamHintByPlayer = {}; for (const p of props || []) { const k = norm(p.player); if (!k || teamHintByPlayer[k]) continue; const hint = [p.home_team, p.away_team].filter(Boolean); if (hint.length) teamHintByPlayer[k] = hint; } // Session 60 (night2/B) — THE STREAKS PRODUCER. The aggregator (streaks + // hot lists) starved because its data producers were all external and // unarmed (tank01-prefetch via n8n, the offline Python grading flow). // The stats resolve above already fetched each slate player's game log — // accumulate it into the `rosterlogs:{sport}` blob rosterLogs.loadRosterLogs // reads FIRST. Zero extra API calls; the pipeline now feeds its own // free layer. const logEntries = []; await mapLimit(players, STATS_CONCURRENCY, async (player) => { try { const teamHint = teamHintByPlayer[norm(player)] || null; const stats = await deps.resolveStats(player, sp, teamHint ? { teamHint } : {}); if (stats && stats.found) { const c = deps.classify(sp, stats.classifierInput || {}); archByPlayer[player] = c.primary ? c.primary.name : null; if (stats.team) teamByPlayer[player] = stats.team; // Wave 2A — capture the resolved athlete id (headshot thread). if (stats.playerId != null) playerIdByPlayer[player] = stats.playerId; if (stats.espnId != null) espnIdByPlayer[player] = stats.espnId; if (Array.isArray(stats.rawLog) && stats.rawLog.length > 0) { logEntries.push({ name: normalizeName(player).display || player, playerId: stats.playerId ?? null, team: stats.team || null, group: stats.group || null, seasonRaw: stats.seasonRaw || null, games: stats.rawLog, }); } } } catch { /* graceful — no badge */ } }); await mergeRosterLogs(sp, logEntries, deps); // Wave 2B — the RELIABLE espnId/headshot source. The stats-resolve espnId // above comes only from espnStatsAdapter (the offline-Python fallback), which // is flaky in prod. ESPN's own schedule→summary feeds (already free, already // called elsewhere) carry each athlete's id AND often a DIRECT headshot href. // Build the index once per snapshot (MLB → {} so its MLBAM path is untouched) // and fill any player the primary resolve left without an id. A direct href is // preferred — it's the exact URL, so it never 404s on a constructed path. let espnIndex = {}; try { espnIndex = (await deps.buildEspnIndex(sp, { cacheGet: deps.cacheGet, cacheSet: deps.cacheSet })) || {}; } catch { espnIndex = {}; /* graceful — every player falls to a monogram */ } const headshotUrlByPlayer = {}; for (const player of players) { const entry = espnIndex[nameKey(player)]; if (!entry) continue; if (espnIdByPlayer[player] == null && entry.espnId != null) espnIdByPlayer[player] = entry.espnId; // A direct ESPN href wins over any constructed URL (most reliable; the only // honest route for soccer, where we never construct an id-based URL). if (entry.headshotHref) headshotUrlByPlayer[player] = entry.headshotHref; } const enriched = graded.map((g) => { const pn = g.player || g.player_name; return { ...g, gradedAt: gradedAtFor(g, oddsByKey, ts), archetype: archByPlayer[pn] || null, team: teamByPlayer[pn] || g.team || null, // Wave 2A — real headshot id (MLBAM for MLB, ESPN for NBA/WNBA), threaded // from the stats resolve above. Absent → PlayerAvatar renders a monogram. playerId: playerIdByPlayer[pn] ?? g.playerId ?? null, espnId: espnIdByPlayer[pn] ?? g.espnId ?? null, // Wave 2B — a RESOLVED absolute headshot URL from ESPN (preferred over the // constructed (sport,id) URL). Absent → the id/monogram path stands. headshotUrl: headshotUrlByPlayer[pn] ?? g.headshotUrl ?? null, }; }); // LOCK-LINE PERSISTENCE (measurement-only). At THIS moment the grades have locked to // their lines (`gradedAt` uses `ts`), so `props` is the multi-book snapshot AS IT // EXISTED AT LOCK. Persist each graded prop's per-book lines to `lock_lines`, // timestamped `ts`, so a future audit can check whether our locked line was stale-high // vs consensus at lock. Built from the in-memory `props` (no Redis re-read → no TTL // race). Best-effort + structurally fenced: nothing on the grade path reads lock_lines, // and the graded slate is byte-identical whether or not this runs. try { const gradedKeys = new Set(enriched.map((g) => `${norm(g.player || g.player_name)}|${String(g.stat_type || g.stat || '').toLowerCase()}`)); const lockRows = deps.lockLineCapture.buildLockRows(sp, props, gradedKeys, { lockedAt: ts }); const lr = await deps.lockLineCapture.persist(lockRows); console.log(`[lock-lines] ${sp}: ${lr.written}/${lr.attempted} multi-book rows persisted at lock${lr.skipped ? ' (skipped — no supabase env)' : ''}${lr.error ? ` ERROR: ${lr.error}` : ''}`); } catch (e) { console.warn(`[lock-lines] ${sp} persist failed (measurement-only, snapshot continues):`, e.message); } // Session 64 — retention persists HERE, after enrichment, so archetype/team/ // opponent are populated. Feature values were captured at grade time and are // NOT touched by the merge (mergeEnrichment only fills the three null fields). // Session 71 — CHAMPION / CHALLENGER. The challenger is computed here, where // the archetype resolve already happened, so grade-time I/O stays at zero. // `enriched` (champion p_win) is READ, never written: the serving projection // is untouched, and the challenger rides alongside it to the ledger. let withChallenger = enriched; try { const challenger = deps.challenger || require('./challengerProjection'); const axes = deps.archetypeAxes || require('./archetypeAxes'); const rowsByKey = await (deps.loadStatcast || loadStatcastRows)(sp); if (rowsByKey && rowsByKey.size) { const classifyFor = (playerName) => { const row = rowsByKey.get(nameKey(playerName || '')); return row ? axes.classifyPlayer(row) : null; }; // Session 77 — attach environment (park × weather) + matchup (platoon) // per grade. The batter hand rides on the statcast row; the rest is // fetched once here. Best-effort: a context failure leaves archetype-only. let contextFor = null; let ctxInternals = null; // hoisted for proj-v1 (opposing pitcher resolution) try { const envCtx = deps.environmentContext || require('./environmentContext'); const ctx = await envCtx.buildContext(sp, { origin: process.env.BACKEND_SELF_ORIGIN || 'http://localhost:3000', }); ctxInternals = ctx._internals || null; // Enrich each grade with the hitter hand the platoon estimate needs // (statcast_aggregates.bats, already loaded above). const handOf = (name) => { const row = rowsByKey.get(nameKey(name || '')); return row ? row.bats : null; }; contextFor = (g) => ctx.contextFor({ ...g, bats: g.bats || handOf(g.player || g.player_name) }); console.log(`[env] ${sp} — ${ctx.stats.games || 0} games, ${ctx.stats.venues_with_weather || 0} weather, ${ctx.stats.opp_declared || 0} opp-SP declared`); } catch (e) { console.warn(`[env] ${sp} context skipped:`, e.message); } withChallenger = await challenger.attachChallenger(enriched, classifyFor, contextFor); const moved = withChallenger.filter((g) => g.challenger_delta).length; const envMoved = withChallenger.filter((g) => g.env_multiplier != null).length; const platoonMoved = withChallenger.filter((g) => (g.challenger_adjustments || []).some((a) => a.axis === 'matchup')).length; console.log(`[challenger] ${sp} — ${moved}/${withChallenger.length} adjusted (env ${envMoved}, platoon ${platoonMoved})`); // Phase A #2 — SECOND challenger: SEASON contact quality (contact-v1), // tagged separately from arch-v1 so each is measured independently. Reuses // the statcast rows already loaded; champion + arch-v1 fields untouched. // Its own try — a second challenger must never break the pipeline either. try { const contact = deps.contactChallenger || require('./contactChallenger'); const refs = contact.buildRefs([...rowsByKey.values()]); const rowForContact = (name) => rowsByKey.get(nameKey(name || '')); withChallenger = await contact.attachContactChallenger(withChallenger, rowForContact, refs); const cNudged = withChallenger.filter((g) => g.contact_delta).length; const cAbstain = withChallenger.filter((g) => g.p_win_contact == null).length; console.log(`[contact-challenger] ${sp} — ${cNudged} nudged, ${cAbstain} abstained / ${withChallenger.length}`); } catch (e) { console.warn(`[contact-challenger] ${sp} skipped:`, e.message); } // proj-v1 — ABSOLUTE matchup projection (MLB batting). Own try; flag-gated; // reuses arch-v1's already-computed park/weather/platoon on each grade so // it adds no duplicate env I/O. NEVER abstains — thin → wide distribution. if (String(process.env.PROJ_V1_ENABLED || '1') !== '0') { try { const projection = deps.projectionChallenger || require('./projectionChallenger'); const { abbrOf } = require('./environmentContext'); const arsenalById = await (deps.loadArsenals || loadPitcherArsenals)(sp); const oppByTeam = (ctxInternals && ctxInternals.oppPitcherByTeam) || new Map(); const mlbAdapter = deps.mlbAdapter || require('./adapters/mlbStatsAdapter'); withChallenger = await projection.attachProjection(withChallenger, { batterRowFor: (g) => rowsByKey.get(nameKey(g.player || g.player_name || '')), parkFor: (g) => (g.env_park_base != null ? Number(g.env_park_base) : null), weatherFor: (g) => (g.env_weather_mod != null ? Number(g.env_weather_mod) : null), platoonFor: (g) => { const a = (g.challenger_adjustments || []).find((x) => x.axis === 'matchup'); return a ? a.multiplier : null; }, arsenalFor: (g) => { const pid = oppByTeam.get(abbrOf(g.team)); return pid != null ? (arsenalById.get(Number(pid)) || null) : null; }, gameLogFor: async (g) => { if (!g.playerId) return []; try { return (await mlbAdapter.getPlayerGameLog(g.playerId)) || []; } catch { return []; } }, }); const projected = withChallenger.filter((g) => g.proj_point != null).length; const projMatchup = withChallenger.filter((g) => (g.proj_factors && g.proj_factors.breakdown || []).some((f) => f.label === 'matchup' && f.present)).length; console.log(`[proj-v1] ${sp} — ${projected}/${withChallenger.length} projected (${projMatchup} with matchup)`); } catch (e) { console.warn(`[proj-v1] ${sp} skipped:`, e.message); } } } } catch (e) { // The challenger must NEVER break the pipeline it is measured inside. console.warn(`[challenger] ${sp} skipped:`, e.message); } // ── FORWARD CALIBRATION + THE SHADOW DUEL ───────────────────────────── // Fitted on games that are OVER, applied to tonight's props. `p_win` is NOT // touched — the counter stays byte-identical and the corrected value rides // beside it, because a calibration map is a correction TO a forecast, not a // different forecast. // // WHAT IS SERVED, AND WHY IT IS A BET RATHER THAN A RESULT. On identical // held-out rows the ISOTONIC map scored BETTER than the low-parameter one // (hits +0.0028, rbi +0.0042, total_bases tied). We serve the low-parameter // map anyway, on the argument that isotonic's in-window edge is daily // structure shared between the fit and evaluation windows and will not // transmit forward. At 19 dates no instrument here can test that argument — // LODO has 1.4-9.3% power — so it is a BET, not evidence. // // So both are computed on every prop and the shadow is logged. Real // out-of-window dates adjudicate it: // // PRE-REGISTERED: once >=10 forward dates have settled that NEITHER map was // fitted on, if isotonic beats low-param with a date-block bootstrap CI // excluding zero, the capacity argument is REFUTED and hits/TB revert to // isotonic. If low-param wins or ties, the bet was right. The season // decides, not the argument. // // Serving is unchanged until that bar is met. `calibrated` is true only inside // a band certified out-of-sample, and it is what `chain.chainAcross` requires // before it will compound anything. if (sp === 'mlb') { for (const stat of CALIBRATION_DEPLOYED) { try { const calSvc = deps.calibrationService || require('./model/lowParamService'); const shadowSvc = deps.shadowCalibrationService || require('./model/calibrationService'); const sbc = require('../utils/supabase').getSupabaseServiceClient(); const calibrator = sbc ? await calSvc.fromLedger(sbc, { sport: 'mlb', stat }) : null; // The shadow must never break serving: its own try, and a null shadow // simply means the duel has no entry for tonight. let shadow = null; try { shadow = sbc ? await shadowSvc.fromLedger(sbc, { sport: 'mlb', stat }) : null; } catch { shadow = null; } if (calibrator) { let marked = 0; let shadowed = 0; for (const g of enriched) { if (String(g.stat_type || g.stat || '').toLowerCase() !== stat) continue; const out = calibrator.calibrate(g.p_win); g.p_win_calibrated = out.p_calibrated; g.p_win_lowparam = out.p_calibrated; // named, so the duel is legible g.calibrated = out.calibrated; g.calibration_reason = out.reason; g.calibration_status = 'provisional'; g.calibration_basis = CALIBRATION_BASIS[stat] || null; if (out.calibrated) marked += 1; if (shadow) { const sh = shadow.calibrate(g.p_win); // SHADOW ONLY. Never read by serving, never by chainAcross. g.p_win_isotonic_shadow = sh.p_calibrated; g.calibration_duel_fitted_through = shadow.fitted_through || null; if (sh.p_calibrated != null) shadowed += 1; } } console.log(`[calibration] ${sp} ${stat} (low-param, PROVISIONAL) — ${marked} stackable; fit n=${calibrator.fit_n} through ${calibrator.fitted_through}, a=${calibrator.model.a} shrink=${calibrator.shrinkage}; shadow logged on ${shadowed}`); } else { console.log(`[calibration] ${sp} ${stat} — no calibrator (thin history); nothing is stackable`); } } catch (e) { console.warn(`[calibration] ${stat} skipped:`, e.message); } } } // LINEUP + BASERUNNER CONTEXT — the input RBI and runs have always needed. // Best-effort and dated: a context failure must never break a snapshot, and a // lineup is a PRE-GAME fact that changes by the hour, so what we knew at grade // time must never be overwritten by what turned out to be true. if (sp === 'mlb') { try { const ctx = deps.lineupContext || require('./lineupContextService'); const sbc = require('../utils/supabase').getSupabaseServiceClient(); if (sbc) { const res = await ctx.refreshContext({ supabase: sbc, sport: 'mlb' }); console.log(`[lineup-context] ${sp} — lineups ${res.lineups}/${res.lineup_rows ?? 0} (${res.games_with_lineups ?? 0} games), opportunity ${res.opportunity}/${res.opportunity_rows ?? 0}${res.reason ? ` reason: ${res.reason}` : ''}`); } } catch (e) { console.warn('[lineup-context] skipped:', e.message); } } await persistRetention(enriched); // Line deltas vs the previous snapshot's locked lines. const prev = await deps.cacheGet(`snapshot:${sp}:latest`); const deltas = computeLineDeltas(enriched, prev && prev.grades); // Lock: previous = old latest, latest = new, grades = enriched. if (prev) await deps.cacheSet(`snapshot:${sp}:previous`, prev, SNAP_TTL); // `updated_at` = grade-LOCK time (advances only on a full snapshot, 5×/day — // grades never change in-game, so this is intentionally stable). `refreshed_at` // = the freshness heartbeat: seeded here at lock time, then bumped every // intraday refresh (intradayRefreshService). The SYNC badge keys off // refreshed_at — measuring the 20-min cadence against the grade-lock field is // what produced the "SIGNAL LIVE vs STALE 8h" contradiction. const snapshot = { sport: sp, updated_at: ts, refreshed_at: ts, grades: enriched, deltas, gradeCount: enriched.length }; await deps.cacheSet(`snapshot:${sp}:latest`, snapshot, SNAP_TTL); // Session 59 — grades:{sport} must outlive the gap between cron runs (up to // 5h) or team rosters / Explore / leaders go dark mid-day. SNAP_TTL (6h), // NOT the legacy 2h gradeSlateService TTL — that gap was why /team showed // "No active props" for players who were on the slate (audit 2.4). await deps.cacheSet(`grades:${sp}`, { grades: enriched, updated_at: ts, source: (odds && odds.provider) || 'odds-api' }, SNAP_TTL); // Ticker exhaust. const events = generateTickerEvents(sp, enriched, deltas, ts); await pushTickerItems(events, deps); // Session 58 — Phase 1 truth infrastructure. (a) Upsert the public model // record (user_id NULL, idempotent — re-runs never duplicate or overwrite // the original lock). (b) Overwrite today's closing_line/odds from the // CURRENT feed — the last capture before game start IS the closing line. // Best-effort: the ledger must never break the snapshot. let ledgerWritten = 0; try { const rec = await deps.ledger.recordPipelineGrades(sp, withChallenger, props, { now: deps.now }); ledgerWritten = rec.written || 0; await deps.ledger.captureClosing(sp, props); } catch (e) { console.warn(`[snapshot] ledger write failed for ${sp}:`, e.message); } // Session 56 — success alert, enriched with the rolling accuracy (if settled). let accPart = ''; try { const acc = await deps.cacheGet(`accuracy:${sp}`); const pct = acc && acc.overall && acc.overall.pct; if (pct != null) accPart = `, ${pct}% accuracy (30d)`; } catch { /* accuracy is best-effort in the alert */ } await deps.notify( `✅ ${sp.toUpperCase()} snapshot: ${enriched.length} props graded, ${deltas.length} deltas${accPart}`, { title: 'VYNDR pipeline', tags: ['white_check_mark'] }, ); return { sport: sp, status: 'ok', gradeCount: enriched.length, ledgerWritten, // Session 64 — surfaced so the scheduler can page when retention silently // writes nothing. Retention is best-effort by design, which makes a broken // write invisible without this. retentionRows, topGrades: enriched.filter((g) => isTopGrade(g.grade)).slice(0, 5).map((g) => ({ player: g.player || g.player_name, stat: g.stat_type || g.stat, grade: g.grade, archetype: g.archetype, })), deltas: deltas.length, duration: deps.nowMs() - start, }; } /** Run snapshots for every active sport sequentially (cron entrypoint). */ // Job 1 — `opts.sports` scopes the run to a subset (the scheduler passes the // hour's per-sport cadence). Absent → every ACTIVE sport (the on-demand // /api/internal/snapshot/all behaviour is unchanged). runSnapshot ignores the // extra key. async function runAllSnapshots(opts = {}) { // An EXPLICIT array is honored verbatim (even empty = run nothing); only an // ABSENT `sports` key means "every active sport" (on-demand /all). const sports = Array.isArray(opts.sports) ? opts.sports : ACTIVE_SPORTS; const results = []; for (const sp of sports) { results.push(await runSnapshot(sp, opts)); } return results; } module.exports = { runSnapshot, runAllSnapshots, computeLineDeltas, generateTickerEvents, pushTickerItems, ACTIVE_SPORTS, CALIBRATION_DEPLOYED, CALIBRATION_BASIS, __internals: { propKey, gradedAtFor, indexOdds, lastName, isTopGrade, DELTA_NOISE, DELTA_MOVE, TICKER_CAP }, };