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

在量化投资领域很多投资者都有过这样的经历看到券商研报中精美的策略回测曲线和诱人的收益数据却苦于没有完整的代码实现只能停留在理论层面。本文将带你完整复现一个基于动量因子和质量因子的ETF双因子轮动策略从数据获取、因子计算到策略回测提供全套可运行的Python源码。本文适合有一定Python基础的量化投资爱好者学完后你将掌握一个完整的量化策略开发流程包括数据接口调用、因子计算、权重分配和回测分析。无论是想学习量化策略开发还是需要一套可复用的策略框架本文都能提供实用价值。1. 策略原理与核心概念1.1 什么是ETF双因子轮动策略ETF双因子轮动策略是一种基于多因子模型的资产配置方法通过两个核心维度筛选ETF并动态调整仓位。该策略的核心思想是在不同市场环境下通过量化指标选择表现优异的ETF品种实现资产的优化配置。策略的两个核心因子动量因子衡量资产的价格趋势强度选择近期表现较好的ETF质量因子评估资产的稳定性和风险特征筛选波动率较低、趋势稳定的ETF1.2 策略运作机制双因子轮动策略的运作包含以下几个关键环节标的池选择确定参与轮动的ETF品种范围因子计算定期计算每个ETF的动量得分和质量得分综合评分将两个因子得分加权合并为综合评分权重分配根据评分结果分配投资权重调仓执行按固定周期调整持仓比例风控管理设置止损、仓位控制等风险管理措施1.3 策略优势与适用场景这种策略的优势在于双因子互补动量因子捕捉趋势质量因子控制风险动态调整根据市场变化自动优化配置分散投资避免单一资产或因子的过度暴露系统化执行减少主观情绪干扰适合在趋势明显的市场环境中运行特别是在板块轮动较为频繁的时期表现较好。2. 环境准备与依赖配置2.1 Python环境要求本策略基于Python 3.8开发需要安装以下核心库# 基础数据处理库 pip install pandas1.4.0 pip install numpy1.21.0 # 量化分析库 pip install tushare1.2.0 pip install baostock0.8.0 pip install empyrical0.5.0 # 回测框架 pip install backtrader1.9.0 # 可视化库 pip install matplotlib3.5.0 pip install seaborn0.11.02.2 数据源配置策略需要配置数据源获取ETF行情数据这里以Tushare为例# data_source.py import tushare as ts import pandas as pd from datetime import datetime, timedelta import time class DataFetcher: def __init__(self, token): 初始化数据接口 :param token: Tushare token self.pro ts.pro_api(token) def get_etf_list(self): 获取ETF基础信息 # 获取基金基础信息 df self.pro.fund_basic(marketE) etf_list df[df[fund_type] ETF].copy() return etf_list[[ts_code, name, list_date, delist_date]] def get_daily_data(self, ts_code, start_date, end_date): 获取ETF日线数据 :param ts_code: ETF代码 :param start_date: 开始日期 :param end_date: 结束日期 try: df self.pro.fund_daily(ts_codets_code, start_datestart_date, end_dateend_date) df[trade_date] pd.to_datetime(df[trade_date]) df df.sort_values(trade_date) return df except Exception as e: print(f获取{ts_code}数据失败: {e}) return None2.3 项目结构设计建议的项目文件结构etf_dual_factor/ ├── data/ # 数据目录 │ ├── raw/ # 原始数据 │ └── processed/ # 处理后的数据 ├── src/ │ ├── data_source.py # 数据获取模块 │ ├── factor_calculator.py # 因子计算模块 │ ├── strategy.py # 策略逻辑模块 │ ├── backtest.py # 回测引擎模块 │ └── utils.py # 工具函数 ├── config/ # 配置文件 │ └── strategy_config.yaml └── results/ # 结果输出 ├── figures/ # 图表 └── reports/ # 报告3. 核心因子计算实现3.1 动量因子计算动量因子衡量资产价格的趋势强度常用的计算方法包括收益率动量、价格动量和相对强弱指标等。# factor_calculator.py import pandas as pd import numpy as np from typing import Dict, List class FactorCalculator: def __init__(self, price_data: Dict[str, pd.DataFrame]): 初始化因子计算器 :param price_data: 各ETF的价格数据字典 self.price_data price_data def calculate_momentum_factor(self, etf_code: str, lookback_periods: List[int] [20, 60, 120]): 计算动量因子得分 :param etf_code: ETF代码 :param lookback_periods: 回看周期列表 :return: 动量因子得分 if etf_code not in self.price_data: return None df self.price_data[etf_code].copy() df df.sort_values(trade_date) momentum_scores [] for period in lookback_periods: if len(df) period: continue # 计算周期收益率 current_close df[close].iloc[-1] past_close df[close].iloc[-period] period_return (current_close - past_close) / past_close # 计算收益率波动率风险调整 returns df[close].pct_change().dropna() if len(returns) period: vol returns.tail(period).std() # 风险调整后的动量得分 momentum_score period_return / vol if vol ! 0 else 0 momentum_scores.append(momentum_score) # 综合多个周期的动量得分 if momentum_scores: return np.mean(momentum_scores) else: return 0 def calculate_quality_factor(self, etf_code: str, volatility_window: int 60): 计算质量因子得分低波动性 趋势稳定性 :param etf_code: ETF代码 :param volatility_window: 波动率计算窗口 :return: 质量因子得分 if etf_code not in self.price_data: return None df self.price_data[etf_code].copy() df df.sort_values(trade_date) if len(df) volatility_window: return 0 returns df[close].pct_change().dropna() # 计算波动率逆向指标波动率越低得分越高 volatility returns.tail(volatility_window).std() # 计算最大回撤稳定性指标 cumulative_returns (1 returns).cumprod() peak cumulative_returns.expanding().max() drawdown (cumulative_returns - peak) / peak max_drawdown drawdown.min() # 计算夏普比率风险调整收益 sharpe_ratio returns.mean() / returns.std() if returns.std() ! 0 else 0 # 综合质量得分波动率和回撤越小越好夏普比率越大越好 quality_score (1 / (1 abs(volatility))) * 0.4 \ (1 / (1 abs(max_drawdown))) * 0.3 \ sharpe_ratio * 0.3 return quality_score3.2 因子标准化与权重配置不同因子的量纲和分布不同需要进行标准化处理def normalize_factors(self, factor_scores: Dict[str, Dict[str, float]]): 标准化因子得分 :param factor_scores: 各ETF的因子得分字典 :return: 标准化后的因子得分 etf_codes list(factor_scores.keys()) momentum_scores [factor_scores[code][momentum] for code in etf_codes] quality_scores [factor_scores[code][quality] for code in etf_codes] # 使用排名标准化 momentum_rank pd.Series(momentum_scores).rank(pctTrue) quality_rank pd.Series(quality_scores).rank(pctTrue) normalized_scores {} for i, code in enumerate(etf_codes): normalized_scores[code] { momentum: momentum_rank.iloc[i], quality: quality_rank.iloc[i] } return normalized_scores def calculate_composite_score(self, normalized_scores: Dict[str, Dict[str, float]], weights: Dict[str, float] None): 计算综合得分 :param normalized_scores: 标准化后的因子得分 :param weights: 因子权重默认动量40%质量60% :return: 综合得分字典 if weights is None: weights {momentum: 0.4, quality: 0.6} composite_scores {} for code, scores in normalized_scores.items(): composite_score (scores[momentum] * weights[momentum] scores[quality] * weights[quality]) composite_scores[code] composite_score return composite_scores4. 完整策略实现4.1 策略逻辑框架基于Backtrader回测框架实现完整的双因子轮动策略# strategy.py import backtrader as bt import pandas as pd from datetime import datetime class DualFactorRotationStrategy(bt.Strategy): params ( (rebalance_days, 20), # 调仓周期 (top_n, 5), # 选择前N只ETF (momentum_weight, 0.4), # 动量因子权重 (quality_weight, 0.6), # 质量因子权重 ) def __init__(self): self.etf_data {} # 存储各ETF数据 self.rebalance_day 0 self.initial_cash self.broker.getvalue() def next(self): # 检查是否到达调仓日 if len(self.data) % self.params.rebalance_days ! 0: return # 计算因子得分 factor_scores self.calculate_factor_scores() # 选择投资标的 selected_etfs self.select_etfs(factor_scores) # 执行调仓 self.rebalance_portfolio(selected_etfs) def calculate_factor_scores(self): 计算各ETF的因子得分 factor_scores {} for data in self.datas: etf_code data._name # 获取价格数据最近120个交易日 prices [data.close[i] for i in range(-120, 0)] if len(prices) 60: # 数据不足 continue # 计算动量因子 momentum_score self.calculate_momentum(prices) # 计算质量因子 quality_score self.calculate_quality(prices) factor_scores[etf_code] { momentum: momentum_score, quality: quality_score } return factor_scores def calculate_momentum(self, prices): 计算动量得分 if len(prices) 120: return 0 # 短期动量20日 short_return (prices[-1] - prices[-20]) / prices[-20] # 中期动量60日 mid_return (prices[-1] - prices[-60]) / prices[-60] # 长期动量120日 long_return (prices[-1] - prices[-120]) / prices[-120] # 综合动量得分 momentum_score (short_return * 0.5 mid_return * 0.3 long_return * 0.2) return momentum_score def calculate_quality(self, prices): 计算质量得分 returns [(prices[i] - prices[i-1]) / prices[i-1] for i in range(1, len(prices))] if len(returns) 60: return 0 # 计算波动率逆向 volatility np.std(returns[-60:]) # 计算最大回撤 cumulative np.cumprod(1 np.array(returns)) peak np.maximum.accumulate(cumulative) drawdown (cumulative - peak) / peak max_dd np.min(drawdown) # 质量得分波动率和回撤越小越好 quality_score 1 / (1 abs(volatility)) * 0.6 1 / (1 abs(max_dd)) * 0.4 return quality_score def select_etfs(self, factor_scores): 根据综合得分选择ETF if not factor_scores: return [] # 计算综合得分 composite_scores {} for code, scores in factor_scores.items(): composite_score (scores[momentum] * self.params.momentum_weight scores[quality] * self.params.quality_weight) composite_scores[code] composite_score # 按得分排序选择前N只 sorted_etfs sorted(composite_scores.items(), keylambda x: x[1], reverseTrue) selected [etf[0] for etf in sorted_etfs[:self.params.top_n]] return selected def rebalance_portfolio(self, selected_etfs): 执行投资组合再平衡 # 清空当前持仓 for data in self.datas: self.close(data) # 等权重分配 weight 1.0 / len(selected_etfs) if selected_etfs else 0 # 买入选中的ETF for etf_code in selected_etfs: data self.getdatabyname(etf_code) value self.broker.getvalue() * weight size value // data.close[0] if size 0: self.buy(datadata, sizesize)4.2 回测引擎配置完整的回测流程实现# backtest.py import backtrader as bt import pandas as pd from datetime import datetime import matplotlib.pyplot as plt class BacktestEngine: def __init__(self, initial_cash1000000, commission0.001): self.cerebro bt.Cerebro() self.cerebro.broker.setcash(initial_cash) self.cerebro.broker.setcommission(commissioncommission) def add_data(self, etf_data_dict): 添加ETF数据到回测引擎 for etf_code, df in etf_data_dict.items(): # 确保数据格式正确 df[trade_date] pd.to_datetime(df[trade_date]) df df.set_index(trade_date) df df.sort_index() # 创建Backtrader数据格式 data bt.feeds.PandasData( datanamedf, datetimeNone, # 使用索引作为时间 openopen, highhigh, lowlow, closeclose, volumevol, openinterest-1 ) self.cerebro.adddata(data, nameetf_code) def run_backtest(self, strategy, strategy_paramsNone): 运行回测 self.cerebro.addstrategy(strategy, **strategy_params or {}) # 添加分析器 self.cerebro.addanalyzer(bt.analyzers.SharpeRatio, _namesharpe) self.cerebro.addanalyzer(bt.analyzers.DrawDown, _namedrawdown) self.cerebro.addanalyzer(bt.analyzers.Returns, _namereturns) self.cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _nametrades) # 运行回测 results self.cerebro.run() return results[0] # 返回策略实例 def plot_results(self): 绘制回测结果 self.cerebro.plot(stylecandlestick, volumeFalse)5. 完整实战案例5.1 数据准备与预处理# main.py import pandas as pd from datetime import datetime, timedelta from data_source import DataFetcher from backtest import BacktestEngine from strategy import DualFactorRotationStrategy def main(): # 初始化数据获取器 token 你的Tushare Token # 需要申请获取 fetcher DataFetcher(token) # 获取ETF列表 etf_list fetcher.get_etf_list() print(f获取到{len(etf_list)}只ETF) # 选择部分流动性较好的ETF作为标的池 target_etfs [510300.SH, 510500.SH, 159915.SZ, 512100.SH, 512000.SH] # 设置回测时间范围 start_date 20200101 end_date 20231231 # 获取历史数据 etf_data {} for etf_code in target_etfs: print(f正在获取{etf_code}数据...) data fetcher.get_daily_data(etf_code, start_date, end_date) if data is not None and len(data) 100: # 确保有足够数据 etf_data[etf_code] data time.sleep(0.5) # 避免请求过于频繁 print(f成功获取{len(etf_data)}只ETF的历史数据) # 初始化回测引擎 engine BacktestEngine(initial_cash1000000) engine.add_data(etf_data) # 设置策略参数 strategy_params { rebalance_days: 20, top_n: 3, momentum_weight: 0.4, quality_weight: 0.6 } # 运行回测 print(开始回测...) result engine.run_backtest(DualFactorRotationStrategy, strategy_params) # 输出回测结果 print(\n 回测结果 ) print(f初始资金: 1,000,000元) print(f最终资产: {result.broker.getvalue():.2f}元) print(f总收益率: {(result.broker.getvalue() / 1000000 - 1) * 100:.2f}%) # 绘制结果图表 engine.plot_results() if __name__ __main__: main()5.2 策略参数优化通过网格搜索寻找最优参数组合# optimization.py import itertools from backtest import BacktestEngine from strategy import DualFactorRotationStrategy def parameter_optimization(etf_data): 参数优化函数 # 定义参数范围 rebalance_days_range [10, 20, 30] top_n_range [2, 3, 4, 5] momentum_weight_range [0.3, 0.4, 0.5, 0.6] best_params None best_return -float(inf) # 网格搜索 for params in itertools.product(rebalance_days_range, top_n_range, momentum_weight_range): rebalance_days, top_n, momentum_weight params quality_weight 1 - momentum_weight if quality_weight 0: continue strategy_params { rebalance_days: rebalance_days, top_n: top_n, momentum_weight: momentum_weight, quality_weight: quality_weight } try: engine BacktestEngine() engine.add_data(etf_data) result engine.run_backtest(DualFactorRotationStrategy, strategy_params) final_value result.broker.getvalue() if final_value best_return: best_return final_value best_params strategy_params except Exception as e: print(f参数{params}回测失败: {e}) continue return best_params, best_return6. 常见问题与解决方案6.1 数据获取问题问题1Tushare接口限制现象获取数据时出现权限错误或频率限制解决方案申请正式版Token设置请求间隔使用本地数据缓存# 添加请求间隔控制 import time from functools import wraps def rate_limit(seconds): def decorator(func): last_called [0.0] wraps(func) def wrapper(*args, **kwargs): elapsed time.time() - last_called[0] left_to_wait seconds - elapsed if left_to_wait 0: time.sleep(left_to_wait) ret func(*args, **kwargs) last_called[0] time.time() return ret return wrapper return decorator class DataFetcher: rate_limit(0.5) # 每0.5秒最多请求一次 def get_daily_data(self, ts_code, start_date, end_date): # 原有实现... pass问题2数据缺失或不连续现象某些ETF在特定时间段没有数据解决方案数据完整性检查填充或剔除缺失数据def validate_data_completeness(self, df, start_date, end_date): 验证数据完整性 expected_dates pd.date_range(startstart_date, endend_date, freqD) actual_dates pd.to_datetime(df[trade_date]).sort_values() missing_dates expected_dates.difference(actual_dates) if len(missing_dates) 0: print(f警告缺失{len(missing_dates)}个交易日数据) # 向前填充缺失值或剔除不连续数据 df_complete df.set_index(trade_date).reindex(expected_dates).ffill() return df_complete.reset_index()6.2 策略执行问题问题3流动性不足导致无法成交现象回测中下单但实际无法成交解决方案添加流动性检查使用下一根K线开盘价成交class ImprovedDualFactorStrategy(DualFactorRotationStrategy): def next(self): # 在调仓日使用下一根K线开盘价成交 if len(self.data) % self.params.rebalance_days 0: # 记录调仓指令在下一个交易日执行 self.rebalance_next_open True if getattr(self, rebalance_next_open, False): self.execute_rebalance() self.rebalance_next_open False def execute_rebalance(self): 在开盘时执行调仓 # 实现调仓逻辑... pass问题4过拟合问题现象在历史数据上表现优异但实盘效果差解决方案使用Walk-Forward分析避免过度优化def walk_forward_analysis(etf_data, train_period2, test_period1): Walk-Forward分析 all_dates sorted(etf_data[list(etf_data.keys())[0]][trade_date]) results [] for i in range(0, len(all_dates) - (train_period test_period) * 252, 63): # 约3个月步长 # 划分训练集和测试集 train_end i train_period * 252 test_end train_end test_period * 252 if test_end len(all_dates): break # 在训练集上优化参数 train_data slice_data(etf_data, all_dates[i], all_dates[train_end]) best_params parameter_optimization(train_data) # 在测试集上验证 test_data slice_data(etf_data, all_dates[train_end], all_dates[test_end]) test_result run_backtest_with_params(test_data, best_params) results.append(test_result) return results7. 策略优化与进阶功能7.1 动态权重调整根据市场状态动态调整因子权重def dynamic_weight_adjustment(self, market_status): 根据市场状态动态调整因子权重 if market_status bull: # 牛市中加大动量因子权重 self.params.momentum_weight 0.6 self.params.quality_weight 0.4 elif market_status bear: # 熊市中加大质量因子权重 self.params.momentum_weight 0.2 self.params.quality_weight 0.8 else: # 震荡市均衡配置 self.params.momentum_weight 0.4 self.params.quality_weight 0.6 def assess_market_status(self, market_data): 评估当前市场状态 # 使用移动平均线判断趋势 short_ma market_data.close.rolling(20).mean() long_ma market_data.close.rolling(60).mean() if short_ma.iloc[-1] long_ma.iloc[-1] and market_data.close.iloc[-1] short_ma.iloc[-1]: return bull # 牛市 elif short_ma.iloc[-1] long_ma.iloc[-1] and market_data.close.iloc[-1] short_ma.iloc[-1]: return bear # 熊市 else: return neutral # 震荡市7.2 风险控制增强添加更严格的风险控制机制class RiskManagedDualFactorStrategy(DualFactorRotationStrategy): params ( (max_drawdown_limit, -0.15), # 最大回撤限制 (volatility_limit, 0.3), # 波动率限制 (position_limit, 0.8), # 单品种仓位限制 ) def __init__(self): super().__init__() self.peak_value self.initial_cash self.in_market True def next(self): # 检查风险控制条件 if not self.check_risk_controls(): if self.in_market: self.go_to_cash() # 清仓离场 return if self.in_market: super().next() else: # 等待重新入场信号 if self.check_reentry_signal(): self.in_market True def check_risk_controls(self): 检查风险控制条件 current_value self.broker.getvalue() # 更新峰值 if current_value self.peak_value: self.peak_value current_value # 计算当前回撤 drawdown (current_value - self.peak_value) / self.peak_value # 检查回撤限制 if drawdown self.params.max_drawdown_limit: return False # 检查波动率限制简化实现 recent_returns [] for data in self.datas: if len(data) 20: returns (data.close[0] - data.close[-20]) / data.close[-20] recent_returns.append(returns) if recent_returns and np.std(recent_returns) self.params.volatility_limit: return False return True def go_to_cash(self): 清仓转为现金 for data in self.datas: self.close(data) self.in_market False7.3 绩效分析报告生成详细的策略绩效报告def generate_performance_report(strategy_instance, etf_data): 生成详细的绩效分析报告 import empyrical as ep import matplotlib.pyplot as plt # 提取净值曲线 portfolio_values [1000000] # 初始净值 dates [] # 模拟计算每日净值简化实现 for i in range(1, 100): # 假设100个交易日 # 实际实现中需要从回测结果中提取 pass returns pd.Series(portfolio_values).pct_change().dropna() # 计算各项指标 total_return ep.cum_returns_final(returns) annual_return ep.annual_return(returns) volatility ep.annual_volatility(returns) sharpe_ratio ep.sharpe_ratio(returns) max_drawdown ep.max_drawdown(returns) # 生成报告 report { 总收益率: f{total_return:.2%}, 年化收益率: f{annual_return:.2%}, 年化波动率: f{volatility:.2%}, 夏普比率: f{sharpe_ratio:.2f}, 最大回撤: f{max_drawdown:.2%}, 卡玛比率: f{annual_return/abs(max_drawdown):.2f} if max_drawdown ! 0 else N/A } return report通过本文的完整实现你不仅得到了一个可运行的ETF双因子轮动策略更重要的是掌握了一套完整的量化策略开发方法论。在实际应用中建议先进行充分的回测验证再考虑小规模实盘测试同时持续优化和监控策略表现。