diff --git a/docker-compose.yml b/docker-compose.yml index fc07f88..e3f9200 100644 --- a/docker-compose.yml +++ b/docker-compose.yml @@ -92,6 +92,28 @@ services: networks: - nginx_proxy_manager_default + replay: + build: + context: . + provenance: false + command: node workers/replay-entrypoint.js + env_file: .env + volumes: + - ./config.json:/app/config.json:ro + - ./data:/data + environment: + NODE_ENV: production + DURIIN_DB: /data/archive.sqlite + INTELLIGENCE_DB: /data/intelligence.sqlite + AUTONOMY_REPLAY_POLL_MS: "${AUTONOMY_REPLAY_POLL_MS:-15000}" + AUTONOMY_REPLAY_DAILY_BUDGET: "${AUTONOMY_REPLAY_DAILY_BUDGET:-100}" + AUTONOMY_REPLAY_WATERMARK_DAYS: "${AUTONOMY_REPLAY_WATERMARK_DAYS:-7}" + cpus: "0.50" + mem_limit: 768m + restart: unless-stopped + networks: + - nginx_proxy_manager_default + calibration: build: context: . diff --git a/package.json b/package.json index 13a24d9..51629e9 100644 --- a/package.json +++ b/package.json @@ -10,7 +10,8 @@ "autonomy:init": "node scripts/initialize-autonomy.js", "autonomy:allowlist": "node scripts/set-autonomy-instrument.js", "autonomy:sync-assets": "node scripts/sync-paper-assets.js", - "autonomy:import-legacy": "node scripts/import-legacy-intelligence.js" + "autonomy:import-legacy": "node scripts/import-legacy-intelligence.js", + "autonomy:replay": "node workers/replay-entrypoint.js" }, "keywords": [], "author": "", diff --git a/public/admin/assets/js/autonomy.js b/public/admin/assets/js/autonomy.js index 341d84b..fc0c204 100644 --- a/public/admin/assets/js/autonomy.js +++ b/public/admin/assets/js/autonomy.js @@ -79,6 +79,12 @@ byId("perf-cohorts").textContent = formatNumber(data.calibration?.length || 0); if (outcomes) byId("performance-note").textContent = `Measured on ${outcomes} matured predictions. Results remain descriptive until the sample is large enough for stable calibration.`; + const replay = data.replay; + byId("replay-status").textContent = replay ? replay.status : "Not started"; + byId("replay-articles").textContent = replay ? formatNumber(replay.processed_articles || 0) : "—"; + byId("replay-evaluations").textContent = replay ? formatNumber(replay.evaluations || 0) : "—"; + byId("replay-watermark").textContent = replay?.watermark_at ? formatRelative(replay.watermark_at) : "—"; + byId("runtime-queue").textContent = `${formatNumber(pendingHistorical, true)} pending`; byId("runtime-coordinator").textContent = pendingHistorical ? "Processing" : "Watching"; if (data.account) { diff --git a/public/admin/pages/autonomy.html b/public/admin/pages/autonomy.html index abfb5b8..6ed6f9a 100644 --- a/public/admin/pages/autonomy.html +++ b/public/admin/pages/autonomy.html @@ -128,6 +128,16 @@ + +
+
06

Historical replay

Isolated from execution
+
+
Run stateStarting…
+
Articles replayed—
+
Walk-forward evaluations—
+
Watermark—
+
+
diff --git a/src/autonomy/coordinator.js b/src/autonomy/coordinator.js index 576b740..10a5017 100644 --- a/src/autonomy/coordinator.js +++ b/src/autonomy/coordinator.js @@ -34,9 +34,22 @@ function normalizeProposal(raw, { informationCutoff, model = 'unknown', promptVe }; } -function verifyEvidence(archiveDb, articleIds) { +function verifyEvidence(archiveDb, articleIds, informationCutoff = null) { const placeholders = articleIds.map(() => '?').join(','); - const rows = archiveDb.prepare(`SELECT id FROM articles WHERE id IN (${placeholders})`).all(...articleIds); + // A replay must only see material which existed at its information cutoff. + // Live proposals retain the simpler existence check. + const cutoffClause = informationCutoff ? ' AND datetime(COALESCE(pub_date_effective, pub_date, ingested_at)) <= datetime(?)' : ''; + let rows; + try { + rows = archiveDb.prepare(`SELECT id FROM articles WHERE id IN (${placeholders})${cutoffClause}`) + .all(...articleIds, ...(informationCutoff ? [informationCutoff] : [])); + } catch (error) { + // Minimal/test archives may not retain publication metadata. A production + // replay archive is required to have it, so this fallback is only for the + // existing live evidence contract. + if (informationCutoff) throw error; + rows = archiveDb.prepare(`SELECT id FROM articles WHERE id IN (${placeholders})`).all(...articleIds); + } const found = new Set(rows.map((row) => row.id)); return articleIds.every((id) => found.has(id)); } @@ -44,7 +57,7 @@ function verifyEvidence(archiveDb, articleIds) { function acceptProposal(intelligenceDb, archiveDb, raw, metadata = {}) { const proposal = normalizeProposal(raw, metadata); for (const prediction of proposal.predictions) { - if (!verifyEvidence(archiveDb, prediction.evidenceArticleIds)) { + if (!verifyEvidence(archiveDb, prediction.evidenceArticleIds, metadata.origin === 'replay' ? proposal.informationCutoff : null)) { throw new Error(`proposal references missing evidence for ${prediction.instrument}`); } const instrument = intelligenceDb.prepare( @@ -60,8 +73,8 @@ function acceptProposal(intelligenceDb, archiveDb, raw, metadata = {}) { const insertPrediction = intelligenceDb.prepare(` INSERT INTO autonomy_predictions (proposal_id, event_id, instrument, direction, event_type, causal_channel, - horizon_days, information_cutoff, evidence_article_ids, invalidation_condition, learning_eligible, strategy_version) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) + horizon_days, information_cutoff, evidence_article_ids, invalidation_condition, learning_eligible, strategy_version, origin, replay_run_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?) `); const tx = intelligenceDb.transaction(() => { const proposalResult = insert.run(metadata.eventId || null, JSON.stringify(proposal), proposal.informationCutoff, @@ -70,7 +83,8 @@ function acceptProposal(intelligenceDb, archiveDb, raw, metadata = {}) { insertPrediction.run(proposalResult.lastInsertRowid, metadata.eventId || null, prediction.instrument, prediction.direction, prediction.eventType, prediction.causalChannel, prediction.horizonDays, proposal.informationCutoff, JSON.stringify(prediction.evidenceArticleIds), prediction.invalidationCondition, - metadata.learningEligible ? 1 : 0, metadata.strategyVersion || 'autonomy-1'); + metadata.learningEligible ? 1 : 0, metadata.strategyVersion || 'autonomy-1', + metadata.origin || 'live', metadata.replayRunId || null); } return Number(proposalResult.lastInsertRowid); }); diff --git a/src/autonomy/schema.js b/src/autonomy/schema.js index 2fe593a..eba4b91 100644 --- a/src/autonomy/schema.js +++ b/src/autonomy/schema.js @@ -1,4 +1,4 @@ -const AUTONOMY_SCHEMA_VERSION = 1; +const AUTONOMY_SCHEMA_VERSION = 2; function initAutonomySchema(db) { db.exec(` @@ -178,12 +178,52 @@ function initAutonomySchema(db) { captured_at TEXT NOT NULL DEFAULT (datetime('now')) ); + -- Historical replay is a separate evidence path. It deliberately never + -- writes to autonomy_decisions or order intents. + CREATE TABLE IF NOT EXISTS autonomy_replay_runs ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + status TEXT NOT NULL DEFAULT 'running' + CHECK (status IN ('running', 'paused', 'complete', 'superseded', 'failed')), + start_at TEXT, + watermark_at TEXT, + cursor_article_id INTEGER NOT NULL DEFAULT 0, + cursor_effective_at TEXT, + processed_articles INTEGER NOT NULL DEFAULT 0, + strategy_version TEXT NOT NULL, + prompt_version TEXT NOT NULL, + coordinator_model TEXT, + price_provider TEXT NOT NULL DEFAULT 'yahoo', + created_at TEXT NOT NULL DEFAULT (datetime('now')), + updated_at TEXT NOT NULL DEFAULT (datetime('now')), + completed_at TEXT, + last_error TEXT + ); + CREATE INDEX IF NOT EXISTS idx_autonomy_replay_runs_active + ON autonomy_replay_runs(status, cursor_article_id); + + CREATE TABLE IF NOT EXISTS autonomy_replay_evaluations ( + prediction_id INTEGER PRIMARY KEY REFERENCES autonomy_predictions(id), + replay_run_id INTEGER NOT NULL REFERENCES autonomy_replay_runs(id), + snapshot_cutoff TEXT NOT NULL, + sample_size INTEGER NOT NULL, + action TEXT NOT NULL CHECK (action IN ('BUY', 'SELL', 'HOLD', 'ABSTAIN')), + calibrated_probability REAL, + expected_excess_return REAL, + rationale TEXT NOT NULL, + created_at TEXT NOT NULL DEFAULT (datetime('now')) + ); + INSERT OR IGNORE INTO autonomy_schema(version) VALUES (${AUTONOMY_SCHEMA_VERSION}); `); for (const statement of [ 'ALTER TABLE autonomy_order_intents ADD COLUMN attempts INTEGER NOT NULL DEFAULT 0', 'ALTER TABLE autonomy_order_intents ADD COLUMN last_error TEXT', 'ALTER TABLE autonomy_predictions ADD COLUMN learning_eligible INTEGER NOT NULL DEFAULT 0', + "ALTER TABLE autonomy_predictions ADD COLUMN origin TEXT NOT NULL DEFAULT 'live'", + 'ALTER TABLE autonomy_predictions ADD COLUMN replay_run_id INTEGER', + 'ALTER TABLE autonomy_replay_runs ADD COLUMN cursor_effective_at TEXT', + "ALTER TABLE autonomy_calibration_snapshots ADD COLUMN source TEXT NOT NULL DEFAULT 'live'", + 'ALTER TABLE autonomy_calibration_snapshots ADD COLUMN replay_run_id INTEGER', ]) { try { db.exec(statement); } catch (_) {} } diff --git a/src/routes/admin.js b/src/routes/admin.js index 8f2859c..7b3394b 100644 --- a/src/routes/admin.js +++ b/src/routes/admin.js @@ -174,7 +174,9 @@ async function adminRoutes(fastify) { const outcomeSummary = intelligenceDb.prepare(` SELECT COUNT(*) AS total, SUM(direction_correct) AS correct, AVG(excess_return) AS average_excess_return - FROM autonomy_outcomes + FROM autonomy_outcomes o + JOIN autonomy_predictions p ON p.id = o.prediction_id + WHERE p.origin = 'live' `).get(); const instruments = intelligenceDb.prepare(` SELECT COUNT(*) AS count FROM autonomy_instruments WHERE active=1 AND tradable=1 @@ -214,6 +216,18 @@ async function adminRoutes(fastify) { expected_excess_return, lower_return, upper_return, created_at FROM autonomy_calibration_snapshots ORDER BY id DESC LIMIT 8 `).all(); + const replay = intelligenceDb.prepare(` + SELECT r.id, r.status, r.watermark_at, r.cursor_article_id, r.cursor_effective_at, + r.processed_articles, r.updated_at, + SUM(CASE WHEN p.status = 'resolved' THEN 1 ELSE 0 END) AS resolved_predictions, + COUNT(p.id) AS predictions, + SUM(o.direction_correct) AS correct_predictions, + (SELECT COUNT(*) FROM autonomy_replay_evaluations e WHERE e.replay_run_id = r.id) AS evaluations + FROM autonomy_replay_runs r + LEFT JOIN autonomy_predictions p ON p.replay_run_id = r.id + LEFT JOIN autonomy_outcomes o ON o.prediction_id = p.id + GROUP BY r.id ORDER BY r.id DESC LIMIT 1 + `).get() || null; return { enabled: true, @@ -232,6 +246,7 @@ async function adminRoutes(fastify) { latestOrders, account, calibration, + replay, generatedAt: new Date().toISOString(), }; }); diff --git a/test/autonomy.test.js b/test/autonomy.test.js index bcb2b7f..190b099 100644 --- a/test/autonomy.test.js +++ b/test/autonomy.test.js @@ -53,6 +53,26 @@ test('accepted proposal preserves evidence and creates immutable prediction', () assert.deepEqual(JSON.parse(intelligence.prepare('SELECT evidence_article_ids FROM autonomy_predictions').get().evidence_article_ids), [7]); }); +test('replay evidence cannot look beyond its information cutoff and remains non-executable', () => { + const archive = new Database(':memory:'); + archive.exec('CREATE TABLE articles (id INTEGER PRIMARY KEY, pub_date_effective TEXT, pub_date TEXT, ingested_at TEXT)'); + archive.prepare("INSERT INTO articles VALUES (7, '2020-01-01T00:00:00Z', NULL, '2020-01-01T00:00:00Z')").run(); + const intelligence = new Database(':memory:'); + initAutonomySchema(intelligence); + intelligence.prepare("INSERT INTO autonomy_instruments(symbol, broker, active, tradable) VALUES ('NVDA', 'test', 1, 1)").run(); + assert.throws(() => acceptProposal(intelligence, archive, { + predictions: [{ instrument: 'NVDA', direction: 'positive', event_type: 'earnings', horizon_days: 10, evidence_article_ids: [7] }], + }, { informationCutoff: '2019-12-31T00:00:00Z', origin: 'replay', replayRunId: 1 }), /missing evidence/); + const proposal = intelligence.prepare("INSERT INTO autonomy_proposals(payload, information_cutoff, status) VALUES ('{}', datetime('now'), 'accepted')").run(); + const prediction = intelligence.prepare(`INSERT INTO autonomy_predictions + (proposal_id, instrument, direction, event_type, horizon_days, information_cutoff, evidence_article_ids, strategy_version, origin) + VALUES (?, 'NVDA', 'positive', 'test', 10, datetime('now'), '[7]', 'test', 'replay')`).run(proposal.lastInsertRowid); + intelligence.prepare("INSERT INTO autonomy_decisions(prediction_id, action, rationale, strategy_version) VALUES (?, 'BUY', 'test', 'test')").run(prediction.lastInsertRowid); + const executable = intelligence.prepare(`SELECT d.id FROM autonomy_decisions d JOIN autonomy_predictions p ON p.id=d.prediction_id + WHERE d.action IN ('BUY','SELL') AND p.origin='live'`).all(); + assert.equal(executable.length, 0); +}); + test('calibration and policy abstain on insufficient evidence', () => { const calibration = calibrateOutcomes([ { excess_return: 0.02, direction_correct: 1 }, diff --git a/workers/calibrationWorker.js b/workers/calibrationWorker.js index e293428..0e8d161 100644 --- a/workers/calibrationWorker.js +++ b/workers/calibrationWorker.js @@ -11,7 +11,7 @@ function refreshCalibration(db, version = `cal-${Date.now()}`) { 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 + WHERE p.status = 'resolved' AND p.learning_eligible = 1 AND p.origin = 'live' `).all().reduce((map, row) => { const key = cohortKey({ direction: row.direction, eventType: row.event_type, horizonDays: row.horizon_days }); if (!map.has(key)) map.set(key, []); @@ -21,8 +21,8 @@ function refreshCalibration(db, version = `cal-${Date.now()}`) { const insert = 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) - VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?) + expected_excess_return, lower_return, upper_return, parent_cohort_key, version, source, replay_run_id) + VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, 'live', NULL) `); const tx = db.transaction(() => { for (const [key, rows] of groups) { @@ -40,7 +40,7 @@ function createDecisions(db, strategyVersion = 'autonomy-1') { const predictions = db.prepare(` SELECT p.* FROM autonomy_predictions p LEFT JOIN autonomy_decisions d ON d.prediction_id = p.id - WHERE d.prediction_id IS NULL AND p.status = 'open' + WHERE d.prediction_id IS NULL AND p.status = 'open' AND p.origin = 'live' `).all(); const latest = db.prepare(` SELECT * FROM autonomy_calibration_snapshots @@ -68,6 +68,44 @@ function createDecisions(db, strategyVersion = 'autonomy-1') { return created; } +// Replay evaluations are walk-forward: each historical prediction is scored +// against calibration data that had matured strictly before its cutoff. They +// are stored in their own ledger, never in autonomy_decisions. +function refreshReplayEvaluations(db) { + const predictions = db.prepare(` + SELECT p.*, o.excess_return, o.direction_correct + FROM autonomy_predictions p JOIN autonomy_outcomes o ON o.prediction_id = p.id + LEFT JOIN autonomy_replay_evaluations e ON e.prediction_id = p.id + WHERE p.origin = 'replay' AND p.status = 'resolved' AND e.prediction_id IS NULL + ORDER BY datetime(p.information_cutoff), p.id LIMIT 200 + `).all(); + const prior = 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.origin = 'replay' AND p.status = 'resolved' + AND datetime(p.information_cutoff, '+' || p.horizon_days || ' days') < datetime(?) + `); + const insert = db.prepare(` + INSERT INTO autonomy_replay_evaluations + (prediction_id, replay_run_id, snapshot_cutoff, sample_size, action, calibrated_probability, expected_excess_return, rationale) + VALUES (?, ?, ?, ?, ?, ?, ?, ?) + `); + const tx = db.transaction(() => { + for (const prediction of predictions) { + const key = cohortKey({ direction: prediction.direction, eventType: prediction.event_type, horizonDays: prediction.horizon_days }); + const rows = prior.all(prediction.information_cutoff).filter((row) => + cohortKey({ direction: row.direction, eventType: row.event_type, horizonDays: row.horizon_days }) === key); + const calibration = rows.length ? calibrateOutcomes(rows) : null; + const decision = calibration ? decide({ ...calibration, direction: prediction.direction }, { minSampleSize: 30 }) + : { action: 'ABSTAIN', rationale: 'walk-forward calibration unavailable' }; + insert.run(prediction.id, prediction.replay_run_id, prediction.information_cutoff, rows.length, decision.action, + calibration?.directionalProbability || null, calibration?.expectedExcessReturn || null, decision.rationale); + } + }); + tx(); + return predictions.length; +} + async function runCalibrationWorker({ intelligencePath, pollMs = 60000, workerId = `calibration-${os.hostname()}-${process.pid}` } = {}) { const db = new Database(intelligencePath); db.pragma('journal_mode = WAL'); @@ -78,7 +116,8 @@ async function runCalibrationWorker({ intelligencePath, pollMs = 60000, workerId 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 decisions = createDecisions(db); - if (groups || decisions) console.log(`[${workerId}] calibration groups=${groups} decisions=${decisions}`); + const replayEvaluations = refreshReplayEvaluations(db); + if (groups || decisions || replayEvaluations) console.log(`[${workerId}] calibration groups=${groups} decisions=${decisions} replay_evaluations=${replayEvaluations}`); } catch (error) { console.error(`[${workerId}] calibration error:`, error.message); } @@ -86,4 +125,4 @@ async function runCalibrationWorker({ intelligencePath, pollMs = 60000, workerId } } -module.exports = { refreshCalibration, createDecisions, runCalibrationWorker }; +module.exports = { refreshCalibration, createDecisions, refreshReplayEvaluations, runCalibrationWorker }; diff --git a/workers/executionWorker.js b/workers/executionWorker.js index 539b71c..0521744 100644 --- a/workers/executionWorker.js +++ b/workers/executionWorker.js @@ -51,8 +51,9 @@ async function runExecutionWorker({ intelligencePath, pollMs = 10000, mode = 'sh } const decisions = db.prepare(` SELECT d.id FROM autonomy_decisions d + JOIN autonomy_predictions p ON p.id = d.prediction_id LEFT JOIN autonomy_order_intents oi ON oi.decision_id = d.id - WHERE oi.id IS NULL AND d.action IN ('BUY', 'SELL') + WHERE oi.id IS NULL AND d.action IN ('BUY', 'SELL') AND p.origin = 'live' ORDER BY d.created_at ASC LIMIT 25 `).all(); for (const decision of decisions) { diff --git a/workers/outcomeAutonomyWorker.js b/workers/outcomeAutonomyWorker.js index 81727b4..05e7716 100644 --- a/workers/outcomeAutonomyWorker.js +++ b/workers/outcomeAutonomyWorker.js @@ -20,7 +20,9 @@ function httpGet(url) { } async function history(symbol) { - const url = `https://query1.finance.yahoo.com/v8/finance/chart/${encodeURIComponent(symbol)}?range=10y&interval=1d`; + // GDELT backfills predate the normal rolling quote window. Use an explicit + // point-in-time range so replay outcomes do not silently become unresolvable. + const url = `https://query1.finance.yahoo.com/v8/finance/chart/${encodeURIComponent(symbol)}?period1=946684800&period2=${Math.floor(Date.now() / 1000)}&interval=1d`; const body = JSON.parse(await httpGet(url)); const result = body?.chart?.result?.[0]; if (!result) return []; diff --git a/workers/replay-entrypoint.js b/workers/replay-entrypoint.js new file mode 100644 index 0000000..a32c409 --- /dev/null +++ b/workers/replay-entrypoint.js @@ -0,0 +1,8 @@ +const path = require('path'); +const { runReplayWorker } = require('./replayWorker'); + +runReplayWorker({ + archivePath: process.env.DURIIN_DB || path.resolve('/data/archive.sqlite'), + intelligencePath: process.env.INTELLIGENCE_DB || path.resolve('/data/intelligence.sqlite'), + pollMs: Number(process.env.AUTONOMY_REPLAY_POLL_MS) || 15000, +}).catch((error) => { console.error('[replay] fatal:', error); process.exit(1); }); diff --git a/workers/replayWorker.js b/workers/replayWorker.js new file mode 100644 index 0000000..8288f49 --- /dev/null +++ b/workers/replayWorker.js @@ -0,0 +1,113 @@ +const os = require('os'); +const fs = require('fs'); +const path = require('path'); +const Database = require('better-sqlite3'); +const { initAutonomySchema } = require('../src/autonomy/schema'); +const { enqueueJob, leaseNextJob, completeJob, failJob } = require('../src/autonomy/jobs'); +const { callCoordinator } = require('../src/autonomy/llm'); +const { acceptProposal, recordRejectedProposal } = require('../src/autonomy/coordinator'); + +function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); } + +function loadConfig() { + const configPath = path.resolve(process.env.DURIIN_CONFIG || path.join(__dirname, '..', 'config.json')); + const raw = JSON.parse(fs.readFileSync(configPath, 'utf8')); + require('dotenv').config({ path: path.resolve(path.dirname(configPath), '.env') }); + raw.openRouter = { ...(raw.openRouter || {}) }; + if (process.env.OPEN_ROUTER_API_KEY) raw.openRouter.apiKey = process.env.OPEN_ROUTER_API_KEY; + if (process.env.OPEN_ROUTER_LLM_MODEL) raw.openRouter.llmModel = process.env.OPEN_ROUTER_LLM_MODEL; + return raw; +} + +function articleTimeColumns(archiveDb) { + const columns = new Set(archiveDb.prepare('PRAGMA table_info(articles)').all().map((row) => row.name)); + const candidates = ['pub_date_effective', 'pub_date', 'ingested_at'].filter((name) => columns.has(name)); + if (!candidates.length) throw new Error('archive articles need publication or ingestion timestamps for replay'); + return { columns, effective: `COALESCE(${candidates.join(', ')})` }; +} + +function replayPrompt(article) { + return `Historical evidence cutoff: ${article.effective_at}\n\n` + + `[Evidence 1] article_id=${article.id}\nTitle: ${article.title || ''}\n${String(article.content || article.description || '').slice(0, 6000)}\n\n` + + `Return JSON only in this shape:\n${JSON.stringify({ predictions: [{ + instrument: 'NVDA', direction: 'positive|negative', event_type: 'stable_enum', + causal_channel: 'short description', horizon_days: 10, evidence_article_ids: [123], + invalidation_condition: 'condition', + }] }, null, 2)}\n\n` + + 'Use only this dated evidence. Return an empty predictions array when there is no clear, tradable hypothesis. Never include probabilities, returns, confidence, position sizes, or actions.'; +} + +function activeRun(db, config) { + let run = db.prepare("SELECT * FROM autonomy_replay_runs WHERE status = 'running' ORDER BY id DESC LIMIT 1").get(); + if (run) return run; + const watermarkDays = Math.max(1, Number(process.env.AUTONOMY_REPLAY_WATERMARK_DAYS) || 7); + const result = db.prepare(` + INSERT INTO autonomy_replay_runs (watermark_at, strategy_version, prompt_version, coordinator_model) + VALUES (datetime('now', ?), 'autonomy-1', 'replay-coordinator-1', ?) + `).run(`-${watermarkDays} days`, config.openRouter.llmModel || 'unknown'); + return db.prepare('SELECT * FROM autonomy_replay_runs WHERE id = ?').get(result.lastInsertRowid); +} + +function scheduleNext(db, archiveDb, run) { + const { columns, effective } = articleTimeColumns(archiveDb); + const content = columns.has('content') ? "content IS NOT NULL AND content != ''" : '1=1'; + const indexFilter = columns.has('is_index_page') ? 'AND (is_index_page = 0 OR is_index_page IS NULL)' : ''; + const cursorFilter = run.cursor_effective_at + ? `AND (datetime(${effective}) > datetime(?) OR (datetime(${effective}) = datetime(?) AND id > ?))` + : ''; + const params = [run.watermark_at]; + if (run.cursor_effective_at) params.push(run.cursor_effective_at, run.cursor_effective_at, run.cursor_article_id); + const article = archiveDb.prepare(` + SELECT id, title, description, content, ${effective} AS effective_at + FROM articles + WHERE ${content} ${indexFilter} AND datetime(${effective}) <= datetime(?) ${cursorFilter} + ORDER BY datetime(${effective}) ASC, id ASC LIMIT 1 + `).get(...params); + if (!article) return null; + enqueueJob(db, { + jobType: 'replay_article', lane: 'historical', priority: 1, entityType: 'article', entityId: article.id, + idempotencyKey: `replay:${run.id}:article:${article.id}`, + }); + return article; +} + +async function runReplayWorker({ archivePath, intelligencePath, workerId = `replay-${os.hostname()}-${process.pid}`, pollMs = 15000 } = {}) { + const archiveDb = new Database(archivePath, { readonly: true }); + const db = new Database(intelligencePath); + db.pragma('journal_mode = WAL'); + db.pragma('busy_timeout = 5000'); + initAutonomySchema(db); + const config = loadConfig(); + const dailyBudget = Math.max(1, Number(process.env.AUTONOMY_REPLAY_DAILY_BUDGET) || 100); + while (true) { + try { + const completedToday = db.prepare("SELECT COUNT(*) AS count FROM autonomy_jobs WHERE job_type='replay_article' AND status='complete' AND date(completed_at) = date('now')").get().count; + if (completedToday >= dailyBudget) { await sleep(Math.max(pollMs, 60000)); continue; } + const run = activeRun(db, config); + scheduleNext(db, archiveDb, run); + const job = leaseNextJob(db, workerId, 300, ['replay_article']); + if (!job) { await sleep(pollMs); continue; } + try { + const { effective } = articleTimeColumns(archiveDb); + const article = archiveDb.prepare(`SELECT id, title, description, content, ${effective} AS effective_at FROM articles WHERE id=?`).get(job.entity_id); + if (!article || !article.effective_at) throw new Error(`replay article ${job.entity_id} is unavailable`); + const raw = await callCoordinator(config, replayPrompt(article)); + try { + acceptProposal(db, archiveDb, raw, { + informationCutoff: article.effective_at, model: config.openRouter.llmModel || 'unknown', + promptVersion: 'replay-coordinator-1', strategyVersion: 'autonomy-1', learningEligible: false, + origin: 'replay', replayRunId: run.id, + }); + } catch (validationError) { + recordRejectedProposal(db, raw, { informationCutoff: article.effective_at, model: config.openRouter.llmModel || 'unknown', promptVersion: 'replay-coordinator-1' }, validationError.message); + } + db.prepare(`UPDATE autonomy_replay_runs SET cursor_article_id=?, cursor_effective_at=?, processed_articles=processed_articles+1, updated_at=datetime('now') WHERE id=?`) + .run(article.id, article.effective_at, run.id); + completeJob(db, job.id, workerId); + } catch (error) { failJob(db, job.id, workerId, error); } + } catch (error) { console.error(`[${workerId}] replay error:`, error.message); } + await sleep(pollMs); + } +} + +module.exports = { articleTimeColumns, replayPrompt, scheduleNext, runReplayWorker };