AlphaFutures/test_data_adapters.py

#!/usr/bin/env python3
# 测试数据适配器

from qihuo_analyzer.data.api_adapters import DataAdapterFactory


def test_adapter(adapter_type):
    """测试数据适配器
    
    Args:
        adapter_type: 适配器类型，可选值：'tqsdk', 'rqdata'
    """
    print(f"\n=== 测试 {adapter_type} 适配器 ===")
    
    # 创建适配器
    adapter = DataAdapterFactory.create_adapter(adapter_type)
    
    # 连接API
    print("连接API...")
    connected = adapter.connect()
    print(f"连接结果: {'成功' if connected else '失败'}")
    
    if connected:
        # 测试获取K线数据
        print("\n测试获取K线数据...")
        kline_data = adapter.get_kline_data('CU2603', '1d', 10)
        if kline_data is not None:
            print(f"成功获取K线数据，数据长度: {len(kline_data)}")
            print(kline_data.head())
        else:
            print("获取K线数据失败")
        
        # 测试获取Tick数据
        print("\n测试获取Tick数据...")
        tick_data = adapter.get_tick_data('CU2603', 100)
        if tick_data is not None:
            print(f"成功获取Tick数据，数据长度: {len(tick_data)}")
            print(tick_data.head())
        else:
            print("获取Tick数据失败")
        
        # 测试获取合约信息
        print("\n测试获取合约信息...")
        contract_info = adapter.get_contract_info('CU2603')
        if contract_info is not None:
            print("成功获取合约信息:")
            for key, value in contract_info.items():
                print(f"{key}: {value}")
        else:
            print("获取合约信息失败")
        
        # 测试批量获取市场数据
        print("\n测试批量获取市场数据...")
        market_data = adapter.get_market_data(['CU2603', 'AL2603'])
        if market_data:
            print("成功获取市场数据:")
            for symbol, data in market_data.items():
                print(f"{symbol}: {data}")
        else:
            print("获取市场数据失败")
        
        # 测试获取所有品种列表
        print("\n测试获取所有品种列表...")
        symbols = adapter.get_all_symbols()
        if symbols:
            print(f"成功获取品种列表，共{len(symbols)}个品种")
            print(f"前10个品种: {symbols[:10]}")
        else:
            print("获取品种列表失败")
        
        # 断开连接
        print("\n断开连接...")
        adapter.disconnect()
    else:
        print("API连接失败，跳过测试")


if __name__ == "__main__":
    # 测试TQSDK适配器
    test_adapter('tqsdk')
    
    # 测试RQData适配器
    test_adapter('rqdata')
    
    print("\n=== 测试完成 ===")