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.
124 lines
4.1 KiB
124 lines
4.1 KiB
#!/usr/bin/env python3
|
|
# 测试枚举合约数据
|
|
|
|
import os
|
|
import sys
|
|
|
|
# 确保能导入项目模块
|
|
sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
|
|
|
|
from qihuo_analyzer.data.data_fetcher import DataFetcher
|
|
|
|
def test_enum_symbols():
|
|
"""测试枚举合约数据"""
|
|
print("=== 测试枚举合约数据 ===")
|
|
|
|
# 创建DataFetcher实例
|
|
fetcher = DataFetcher()
|
|
|
|
# 测试获取所有品种列表
|
|
print("\n1. 测试获取所有品种列表:")
|
|
symbols = fetcher.get_all_symbols()
|
|
print(f"成功获取 {len(symbols)} 个品种")
|
|
print("所有品种:", symbols)
|
|
|
|
# 测试获取按交易所划分的品种列表
|
|
print("\n2. 测试获取按交易所划分的品种列表:")
|
|
symbols_by_exchange = fetcher.get_all_symbols_by_exchange()
|
|
for exchange, products in symbols_by_exchange.items():
|
|
print(f"\n交易所: {exchange}")
|
|
print(f"品种数量: {len(products)}")
|
|
for product, contracts in products.items():
|
|
print(f" {product}: {contracts[:3]}... (共{len(contracts)}个合约)")
|
|
|
|
# 测试用户指定的品种是否都存在
|
|
print("\n3. 测试用户指定的品种是否都存在:")
|
|
specified_products = [
|
|
"金", "银", "铜", "镍", "锡", "玻璃", "烧碱", "纯碱", "焦煤",
|
|
"螺纹钢", "氧化铝", "甲醇", "PVC", "燃油", "原油", "铝", "棕榈油",
|
|
"碳酸锂", "工业硅", "橡胶", "合成橡胶", "锌", "20号胶", "多晶硅",
|
|
"中证1000", "中证500", "低硫燃油", "上证50"
|
|
]
|
|
|
|
# 品种中文名称映射
|
|
product_name_map = {
|
|
'AU': '金',
|
|
'AG': '银',
|
|
'CU': '铜',
|
|
'NI': '镍',
|
|
'SN': '锡',
|
|
'FG': '玻璃',
|
|
'LY': '烧碱',
|
|
'SA': '纯碱',
|
|
'JM': '焦煤',
|
|
'RB': '螺纹钢',
|
|
'ALO': '氧化铝',
|
|
'MA': '甲醇',
|
|
'V': 'PVC',
|
|
'FU': '燃油',
|
|
'SC': '原油',
|
|
'AL': '铝',
|
|
'P': '棕榈油',
|
|
'LI': '碳酸锂',
|
|
'SI': '工业硅',
|
|
'RU': '橡胶',
|
|
'BR': '合成橡胶',
|
|
'ZN': '锌',
|
|
'NR': '20号胶',
|
|
'SP': '多晶硅',
|
|
'IM': '中证1000',
|
|
'IC': '中证500',
|
|
'LU': '低硫燃油',
|
|
'IH': '上证50'
|
|
}
|
|
|
|
# 检查每个指定的品种
|
|
found_products = []
|
|
missing_products = []
|
|
|
|
for product_code, product_name in product_name_map.items():
|
|
# 检查是否在模拟品种列表中
|
|
mock_symbols = fetcher._get_mock_all_symbols()
|
|
product_found = any(symbol.startswith(product_code) for symbol in mock_symbols)
|
|
|
|
# 检查是否在按交易所划分的列表中
|
|
exchange_found = False
|
|
for exchange, products in symbols_by_exchange.items():
|
|
if product_code in products:
|
|
exchange_found = True
|
|
break
|
|
|
|
if product_found or exchange_found:
|
|
found_products.append(product_name)
|
|
print(f"✓ 找到: {product_name} ({product_code})")
|
|
else:
|
|
missing_products.append(product_name)
|
|
print(f"✗ 缺失: {product_name} ({product_code})")
|
|
|
|
print(f"\n4. 检查结果:")
|
|
print(f"找到的品种: {len(found_products)}/{len(specified_products)}")
|
|
print(f"缺失的品种: {len(missing_products)}/{len(specified_products)}")
|
|
|
|
if missing_products:
|
|
print(f"缺失的品种: {missing_products}")
|
|
else:
|
|
print("✓ 所有用户指定的品种都已找到！")
|
|
|
|
# 测试获取K线数据
|
|
print("\n5. 测试获取K线数据:")
|
|
test_symbols = ['AU2603', 'AG2603', 'CU2603', 'NI2603', 'SI2603']
|
|
for symbol in test_symbols:
|
|
try:
|
|
data = fetcher.get_kline_data(symbol, "1h", 10)
|
|
if data is not None:
|
|
print(f"✓ 成功获取 {symbol} 的K线数据 ({len(data)}条)")
|
|
else:
|
|
print(f"✗ 无法获取 {symbol} 的K线数据")
|
|
except Exception as e:
|
|
print(f"✗ 获取 {symbol} 数据失败: {e}")
|
|
|
|
print("\n=== 测试完成 ===")
|
|
|
|
if __name__ == "__main__":
|
|
test_enum_symbols()
|