From eb6a07548c5140a607696428765ed8e6c8bd2299 Mon Sep 17 00:00:00 2001 From: aszerW Date: Wed, 29 Apr 2026 22:20:13 +0800 Subject: [PATCH] =?UTF-8?q?feat(=E5=85=A8=E7=90=83=E5=B8=82=E5=9C=BA):=20?= =?UTF-8?q?=E8=BF=81=E7=A7=BB=E8=81=9A=E5=AE=BD=E7=AD=96=E7=95=A5=E4=B8=BA?= =?UTF-8?q?Tushare=E7=8B=AC=E7=AB=8B=E5=9B=9E=E6=B5=8B=E7=89=88=E6=9C=AC?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 从聚宽平台迁移全球市场ETF轮动策略，复用动量.py核心模块。ETF池: 纳指100/日经225/德国DAX/黄金/有色金属/南方原油/30年国债/红利低波/创业板。回测(2019~2026): CAGR=44.29%, Sharpe=1.50, MaxDD=-16.93%, Calmar=2.62, 盈利年份8/8, 跑赢基准7/8 --- 全球市场.py | 299 ++++++++++++++++++++++++++++++++++++++++++++++++++++ 1 file changed, 299 insertions(+) create mode 100644 全球市场.py diff --git a/全球市场.py b/全球市场.py new file mode 100644 index 0000000..66f1fff --- /dev/null +++ b/全球市场.py @@ -0,0 +1,299 @@ +""" +全球市场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)