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.
103 lines
3.1 KiB
103 lines
3.1 KiB
#!/usr/bin/env python3
|
|
# 测试TQSDK合约代码格式修复
|
|
|
|
import os
|
|
import sys
|
|
|
|
# 确保能导入项目模块
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from qihuo_analyzer.data.api_adapters.tqsdk_adapter import TqSdkAdapter
|
|
from qihuo_analyzer.data.data_fetcher import DataFetcher
|
|
|
|
def test_tq_symbol_format():
|
|
"""测试TQSDK合约代码格式"""
|
|
print("=== 测试TQSDK合约代码格式修复 ===")
|
|
|
|
# 测试TQSDK适配器的合约代码转换
|
|
print("\n1. 测试TQSDK适配器合约代码转换:")
|
|
adapter = TqSdkAdapter()
|
|
|
|
test_symbols = [
|
|
"TA2603", # PTA
|
|
"CF2603", # 棉花
|
|
"SR2603", # 白糖
|
|
"MA2603", # 甲醇
|
|
"ZC2603", # 动力煤
|
|
"FG2603", # 玻璃
|
|
"RM2603", # 菜籽粕
|
|
"OI2603", # 菜籽油
|
|
"SA2603", # 纯碱
|
|
"LY2603", # 烧碱
|
|
"CU2603", # 铜
|
|
"AL2603", # 铝
|
|
"ZN2603", # 锌
|
|
"PB2603", # 铅
|
|
"NI2603", # 镍
|
|
"SN2603", # 锡
|
|
"AU2603", # 黄金
|
|
"AG2603", # 白银
|
|
"RB2603", # 螺纹钢
|
|
"HC2603", # 热轧卷板
|
|
"BU2603", # 沥青
|
|
"RU2603", # 橡胶
|
|
"FU2603", # 燃油
|
|
"I2603", # 铁矿石
|
|
"J2603", # 焦炭
|
|
"JM2603", # 焦煤
|
|
"A2603", # 大豆
|
|
"B2603", # 豆粕
|
|
"M2603", # 豆粕
|
|
"Y2603", # 豆油
|
|
"P2603", # 棕榈油
|
|
"C2603", # 玉米
|
|
"CS2603", # 玉米淀粉
|
|
"L2603", # 聚乙烯
|
|
"V2603", # 聚氯乙烯
|
|
"PP2603", # 聚丙烯
|
|
]
|
|
|
|
for symbol in test_symbols:
|
|
try:
|
|
tq_symbol = adapter._convert_symbol(symbol)
|
|
print(f"✓ {symbol} → {tq_symbol}")
|
|
except Exception as e:
|
|
print(f"✗ {symbol} → 转换失败: {e}")
|
|
|
|
# 测试DataFetcher获取PTA数据
|
|
print("\n2. 测试DataFetcher获取PTA数据:")
|
|
fetcher = DataFetcher()
|
|
|
|
try:
|
|
# 测试获取PTA K线数据
|
|
print("测试获取 TA2603 的K线数据...")
|
|
kline_data = fetcher.get_kline_data("TA2603", "1h", 10)
|
|
if kline_data is not None:
|
|
print(f"✓ 成功获取 {len(kline_data)} 条PTA K线数据")
|
|
print("数据示例:")
|
|
print(kline_data.head())
|
|
else:
|
|
print("✗ 无法获取PTA K线数据")
|
|
except Exception as e:
|
|
print(f"✗ 获取PTA数据失败: {e}")
|
|
|
|
# 测试获取其他郑州商品交易所数据
|
|
print("\n3. 测试获取其他郑州商品交易所数据:")
|
|
czce_symbols = ["CF2603", "SR2603", "MA2603", "FG2603"]
|
|
|
|
for symbol in czce_symbols:
|
|
try:
|
|
print(f"测试获取 {symbol} 的K线数据...")
|
|
kline_data = fetcher.get_kline_data(symbol, "1h", 5)
|
|
if kline_data is not None:
|
|
print(f"✓ 成功获取 {symbol} 的K线数据")
|
|
else:
|
|
print(f"✗ 无法获取 {symbol} 的K线数据")
|
|
except Exception as e:
|
|
print(f"✗ 获取 {symbol} 数据失败: {e}")
|
|
|
|
print("\n=== 测试完成 ===")
|
|
|
|
if __name__ == "__main__":
|
|
test_tq_symbol_format()
|