如何把akshare行情无缝切换miniqmt数据源

from xtquant import xtdata
import pandas as pd
import time


def stock_zh_a_hist_miniqmt(symbol, start_date, end_date, adjust=""):
    """
    正确地将MiniQMT数据转换为AkShare格式的DataFrame。
    """
    # 1. 转换股票代码格式
    if symbol.startswith(('0', '3')):
        xt_code = f"{symbol}.SZ"
    elif symbol.startswith('6'):
        xt_code = f"{symbol}.SH"
    elif symbol.startswith(('9', '8')):
        xt_code = f"{symbol}.BJ"
    else:
        print(f"无法识别的股票代码: {symbol}")
        return pd.DataFrame()

    # 2. 设置复权类型
    dividend_type = "none"
    if adjust == "qfq":
        dividend_type = "front"
    elif adjust == "hfq":
        dividend_type = "back"

    try:
        # 3. 下载数据
        print(f"开始下载 {symbol} 的历史数据...")
        xtdata.download_history_data(xt_code, '1d', start_date, end_date)

        # 等待数据下载完成
        time.sleep(3)

        # 4. 获取市场数据
        print(f"获取 {symbol} 的市场数据...")
        market_data_dict = xtdata.get_market_data_ex(
            field_list=['open', 'high', 'low', 'close', 'volume', 'amount', 'time'],  # 明确指定需要的字段
            stock_list=[xt_code],
            period='1d',
            start_time=start_date,
            end_time=end_date,
            dividend_type=dividend_type,
            count=-1
        )

        print("获取到的数据字典结构:", list(market_data_dict.keys()) if market_data_dict else "空")

        
        if market_data_dict is None or xt_code not in market_data_dict:
            print(f"未获取到 {symbol} 的数据。")
            return pd.DataFrame()

        # 提取该股票对应的数据
        stock_data = market_data_dict[xt_code]
        #print("数据列名:", stock_data.columns.tolist() if hasattr(stock_data, 'columns') else "非DataFrame格式")

        # 5. 创建新的DataFrame并处理时间列
        result_data = {}

        # 字段映射
        field_mapping = {
            'time': '日期',
            'open': '开盘',
            'high': '最高',
            'low': '最低',
            'close': '收盘',
            'volume': '成交量',
            'amount': '成交额'
        }

        # 复制需要的字段
        for xt_field, ak_field in field_mapping.items():
            if hasattr(stock_data, 'columns') and xt_field in stock_data.columns:
                result_data[ak_field] = stock_data[xt_field].values
            elif isinstance(stock_data, dict) and xt_field in stock_data:
                result_data[ak_field] = stock_data[xt_field]

        # 创建DataFrame
        df = pd.DataFrame(result_data)

        if df.empty:
            print("转换后的DataFrame为空")
            return df

        print("转换前的时间数据样例:", df['日期'].head(3).tolist() if '日期' in df.columns else "无日期列")

        # 6. 处理时间格式
        if '日期' in df.columns:
            if '日期' in df.columns:
                # 方案A：如果时间是数值时间戳（毫秒）
                if df['日期'].dtype in [int, float]:
                    # 尝试毫秒时间戳转换
                    df['日期'] = pd.to_datetime(df['日期'], unit='ms')
                    df['日期'] = pd.to_datetime(df['日期']).dt.strftime('%Y%m%d')


        # 7. 添加股票代码列
        df['股票代码'] = symbol

        # 8. 按日期排序
        if '日期' in df.columns:
            df = df.sort_values('日期').reset_index(drop=True)

        return df

    except Exception as e:
        print(f"处理股票 {symbol} 时发生错误: {e}")
        import traceback
        traceback.print_exc()
        return pd.DataFrame()


# 使用示例和测试
if __name__ == "__main__":
    # 测试不同的股票和日期
    test_cases = [
        ('000001', '20250101', '20250930', ""),  # 平安银行
        ('600519', '20250101', '20250930', "qfq"),  # 贵州茅台，前复权
    ]

    for symbol, start, end, adj in test_cases:
        print(f"\n正在测试 {symbol}，时间段: {start} 到 {end}，复权: {adj}")
        df = stock_zh_a_hist_miniqmt(symbol, start, end, adjust=adj)
        print(df.head(10))