aszerW
c905230a40
Archive legacy framework and utility modules that are no longer referenced by the active core (datasource/ and rotation/): - framework/ -> archive/framework/ - framework_v2/ -> archive/framework_v2/ - strategies/ -> archive/strategies/ - config/ -> archive/config/ - visualization/ -> archive/visualization/ - scripts/ -> archive/scripts/ - tests/ -> archive/tests/ - run_rotation.py, run_us_rotation.py -> archive/single_files/ - compare_*.py, test_api_dates.py -> archive/single_files/
115 lines
3.3 KiB
Python
115 lines
3.3 KiB
Python
#!/usr/bin/env python3
|
||
"""
|
||
轮动策略回测入口脚本
|
||
|
||
用法:
|
||
python scripts/run_rotation.py --config strategies/rotation/config.yaml
|
||
python scripts/run_rotation.py --config strategies/rotation/config.yaml --save-path results/report
|
||
"""
|
||
|
||
import sys
|
||
import argparse
|
||
from pathlib import Path
|
||
from datetime import datetime
|
||
|
||
# 添加项目根目录到路径
|
||
project_root = Path(__file__).parent.parent
|
||
sys.path.insert(0, str(project_root))
|
||
|
||
# 加载环境变量
|
||
from dotenv import load_dotenv
|
||
load_dotenv()
|
||
|
||
from strategies.rotation.strategy import RotationStrategy
|
||
|
||
|
||
def load_config(config_path: str) -> dict:
|
||
"""加载配置"""
|
||
import yaml
|
||
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='strategies/rotation/config.yaml',
|
||
help='配置文件路径'
|
||
)
|
||
parser.add_argument(
|
||
'--save-path',
|
||
type=str,
|
||
default=None,
|
||
help='报告保存路径前缀'
|
||
)
|
||
parser.add_argument(
|
||
'--no-api',
|
||
action='store_true',
|
||
help='不使用Flask API，使用本地数据源'
|
||
)
|
||
|
||
args = parser.parse_args()
|
||
|
||
start_time = datetime.now()
|
||
|
||
print("=" * 60)
|
||
print(" ETF轮动策略 回测系统")
|
||
print("=" * 60)
|
||
|
||
# 加载配置
|
||
print(f"\n加载配置: {args.config}")
|
||
config = load_config(args.config)
|
||
|
||
# 显示配置摘要
|
||
code_list = list(config.get('code_list', {}).keys())
|
||
print(f"候选标的: {len(code_list)} 只")
|
||
print(f"回测区间: {config.get('start_date', 'N/A')} ~ {config.get('end_date', 'N/A')}")
|
||
print(f"因子类型: {config.get('factor_type', 'momentum')}")
|
||
print(f"窗口天数: {config.get('n_days', 25)}")
|
||
print(f"选股数量: {config.get('select_num', 3)}")
|
||
print(f"调仓周期: {config.get('rebalance_days', 1)} 天")
|
||
print(f"交易成本: {config.get('trade_cost', 0.001):.2%}")
|
||
|
||
# 初始化策略
|
||
print("\n初始化策略...")
|
||
strategy = RotationStrategy.from_yaml(args.config)
|
||
|
||
# 设置保存路径
|
||
if args.save_path is None:
|
||
report_date = datetime.now().strftime('%Y%m%d')
|
||
args.save_path = f"results/report_{report_date}"
|
||
|
||
# 执行回测
|
||
print("\n" + "=" * 60)
|
||
print("开始回测...")
|
||
print("=" * 60)
|
||
|
||
# 使用Flask API或本地数据源
|
||
use_flask_api = not args.no_api
|
||
data = strategy.get_data(use_flask_api=use_flask_api)
|
||
|
||
result = strategy.run_backtest(data=data, save_path=args.save_path)
|
||
|
||
# 输出结果
|
||
if result.get('result') is not None:
|
||
final_nav = result['result']['策略净值'].iloc[-1]
|
||
total_return = (final_nav - 1) * 100
|
||
|
||
print("\n" + "=" * 60)
|
||
print("回测完成!")
|
||
print("=" * 60)
|
||
print(f"最终净值: {final_nav:.4f}")
|
||
print(f"累计收益: {total_return:.2f}%")
|
||
print(f"调仓次数: {len(result.get('rebalance_events', []))} 次")
|
||
print(f"报告保存: {args.save_path}_*.csv")
|
||
|
||
elapsed = datetime.now() - start_time
|
||
print(f"\n总耗时: {elapsed.total_seconds():.1f}秒")
|
||
|
||
return result
|
||
|
||
|
||
if __name__ == '__main__':
|
||
main() |