MomentumLab/app/backend/src/services/dataSyncService.ts

import prisma from '../config/database';
import { cache } from '../config/redis';
import logger from '../utils/logger';
import { externalDataSourceService } from './externalDataSourceService';

export class DataSyncService {
  // 延迟辅助函数
  private delay(ms: number): Promise<void> {
    return new Promise((resolve) => setTimeout(resolve, ms));
  }

  // 同步实时行情（模拟数据）
  async syncRealTimeQuotes(): Promise<void> {
    try {
      logger.info('Starting real-time quotes sync...');
      
      const stocks = await prisma.stock.findMany();
      const now = new Date();
      
      for (const stock of stocks.slice(0, 100)) {
        try {
          const basePrice = 10 + Math.random() * 100;
          const changePercent = (Math.random() * 10 - 5);
          const price = basePrice * (1 + changePercent / 100);
          
          // 使用 create，因为每次时间都不一样
          await prisma.stockQuote.create({
            data: {
              stockCode: stock.code,
              price: price,
              open: basePrice * (1 + (Math.random() * 4 - 2) / 100),
              high: price * (1 + Math.random() * 0.02),
              low: price * (1 - Math.random() * 0.02),
              preClose: basePrice,
              volume: BigInt(Math.floor(Math.random() * 10000000)),
              turnover: BigInt(Math.floor(Math.random() * 100000000)),
              changePercent: changePercent,
              quoteTime: now,
            },
          });
        } catch (error) {
          logger.error(`Failed to sync quote for ${stock.code}:`, error);
        }
      }

      logger.info('Real-time quotes sync completed');
    } catch (error) {
      logger.error('Failed to sync real-time quotes:', error);
      throw error;
    }
  }

  // 同步版块行情（模拟数据）
  async syncSectorQuotes(): Promise<void> {
    try {
      logger.info('Starting sector quotes sync...');
      
      const sectors = await prisma.sector.findMany();
      const now = new Date();
      
      for (const sector of sectors) {
        try {
          const changePercent = Math.random() * 6 - 3;
          
          await prisma.sectorQuote.create({
            data: {
              sectorCode: sector.code,
              current: 1000 + Math.random() * 500,
              change: changePercent * 10,
              changePercent: changePercent,
              volume: BigInt(Math.floor(Math.random() * 1000000000)),
              turnover: BigInt(Math.floor(Math.random() * 10000000000)),
              quoteTime: now,
            },
          });
        } catch (error) {
          logger.error(`Failed to sync sector quote for ${sector.code}:`, error);
        }
      }

      logger.info('Sector quotes sync completed');
    } catch (error) {
      logger.error('Failed to sync sector quotes:', error);
      throw error;
    }
  }

  // 同步K线数据（单只股票）
  async syncKLineData(stockCode: string, period: string = 'day'): Promise<void> {
    try {
      logger.info(`Starting K-line sync for ${stockCode} (${period})...`);
      
      // 获取最新K线日期
      const latestKLine = await prisma.stockKLine.findFirst({
        where: { stockCode, period },
        orderBy: { date: 'desc' },
      });

      const startDate = latestKLine 
        ? new Date(latestKLine.date.getTime() + 24 * 60 * 60 * 1000)
        : new Date(Date.now() - 365 * 24 * 60 * 60 * 1000);
      
      const endDate = new Date();

      // 使用模拟数据
      const days = Math.min(100, Math.ceil((endDate.getTime() - startDate.getTime()) / (24 * 60 * 60 * 1000)));
      
      for (let i = days; i >= 0; i--) {
        const date = new Date(endDate);
        date.setDate(date.getDate() - i);
        
        const basePrice = 10 + Math.random() * 100;
        const change = (Math.random() * 0.1 - 0.05);
        const close = basePrice * (1 + change);
        const open = basePrice * (1 + (Math.random() * 0.04 - 0.02));
        const high = Math.max(open, close) * (1 + Math.random() * 0.02);
        const low = Math.min(open, close) * (1 - Math.random() * 0.02);

        try {
          await prisma.stockKLine.upsert({
            where: {
              stockCode_period_date: {
                stockCode,
                period,
                date,
              },
            },
            update: {
              open,
              high,
              low,
              close,
              volume: BigInt(Math.floor(Math.random() * 10000000)),
            },
            create: {
              stockCode,
              period,
              date,
              open,
              high,
              low,
              close,
              volume: BigInt(Math.floor(Math.random() * 10000000)),
            },
          });
        } catch (error) {
          logger.error(`Failed to upsert K-line for ${stockCode} on ${date}:`, error);
        }
      }

      logger.info(`K-line sync completed for ${stockCode}`);
    } catch (error) {
      logger.error(`Failed to sync K-line for ${stockCode}:`, error);
      throw error;
    }
  }

