尧图网站开发 
全天候五福ETF轮动策略框架深度拆解最近和几位做量化投资的朋友聊天发现一个挺有意思的现象大家手里的策略模型越来越复杂但真正能稳定跑赢宽基指数的却不多。很多人把精力花在追逐热点、预测短期涨跌上结果往往是“一顿操作猛如虎回头一看原地杵”。这背后反映出一个核心问题对于大多数非机构投资者而言一个策略的长期生命力往往不在于它有多“聪明”而在于它的逻辑是否足够清晰、稳健以及是否具备良好的风险分散能力。今天要拆解的“全天候五福ETF轮动策略”就是一个典型的、试图用清晰规则和资产分散来穿越周期的思路。它不追求抓住每一个波段而是通过一套固定的轮动规则在几类具有不同风险收益特征的ETF之间进行切换目标是实现长期相对稳健的收益。这个名字听起来有点“玄学”但内核其实是一套可量化、可回溯、可执行的系统化交易框架。本文将为你彻底拆解这个策略的逻辑、实现方法、潜在优势与陷阱并提供一套完整的Python回测代码。无论你是量化新手想了解策略构建还是老手在寻找稳健的配置思路这篇文章都能给你带来实实在在的参考。1. 策略核心逻辑它到底想解决什么问题在深入代码之前我们必须先理解这个策略的灵魂。它的目标不是“暴富”而是试图解决普通投资者在资产配置中的几个典型痛点择时困境普通投资者难以判断市场顶底频繁交易容易追涨杀跌。资产单一风险重仓单一行业或主题净值波动大体验差。人性弱点贪婪和恐惧导致计划无法严格执行。“全天候五福ETF轮动策略”的应对思路是规则化用明确的、量化的规则如均线、动量代替主观判断消除情绪干扰。分散化同时关注多个不同类别的资产如沪深300代表大盘中证500代表中小盘黄金、债券等不把鸡蛋放在一个篮子里。轮动化定期如每周或每月根据预设规则将资金切换到当前“状态最好”的1-2个ETF上实现“强者恒强”的动量效应或“均值回归”的逆向效应。所谓“五福”通常指代五类不同属性的ETF构建一个简易的“全天候”组合。常见的构成思路是权益类进攻沪深300ETF大盘、中证500ETF中小盘。固收类防守国债ETF或企业债ETF。商品类抗通胀/避险黄金ETF。海外类分散恒生ETF或纳斯达克ETF。策略的核心决策就变成了在每一个调仓日如何从这“五福”中选出下一个持有周期内最值得持有的标的。2. 核心概念与轮动引擎原理要实现轮动需要一个“评选标准”或“信号引擎”。这是策略的“大脑”决定了资金的流向。常见的引擎有2.1 动量Momentum模型通俗理解追涨杀跌的量化版。认为近期上涨的资产短期内继续上涨的概率更大。常用指标时间序列动量计算过去N日如20日、60日的收益率排名最高者胜出。截面动量在同一个时间点比较所有备选资产在过去一段时间的表现选择最强的。均线系统价格位于某条均线如20日均线之上视为强势之下视为弱势。轮动只在处于“强势”状态的资产中进行。2.2 均值回归Mean Reversion模型通俗理解高抛低吸的量化版。认为资产价格围绕其价值中枢波动涨多了会跌跌多了会涨。常用指标RSI相对强弱指数当RSI低于某一阈值如30时视为超卖可能买入高于某一阈值如70时视为超买可能卖出。布林带Bollinger Bands价格触及下轨时买入触及上轨时卖出。2.3 风险平价Risk Parity模型通俗理解不追求收益最高而追求风险最均衡。让组合中各类资产对整体风险的贡献度相等。实现方式需要计算资产的历史波动率、相关性并通过优化算法分配权重。计算相对复杂但理念非常先进。“全天候五福”策略通常采用“动量趋势过滤”的复合模型。例如趋势过滤首先用长期均线如120日线判断整个市场或某个资产是否处于可投资状态线上为可投线下则空仓或持有债券。动量选拔在通过趋势过滤的资产池中再用短期动量如20日收益率进行排名选择排名第一的资产持有。这种设计兼顾了中长期趋势和短期强度目的是在牛市跟得上在熊市能躲开。3. 环境准备与数据获取在开始构建策略之前我们需要准备好Python环境和数据源。3.1 环境配置建议使用Anaconda创建独立的Python环境。# 创建新环境 conda create -n etf_rotation python3.9 conda activate etf_rotation # 安装核心库 pip install pandas numpy matplotlib pip install jqdatasdk # 聚宽数据API需注册获取账号 # 或使用其他数据源如 akshare, tushare, baostock 等 # pip install akshare本文示例将使用akshare作为免费数据源它无需注册但数据可能有延迟适合学习回测。3.2 备选ETF池定义我们需要确定具体的“五福”标的及其交易代码。以下是一个示例池以沪深市场为例资产类别代表指数/资产常用ETF代码说明大盘权益沪深300510300.SH(华泰柏瑞沪深300ETF)市场基准大盘蓝筹中小盘权益中证500510500.SH(南方中证500ETF)成长风格波动较大固收防守国债511010.SH(国泰上证5年期国债ETF)低风险与股市常呈负相关商品避险黄金518880.SH(华安黄金ETF)抗通胀避险资产海外分散恒生指数159920.SZ(华夏恒生ETF)投资港股分散A股风险注意实际回测中需注意ETF的上市日期避免使用上市前的数据。4. 策略流程拆解与代码实现我们将实现一个基于“双均线趋势过滤 动量排名”的经典轮动策略。策略规则调仓周期每月第一个交易日调仓。趋势过滤计算每个ETF的收盘价相对于其120日简单移动平均线SMA的位置。只有当Close SMA120时该ETF才具备被选资格。动量排名在所有通过趋势过滤的ETF中计算它们过去20个交易日的收益率动量。选择标的持有动量排名第一的ETF。空仓规则如果所有ETF都不满足Close SMA120即均线之下则整个组合空仓持有现金。4.1 数据获取与预处理首先我们编写一个函数来获取多只ETF的历史数据。# 文件data_fetcher.py import akshare as ak import pandas as pd import numpy as np from datetime import datetime, timedelta def fetch_etf_data(etf_codes, start_date2015-01-01, end_dateNone): 获取多只ETF的复权价格数据 :param etf_codes: dict, {‘代码’: ‘名称’} :param start_date: str, 开始日期 :param end_date: str, 结束日期默认为今天 :return: pd.DataFrame, 列名为代码值为复权收盘价 if end_date is None: end_date datetime.now().strftime(%Y-%m-%d) all_data {} for code, name in etf_codes.items(): print(f正在获取 {name}({code}) 的数据...) try: # 使用akshare的基金接口这里以‘基金历史净值’为例实际需根据代码调整接口 # 注意akshare接口可能变化且部分ETF需使用不同函数 # 此处为示例逻辑可能需要替换为正确的akshare函数或使用其他数据源 if code.endswith(.SH): # 示例上海证券交易所ETF使用fund_etf_hist_sina df ak.fund_etf_hist_sina(symbolcode.replace(.SH, )) else: # 示例深圳证券交易所ETF df ak.fund_etf_hist_sina(symbolcode.replace(.SZ, )) # 假设返回的DataFrame有date和close列 df[date] pd.to_datetime(df[date]) df.set_index(date, inplaceTrue) # 选取时间区间 df df.loc[start_date:end_date] if not df.empty: # 使用复权收盘价这里假设close已是复权价 all_data[code] df[close] else: print(f 警告{code} 在指定时间段无数据) except Exception as e: print(f 获取 {code} 数据失败: {e}) continue # 合并所有数据 price_df pd.DataFrame(all_data) price_df price_df.sort_index() # 按日期排序 price_df price_df.ffill().dropna() # 前向填充并删除仍有空值的行 print(f数据获取完成时间范围{price_df.index[0]} 至 {price_df.index[-1]} 共{len(price_df)}个交易日。) return price_df # 定义我们的“五福”ETF池 etf_pool { 510300.SH: 沪深300ETF, 510500.SH: 中证500ETF, 511010.SH: 国债ETF, 518880.SH: 黄金ETF, 159920.SZ: 恒生ETF } # 获取数据 (实际操作时请检查akshare接口可用性或替换为其他稳定数据源) # price_df fetch_etf_data(etf_pool, start_date2018-01-01) # 由于akshare接口可能不稳定这里我们先模拟一个数据框架用于后续策略逻辑演示 print(注意实际应用中请使用稳定的数据源如聚宽、Tushare Pro、Baostock获取精确的复权价格数据。)4.2 策略信号生成接下来我们根据规则计算交易信号。# 文件strategy_signal.py def calculate_signals(price_df, fast_window20, slow_window120): 计算双均线过滤下的动量轮动信号 :param price_df: 价格DataFrame :param fast_window: 动量计算窗口短期 :param slow_window: 趋势过滤窗口长期 :return: signal_df (信号DataFrame), 1表示持有0表示不持有 signal_df pd.DataFrame(indexprice_df.index, columnsprice_df.columns) signal_df[:] 0 # 初始化为0 # 计算慢速均线趋势过滤线 sma_slow price_df.rolling(windowslow_window, min_periods1).mean() # 计算快速动量过去fast_window日的收益率 # 使用.pct_change计算收益率然后计算滚动窗口内的总收益 momentum price_df.pct_change(fast_window) for i in range(slow_window, len(price_df)): current_date price_df.index[i] # 趋势过滤找出当前价格在慢速均线之上的ETF above_sma price_df.iloc[i] sma_slow.iloc[i] eligible_etfs above_sma[above_sma].index.tolist() if eligible_etfs: # 动量排名在符合条件的ETF中选择过去动量最大的一个 # 使用.iloc[i]获取当前行的动量值但动量计算是基于过去窗口的所以是有效的 current_momentum momentum.loc[current_date, eligible_etfs] best_etf current_momentum.idxmax() # 动量最大者 signal_df.loc[current_date, best_etf] 1 else: # 所有ETF都在均线之下空仓信号全为0 pass return signal_df # 假设我们有一个价格DataFrame price_df # signal_df calculate_signals(price_df)4.3 策略回测与绩效分析有了信号我们就可以模拟交易计算净值曲线和各项指标。# 文件backtest_engine.py def simple_backtest(price_df, signal_df, initial_capital1000000, commission_rate0.0003): 简单回测引擎不考虑涨停跌停、交易数量取整等细节适合原理演示 :param price_df: 价格数据 :param signal_df: 交易信号 :param initial_capital: 初始本金 :param commission_rate: 单边交易佣金率万三为例 :return: 回测结果字典包含净值、持仓等 # 对齐数据 common_index price_df.index.intersection(signal_df.index) price_df price_df.loc[common_index] signal_df signal_df.loc[common_index] # 初始化 capital initial_capital position pd.DataFrame(0, indexprice_df.index, columnsprice_df.columns) # 持仓数量 cash pd.Series(0.0, indexprice_df.index) # 现金 cash.iloc[0] capital nav pd.Series(0.0, indexprice_df.index) # 资产净值 nav.iloc[0] capital # 上一期信号用于判断是否调仓 prev_signal pd.Series(0, indexprice_df.columns) for i in range(1, len(price_df)): current_date price_df.index[i] prev_date price_df.index[i-1] current_signal signal_df.iloc[i] current_price price_df.iloc[i] # 判断是否需要调仓信号发生变化 if not current_signal.equals(prev_signal): # 先清空上一期持仓按昨日价格计算价值简化处理 # 实际中应在调仓日以开盘价卖出此处用昨日收盘价近似 sell_price price_df.iloc[i-1] position_value (position.iloc[i-1] * sell_price).sum() # 扣除卖出佣金 cash_after_sell cash.iloc[i-1] position_value * (1 - commission_rate) # 买入新的标的 target_etf current_signal[current_signal 1].index if len(target_etf) 0: target_etf target_etf[0] # 计算可买数量按当前价格并预留佣金 buy_price current_price[target_etf] # 扣除买入佣金 available_cash_for_buy cash_after_sell / (1 commission_rate) shares_to_buy int(available_cash_for_buy / buy_price) position.iloc[i, price_df.columns.get_loc(target_etf)] shares_to_buy cash.iloc[i] cash_after_sell - shares_to_buy * buy_price * (1 commission_rate) else: # 空仓 position.iloc[i] 0 cash.iloc[i] cash_after_sell else: # 持仓不变市值随价格波动 position.iloc[i] position.iloc[i-1] cash.iloc[i] cash.iloc[i-1] # 计算当日净值 market_value (position.iloc[i] * current_price).sum() nav.iloc[i] cash.iloc[i] market_value # 更新上一期信号 prev_signal current_signal.copy() # 计算收益率序列 returns nav.pct_change().fillna(0) # 计算基础绩效指标 total_return (nav.iloc[-1] / nav.iloc[0]) - 1 annual_return (1 total_return) ** (252 / len(nav)) - 1 # 年化交易日按252计 volatility returns.std() * np.sqrt(252) # 年化波动率 if volatility 0: sharpe_ratio annual_return / volatility # 夏普比率无风险利率设为0简化 else: sharpe_ratio np.nan # 最大回撤 cummax nav.cummax() drawdown (nav - cummax) / cummax max_drawdown drawdown.min() results { nav: nav, returns: returns, position: position, cash: cash, total_return: total_return, annual_return: annual_return, volatility: volatility, sharpe_ratio: sharpe_ratio, max_drawdown: max_drawdown, drawdown: drawdown } return results # 假设已有 price_df 和 signal_df # results simple_backtest(price_df, signal_df)5. 运行结果分析与可视化回测完成后我们需要直观地看到策略表现。我们将策略净值与简单买入持有例如沪深300ETF进行对比。# 文件visualization.py import matplotlib.pyplot as plt import matplotlib.dates as mdates def plot_backtest_results(results, benchmark_price_series, benchmark_nameBenchmark): 绘制回测结果图 :param results: 回测引擎返回的结果字典 :param benchmark_price_series: 基准价格序列如沪深300ETF价格 :param benchmark_name: 基准名称 nav results[nav] drawdown results[drawdown] # 将基准价格归一化以便与策略净值比较 benchmark_nav benchmark_price_series / benchmark_price_series.iloc[0] * nav.iloc[0] fig, axes plt.subplots(2, 1, figsize(14, 10), sharexTrue) # 图1净值曲线对比 ax1 axes[0] ax1.plot(nav.index, nav.values, labelfRotation Strategy (Ann.Ret{results[annual_return]:.2%}, Sharpe{results[sharpe_ratio]:.2f}), linewidth2) ax1.plot(benchmark_nav.index, benchmark_nav.values, labelf{benchmark_name} (Buy Hold), linestyle--, linewidth1.5) ax1.set_ylabel(Net Asset Value) ax1.set_title(ETF Rotation Strategy vs Buy Hold) ax1.legend() ax1.grid(True, whichmajor, linestyle--, linewidth0.5, alpha0.7) # 图2回撤曲线 ax2 axes[1] ax2.fill_between(drawdown.index, 0, drawdown.values * 100, colorred, alpha0.3) ax2.plot(drawdown.index, drawdown.values * 100, colordarkred, linewidth1) ax2.set_ylabel(Drawdown (%)) ax2.set_xlabel(Date) ax2.set_title(fStrategy Drawdown (Max DD{results[max_drawdown]:.2%})) ax2.grid(True, whichmajor, linestyle--, linewidth0.5, alpha0.7) # 格式化x轴日期 ax2.xaxis.set_major_formatter(mdates.DateFormatter(%Y-%m)) ax2.xaxis.set_major_locator(mdates.YearLocator()) plt.xticks(rotation45) plt.tight_layout() plt.show() # 打印关键指标 print(*50) print(策略绩效概览) print(*50) print(f累计收益率: {results[total_return]:.2%}) print(f年化收益率: {results[annual_return]:.2%}) print(f年化波动率: {results[volatility]:.2%}) print(f夏普比率: {results[sharpe_ratio]:.2f}) print(f最大回撤: {results[max_drawdown]:.2%}) print(*50) # 假设 benchmark_price 是沪深300ETF的价格序列 # plot_backtest_results(results, benchmark_price, CSI 300 ETF)6. 策略的潜在问题与优化方向没有一个策略是完美的。在激动地看到回测曲线之前我们必须清醒地认识到这个框架的局限性。6.1 常见问题与陷阱过度拟合Overfitting这是量化策略最大的敌人。你可能通过反复测试找到一组在历史数据上表现完美的参数如均线周期20/120调仓周期每月但这组参数在未来很可能失效。应对使用样本外数据测试、进行参数敏感性分析、避免使用过多复杂规则。幸存者偏差Survivorship Bias我们使用的ETF都是目前存活且活跃的但回测期初可能有些ETF还没上市或者有些已退市。这会导致回测结果过于乐观。应对使用“点时间”数据确保回测中只使用在当时已上市的数据。交易成本与冲击成本我们的简单回测假设可以立即以收盘价成交且佣金费率固定。现实中大额交易会产生冲击成本调仓时可能无法理想成交。应对在回测中引入更复杂的交易模型如按开盘价成交并加入固定的滑点如0.1%。数据透视偏差Look-ahead Bias在计算信号时错误地使用了未来的数据。例如在T日收盘后我们只能用T日及之前的数据计算T日的信号而不能用到T1日的开盘价。应对确保所有计算在时间轴上是严格滞后的。上面的示例代码已注意此点。策略同质化与失效如果一个简单的轮动策略被广泛知晓和使用其有效性可能会下降因为大量的资金涌入会侵蚀策略的Alpha超额收益。6.2 策略优化与进阶思路如果你理解了基础框架并想进一步探索可以考虑以下方向信号引擎升级多因子综合不止看动量可以加入估值PE/PB、波动率、换手率等其他因子进行加权打分。机器学习模型使用逻辑回归、随机森林等模型基于多维度特征预测下一期最优资产。但需严防过拟合。风险预算模型不单纯追求收益最高而是控制组合的整体风险向风险平价Risk Parity思路靠拢。资产池扩展增加资产类别纳入海外债、原油、REITs等更多不相关资产。细分行业ETF在权益内部轮动到不同的行业ETF如消费、科技、医药、金融。风控机制强化动态止损当持有资产从入场点回撤超过一定幅度时强制平仓。波动率控制在市场整体波动率急剧放大时降低仓位或转向更保守的资产。相关性监控当资产池内资产相关性突然升高时同涨同跌意味着分散失效需触发风控。仓位管理优化不止持有一只可以持有排名前2或前3的资产并动态分配权重如按动量大小分配。金字塔加减仓在趋势确认后分批加仓在趋势减弱时分批减仓。7. 最佳实践与工程化建议如果你想将此类策略用于实盘或更严肃的研究以下几点至关重要数据质量是生命线使用付费的、稳定的数据源如聚宽、万得、Tushare Pro确保数据的准确性和完整性除权除息、停牌等。建立本地数据缓存避免每次回测都重复请求网络数据。回测框架要严谨使用成熟的回测框架如Zipline,Backtrader,Qlib它们处理了众多细节成交顺序、税费、涨跌停等。进行前向分析Walk Forward Analysis, WFA将历史数据分成多个训练集和测试集滚动优化和测试检验策略的稳健性。参数敏感性分析对关键参数均线周期、动量窗口、调仓频率进行网格搜索观察策略绩效是否在参数小范围变动时发生剧烈变化。如果变化剧烈说明策略不稳定。日志与监控在策略中植入详细的日志记录记录每一次调仓的逻辑、信号值、成交情况。实盘运行时需要有独立的监控程序定期检查策略状态、持仓与账户的一致性。永远敬畏市场任何量化模型都是对历史规律的总结。市场结构会变黑天鹅事件会发生。实盘前必须经过长时间的模拟盘Paper Trading验证。投入实盘的资金应是你能承受损失的部分并做好策略阶段性失效的准备。“全天候五福ETF轮动策略”为我们提供了一个从零开始构建系统化投资框架的优秀范本。它的价值不在于提供一个“圣杯”而在于展示了一种用规则战胜情绪、用分散应对不确定、用回溯检验想法的理性投资方法论。通过本文的拆解你应该已经掌握了从数据获取、信号生成、回测验证到风险分析的完整链条。下一步我建议你亲手复现使用稳定的数据源将本文的代码跑通观察策略在不同历史阶段如牛市、熊市、震荡市的表现。大胆修改尝试更换资产池、修改信号规则比如用RSI替代均线过滤、调整调仓周期感受策略绩效的变化。深入思考这个策略最大的弱点可能在哪里在什么样的市场环境下它会持续亏损如何改进才能让它更适应未来的市场量化交易的世界没有终点每一个策略的终点都是新的起点。希望这个“五福”框架能成为你探索这个迷人领域的第一块坚实跳板。 
全天候五福ETF轮动策略框架深度拆解最近和几位做量化投资的朋友聊天发现一个挺有意思的现象大家手里的策略模型越来越复杂但真正能稳定跑赢宽基指数的却不多。很多人把精力花在追逐热点、预测短期涨跌上结果往往是“一顿操作猛如虎回头一看原地杵”。这背后反映出一个核心问题对于大多数非机构投资者而言一个策略的长期生命力往往不在于它有多“聪明”而在于它的逻辑是否足够清晰、稳健以及是否具备良好的风险分散能力。今天要拆解的“全天候五福ETF轮动策略”就是一个典型的、试图用清晰规则和资产分散来穿越周期的思路。它不追求抓住每一个波段而是通过一套固定的轮动规则在几类具有不同风险收益特征的ETF之间进行切换目标是实现长期相对稳健的收益。这个名字听起来有点“玄学”但内核其实是一套可量化、可回溯、可执行的系统化交易框架。本文将为你彻底拆解这个策略的逻辑、实现方法、潜在优势与陷阱并提供一套完整的Python回测代码。无论你是量化新手想了解策略构建还是老手在寻找稳健的配置思路这篇文章都能给你带来实实在在的参考。1. 策略核心逻辑它到底想解决什么问题在深入代码之前我们必须先理解这个策略的灵魂。它的目标不是“暴富”而是试图解决普通投资者在资产配置中的几个典型痛点择时困境普通投资者难以判断市场顶底频繁交易容易追涨杀跌。资产单一风险重仓单一行业或主题净值波动大体验差。人性弱点贪婪和恐惧导致计划无法严格执行。“全天候五福ETF轮动策略”的应对思路是规则化用明确的、量化的规则如均线、动量代替主观判断消除情绪干扰。分散化同时关注多个不同类别的资产如沪深300代表大盘中证500代表中小盘黄金、债券等不把鸡蛋放在一个篮子里。轮动化定期如每周或每月根据预设规则将资金切换到当前“状态最好”的1-2个ETF上实现“强者恒强”的动量效应或“均值回归”的逆向效应。所谓“五福”通常指代五类不同属性的ETF构建一个简易的“全天候”组合。常见的构成思路是权益类进攻沪深300ETF大盘、中证500ETF中小盘。固收类防守国债ETF或企业债ETF。商品类抗通胀/避险黄金ETF。海外类分散恒生ETF或纳斯达克ETF。策略的核心决策就变成了在每一个调仓日如何从这“五福”中选出下一个持有周期内最值得持有的标的。2. 核心概念与轮动引擎原理要实现轮动需要一个“评选标准”或“信号引擎”。这是策略的“大脑”决定了资金的流向。常见的引擎有2.1 动量Momentum模型通俗理解追涨杀跌的量化版。认为近期上涨的资产短期内继续上涨的概率更大。常用指标时间序列动量计算过去N日如20日、60日的收益率排名最高者胜出。截面动量在同一个时间点比较所有备选资产在过去一段时间的表现选择最强的。均线系统价格位于某条均线如20日均线之上视为强势之下视为弱势。轮动只在处于“强势”状态的资产中进行。2.2 均值回归Mean Reversion模型通俗理解高抛低吸的量化版。认为资产价格围绕其价值中枢波动涨多了会跌跌多了会涨。常用指标RSI相对强弱指数当RSI低于某一阈值如30时视为超卖可能买入高于某一阈值如70时视为超买可能卖出。布林带Bollinger Bands价格触及下轨时买入触及上轨时卖出。2.3 风险平价Risk Parity模型通俗理解不追求收益最高而追求风险最均衡。让组合中各类资产对整体风险的贡献度相等。实现方式需要计算资产的历史波动率、相关性并通过优化算法分配权重。计算相对复杂但理念非常先进。“全天候五福”策略通常采用“动量趋势过滤”的复合模型。例如趋势过滤首先用长期均线如120日线判断整个市场或某个资产是否处于可投资状态线上为可投线下则空仓或持有债券。动量选拔在通过趋势过滤的资产池中再用短期动量如20日收益率进行排名选择排名第一的资产持有。这种设计兼顾了中长期趋势和短期强度目的是在牛市跟得上在熊市能躲开。3. 环境准备与数据获取在开始构建策略之前我们需要准备好Python环境和数据源。3.1 环境配置建议使用Anaconda创建独立的Python环境。# 创建新环境 conda create -n etf_rotation python3.9 conda activate etf_rotation # 安装核心库 pip install pandas numpy matplotlib pip install jqdatasdk # 聚宽数据API需注册获取账号 # 或使用其他数据源如 akshare, tushare, baostock 等 # pip install akshare本文示例将使用akshare作为免费数据源它无需注册但数据可能有延迟适合学习回测。3.2 备选ETF池定义我们需要确定具体的“五福”标的及其交易代码。以下是一个示例池以沪深市场为例资产类别代表指数/资产常用ETF代码说明大盘权益沪深300510300.SH(华泰柏瑞沪深300ETF)市场基准大盘蓝筹中小盘权益中证500510500.SH(南方中证500ETF)成长风格波动较大固收防守国债511010.SH(国泰上证5年期国债ETF)低风险与股市常呈负相关商品避险黄金518880.SH(华安黄金ETF)抗通胀避险资产海外分散恒生指数159920.SZ(华夏恒生ETF)投资港股分散A股风险注意实际回测中需注意ETF的上市日期避免使用上市前的数据。4. 策略流程拆解与代码实现我们将实现一个基于“双均线趋势过滤 动量排名”的经典轮动策略。策略规则调仓周期每月第一个交易日调仓。趋势过滤计算每个ETF的收盘价相对于其120日简单移动平均线SMA的位置。只有当Close SMA120时该ETF才具备被选资格。动量排名在所有通过趋势过滤的ETF中计算它们过去20个交易日的收益率动量。选择标的持有动量排名第一的ETF。空仓规则如果所有ETF都不满足Close SMA120即均线之下则整个组合空仓持有现金。4.1 数据获取与预处理首先我们编写一个函数来获取多只ETF的历史数据。# 文件data_fetcher.py import akshare as ak import pandas as pd import numpy as np from datetime import datetime, timedelta def fetch_etf_data(etf_codes, start_date2015-01-01, end_dateNone): 获取多只ETF的复权价格数据 :param etf_codes: dict, {‘代码’: ‘名称’} :param start_date: str, 开始日期 :param end_date: str, 结束日期默认为今天 :return: pd.DataFrame, 列名为代码值为复权收盘价 if end_date is None: end_date datetime.now().strftime(%Y-%m-%d) all_data {} for code, name in etf_codes.items(): print(f正在获取 {name}({code}) 的数据...) try: # 使用akshare的基金接口这里以‘基金历史净值’为例实际需根据代码调整接口 # 注意akshare接口可能变化且部分ETF需使用不同函数 # 此处为示例逻辑可能需要替换为正确的akshare函数或使用其他数据源 if code.endswith(.SH): # 示例上海证券交易所ETF使用fund_etf_hist_sina df ak.fund_etf_hist_sina(symbolcode.replace(.SH, )) else: # 示例深圳证券交易所ETF df ak.fund_etf_hist_sina(symbolcode.replace(.SZ, )) # 假设返回的DataFrame有date和close列 df[date] pd.to_datetime(df[date]) df.set_index(date, inplaceTrue) # 选取时间区间 df df.loc[start_date:end_date] if not df.empty: # 使用复权收盘价这里假设close已是复权价 all_data[code] df[close] else: print(f 警告{code} 在指定时间段无数据) except Exception as e: print(f 获取 {code} 数据失败: {e}) continue # 合并所有数据 price_df pd.DataFrame(all_data) price_df price_df.sort_index() # 按日期排序 price_df price_df.ffill().dropna() # 前向填充并删除仍有空值的行 print(f数据获取完成时间范围{price_df.index[0]} 至 {price_df.index[-1]} 共{len(price_df)}个交易日。) return price_df # 定义我们的“五福”ETF池 etf_pool { 510300.SH: 沪深300ETF, 510500.SH: 中证500ETF, 511010.SH: 国债ETF, 518880.SH: 黄金ETF, 159920.SZ: 恒生ETF } # 获取数据 (实际操作时请检查akshare接口可用性或替换为其他稳定数据源) # price_df fetch_etf_data(etf_pool, start_date2018-01-01) # 由于akshare接口可能不稳定这里我们先模拟一个数据框架用于后续策略逻辑演示 print(注意实际应用中请使用稳定的数据源如聚宽、Tushare Pro、Baostock获取精确的复权价格数据。)4.2 策略信号生成接下来我们根据规则计算交易信号。# 文件strategy_signal.py def calculate_signals(price_df, fast_window20, slow_window120): 计算双均线过滤下的动量轮动信号 :param price_df: 价格DataFrame :param fast_window: 动量计算窗口短期 :param slow_window: 趋势过滤窗口长期 :return: signal_df (信号DataFrame), 1表示持有0表示不持有 signal_df pd.DataFrame(indexprice_df.index, columnsprice_df.columns) signal_df[:] 0 # 初始化为0 # 计算慢速均线趋势过滤线 sma_slow price_df.rolling(windowslow_window, min_periods1).mean() # 计算快速动量过去fast_window日的收益率 # 使用.pct_change计算收益率然后计算滚动窗口内的总收益 momentum price_df.pct_change(fast_window) for i in range(slow_window, len(price_df)): current_date price_df.index[i] # 趋势过滤找出当前价格在慢速均线之上的ETF above_sma price_df.iloc[i] sma_slow.iloc[i] eligible_etfs above_sma[above_sma].index.tolist() if eligible_etfs: # 动量排名在符合条件的ETF中选择过去动量最大的一个 # 使用.iloc[i]获取当前行的动量值但动量计算是基于过去窗口的所以是有效的 current_momentum momentum.loc[current_date, eligible_etfs] best_etf current_momentum.idxmax() # 动量最大者 signal_df.loc[current_date, best_etf] 1 else: # 所有ETF都在均线之下空仓信号全为0 pass return signal_df # 假设我们有一个价格DataFrame price_df # signal_df calculate_signals(price_df)4.3 策略回测与绩效分析有了信号我们就可以模拟交易计算净值曲线和各项指标。# 文件backtest_engine.py def simple_backtest(price_df, signal_df, initial_capital1000000, commission_rate0.0003): 简单回测引擎不考虑涨停跌停、交易数量取整等细节适合原理演示 :param price_df: 价格数据 :param signal_df: 交易信号 :param initial_capital: 初始本金 :param commission_rate: 单边交易佣金率万三为例 :return: 回测结果字典包含净值、持仓等 # 对齐数据 common_index price_df.index.intersection(signal_df.index) price_df price_df.loc[common_index] signal_df signal_df.loc[common_index] # 初始化 capital initial_capital position pd.DataFrame(0, indexprice_df.index, columnsprice_df.columns) # 持仓数量 cash pd.Series(0.0, indexprice_df.index) # 现金 cash.iloc[0] capital nav pd.Series(0.0, indexprice_df.index) # 资产净值 nav.iloc[0] capital # 上一期信号用于判断是否调仓 prev_signal pd.Series(0, indexprice_df.columns) for i in range(1, len(price_df)): current_date price_df.index[i] prev_date price_df.index[i-1] current_signal signal_df.iloc[i] current_price price_df.iloc[i] # 判断是否需要调仓信号发生变化 if not current_signal.equals(prev_signal): # 先清空上一期持仓按昨日价格计算价值简化处理 # 实际中应在调仓日以开盘价卖出此处用昨日收盘价近似 sell_price price_df.iloc[i-1] position_value (position.iloc[i-1] * sell_price).sum() # 扣除卖出佣金 cash_after_sell cash.iloc[i-1] position_value * (1 - commission_rate) # 买入新的标的 target_etf current_signal[current_signal 1].index if len(target_etf) 0: target_etf target_etf[0] # 计算可买数量按当前价格并预留佣金 buy_price current_price[target_etf] # 扣除买入佣金 available_cash_for_buy cash_after_sell / (1 commission_rate) shares_to_buy int(available_cash_for_buy / buy_price) position.iloc[i, price_df.columns.get_loc(target_etf)] shares_to_buy cash.iloc[i] cash_after_sell - shares_to_buy * buy_price * (1 commission_rate) else: # 空仓 position.iloc[i] 0 cash.iloc[i] cash_after_sell else: # 持仓不变市值随价格波动 position.iloc[i] position.iloc[i-1] cash.iloc[i] cash.iloc[i-1] # 计算当日净值 market_value (position.iloc[i] * current_price).sum() nav.iloc[i] cash.iloc[i] market_value # 更新上一期信号 prev_signal current_signal.copy() # 计算收益率序列 returns nav.pct_change().fillna(0) # 计算基础绩效指标 total_return (nav.iloc[-1] / nav.iloc[0]) - 1 annual_return (1 total_return) ** (252 / len(nav)) - 1 # 年化交易日按252计 volatility returns.std() * np.sqrt(252) # 年化波动率 if volatility 0: sharpe_ratio annual_return / volatility # 夏普比率无风险利率设为0简化 else: sharpe_ratio np.nan # 最大回撤 cummax nav.cummax() drawdown (nav - cummax) / cummax max_drawdown drawdown.min() results { nav: nav, returns: returns, position: position, cash: cash, total_return: total_return, annual_return: annual_return, volatility: volatility, sharpe_ratio: sharpe_ratio, max_drawdown: max_drawdown, drawdown: drawdown } return results # 假设已有 price_df 和 signal_df # results simple_backtest(price_df, signal_df)5. 运行结果分析与可视化回测完成后我们需要直观地看到策略表现。我们将策略净值与简单买入持有例如沪深300ETF进行对比。# 文件visualization.py import matplotlib.pyplot as plt import matplotlib.dates as mdates def plot_backtest_results(results, benchmark_price_series, benchmark_nameBenchmark): 绘制回测结果图 :param results: 回测引擎返回的结果字典 :param benchmark_price_series: 基准价格序列如沪深300ETF价格 :param benchmark_name: 基准名称 nav results[nav] drawdown results[drawdown] # 将基准价格归一化以便与策略净值比较 benchmark_nav benchmark_price_series / benchmark_price_series.iloc[0] * nav.iloc[0] fig, axes plt.subplots(2, 1, figsize(14, 10), sharexTrue) # 图1净值曲线对比 ax1 axes[0] ax1.plot(nav.index, nav.values, labelfRotation Strategy (Ann.Ret{results[annual_return]:.2%}, Sharpe{results[sharpe_ratio]:.2f}), linewidth2) ax1.plot(benchmark_nav.index, benchmark_nav.values, labelf{benchmark_name} (Buy Hold), linestyle--, linewidth1.5) ax1.set_ylabel(Net Asset Value) ax1.set_title(ETF Rotation Strategy vs Buy Hold) ax1.legend() ax1.grid(True, whichmajor, linestyle--, linewidth0.5, alpha0.7) # 图2回撤曲线 ax2 axes[1] ax2.fill_between(drawdown.index, 0, drawdown.values * 100, colorred, alpha0.3) ax2.plot(drawdown.index, drawdown.values * 100, colordarkred, linewidth1) ax2.set_ylabel(Drawdown (%)) ax2.set_xlabel(Date) ax2.set_title(fStrategy Drawdown (Max DD{results[max_drawdown]:.2%})) ax2.grid(True, whichmajor, linestyle--, linewidth0.5, alpha0.7) # 格式化x轴日期 ax2.xaxis.set_major_formatter(mdates.DateFormatter(%Y-%m)) ax2.xaxis.set_major_locator(mdates.YearLocator()) plt.xticks(rotation45) plt.tight_layout() plt.show() # 打印关键指标 print(*50) print(策略绩效概览) print(*50) print(f累计收益率: {results[total_return]:.2%}) print(f年化收益率: {results[annual_return]:.2%}) print(f年化波动率: {results[volatility]:.2%}) print(f夏普比率: {results[sharpe_ratio]:.2f}) print(f最大回撤: {results[max_drawdown]:.2%}) print(*50) # 假设 benchmark_price 是沪深300ETF的价格序列 # plot_backtest_results(results, benchmark_price, CSI 300 ETF)6. 策略的潜在问题与优化方向没有一个策略是完美的。在激动地看到回测曲线之前我们必须清醒地认识到这个框架的局限性。6.1 常见问题与陷阱过度拟合Overfitting这是量化策略最大的敌人。你可能通过反复测试找到一组在历史数据上表现完美的参数如均线周期20/120调仓周期每月但这组参数在未来很可能失效。应对使用样本外数据测试、进行参数敏感性分析、避免使用过多复杂规则。幸存者偏差Survivorship Bias我们使用的ETF都是目前存活且活跃的但回测期初可能有些ETF还没上市或者有些已退市。这会导致回测结果过于乐观。应对使用“点时间”数据确保回测中只使用在当时已上市的数据。交易成本与冲击成本我们的简单回测假设可以立即以收盘价成交且佣金费率固定。现实中大额交易会产生冲击成本调仓时可能无法理想成交。应对在回测中引入更复杂的交易模型如按开盘价成交并加入固定的滑点如0.1%。数据透视偏差Look-ahead Bias在计算信号时错误地使用了未来的数据。例如在T日收盘后我们只能用T日及之前的数据计算T日的信号而不能用到T1日的开盘价。应对确保所有计算在时间轴上是严格滞后的。上面的示例代码已注意此点。策略同质化与失效如果一个简单的轮动策略被广泛知晓和使用其有效性可能会下降因为大量的资金涌入会侵蚀策略的Alpha超额收益。6.2 策略优化与进阶思路如果你理解了基础框架并想进一步探索可以考虑以下方向信号引擎升级多因子综合不止看动量可以加入估值PE/PB、波动率、换手率等其他因子进行加权打分。机器学习模型使用逻辑回归、随机森林等模型基于多维度特征预测下一期最优资产。但需严防过拟合。风险预算模型不单纯追求收益最高而是控制组合的整体风险向风险平价Risk Parity思路靠拢。资产池扩展增加资产类别纳入海外债、原油、REITs等更多不相关资产。细分行业ETF在权益内部轮动到不同的行业ETF如消费、科技、医药、金融。风控机制强化动态止损当持有资产从入场点回撤超过一定幅度时强制平仓。波动率控制在市场整体波动率急剧放大时降低仓位或转向更保守的资产。相关性监控当资产池内资产相关性突然升高时同涨同跌意味着分散失效需触发风控。仓位管理优化不止持有一只可以持有排名前2或前3的资产并动态分配权重如按动量大小分配。金字塔加减仓在趋势确认后分批加仓在趋势减弱时分批减仓。7. 最佳实践与工程化建议如果你想将此类策略用于实盘或更严肃的研究以下几点至关重要数据质量是生命线使用付费的、稳定的数据源如聚宽、万得、Tushare Pro确保数据的准确性和完整性除权除息、停牌等。建立本地数据缓存避免每次回测都重复请求网络数据。回测框架要严谨使用成熟的回测框架如Zipline,Backtrader,Qlib它们处理了众多细节成交顺序、税费、涨跌停等。进行前向分析Walk Forward Analysis, WFA将历史数据分成多个训练集和测试集滚动优化和测试检验策略的稳健性。参数敏感性分析对关键参数均线周期、动量窗口、调仓频率进行网格搜索观察策略绩效是否在参数小范围变动时发生剧烈变化。如果变化剧烈说明策略不稳定。日志与监控在策略中植入详细的日志记录记录每一次调仓的逻辑、信号值、成交情况。实盘运行时需要有独立的监控程序定期检查策略状态、持仓与账户的一致性。永远敬畏市场任何量化模型都是对历史规律的总结。市场结构会变黑天鹅事件会发生。实盘前必须经过长时间的模拟盘Paper Trading验证。投入实盘的资金应是你能承受损失的部分并做好策略阶段性失效的准备。“全天候五福ETF轮动策略”为我们提供了一个从零开始构建系统化投资框架的优秀范本。它的价值不在于提供一个“圣杯”而在于展示了一种用规则战胜情绪、用分散应对不确定、用回溯检验想法的理性投资方法论。通过本文的拆解你应该已经掌握了从数据获取、信号生成、回测验证到风险分析的完整链条。下一步我建议你亲手复现使用稳定的数据源将本文的代码跑通观察策略在不同历史阶段如牛市、熊市、震荡市的表现。大胆修改尝试更换资产池、修改信号规则比如用RSI替代均线过滤、调整调仓周期感受策略绩效的变化。深入思考这个策略最大的弱点可能在哪里在什么样的市场环境下它会持续亏损如何改进才能让它更适应未来的市场量化交易的世界没有终点每一个策略的终点都是新的起点。希望这个“五福”框架能成为你探索这个迷人领域的第一块坚实跳板。