fix: 回测细节导出、交易日历测试和动量因子修复

修复项:
- export_backtest_detail.py: 统一回测导出脚本的数据源调用逻辑
- test_trading_calendar.py: 交易日历功能测试
- verify_fix_result.py: 修复结果验证
- verify_mode_b.py: 模式 B 验证

策略修复:
- momentum.py: 动量因子计算优化
- strategy.py: StrategyBase 数据获取修复(fetch_indices 返回 dict)
This commit is contained in:
2026-05-24 14:26:35 +08:00
parent 0954458114
commit 5212b004dc
5 changed files with 999 additions and 0 deletions

View File

@@ -0,0 +1,164 @@
#!/usr/bin/env python3
"""
测试交易日历 API
"""
import sys
from pathlib import Path
import requests
# Flask 服务地址
FLASK_API_URL = "http://localhost:80"
def test_calendar_api():
"""测试交易日历 API"""
print("\n" + "="*80)
print("📅 交易日历 API 测试")
print("="*80)
# 测试 1: A 股
print("\n[1] 测试 A 股交易日历...")
url = f"{FLASK_API_URL}/api/v1/trading-calendar"
params = {"market": "A", "start": "2024-01-01", "end": "2024-01-31"}
try:
response = requests.get(url, params=params, timeout=10)
if response.status_code == 200:
data = response.json()
print(f" ✅ 成功: {data['count']} 个交易日")
print(f" 市场: {data['market']}")
print(f" 交易所: {data['exchange']}")
print(f" 日期范围: {data['start']} ~ {data['end']}")
print(f" 前5个交易日: {data['trading_dates'][:5]}")
else:
print(f" ❌ 失败: {response.status_code}")
print(f" 响应: {response.json()}")
except Exception as e:
print(f" ❌ 异常: {e}")
# 测试 2: 美股
print("\n[2] 测试美股交易日历...")
params = {"market": "US", "start": "2024-01-01", "end": "2024-01-31"}
try:
response = requests.get(url, params=params, timeout=10)
if response.status_code == 200:
data = response.json()
print(f" ✅ 成功: {data['count']} 个交易日")
print(f" 市场: {data['market']}")
print(f" 交易所: {data['exchange']}")
print(f" 前5个交易日: {data['trading_dates'][:5]}")
else:
print(f" ❌ 失败: {response.status_code}")
print(f" 响应: {response.json()}")
except Exception as e:
print(f" ❌ 异常: {e}")
# 测试 3: 港股
print("\n[3] 测试港股交易日历...")
params = {"market": "HK", "start": "2024-01-01", "end": "2024-01-31"}
try:
response = requests.get(url, params=params, timeout=10)
if response.status_code == 200:
data = response.json()
print(f" ✅ 成功: {data['count']} 个交易日")
print(f" 市场: {data['market']}")
print(f" 交易所: {data['exchange']}")
print(f" 前5个交易日: {data['trading_dates'][:5]}")
else:
print(f" ❌ 失败: {response.status_code}")
print(f" 响应: {response.json()}")
except Exception as e:
print(f" ❌ 异常: {e}")
# 测试 4: 日历信息
print("\n[4] 测试日历信息...")
url_info = f"{FLASK_API_URL}/api/v1/calendar/info"
try:
response = requests.get(url_info, timeout=10)
if response.status_code == 200:
data = response.json()
print(f" ✅ 成功")
print(f" 支持的市场:")
for market, info in data.get('supported_markets', {}).items():
print(f" {market}: {info['name']} ({info['method']})")
print(f" pandas_market_calendars: {'✅ 已安装' if data.get('pandas_market_calendars_installed') else '❌ 未安装'}")
else:
print(f" ❌ 失败: {response.status_code}")
except Exception as e:
print(f" ❌ 异常: {e}")
def test_local_fetcher():
"""测试本地 UniversalDataFetcher"""
print("\n" + "="*80)
print("🧪 本地 UniversalDataFetcher 测试")
print("="*80)
sys.path.insert(0, str(Path(__file__).parent.parent))
try:
from datasource.universal_fetcher import UniversalDataFetcher
fetcher = UniversalDataFetcher()
# 测试 A 股
print("\n[1] A 股交易日历 (2024年)...")
cal_a = fetcher.get_trading_calendar('A', '2024-01-01', '2024-12-31')
print(f"{len(cal_a)} 个交易日")
print(f" 前5天: {list(cal_a[:5])}")
# 测试美股
print("\n[2] 美股交易日历 (2024年)...")
cal_us = fetcher.get_trading_calendar('US', '2024-01-01', '2024-12-31')
print(f"{len(cal_us)} 个交易日")
print(f" 前5天: {list(cal_us[:5])}")
# 测试港股
print("\n[3] 港股交易日历 (2024年)...")
cal_hk = fetcher.get_trading_calendar('HK', '2024-01-01', '2024-12-31')
print(f"{len(cal_hk)} 个交易日")
print(f" 前5天: {list(cal_hk[:5])}")
# 日历信息
print("\n[4] 日历支持信息...")
info = fetcher.get_calendar_info()
print(f" ✅ 支持 {len(info['supported_markets'])} 个市场")
except Exception as e:
print(f" ❌ 失败: {e}")
import traceback
traceback.print_exc()
def main():
print("\n" + "="*80)
print("📅 交易日历功能测试")
print("="*80)
# 测试 1: 本地 fetcher
test_local_fetcher()
# 测试 2: Flask API如果服务在运行
print("\n" + "="*80)
print("🌐 测试 Flask API 端点")
print("="*80)
print(f"\nAPI 地址: {FLASK_API_URL}")
print("注意: 需要 Flask 服务正在运行")
try:
response = requests.get(f"{FLASK_API_URL}/health", timeout=3)
if response.status_code == 200:
print("✅ Flask 服务可访问")
test_calendar_api()
else:
print(f"⚠️ Flask 服务返回 {response.status_code},跳过 API 测试")
except:
print("⚠️ Flask 服务未运行,跳过 API 测试")
print("\n" + "="*80)
print("✅ 测试完成")
print("="*80)
if __name__ == "__main__":
main()