AlphaFuturesPro/backend/python_service/tqapi_service.py

import asyncio
import datetime
from tqsdk import TqApi, TqAuth
from typing import List, Dict, Any

class TqApiService:
    # 类级缓存，存储用户名密码对应的TqApiService实例
    _instance_cache = {}
    
    def __init__(self):
        self.api = None
        self.connected = False
        self.username = None
        self.password = None
    
    @classmethod
    async def get_instance(cls, username: str, password: str):
        """获取或创建TqApiService实例"""
        # 生成缓存键
        cache_key = f"{username}:{password}"
        
        # 检查缓存中是否存在实例
        if cache_key in cls._instance_cache:
            instance = cls._instance_cache[cache_key]
            # 检查实例是否仍然连接
            if instance.connected:
                print(f"使用缓存的TqApiService实例 for {username}")
                return instance
            else:
                # 实例已断开连接，从缓存中移除
                print(f"TqApiService实例 for {username} 已断开连接，从缓存中移除")
                del cls._instance_cache[cache_key]
        
        # 创建新实例
        print(f"创建新的TqApiService实例 for {username}")
        instance = cls()
        instance.username = username
        instance.password = password
        # 连接到天勤服务器
        await instance.connect(username, password)
        # 存入缓存
        cls._instance_cache[cache_key] = instance
        return instance
    
    async def connect(self, username: str, password: str) -> bool:
        """连接到天勤服务器"""
        try:
            # 立即返回，不等待TqApi实例完全连接到天勤服务器
            # 连接过程将在后台进行
            self.username = username
            self.password = password
            self.connected = True
            # 创建后台任务来初始化API实例和运行事件循环
            asyncio.create_task(self._initialize_api())
            return True
        except Exception as e:
            print(f"连接失败: {e}")
            self.connected = False
            return False
    
    async def _initialize_api(self):
        """初始化TqApi实例并运行事件循环"""
        try:
            print("正在初始化TqApi实例...")
            # 创建TqApi实例
            self.api = TqApi(auth=TqAuth(self.username, self.password), debug=False)
            print("TqApi实例初始化完成")
            # 运行事件循环
            await self._run_api()
        except Exception as e:
            print(f"初始化TqApi实例失败: {e}")
            self.connected = False
    
    async def _run_api(self):
        """运行TQAPI事件循环"""
        try:
            while self.connected and self.api:
                try:
                    # 非阻塞方式检查更新
                    self.api.wait_update(0.1)  # 超时0.1秒，避免完全阻塞
                except Exception as e:
                    print(f"API更新出错: {e}")
                    await asyncio.sleep(0.1)
                # 让出CPU时间，避免阻塞其他任务
                await asyncio.sleep(0.01)
        except Exception as e:
            print(f"API运行出错: {e}")
            self.connected = False
    
    async def get_contracts(self) -> List[Dict[str, Any]]:
        """获取合约列表"""
        if not self.connected:
            raise Exception("未连接到天勤服务器")
        
        if not self.api:
            # API实例尚未初始化，返回空列表
            print("API实例尚未初始化，返回空合约列表")
            return []
        
        try:
            # 获取所有合约
            contracts = self.api.get_contracts()
            # 过滤期货合约，排除期权等其他类型
            futures_contracts = []
            for symbol, contract in contracts.items():
                # 只保留期货合约
                if contract["product_class"] == "FUTURE":
                    futures_contracts.append({
                        "symbol": symbol,
                        "name": contract["name"],
                        "exchange": contract["exchange"],
                        "product_id": contract["product_id"],
                        "price_tick": contract["price_tick"],
                        "volume_multiple": contract["volume_multiple"],
                        "margin_rate": contract["margin_rate"],
                        "expire_datetime": contract["expire_datetime"]
                    })
            return futures_contracts
        except Exception as e:
            print(f"获取合约列表失败: {e}")
            # 返回空列表，避免整个请求失败
            return []
    
    async def get_contract(self, symbol: str) -> Dict[str, Any]:
        """获取合约详情"""
        if not self.connected:
            raise Exception("未连接到天勤服务器")
        
        if not self.api:
            # API实例尚未初始化，返回默认值
            print("API实例尚未初始化，返回默认合约详情")
            return {
                "symbol": symbol,
                "name": "",
                "exchange": "",
                "product_id": "",
                "price_tick": 0,
                "volume_multiple": 0,
                "margin_rate": 0,
                "expire_datetime": 0
            }
        
        try:
            # 获取合约详情
            contract = self.api.get_contract(symbol)
            return {
                "symbol": symbol,
                "name": contract["name"],
                "exchange": contract["exchange"],
                "product_id": contract["product_id"],
                "price_tick": contract["price_tick"],
                "volume_multiple": contract["volume_multiple"],
                "margin_rate": contract["margin_rate"],
                "expire_datetime": contract["expire_datetime"]
            }
        except Exception as e:
            print(f"获取合约详情失败: {e}")
            # 返回默认值，避免整个请求失败
            return {
                "symbol": symbol,
                "name": "",
                "exchange": "",
                "product_id": "",
                "price_tick": 0,
                "volume_multiple": 0,
                "margin_rate": 0,
                "expire_datetime": 0
            }
    
    async def get_klines(self, symbol: str, period: str, count: int) -> List[Dict[str, Any]]:
        """获取K线数据"""
        if not self.connected:
            raise Exception("未连接到天勤服务器")
        
        if not self.api:
            # API实例尚未初始化，返回空列表
            print("API实例尚未初始化，返回空K线数据列表")
            return []
        
        try:
            # 转换周期格式
            duration_seconds = self._convert_period(period)
            # 获取K线数据
            klines = self.api.get_kline_serial(symbol, duration_seconds, data_length=count)
            # 转换为前端需要的格式
            result = []
            for i in range(len(klines)):
                result.append({
                    "time": int(klines.iloc[i]["datetime"].timestamp()),
                    "open": float(klines.iloc[i]["open"]),
                    "high": float(klines.iloc[i]["high"]),
                    "low": float(klines.iloc[i]["low"]),
                    "close": float(klines.iloc[i]["close"]),
                    "volume": float(klines.iloc[i]["volume"])
                })
            return result
        except Exception as e:
            print(f"获取K线数据失败: {e}")
            # 返回空列表，避免整个请求失败
            return []
    
    async def get_tick(self, symbol: str) -> Dict[str, Any]:
        """获取tick数据"""
        print(f"[TqApiService] 开始获取 tick 数据，symbol: {symbol}")
        try:
            print(f"[TqApiService] 检查连接状态: {self.connected}")
            if not self.connected:
                # 未连接到天勤服务器，返回默认数据
                print("[TqApiService] 未连接到天勤服务器，返回默认tick数据")
                return {
                  "last_price": 0,
                  "pre_close": 0,
                  "open": 0,
                  "high": 0,
                  "low": 0,
                  "volume": 0,
                  "open_interest": 0,
                  "bid_price1": 0,
                  "bid_volume1": 0,
                  "ask_price1": 0,
                  "ask_volume1": 0,
                  "datetime": int(datetime.datetime.now().timestamp())
                }
            
            print(f"[TqApiService] 检查API实例状态: {self.api is not None}")
            if not self.api:
                # API实例尚未初始化，返回默认数据
                print("[TqApiService] API实例尚未初始化，返回默认tick数据")
                return {
                  "last_price": 0,
                  "pre_close": 0,
                  "open": 0,
                  "high": 0,
                  "low": 0,
                  "volume": 0,
                  "open_interest": 0,
                  "bid_price1": 0,
                  "bid_volume1": 0,
                  "ask_price1": 0,
                  "ask_volume1": 0,
                  "datetime": int(datetime.datetime.now().timestamp())
                }
            
            # 获取实时行情
            print(f"[TqApiService] 正在调用 self.api.get_quote({symbol})...")
            quote = self.api.get_quote(symbol)
            print(f"[TqApiService] 获取 quote 对象成功")
            
            # 尝试多次获取行情更新，避免单次超时
            max_attempts = 3
            print(f"[TqApiService] 开始尝试获取行情更新，最多 {max_attempts} 次")
            for attempt in range(max_attempts):
                try:
                    # 等待行情更新，设置超时
                    print(f"[TqApiService] 尝试 {attempt + 1}/{max_attempts}: 调用 self.api.wait_update(1.0)...")
                    updated = self.api.wait_update(1.0)  # 1秒超时
                    print(f"[TqApiService] 尝试 {attempt + 1}/{max_attempts}: wait_update 返回: {updated}")
                    if updated:
                        print(f"[TqApiService] 获取行情更新成功")
                        break
                    print(f"[TqApiService] 尝试 {attempt + 1}/{max_attempts}: 获取 {symbol} 的行情信息超时")
                except Exception as e:
                    print(f"[TqApiService] 尝试 {attempt + 1}/{max_attempts}: 获取 {symbol} 的行情信息出错: {e}")
                    await asyncio.sleep(0.5)
            
            # 即使没有更新，也尝试返回当前数据
            print(f"[TqApiService] 正在构建返回数据...")
            result = {
              "last_price": float(quote.get("last_price", 0)),
              "pre_close": float(quote.get("pre_close", 0)),
              "open": float(quote.get("open", 0)),
              "high": float(quote.get("high", 0)),
              "low": float(quote.get("low", 0)),
              "volume": float(quote.get("volume", 0)),
              "open_interest": float(quote.get("open_interest", 0)),
              "bid_price1": float(quote.get("bid_price1", 0)),
              "bid_volume1": float(quote.get("bid_volume1", 0)),
              "ask_price1": float(quote.get("ask_price1", 0)),
              "ask_volume1": float(quote.get("ask_volume1", 0)),
              "datetime": int(quote.get("datetime", datetime.datetime.now()).timestamp())
            }
            print(f"[TqApiService] 获取 tick 数据成功，返回: {result}")
            return result
        except Exception as e:
            print(f"[TqApiService] 获取tick数据失败: {e}")
            import traceback
            traceback.print_exc()
            # 返回默认数据，避免整个请求失败
            default_result = {
              "last_price": 0,
              "pre_close": 0,
              "open": 0,
              "high": 0,
              "low": 0,
              "volume": 0,
              "open_interest": 0,
              "bid_price1": 0,
              "bid_volume1": 0,
              "ask_price1": 0,
              "ask_volume1": 0,
              "datetime": int(datetime.datetime.now().timestamp())
            }
            print(f"[TqApiService] 返回默认数据: {default_result}")
            return default_result
    
    async def disconnect(self) -> bool:
        """断开连接"""
        try:
            if self.api:
                self.connected = False
                self.api.close()
                self.api = None
            return True
        except Exception as e:
            print(f"断开连接失败: {e}")
            return False
    
    def _convert_period(self, period: str) -> int:
        """转换周期格式为秒"""
        period_map = {
            "1M": 60,
            "5M": 300,
            "15M": 900,
            "30M": 1800,
            "1H": 3600,
            "4H": 14400,
            "1D": 86400
        }
        return period_map.get(period, 3600)  # 默认1小时