amazing-data-service/backend/app/services/amazing_data_adapter.py

"""
AmazingData SDK适配器
封装AmazingData SDK的所有接口
"""
import logging
from typing import List, Dict, Callable, Optional, Tuple
from datetime import date
import pandas as pd

logger = logging.getLogger(__name__)


class AmazingDataAdapter:
    """AmazingData SDK适配器"""
    
    def __init__(self, config: Dict):
        """
        初始化适配器
        
        Args:
            config: SDK配置字典
                - username: 用户名
                - password: 密码
                - host: 服务器地址
                - port: 端口
                - local_path: 本地缓存路径
        """
        self.config = config
        self._ad = None
        self._base_data = None
        self._market_data = None
        self._info_data = None
        self._calendar = None
        self._is_connected = False
    
    def connect(self) -> bool:
        """
        建立与AmazingData SDK的连接
        
        Returns:
            是否连接成功
        """
        try:
            import AmazingData as ad
            
            ad.login(
                username=self.config["username"],
                password=self.config["password"],
                host=self.config["host"],
                port=self.config["port"]
            )
            
            self._ad = ad
            
            try:
                self._base_data = ad.BaseData()
            except Exception as e:
                logger.error(f"AmazingData SDK登录验证失败: {str(e)}")
                return False
            
            self._info_data = ad.InfoData()
            self._calendar = self._base_data.get_calendar()
            self._market_data = ad.MarketData(self._calendar)
            
            self._is_connected = True
            logger.info("AmazingData SDK连接成功")
            return True
            
        except ImportError:
            logger.error("AmazingData SDK未安装，请安装SDK后再试")
            return False
        except Exception as e:
            logger.error(f"AmazingData SDK连接失败: {str(e)}")
            return False
    
    def disconnect(self):
        """断开SDK连接"""
        if self._is_connected and self._ad:
            try:
                self._ad.logout(self.config["username"])
            except Exception as e:
                logger.warning(f"断开SDK连接时出错: {str(e)}")
        
        self._is_connected = False
        self._ad = None
        self._base_data = None
        self._market_data = None
        self._info_data = None
        self._calendar = None
        logger.info("AmazingData SDK已断开")
    
    def is_connected(self) -> bool:
        """检查是否已连接"""
        return self._is_connected
    
    # ==================== 基础数据接口 ====================
    
    def get_code_list(self, security_type: str) -> List[str]:
        """
        获取代码列表
        
        Args:
            security_type: 证券类型
                - EXTRA_STOCK_A: 沪深A股
                - EXTRA_FUTURE: 期货
                - EXTRA_ETF: ETF
                - EXTRA_INDEX_A: 指数
        
        Returns:
            代码列表
        """
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            if security_type == "EXTRA_FUTURE":
                return self._base_data.get_future_code_list()
            elif security_type == "EXTRA_OPTION":
                return self._base_data.get_option_code_list()
            else:
                return self._base_data.get_code_list(security_type)
        except Exception as e:
            logger.error(f"获取代码列表失败: {str(e)}")
            return []
    
    def get_future_varieties(self) -> List[str]:
        """
        获取期货品种列表
        
        Returns:
            品种代码列表 (如: IF, IC, al, zn, etc.)
        """
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            codes = self._base_data.get_future_code_list()
            varieties = set()
            for code in codes:
                # 代码格式: 品种代码+月份+交易所
                # 如: IF2401.CFE, al2702.SHF
                parts = code.split('.')
                if len(parts) >= 1:
                    contract = parts[0]
                    # 提取品种代码（去掉月份部分）
                    # 月份格式: YYMM (如 2401, 2702)
                    variety = contract[:-4] if len(contract) > 4 else contract
                    varieties.add(variety)
            return sorted(list(varieties))
        except Exception as e:
            logger.error(f"获取期货品种列表失败: {str(e)}")
            return []
    
    def get_main_contract(self, variety: str) -> Optional[str]:
        """
        获取指定品种的当前主力合约
        
        Args:
            variety: 品种代码 (如: IF, IC, al, zn)
        
        Returns:
            主力合约代码 (如: IF2401.CFE)
        """
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            from datetime import datetime
            
            codes = self._base_data.get_future_code_list()
            current_year = datetime.now().year
            current_month = datetime.now().month
            
            # 筛选该品种的合约
            variety_codes = []
            for code in codes:
                parts = code.split('.')
                if len(parts) >= 1:
                    contract = parts[0]
                    if contract.startswith(variety) and len(contract) > len(variety):
                        variety_codes.append(code)
            
            if not variety_codes:
                return None
            
            # 找到最近月份的合约作为主力合约
            # 主力合约通常是当前月份或下个月份的合约
            def get_contract_month(code):
                parts = code.split('.')
                contract = parts[0]
                month_str = contract[-4:]  # YYMM
                try:
                    year = int(month_str[:2]) + 2000
                    month = int(month_str[2:])
                    return (year, month)
                except:
                    return (9999, 99)
            
            # 按月份排序
            variety_codes.sort(key=get_contract_month)
            
            # 找到第一个大于当前月份的合约
            for code in variety_codes:
                year, month = get_contract_month(code)
                if year >= current_year and month >= current_month:
                    return code
            
            # 如果没有找到，返回最后一个合约
            return variety_codes[-1] if variety_codes else None
            
        except Exception as e:
            logger.error(f"获取主力合约失败: {str(e)}")
            return None
    
    def get_all_main_contracts(self) -> Dict[str, str]:
        """
        获取所有品种的主力合约
        
        Returns:
            字典 {品种: 主力合约}
        """
        varieties = self.get_future_varieties()
        main_contracts = {}
        for variety in varieties:
            main_contract = self.get_main_contract(variety)
            if main_contract:
                main_contracts[variety] = main_contract
        return main_contracts
    
    def get_code_info(self, security_type: str) -> pd.DataFrame:
        """
        获取证券信息
        
        Args:
            security_type: 证券类型
        
        Returns:
            证券信息DataFrame
        """
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            return self._base_data.get_code_info(security_type)
        except Exception as e:
            logger.error(f"获取证券信息失败: {str(e)}")
            return pd.DataFrame()
    
    def get_trading_calendar(self, market: str) -> List[int]:
        """
        获取交易日历
        
        Args:
            market: 市场代码 (SH, SZ, CFE)
        
        Returns:
            交易日列表 (YYYYMMDD格式)
        """
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            if isinstance(self._calendar, list):
                return self._calendar
            elif hasattr(self._calendar, 'get_trading_calendar'):
                return self._calendar.get_trading_calendar(market)
            else:
                return []
        except Exception as e:
            logger.error(f"获取交易日历失败: {str(e)}")
            return []
    
    def get_adj_factor(self, codes: List[str]) -> pd.DataFrame:
        """获取复权因子"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            return self._base_data.get_adj_factor(codes)
        except Exception as e:
            logger.error(f"获取复权因子失败: {str(e)}")
            return pd.DataFrame()
    
    def get_backward_factor(self, codes: List[str]) -> pd.DataFrame:
        """获取后复权因子"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            return self._base_data.get_backward_factor(codes)
        except Exception as e:
            logger.error(f"获取后复权因子失败: {str(e)}")
            return pd.DataFrame()
    
    # ==================== 历史行情接口 ====================
    
    def get_kline(
        self,
        codes: List[str],
        start_date,
        end_date,
        period: str = "daily"
    ) -> Dict[str, pd.DataFrame]:
        """
        获取K线数据
        
        Args:
            codes: 代码列表
            start_date: 开始日期 (YYYYMMDD 或 date对象)
            end_date: 结束日期 (YYYYMMDD 或 date对象)
            period: 周期 (daily, min1, min5, min15, min30, min60)
        
        Returns:
            字典 {code: DataFrame}
        """
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        # 转换日期格式
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        period_map = {
            "daily": 10008,
            "min1": 10000,
            "min5": 10002,
            "min15": 10004,
            "min30": 10005,
            "min60": 10006,
        }
        
        period_value = period_map.get(period, 10008)
        
        try:
            return self._market_data.query_kline(codes, start_date, end_date, period_value)
        except Exception as e:
            logger.error(f"获取K线数据失败: {str(e)}")
            return {code: pd.DataFrame() for code in codes}
    
    def get_snapshot(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> Dict[str, pd.DataFrame]:
        """
        获取历史快照数据
        
        Args:
            codes: 代码列表
            start_date: 开始日期
            end_date: 结束日期
        
        Returns:
            字典 {code: DataFrame}
        """
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._market_data.query_snapshot(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取快照数据失败: {str(e)}")
            return {code: pd.DataFrame() for code in codes}
    
    # ==================== 财务数据接口 ====================
    
    def get_balance_sheet(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> Dict[str, pd.DataFrame]:
        """获取资产负债表"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_balance_sheet(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取资产负债表失败: {str(e)}")
            return {code: pd.DataFrame() for code in codes}
    
    def get_cash_flow(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> Dict[str, pd.DataFrame]:
        """获取现金流量表"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_cash_flow(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取现金流量表失败: {str(e)}")
            return {code: pd.DataFrame() for code in codes}
    
    def get_income_statement(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> Dict[str, pd.DataFrame]:
        """获取利润表"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_income_statement(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取利润表失败: {str(e)}")
            return {code: pd.DataFrame() for code in codes}
    
    def get_profit_express(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> pd.DataFrame:
        """获取业绩快报"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_profit_express(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取业绩快报失败: {str(e)}")
            return pd.DataFrame()
    
    def get_profit_notice(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> pd.DataFrame:
        """获取业绩预告"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_profit_notice(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取业绩预告失败: {str(e)}")
            return pd.DataFrame()
    
    # ==================== 股东股本接口 ====================
    
    def get_top10_shareholders(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> pd.DataFrame:
        """获取十大股东"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_top10_shareholders(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取十大股东失败: {str(e)}")
            return pd.DataFrame()
    
    def get_shareholder_count(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> pd.DataFrame:
        """获取股东户数"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_shareholder_count(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取股东户数失败: {str(e)}")
            return pd.DataFrame()
    
    def get_equity_structure(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> pd.DataFrame:
        """获取股本结构"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_equity_structure(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取股本结构失败: {str(e)}")
            return pd.DataFrame()
    
    # ==================== 融资融券接口 ====================
    
    def get_margin_summary(
        self,
        start_date,
        end_date
    ) -> pd.DataFrame:
        """获取融资融券汇总"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_margin_summary(start_date, end_date)
        except Exception as e:
            logger.error(f"获取融资融券汇总失败: {str(e)}")
            return pd.DataFrame()
    
    def get_margin_detail(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> Dict[str, pd.DataFrame]:
        """获取融资融券明细"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_margin_detail(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取融资融券明细失败: {str(e)}")
            return {code: pd.DataFrame() for code in codes}
    
    # ==================== 指数数据接口 ====================
    
    def get_index_constituents(self, codes: List[str]) -> Dict[str, pd.DataFrame]:
        """获取指数成分股"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            return self._info_data.get_index_constituents(codes)
        except Exception as e:
            logger.error(f"获取指数成分股失败: {str(e)}")
            return {code: pd.DataFrame() for code in codes}
    
    def get_index_weights(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> Dict[str, pd.DataFrame]:
        """获取指数权重"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_index_weights(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取指数权重失败: {str(e)}")
            return {code: pd.DataFrame() for code in codes}
    
    # ==================== ETF数据接口 ====================
    
    def get_etf_pcf(self, codes: List[str]) -> Tuple[pd.DataFrame, Dict[str, pd.DataFrame]]:
        """获取ETF申赎数据"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            return self._info_data.get_etf_pcf(codes)
        except Exception as e:
            logger.error(f"获取ETF申赎数据失败: {str(e)}")
            return pd.DataFrame(), {}
    
    def get_fund_share(
        self,
        codes: List[str],
        start_date,
        end_date
    ) -> Dict[str, pd.DataFrame]:
        """获取基金份额数据"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        if isinstance(start_date, date):
            start_date = int(start_date.strftime("%Y%m%d"))
        if isinstance(end_date, date):
            end_date = int(end_date.strftime("%Y%m%d"))
        
        try:
            return self._info_data.get_fund_share(codes, start_date, end_date)
        except Exception as e:
            logger.error(f"获取基金份额数据失败: {str(e)}")
            return {code: pd.DataFrame() for code in codes}
    
    # ==================== 可转债数据接口 ====================
    
    def get_kzz_issuance(self, codes: List[str]) -> Dict[str, pd.DataFrame]:
        """获取可转债发行数据"""
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            return self._info_data.get_kzz_issuance(codes)
        except Exception as e:
            logger.error(f"获取可转债发行数据失败: {str(e)}")
            return {code: pd.DataFrame() for code in codes}
    
    # ==================== 实时数据订阅 ====================
    
    def subscribe_snapshot(self, codes: List[str], callback: Callable):
        """
        订阅实时快照
        
        Args:
            codes: 代码列表
            callback: 回调函数，接收(code, data)参数
        """
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            # 使用实时数据接口订阅
            realtime = self._ad.RealtimeData()
            realtime.subscribe_snapshot(codes, callback)
        except Exception as e:
            logger.error(f"订阅实时快照失败: {str(e)}")
    
    def subscribe_kline(
        self,
        codes: List[str],
        period: str,
        callback: Callable
    ):
        """
        订阅实时K线
        
        Args:
            codes: 代码列表
            period: 周期
            callback: 回调函数
        """
        if not self._is_connected:
            raise RuntimeError("SDK未连接")
        
        try:
            realtime = self._ad.RealtimeData()
            realtime.subscribe_kline(codes, period, callback)
        except Exception as e:
            logger.error(f"订阅实时K线失败: {str(e)}")
    
    def unsubscribe(self, codes: List[str] = None):
        """
        取消订阅
        
        Args:
            codes: 代码列表，None表示取消所有
        """
        if not self._is_connected:
            return
        
        try:
            realtime = self._ad.RealtimeData()
            realtime.unsubscribe(codes)
        except Exception as e:
            logger.error(f"取消订阅失败: {str(e)}")