From 4aa2367971b170d857e1402a2689c6e66027b029 Mon Sep 17 00:00:00 2001 From: ImBenji Date: Thu, 13 Aug 2026 19:10:59 +0100 Subject: [PATCH] fix: calibrate from replay outcomes --- test/autonomy.test.js | 48 +++++++++++++++++++++++++++++ workers/calibrationWorker.js | 59 ++++++++++++++++++++++++++++++------ 2 files changed, 97 insertions(+), 10 deletions(-) diff --git a/test/autonomy.test.js b/test/autonomy.test.js index ba04966..c82a128 100644 --- a/test/autonomy.test.js +++ b/test/autonomy.test.js @@ -12,6 +12,7 @@ const { calculateOutcome } = require('../src/autonomy/outcomes'); const { createOrderIntent } = require('../src/autonomy/orderIntents'); const { enqueueCoordinatorEvent, reconcileArchiveBatch, reconcileLiveBatch } = require('../workers/autonomyWorker'); const { scheduleNext } = require('../workers/replayWorker'); +const { refreshHistoricalCalibration, createDecisions } = require('../workers/calibrationWorker'); test('autonomy schema and leased jobs are restart-safe', () => { const db = new Database(':memory:'); @@ -120,6 +121,53 @@ test('calibration and policy abstain on insufficient evidence', () => { assert.equal(decide({ direction: 'negative', probability: 0.8, expectedExcessReturn: -0.02, lowerReturn: -0.04, sampleSize: 40 }).action, 'SELL'); }); +test('historical replay outcomes create replay calibration snapshots', () => { + const db = new Database(':memory:'); + initAutonomySchema(db); + const proposal = db.prepare("INSERT INTO autonomy_proposals(payload, information_cutoff, status) VALUES ('{}', '2020-01-01T00:00:00Z', 'accepted')").run(); + const prediction = db.prepare(` + INSERT INTO autonomy_predictions + (proposal_id, instrument, direction, event_type, horizon_days, information_cutoff, evidence_article_ids, + learning_eligible, strategy_version, origin, replay_run_id, status) + VALUES (?, 'NVDA', 'positive', 'earnings', 10, '2020-01-01T00:00:00Z', '[1]', 0, 'test', 'replay', 7, 'resolved') + `).run(proposal.lastInsertRowid); + db.prepare(` + INSERT INTO autonomy_outcomes(prediction_id, excess_return, direction_correct) + VALUES (?, 0.04, 1) + `).run(prediction.lastInsertRowid); + + assert.equal(refreshHistoricalCalibration(db, 'test-cal'), 1); + const snapshot = db.prepare("SELECT source, replay_run_id, sample_size, directional_probability FROM autonomy_calibration_snapshots").get(); + assert.equal(snapshot.source, 'replay'); + assert.equal(snapshot.replay_run_id, 7); + assert.equal(snapshot.sample_size, 1); + assert(snapshot.directional_probability > 0.5); +}); + +test('live decisions map calibration snapshot fields into policy inputs', () => { + const db = new Database(':memory:'); + initAutonomySchema(db); + const proposal = db.prepare("INSERT INTO autonomy_proposals(payload, information_cutoff, status) VALUES ('{}', datetime('now'), 'accepted')").run(); + const prediction = db.prepare(` + INSERT INTO autonomy_predictions + (proposal_id, instrument, direction, event_type, horizon_days, information_cutoff, evidence_article_ids, + learning_eligible, strategy_version, origin, status) + VALUES (?, 'NVDA', 'positive', 'earnings', 10, datetime('now'), '[1]', 1, 'test', 'live', 'open') + `).run(proposal.lastInsertRowid); + db.prepare(` + INSERT INTO autonomy_calibration_snapshots + (cohort_key, sample_size, effective_sample_size, directional_probability, expected_excess_return, + lower_return, upper_return, parent_cohort_key, version, source) + VALUES ('unknown|earnings|10|positive', 40, 42, 0.7, 0.02, -0.01, 0.06, NULL, 'test-cal', 'replay') + `).run(); + + assert.equal(createDecisions(db), 1); + const decision = db.prepare('SELECT * FROM autonomy_decisions WHERE prediction_id=?').get(prediction.lastInsertRowid); + assert.equal(decision.action, 'BUY'); + assert.equal(decision.calibrated_probability, 0.7); + assert.equal(decision.expected_excess_return, 0.02); +}); + test('paper execution is allowlisted, bounded and idempotent', () => { const intent = validatePaperIntent({ decisionId: 12, instrument: 'NVDA', action: 'BUY', notional: 100 }, { tradable: true, maxNotional: 500 }); const broker = createSimulator(); diff --git a/workers/calibrationWorker.js b/workers/calibrationWorker.js index 0e8d161..8d5eda9 100644 --- a/workers/calibrationWorker.js +++ b/workers/calibrationWorker.js @@ -6,13 +6,26 @@ const { decide } = require('../src/autonomy/policy'); function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); } -function refreshCalibration(db, version = `cal-${Date.now()}`) { +function snapshotToDecisionInput(snapshot, direction) { + return { + direction, + probability: snapshot.directional_probability, + expectedExcessReturn: snapshot.expected_excess_return, + lowerReturn: snapshot.lower_return, + upperReturn: snapshot.upper_return, + sampleSize: snapshot.sample_size, + }; +} + +function refreshCalibration(db, version = `cal-${Date.now()}`, { origin = 'live', source = origin, replayRunId = null } = {}) { + const learningClause = origin === 'live' ? 'AND p.learning_eligible = 1' : ''; + const replayClause = replayRunId ? 'AND p.replay_run_id = @replayRunId' : ''; const groups = db.prepare(` SELECT p.direction, p.event_type, p.horizon_days, o.* FROM autonomy_predictions p JOIN autonomy_outcomes o ON o.prediction_id = p.id - WHERE p.status = 'resolved' AND p.learning_eligible = 1 AND p.origin = 'live' - `).all().reduce((map, row) => { + WHERE p.status = 'resolved' AND p.origin = @origin ${learningClause} ${replayClause} + `).all({ origin, replayRunId }).reduce((map, row) => { const key = cohortKey({ direction: row.direction, eventType: row.event_type, horizonDays: row.horizon_days }); if (!map.has(key)) map.set(key, []); map.get(key).push(row); @@ -22,20 +35,44 @@ function refreshCalibration(db, version = `cal-${Date.now()}`) { INSERT INTO autonomy_calibration_snapshots (cohort_key, sample_size, effective_sample_size, directional_probability, expected_excess_return, lower_return, upper_return, parent_cohort_key, version, source, replay_run_id) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', NULL) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `); const tx = db.transaction(() => { for (const [key, rows] of groups) { - if (db.prepare("SELECT 1 FROM autonomy_calibration_snapshots WHERE cohort_key = ? AND version = ?").get(key, version)) continue; + if (db.prepare(` + SELECT 1 FROM autonomy_calibration_snapshots + WHERE cohort_key = ? AND version = ? AND source = ? AND COALESCE(replay_run_id, 0) = COALESCE(?, 0) + `).get(key, version, source, replayRunId)) continue; const result = calibrateOutcomes(rows); insert.run(key, result.sampleSize, result.effectiveSampleSize, result.directionalProbability, - result.expectedExcessReturn, result.lowerReturn, result.upperReturn, null, version); + result.expectedExcessReturn, result.lowerReturn, result.upperReturn, null, version, source, replayRunId); } }); tx(); return groups.size; } +function refreshHistoricalCalibration(db, version = `replay-cal-${Date.now()}`) { + const runs = db.prepare(` + SELECT DISTINCT replay_run_id AS replayRunId + FROM autonomy_predictions + WHERE origin = 'replay' AND replay_run_id IS NOT NULL + ORDER BY replay_run_id + `).all(); + let groups = 0; + for (const run of runs) { + groups += refreshCalibration(db, `${version}-run-${run.replayRunId}`, { + origin: 'replay', + source: 'replay', + replayRunId: run.replayRunId, + }); + } + if (!runs.length) { + groups += refreshCalibration(db, version, { origin: 'replay', source: 'replay' }); + } + return groups; +} + function createDecisions(db, strategyVersion = 'autonomy-1') { const predictions = db.prepare(` SELECT p.* FROM autonomy_predictions p @@ -57,7 +94,7 @@ function createDecisions(db, strategyVersion = 'autonomy-1') { const key = cohortKey({ direction: prediction.direction, eventType: prediction.event_type, horizonDays: prediction.horizon_days }); const calibration = latest.get(key); const decision = calibration - ? decide({ ...calibration, direction: prediction.direction }, { minSampleSize: 30 }) + ? decide(snapshotToDecisionInput(calibration, prediction.direction), { minSampleSize: 30 }) : { action: 'ABSTAIN', rationale: 'calibration unavailable' }; insert.run(prediction.id, decision.action, calibration?.directional_probability || null, calibration?.expected_excess_return || null, decision.rationale, strategyVersion); @@ -114,10 +151,12 @@ async function runCalibrationWorker({ intelligencePath, pollMs = 60000, workerId while (true) { try { const state = db.prepare('SELECT COUNT(*) AS count, COALESCE(MAX(prediction_id), 0) AS max_id FROM autonomy_outcomes').get(); - const groups = refreshCalibration(db, `cal-${state.count}-${state.max_id}`); + const version = `cal-${state.count}-${state.max_id}`; + const groups = refreshCalibration(db, version); + const historicalGroups = refreshHistoricalCalibration(db, version); const decisions = createDecisions(db); const replayEvaluations = refreshReplayEvaluations(db); - if (groups || decisions || replayEvaluations) console.log(`[${workerId}] calibration groups=${groups} decisions=${decisions} replay_evaluations=${replayEvaluations}`); + if (groups || historicalGroups || decisions || replayEvaluations) console.log(`[${workerId}] calibration groups=${groups} historical_groups=${historicalGroups} decisions=${decisions} replay_evaluations=${replayEvaluations}`); } catch (error) { console.error(`[${workerId}] calibration error:`, error.message); } @@ -125,4 +164,4 @@ async function runCalibrationWorker({ intelligencePath, pollMs = 60000, workerId } } -module.exports = { refreshCalibration, createDecisions, refreshReplayEvaluations, runCalibrationWorker }; +module.exports = { refreshCalibration, refreshHistoricalCalibration, createDecisions, refreshReplayEvaluations, runCalibrationWorker };