"""
TradingAgents AkShare数据适配器

该模块提供AkShare免费数据源的完整集成，替代所有付费数据源：
- 完全免费，无API密钥要求
- 覆盖A股、港股、美股、期货、指数等全市场数据
- 实时行情、历史数据、资金流向、技术指标
- 智能缓存和错误处理机制

@author TradingAgents团队
@version MVP-2.0 (AkShare集成版)
@since 2024-12
"""

import os
import time
import json
import sqlite3
import akshare as ak
import pandas as pd
from datetime import datetime, timedelta
from typing import Dict, Any, List, Optional, Union
from dataclasses import dataclass
import logging
from functools import wraps

# 配置日志
logging.basicConfig(level=logging.INFO)
logger = logging.getLogger(__name__)


# --- 新增：通用重试装饰器 ---
def retry_on_failure(max_retries: int, retry_delay: float):
    """
    一个装饰器，用于在函数调用失败时自动重试。
    
    @param {int} max_retries - 最大重试次数。
    @param {float} retry_delay - 每次重试之间的基础延迟秒数。
    """
    def decorator(func):
        @wraps(func)
        def wrapper(*args, **kwargs):
            # args[0] is 'self' for instance methods that are part of a class
            self_instance = args[0] if args else None
            
            # 从实例配置中获取重试参数，如果不存在则使用装饰器默认值
            effective_max_retries = max_retries
            effective_retry_delay = retry_delay
            if self_instance and hasattr(self_instance, 'config'):
                effective_max_retries = getattr(self_instance.config, 'max_retries', max_retries)
                effective_retry_delay = getattr(self_instance.config, 'retry_delay', retry_delay)

            for attempt in range(effective_max_retries):
                try:
                    return func(*args, **kwargs)
                except Exception as e:
                    logger.warning(f"调用 {func.__name__} 失败 (尝试 {attempt + 1}/{effective_max_retries}): {e}")
                    if attempt + 1 == effective_max_retries:
                        logger.error(f"调用 {func.__name__} 在 {effective_max_retries} 次尝试后最终失败。")
                        # 重新抛出自定义异常，以便上层统一处理
                        raise AkShareError(f"AkShare API '{func.__name__}' 在多次重试后最终失败: {e}") from e
                    time.sleep(effective_retry_delay * (attempt + 1)) # 线性增加延迟
        return wrapper
    return decorator


@dataclass
class AkShareConfig:
    """
    AkShare配置类
    
    @description 管理AkShare数据源的配置参数
    """
    cache_ttl_hours: int = 6
    enable_fallback: bool = True
    user_agent: str = "TradingAgents-AkShare/2.0"
    max_retries: int = 3
    retry_delay: float = 1.0


class AkShareCacheManager:
    """
    AkShare智能缓存管理器
    
    @description 专为AkShare数据优化的缓存系统
    """
    
    def __init__(self, cache_db_path: str = "data/cache/akshare_cache.db"):
        self.cache_db_path = cache_db_path
        self._ensure_cache_dir()
        self._init_cache_db()
    
    def _ensure_cache_dir(self):
        """确保缓存目录存在"""
        os.makedirs(os.path.dirname(self.cache_db_path), exist_ok=True)
    
    def _init_cache_db(self):
        """初始化缓存数据库"""
        with sqlite3.connect(self.cache_db_path) as conn:
            conn.execute("""
                CREATE TABLE IF NOT EXISTS akshare_cache (
                    key TEXT PRIMARY KEY,
                    data TEXT NOT NULL,
                    timestamp REAL NOT NULL,
                    ttl_hours INTEGER NOT NULL,
                    data_type TEXT NOT NULL
                )
            """)
            conn.execute("""
                CREATE INDEX IF NOT EXISTS idx_akshare_timestamp 
                ON akshare_cache(timestamp)
            """)
    
    def get_cached_data(self, key: str, max_age_hours: Union[int, float] = 24) -> Optional[pd.DataFrame]:
        """获取缓存的DataFrame数据"""
        try:
            with sqlite3.connect(self.cache_db_path) as conn:
                cursor = conn.execute(
                    "SELECT data, timestamp, ttl_hours FROM akshare_cache WHERE key = ?",
                    (key,)
                )
                result = cursor.fetchone()
                
                if result:
                    data_str, timestamp, ttl_hours = result
                    age_hours = (time.time() - timestamp) / 3600
                    
                    if age_hours < min(float(max_age_hours), float(ttl_hours)):
                        logger.info(f"AkShare缓存命中: {key} (年龄: {age_hours:.1f}小时)")
                        return pd.read_json(data_str, orient='records')
                    else:
                        conn.execute("DELETE FROM akshare_cache WHERE key = ?", (key,))
                        logger.info(f"AkShare缓存过期已删除: {key}")
                
                return None
        except Exception as e:
            logger.error(f"获取AkShare缓存失败: {e}")
            return None
    
    def set_cached_data(self, key: str, data: pd.DataFrame, ttl_hours: Union[int, float] = 24, data_type: str = "general") -> bool:
        """设置缓存的DataFrame数据"""
        try:
            with sqlite3.connect(self.cache_db_path) as conn:
                data_json = data.to_json(orient='records', date_format='iso')
                conn.execute(
                    "INSERT OR REPLACE INTO akshare_cache (key, data, timestamp, ttl_hours, data_type) VALUES (?, ?, ?, ?, ?)",
                    (key, data_json, time.time(), ttl_hours, data_type)
                )
                logger.info(f"AkShare缓存已设置: {key} (TTL: {ttl_hours}小时)")
                return True
        except Exception as e:
            logger.error(f"设置AkShare缓存失败: {e}")
            return False


