Duriin-API/workers/calibrationWorker.js

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 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.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);
    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, source, replay_run_id)
    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 = ? 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, 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
    LEFT JOIN autonomy_decisions d ON d.prediction_id = p.id
    WHERE d.prediction_id IS NULL AND p.status = 'open' AND p.origin = 'live'
  `).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(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);
      created++;
    }
  });
  tx();
  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');
  db.pragma('busy_timeout = 5000');
  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 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 || 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);
    }
    await sleep(pollMs);
  }
}

module.exports = { refreshCalibration, refreshHistoricalCalibration, createDecisions, refreshReplayEvaluations, runCalibrationWorker };