  // 批量同步所有股票K线数据
  async syncAllStocksKLine(period: string = 'day'): Promise<void> {
    try {
      logger.info(`Starting batch K-line sync for period: ${period}...`);
      
      const stocks = await prisma.stock.findMany({
        select: { code: true },
      });

      let successCount = 0;
      let failCount = 0;

      for (const stock of stocks) {
        try {
          await this.syncKLineData(stock.code, period);
          successCount++;
          
          await this.delay(100);
        } catch (error) {
          logger.error(`Failed to sync K-line for ${stock.code}:`, error);
          failCount++;
        }
      }

      logger.info(`Batch K-line sync completed: ${successCount} success, ${failCount} failed`);
    } catch (error) {
      logger.error('Failed to batch sync K-line data:', error);
      throw error;
    }
  }

  // 同步所有股票日行情数据（从外部数据源）
  async syncAllStocks(): Promise<void> {
    try {
      logger.info('[syncAllStocks] 开始同步所有股票日行情数据...');
      
      // 检查外部数据源是否可用
      if (!externalDataSourceService.isEnabled()) {
        throw new Error('数据源未配置，无法同步股票数据');
      }

      // 1. 获取本地数据库最新日期
      const latestQuote = await prisma.stockDailyQuote.findFirst({
        orderBy: { tradeDay: 'desc' },
        select: { tradeDay: true },
      });
      
      const startDate = latestQuote?.tradeDay 
        ? new Date(latestQuote.tradeDay.getTime() + 24 * 60 * 60 * 1000)
        : new Date(Date.now() - 30 * 24 * 60 * 60 * 1000); // 默认30天前
      
      logger.info(`[syncAllStocks] 本地最新日期: ${latestQuote?.tradeDay?.toISOString().split('T')[0] || '无'}, 开始日期: ${startDate.toISOString().split('T')[0]}`);

      // 2. 获取交易日历（返回交易日字符串数组 YYYYMMDD）
      const tradingDateList = await externalDataSourceService.getTradingDates(
        startDate.toISOString().split('T')[0].replace(/-/g, ''),
        new Date().toISOString().split('T')[0].replace(/-/g, '')
      );
      
      logger.info(`[syncAllStocks] 需要同步 ${tradingDateList.length} 个交易日: ${tradingDateList.join(', ')}`);

      if (tradingDateList.length === 0) {
        logger.info('[syncAllStocks] 没有需要同步的交易日');
        return;
      }

      // 3. 获取所有股票代码
      const symbols = await externalDataSourceService.getSymbols({ 
        type: 'stock',
        limit: 6000 
      });
      
      // 保留完整的 symbol 信息，用于后续调用 API
      const stockList = symbols
        .filter(s => s.symbol_id && s.status === 'active') // 只保留活跃股票
        .map(s => ({
          code: s.symbol_id?.replace(/\.(SZ|SH|BJ)$/i, '') || '', // 纯代码，用于数据库
          symbol_id: s.symbol_id || '', // 完整代码（带后缀），用于 API 调用
          name: s.name || '',
          exchange: s.exchange || '',
        }))
        .filter(s => s.code && s.symbol_id); // 过滤掉无效数据
      
      logger.info(`[syncAllStocks] 获取到 ${stockList.length} 只活跃股票，样例: ${stockList.slice(0, 5).map(s => s.symbol_id).join(', ')}`);

      // 4. 按日期遍历，批量获取并保存数据
      let totalSyncedDays = 0;
      let totalSyncedRecords = 0;

      for (const tradeDate of tradingDateList) {
        try {
          logger.info(`[syncAllStocks] 同步 ${tradeDate} 的数据...`);
          
          // 格式化日期 YYYY-MM-DD
          const formattedDate = `${tradeDate.slice(0, 4)}-${tradeDate.slice(4, 6)}-${tradeDate.slice(6, 8)}`;
          
          // 批量获取该日所有股票的行情数据
          // 分批处理，每批10只股票
          const batchSize = 10;
          let daySyncedCount = 0;
          
          for (let i = 0; i < stockList.length; i += batchSize) {
            const batch = stockList.slice(i, i + batchSize);
            
            try {
              // 调用外部数据源批量获取日行情
              const quotes = await this.fetchDailyQuotes(batch, formattedDate);
              
              // 批量保存到数据库
              await this.saveDailyQuotes(quotes, formattedDate);
              
              daySyncedCount += quotes.length;
              totalSyncedRecords += quotes.length;
              
              // 避免请求过快
              await this.delay(200);
            } catch (error) {
              logger.error(`[syncAllStocks] 批量获取 ${formattedDate} 第 ${i/batchSize + 1} 批数据失败:`, error);
            }
          }
          
          totalSyncedDays++;
          logger.info(`[syncAllStocks] ${formattedDate} 同步完成: ${daySyncedCount} 条记录`);
          
        } catch (error) {
          logger.error(`[syncAllStocks] 同步 ${tradeDate} 失败:`, error);
        }
      }

      logger.info(`[syncAllStocks] 同步完成: ${totalSyncedDays} 天, ${totalSyncedRecords} 条记录`);
    } catch (error) {
      logger.error('[syncAllStocks] 同步失败:', error);
      throw new Error('数据源异常，股票同步失败: ' + (error as Error).message);
    }
  }

