/** * 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 { 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[] = []; constructor(config?: BotConfig) { this.config = config || loadConfig(); } /** * Start the bot. */ async start(): Promise { 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 { 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 { 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'); } }