const { logger } = require('../utils/logger'); const { executeQuery } = require('../config/database'); const { broadcastPriceUpdate, broadcastMultiplePriceUpdates } = require('./websocketService'); const { addTick } = require('./candlestickService'); // In-memory cache for current prices const priceCache = new Map(); const symbolSubscriptions = new Map(); // Simulated price data for development const generateMockPriceData = (symbol) => { const basePrice = priceCache.get(symbol)?.close || 1.0; const change = (Math.random() - 0.5) * 0.02; // ±1% change const newPrice = basePrice * (1 + change); return { symbol, open: basePrice, high: Math.max(basePrice, newPrice) * (1 + Math.random() * 0.001), low: Math.min(basePrice, newPrice) * (1 - Math.random() * 0.001), close: newPrice, volume: Math.floor(Math.random() * 1000000) + 100000, timestamp: new Date(), timeframe: 'H1' }; }; class PriceUpdateService { constructor() { this.isRunning = false; this.updateInterval = null; this.supportedSymbols = []; this.updateCallbacks = []; this.initialize(); } initialize() { // Load supported symbols from environment this.supportedSymbols = (process.env.SUPPORTED_SYMBOLS || 'EURUSD,GBPUSD,USDJPY') .split(',') .map(symbol => symbol.trim()); logger.info(`Price update service initialized with symbols: ${this.supportedSymbols.join(', ')}`); // Initialize price cache with mock data this.initializePriceCache(); } initializePriceCache() { this.supportedSymbols.forEach(symbol => { priceCache.set(symbol, generateMockPriceData(symbol)); }); logger.info('Price cache initialized'); } start() { if (this.isRunning) { logger.warn('Price update service is already running'); return; } const interval = parseInt(process.env.PRICE_UPDATE_INTERVAL) || 1000; this.updateInterval = setInterval(() => { this.updatePrices(); }, interval); this.isRunning = true; logger.info(`Price update service started with ${interval}ms interval`); // Start cleanup interval setInterval(() => { this.cleanup(); }, 60000); // Cleanup every minute } stop() { if (!this.isRunning) { return; } if (this.updateInterval) { clearInterval(this.updateInterval); this.updateInterval = null; } this.isRunning = false; logger.info('Price update service stopped'); } async updatePrices() { try { const updates = []; for (const symbol of this.supportedSymbols) { const priceData = await this.generatePriceUpdate(symbol); updates.push(priceData); // Update cache priceCache.set(symbol, priceData); // Store in database (if needed) await this.storePriceData(priceData); } // Broadcast updates to WebSocket clients if (updates.length > 0) { broadcastMultiplePriceUpdates(updates); } // Call registered update callbacks this.updateCallbacks.forEach(callback => { try { callback(updates); } catch (error) { logger.error('Error in price update callback:', error); } }); } catch (error) { logger.error('Error updating prices:', error); } } async generatePriceUpdate(symbol) { // In a real implementation, this would fetch from external APIs // For now, we'll generate mock data const currentPrice = priceCache.get(symbol); const change = (Math.random() - 0.5) * 0.02; // ±1% change const newPrice = currentPrice ? currentPrice.close * (1 + change) : 1.0; const priceData = { symbol, open: currentPrice ? currentPrice.close : newPrice, high: Math.max(currentPrice ? currentPrice.close : newPrice, newPrice) * (1 + Math.random() * 0.001), low: Math.min(currentPrice ? currentPrice.close : newPrice, newPrice) * (1 - Math.random() * 0.001), close: newPrice, volume: Math.floor(Math.random() * 1000000) + 100000, timestamp: new Date(), timeframe: 'H1' }; // Add tick data for candlestick aggregation // Generate multiple ticks within this price update period const tickCount = Math.floor(Math.random() * 5) + 3; // 3-7 ticks per update for (let i = 0; i < tickCount; i++) { const tickPrice = newPrice + (Math.random() - 0.5) * 0.001; // Small variations const tickTimestamp = new Date(priceData.timestamp.getTime() - (tickCount - i) * 100); // Spread over the interval addTick(symbol, tickPrice, tickTimestamp); } return priceData; } async storePriceData(priceData) { try { const dbType = process.env.DB_TYPE || 'sqlite'; if (dbType === 'sqlite') { await this.storePriceDataSQLite(priceData); } else if (dbType === 'mysql') { await this.storePriceDataMySQL(priceData); } else if (dbType === 'postgres') { await this.storePriceDataPostgreSQL(priceData); } } catch (error) { logger.error('Error storing price data:', error); } } async storePriceDataSQLite(priceData) { const query = ` INSERT OR IGNORE INTO price_data (symbol, timeframe, timestamp, open, high, low, close, volume) VALUES (?, ?, ?, ?, ?, ?, ?, ?) `; await executeQuery(query, [ priceData.symbol, priceData.timeframe, priceData.timestamp, priceData.open, priceData.high, priceData.low, priceData.close, priceData.volume ]); } async storePriceDataMySQL(priceData) { const query = ` INSERT IGNORE INTO price_data (symbol, timeframe, timestamp, open, high, low, close, volume) VALUES (?, ?, ?, ?, ?, ?, ?, ?) `; await executeQuery(query, [ priceData.symbol, priceData.timeframe, priceData.timestamp, priceData.open, priceData.high, priceData.low, priceData.close, priceData.volume ]); } async storePriceDataPostgreSQL(priceData) { const query = ` INSERT INTO price_data (symbol, timeframe, timestamp, open, high, low, close, volume) VALUES (?, ?, ?, ?, ?, ?, ?, ?) ON CONFLICT DO NOTHING `; await executeQuery(query, [ priceData.symbol, priceData.timeframe, priceData.timestamp, priceData.open, priceData.high, priceData.low, priceData.close, priceData.volume ]); } // Get current price for a symbol getCurrentPrice(symbol) { return priceCache.get(symbol); } // Get current prices for multiple symbols getCurrentPrices(symbols = null) { const targetSymbols = symbols || this.supportedSymbols; const prices = {}; targetSymbols.forEach(symbol => { prices[symbol] = priceCache.get(symbol); }); return prices; } // Get historical data for a symbol async getHistoricalData(symbol, timeframe = 'H1', limit = 1000) { try { const dbType = process.env.DB_TYPE || 'sqlite'; let query; if (dbType === 'sqlite') { query = ` SELECT * FROM price_data WHERE symbol = ? AND timeframe = ? ORDER BY timestamp DESC LIMIT ? `; } else { query = ` SELECT * FROM price_data WHERE symbol = ? AND timeframe = ? ORDER BY timestamp DESC LIMIT ? `; } const params = [symbol, timeframe, limit]; const data = await executeQuery(query, params); return data.reverse(); // Return in chronological order } catch (error) { logger.error(`Error fetching historical data for ${symbol}:`, error); return []; } } // Register callback for price updates onPriceUpdate(callback) { this.updateCallbacks.push(callback); } // Remove price update callback removePriceUpdateCallback(callback) { const index = this.updateCallbacks.indexOf(callback); if (index > -1) { this.updateCallbacks.splice(index, 1); } } // Cleanup old data async cleanup() { try { const cutoffDate = new Date(); cutoffDate.setDate(cutoffDate.getDate() - 30); // Keep 30 days of data const dbType = process.env.DB_TYPE || 'sqlite'; let query; if (dbType === 'sqlite') { query = 'DELETE FROM price_data WHERE timestamp < ?'; } else { query = 'DELETE FROM price_data WHERE timestamp < ?'; } const params = [cutoffDate]; await executeQuery(query, params); logger.info('Price data cleanup completed'); } catch (error) { logger.error('Error during price data cleanup:', error); } } // Get service statistics getStats() { return { isRunning: this.isRunning, supportedSymbols: this.supportedSymbols, cachedSymbols: Array.from(priceCache.keys()), updateCallbacks: this.updateCallbacks.length, cacheSize: priceCache.size, uptime: process.uptime() }; } } // Create singleton instance const priceUpdateService = new PriceUpdateService(); // Export functions for external use const startPriceUpdates = (io) => { priceUpdateService.start(); }; const stopPriceUpdates = () => { priceUpdateService.stop(); }; const getCurrentPrice = (symbol) => { return priceUpdateService.getCurrentPrice(symbol); }; const getCurrentPrices = (symbols) => { return priceUpdateService.getCurrentPrices(symbols); }; const getHistoricalData = (symbol, timeframe, limit) => { return priceUpdateService.getHistoricalData(symbol, timeframe, limit); }; const onPriceUpdate = (callback) => { priceUpdateService.onPriceUpdate(callback); }; const getPriceUpdateStats = () => { return priceUpdateService.getStats(); }; module.exports = { startPriceUpdates, stopPriceUpdates, getCurrentPrice, getCurrentPrices, getHistoricalData, onPriceUpdate, getPriceUpdateStats, priceUpdateService };