class AkShareStockAdapter:
    """
    AkShare股票数据适配器
    
    @description 提供A股、港股、美股的统一数据接口
    """
    
    def __init__(self, config: AkShareConfig):
        self.config = config
        self.cache = AkShareCacheManager()
        self._us_symbol_map: Optional[Dict[str, str]] = None  # 内存缓存美股代码映射
    
    @retry_on_failure(max_retries=2, retry_delay=2)
    def _update_us_symbol_map(self) -> Dict[str, str]:
        """获取并缓存美股代码映射表"""
        logger.info("正在从AkShare获取最新美股代码映射表...")
        df = ak.stock_us_spot_em()
        if '代码' in df.columns:
            # AkShare美股代码格式为 "105.AAPL"，我们需要提取 "AAPL" 作为键
            symbol_map = {row['代码'].split('.')[-1]: row['代码'] for _, row in df.iterrows()}
            self._us_symbol_map = symbol_map
            logger.info(f"美股代码映射表更新成功，共 {len(self._us_symbol_map)} 个代码。")
            # 将映射表也存入磁盘缓存，避免每次重启都重新获取
            map_df = pd.DataFrame(list(self._us_symbol_map.items()), columns=['common_symbol', 'ak_symbol'])
            self.cache.set_cached_data("us_symbol_map", map_df, ttl_hours=24 * 7) # 缓存一周
            return self._us_symbol_map
        raise AkShareError("无法从AkShare获取有效的美股'代码'列。")

    def _get_us_symbol(self, common_symbol: str) -> str:
        """
        根据通用代码获取AkShare专用美股代码。
        优先使用内存缓存，其次是磁盘缓存，最后是网络获取。
        """
        common_symbol = common_symbol.upper()
        
        if self._us_symbol_map is None:
            # 尝试从磁盘缓存加载
            cached_map_df = self.cache.get_cached_data("us_symbol_map", max_age_hours=24*7)
            if cached_map_df is not None and not cached_map_df.empty:
                self._us_symbol_map = dict(zip(cached_map_df['common_symbol'], cached_map_df['ak_symbol']))
                logger.info("已从磁盘缓存加载美股代码映射表。")
            else:
                # 缓存没有，从网络获取
                self._update_us_symbol_map()
        
        # Linter类型守卫: 经过上面的逻辑，_us_symbol_map不应为None
        if self._us_symbol_map is None:
            raise AkShareError("未能成功加载或更新美股代码映射表。")

        ak_symbol = self._us_symbol_map.get(common_symbol)
        
        if ak_symbol is None:
            logger.warning(f"在缓存的映射中未找到美股代码 {common_symbol}，尝试强制刷新...")
            self._update_us_symbol_map() # 强制刷新
            
            # Linter类型守卫: 再次检查
            if self._us_symbol_map is None:
                 raise AkShareError("刷新后，美股代码映射表仍然无法加载。")
            
            ak_symbol = self._us_symbol_map.get(common_symbol)
            if ak_symbol is None:
                 raise AkShareError(f"在AkShare代码映射中最终也未找到美股代码: {common_symbol}")
        
        return ak_symbol

    @retry_on_failure(max_retries=3, retry_delay=1)
    def get_stock_realtime(self, symbol: str, market: str = "auto") -> Dict[str, Any]:
        """
        获取股票实时行情
        
        @param {str} symbol - 股票代码
        @param {str} market - 市场类型 ("auto", "a_stock", "hk_stock", "us_stock")
        @returns {Dict[str, Any]} 实时行情数据
        """
        cache_key = f"realtime_{market}_{symbol}"
        
        # 尝试从缓存获取 (实时数据缓存5分钟)
        cached_data = self.cache.get_cached_data(cache_key, max_age_hours=0.083)  # 5分钟
        if cached_data is not None and not cached_data.empty:
            return self._format_realtime_data(cached_data.iloc[0].to_dict(), symbol, market)
        
        # 自动检测市场类型
        if market == "auto":
            market = self._detect_market(symbol)
        
        # 根据市场获取数据
        if market == "a_stock":
            df = ak.stock_zh_a_spot_em()
            stock_data = df[df['代码'] == symbol]
        elif market == "hk_stock":
            df = ak.stock_hk_spot_em()
            stock_data = df[df['代码'] == symbol]
        elif market == "us_stock":
            # 对于美股，实时数据从历史数据最后一条记录获取
            hist_df = self.get_stock_history(symbol, market="us_stock")
            if hist_df.empty:
                 raise ValueError(f"无法获取美股 {symbol} 的历史数据来推断实时价格。")
            
            # 将最后一行历史数据转换为实时数据格式
            latest_data = hist_df.iloc[-1].to_dict()
            stock_data = pd.DataFrame([latest_data])
            # 字段名已经在 _standardize_history_columns 中被标准化了
        else:
            raise ValueError(f"不支持的市场类型: {market}")
        
        if stock_data.empty:
            raise ValueError(f"未找到股票代码: {symbol}")
        
        # 缓存数据
        self.cache.set_cached_data(cache_key, stock_data, 0.083, "realtime")
        
        result = self._format_realtime_data(stock_data.iloc[0].to_dict(), symbol, market)
        logger.info(f"AkShare实时行情获取成功: {symbol} ({market})")
        return result
    
    def _detect_market(self, symbol: str) -> str:
        """自动检测股票市场类型"""
        symbol = symbol.upper()
        
        # A股判断
        if (symbol.startswith(('00', '30')) and len(symbol) == 6) or \
           (symbol.startswith('6') and len(symbol) == 6):
            return "a_stock"
        
        # 港股判断
        if symbol.startswith('0') and len(symbol) == 5:
            return "hk_stock"
        
        # 美股判断 (默认)
        return "us_stock"
    
    def _format_realtime_data(self, raw_data: Dict, symbol: str, market: str) -> Dict[str, Any]:
        """格式化实时数据"""
        try:
            if market == "a_stock":
                return {
                    "symbol": symbol,
                    "market": market,
                    "name": raw_data.get('名称', ''),
                    "current_price": raw_data.get('最新价', 0),
                    "change": raw_data.get('涨跌额', 0),
                    "change_percent": raw_data.get('涨跌幅', 0),
                    "volume": raw_data.get('成交量', 0),
                    "turnover": raw_data.get('成交额', 0),
                    "high": raw_data.get('最高', 0),
                    "low": raw_data.get('最低', 0),
                    "open": raw_data.get('今开', 0),
                    "prev_close": raw_data.get('昨收', 0),
                    "amplitude": raw_data.get('振幅', 0),
                    "pe_ratio": raw_data.get('市盈率-动态', 0),
                    "market_cap": raw_data.get('总市值', 0),
                    "timestamp": datetime.now().isoformat(),
                    "data_source": "AkShare-A股"
                }
            elif market == "hk_stock":
                return {
                    "symbol": symbol,
                    "market": market,
                    "name": raw_data.get('名称', ''),
                    "current_price": raw_data.get('最新价', 0),
                    "change": raw_data.get('涨跌额', 0),
                    "change_percent": raw_data.get('涨跌幅', 0),
                    "volume": raw_data.get('成交量', 0),
                    "turnover": raw_data.get('成交额', 0),
                    "high": raw_data.get('最高', 0),
                    "low": raw_data.get('最低', 0),
                    "open": raw_data.get('今开', 0),
                    "prev_close": raw_data.get('昨收', 0),
                    "timestamp": datetime.now().isoformat(),
                    "data_source": "AkShare-港股"
                }
            elif market == "us_stock":
                return {
                    "symbol": symbol,
                    "market": market,
                    "name": raw_data.get('名称', ''),
                    "current_price": raw_data.get('最新价', 0),
                    "change": raw_data.get('涨跌额', 0),
                    "change_percent": raw_data.get('涨跌幅', 0),
                    "volume": raw_data.get('成交量', 0),
                    "turnover": raw_data.get('成交额', 0),
                    "high": raw_data.get('最高价', 0),
                    "low": raw_data.get('最低价', 0),
                    "open": raw_data.get('开盘价', 0),
                    "prev_close": raw_data.get('昨收价', 0),
                    "market_cap": raw_data.get('总市值', 0),
                    "pe_ratio": raw_data.get('市盈率', 0),
                    "timestamp": datetime.now().isoformat(),
                    "data_source": "AkShare-美股"
                }
            else:
                return {"error": f"未知市场类型: {market}"}
                
        except Exception as e:
            logger.error(f"格式化实时数据失败: {e}")
            return {"error": str(e)}
    
    @retry_on_failure(max_retries=3, retry_delay=1)
    def get_stock_history(self, symbol: str, period: str = "1y", market: str = "auto") -> pd.DataFrame:
        """
        获取股票历史数据
        
        @param {str} symbol - 股票代码
        @param {str} period - 时间周期 ("1d", "1w", "1m", "3m", "6m", "1y", "2y")
        @param {str} market - 市场类型
        @returns {pd.DataFrame} 历史K线数据
        """
        cache_key = f"history_{market}_{symbol}_{period}"
        
        # 尝试从缓存获取 (历史数据缓存6小时)
        cached_data = self.cache.get_cached_data(cache_key, self.config.cache_ttl_hours)
        if cached_data is not None:
            return cached_data
        
        if market == "auto":
            market = self._detect_market(symbol)
        
        # 计算日期范围
        end_date = datetime.now()
        start_date = end_date - timedelta(days=365 * 2) # 拉取两年数据以确保有足够信息
        if period == "1d":
            start_date = end_date - timedelta(days=1)
        elif period == "1w":
            start_date = end_date - timedelta(weeks=1)
        elif period == "1m":
            start_date = end_date - timedelta(days=30)
        elif period == "3m":
            start_date = end_date - timedelta(days=90)
        elif period == "6m":
            start_date = end_date - timedelta(days=180)
        elif period == "1y":
            start_date = end_date - timedelta(days=365)
        
        # 根据市场获取历史数据
        if market == "a_stock":
            df = ak.stock_zh_a_hist(symbol=symbol, period="daily", adjust="qfq")
        elif market == "hk_stock":
            df = ak.stock_hk_daily(symbol=symbol)
        elif market == "us_stock":
            # 美股直接使用symbol查询
            start_str = start_date.strftime('%Y%m%d')
            end_str = end_date.strftime('%Y%m%d')
            # 使用 `stock_us_daily` 这个接口可能更稳定，因为它来自新浪
            df = ak.stock_us_daily(symbol=symbol, adjust="qfq")
            # 筛选日期
            df['date'] = pd.to_datetime(df['date'])
            df = df[(df['date'] >= start_date) & (df['date'] <= end_date)]

        else:
            raise ValueError(f"不支持的市场类型: {market}")
        
        if df.empty:
            raise ValueError(f"未找到历史数据: {symbol}")
        
        # 标准化列名
        df = self._standardize_history_columns(df, market)
        
        # 缓存数据
        self.cache.set_cached_data(cache_key, df, self.config.cache_ttl_hours, "history")
        
        logger.info(f"AkShare历史数据获取成功: {symbol} ({market}), {len(df)}条记录")
        return df
    
    def _standardize_history_columns(self, df: pd.DataFrame, market: str) -> pd.DataFrame:
        """标准化历史数据列名"""
        try:
            if market == "a_stock":
                # A股列名映射
                column_mapping = {
                    '日期': 'date',
                    '开盘': 'open',
                    '收盘': 'close',
                    '最高': 'high',
                    '最低': 'low',
                    '成交量': 'volume',
                    '成交额': 'turnover',
                    '振幅': 'amplitude',
                    '涨跌幅': 'change_percent',
                    '涨跌额': 'change',
                    '换手率': 'turnover_rate'
                }
            elif market == "hk_stock":
                # 港股列名映射
                column_mapping = {
                    '日期': 'date',
                    '开盘': 'open',
                    '收盘': 'close',
                    '最高': 'high',
                    '最低': 'low',
                    '成交量': 'volume',
                    '成交额': 'turnover'
                }
            elif market == "us_stock":
                # 美股列名映射
                column_mapping = {
                    '日期': 'date',
                    '开盘': 'open',
                    '收盘': 'close',
                    '最高': 'high',
                    '最低': 'low',
                    '成交量': 'volume',
                    '成交额': 'turnover',
                    '振幅': 'amplitude',
                    '涨跌幅': 'change_percent',
                    '涨跌额': 'change',
                    '换手率': 'turnover_rate'
                }
            else:
                return df
            
            # 重命名列
            df = df.rename(columns=column_mapping)
            
            # 确保日期列为datetime类型
            if 'date' in df.columns:
                df['date'] = pd.to_datetime(df['date'])
                df = df.set_index('date')
            
            return df
            
        except Exception as e:
            logger.error(f"标准化列名失败: {e}")
            return df