  // 批量获取日行情数据（辅助方法）
  private async fetchDailyQuotes(
    stocks: { code: string; symbol_id: string; name: string; exchange: string }[],
    tradeDate: string
  ): Promise<any[]> {
    // 使用 K 线接口获取日线数据
    const quotes: any[] = [];
    
    for (const stock of stocks) {
      try {
        // 使用完整的 symbol_id（带后缀，如 000001.SZ）调用 API
        const klines = await externalDataSourceService.getKLines(stock.symbol_id, 'day', {
          startDate: tradeDate.replace(/-/g, ''),
          endDate: tradeDate.replace(/-/g, ''),
          limit: 1,
        });
        
        if (klines && klines.length > 0) {
          const k = klines[0];
          // 使用纯代码（不带后缀）保存到数据库
          quotes.push({
            stockCode: stock.code,
            stockName: stock.name,
            open: k.open,
            close: k.close,
            high: k.high,
            low: k.low,
            volume: k.volume,
            amount: k.amount || 0,
            differrange: ((k.close - k.open) / k.open) * 100,
          });
        }
      } catch (error) {
        logger.warn(`[fetchDailyQuotes] 获取 ${stock.symbol_id} ${tradeDate} 数据失败`);
      }
    }
    
    return quotes;
  }

  // 批量保存日行情数据（辅助方法）
  private async saveDailyQuotes(quotes: any[], tradeDate: string): Promise<void> {
    const date = new Date(tradeDate);
    
    for (const quote of quotes) {
      try {
        await prisma.stockDailyQuote.upsert({
          where: {
            stockCode_tradeDay: {
              stockCode: quote.stockCode,
              tradeDay: date,
            },
          },
          update: {
            open: quote.open,
            close: quote.close,
            high: quote.high,
            low: quote.low,
            volume: BigInt(quote.volume || 0),
            amount: BigInt(quote.amount || 0),
            differrange: quote.differrange,
          },
          create: {
            stockCode: quote.stockCode,
            tradeDay: date,
            open: quote.open,
            close: quote.close,
            high: quote.high,
            low: quote.low,
            volume: BigInt(quote.volume || 0),
            amount: BigInt(quote.amount || 0),
            differrange: quote.differrange,
          },
        });
      } catch (error) {
        logger.error(`[saveDailyQuotes] 保存 ${quote.stockCode} 失败:`, error);
      }
    }
  }

  // 同步版块数据
  async syncSectors(): Promise<void> {
    try {
      logger.info('Starting sync sectors...');
      
      const sectors = [
        { name: '半导体', code: '880491' },
        { name: '新能源', code: '880952' },
        { name: '医药生物', code: '880122' },
        { name: '白酒', code: '880952' },
        { name: '银行', code: '880471' },
        { name: '证券', code: '880472' },
        { name: '保险', code: '880473' },
        { name: '房地产', code: '880482' },
        { name: '汽车', code: '880391' },
        { name: '电子', code: '880494' },
      ];

      for (const sector of sectors) {
        try {
          await prisma.sector.upsert({
            where: { code: sector.code },
            update: { name: sector.name },
            create: sector,
          });
        } catch (error) {
          logger.error(`Failed to upsert sector ${sector.code}:`, error);
        }
      }

      logger.info(`Synced ${sectors.length} sectors`);
    } catch (error) {
      logger.error('Failed to sync sectors:', error);
      throw error;
    }
  }

