fin_data_unify_plat/backend/app/services/kline/futures_service.py

"""
期货 K 线服务 v2.2
支持多周期、持仓量、结算价等期货特有数据
"""
import logging
from datetime import datetime, timedelta
from typing import List, Optional, Dict, Any
from enum import Enum

from sqlalchemy.orm import Session

from app.models.kline import (
    Frequency, FuturesKLineItem, FuturesKLineData,
    FuturesSymbolInfo, FuturesContractInfo, FuturesKLineQuery
)
from app.repositories.kline.futures_repository import FuturesKLineRepository
from app.services.cache_service import cache_service

logger = logging.getLogger(__name__)


# 支持的期货 K 线周期
FUTURES_FREQUENCIES = [
    Frequency.FREQ_1M,
    Frequency.FREQ_5M,
    Frequency.FREQ_15M,
    Frequency.FREQ_30M,
    Frequency.FREQ_1H,
    Frequency.FREQ_1D,
    Frequency.FREQ_1W,
    Frequency.FREQ_1MONTH,
]


class FuturesKLineService:
    """期货 K 线服务"""
    
    def __init__(self, db: Session):
        self.repository = FuturesKLineRepository(db)
        self.db = db
    
    async def query_klines(
        self,
        symbol: str,
        freq: Frequency,
        start: datetime,
        end: datetime,
        use_cache: bool = True
    ) -> FuturesKLineData:
        """
        查询期货 K 线数据
        
        Args:
            symbol: 合约代码 (如 IF2406, AG2605.SHF)
            freq: K 线周期
            start: 开始时间
            end: 结束时间
            use_cache: 是否使用缓存
            
        Returns:
            FuturesKLineData: K 线数据响应
            
        Raises:
            ValueError: 参数验证失败
        """
        # 参数验证
        self._validate_params(symbol, freq, start, end)
        
        # 验证周期是否支持
        if freq not in FUTURES_FREQUENCIES:
            raise ValueError(f"不支持的期货 K 线周期: {freq}")
        
        # 尝试从缓存获取
        cache_key = None
        if use_cache:
            cache_key = f"futures_kline:{symbol}:{freq.value}:{start.strftime('%Y%m%d')}:{end.strftime('%Y%m%d')}"
            cached = await cache_service.get(cache_key)
            if cached:
                logger.info(f"缓存命中: {cache_key}")
                return FuturesKLineData(**cached)
        
        # 查询数据库
        items = self.repository.get_klines(symbol, freq, start, end)
        
        # 如果没有数据，尝试从适配器获取
        if not items:
            logger.info(f"数据库无 {symbol} 数据，尝试从数据源获取")
            items = await self._fetch_from_adapter(symbol, freq, start, end)
            
            # 保存到数据库
            if items:
                self._save_klines_to_db(symbol, freq, items)
        
        # 获取品种信息
        symbol_info = self.repository.get_symbol_info(symbol)
        
        result = FuturesKLineData(
            symbol=symbol,
            name=symbol_info.name if symbol_info else "",
            freq=freq,
            count=len(items),
            items=items
        )
        
        # 写入缓存
        if cache_key and items:
            await cache_service.set(cache_key, result.model_dump(), expire=300)  # 5分钟缓存
        
        return result
    
    async def query_klines_batch(
        self,
        symbols: List[str],
        freq: Frequency,
        start: datetime,
        end: datetime,
        use_cache: bool = True
    ) -> Dict[str, FuturesKLineData]:
        """
        批量查询期货 K 线数据
        
        Args:
            symbols: 合约代码列表 (最多100个)
            freq: K 线周期
            start: 开始时间
            end: 结束时间
            use_cache: 是否使用缓存
            
        Returns:
            Dict[str, FuturesKLineData]: 合约代码 -> K线数据映射
        """
        if len(symbols) > 100:
            raise ValueError("批量查询最多支持 100 个合约")
        
        results = {}
        for symbol in symbols:
            try:
                results[symbol] = await self.query_klines(
                    symbol, freq, start, end, use_cache
                )
            except Exception as e:
                logger.error(f"查询 {symbol} 失败: {e}")
                results[symbol] = FuturesKLineData(
                    symbol=symbol,
                    name="",
                    freq=freq,
                    count=0,
                    items=[],
                    error=str(e)
                )
        
        return results
    
    async def get_main_contract_klines(
        self,
        product_code: str,
        freq: Frequency,
        start: datetime,
        end: datetime
    ) -> Optional[FuturesKLineData]:
        """
        获取主力合约 K 线数据
        
        Args:
            product_code: 品种代码 (如 IF, IC, AG)
            freq: K 线周期
            start: 开始时间
            end: 结束时间
            
        Returns:
            Optional[FuturesKLineData]: 主力合约 K 线数据
        """
        # 获取主力合约
        main_contract = self.repository.get_main_contract(product_code)
        if not main_contract:
            logger.warning(f"品种 {product_code} 无主力合约")
            return None
        
        return await self.query_klines(main_contract, freq, start, end)
    
    def get_active_contracts(
        self,
        product_code: str,
        limit: int = 10
    ) -> List[FuturesContractInfo]:
        """
        获取活跃合约列表
        
        Args:
            product_code: 品种代码
            limit: 返回数量
            
        Returns:
            List[FuturesContractInfo]: 活跃合约列表
        """
        return self.repository.get_active_contracts(product_code, limit)
    
    def get_contract_info(self, symbol: str) -> Optional[FuturesContractInfo]:
        """
        获取合约信息
        
        Args:
            symbol: 合约代码
            
        Returns:
            Optional[FuturesContractInfo]: 合约信息
        """
        return self.repository.get_contract_info(symbol)
    
    def _validate_params(
        self,
        symbol: str,
        freq: Frequency,
        start: datetime,
        end: datetime
    ) -> None:
        """验证查询参数"""
        if not symbol:
            raise ValueError("合约代码不能为空")
        
        if start >= end:
            raise ValueError("开始时间必须早于结束时间")
        
        # 时间范围限制
        max_days = {
            Frequency.FREQ_1M: 30,
            Frequency.FREQ_5M: 60,
            Frequency.FREQ_15M: 90,
            Frequency.FREQ_30M: 180,
            Frequency.FREQ_1H: 365,
            Frequency.FREQ_1D: 3650,  # 10年
            Frequency.FREQ_1W: 3650,
            Frequency.FREQ_1MONTH: 3650,
        }
        
        days = (end - start).days
        max_allowed = max_days.get(freq, 365)
        if days > max_allowed:
            raise ValueError(f"{freq.value} 周期最多查询 {max_allowed} 天数据")
    
    async def _fetch_from_adapter(
        self,
        symbol: str,
        freq: Frequency,
        start: datetime,
        end: datetime
    ) -> List[FuturesKLineItem]:
        """
        从数据源适配器获取 K 线数据
        
        注意: 此方法修复了原代码中的异步问题
        原代码错误地使用 asyncio.new_event_loop()
        """
        try:
            from app.services.amazing_data_service import AmazingDataService
            
            # 获取服务实例
            service = AmazingDataService()
            
            # 转换频率格式
            period_map = {
                Frequency.FREQ_1M: "min1",
                Frequency.FREQ_5M: "min5",
                Frequency.FREQ_15M: "min15",
                Frequency.FREQ_30M: "min30",
                Frequency.FREQ_1H: "min60",
                Frequency.FREQ_1D: "day",
                Frequency.FREQ_1W: "week",
                Frequency.FREQ_1MONTH: "month",
            }
            period = period_map.get(freq, "day")
            
            # 使用上下文管理器确保连接正确关闭
            async with service.get_connection() as conn:
                # 调用适配器获取数据
                df = await conn.get_kline(
                    symbol=symbol,
                    period=period,
                    start_time=start.strftime("%Y%m%d%H%M%S"),
                    end_time=end.strftime("%Y%m%d%H%M%S")
                )
            
            if df is None or df.empty:
                logger.warning(f"适配器返回空数据: {symbol}")
                return []
            
            # 转换为 KLineItem 列表
            items = []
            for _, row in df.iterrows():
                items.append(FuturesKLineItem(
                    symbol=symbol,
                    time=row.get('time', row.name) if isinstance(row.get('time'), datetime) 
                         else datetime.strptime(str(row.get('time', row.name)), "%Y-%m-%d %H:%M:%S"),
                    open=float(row.get('open', 0)),
                    high=float(row.get('high', 0)),
                    low=float(row.get('low', 0)),
                    close=float(row.get('close', 0)),
                    volume=int(row.get('volume', 0)),
                    open_interest=int(row.get('open_interest', 0)) if 'open_interest' in row else 0,
                    settlement_price=float(row.get('settlement', 0)) if 'settlement' in row else None,
                    trade_date=row.get('trade_date', row.get('time', row.name).date() if hasattr(row.get('time', row.name), 'date') else None)
                ))
            
            logger.info(f"从适配器获取 {symbol} {freq.value} 数据 {len(items)} 条")
            return items
            
        except ImportError:
            logger.warning("AmazingDataService 未安装，无法从适配器获取数据")
            return []
        except Exception as e:
            logger.error(f"从适配器获取数据失败: {e}")
            return []
    
    def _save_klines_to_db(
        self,
        symbol: str,
        freq: Frequency,
        items: List[FuturesKLineItem]
    ) -> int:
        """
        保存 K 线数据到数据库
        
        Args:
            symbol: 合约代码
            freq: K 线周期
            items: K 线数据列表
            
        Returns:
            int: 保存的数量
        """
        try:
            count = self.repository.save_klines(symbol, freq, items)
            logger.info(f"保存 {symbol} {freq.value} 数据 {count} 条")
            return count
        except Exception as e:
            logger.error(f"保存数据失败: {e}")
            return 0
    
    async def sync_klines(
        self,
        symbol: str,
        freq: Frequency,
        days: int = 30
    ) -> Dict[str, Any]:
        """
        同步 K 线数据
        
        Args:
            symbol: 合约代码
            freq: K 线周期
            days: 同步天数
            
        Returns:
            Dict: 同步结果
        """
        end = datetime.now()
        start = end - timedelta(days=days)
        
        # 获取最新时间戳
        latest = self.repository.get_latest_timestamp(symbol, freq)
        if latest and latest > start:
            start = latest + timedelta(seconds=1)
        
        if start >= end:
            return {
                "symbol": symbol,
                "freq": freq.value,
                "status": "already_synced",
                "count": 0
            }
        
        # 从适配器获取数据
        items = await self._fetch_from_adapter(symbol, freq, start, end)
        
        # 保存到数据库
        count = self._save_klines_to_db(symbol, freq, items)
        
        return {
            "symbol": symbol,
            "freq": freq.value,
            "status": "synced",
            "count": count,
            "start": start.isoformat(),
            "end": end.isoformat()
        }


# 服务工厂函数
def get_futures_kline_service(db: Session) -> FuturesKLineService:
    """获取期货 K 线服务实例"""
    return FuturesKLineService(db)