Sessions 5-7a: 955 tests, deployment ready
This commit is contained in:
@@ -0,0 +1,303 @@
|
||||
/**
|
||||
* Grading pipeline orchestrator.
|
||||
*
|
||||
* Called by n8n at 10:30 AM, 1 PM, 4 PM, 6 PM ET (and on demand from the
|
||||
* /api/grading/pipeline endpoint). For one sport per call, it:
|
||||
*
|
||||
* 1. Pulls today's scoreboard from the sport config's ESPN endpoint.
|
||||
* We do NOT call SharpAPI for the slate — only for player props per
|
||||
* game. Scoreboard is the source of truth for which games exist.
|
||||
* 2. For each game, fetches player props via SharpAPI.
|
||||
* 3. For each prop, builds a feature vector + trap composite +
|
||||
* consistency score, then asks Engine 1 to grade.
|
||||
* 4. Persists the grade to grade_history.
|
||||
* 5. Queues A/B-tier grades for Engine 2.
|
||||
* 6. Drains the Engine 2 queue (best-effort, one batch).
|
||||
*
|
||||
* Failure semantics:
|
||||
* - SharpAPI down → 0 props graded, summary still returns.
|
||||
* - Per-prop error → log + skip, other props continue.
|
||||
* - Engine 2 queue failure → does not affect Engine 1 grades that
|
||||
* are already in the database.
|
||||
*/
|
||||
|
||||
const axios = require('axios');
|
||||
const { getSportConfig } = require('../../config/sports');
|
||||
const { getSupabaseServiceClient } = require('../../utils/supabase');
|
||||
const featureCache = require('./featureCache');
|
||||
const trapDetection = require('./trapDetection');
|
||||
const consistencyScore = require('./consistencyScore');
|
||||
const engine1 = require('./engine1');
|
||||
const engine2 = require('./engine2');
|
||||
const gameLogService = require('./gameLogService');
|
||||
const probabilityEstimator = require('./probabilityEstimator');
|
||||
const sharpApi = require('../adapters/sharpApiAdapter');
|
||||
|
||||
const HTTP_TIMEOUT_MS = 15_000;
|
||||
|
||||
async function fetchTodaysGames(sportCfg) {
|
||||
try {
|
||||
const res = await axios.get(sportCfg.espnScoreboard, { timeout: HTTP_TIMEOUT_MS });
|
||||
const events = res.data?.events || [];
|
||||
return events.map((ev) => {
|
||||
const comp = ev?.competitions?.[0];
|
||||
const teams = (comp?.competitors || []).reduce((acc, t) => {
|
||||
const role = t?.homeAway === 'home' ? 'home' : 'away';
|
||||
acc[role] = { id: t?.id, abbr: t?.team?.abbreviation, name: t?.team?.displayName };
|
||||
return acc;
|
||||
}, {});
|
||||
return {
|
||||
gameId: String(ev.id),
|
||||
gameDate: ev?.date,
|
||||
home: teams.home,
|
||||
away: teams.away,
|
||||
state: ev?.status?.type?.state,
|
||||
};
|
||||
});
|
||||
} catch (err) {
|
||||
console.warn('[orchestrator] scoreboard fetch failed:', err.message);
|
||||
return [];
|
||||
}
|
||||
}
|
||||
|
||||
async function buildPropContext(prop, game, sport) {
|
||||
// Determine whether this prop's player is on home or away team. We
|
||||
// don't have a roster lookup at this point of the pipeline; the orchestrator
|
||||
// treats prop.team (if SharpAPI provides) as the canonical, falling back
|
||||
// to "unknown" for home_away.
|
||||
const team = prop.team || prop.teamAbbr;
|
||||
const isHome = team && game.home?.abbr === team;
|
||||
const opponentAbbr = isHome ? game.away?.abbr : game.home?.abbr;
|
||||
return {
|
||||
playerId: prop.playerId || prop.player_id || null,
|
||||
playerName: prop.player,
|
||||
statType: prop.statType || prop.stat_type,
|
||||
sport,
|
||||
line: Number(prop.line),
|
||||
direction: prop.direction || 'over',
|
||||
teamAbbr: team,
|
||||
opponentAbbr,
|
||||
gameId: game.gameId,
|
||||
gameContext: {
|
||||
home_away: team ? (isHome ? 'home' : 'away') : null,
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
async function gradeProp(prop, game, sport) {
|
||||
const ctx = await buildPropContext(prop, game, sport);
|
||||
|
||||
// Feature vector — every signal computed in 6b.
|
||||
const featurePayload = await featureCache.getFeatures({
|
||||
playerId: ctx.playerId,
|
||||
playerName: ctx.playerName,
|
||||
statType: ctx.statType,
|
||||
sport: ctx.sport,
|
||||
teamAbbr: ctx.teamAbbr,
|
||||
opponentAbbr: ctx.opponentAbbr,
|
||||
gameId: ctx.gameId,
|
||||
gameContext: ctx.gameContext,
|
||||
});
|
||||
const features = featurePayload?.features || {};
|
||||
|
||||
// Trap detector — uses features + lineMovement snapshots already in DB.
|
||||
const trap = await trapDetection.getTrapScore({
|
||||
playerName: ctx.playerName,
|
||||
statType: ctx.statType,
|
||||
sport: ctx.sport,
|
||||
gameId: ctx.gameId,
|
||||
gameContext: ctx.gameContext,
|
||||
features,
|
||||
odds: { playerLine: ctx.line, consensus: prop.consensus },
|
||||
});
|
||||
|
||||
// Consistency — Engine 2 uses this verbatim in its prompt.
|
||||
let consistency = { consistency: 'unknown', score: null, games: 0 };
|
||||
let gameLogs = null;
|
||||
try {
|
||||
gameLogs = await gameLogService.getGameLogs(ctx.playerName, ctx.sport, 20);
|
||||
if (gameLogs && gameLogs.length) {
|
||||
consistency = await consistencyScore.getConsistency({
|
||||
playerName: ctx.playerName,
|
||||
sport: ctx.sport,
|
||||
statType: ctx.statType,
|
||||
gameLogs,
|
||||
});
|
||||
}
|
||||
} catch (err) {
|
||||
console.warn('[orchestrator] consistency failed for', ctx.playerName, err.message);
|
||||
}
|
||||
|
||||
// P(Over) — quantile-based probability from game logs. We pass the same
|
||||
// game logs to the estimator that consistency uses, so both views agree
|
||||
// on the same data window. Null if no logs (Python service down).
|
||||
let probability = { p_over: null, p_under: null, components: {}, reason: 'no_logs' };
|
||||
if (gameLogs && gameLogs.length) {
|
||||
probability = probabilityEstimator.estimateProbability({
|
||||
gameLogs,
|
||||
line: ctx.line,
|
||||
statType: ctx.statType,
|
||||
features,
|
||||
});
|
||||
}
|
||||
|
||||
// Engine 1 — rule-based, deterministic.
|
||||
const result = engine1.gradeProp({
|
||||
features,
|
||||
trap,
|
||||
consistency,
|
||||
prop: { line: ctx.line, direction: ctx.direction },
|
||||
});
|
||||
|
||||
return { ctx, features, trap, consistency, probability, engine1Result: result };
|
||||
}
|
||||
|
||||
async function persistGrade(graded, prop, sport) {
|
||||
const supabase = getSupabaseServiceClient();
|
||||
const { ctx, engine1Result, trap, consistency, features, probability } = graded;
|
||||
const row = {
|
||||
player_id: ctx.playerId,
|
||||
player_name: ctx.playerName,
|
||||
sport,
|
||||
stat_type: ctx.statType,
|
||||
line: ctx.line,
|
||||
direction: ctx.direction,
|
||||
grade: engine1Result.grade,
|
||||
projection: Number.isFinite(features.l5_avg) ? features.l5_avg : null,
|
||||
// modeled_prob is the implied probability from Engine 1's grade tier;
|
||||
// p_over is the quantile-based probability from game logs. Both useful
|
||||
// — the former for grade-vs-line edge math, the latter for UI display.
|
||||
modeled_prob: Number.isFinite(engine1Result?.confidence) ? engine1Result.confidence : null,
|
||||
implied_prob: null,
|
||||
p_over: Number.isFinite(probability?.p_over) ? probability.p_over : null,
|
||||
// factors drive the weight adjuster: each resolved prop's factors get
|
||||
// nudged based on hit/miss outcome. Stored as JSONB so we can also
|
||||
// surface them in the UI "why this grade" tooltip.
|
||||
factors: Array.isArray(engine1Result?.all_factors)
|
||||
? engine1Result.all_factors
|
||||
: (Array.isArray(engine1Result?.top_factors) ? engine1Result.top_factors : null),
|
||||
game_date: new Date().toISOString().slice(0, 10),
|
||||
game_id: ctx.gameId,
|
||||
};
|
||||
const { data, error } = await supabase.from('grade_history').insert(row).select('id').single();
|
||||
if (error) {
|
||||
console.warn('[orchestrator] grade_history insert failed:', error.message);
|
||||
return null;
|
||||
}
|
||||
// Hand the gradeId + full context to engine2 so it can build a prompt.
|
||||
engine2.queueAnalysis(data.id, {
|
||||
player_name: ctx.playerName,
|
||||
team: ctx.teamAbbr,
|
||||
sport,
|
||||
direction: ctx.direction,
|
||||
line: ctx.line,
|
||||
stat_type: ctx.statType,
|
||||
home_team: prop._home,
|
||||
away_team: prop._away,
|
||||
game_date: row.game_date,
|
||||
engine1_grade: engine1Result.grade,
|
||||
engine1_factors: engine1Result.top_factors,
|
||||
features,
|
||||
trap,
|
||||
consistency,
|
||||
probability,
|
||||
recentGames: [],
|
||||
});
|
||||
return data.id;
|
||||
}
|
||||
|
||||
async function gradeProps(props, game, sport) {
|
||||
const out = [];
|
||||
for (const prop of props) {
|
||||
try {
|
||||
const graded = await gradeProp(prop, game, sport);
|
||||
const gradeId = await persistGrade(graded, { ...prop, _home: game.home?.name, _away: game.away?.name }, sport);
|
||||
out.push({ gradeId, grade: graded.engine1Result.grade, prop });
|
||||
} catch (err) {
|
||||
console.warn('[orchestrator] gradeProp failed for', prop?.player, err.message);
|
||||
}
|
||||
}
|
||||
return out;
|
||||
}
|
||||
|
||||
async function runPipeline(sport, options = {}) {
|
||||
const start = Date.now();
|
||||
let sportCfg;
|
||||
try { sportCfg = getSportConfig(sport); }
|
||||
catch (err) { return { error: err.message, sport, games_processed: 0, props_graded: 0, duration_ms: Date.now() - start }; }
|
||||
|
||||
const games = await fetchTodaysGames(sportCfg);
|
||||
if (games.length === 0) {
|
||||
return { sport, games_processed: 0, props_graded: 0, engine2_queued: 0, errors: 0, duration_ms: Date.now() - start };
|
||||
}
|
||||
|
||||
let propsGraded = 0;
|
||||
let errors = 0;
|
||||
let engine2Queued = 0;
|
||||
for (const game of games) {
|
||||
let props;
|
||||
try {
|
||||
props = await sharpApi.getPlayerProps(sport, game.gameId);
|
||||
} catch (err) {
|
||||
console.warn('[orchestrator] sharpApi failed for', game.gameId, err.message);
|
||||
errors += 1;
|
||||
continue;
|
||||
}
|
||||
if (!Array.isArray(props) || props.length === 0) continue;
|
||||
const before = engine2.getQueueSize();
|
||||
const graded = await gradeProps(props, game, sport);
|
||||
propsGraded += graded.length;
|
||||
engine2Queued += engine2.getQueueSize() - before;
|
||||
}
|
||||
|
||||
// Drain the Engine 2 queue with a bounded loop. Each processQueue()
|
||||
// call handles ENGINE2_BATCH_SIZE items, so for slates of ~50+ A/B
|
||||
// grades one call would leave most of the queue parked. Cap at 5
|
||||
// iterations (≈50 props per pipeline run with default batch size)
|
||||
// — beyond that, the next pipeline cycle picks up the remainder.
|
||||
let engine2Summary = { processed: 0, succeeded: 0, failed: 0, remaining: engine2.getQueueSize() };
|
||||
if (!options.skipEngine2) {
|
||||
const MAX_DRAIN_ITERS = 5;
|
||||
let drainIters = 0;
|
||||
const totals = { processed: 0, succeeded: 0, failed: 0 };
|
||||
while (engine2.getQueueSize() > 0 && drainIters < MAX_DRAIN_ITERS) {
|
||||
const round = await engine2.processQueue();
|
||||
totals.processed += round.processed || 0;
|
||||
totals.succeeded += round.succeeded || 0;
|
||||
totals.failed += round.failed || 0;
|
||||
drainIters += 1;
|
||||
// If a round processes 0 items, the queue is stuck (likely
|
||||
// disabled or all calls failing) — break early instead of looping.
|
||||
if ((round.processed || 0) === 0) break;
|
||||
}
|
||||
engine2Summary = { ...totals, remaining: engine2.getQueueSize(), iterations: drainIters };
|
||||
}
|
||||
|
||||
return {
|
||||
sport,
|
||||
games_processed: games.length,
|
||||
props_graded: propsGraded,
|
||||
engine2_queued: engine2Queued,
|
||||
engine2_summary: engine2Summary,
|
||||
errors,
|
||||
duration_ms: Date.now() - start,
|
||||
};
|
||||
}
|
||||
|
||||
function getEngineStatus() {
|
||||
return {
|
||||
engine2_queue_size: engine2.getQueueSize(),
|
||||
adapters_configured: {
|
||||
sharp_api: sharpApi.configured(),
|
||||
open_router: require('../adapters/openRouterAdapter').configured(),
|
||||
},
|
||||
};
|
||||
}
|
||||
|
||||
module.exports = {
|
||||
runPipeline,
|
||||
gradeProps,
|
||||
gradeProp,
|
||||
getEngineStatus,
|
||||
__internals: { fetchTodaysGames, buildPropContext, persistGrade },
|
||||
};
|
||||
Reference in New Issue
Block a user