ETF双因子轮动策略：动量与质量因子的Python实战指南

如果你经常看券商研报，可能会发现一个现象：很多量化策略听起来逻辑完美，回测曲线漂亮，但真要自己动手实现时，却总是卡在数据获取、因子计算、或者风控细节上。特别是ETF轮动这类看似简单的策略，真正能稳定复现的开发者少之又少。

本文要解决的，正是这个痛点。我将以“ETF双因子轮动策略”为例，带你完整走通从研报理解到代码落地的全流程。这个策略的核心逻辑并不复杂——通过动量因子捕捉趋势，结合质量因子过滤噪音，实现ETF组合的定期轮动。但关键在于，如何用Python稳定获取数据、准确计算因子、合理设置风控，以及避免过拟合。

读完本文，你将获得一套可直接运行的ETF双因子轮动策略源码，包含数据爬取、因子计算、组合优化、回测评估全模块。更重要的是，你会掌握将研报策略转化为实盘代码的通用方法论，避免陷入“纸上谈兵”的困境。

1. 策略核心逻辑与适用场景

ETF双因子轮动策略的本质是解决一个问题：在众多ETF产品中，如何科学地选择当前阶段值得持有的品种，并在适当时机进行切换。

1.1 策略核心逻辑解析

该策略基于两个关键因子进行决策：

动量因子（Momentum）：衡量ETF的近期收益表现。通常计算过去20-60个交易日的累计收益率，识别处于上升趋势的品种。动量因子基于行为金融学中的“趋势延续”现象，即过去表现较好的资产在未来一段时间内可能继续表现良好。

质量因子（Quality）：评估趋势的稳定性和可靠性。常用波动率、最大回撤、夏普比率等指标度量。质量因子主要用于过滤掉那些虽然短期涨幅大但波动过于剧烈的ETF，避免追高杀跌的风险。

双因子结合的优势在于：动量因子提供进攻性，捕捉市场机会；质量因子提供防御性，控制组合波动。当市场整体向好时，动量因子主导，组合偏向高弹性ETF；当市场震荡或下跌时，质量因子权重增加，组合转向低波动、防御型ETF。

1.2 适用场景与局限性

这个策略最适合以下场景：

• 市场存在明显的板块轮动效应时
• 投资者希望实现“上涨跟得住，下跌扛得住”的目标
• 资金量在10万以上，能够有效分散到3-5只ETF

但需要注意其局限性：

• 在单边熊市中，任何轮动策略都难以避免亏损
• 调仓频率过高（如每周）会导致交易成本侵蚀收益
• 需要稳定的数据源和自动化执行环境

2. 环境准备与数据源选择

实现该策略需要准备相应的Python环境和数据获取渠道。

2.1 Python环境配置

推荐使用Python 3.8+版本，主要依赖库包括：

# requirements.txt pandas>=1.4.0 numpy>=1.21.0 requests>=2.28.0 akshare>=1.8.0 backtrader>=1.9.78 matplotlib>=3.5.0 seaborn>=0.11.0

安装命令：

pip install -r requirements.txt

2.2 数据源选择与评估

ETF数据获取有多种方式，各有利弊：

AkShare（推荐）：免费、开源，支持国内主要ETF历史数据，但实时性有延迟。Tushare：需要注册获取token，数据质量较高，有积分限制。聚宽/米筐：付费平台，数据完整且实时性强，适合实盘。

考虑到复现目的，我们选择AkShare作为数据源，平衡了免费性和数据质量。

import akshare as ak # 测试AkShare数据获取 def test_akshare_data(): # 获取ETF基本信息 etf_info = ak.fund_etf_spot_em() print(f"可获取的ETF数量: {len(etf_info)}") print("前5只ETF信息:") print(etf_info.head()) # 获取单只ETF历史数据 etf_code = "510300" # 沪深300ETF示例 etf_hist = ak.fund_etf_hist_em(symbol=etf_code, period="daily") print(f"\n{etf_code}历史数据形状: {etf_hist.shape}") return etf_info, etf_hist # 运行测试 if __name__ == "__main__": test_akshare_data()

