/** * 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) */ const SNAP_TTL = 6 * 3600; // 6h — a snapshot is valid until the next run const GRADES_TTL = 2 * 3600; // matches gradeSlateService 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. */ function indexOdds(props) { const map = {}; for (const p of props || []) { const k = `${norm(p.player)}|${String(p.stat_type || '').toLowerCase()}`; 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({ tag: 'SCAN', color: 'var(--g-a)', ts, sport, // sport tag → dedupe one SCAN per sport text: `${sport.toUpperCase()} slate scanned · ${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 SCAN event's sport, from the event field or its text prefix // (defends ticker items written before the `sport` field existed). function scanSportOf(e) { if (e.tag !== 'SCAN') return null; if (e.sport) return String(e.sport).toLowerCase(); const m = String(e.text || '').match(/^([a-z]+)\s+slate 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 SCAN per sport: drop existing SCAN events for any sport // that has a fresh SCAN in this batch. MOVE/GRADE events are time-specific and // preserved. const freshScanSports = new Set(events.map(scanSportOf).filter(Boolean)); const pruned = arr.filter((e) => { const sp = scanSportOf(e); return !(sp && freshScanSports.has(sp)); }); const merged = [...events, ...pruned].slice(0, TICKER_CAP); await deps.cacheSet('ticker:items', merged, TICKER_TTL); } 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. */ 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()), }; const start = deps.nowMs(); const ts = deps.now(); let odds; try { odds = await deps.getOdds(sp); } catch (e) { return { sport: sp, status: 'error', reason: e.message, gradeCount: 0 }; } const props = (odds && Array.isArray(odds.props)) ? odds.props : []; if (props.length === 0) return { sport: sp, status: 'skipped', reason: 'no props', gradeCount: 0 }; // 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; }, }); const rawGraded = (envelope && Array.isArray(envelope.grades)) ? envelope.grades : []; if (rawGraded.length === 0) 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 = {}; await mapLimit(players, STATS_CONCURRENCY, async (player) => { try { const stats = await deps.resolveStats(player, sp); if (stats && stats.found) { const c = deps.classify(sp, stats.classifierInput || {}); archByPlayer[player] = c.primary ? c.primary.name : null; } } catch { /* graceful — no badge */ } }); const enriched = graded.map((g) => ({ ...g, gradedAt: gradedAtFor(g, oddsByKey, ts), archetype: archByPlayer[g.player || g.player_name] || null, })); // 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); const snapshot = { sport: sp, updated_at: ts, grades: enriched, deltas, gradeCount: enriched.length }; await deps.cacheSet(`snapshot:${sp}:latest`, snapshot, SNAP_TTL); await deps.cacheSet(`grades:${sp}`, { grades: enriched, updated_at: ts, source: (odds && odds.provider) || 'odds-api' }, GRADES_TTL); // Ticker exhaust. const events = generateTickerEvents(sp, enriched, deltas, ts); await pushTickerItems(events, deps); return { sport: sp, status: 'ok', gradeCount: enriched.length, 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). */ async function runAllSnapshots(opts = {}) { const results = []; for (const sp of ACTIVE_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 }, };