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.
91 lines
2.8 KiB
91 lines
2.8 KiB
#!/usr/bin/env python3
|
|
# 测试铁矿石合约代码修复
|
|
|
|
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_iron_ore_contract():
|
|
"""测试铁矿石合约代码处理"""
|
|
print("=== 测试铁矿石合约代码修复 ===")
|
|
|
|
# 测试TQSDK适配器的合约代码转换
|
|
print("\n1. 测试TQSDK适配器合约代码转换:")
|
|
adapter = TqSdkAdapter()
|
|
|
|
test_symbols = [
|
|
"I2603", # 铁矿石
|
|
"J2603", # 焦炭
|
|
"JM2603", # 焦煤
|
|
"CU2603", # 铜
|
|
"AL2603", # 铝
|
|
"AU2603", # 黄金
|
|
"AG2603", # 白银
|
|
"RB2603", # 螺纹钢
|
|
"P2603", # 棕榈油
|
|
"V2603", # PVC
|
|
"MA2603", # 甲醇
|
|
"FG2603", # 玻璃
|
|
"SA2603", # 纯碱
|
|
"LY2603", # 烧碱
|
|
"LI2603", # 碳酸锂
|
|
"SI2603", # 工业硅
|
|
"SP2603", # 多晶硅
|
|
"RU2603", # 橡胶
|
|
"NR2603", # 20号胶
|
|
"FU2603", # 燃油
|
|
"SC2603", # 原油
|
|
"LU2603", # 低硫燃油
|
|
"ALO2603",# 氧化铝
|
|
"BR2603", # 合成橡胶
|
|
]
|
|
|
|
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获取铁矿石数据
|
|
print("\n2. 测试DataFetcher获取铁矿石数据:")
|
|
fetcher = DataFetcher()
|
|
|
|
try:
|
|
# 测试获取铁矿石K线数据
|
|
print("测试获取 I2603 的K线数据...")
|
|
kline_data = fetcher.get_kline_data("I2603", "1h", 10)
|
|
if kline_data is not None:
|
|
print(f"✓ 成功获取 {len(kline_data)} 条铁矿石K线数据")
|
|
print("数据示例:")
|
|
print(kline_data.head())
|
|
else:
|
|
print("✗ 无法获取铁矿石K线数据")
|
|
except Exception as e:
|
|
print(f"✗ 获取铁矿石数据失败: {e}")
|
|
|
|
# 测试其他单字符品种
|
|
print("\n3. 测试其他单字符品种:")
|
|
single_char_symbols = ["I2603", "J2603", "P2603", "C2603", "L2603"]
|
|
|
|
for symbol in single_char_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_iron_ore_contract()
|