class AkShareIndexAdapter:
    """
    AkShare指数数据适配器
    
    @description 提供各类指数的实时和历史数据
    """
    
    def __init__(self, config: AkShareConfig):
        self.config = config
        self.cache = AkShareCacheManager()
    
    @retry_on_failure(max_retries=3, retry_delay=1)
    def get_major_indices(self) -> Dict[str, Any]:
        """获取全球主要指数的实时行情"""
        cache_key = "major_indices_realtime"
        
        # 尝试从缓存获取 (实时数据缓存5分钟)
        cached_data = self.cache.get_cached_data(cache_key, max_age_hours=0.083)
        if cached_data is not None and not cached_data.empty:
            return self._format_indices_data(cached_data)
        
        try:
            # 获取沪深重要指数
            df = ak.stock_zh_index_spot_em(symbol="沪深重要指数")
            
            # 缓存数据
            self.cache.set_cached_data(cache_key, df, 0.083, "indices")
            
            result = self._format_indices_data(df)
            logger.info(f"AkShare主要指数数据获取成功: {len(df)}个指数")
            return result
            
        except Exception as e:
            logger.error(f"获取主要指数失败: {e}")
            raise AkShareError(f"获取主要指数失败: {e}")
    
    def get_index_history(self, symbol: str, period: str = "1y") -> pd.DataFrame:
        """获取指数历史数据"""
        cache_key = f"index_history_{symbol}_{period}"
        
        # 尝试从缓存获取
        cached_data = self.cache.get_cached_data(cache_key, self.config.cache_ttl_hours)
        if cached_data is not None:
            return cached_data
        
        try:
            df = ak.stock_zh_index_daily(symbol=symbol)
            
            if df.empty:
                raise ValueError(f"未找到指数数据: {symbol}")
            
            # 根据周期过滤数据
            if period != "all":
                days_map = {"1d": 1, "1w": 7, "1m": 30, "3m": 90, "6m": 180, "1y": 365, "2y": 730}
                days = days_map.get(period, 365)
                cutoff_date = datetime.now() - timedelta(days=days)
                df = df[df.index >= cutoff_date]
            
            # 缓存数据
            self.cache.set_cached_data(cache_key, df, self.config.cache_ttl_hours, "index_history")
            
            logger.info(f"AkShare指数历史数据获取成功: {symbol}, {len(df)}条记录")
            return df
            
        except Exception as e:
            logger.error(f"获取指数历史数据失败: {symbol} - {e}")
            raise AkShareError(f"获取指数历史数据失败: {e}")
    
    def _format_indices_data(self, df: pd.DataFrame) -> Dict[str, Any]:
        """格式化指数数据"""
        try:
            indices = {}
            
            for _, row in df.iterrows():
                code = row.get('代码', '')
                indices[code] = {
                    "code": code,
                    "name": row.get('名称', ''),
                    "current": row.get('最新价', 0),
                    "change": row.get('涨跌额', 0),
                    "change_percent": row.get('涨跌幅', 0),
                    "volume": row.get('成交量', 0),
                    "turnover": row.get('成交额', 0),
                    "high": row.get('最高', 0),
                    "low": row.get('最低', 0),
                    "open": row.get('今开', 0),
                    "prev_close": row.get('昨收', 0)
                }
            
            return {
                "indices": indices,
                "timestamp": datetime.now().isoformat(),
                "data_source": "AkShare-指数",
                "count": len(indices)
            }
            
        except Exception as e:
            logger.error(f"格式化指数数据失败: {e}")
            return {"error": str(e)}


