You can not select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
123 lines
3.4 KiB
123 lines
3.4 KiB
#!/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()
|