在量化投资领域，很多投资者都有过这样的经历：看到券商研报中精美的策略回测曲线和诱人的收益数据，却苦于没有完整的代码实现，只能停留在理论层面。本文将带你完整复现一个基于动量因子和质量因子的ETF双因子轮动策略，从数据获取、因子计算到策略回测，提供全套可运行的Python源码。

本文适合有一定Python基础的量化投资爱好者，学完后你将掌握一个完整的量化策略开发流程，包括数据接口调用、因子计算、权重分配和回测分析。无论是想学习量化策略开发，还是需要一套可复用的策略框架，本文都能提供实用价值。

1. 策略原理与核心概念

1.1 什么是ETF双因子轮动策略

ETF双因子轮动策略是一种基于多因子模型的资产配置方法，通过两个核心维度筛选ETF并动态调整仓位。该策略的核心思想是：在不同市场环境下，通过量化指标选择表现优异的ETF品种，实现资产的优化配置。

策略的两个核心因子：

• 动量因子：衡量资产的价格趋势强度，选择近期表现较好的ETF
• 质量因子：评估资产的稳定性和风险特征，筛选波动率较低、趋势稳定的ETF

1.2 策略运作机制

双因子轮动策略的运作包含以下几个关键环节：

1. 标的池选择：确定参与轮动的ETF品种范围
2. 因子计算：定期计算每个ETF的动量得分和质量得分
3. 综合评分：将两个因子得分加权合并为综合评分
4. 权重分配：根据评分结果分配投资权重
5. 调仓执行：按固定周期调整持仓比例
6. 风控管理：设置止损、仓位控制等风险管理措施

1.3 策略优势与适用场景

这种策略的优势在于：

• 双因子互补：动量因子捕捉趋势，质量因子控制风险
• 动态调整：根据市场变化自动优化配置
• 分散投资：避免单一资产或因子的过度暴露
• 系统化执行：减少主观情绪干扰

适合在趋势明显的市场环境中运行，特别是在板块轮动较为频繁的时期表现较好。

2. 环境准备与依赖配置

2.1 Python环境要求

本策略基于Python 3.8+开发，需要安装以下核心库：

# 基础数据处理库 pip install pandas>=1.4.0 pip install numpy>=1.21.0 # 量化分析库 pip install tushare>=1.2.0 pip install baostock>=0.8.0 pip install empyrical>=0.5.0 # 回测框架 pip install backtrader>=1.9.0 # 可视化库 pip install matplotlib>=3.5.0 pip install seaborn>=0.11.0

2.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(market='E') 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_code=ts_code, start_date=start_date, end_date=end_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 None

2.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_score

3.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(pct=True) quality_rank = pd.Series(quality_scores).rank(pct=True) 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_scores

4. 完整策略实现

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(), key=lambda x: x[1], reverse=True) 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(data=data, size=size)

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_cash=1000000, commission=0.001): self.cerebro = bt.Cerebro() self.cerebro.broker.setcash(initial_cash) self.cerebro.broker.setcommission(commission=commission) 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( dataname=df, datetime=None, # 使用索引作为时间 open='open', high='high', low='low', close='close', volume='vol', openinterest=-1 ) self.cerebro.adddata(data, name=etf_code) def run_backtest(self, strategy, strategy_params=None): """运行回测""" self.cerebro.addstrategy(strategy, **strategy_params or {}) # 添加分析器 self.cerebro.addanalyzer(bt.analyzers.SharpeRatio, _name='sharpe') self.cerebro.addanalyzer(bt.analyzers.DrawDown, _name='drawdown') self.cerebro.addanalyzer(bt.analyzers.Returns, _name='returns') self.cerebro.addanalyzer(bt.analyzers.TradeAnalyzer, _name='trades') # 运行回测 results = self.cerebro.run() return results[0] # 返回策略实例 def plot_results(self): """绘制回测结果""" self.cerebro.plot(style='candlestick', volume=False)

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_cash=1000000) 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_return

6. 常见问题与解决方案

6.1 数据获取问题

问题1：Tushare接口限制

• 现象：获取数据时出现权限错误或频率限制
• 解决方案：申请正式版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(start=start_date, end=end_date, freq='D') 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_period=2, test_period=1): """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 results

7. 策略优化与进阶功能

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 = False

7.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双因子轮动策略，更重要的是掌握了一套完整的量化策略开发方法论。在实际应用中，建议先进行充分的回测验证，再考虑小规模实盘测试，同时持续优化和监控策略表现。