class AkShareFundFlowAdapter:
    """
    AkShare资金流向数据适配器
    
    @description 提供个股、概念、行业的资金流向分析
    """
    
    def __init__(self, config: AkShareConfig):
        self.config = config
        self.cache = AkShareCacheManager()
    
    @retry_on_failure(max_retries=3, retry_delay=1)
    def get_individual_fund_flow(self, time_period: str = "即时") -> pd.DataFrame:
        """获取个股资金流向"""
        cache_key = f"fund_flow_individual_{time_period}"
        
        # 尝试从缓存获取
        cached_data = self.cache.get_cached_data(cache_key, 0.25)  # 15分钟缓存
        if cached_data is not None:
            return cached_data
        
        try:
            df = ak.stock_fund_flow_individual(symbol=time_period)
            
            # 缓存数据
            self.cache.set_cached_data(cache_key, df, 0.25, "fund_flow")
            
            logger.info(f"AkShare个股资金流向获取成功: {time_period}, {len(df)}条记录")
            return df
            
        except Exception as e:
            logger.error(f"获取个股资金流向失败: {e}")
            raise AkShareError(f"获取个股资金流向失败: {e}")
    
    @retry_on_failure(max_retries=3, retry_delay=1)
    def get_concept_fund_flow(self, time_period: str = "即时") -> pd.DataFrame:
        """获取概念资金流向"""
        cache_key = f"fund_flow_concept_{time_period}"
        
        # 尝试从缓存获取
        cached_data = self.cache.get_cached_data(cache_key, 0.25)  # 15分钟缓存
        if cached_data is not None:
            return cached_data
        
        try:
            df = ak.stock_fund_flow_concept(symbol=time_period)
            
            # 缓存数据
            self.cache.set_cached_data(cache_key, df, 0.25, "fund_flow")
            
            logger.info(f"AkShare概念资金流向获取成功: {time_period}, {len(df)}条记录")
            return df
            
        except Exception as e:
            logger.error(f"获取概念资金流向失败: {e}")
            raise AkShareError(f"获取概念资金流向失败: {e}")


