AlphaFuturesPro/backend/src/services/datasource/TQDataSource.ts

// TQSDK数据源实现
import { DataSource } from './DataSource';
import TQSDK from 'tqsdk';
import WebSocket from 'ws';

export class TQDataSource implements DataSource {
  private tq: any = null;
  private initialized: boolean = false;
  private config: {
    username?: string;
    password?: string;
    timeout?: number;
    retries?: number;
    maxConnections?: number;
  };

  constructor(config: any = {}) {
    console.log('使用TQSDK数据源初始化...');
    this.config = {
      username: config.username || '',
      password: config.password || '',
      timeout: config.timeout || 30000,
      retries: config.retries || 3,
      maxConnections: config.maxConnections || 5
    };
    console.log('TQSDK数据源配置:', this.config);
  }

  async initialize(): Promise<boolean> {
    try {
      console.log('开始初始化TQSDK数据源...');
      
      // 创建TQSDK实例，使用配置的参数
      console.log('创建TQSDK实例...');
      
      // 初始化TQSDK，传入WebSocket对象
      this.tq = new TQSDK({ autoInit: true }, { WebSocket });
      
      console.log('TQSDK实例创建成功，等待就绪...');
      
      // 等待初始化完成，设置超时
      const timeout = this.config.timeout || 10000;
      console.log('等待TQSDK连接，超时时间:', timeout, 'ms');
      
      await new Promise((resolve, reject) => {
        // 设置超时
        const timeoutId = setTimeout(() => {
          console.error('TQSDK连接超时');
          reject(new Error('TQSDK连接超时'));
        }, timeout);
        
        // 监听就绪事件
        this.tq?.on('ready', () => {
          console.log('TQSDK就绪');
          clearTimeout(timeoutId);
          resolve(true);
        });
        
        // 监听错误事件
        this.tq?.on('error', (error: any) => {
          console.error('TQSDK错误:', error);
          clearTimeout(timeoutId);
          reject(error);
        });
      });
      
      this.initialized = true;
      console.log('TQSDK数据源初始化成功');
      return true;
    } catch (error) {
      console.error('TQSDK数据源初始化失败:', error);
      // 清理资源
      if (this.tq) {
        try {
          if (typeof this.tq.close === 'function') {
            this.tq.close();
          } else {
            console.log('TqApi实例没有close方法');
          }
        } catch (closeError) {
          console.error('关闭TQSDK连接失败:', closeError);
        }
        this.tq = null;
      }
      this.initialized = false;
      return false;
    }
  }

  async getContractList(): Promise<any[]> {
    if (!this.initialized || !this.tq) {
      throw new Error('TQSDK数据源未初始化');
    }

    try {
      // 使用TQSDK的getQuotesByInput方法获取合约列表
      // 这里使用空字符串搜索，获取所有合约
      const contracts = this.tq.getQuotesByInput('', {
        future: true,
        future_index: true,
        future_cont: true,
        option: false,
        combine: false
      });
      
      // 转换合约格式
      const futuresContracts = contracts.map((symbol: string) => {
        try {
          const quote = this.tq.getQuote(symbol);
          return {
            symbol: symbol,
            name: quote.instrument_name,
            exchange: symbol.split('.')[0]
          };
        } catch (error) {
          console.error(`获取合约${symbol}信息失败:`, error);
          return null;
        }
      }).filter((contract: any) => contract !== null);
      
      return futuresContracts;
    } catch (error) {
      console.error('获取合约列表失败:', error);
      throw error;
    }
  }

  async getContractDetail(symbol: string): Promise<any> {
    if (!this.initialized || !this.tq) {
      throw new Error('TQSDK数据源未初始化');
    }

    try {
      // 获取合约详情
      const quote = this.tq.getQuote(symbol);
      return {
        symbol: symbol,
        name: quote.instrument_name,
        exchange: symbol.split('.')[0],
        product_id: quote.product_id,
        price_tick: quote.price_tick,
        volume_multiple: quote.volume_multiple,
        margin_rate: quote.margin_rate,
        expire_datetime: quote.expire_datetime
      };
    } catch (error) {
      console.error(`获取合约${symbol}详情失败:`, error);
      throw error;
    }
  }

