Files
vyndr/src/services/snapshotService.js
T
builtbykev b6f12daa98 Contact-quality challenger (contact-v1) — nominate, don't swap
Phase A #2: the champion grade (l5/l20 result-based form) is a HYPOTHESIS that
contact quality predicts better — unmeasured on our props, with zero settled
p_win yet. Swapping l5/l20 (the champion's two heaviest ±1.0 factors) blind
could degrade the core grade undetectably for weeks. So this NOMINATES contact
quality as a second challenger, records what it WOULD project per prop, and lets
the settled ledger decide. Nothing users see changes; the champion is untouched.

- src/services/contactChallenger.js — pure, mirrors challengerProjection. Log-
  odds lean (capped, never a re-forecast) from SEASON contact quality vs league
  percentiles. Metric→prop mapping is the whole game: barrel_pct→HR,
  hard_hit_pct→TB/doubles, k_pct-INVERSE→hits (singles resolve on contact
  frequency, not barrels), k_pct→batter K. rbi/runs/walks ABSTAIN (opportunity/
  discipline — no clean contact predictor). Honest-absent: thin (<50 PA)/absent/
  unmapped/non-batter → p_win_contact NULL (no projection), never a fallback;
  "measured but unremarkable" is distinct (equals champion, delta 0).
- Wired in snapshotService AFTER arch-v1, reusing the already-loaded statcast
  rows; its own try so a second challenger can't break the pipeline. Reads
  g.p_win, never writes it.
- Retained SEPARATELY on the ledger (p_win_contact/contact_delta/
  contact_adjustments/contact_version='contact-v1') so each challenger's marginal
  contribution is measured independently; ledger_entries.stat gives per-prop-type
  segmentation. Migration 031 (applied to prod).

Phase 0 (prod-verified): statcast_aggregates is SEASON cumulative (not rolling),
48h stale now but season-scoped so ~8 PA/600 is negligible; 100% of graded
hitters covered, 92% at ≥50 PA; no xBA/xwOBA in the feed. Forward-only,
version-stamped (contact_version null on pre-nomination rows). Promotion is a
LATER decision on settled evidence, per prop type — never asserted here.

contactChallenger 14/14; snapshot/ledger/arch-v1 suites 80 green.

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>
Claude-Session: https://claude.ai/code/session_01VCNgGSt5qvcLxaeQqa7Zpj
2026-07-22 22:42:55 -04:00

673 lines
33 KiB
JavaScript
Raw Blame History

This file contains ambiguous Unicode characters
This file contains Unicode characters that might be confused with other characters. If you think that this is intentional, you can safely ignore this warning. Use the Escape button to reveal them.
/**
* 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. */
function indexOdds(props) {
const map = {};
const bothSides = (p) => p && p.over_odds != null && p.under_odds != null;
for (const p of props || []) {
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; }
}
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'),
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);
}
// 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;
await deps.gradeAndCacheSlate(sp, props, {
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,
};
});
// 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;
try {
const envCtx = deps.environmentContext || require('./environmentContext');
const ctx = await envCtx.buildContext(sp, {
origin: process.env.BACKEND_SELF_ORIGIN || 'http://localhost:3000',
});
// 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);
}
}
} catch (e) {
// The challenger must NEVER break the pipeline it is measured inside.
console.warn(`[challenger] ${sp} 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,
__internals: { propKey, gradedAtFor, indexOdds, lastName, isTopGrade, DELTA_NOISE, DELTA_MOVE, TICKER_CAP },
};