""" 全球市场ETF轮动策略 - 本地回测版本 原始策略来源：聚宽 https://www.joinquant.com/post/1399 核心逻辑（与动量策略共用）： 1. 加权线性回归（权重1→2递增）计算趋势得分 2. score = 年化收益率 × R² 3. ATR动态调整回看窗口（20~60天） 4. 崩盘过滤：连续3天任一天跌>5%则得分归零 5. 溢价过滤：溢价率≥5%则降权 6. 全仓单一品种轮动 ETF池：全球化配置 纳指100 / 日经225 / 德国DAX / 黄金 / 有色金属 / 南方原油 / 30年国债 / 红利低波 / 创业板 """ import sys import math import warnings from pathlib import Path from datetime import datetime import numpy as np import pandas as pd warnings.filterwarnings("ignore") # 添加项目根目录 sys.path.insert(0, str(Path(__file__).parent)) from dotenv import load_dotenv load_dotenv() # ==================== 策略配置 ==================== CONFIG = { # 全球市场ETF池（聚宽代码 -> Tushare代码映射） 'etf_pool': { '513100.SH': '纳指100ETF', '513520.SH': '日经225ETF', '513030.SH': '德国DAX ETF', '518880.SH': '黄金ETF华安', '159980.SZ': '有色金属ETF', '501018.SH': '南方原油LOF', '511090.SH': '30年国债ETF', '512890.SH': '红利低波ETF', '159915.SZ': '创业板ETF易方达', }, 'target_num': 1, # 持仓数量 'auto_day': True, # 是否启用动态周期 'fixed_days': 25, # 固定回看天数 'min_days': 20, # 动态周期最小值 'max_days': 60, # 动态周期最大值 'premium_threshold': 5.0, # 溢价率阈值(%) 'trade_cost': 0.001, # 单次交易成本(双边) 'start_date': '2019-01-01', 'benchmark': '000300.SH', # 基准：沪深300 } # ==================== 复用动量策略核心模块 ==================== from 动量 import ( fetch_all_etf_data, fetch_etf_nav_data, calc_atr, calc_weighted_momentum_score, apply_crash_filter, calc_premium_rate, print_performance, print_yearly_returns, ) # ==================== 回测引擎 ==================== def run_backtest(config: dict): """执行回测""" end_date = datetime.now().strftime('%Y-%m-%d') etf_pool = config['etf_pool'] etf_codes = list(etf_pool.keys()) print("=" * 60) print(" 全球市场ETF轮动策略 - 本地回测") print("=" * 60) print(f" 候选ETF: {len(etf_codes)} 只") for code, name in etf_pool.items(): print(f" {code} {name}") print(f" 持仓数量: {config['target_num']}") print(f" 动态周期: {'开启' if config['auto_day'] else '关闭'}") if config['auto_day']: print(f" 回看范围: {config['min_days']}~{config['max_days']} 天") else: print(f" 固定回看: {config['fixed_days']} 天") print(f" 回测区间: {config['start_date']} ~ {end_date}") # 1. 获取数据 print(f"\n{'='*60}") print("下载ETF价格数据...") all_data = fetch_all_etf_data(etf_codes, config['start_date'], end_date, etf_pool) print("\n下载ETF净值数据...") nav_data = fetch_etf_nav_data(etf_codes, config['start_date'], end_date) print(f" 净值数据: {len(nav_data)} 只") if not all_data: print("无数据，退出") return # 2. 构建交易日历 all_dates = set() for df in all_data.values(): all_dates.update(df.index.tolist()) trade_dates = sorted(all_dates) trade_dates = [d for d in trade_dates if d >= pd.Timestamp(config['start_date'])] print(f"\n交易日数: {len(trade_dates)}") print(f"区间: {trade_dates[0].strftime('%Y-%m-%d')} ~ {trade_dates[-1].strftime('%Y-%m-%d')}") # 3. 逐日回测 print(f"\n{'='*60}") print("开始回测...") print("=" * 60) max_lookback = config['max_days'] + 10 holding = None daily_returns = [] signals = [] for i, today in enumerate(trade_dates): # 计算每只ETF的得分 scores = {} score_details = {} for code in etf_codes: if code not in all_data: continue df = all_data[code] hist = df[df.index <= today].tail(max_lookback + 1) if len(hist) < config['min_days']: continue close_arr = hist['close'].values if config['auto_day']: if len(hist) < max_lookback: lookback = config['fixed_days'] else: long_atr = calc_atr(hist['high'], hist['low'], hist['close'], config['max_days']) short_atr = calc_atr(hist['high'], hist['low'], hist['close'], config['min_days']) la = long_atr.iloc[-1] sa = short_atr.iloc[-1] if la > 0 and not np.isnan(la) and not np.isnan(sa): ratio = min(0.9, sa / la) lookback = int(config['min_days'] + (config['max_days'] - config['min_days']) * (1 - ratio)) else: lookback = config['fixed_days'] prices = close_arr[-lookback:] else: prices = close_arr[-config['fixed_days']:] if len(prices) < 5: continue result = calc_weighted_momentum_score(prices) score = result['score'] score = apply_crash_filter(close_arr, score) # 溢价过滤 if code in nav_data: nav_df = nav_data[code] nav_row = nav_df[nav_df.index <= today] if not nav_row.empty: nav_val = nav_row.iloc[-1]['nav'] etf_price = close_arr[-1] premium = calc_premium_rate(etf_price, nav_val) if premium >= config['premium_threshold']: score -= 1 if 0 < score < 6: scores[code] = score score_details[code] = result # 选出排名最高的标的 if scores: ranked = sorted(scores.items(), key=lambda x: x[1], reverse=True) target = ranked[0][0] else: target = None # 计算当日收益 if holding is not None and holding in all_data: df_h = all_data[holding] if today in df_h.index: prev_dates = df_h[df_h.index < today].index if len(prev_dates) > 0: prev_date = prev_dates[-1] prev_price = df_h.loc[prev_date, 'close'] today_price = df_h.loc[today, 'close'] daily_ret = today_price / prev_price - 1 else: daily_ret = 0.0 else: daily_ret = 0.0 else: daily_ret = 0.0 # 调仓成本 trade_cost = 0.0 if target != holding: trade_cost = config['trade_cost'] if holding is not None: signals.append({ 'date': today, 'action': '调仓', 'from': holding, 'to': target or '空仓', 'score': scores.get(target, 0) if target else 0, }) holding = target daily_returns.append({ 'date': today, 'daily_return': daily_ret - trade_cost if trade_cost > 0 else daily_ret, 'holding': holding or '空仓', }) # 4. 计算绩效 result_df = pd.DataFrame(daily_returns).set_index('date') result_df['nav'] = (1 + result_df['daily_return']).cumprod() # 基准数据 benchmark_code = config['benchmark'] print(f"\n获取基准数据 {benchmark_code}...") import os, tushare as ts pro = ts.pro_api(os.getenv("TUSHARE_TOKEN")) bench_df = pro.index_daily( ts_code=benchmark_code, start_date=config['start_date'].replace('-', ''), end_date=end_date.replace('-', ''), ) if bench_df is not None and not bench_df.empty: bench_df['date'] = pd.to_datetime(bench_df['trade_date']) bench_df = bench_df.set_index('date').sort_index() bench_close = bench_df['close'].reindex(result_df.index, method='ffill') result_df['bench_return'] = bench_close / bench_close.iloc[0] else: result_df['bench_return'] = 1.0 # 5. 输出绩效报告 print_performance(result_df, signals, config) # 6. 年度收益统计 print_yearly_returns(result_df) # 7. 生成图表 save_chart(result_df, config) return result_df # ==================== 图表生成 ==================== def save_chart(result_df: pd.DataFrame, config: dict): """生成净值曲线图""" try: import matplotlib matplotlib.use('Agg') import matplotlib.pyplot as plt matplotlib.rcParams['font.sans-serif'] = ['Arial Unicode MS', 'SimHei', 'DejaVu Sans'] matplotlib.rcParams['axes.unicode_minus'] = False fig, (ax1, ax2) = plt.subplots(2, 1, figsize=(14, 8), height_ratios=[3, 1], gridspec_kw={'hspace': 0.3}) ax1.plot(result_df.index, result_df['nav'], label='全球市场轮动', linewidth=1.5, color='#2ecc71') ax1.plot(result_df.index, result_df['bench_return'], label='沪深300', linewidth=1, color='#95a5a6') ax1.set_title('全球市场ETF轮动策略 净值曲线', fontsize=14, fontweight='bold') ax1.legend(loc='upper left') ax1.grid(True, alpha=0.3) ax1.set_ylabel('净值') peak = result_df['nav'].cummax() drawdown = (result_df['nav'] - peak) / peak ax2.fill_between(result_df.index, drawdown, 0, alpha=0.4, color='#e74c3c') ax2.set_title('回撤', fontsize=12) ax2.set_ylabel('回撤') ax2.grid(True, alpha=0.3) chart_path = Path(__file__).parent / 'results' / 'global_market_chart.png' chart_path.parent.mkdir(exist_ok=True) fig.savefig(chart_path, dpi=150, bbox_inches='tight') plt.close(fig) print(f"\n报告图表已保存: {chart_path}") except Exception as e: print(f"\n图表生成失败: {e}") # ==================== 主入口 ==================== if __name__ == "__main__": run_backtest(CONFIG)