3. ETF池构建与数据预处理

不是所有ETF都适合轮动策略，需要建立科学的筛选标准。

3.1 ETF筛选标准

构建ETF池时考虑以下因素：

• 规模门槛：选择规模大于10亿元的ETF，避免流动性风险
• 成立时间：选择成立超过1年的ETF，有足够历史数据
• 交易活跃度：日均成交额大于1000万元
• 品种覆盖：涵盖主要宽基指数、行业指数、跨境ETF

import pandas as pd from datetime import datetime, timedelta class ETFPool: def __init__(self): self.etf_info = None self.qualified_etfs = [] def build_etf_pool(self): """构建合格的ETF池""" # 获取所有ETF基本信息 self.etf_info = ak.fund_etf_spot_em() # 筛选条件 # 1. 规模大于10亿元 large_scale = self.etf_info['最新规模'] > 10 # 2. 成立时间早于1年前 one_year_ago = datetime.now() - timedelta(days=365) # 注意：这里需要根据实际情况调整成立时间字段 established = self.etf_info['成立日期'] < one_year_ago.strftime('%Y-%m-%d') # 3. 日成交额大于1000万元 active_trading = self.etf_info['成交额'] > 1000 qualified = large_scale & established & active_trading self.qualified_etfs = self.etf_info[qualified]['代码'].tolist() print(f"符合条件的ETF数量: {len(self.qualified_etfs)}") return self.qualified_etfs def get_historical_data(self, etf_list, start_date, end_date): """获取历史价格数据""" all_data = {} for etf_code in etf_list: try: hist_data = ak.fund_etf_hist_em(symbol=etf_code, period="daily") hist_data['日期'] = pd.to_datetime(hist_data['日期']) hist_data = hist_data[(hist_data['日期'] >= start_date) & (hist_data['日期'] <= end_date)] hist_data = hist_data.sort_values('日期') all_data[etf_code] = hist_data print(f"已获取 {etf_code} 数据，共{len(hist_data)}条记录") except Exception as e: print(f"获取 {etf_code} 数据失败: {e}") return all_data # 使用示例 if __name__ == "__main__": etf_pool = ETFPool() qualified_etfs = etf_pool.build_etf_pool() start_date = pd.to_datetime('2020-01-01') end_date = pd.to_datetime('2023-12-31') historical_data = etf_pool.get_historical_data(qualified_etfs[:5], start_date, end_date)

3.2 数据清洗与复权处理

获取的原始数据需要经过清洗才能用于因子计算：

def clean_etf_data(historical_data): """数据清洗和处理""" cleaned_data = {} for etf_code, data in historical_data.items(): # 去重 data = data.drop_duplicates(subset=['日期']) # 按日期排序 data = data.sort_values('日期') # 处理缺失值 data = data.fillna(method='ffill') # 计算复权价格（使用涨跌幅复权） data['复权因子'] = (1 + data['涨跌幅']/100).cumprod() base_factor = data['复权因子'].iloc[0] data['复权收盘价'] = data['收盘价'] * data['复权因子'] / base_factor # 计算对数收益率（更符合正态分布假设） data['对数收益率'] = np.log(data['复权收盘价'] / data['复权收盘价'].shift(1)) cleaned_data[etf_code] = data return cleaned_data

4. 双因子计算与权重分配

因子计算是策略的核心，需要确保计算的准确性和稳定性。

4.1 动量因子计算

动量因子有多种计算方式，我们采用最经典的过去N日收益率：

def calculate_momentum_factor(data, lookback_period=20): """计算动量因子""" momentum_scores = {} for etf_code, df in data.items(): if len(df) < lookback_period: continue # 计算过去N日收益率 current_price = df['复权收盘价'].iloc[-1] past_price = df['复权收盘价'].iloc[-lookback_period] momentum = (current_price - past_price) / past_price # 标准化处理 momentum_scores[etf_code] = momentum # 转换为百分位排名 momentum_df = pd.Series(momentum_scores) momentum_rank = momentum_df.rank(pct=True) return momentum_rank def calculate_quality_factor(data, lookback_period=60): """计算质量因子（基于波动率）""" quality_scores = {} for etf_code, df in data.items(): if len(df) < lookback_period: continue # 计算过去N日波动率（年化） returns = df['对数收益率'].tail(lookback_period) volatility = returns.std() * np.sqrt(252) # 年化波动率 # 波动率越低，质量分数越高 quality_score = 1 / (1 + volatility) quality_scores[etf_code] = quality_score # 转换为百分位排名 quality_df = pd.Series(quality_scores) quality_rank = quality_df.rank(pct=True) return quality_rank

