KalBot/worker.js

import { MarketTracker } from './lib/market/tracker.js';
import { PaperEngine } from './lib/paper/engine.js';
import { MartingaleStrategy } from './lib/strategies/martingale.js';
import { MartingaleAlphaStrategy } from './lib/strategies/martingale-alpha.js';
import { ThresholdStrategy } from './lib/strategies/threshold.js';
import { getMarket } from './lib/kalshi/rest.js';
import { db } from './lib/db.js';
import { notify } from './lib/notify.js';
import fs from 'fs';

const STATE_FILE = '/tmp/kalbot-state.json';
const HEARTBEAT_MS = 2000;

async function main() {
  console.log('=== Kalbot Worker Starting ===');

  await db.connect();

  const paper = new PaperEngine(1000);
  await paper.init();

  const strategies = [
    new MartingaleStrategy({ threshold: 70, baseBet: 1, maxDoublings: 5 }),
    new MartingaleAlphaStrategy({ minPct: 40, maxPct: 60, baseBet: 1, maxRounds: 3 }),
    new ThresholdStrategy({ triggerPct: 65, betSize: 1 })
  ];

  // Ensure each strategy has a paper account initialized
  for (const s of strategies) {
    paper._getAccount(s.name);
  }

  console.log(`[Worker] Loaded ${strategies.length} strategies: ${strategies.map((s) => s.name).join(', ')}`);

  // Settle any orphaned positions from previous runs before starting
  try {
    await paper.settleOrphans(getMarket);
  } catch (e) {
    console.error('[Worker] Orphan settlement error:', e.message);
  }

  const tracker = new MarketTracker();
  let latestMarketState = null;
  let heartbeatTimer = null;

  writeState(latestMarketState, paper, strategies);

  tracker.on('update', async (state) => {
    latestMarketState = state || null;
    writeState(latestMarketState, paper, strategies);

    if (!state) return;
    if (paper._resetting) return;

    for (const strategy of strategies) {
      if (!strategy.enabled) continue;

      const signal = strategy.evaluate(state);
      if (signal) {
        console.log(`[Worker] Signal from ${strategy.name}: ${signal.side} @ ${signal.price}¢ — ${signal.reason}`);

        if (strategy.mode === 'paper') {
          await paper.executeTrade(signal, state);
        }
      }
    }

    writeState(latestMarketState, paper, strategies);
  });

  tracker.on('settled', async ({ ticker, result }) => {
    console.log(`[Worker] Market ${ticker} settled: ${result}`);

    if (paper._resetting) {
      console.log(`[Worker] Skipping settlement for ${ticker} — reset in progress`);
      return;
    }

    const settledPositions = await paper.settle(ticker, result);

    for (const strategy of strategies) {
      if (!settledPositions) continue;
      for (const trade of settledPositions) {
        strategy.onSettlement(trade.result, trade);
      }
    }

    await notify(
      `Market ${ticker} settled: ${result?.toUpperCase() || 'unknown'}`,
      'Market Settled',
      'default',
      'chart_with_upwards_trend'
    );

    writeState(latestMarketState, paper, strategies);
  });

  await tracker.start();
  await notify('🤖 Kalbot Worker started!', 'Kalbot Online', 'low', 'robot,green_circle');

  heartbeatTimer = setInterval(() => {
    writeState(latestMarketState, paper, strategies);
  }, HEARTBEAT_MS);

  // Check for reset flag periodically
  setInterval(async () => {
    try {
      if (fs.existsSync('/tmp/kalbot-reset-flag')) {
        fs.unlinkSync('/tmp/kalbot-reset-flag');
        console.log('[Worker] Reset flag detected — resetting all paper data');
        await paper.resetAll();
        for (const s of strategies) {
          if (s.consecutiveLosses !== undefined) s.consecutiveLosses = 0;
          if (s.currentBetSize !== undefined) s.currentBetSize = s.config.baseBet;
          if (s.round !== undefined) s.round = 0;
          if (s.cycleWins !== undefined) s.cycleWins = 0;
          if (s.cycleLosses !== undefined) s.cycleLosses = 0;
          if (s.totalCycles !== undefined) s.totalCycles = 0;
          s.lastTradeTicker = null;
          s.lastTradeTime = 0;
        }
        writeState(latestMarketState, paper, strategies);
        await notify('🔄 Paper trading reset by admin', 'Kalbot Reset', 'default', 'recycle');
      }
    } catch {
      // ignore
    }
  }, 1000);

  console.log('[Worker] Running. Press Ctrl+C to stop.');

  const shutdown = async (signal) => {
    console.log(`\n[Worker] ${signal} received. Shutting down...`);
    clearInterval(heartbeatTimer);
    tracker.stop();
    await notify('🔴 Kalbot Worker stopped', 'Kalbot Offline', 'high', 'robot,red_circle');
    process.exit(0);
  };

  process.on('SIGINT', () => shutdown('SIGINT'));
  process.on('SIGTERM', () => shutdown('SIGTERM'));
}

function writeState(marketState, paper, strategies) {
  const data = {
    market: marketState,
    paper: paper.getStats(),
    paperByStrategy: paper.getPerStrategyStats(),
    strategies: strategies.map((s) => s.toJSON()),
    workerUptime: process.uptime(),
    lastUpdate: Date.now()
  };

  try {
    fs.writeFileSync(STATE_FILE, JSON.stringify(data));
  } catch (e) {
    // Non-critical
  }
}

main().catch((err) => {
  console.error('[Worker] Fatal error:', err);
  process.exit(1);
});