  async getKlineData(symbol: string, period: string, count: number): Promise<any[]> {
    if (!this.initialized || !this.tq) {
      throw new Error('TQSDK数据源未初始化');
    }

    try {
      // 转换周期格式
      const duration = this.convertPeriodToDuration(period);
      // 获取K线数据
      const klines = this.tq.getKlines(symbol, duration);
      return klines.data || [];
    } catch (error) {
      console.error(`获取合约${symbol}K线数据失败:`, error);
      throw error;
    }
  }

  async getTickData(symbol: string): Promise<any> {
    if (!this.initialized || !this.tq) {
      throw new Error('TQSDK数据源未初始化');
    }

    try {
      // 获取实时行情数据
      const quote = this.tq.getQuote(symbol);
      return {
        last_price: quote.last_price,
        price_change: quote.last_price - quote.pre_settlement,
        pre_close: quote.pre_close,
        open: quote.open,
        high: quote.high,
        low: quote.low,
        volume: quote.volume,
        open_interest: quote.open_interest,
        bid_price1: quote.bid_price1,
        bid_volume1: quote.bid_volume1,
        ask_price1: quote.ask_price1,
        ask_volume1: quote.ask_volume1,
        datetime: quote.datetime
      };
    } catch (error) {
      console.error(`获取合约${symbol}实时行情数据失败:`, error);
      throw error;
    }
  }

  async getMarketOverview(): Promise<any[]> {
    if (!this.initialized || !this.tq) {
      throw new Error('TQSDK数据源未初始化');
    }

    try {
      // 获取所有合约
      const contracts = await this.getContractList();
      // 限制获取的合约数量，避免请求过多
      const limitedContracts = contracts.slice(0, 20);
      // 获取每个合约的实时行情
      const overview = [];
      for (const contract of limitedContracts) {
        try {
          const tick = await this.getTickData(contract.symbol);
          overview.push({
            symbol: contract.symbol,
            name: contract.name,
            price: tick.last_price,
            change: tick.price_change,
            change_percent: tick.price_change / tick.pre_close * 100,
            volume: tick.volume,
            open_interest: tick.open_interest
          });
        } catch (error) {
          console.error(`获取合约${contract.symbol}行情失败:`, error);
        }
      }
      return overview;
    } catch (error) {
      console.error('获取市场概览失败:', error);
      throw error;
    }
  }

  async getHistoricalTrades(symbol: string, start: number, end: number): Promise<any[]> {
    if (!this.initialized || !this.tq) {
      throw new Error('TQSDK数据源未初始化');
    }

    try {
      // TQSDK Node.js版本可能不支持直接获取历史成交数据
      // 这里返回空数组，实际使用时需要根据TQSDK文档调整
      console.warn('TQSDK Node.js版本可能不支持直接获取历史成交数据');
      return [];
    } catch (error) {
      console.error(`获取合约${symbol}历史成交数据失败:`, error);
      throw error;
    }
  }

  async close(): Promise<void> {
    if (this.tq) {
      try {
        // TQSDK Node.js版本可能没有close方法
        // 这里尝试关闭websocket连接
        if (this.tq.quotesWs && typeof this.tq.quotesWs.close === 'function') {
          this.tq.quotesWs.close();
          console.log('TQSDK行情WebSocket连接已关闭');
        }
        console.log('TQSDK连接已关闭');
      } catch (error) {
        console.error('关闭TQSDK连接失败:', error);
      }
      this.tq = null;
      this.initialized = false;
      console.log('TQSDK数据源已关闭');
    }
  }

  // 转换周期格式为TQSDK所需的纳秒格式
  private convertPeriodToDuration(period: string): number {
    switch (period) {
      case '1M':
        return 60 * 1e9; // 1分钟
      case '5M':
        return 5 * 60 * 1e9; // 5分钟
      case '15M':
        return 15 * 60 * 1e9; // 15分钟
      case '30M':
        return 30 * 60 * 1e9; // 30分钟
      case '1H':
        return 60 * 60 * 1e9; // 1小时
      case '4H':
        return 4 * 60 * 60 * 1e9; // 4小时
      case '1D':
        return 24 * 60 * 60 * 1e9; // 1天
      case '1W':
        return 7 * 24 * 60 * 60 * 1e9; // 1周
      default:
        return 60 * 1e9; // 默认1分钟
    }
  }
}