class AkShareFuturesAdapter:
    """
    AkShare期货数据适配器
    
    @description 提供国内外期货实时和历史数据
    """
    
    def __init__(self, config: AkShareConfig):
        self.config = config
        self.cache = AkShareCacheManager()
    
    @retry_on_failure(max_retries=3, retry_delay=1)
    def get_domestic_futures(self) -> pd.DataFrame:
        """
        获取国内期货实时行情
        """
        cache_key = "futures_domestic_realtime"
        # 缓存1分钟
        cached_data = self.cache.get_cached_data(cache_key, max_age_hours=1/60)
        if cached_data is not None:
            return cached_data
        try:
            # 修复：新版akshare.futures_zh_spot()不再需要subscribe_list参数
            shfe_df = ak.futures_zh_spot(market="SHFE") #上海期货交易所
            dce_df = ak.futures_zh_spot(market="DCE")  #大连商品交易所
            czce_df = ak.futures_zh_spot(market="CZCE") #郑州商品交易所
            
            # 合并数据
            all_futures = pd.concat([shfe_df, dce_df, czce_df], ignore_index=True)
            self.cache.set_cached_data(cache_key, all_futures, 1/60, "futures")
            logger.info(f"AkShare国内期货数据获取成功: {len(all_futures)}个合约")
            return all_futures
            
        except Exception as e:
            logger.error(f"获取国内期货数据失败: {e}")
            raise AkShareError(f"获取期货数据失败: {e}")
    
    @retry_on_failure(max_retries=3, retry_delay=1)
    def get_foreign_futures(self) -> pd.DataFrame:
        """
        获取国外主要财经网站的期货实时行情
        """
        cache_key = "futures_foreign_realtime"
        # 缓存1分钟
        cached_data = self.cache.get_cached_data(cache_key, max_age_hours=1/60)
        if cached_data is not None:
            return cached_data
        try:
            # 数据源: 新浪财经
            df = ak.futures_foreign_spot_sina()
            self.cache.set_cached_data(cache_key, df, 1/60, "futures")
            logger.info(f"AkShare国外期货数据获取成功: {len(df)}个合约")
            return df
            
        except Exception as e:
            logger.error(f"获取国外期货数据失败: {e}")
            raise AkShareError(f"获取期货数据失败: {e}")


