AlphaFutures/test_config_switch.py

#!/usr/bin/env python3
# 测试通过配置文件切换数据源

import os
import sys

# 确保能导入项目模块
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))

# 从配置文件加载配置
from config import (
    DATA_ADAPTER_TYPE,
    TQSDK_USERNAME,
    TQSDK_PASSWORD,
    RQDATA_USERNAME,
    RQDATA_PASSWORD,
    DEBUG
)

# 设置环境变量
def setup_environment():
    """根据配置文件设置环境变量"""
    print("=== 设置环境变量 ===")
    
    # 设置数据源类型
    os.environ["DATA_ADAPTER_TYPE"] = DATA_ADAPTER_TYPE
    print(f"数据源类型: {DATA_ADAPTER_TYPE}")
    
    # 设置TQSDK账号
    if TQSDK_USERNAME:
        os.environ["TQSDK_USERNAME"] = TQSDK_USERNAME
        os.environ["TQSDK_PASSWORD"] = TQSDK_PASSWORD
        print("TQSDK账号: 已配置")
    else:
        print("TQSDK账号: 未配置（将使用模拟数据）")
    
    # 设置RQData账号
    if RQDATA_USERNAME:
        os.environ["RQDATA_USERNAME"] = RQDATA_USERNAME
        os.environ["RQDATA_PASSWORD"] = RQDATA_PASSWORD
        print("RQData账号: 已配置")
    else:
        print("RQData账号: 未配置（将使用模拟数据）")
    
    # 设置调试模式
    if DEBUG:
        os.environ["DEBUG"] = "True"
        print("调试模式: 开启")
    else:
        print("调试模式: 关闭")
    
    print("环境变量设置完成！\n")

# 测试数据源连接
def test_data_source():
    """测试数据源连接和基本功能"""
    print("=== 测试数据源功能 ===")
    
    try:
        from qihuo_analyzer.data.data_fetcher import DataFetcher
        
        # 创建数据获取器实例
        print("创建DataFetcher实例...")
        fetcher = DataFetcher()
        
        # 查看使用的适配器
        adapter_name = fetcher.adapter.__class__.__name__
        print(f"当前使用的适配器: {adapter_name}")
        
        # 测试获取品种列表
        print("\n测试获取品种列表...")
        symbols = fetcher.get_all_symbols()
        if symbols:
            print(f"成功获取 {len(symbols)} 个品种")
            print("前10个品种:", symbols[:10])
        else:
            print("获取品种列表失败")
        
        # 测试获取K线数据
        print("\n测试获取K线数据...")
        test_symbol = symbols[0] if symbols else "CU2603"
        kline_data = fetcher.get_kline_data(
            symbol=test_symbol,
            duration="1h",  # 1小时
            count=50
        )
        
        if kline_data is not None:
            print(f"成功获取 {len(kline_data)} 条K线数据")
            print("数据示例:")
            print(kline_data.head())
        else:
            print("获取K线数据失败")
        
        print("\n=== 测试完成 ===")
        return True
        
    except Exception as e:
        print(f"测试失败: {e}")
        import traceback
        traceback.print_exc()
        return False

# 主函数
def main():
    """主测试函数"""
    print("开始测试通过配置文件切换数据源...\n")
    
    # 1. 设置环境变量
    setup_environment()
    
    # 2. 测试数据源
    success = test_data_source()
    
    if success:
        print("\n✓ 数据源切换测试成功！")
        print(f"✓ 当前使用的数据源类型: {DATA_ADAPTER_TYPE}")
    else:
        print("\n✗ 数据源切换测试失败！")

if __name__ == "__main__":
    main()