  // 计算动量分数
  async calculateMomentumScores(): Promise<void> {
    try {
      logger.info('Starting calculate momentum scores...');
      
      const stocks = await prisma.stock.findMany({
        include: {
          klines: {
            where: { period: 'day' },
            orderBy: { date: 'desc' },
            take: 20,
          },
        },
      });

      for (const stock of stocks) {
        try {
          if (stock.klines.length < 5) continue;
          
          const prices = stock.klines.map(k => k.close).reverse();
          const ma5 = prices.slice(-5).reduce((a, b) => a + b, 0) / 5;
          const ma10 = prices.slice(-10).reduce((a, b) => a + b, 0) / 10;
          const ma20 = prices.slice(-20).reduce((a, b) => a + b, 0) / 20;
          
          const currentPrice = prices[prices.length - 1];
          const priceChange = (currentPrice - prices[0]) / prices[0] * 100;
          
          let score = 50;
          if (currentPrice > ma5) score += 10;
          if (currentPrice > ma10) score += 10;
          if (currentPrice > ma20) score += 10;
          if (priceChange > 5) score += 10;
          if (priceChange > 10) score += 10;
          
          const today = new Date();
          today.setHours(0, 0, 0, 0);
          
          await prisma.momentumStock.upsert({
            where: {
              stockCode_date: {
                stockCode: stock.code,
                date: today,
              },
            },
            update: {
              momentumScore: score,
            },
            create: {
              stockCode: stock.code,
              momentumScore: score,
              volumeRatio: 1 + Math.random() * 2,
              breakThrough: score > 70,
              date: today,
            },
          });
        } catch (error) {
          logger.error(`Failed to calculate momentum for ${stock.code}:`, error);
        }
      }

      logger.info('Momentum scores calculation completed');
    } catch (error) {
      logger.error('Failed to calculate momentum scores:', error);
      throw error;
    }
  }

  // 同步市场指数
  async syncMarketIndices(): Promise<void> {
    try {
      logger.info('Starting sync market indices...');
      
      const indices = [
        { name: '上证指数', code: '000001' },
        { name: '深证成指', code: '399001' },
        { name: '创业板指', code: '399006' },
      ];

      for (const index of indices) {
        try {
          await prisma.marketIndex.upsert({
            where: { code: index.code },
            update: {
              current: 3000 + Math.random() * 500,
              change: Math.random() * 40 - 20,
              changePercent: Math.random() * 2 - 1,
              volume: BigInt(Math.floor(Math.random() * 1000000000)),
              turnover: BigInt(Math.floor(Math.random() * 10000000000)),
            },
            create: {
              name: index.name,
              code: index.code,
              current: 3000 + Math.random() * 500,
              change: Math.random() * 40 - 20,
              changePercent: Math.random() * 2 - 1,
              volume: BigInt(Math.floor(Math.random() * 1000000000)),
              turnover: BigInt(Math.floor(Math.random() * 10000000000)),
            },
          });
        } catch (error) {
          logger.error(`Failed to upsert market index ${index.code}:`, error);
        }
      }

      logger.info('Market indices sync completed');
    } catch (error) {
      logger.error('Failed to sync market indices:', error);
      throw error;
    }
  }

  // 初始化基础数据
  async initBaseData(): Promise<void> {
    try {
      logger.info('Initializing base data...');
      
      // 检查外部数据源是否可用
      if (externalDataSourceService.isEnabled()) {
        try {
          // 同步股票
          await this.syncAllStocks();
        } catch (error) {
          logger.warn('Stock sync failed during init, will retry later:', error);
        }
      } else {
        logger.warn('External data source not enabled, skipping stock sync during init');
      }
      
      // 同步版块
      await this.syncSectors();
      
      // 同步市场指数
      await this.syncMarketIndices();

      logger.info('Base data initialization completed');
    } catch (error) {
      logger.error('Failed to initialize base data:', error);
    }
  }
}

export const dataSyncService = new DataSyncService();