304 lines
11 KiB
JavaScript
304 lines
11 KiB
JavaScript
/**
|
|
* 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 },
|
|
};
|