poly-weather/packages/trading-engine/src/bot.ts

/**
 * Main bot orchestrator.
 * Manages the lifecycle of the trading pipeline, polling, and WebSocket feeds.
 */

import type { BotConfig, CityConfig, ModelForecasts } from '@poly-weather/core';
import { createChildLogger, loadConfig, TARGET_CITIES } from '@poly-weather/core';
import { startModelPolling } from '@poly-weather/weather-engine';
import {
  CLOBTradingClient,
  OrderManager,
  WebSocketFeed,
  scanWeatherMarkets,
  groupMarketsByEvent,
  PositionSyncService,
} from '@poly-weather/market-engine';
import { TradingPipeline } from './pipeline.js';
import { HealthCheckServer } from './health-check.js';
import {
  disconnectDb,
  getPendingTrades,
  updateTradeStatus,
  getUnsettledTrades,
  settleTradePnl,
  aggregateDailyMetrics,
} from './db.js';

const logger = createChildLogger('bot');

function sleep(ms: number): Promise<void> {
  return new Promise((resolve) => setTimeout(resolve, ms));
}

export class WeatherArbitrageBot {
  private config: BotConfig;
  private pipeline!: TradingPipeline;
  private healthCheck!: HealthCheckServer;
  private positionSync!: PositionSyncService;
  private wsFeed: WebSocketFeed | null = null;
  private stopModelPolling: (() => void) | null = null;
  private running = false;
  private intervals: ReturnType<typeof setInterval>[] = [];

  constructor(config?: BotConfig) {
    this.config = config || loadConfig();
  }

  /**
   * Start the bot.
   */
  async start(): Promise<void> {
    logger.info(
      {
        dryRun: this.config.dryRun,
        cities: this.config.cities.length,
        logLevel: this.config.logLevel,
      },
      '🤖 Weather Arbitrage Bot starting...',
    );

    // Initialize CLOB client with Polymarket wallet config
    const clobClient = new CLOBTradingClient(
      this.config.polymarket,
      this.config.dryRun,
    );

    // Initialize CLOB auth (derive or load API keys)
    if (!this.config.dryRun) {
      await clobClient.initialize();
    }

    // Initialize order manager
    const orderManager = new OrderManager(clobClient, this.config.dryRun);

    // Initialize position sync
    this.positionSync = new PositionSyncService(
      clobClient,
      { getPendingTrades, updateTradeStatus, getUnsettledTrades, settleTradePnl },
      this.config.dryRun,
    );

    // Initialize pipeline
    this.pipeline = new TradingPipeline(
      orderManager,
      this.config.cities,
      this.config.dryRun,
    );

    // Start health check server
    this.healthCheck = new HealthCheckServer(this.config.dryRun);
    this.healthCheck.setAutoExecuteCityCallback((city, enabled) => {
      this.pipeline.setAutoExecuteCity(city, enabled);
      this.healthCheck.setAutoExecuteCities(this.pipeline.autoExecuteCities);
    });
    this.healthCheck.start(this.config.healthPort);

    this.running = true;

    // Sync positions, settle resolved markets, aggregate metrics on startup
    await this.positionSync.syncOnStartup();
    await this.positionSync.settleTrades();
    await aggregateDailyMetrics();

    // Run initial cycle
    logger.info('Running initial pipeline cycle...');
    await this.runCycle();

    // Start model polling on cron schedule
    this.stopModelPolling = startModelPolling(
      this.config.cities,
      async (_city: CityConfig, _forecasts: ModelForecasts) => {
        // On new model data, immediately re-evaluate
        if (this.running) {
          logger.info('New model data detected, running evaluation...');
          await this.runCycle();
        }
      },
    );

    // Start periodic full scan (every 2 minutes)
    const scanInterval = setInterval(async () => {
      if (this.running) {
        await this.runCycle();
      }
    }, 120_000);
    this.intervals.push(scanInterval);

    // Start position sync + settlement + metrics aggregation (every 60 seconds)
    const syncInterval = setInterval(async () => {
      if (this.running) {
        try {
          if (!this.config.dryRun) {
            await this.positionSync.syncPendingOrders();
          }
          await this.positionSync.settleTrades();
          await aggregateDailyMetrics();
        } catch (err) {
          logger.error({ err }, 'Position sync / settlement / metrics failed');
        }
      }
    }, 60_000);
    this.intervals.push(syncInterval);

    // Start market re-discovery (every 30 minutes)
    const discoveryInterval = setInterval(async () => {
      if (this.running) {
        try {
          logger.info('Refreshing market discovery...');
          const newMarkets = await scanWeatherMarkets();
          const grouped = groupMarketsByEvent(newMarkets);
          logger.info({ marketCount: grouped.size }, 'Markets refreshed');

          // Update WebSocket subscriptions
          if (this.wsFeed) {
            const tokenIds = [...grouped.values()].flatMap((m) =>
              m.buckets.flatMap((b) => [b.yesTokenId, b.noTokenId]),
            );
            this.wsFeed.updateSubscriptions(tokenIds);
          }
        } catch (err) {
          logger.error({ err }, 'Market refresh failed');
        }
      }
    }, 1_800_000);
    this.intervals.push(discoveryInterval);

    logger.info(
      {
        dryRun: this.config.dryRun,
        healthPort: this.config.healthPort,
      },
      '✅ Bot is running',
    );
  }

  private async runCycle(): Promise<void> {
    try {
      const result = await this.pipeline.runCycle();
      this.healthCheck.updateCycleMetrics(result);

      logger.info(
        {
          markets: result.marketsEvaluated,
          signals: result.signalsGenerated,
          orders: result.ordersPlaced,
          errors: result.errors.length,
        },
        'Cycle complete',
      );
    } catch (err) {
      logger.error({ err }, 'Cycle failed');
    }
  }

  /** Toggle auto-execute for a specific city at runtime. */
  setAutoExecuteCity(city: string, enabled: boolean): void {
    if (this.pipeline) {
      this.pipeline.setAutoExecuteCity(city, enabled);
    }
  }

  /** Get cities with auto-execute enabled. */
  get autoExecuteCities(): string[] {
    return this.pipeline?.autoExecuteCities ?? [];
  }

  /**
   * Stop the bot gracefully.
   */
  async stop(): Promise<void> {
    logger.info('Stopping bot...');
    this.running = false;

    // Stop intervals
    this.intervals.forEach(clearInterval);
    this.intervals = [];

    // Stop model polling
    if (this.stopModelPolling) {
      this.stopModelPolling();
    }

    // Disconnect WebSocket
    if (this.wsFeed) {
      this.wsFeed.disconnect();
    }

    // Stop health check
    this.healthCheck.stop();

    // Disconnect database
    await disconnectDb();

    logger.info('🛑 Bot stopped');
  }
}