class AkShareMasterAdapter:
    """
    AkShare主适配器
    
    @description 统一管理所有AkShare数据源，提供完整的数据服务
    """
    
    def __init__(self, config: Optional[AkShareConfig] = None):
        """
        初始化AkShare主适配器
        
        @param {Optional[AkShareConfig]} config - 配置对象
        """
        self.config = config or AkShareConfig()
        
        # 初始化各个子适配器
        self.stock = AkShareStockAdapter(self.config)
        self.index = AkShareIndexAdapter(self.config)
        self.fund_flow = AkShareFundFlowAdapter(self.config)
        self.futures = AkShareFuturesAdapter(self.config)
        
        # 成本统计
        self.monthly_cost = 0.0  # AkShare完全免费
        
        logger.info("AkShare主适配器初始化完成 - 100%免费数据源")
    
    @retry_on_failure(max_retries=2, retry_delay=2)
    def get_market_overview(self) -> Dict[str, Any]:
        """获取市场概览数据"""
        logger.info("开始获取AkShare全市场宏观概览...")
        overview = {}
        errors = {}

        # 使用并行或异步获取可以加速，此处为简化采用串行
        try:
            overview['major_indices'] = self.index.get_major_indices()
        except AkShareError as e:
            errors['major_indices'] = str(e)
            overview['major_indices'] = {}

        try:
            overview['concept_fund_flow'] = self.fund_flow.get_concept_fund_flow("即时").to_dict('records')
        except AkShareError as e:
            errors['concept_fund_flow'] = str(e)
            overview['concept_fund_flow'] = []

        try:
            overview['domestic_futures'] = self.futures.get_domestic_futures().to_dict('records')
        except AkShareError as e:
            errors['domestic_futures'] = str(e)
            overview['domestic_futures'] = []

        if errors:
            logger.warning(f"获取市场概览时发生错误: {errors}")

        logger.info("AkShare全市场宏观概览获取完成。")
        return overview
    
    @retry_on_failure(max_retries=2, retry_delay=1)
    def search_stock(self, query: str) -> List[Dict[str, Any]]:
        """搜索股票"""
        query = query.upper()
        results = []
        
        # 搜索A股
        try:
            df_a = ak.stock_zh_a_spot_em()
            # 简化搜索：同时匹配代码和名称
            matched_a = df_a[df_a['代码'].str.contains(query, na=False) | df_a['名称'].str.contains(query, na=False)]
            for _, row in matched_a.iterrows():
                results.append({'symbol': row['代码'], 'name': row['名称'], 'market': 'A股'})
        except Exception as e:
            logger.warning(f"搜索A股失败: {e}")
            
        # 搜索美股
        try:
            df_us = ak.stock_us_spot_em()
            matched_us = df_us[df_us['代码'].str.contains(query, na=False) | df_us['名称'].str.contains(query, na=False)]
            for _, row in matched_us.iterrows():
                 results.append({'symbol': row['代码'], 'name': row['名称'], 'market': '美股'})
        except Exception as e:
            logger.warning(f"搜索美股失败: {e}")

        # 搜索港股
        try:
            df_hk = ak.stock_hk_spot_em()
            matched_hk = df_hk[df_hk['代码'].str.contains(query, na=False) | df_hk['名称'].str.contains(query, na=False)]
            for _, row in matched_hk.iterrows():
                results.append({'symbol': row['代码'], 'name': row['名称'], 'market': '港股'})
        except Exception as e:
            logger.warning(f"搜索港股失败: {e}")
            
        logger.info(f"AkShare搜索'{query}'完成, 找到{len(results)}个结果。")
        return results
    
    def get_stock_data(self, symbol: str, market: str = "auto") -> Dict[str, Any]:
        """
        获取单只股票的综合数据
        
        @param {str} symbol - 股票代码
        @param {str} market - 市场 ("auto", "a_stock", "hk_stock", "us_stock")
        @returns {Dict[str, Any]} 包含实时和历史数据的字典
        """
        try:
            if market == "auto":
                market = self.stock._detect_market(symbol)
            
            # 修正: 美股需要特殊处理以组合数据
            if market == "us_stock":
                return self.get_us_stock_data(symbol)

            realtime_data = self.stock.get_stock_realtime(symbol, market=market)
            history_data = self.stock.get_stock_history(symbol, market=market)
            
            return {
                "symbol": symbol,
                "market": market,
                "realtime": realtime_data,
                "history": history_data.to_dict('records')
            }
        except Exception as e:
            logger.error(f"获取 {symbol} 的综合数据失败: {e}")
            raise AkShareError(f"获取综合数据失败: {e}")

    @retry_on_failure(max_retries=2, retry_delay=2)
    def get_us_stock_data(self, symbol: str) -> Dict[str, Any]:
        """专门用于获取美股综合数据的方法"""
        logger.info(f"开始为美股 {symbol} 组合AkShare数据...")

        # 1. 获取历史数据
        history_df = self.stock.get_stock_history(symbol, market="us_stock")
        if history_df.empty:
            raise AkShareError(f"无法为 {symbol} 获取历史数据。")

        # 2. 获取公司基本面信息
        try:
            info_df = ak.stock_individual_basic_info_us_xq(symbol=symbol)
            info_dict = dict(zip(info_df['item'], info_df['value']))
        except Exception as e:
            logger.warning(f"获取美股 {symbol} 基本信息失败: {e}")
            info_dict = {}

        # 3. 组合实时数据 (从历史数据中提取)
        latest_hist = history_df.iloc[-1]
        change_val = latest_hist.get("change", 0.0) or 0.0 # 增加空值检查
        realtime_data = {
            "symbol": symbol,
            "name": info_dict.get("org_name_cn", symbol),
            "current_price": latest_hist.get("close"),
            "previous_close": latest_hist.get("close", 0.0) - change_val,
            "open": latest_hist.get("open"),
            "high": latest_hist.get("high"),
            "low": latest_hist.get("low"),
            "volume": latest_hist.get("volume"),
            "market_cap": info_dict.get("total_shares", 0) * latest_hist.get("close", 0),
            "pe_ratio": info_dict.get("pe_lyr"),
            "exchange": "NASDAQ" if ".O" not in info_dict.get("td_mkt", "") else "NYSE",
            "currency": "USD"
        }

        return {
            "symbol": symbol,
            "market": "us_stock",
            "realtime": realtime_data,
            "history": history_df.to_dict('records'),
            "info_from_akshare": info_dict
        }

    def get_cost_analysis(self) -> Dict[str, Any]:
        """获取成本分析"""
        return {
            "provider": "AkShare",
            "cost_per_call": 0.0,
            "monthly_cost_estimate": 0.0,
            "cost_model": "完全免费",
            "notes": "AkShare是中国领先的开源金融数据接口库，提供免费的数据服务。"
        }


