Files
Duriin-API/workers/outcomeAutonomyWorker.js
T

84 lines
4.0 KiB
JavaScript

const os = require('os');
const https = require('https');
const { openRuntimeDb } = require('../src/db/runtime');
const { initAutonomySchema } = require('../src/autonomy/schema');
const { calculateOutcome } = require('../src/autonomy/outcomes');
function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); }
function httpGet(url) {
return new Promise((resolve, reject) => {
const request = https.get(url, { headers: { 'User-Agent': 'duriin-autonomy/1.0' } }, (response) => {
let body = '';
response.setEncoding('utf8');
response.on('data', (chunk) => { body += chunk; });
response.on('end', () => response.statusCode >= 200 && response.statusCode < 300
? resolve(body) : reject(new Error(`market data returned ${response.statusCode}`)));
});
request.setTimeout(15000, () => request.destroy(new Error('market data timeout')));
request.on('error', reject);
});
}
async function history(symbol) {
// 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 [];
return (result.timestamp || []).map((timestamp, index) => ({
date: new Date(timestamp * 1000).toISOString().slice(0, 10),
close: result.indicators?.quote?.[0]?.close?.[index],
})).filter((row) => Number.isFinite(row.close));
}
async function resolveAutonomyOutcomes({ intelligencePath, workerId = `outcome-${os.hostname()}-${process.pid}`, pollMs = 60000 } = {}) {
const db = openRuntimeDb(intelligencePath, { schema: 'intelligence' });
db.pragma('journal_mode = WAL');
db.pragma('busy_timeout = 5000');
initAutonomySchema(db);
const cache = new Map();
while (true) {
const predictions = db.prepare(`
SELECT p.* FROM autonomy_predictions p
LEFT JOIN autonomy_outcomes o ON o.prediction_id = p.id
WHERE p.status = 'open' AND o.prediction_id IS NULL
AND datetime(p.information_cutoff, '+' || p.horizon_days || ' days') <= datetime('now')
ORDER BY p.information_cutoff ASC LIMIT 25
`).all();
for (const prediction of predictions) {
try {
if (!cache.has(prediction.instrument)) cache.set(prediction.instrument, await history(prediction.instrument));
if (!cache.has('SPY')) cache.set('SPY', await history('SPY'));
const result = calculateOutcome(prediction, cache.get(prediction.instrument), cache.get('SPY'));
if (!result) {
db.prepare("UPDATE autonomy_predictions SET status = 'unresolvable' WHERE id = ?").run(prediction.id);
continue;
}
db.prepare(`
INSERT INTO autonomy_outcomes
(prediction_id, price_0, price_horizon, benchmark_0, benchmark_horizon, excess_return, direction_correct, error_type)
VALUES (?, ?, ?, ?, ?, ?, ?, ?)
ON CONFLICT(prediction_id) DO UPDATE SET
price_0=excluded.price_0,
price_horizon=excluded.price_horizon,
benchmark_0=excluded.benchmark_0,
benchmark_horizon=excluded.benchmark_horizon,
excess_return=excluded.excess_return,
direction_correct=excluded.direction_correct,
error_type=excluded.error_type,
evaluated_at=datetime('now')
`).run(prediction.id, result.price0, result.priceHorizon, result.benchmark0, result.benchmarkHorizon,
result.excessReturn, result.directionCorrect, result.directionCorrect ? null : 'direction_error');
db.prepare("UPDATE autonomy_predictions SET status = 'resolved' WHERE id = ?").run(prediction.id);
} catch (error) {
console.error(`[autonomy-outcome] ${workerId} prediction ${prediction.id}:`, error.message);
}
await sleep(800);
}
await sleep(pollMs);
}
}
module.exports = { calculateOutcome, resolveAutonomyOutcomes };