`src/workers/cron-runner.js`

62 lines
const cron = require('node-cron');

const BASE_URL = 'http://127.0.0.1:7800';

async function runWorker(name, url, body) {
  const start = Date.now();
  console.log(`[Cron] Starting ${name}...`);
  try {
    const res = await fetch(url, {
      method: 'POST',
      headers: { 'Content-Type': 'application/json' },
      body: JSON.stringify(body),
    });
    const data = await res.json();
    console.log(`[Cron] ${name} completed in ${Date.now() - start}ms:`, JSON.stringify(data).substring(0, 200));
    return data;
  } catch (err) {
    console.error(`[Cron] ${name} error:`, err.message);
    return { error: err.message };
  }
}

// Market ingest: every 30 minutes
cron.schedule('*/30 * * * *', () => {
  runWorker('market_ingest', `${BASE_URL}/api/markets`, { action: 'ingest' });
});

// Weather fetch all cities: every 2 hours
cron.schedule('0 */2 * * *', () => {
  runWorker('weather_fetch_all', `${BASE_URL}/api/weather`, { action: 'fetch_all' });
});

// Run predictions: every hour
cron.schedule('0 * * * *', () => {
  runWorker('run_predictions', `${BASE_URL}/api/models`, { action: 'run_predictions' });
});

// Daily risk reset at midnight UTC
cron.schedule('0 0 * * *', async () => {
  console.log('[Cron] Daily risk reset');
  try {
    const { query } = require('../lib/db');
    await query("UPDATE risk_state SET daily_pnl = 0, updated_at = NOW() WHERE id = (SELECT id FROM risk_state ORDER BY id DESC LIMIT 1)");
    console.log('[Cron] Daily P&L reset to 0');
  } catch (err) {
    console.error('[Cron] Risk reset error:', err.message);
  }
});

// Health check ping: every 5 minutes
cron.schedule('*/5 * * * *', async () => {
  try {
    const res = await fetch(`${BASE_URL}/api/health`);
    if (!res.ok) console.warn('[Cron] Health check failed:', res.status);
  } catch (err) {
    console.error('[Cron] Health check error:', err.message);
  }
});

console.log('[Cron] Bertha cron runner started');
console.log('[Cron] Schedules: market_ingest(30m), weather_fetch(2h), predictions(1h), risk_reset(midnight), health(5m)');