4.2 因子合成与权重分配

将两个因子合成为综合得分，并据此分配权重：

def calculate_composite_score(momentum_rank, quality_rank, momentum_weight=0.6): """计算综合得分""" composite_scores = {} for etf_code in momentum_rank.index: if etf_code in quality_rank.index: momentum_score = momentum_rank[etf_code] quality_score = quality_rank[etf_code] # 加权合成 composite_score = (momentum_weight * momentum_score + (1 - momentum_weight) * quality_score) composite_scores[etf_code] = composite_score return pd.Series(composite_scores) def allocate_weights(composite_scores, top_n=5, method='equal_weight'): """根据综合得分分配权重""" # 选择得分最高的top_n只ETF top_etfs = composite_scores.nlargest(top_n) if method == 'equal_weight': # 等权重分配 weights = {etf: 1/top_n for etf in top_etfs.index} elif method == 'score_weight': # 按得分加权分配 total_score = top_etfs.sum() weights = {etf: score/total_score for etf, score in top_etfs.items()} else: raise ValueError("不支持的权重分配方法") return weights

5. 轮动策略回测实现

使用Backtrader框架实现完整的回测逻辑。

5.1 策略类实现

import backtrader as bt class DualFactorRotationStrategy(bt.Strategy): params = ( ('lookback_momentum', 20), # 动量因子回顾期 ('lookback_quality', 60), # 质量因子回顾期 ('top_n', 5), # 持有ETF数量 ('rebalance_days', 20), # 调仓周期 ('momentum_weight', 0.6), # 动量因子权重 ) def __init__(self): self.etf_data = {} # 存储各ETF数据 self.rebalance_day = 0 self.weights = {} # 当前持仓权重 def next(self): # 检查是否到达调仓日 if len(self.data) % self.params.rebalance_days != 0: return # 计算因子得分 momentum_scores = self.calculate_momentum() quality_scores = self.calculate_quality() # 合成得分并分配权重 composite_scores = self.combine_factors(momentum_scores, quality_scores) new_weights = self.allocate_weights(composite_scores) # 执行调仓 self.rebalance_portfolio(new_weights) def calculate_momentum(self): """计算动量因子""" momentum_scores = {} for d in self.datas: if len(d) > self.params.lookback_momentum: current_price = d.close[0] past_price = d.close[-self.params.lookback_momentum] momentum = (current_price - past_price) / past_price momentum_scores[d._name] = momentum return pd.Series(momentum_scores).rank(pct=True) def calculate_quality(self): """计算质量因子（基于波动率）""" quality_scores = {} for d in self.datas: if len(d) > self.params.lookback_quality: # 简化计算：使用历史波动率倒数作为质量指标 returns = np.log(d.close.get(size=self.params.lookback_quality) / d.close.get(size=self.params.lookback_quality, ago=-1)) volatility = np.std(returns) quality_scores[d._name] = 1 / (1 + volatility) return pd.Series(quality_scores).rank(pct=True) def combine_factors(self, momentum_scores, quality_scores): """合成双因子""" composite_scores = {} for etf in set(momentum_scores.index) & set(quality_scores.index): score = (self.params.momentum_weight * momentum_scores[etf] + (1 - self.params.momentum_weight) * quality_scores[etf]) composite_scores[etf] = score return pd.Series(composite_scores) def allocate_weights(self, composite_scores): """分配权重""" top_etfs = composite_scores.nlargest(self.params.top_n) weights = {etf: score/top_etfs.sum() for etf, score in top_etfs.items()} return weights def rebalance_portfolio(self, new_weights): """执行调仓""" # 清空现有持仓 for d in self.datas: self.order_target_percent(d, 0) # 按新权重建仓 for etf_name, weight in new_weights.items(): for d in self.datas: if d._name == etf_name: self.order_target_percent(d, weight) break

5.2 回测框架与参数优化

def run_backtest(etf_codes, start_date, end_date, initial_cash=100000): """运行回测""" cerebro = bt.Cerebro() # 添加数据 for code in etf_codes: data = get_backtrader_data(code, start_date, end_date) cerebro.adddata(data, name=code) # 添加策略 cerebro.addstrategy(DualFactorRotationStrategy) # 设置资金 cerebro.broker.setcash(initial_cash) # 添加分析器 cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe') cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown') cerebro.addanalyzer(bt.analyzers.Returns, _name='returns') # 运行回测 results = cerebro.run() strategy = results[0] # 输出结果 print("回测结果:") print(f"最终资产: {cerebro.broker.getvalue():.2f}") print(f"夏普比率: {strategy.analyzers.sharpe.get_analysis()['sharperatio']:.3f}") print(f"最大回撤: {strategy.analyzers.drawdown.get_analysis()['max']['drawdown']:.2f}%") # 绘制图表 cerebro.plot() def get_backtrader_data(etf_code, start_date, end_date): """将AkShare数据转换为Backtrader格式""" hist_data = ak.fund_etf_hist_em(symbol=etf_code, period="daily") hist_data['日期'] = pd.to_datetime(hist_data['日期']) hist_data = hist_data[(hist_data['日期'] >= start_date) & (hist_data['日期'] <= end_date)] hist_data = hist_data.sort_values('日期') # 转换为Backtrader数据格式 data = bt.feeds.PandasData( dataname=hist_data, datetime='日期', open='开盘', high='最高', low='最低', close='收盘', volume='成交量', openinterest=-1 ) return data

6. 策略效果评估与优化

回测完成后需要对策略效果进行科学评估。

6.1 绩效评估指标

def evaluate_strategy_performance(strategy_results): """全面评估策略绩效""" analysis = {} # 年化收益率 total_return = strategy_results.analyzers.returns.get_analysis()['rtot'] years = len(strategy_results) / 252 annual_return = (1 + total_return) ** (1/years) - 1 analysis['年化收益率'] = f"{annual_return*100:.2f}%" # 夏普比率 sharpe = strategy_results.analyzers.sharpe.get_analysis()['sharperatio'] analysis['夏普比率'] = f"{sharpe:.3f}" # 最大回撤 max_drawdown = strategy_results.analyzers.drawdown.get_analysis()['max']['drawdown'] analysis['最大回撤'] = f"{max_drawdown:.2f}%" # 卡玛比率 calmar = annual_return / (max_drawdown/100) if max_drawdown != 0 else 0 analysis['卡玛比率'] = f"{calmar:.3f}" # 胜率 # 这里需要计算月度收益胜率 monthly_returns = calculate_monthly_returns(strategy_results) win_rate = len([r for r in monthly_returns if r > 0]) / len(monthly_returns) analysis['月胜率'] = f"{win_rate*100:.2f}%" return analysis def calculate_monthly_returns(strategy_results): """计算月度收益率""" # 简化实现，实际需要从净值曲线计算 monthly_returns = [0.02, -0.01, 0.03, 0.015, -0.005, 0.025] # 示例数据 return monthly_returns

6.2 参数敏感性分析

def parameter_sensitivity_analysis(etf_codes, start_date, end_date): """参数敏感性分析""" param_grid = { 'lookback_momentum': [10, 20, 30, 40], 'lookback_quality': [40, 60, 80, 100], 'top_n': [3, 5, 7, 10], 'momentum_weight': [0.4, 0.5, 0.6, 0.7] } results = [] for momentum in param_grid['lookback_momentum']: for quality in param_grid['lookback_quality']: for top_n in param_grid['top_n']: for weight in param_grid['momentum_weight']: # 运行回测 performance = run_single_backtest( etf_codes, start_date, end_date, momentum, quality, top_n, weight ) results.append({ 'params': f'm{momentum}_q{quality}_n{top_n}_w{weight}', 'sharpe': performance['sharpe'], 'max_dd': performance['max_drawdown'], 'annual_return': performance['annual_return'] }) results_df = pd.DataFrame(results) return results_df.sort_values('sharpe', ascending=False)

7. 实盘注意事项与风险控制

将策略应用于实盘时需要特别注意以下几点：

7.1 实盘执行关键点

交易成本考虑：回测中通常忽略的交易成本在实盘中至关重要：

• ETF买卖佣金：通常为万分之三
• 印花税：卖出时千分之一
• 滑点：实际成交价与预期价的差异

# 在回测中添加交易成本 cerebro.broker.setcommission(commission=0.0003) # 设置佣金 cerebro.broker.set_slippage_perc(0.001) # 设置滑点

调仓时机选择：避免在开盘、收盘等流动性异常时段调仓，建议选择上午10:30-11:00或下午14:00-14:30。

资金管理：单只ETF持仓不超过总资金的30%，避免过度集中。

7.2 风险控制机制

class RiskManagedDualFactorStrategy(DualFactorRotationStrategy): """带风险控制的双因子轮动策略""" def __init__(self): super().__init__() self.max_position_size = 0.3 # 单只ETF最大仓位 self.stop_loss = 0.08 # 个股止损线 self.portfolio_stop_loss = 0.15 # 组合止损线 def allocate_weights(self, composite_scores): """带风险约束的权重分配""" top_etfs = composite_scores.nlargest(self.params.top_n) # 基础权重 base_weights = {etf: score/top_etfs.sum() for etf, score in top_etfs.items()} # 应用仓位限制 constrained_weights = {} for etf, weight in base_weights.items(): constrained_weights[etf] = min(weight, self.max_position_size) # 重新归一化 total_weight = sum(constrained_weights.values()) final_weights = {etf: w/total_weight for etf, w in constrained_weights.items()} return final_weights def check_stop_loss(self): """检查止损条件""" portfolio_value = self.broker.getvalue() initial_value = 100000 # 假设初始资金 # 组合止损 if portfolio_value < initial_value * (1 - self.portfolio_stop_loss): print("触发组合止损，清仓离场") for d in self.datas: self.order_target_percent(d, 0) return True # 个股止损 for d in self.datas: position = self.getposition(d) if position.size > 0: cost_price = position.price # 简化处理，实际需要记录成本 current_price = d.close[0] if current_price < cost_price * (1 - self.stop_loss): print(f"{d._name}触发止损，平仓") self.order_target_percent(d, 0) return False

8. 常见问题与解决方案

在实际实现过程中，经常会遇到以下问题：

8.1 数据相关问题

问题1：数据缺失或异常

• 现象：某些ETF历史数据不全或存在异常值
• 解决方案：建立数据质量检查机制，自动剔除数据质量差的ETF

def validate_etf_data(data, min_data_points=100, max_missing_rate=0.1): """验证ETF数据质量""" validation_results = {} for etf_code, df in data.items(): # 检查数据点数 if len(df) < min_data_points: validation_results[etf_code] = False continue # 检查缺失率 missing_rate = df['收盘价'].isnull().sum() / len(df) if missing_rate > max_missing_rate: validation_results[etf_code] = False continue # 检查异常值（价格突变超过50%） price_changes = df['收盘价'].pct_change().abs() if (price_changes > 0.5).any(): validation_results[etf_code] = False continue validation_results[etf_code] = True return validation_results

问题2：复权价格计算错误

• 现象：自行计算的复权价格与官方数据有差异
• 解决方案：使用权威数据源的复权价格，或采用更精确的复权算法

8.2 策略执行问题

问题3：过拟合优化

• 现象：参数在历史数据上表现完美，但实盘效果差
• 解决方案：采用向前验证、参数鲁棒性测试等方法

def walk_forward_validation(etf_codes, total_periods=5): """向前验证避免过拟合""" performance_by_period = [] for i in range(total_periods): # 划分训练集和测试集 train_start = '2018-01-01' train_end = f'202{