# 自定义异常
class AkShareError(Exception):
    """AkShare数据获取异常"""
    pass


# 工厂函数
def create_akshare_adapter(config: Optional[AkShareConfig] = None) -> AkShareMasterAdapter:
    """
    创建AkShare适配器实例
    
    @param {Optional[AkShareConfig]} config - 配置对象
    @returns {AkShareMasterAdapter} AkShare主适配器实例
    """
    if config is None:
        config = AkShareConfig()
    
    logger.info("AkShare主适配器初始化完成 - 100%免费数据源")
    return AkShareMasterAdapter(config=config)


if __name__ == "__main__":
    # 测试代码
    print("=== AkShare数据适配器测试 ===")
    
    # 创建适配器
    adapter = create_akshare_adapter()
    
    # 测试市场概览
    try:
        overview = adapter.get_market_overview()
        print(f"✅ 市场概览测试成功")
        print(f"   指数数量: {len(overview.get('indices', {}))}")
        print(f"   数据源: {overview.get('data_source')}")
        print(f"   成本节省: {overview.get('cost_info', {}).get('cost_savings')}")
    except Exception as e:
        print(f"❌ 市场概览测试失败: {e}")
    
    # 测试股票搜索
    try:
        search_results = adapter.search_stock("平安")
        print(f"✅ 股票搜索测试成功: 找到{len(search_results)}只股票")
    except Exception as e:
        print(f"❌ 股票搜索测试失败: {e}")
    
    # 测试成本分析
    try:
        cost_analysis = adapter.get_cost_analysis()
        print(f"✅ 成本分析测试成功")
        print(f"   月度节省: ${cost_analysis['savings']['monthly']}")
        print(f"   年度节省: ${cost_analysis['savings']['annual']}")
    except Exception as e:
        print(f"❌ 成本分析测试失败: {e}")
    
    print("\n=== AkShare集成完成 ===")
    print("🎉 100%免费数据源，完全替代FinnHub！")