market-data-service/services/priceUpdateService.js

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
};