etf/archive/legacy_scripts/scripts/run_rotation.py

#!/usr/bin/env python3
"""
ETF轮动策略回测入口

用法:
    python scripts/run_rotation.py
    python scripts/run_rotation.py --config config/strategies/rotation.yaml
"""

import sys
import time
import yaml
import argparse
from pathlib import Path

# 添加项目根目录到路径
project_root = Path(__file__).parent.parent
sys.path.insert(0, str(project_root))

from strategies.rotation.engine import RotationStrategy
from strategies.rotation.portfolio import track_positions, save_trades
from strategies.rotation.report import generate_performance_report
from config.settings import DEFAULT_CODE_NAME_MAP, DEFAULT_BENCHMARK_NAME


def load_config(config_path: str) -> dict:
    """加载配置文件"""
    with open(config_path, "r", encoding="utf-8") as f:
        return yaml.safe_load(f)


def main():
    parser = argparse.ArgumentParser(description="ETF轮动策略回测")
    parser.add_argument(
        "--config",
        type=str,
        default="config/strategies/rotation.yaml",
        help="配置文件路径",
    )
    parser.add_argument(
        "--save-path",
        type=str,
        default="results/report",
        help="报告保存路径前缀",
    )
    args = parser.parse_args()

    start_time = time.time()

    print("=" * 60)
    print("        ETF轮动策略 回测系统")
    print("=" * 60)

    # 加载配置
    config = load_config(args.config)

    # 如果未设置 end_date，默认使用最新日期
    if not config.get('end_date'):
        from datetime import datetime
        config['end_date'] = datetime.now().strftime('%Y-%m-%d')

    # 从配置中读取 code_list（新的配置格式：{代码: {name, etf, market}}）
    code_list_config = config.get('code_list', {})
    
    # 提取代码列表和名称映射
    if isinstance(code_list_config, dict):
        code_list = list(code_list_config.keys())
        # 构建 code_name_map: {代码: 名称}
        code_name_map = {}
        for code, cfg in code_list_config.items():
            if isinstance(cfg, dict):
                code_name_map[code] = cfg.get('name', code)
            else:
                # 兼容旧格式
                code_name_map[code] = cfg
    else:
        # 兼容旧格式（列表）
        code_list = code_list_config
        code_name_map = DEFAULT_CODE_NAME_MAP
        code_list_config = {}

    benchmark_config = config.get('benchmark', {})
    benchmark_name = benchmark_config.get('name', DEFAULT_BENCHMARK_NAME)

    print(f"\n配置文件: {args.config}")
    print(f"候选标的: {len(code_list)} 只")
    
    # 统计ETF映射情况
    etf_count = sum(1 for cfg in code_list_config.values() if isinstance(cfg, dict) and cfg.get('etf'))
    crypto_count = sum(1 for cfg in code_list_config.values() if isinstance(cfg, dict) and cfg.get('market') == 'CRYPTO')
    print(f"  - ETF映射: {etf_count} 只")
    print(f"  - 直接交易: {crypto_count} 只（加密货币）")
    
    print(f"回测区间: {config['start_date']} ~ {config['end_date']}")
    print(f"因子类型: {config['factor_type']}")
    print(f"窗口天数: {config['n_days']}")
    print(f"选中数量: {config['select_num']}")
    print(f"调仓周期: {config['rebalance_days']} 天")
    print(f"交易成本: {config['trade_cost']:.2%}")

    # 保持 config 中的 code_list 为完整配置格式（用于引擎内部解析）
    # 不需要修改 config['code_list']，引擎会直接使用原始配置

    # 创建策略实例
    strategy = RotationStrategy(config)

    # 运行回测
    print("\n" + "=" * 60)
    print("开始回测...")
    print("=" * 60)

    backtest_result = strategy.run()

    # 持仓跟踪
    print("\n" + "=" * 60)
    print("持仓跟踪...")
    print("=" * 60)

    trades_df, summary_df = track_positions(
        backtest_result,
        code_name_map=code_name_map,
        select_num=config["select_num"],
    )
    save_trades(trades_df, summary_df, save_path=args.save_path)

    # 生成绩效报告
    print("\n" + "=" * 60)
    print("生成绩效报告...")
    print("=" * 60)

    metrics = generate_performance_report(
        backtest_result,
        strategy.valid_codes,
        code_name_map=code_name_map,
        benchmark_name=benchmark_name,
        save_path=args.save_path,
        select_num=config["select_num"],
        code_config=code_list_config,  # 传入完整配置以显示ETF映射
        index_data=strategy.index_data,      # 传入指数数据
        etf_price_data=strategy.etf_data,    # 传入ETF价格数据
        etf_nav_data_raw=strategy.etf_nav_data,  # 传入ETF净值数据
    )

    elapsed = time.time() - start_time
    print(f"\n总耗时: {elapsed:.1f}秒")

    return metrics


if __name__ == "__main__":
    main()