如果你经常看券商研报可能会发现一个现象很多量化策略听起来逻辑完美回测曲线漂亮但真要自己动手实现时却总是卡在数据获取、因子计算、或者风控细节上。特别是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 pandas1.4.0 numpy1.21.0 requests2.28.0 akshare1.8.0 backtrader1.9.78 matplotlib3.5.0 seaborn0.11.0安装命令pip install -r requirements.txt2.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(symboletf_code, perioddaily) 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万元品种覆盖涵盖主要宽基指数、行业指数、跨境ETFimport 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(days365) # 注意这里需要根据实际情况调整成立时间字段 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(symboletf_code, perioddaily) 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(methodffill) # 计算复权价格使用涨跌幅复权 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_data4. 双因子计算与权重分配因子计算是策略的核心需要确保计算的准确性和稳定性。4.1 动量因子计算动量因子有多种计算方式我们采用最经典的过去N日收益率def calculate_momentum_factor(data, lookback_period20): 计算动量因子 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(pctTrue) return momentum_rank def calculate_quality_factor(data, lookback_period60): 计算质量因子基于波动率 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(pctTrue) return quality_rank4.2 因子合成与权重分配将两个因子合成为综合得分并据此分配权重def calculate_composite_score(momentum_rank, quality_rank, momentum_weight0.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_n5, methodequal_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 weights5. 轮动策略回测实现使用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(pctTrue) def calculate_quality(self): 计算质量因子基于波动率 quality_scores {} for d in self.datas: if len(d) self.params.lookback_quality: # 简化计算使用历史波动率倒数作为质量指标 returns np.log(d.close.get(sizeself.params.lookback_quality) / d.close.get(sizeself.params.lookback_quality, ago-1)) volatility np.std(returns) quality_scores[d._name] 1 / (1 volatility) return pd.Series(quality_scores).rank(pctTrue) 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) break5.2 回测框架与参数优化def run_backtest(etf_codes, start_date, end_date, initial_cash100000): 运行回测 cerebro bt.Cerebro() # 添加数据 for code in etf_codes: data get_backtrader_data(code, start_date, end_date) cerebro.adddata(data, namecode) # 添加策略 cerebro.addstrategy(DualFactorRotationStrategy) # 设置资金 cerebro.broker.setcash(initial_cash) # 添加分析器 cerebro.addanalyzer(bt.analyzers.SharpeRatio, _namesharpe) cerebro.addanalyzer(bt.analyzers.DrawDown, _namedrawdown) cerebro.addanalyzer(bt.analyzers.Returns, _namereturns) # 运行回测 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(symboletf_code, perioddaily) 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( datanamehist_data, datetime日期, open开盘, high最高, low最低, close收盘, volume成交量, openinterest-1 ) return data6. 策略效果评估与优化回测完成后需要对策略效果进行科学评估。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_returns6.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: fm{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, ascendingFalse)7. 实盘注意事项与风险控制将策略应用于实盘时需要特别注意以下几点7.1 实盘执行关键点交易成本考虑回测中通常忽略的交易成本在实盘中至关重要ETF买卖佣金通常为万分之三印花税卖出时千分之一滑点实际成交价与预期价的差异# 在回测中添加交易成本 cerebro.broker.setcommission(commission0.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 False8. 常见问题与解决方案在实际实现过程中经常会遇到以下问题8.1 数据相关问题问题1数据缺失或异常现象某些ETF历史数据不全或存在异常值解决方案建立数据质量检查机制自动剔除数据质量差的ETFdef validate_etf_data(data, min_data_points100, max_missing_rate0.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_periods5): 向前验证避免过拟合 performance_by_period [] for i in range(total_periods): # 划分训练集和测试集 train_start 2018-01-01 train_end f202{
ETF双因子轮动策略：动量与质量因子的Python实战指南
发布时间：2026/7/15 9:42:20
如果你经常看券商研报可能会发现一个现象很多量化策略听起来逻辑完美回测曲线漂亮但真要自己动手实现时却总是卡在数据获取、因子计算、或者风控细节上。特别是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 pandas1.4.0 numpy1.21.0 requests2.28.0 akshare1.8.0 backtrader1.9.78 matplotlib3.5.0 seaborn0.11.0安装命令pip install -r requirements.txt2.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(symboletf_code, perioddaily) 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万元品种覆盖涵盖主要宽基指数、行业指数、跨境ETFimport 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(days365) # 注意这里需要根据实际情况调整成立时间字段 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(symboletf_code, perioddaily) 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(methodffill) # 计算复权价格使用涨跌幅复权 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_data4. 双因子计算与权重分配因子计算是策略的核心需要确保计算的准确性和稳定性。4.1 动量因子计算动量因子有多种计算方式我们采用最经典的过去N日收益率def calculate_momentum_factor(data, lookback_period20): 计算动量因子 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(pctTrue) return momentum_rank def calculate_quality_factor(data, lookback_period60): 计算质量因子基于波动率 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(pctTrue) return quality_rank4.2 因子合成与权重分配将两个因子合成为综合得分并据此分配权重def calculate_composite_score(momentum_rank, quality_rank, momentum_weight0.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_n5, methodequal_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 weights5. 轮动策略回测实现使用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(pctTrue) def calculate_quality(self): 计算质量因子基于波动率 quality_scores {} for d in self.datas: if len(d) self.params.lookback_quality: # 简化计算使用历史波动率倒数作为质量指标 returns np.log(d.close.get(sizeself.params.lookback_quality) / d.close.get(sizeself.params.lookback_quality, ago-1)) volatility np.std(returns) quality_scores[d._name] 1 / (1 volatility) return pd.Series(quality_scores).rank(pctTrue) 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) break5.2 回测框架与参数优化def run_backtest(etf_codes, start_date, end_date, initial_cash100000): 运行回测 cerebro bt.Cerebro() # 添加数据 for code in etf_codes: data get_backtrader_data(code, start_date, end_date) cerebro.adddata(data, namecode) # 添加策略 cerebro.addstrategy(DualFactorRotationStrategy) # 设置资金 cerebro.broker.setcash(initial_cash) # 添加分析器 cerebro.addanalyzer(bt.analyzers.SharpeRatio, _namesharpe) cerebro.addanalyzer(bt.analyzers.DrawDown, _namedrawdown) cerebro.addanalyzer(bt.analyzers.Returns, _namereturns) # 运行回测 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(symboletf_code, perioddaily) 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( datanamehist_data, datetime日期, open开盘, high最高, low最低, close收盘, volume成交量, openinterest-1 ) return data6. 策略效果评估与优化回测完成后需要对策略效果进行科学评估。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_returns6.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: fm{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, ascendingFalse)7. 实盘注意事项与风险控制将策略应用于实盘时需要特别注意以下几点7.1 实盘执行关键点交易成本考虑回测中通常忽略的交易成本在实盘中至关重要ETF买卖佣金通常为万分之三印花税卖出时千分之一滑点实际成交价与预期价的差异# 在回测中添加交易成本 cerebro.broker.setcommission(commission0.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 False8. 常见问题与解决方案在实际实现过程中经常会遇到以下问题8.1 数据相关问题问题1数据缺失或异常现象某些ETF历史数据不全或存在异常值解决方案建立数据质量检查机制自动剔除数据质量差的ETFdef validate_etf_data(data, min_data_points100, max_missing_rate0.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_periods5): 向前验证避免过拟合 performance_by_period [] for i in range(total_periods): # 划分训练集和测试集 train_start 2018-01-01 train_end f202{
