const os = require('os'); const Database = require('better-sqlite3'); const { initAutonomySchema } = require('../src/autonomy/schema'); const { calibrateOutcomes, cohortKey } = require('../src/autonomy/calibration'); const { decide } = require('../src/autonomy/policy'); function sleep(ms) { return new Promise((resolve) => setTimeout(resolve, ms)); } function refreshCalibration(db, version = `cal-${Date.now()}`) { 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 `).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, []); map.get(key).push(row); return map; }, new Map()); 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 (?, ?, ?, ?, ?, ?, ?, ?, ?) `); 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; const result = calibrateOutcomes(rows); insert.run(key, result.sampleSize, result.effectiveSampleSize, result.directionalProbability, result.expectedExcessReturn, result.lowerReturn, result.upperReturn, null, version); } }); tx(); return groups.size; } 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' `).all(); const latest = db.prepare(` SELECT * FROM autonomy_calibration_snapshots WHERE cohort_key = ? ORDER BY created_at DESC, id DESC LIMIT 1 `); const insert = db.prepare(` INSERT INTO autonomy_decisions (prediction_id, action, calibrated_probability, expected_excess_return, rationale, strategy_version) VALUES (?, ?, ?, ?, ?, ?) `); let created = 0; const tx = db.transaction(() => { for (const prediction of predictions) { 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 }) : { action: 'ABSTAIN', rationale: 'calibration unavailable' }; insert.run(prediction.id, decision.action, calibration?.directional_probability || null, calibration?.expected_excess_return || null, decision.rationale, strategyVersion); created++; } }); tx(); return created; } async function runCalibrationWorker({ intelligencePath, pollMs = 60000, workerId = `calibration-${os.hostname()}-${process.pid}` } = {}) { const db = new Database(intelligencePath); db.pragma('journal_mode = WAL'); initAutonomySchema(db); 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 decisions = createDecisions(db); if (groups || decisions) console.log(`[${workerId}] calibration groups=${groups} decisions=${decisions}`); } catch (error) { console.error(`[${workerId}] calibration error:`, error.message); } await sleep(pollMs); } } module.exports = { refreshCalibration, createDecisions, runCalibrationWorker };