aszerW
0954458114
配置文件: - rotation_global.yaml: 扁平化资产池配置示例，演示 group 策略分组 * 13 个标的覆盖 7 个策略分组（US_TECH, CN_GROWTH, JP_BROAD, EU_BROAD, HK_TECH, COMMODITY, FIXED_INCOME） * signal_source/trade_source 分离配置（跨市场场景） * 分散化选股配置示例（注释状态） * 默认使用 Flask API 数据源 测试用例: - test_flat_asset_pool.py: 7/7 测试通过 * 扁平配置加载验证 * 策略分组功能测试（by_group, groups, count） * 信号/交易标的获取（get_signal_codes, get_trade_codes） * 信号→交易映射（get_signal_to_trade_mapping） * 分散化配置验证 * 标的配置详情验证 - test_config.py: 配置加载器测试 - test_simple_rotation.py: 简单轮动策略端到端测试
129 lines
3.9 KiB
Python
129 lines
3.9 KiB
Python
"""
|
|
测试简单轮动策略
|
|
|
|
验证完整流程：
|
|
1. 配置加载
|
|
2. 策略初始化
|
|
3. 数据获取
|
|
4. 因子计算
|
|
5. 信号生成
|
|
6. 回测执行
|
|
"""
|
|
|
|
import sys
|
|
from pathlib import Path
|
|
import os
|
|
|
|
# 添加项目根目录到路径
|
|
project_root = Path(__file__).parent.parent
|
|
if str(project_root) not in sys.path:
|
|
sys.path.insert(0, str(project_root))
|
|
|
|
from framework_v2.config import load_config
|
|
from framework_v2.strategies.rotation.simple import SimpleRotationStrategy
|
|
|
|
|
|
def test_simple_rotation():
|
|
"""测试简单轮动策略完整流程"""
|
|
|
|
print("\n" + "=" * 70)
|
|
print(" 简单轮动策略端到端测试")
|
|
print("=" * 70)
|
|
|
|
# 设置环境变量
|
|
os.environ['FLASK_API_URL'] = 'https://k3s.tokenpluse.xyz'
|
|
|
|
# 1. 加载配置
|
|
print("\n[1/6] 加载配置...")
|
|
config_path = Path(__file__).parent.parent / 'strategies' / 'rotation' / 'config_simple.yaml'
|
|
config = load_config(str(config_path))
|
|
print(f" ✓ 配置加载成功")
|
|
print(f" 策略: {config.metadata.strategy}")
|
|
print(f" 标的: {list(config.asset_pools.equity.keys())}")
|
|
print(f" 回测: {config.backtest.start_date} ~ {config.backtest.end_date}")
|
|
|
|
# 2. 初始化策略
|
|
print("\n[2/6] 初始化策略...")
|
|
strategy = SimpleRotationStrategy(config)
|
|
print(f" ✓ 策略初始化成功")
|
|
print(f" 名称: {strategy.name}")
|
|
print(f" 动量窗口: {config.factor.n_days} 天")
|
|
print(f" 选股数量: {strategy.select_num}")
|
|
|
|
# 3. 获取数据
|
|
print("\n[3/6] 获取数据...")
|
|
codes = strategy.get_codes()
|
|
print(f" 标的列表: {codes}")
|
|
|
|
data = strategy.get_data()
|
|
print(f" ✓ 获取 {len(data)} 个标的")
|
|
for code, df in data.items():
|
|
print(f" {code}: {len(df)} 天 ({df.index[0].date()} ~ {df.index[-1].date()})")
|
|
|
|
# 4. 计算因子
|
|
print("\n[4/6] 计算因子...")
|
|
factors = strategy.compute_factors(data)
|
|
print(f" ✓ 计算 {len(factors)} 个因子")
|
|
for code, factor in factors.items():
|
|
print(f" {code}: {len(factor)} 值, 范围 [{factor.min():.4f}, {factor.max():.4f}]")
|
|
|
|
# 5. 生成信号
|
|
print("\n[5/6] 生成信号...")
|
|
signals = strategy.generate_signals(factors)
|
|
n_signals = signals.sum().sum()
|
|
print(f" ✓ 生成 {signals.shape[0]} 个交易日信号")
|
|
print(f" 总信号数: {n_signals}")
|
|
print(f" 平均每日持仓: {signals.mean().mean():.2%}")
|
|
|
|
# 6. 仓位管理
|
|
print("\n[6/6] 仓位管理...")
|
|
positions = strategy.manage_positions(signals)
|
|
print(f" ✓ 仓位分配完成")
|
|
print(f" 权重和: {positions.sum(axis=1).mean():.2%}")
|
|
|
|
# 7. 执行回测
|
|
print("\n执行回测...")
|
|
result = strategy._execute_backtest(positions, data)
|
|
|
|
# 打印结果
|
|
print("\n" + "=" * 70)
|
|
print(" 回测结果")
|
|
print("=" * 70)
|
|
|
|
metrics = result['metrics']
|
|
print(f"\n 总收益率: {metrics['total_return']:.2%}")
|
|
print(f" 年化收益: {metrics['annual_return']:.2%}")
|
|
print(f" 最大回撤: {metrics['max_drawdown']:.2%}")
|
|
print(f" 夏普比率: {metrics['sharpe_ratio']:.2f}")
|
|
print(f" 交易天数: {metrics['n_days']}")
|
|
|
|
# 验证结果
|
|
print("\n" + "=" * 70)
|
|
print(" 验证")
|
|
print("=" * 70)
|
|
|
|
assert metrics['total_return'] != 0, "总收益率不应为 0"
|
|
print(" ✓ 总收益率有效")
|
|
|
|
assert len(result['equity_curve']) > 0, "净值曲线不应为空"
|
|
print(" ✓ 净值曲线有效")
|
|
|
|
assert positions.sum(axis=1).max() <= 1.01, "权重和不应超过 100%"
|
|
print(" ✓ 仓位权重有效")
|
|
|
|
print("\n" + "=" * 70)
|
|
print(" ✓ 所有测试通过")
|
|
print("=" * 70 + "\n")
|
|
|
|
return result
|
|
|
|
|
|
if __name__ == "__main__":
|
|
try:
|
|
result = test_simple_rotation()
|
|
except Exception as e:
|
|
print(f"\n✗ 测试失败: {e}")
|
|
import traceback
|
|
traceback.print_exc()
|
|
sys.exit(1)
|