ETF双因子轮动策略：动量与质量因子的Python实现与回测

你是不是经常看到券商研报里各种高大上的量化策略感觉逻辑清晰、收益诱人但真要自己动手实现时却无从下手今天这篇文章我要帮你彻底解决这个问题。我们将完整复现一个经典的ETF双因子轮动策略从数据获取到策略回测再到实盘信号生成提供全套可运行的Python源码。这个策略的核心逻辑很简单动量收益趋势 质量趋势稳定性/低波动双维度筛选ETF结合加权仓位分配在市场不同阶段动态调整持仓。但真正有价值的不只是代码本身而是整个工程化实现的思路。很多人在复现研报策略时容易踩的坑数据源不稳定、因子计算有偏差、回测逻辑不严谨、实盘对接复杂。本文将逐一拆解这些难点让你不仅能跑通这个策略更能掌握量化策略开发的完整方法论。1. 这篇文章真正要解决的问题为什么券商研报的策略看起来很美但自己就是复现不出来问题通常出在三个层面数据层面研报往往使用清洗过的理想数据而真实环境中你需要自己处理数据缺失、复权、停牌等问题。比如ETF的规模数据、净值数据如何稳定获取工程层面策略逻辑转化为可维护的代码需要合理的架构设计。单因子计算、多因子合成、权重分配、风控规则这些模块如何组织才能既清晰又高效回测层面简单的收益率计算很容易但真实的交易成本、滑点、调仓频率等因素会显著影响最终收益。如何构建一个可信的回测系统本文将基于Python生态中成熟的工具链pandas、akshare、backtrader等提供一个工业级的解决方案。适合有一定Python基础想要系统学习量化策略开发的读者。2. 基础概念与核心原理2.1 什么是ETF双因子轮动双因子轮动策略的核心思想是通过两个维度的指标来评估ETF的投资价值并定期调整持仓。动量因子衡量资产的价格趋势强度。通常用过去一段时间如20日、60日的收益率来表示。动量效应基于行为金融学中的追涨杀跌心理趋势一旦形成往往会持续。质量因子衡量趋势的稳定性和风险调整后的收益质量。常用波动率、夏普比率、最大回撤等指标。质量因子帮助过滤掉高波动、高风险的趋势选择更稳健的标的。2.2 策略核心逻辑拆解策略的完整流程包括标的池选择确定参与轮动的ETF范围如宽基指数、行业ETF等因子计算定期计算每个ETF的动量因子和质量因子综合评分将两个因子标准化后加权求和得到综合得分权重分配根据得分分配仓位得分越高权重越大调仓执行定期如月度调整持仓至新的权重分配这种方法的优势在于牛市时动量因子主导抓住强势品种震荡市时质量因子主导控制回撤风险。3. 环境准备与前置条件3.1 Python环境要求推荐使用Python 3.8版本主要依赖库包括# 创建conda环境可选 conda create -n quant_strategy python3.8 conda activate quant_strategy # 安装核心依赖 pip install pandas1.5.3 pip install numpy1.24.3 pip install akshare1.10.0 # 免费金融数据接口 pip install backtrader1.9.78.123 # 回测框架 pip install matplotlib3.7.1 # 绘图 pip install requests2.31.0 # 网络请求3.2 数据源配置我们使用akshare作为主要数据源它提供了丰富的免费金融数据接口import akshare as ak import pandas as pd import numpy as np from datetime import datetime, timedelta import warnings warnings.filterwarnings(ignore) # 测试数据接口是否正常 def test_data_source(): try: # 测试获取ETF基本信息 etf_info ak.fund_etf_spot_em() print(f成功获取{len(etf_info)}只ETF基本信息) return True except Exception as e: print(f数据接口测试失败: {e}) return False if __name__ __main__: test_data_source()4. ETF数据获取与处理模块4.1 构建ETF标的池首先需要确定策略操作的ETF范围。我们选择流动性好、规模较大的宽基指数ETFclass ETFDataManager: def __init__(self): self.etf_pool { 510300: 沪深300ETF, # 沪深300 510500: 中证500ETF, # 中证500 159915: 创业板ETF, # 创业板 512100: 中证1000ETF, # 中证1000 518880: 黄金ETF, # 黄金防御性品种 511010: 国债ETF, # 国债防御性品种 } def get_etf_daily_data(self, etf_code, start_date, end_date): 获取单只ETF的日线数据 try: # 通过akshare获取ETF日线数据 df ak.fund_etf_hist_em(symboletf_code, period每日, start_datestart_date, end_dateend_date) if df.empty: return None # 数据清洗和格式标准化 df[日期] pd.to_datetime(df[日期]) df.set_index(日期, inplaceTrue) df.sort_index(inplaceTrue) # 重命名列以便后续处理 df.rename(columns{ 开盘: open, 收盘: close, 最高: high, 最低: low, 成交量: volume }, inplaceTrue) return df[[open, high, low, close, volume]] except Exception as e: print(f获取{etf_code}数据失败: {e}) return None def get_all_etf_data(self, start_date20200101, end_date20231231): 获取所有ETF标的的历史数据 all_data {} for etf_code in self.etf_pool.keys(): print(f正在获取{etf_code}数据...) data self.get_etf_daily_data(etf_code, start_date, end_date) if data is not None and len(data) 0: all_data[etf_code] data # 添加延时避免请求过于频繁 import time time.sleep(0.5) print(f成功获取{len(all_data)}只ETF数据) return all_data # 测试数据获取 if __name__ __main__: data_manager ETFDataManager() etf_data data_manager.get_all_etf_data(20230101, 20231231) for code, data in etf_data.items(): print(f{code}: {len(data)}个交易日数据)4.2 数据质量检查与清洗真实数据往往存在各种问题需要进行严格的清洗def data_quality_check(etf_data): ETF数据质量检查与清洗 cleaned_data {} for etf_code, df in etf_data.items(): # 检查数据完整性 original_length len(df) # 1. 去除重复日期 df df[~df.index.duplicated(keepfirst)] # 2. 处理缺失值前后填充 df df.fillna(methodffill).fillna(methodbfill) # 3. 检查价格合理性去除明显异常值 price_cols [open, high, low, close] for col in price_cols: # 价格应为正数且在一定范围内 df df[df[col] 0] df df[df[col] 100] # 假设ETF价格不会超过100元 # 4. 确保时间序列连续性 all_dates pd.date_range(startdf.index.min(), enddf.index.max(), freqD) df df.reindex(all_dates) df df.fillna(methodffill) # 向前填充非交易日 print(f{etf_code}: 原始{original_length}条, 清洗后{len(df)}条) cleaned_data[etf_code] df return cleaned_data5. 双因子计算引擎5.1 动量因子计算动量因子我们采用经典的时间序列动量计算方法class FactorCalculator: def __init__(self, lookback_periods[20, 60, 120]): self.lookback_periods lookback_periods def calculate_momentum_factor(self, price_series): 计算动量因子 - 多个时间窗口的收益率 momentum_factors {} for period in self.lookback_periods: if len(price_series) period: continue # 当前价格 / period天前价格 - 1 momentum (price_series / price_series.shift(period) - 1) momentum_factors[fmomentum_{period}] momentum return pd.DataFrame(momentum_factors) def calculate_quality_factor(self, price_series, window20): 计算质量因子 - 基于波动率和风险调整收益 if len(price_series) window: return pd.DataFrame() # 1. 波动率年化 returns price_series.pct_change() volatility returns.rolling(windowwindow).std() * np.sqrt(252) # 2. 夏普比率简化版假设无风险收益为0 sharpe_ratio returns.rolling(windowwindow).mean() / returns.rolling(windowwindow).std() sharpe_ratio sharpe_ratio * np.sqrt(252) # 3. 最大回撤滚动窗口 rolling_max price_series.rolling(windowwindow, min_periods1).max() drawdown (price_series - rolling_max) / rolling_max quality_factors { volatility: volatility, sharpe_ratio: sharpe_ratio, max_drawdown: drawdown } return pd.DataFrame(quality_factors) def calculate_comprehensive_score(self, etf_data): 计算综合得分动量因子 质量因子 all_scores {} for etf_code, df in etf_data.items(): price_series df[close] # 计算动量因子 momentum_df self.calculate_momentum_factor(price_series) # 计算质量因子 quality_df self.calculate_quality_factor(price_series) if momentum_df.empty or quality_df.empty: continue # 因子标准化Z-score factors_df pd.concat([momentum_df, quality_df], axis1) factors_normalized (factors_df - factors_df.mean()) / factors_df.std() # 因子方向调整我们希望动量越大越好波动率越小越好 factors_normalized[volatility] -factors_normalized[volatility] # 波动率取负 factors_normalized[max_drawdown] -factors_normalized[max_drawdown] # 回撤取负 # 综合得分等权重 factors_normalized[composite_score] factors_normalized.mean(axis1) all_scores[etf_code] factors_normalized[composite_score] return pd.DataFrame(all_scores) # 测试因子计算 if __name__ __main__: # 生成示例数据测试 dates pd.date_range(2023-01-01, 2023-12-31, freqD) sample_data pd.DataFrame({ close: np.cumsum(np.random.randn(len(dates)) * 0.01) 100 }, indexdates) calculator FactorCalculator() momentum_factors calculator.calculate_momentum_factor(sample_data[close]) quality_factors calculator.calculate_quality_factor(sample_data[close]) print(动量因子示例:) print(momentum_factors.tail()) print(\n质量因子示例:) print(quality_factors.tail())5.2 因子合成与权重分配多个因子需要合理合成才能产生更好的效果class WeightAllocator: def __init__(self, top_n3, min_weight0.1): self.top_n top_n # 选择前N只ETF self.min_weight min_weight # 最小权重 def calculate_weights(self, scores_df): 基于综合得分计算权重分配 weights_df pd.DataFrame(indexscores_df.index, columnsscores_df.columns) for date in scores_df.index: daily_scores scores_df.loc[date].dropna() if len(daily_scores) 0: continue # 选择得分最高的top_n只ETF top_etfs daily_scores.nlargest(self.top_n) if len(top_etfs) 0: continue # 基于得分计算权重得分越高权重越大 total_score top_etfs.sum() weights top_etfs / total_score # 确保每只ETF至少有最小权重 if len(weights) 0: min_adj_weights weights.clip(lowerself.min_weight) # 重新归一化 adj_weights min_adj_weights / min_adj_weights.sum() for etf, weight in adj_weights.items(): weights_df.loc[date, etf] weight return weights_df.fillna(0) def generate_rebalance_dates(self, start_date, end_date, freqM): 生成调仓日期月末调仓 dates pd.date_range(startstart_date, endend_date, freqfreq) # 调整为月末 rebalance_dates [date.replace(day1) pd.offsets.MonthEnd(0) for date in dates] return rebalance_dates # 权重分配测试 def test_weight_allocation(): # 模拟得分数据 dates pd.date_range(2023-01-01, 2023-03-01, freqD) etf_codes [510300, 510500, 159915, 512100] # 生成随机得分 np.random.seed(42) scores_data np.random.randn(len(dates), len(etf_codes)) * 0.1 0.5 scores_df pd.DataFrame(scores_data, indexdates, columnsetf_codes) allocator WeightAllocator(top_n2, min_weight0.2) weights allocator.calculate_weights(scores_df) print(权重分配示例:) print(weights.tail(10)) if __name__ __main__: test_weight_allocation()6. 回测系统实现6.1 基于Backtrader的回测引擎使用成熟的backtrader框架构建回测系统import backtrader as bt import backtrader.analyzers as btanalyzers class DualFactorRotationStrategy(bt.Strategy): params ( (rebalance_freq, 21), # 调仓频率交易日 (lookback_period, 60), # 回看周期 (top_n, 3), # 持仓数量 ) def __init__(self): self.etf_data {} # 存储各ETF数据 self.rebalance_day 0 self.weights_df None # 权重数据 # 记录交易日志 self.trade_log [] def next(self): # 调仓逻辑 if len(self.data) % self.params.rebalance_freq 0: self.rebalance_portfolio() def rebalance_portfolio(self): 执行调仓操作 current_date self.data.datetime.date(0) if self.weights_df is None or current_date not in self.weights_df.index: return # 获取当前日期权重 daily_weights self.weights_df.loc[current_date] # 清空现有持仓 for data in self.datas: self.close(data) # 按新权重建仓 portfolio_value self.broker.getvalue() for etf_code, weight in daily_weights.items(): if weight 0.01: # 权重大于1%才交易 data self.get_etf_data(etf_code) if data is not None: target_value portfolio_value * weight price data.close[0] size int(target_value / price) if size 0: self.buy(datadata, sizesize) # 记录交易 self.trade_log.append({ date: current_date, etf: etf_code, action: BUY, size: size, price: price, weight: weight }) def get_etf_data(self, etf_code): 根据代码获取对应的data对象 for data in self.datas: if data._name etf_code: return data return None def run_backtest(etf_data, weights_df, initial_cash1000000): 运行回测 cerebro bt.Cerebro() cerebro.broker.setcash(initial_cash) # 添加交易数据 for etf_code, data in etf_data.items(): # 将DataFrame转换为backtrader数据格式 data_df data.reset_index() data_df[datetime] pd.to_datetime(data_df[日期]) data_df.set_index(datetime, inplaceTrue) data_feed bt.feeds.PandasData( datanamedata_df, datetimeNone, # 已经设置为index openopen, highhigh, lowlow, closeclose, volumevolume ) cerebro.adddata(data_feed, nameetf_code) # 添加策略 cerebro.addstrategy(DualFactorRotationStrategy, weights_dfweights_df) # 添加分析器 cerebro.addanalyzer(btanalyzers.SharpeRatio, _namesharpe) cerebro.addanalyzer(btanalyzers.DrawDown, _namedrawdown) cerebro.addanalyzer(btanalyzers.Returns, _namereturns) cerebro.addanalyzer(btanalyzers.TradeAnalyzer, _nametrades) # 运行回测 print(初始资金: %.2f % cerebro.broker.getvalue()) results cerebro.run() print(最终资金: %.2f % cerebro.broker.getvalue()) # 返回回测结果 strat results[0] return { sharpe: strat.analyzers.sharpe.get_analysis(), drawdown: strat.analyzers.drawdown.get_analysis(), returns: strat.analyzers.returns.get_analysis(), trades: strat.analyzers.trades.get_analysis(), trade_log: strat.trade_log } # 回测结果分析 def analyze_backtest_results(results): 分析回测结果 sharpe results[sharpe] drawdown results[drawdown] returns results[returns] trades results[trades] print(\n 回测结果分析 ) print(f年化夏普比率: {sharpe.get(sharperatio, 0):.3f}) print(f最大回撤: {drawdown.get(max, {}).get(drawdown, 0):.2f}%) print(f年化收益率: {returns.get(rnorm100, 0):.2f}%) print(f总交易次数: {trades.get(total, {}).get(total, 0)}) # 绘制交易日志 trade_df pd.DataFrame(results[trade_log]) if not trade_df.empty: print(f\n最近5次交易:) print(trade_df.tail())6.2 完整的策略回测流程将各个模块组合成完整的回测流水线def complete_strategy_backtest(start_date20200101, end_date20231231): 完整的策略回测流程 print( ETF双因子轮动策略回测开始 ) # 1. 数据获取 print(步骤1: 获取ETF数据...) data_manager ETFDataManager() raw_etf_data data_manager.get_all_etf_data(start_date, end_date) # 2. 数据清洗 print(步骤2: 数据质量检查与清洗...) cleaned_etf_data data_quality_check(raw_etf_data) # 3. 因子计算 print(步骤3: 计算双因子...) factor_calculator FactorCalculator() scores_df factor_calculator.calculate_comprehensive_score(cleaned_etf_data) # 4. 权重分配 print(步骤4: 计算权重分配...) weight_allocator WeightAllocator(top_n3, min_weight0.1) weights_df weight_allocator.calculate_weights(scores_df) # 5. 回测执行 print(步骤5: 执行回测...) results run_backtest(cleaned_etf_data, weights_df) # 6. 结果分析 print(步骤6: 分析回测结果...) analyze_backtest_results(results) return { etf_data: cleaned_etf_data, scores_df: scores_df, weights_df: weights_df, backtest_results: results } if __name__ __main__: # 运行完整回测使用较短时间范围测试 results complete_strategy_backtest(20230101, 20231231)7. 实盘信号生成与监控7.1 生成交易信号回测验证有效后可以生成实盘交易信号class TradingSignalGenerator: def __init__(self, model_pathNone): self.factor_calculator FactorCalculator() self.weight_allocator WeightAllocator() def generate_signals(self, etf_data): 生成最新的交易信号 # 计算最新因子得分 scores_df self.factor_calculator.calculate_comprehensive_score(etf_data) if scores_df.empty: return None # 获取最新得分 latest_scores scores_df.iloc[-1].dropna() # 计算权重 latest_weights self.weight_allocator.calculate_weights( pd.DataFrame([latest_scores]) ).iloc[0] # 生成交易信号 signals {} for etf_code, weight in latest_weights.items(): if weight 0.01: # 权重超过1%生成信号 signals[etf_code] { action: BUY if weight 0.05 else HOLD, target_weight: round(weight, 4), score: round(latest_scores.get(etf_code, 0), 4) } return signals def generate_signal_report(self, signals, etf_data): 生成信号报告 if not signals: return 暂无有效交易信号 report [ ETF双因子轮动策略交易信号 ] report.append(f生成时间: {datetime.now().strftime(%Y-%m-%d %H:%M:%S)}) report.append(\n推荐持仓:) for etf_code, signal in signals.items(): if signal[action] BUY: latest_price etf_data[etf_code][close].iloc[-1] report.append( f- {etf_code}: {signal[action]} | f目标权重: {signal[target_weight]*100}% | f综合得分: {signal[score]} | f最新价格: {latest_price:.3f} ) return \n.join(report) # 实时信号生成示例 def demo_realtime_signals(): 演示实时信号生成 data_manager ETFDataManager() # 获取最近一年的数据模拟实时环境 end_date datetime.now().strftime(%Y%m%d) start_date (datetime.now() - timedelta(days365)).strftime(%Y%m%d) etf_data data_manager.get_all_etf_data(start_date, end_date) cleaned_data data_quality_check(etf_data) signal_generator TradingSignalGenerator() signals signal_generator.generate_signals(cleaned_data) if signals: report signal_generator.generate_signal_report(signals, cleaned_data) print(report) else: print(无法生成有效交易信号) if __name__ __main__: demo_realtime_signals()7.2 策略监控与调优实盘运行需要持续的监控和优化class StrategyMonitor: def __init__(self): self.performance_history [] def track_daily_performance(self, date, portfolio_value, signals): 每日业绩跟踪 self.performance_history.append({ date: date, portfolio_value: portfolio_value, signals: signals.copy() if signals else {} }) def analyze_strategy_stability(self, window30): 分析策略稳定性 if len(self.performance_history) window: return 数据不足需要更多历史记录 recent_perf self.performance_history[-window:] values [item[portfolio_value] for item in recent_perf] returns np.diff(values) / values[:-1] volatility np.std(returns) * np.sqrt(252) sharpe np.mean(returns) / np.std(returns) * np.sqrt(252) if np.std(returns) 0 else 0 return { 近期波动率: f{volatility:.3f}, 近期夏普比率: f{sharpe:.3f}, 收益稳定性: 良好 if volatility 0.2 else 需关注 } def generate_monitoring_report(self): 生成监控报告 if not self.performance_history: return 暂无监控数据 latest self.performance_history[-1] stability self.analyze_strategy_stability() report [ 策略监控报告 , f报告时间: {datetime.now().strftime(%Y-%m-%d %H:%M:%S)}, f最新净值: {latest[portfolio_value]:,.2f}, f信号数量: {len(latest.get(signals, {}))}, \n策略稳定性分析: ] for key, value in stability.items(): report.append(f- {key}: {value}) return \n.join(report)8. 常见问题与排查思路在实现过程中可能会遇到的各种问题及解决方案问题现象可能原因排查方式解决方案数据获取失败网络问题/接口变更检查akshare版本和网络连接更新akshare添加重试机制因子计算异常数据缺失或格式错误检查数据质量和长度增加数据验证步骤处理边界情况回测结果异常前视偏差或逻辑错误检查时间戳对齐情况确保使用历史数据进行计算权重分配不合理因子极端值影响检查因子分布情况添加因子缩尾处理交易频率过高调仓周期设置过短分析交易成本影响调整调仓频率添加过滤条件8.1 典型错误示例与修正错误示例1前视偏差# 错误写法使用未来数据 def wrong_factor_calculation(df): # 使用当天的收盘价计算动量错误 df[momentum] df[close] / df[close].shift(20) - 1 return df # 正确写法使用历史数据 def correct_factor_calculation(df): # 使用前一天的收盘价计算动量正确 df[momentum] df[close].shift(1) / df[close].shift(21) - 1 return df错误示例2数据未对齐# 错误写法直接合并不同时间序列 def wrong_data_merge(df1, df2): merged pd.merge(df1, df2, left_indexTrue, right_indexTrue) # 可能丢失数据 return merged # 正确写法先确保时间索引对齐 def correct_data_merge(df1, df2): # 使用相同的时间索引 common_index df1.index.intersection(df2.index) merged pd.merge(df1.loc[common_index], df2.loc[common_index], left_indexTrue, right_indexTrue) return merged9. 最佳实践与工程建议9.1 代码组织规范良好的项目结构是策略可持续迭代的基础etf_dual_factor/ ├── data/ # 数据模块 │ ├── manager.py # 数据管理 │ └── validator.py # 数据验证 ├── factors/ # 因子模块 │ ├── calculator.py # 因子计算 │ └── composite.py # 因子合成 ├── portfolio/ # 组合模块 │ ├── allocator.py # 权重分配 │ └── rebalance.py # 调仓逻辑 ├── backtest/ # 回测模块 │ ├── engine.py # 回测引擎 │ └── analyzer.py # 结果分析 ├── live/ # 实盘模块 │ ├── signals.py # 信号生成 │ └── monitor.py # 策略监控 └── config/ # 配置模块 ├── settings.py # 参数配置 └── constants.py # 常量定义9.2 参数优化与验证策略参数需要系统化的优化流程class ParameterOptimizer: def __init__(self, etf_data): self.etf_data etf_data def grid_search_parameters(self, param_grid): 网格搜索最优参数 best_params None best_sharpe -float(inf) results [] # 参数组合遍历 for lookback in param_grid[lookback_periods]: for top_n in param_grid[top_n_values]: for min_weight in param_grid[min_weights]: # 使用当前参数运行回测 try: factor_calc FactorCalculator(lookback_periods[lookback]) scores_df factor_calc.calculate_comprehensive_score(self.etf_data) weight_alloc WeightAllocator(top_ntop_n, min_weightmin_weight) weights_df weight_alloc.calculate_weights(scores_df) backtest_results run_backtest(self.etf_data, weights_df) sharpe backtest_results[sharpe].get(sharperatio, -999) results.append({ params: {lookback: lookback, top_n: top_n, min_weight: min_weight}, sharpe: sharpe }) if sharpe best_sharpe: best_sharpe sharpe best_params {lookback: lookback, top_n: top_n, min_weight: min_weight} except Exception as e: print(f参数组合失败: {e}) continue return best_params, best_sharpe, results # 参数优化示例 def demo_parameter_optimization(): 演示参数优化过程 data_manager ETFDataManager() etf_data data_manager.get_all_etf_data(20220101, 20231231) cleaned_data data_quality_check(etf_data) optimizer ParameterOptimizer(cleaned_data) param_grid { lookback_periods: [20, 60, 120], top_n_values: [2, 3, 4], min_weights: [0.1, 0.15, 0.2] } best_params, best_sharpe, all_results optimizer.grid_search_parameters(param_grid) print(f最优参数: {best_params}) print(f最优夏普比率: {best_sharpe:.3f})9.3 风险控制与资金管理实盘交易必须重视风险控制class RiskManager: def __init__(self, max_position_size0.3, max_drawdown0.15): self.max_position_size max_position_size # 单标的最大仓位 self.max_drawdown max_drawdown # 最大回撤阈值 def validate_position_weights(self, weights): 验证仓位权重合理性 # 检查单标的风险 for etf, weight in weights.items(): if weight self.max_position_size: print(f警告: {etf}权重{weight:.1%}超过限制{self.max_position_size:.1%}) return False # 检查集中度风险 top_weights sorted(weights.values(), reverseTrue)[:2] if len(top_weights) 2 and sum(top_weights) 0.6: print(f警告: 前两大权重合计{sum(top_weights):.1%}集中度偏高) return False return True def check_drawdown_breach(self, current_drawdown): 检查回撤是否超限 if current_drawdown self.max_drawdown: print(f紧急: 当前回撤{current_drawdown:.1%}超过阈值{self.max_drawdown:.1%}) return True return False def apply_emergency_measures(self, current_drawdown): 执行紧急风控措施 if self.check_drawdown_breach(current_drawdown):