这类量化策略复现文章最怕的就是只给代码不给思路读者要么看不懂要么跑不通。我帮你把整个ETF双因子轮动的实现流程拆解成可落地的步骤从数据获取到策略回测每个环节都配上可运行的代码和参数说明。1. 先搞清楚双因子轮动到底在解决什么问题很多人一看到“双因子”就觉得复杂其实核心逻辑很简单用两个指标比如动量和波动率来筛选ETF定期轮换持有表现更好的品种。这种策略适合震荡市和趋势市但在单边下跌行情中也需要配合止损。我一般会先确认策略的基本假设动量因子近期涨幅大的ETF更可能继续上涨波动率因子波动率适中的品种更适合持有轮动周期通常按月或按周调仓实际操作中最容易出问题的是因子计算方法和调仓逻辑。很多人直接照搬论文公式但忽略了自己交易环境下的实际约束。2. 数据获取与清洗的实操细节没有可靠的数据再好的策略也是空谈。我建议先从免费的公开数据源开始验证策略逻辑。2.1 ETF基础信息获取东方财富、新浪财经等平台都能获取ETF日线数据但需要注意几个关键点import pandas as pd import requests import json def get_etf_daily(code, start_date, end_date): 获取ETF日线数据 code: ETF代码如510300 start_date: 开始日期 20200101 end_date: 结束日期 20231231 # 这里用模拟数据演示实际需要调用相应API # 注意处理网络请求超时和数据格式异常 try: # 模拟数据返回 dates pd.date_range(start_date, end_date, freqD) data { date: dates, close: np.random.randn(len(dates)).cumsum() 100, volume: np.random.randint(1000000, 5000000, len(dates)) } df pd.DataFrame(data) df[code] code return df except Exception as e: print(f获取{code}数据失败: {e}) return None重要提醒实际使用时一定要检查数据完整性特别是是否有停牌日的缺失数据复权价格是否正确成交量是否为0的异常情况2.2 数据清洗标准流程我习惯用这套清洗流程能避免大部分数据质量问题def clean_etf_data(df): 数据清洗函数 # 1. 按日期排序 df df.sort_values(date) # 2. 处理缺失值 df df.dropna(subset[close]) # 3. 去除成交量为0的日期通常为停牌 df df[df[volume] 0] # 4. 检查价格连续性 price_changes df[close].pct_change() # 单日涨跌幅超过20%需要重点检查 abnormal_days df[abs(price_changes) 0.2] if len(abnormal_days) 0: print(f发现{len(abnormal_days)}个价格异常日请核对数据) return df清洗完成后建议保存到本地数据库或文件避免每次重新下载。3. 双因子计算的具体实现因子计算是策略的核心不同的计算方式会导致结果差异很大。3.1 动量因子计算动量因子不是简单的收益率计算需要考虑时间周期和计算方法def calculate_momentum(df, window20): 计算动量因子 window: 回顾周期默认20个交易日 df df.copy() # 方法1简单收益率动量 df[momentum_ret] df[close].pct_change(window) # 方法2累计收益率动量更平滑 df[cumulative_ret] (df[close] / df[close].shift(window)) - 1 # 方法3风险调整后动量夏普比率思路 returns df[close].pct_change() df[momentum_sharpe] returns.rolling(window).mean() / returns.rolling(window).std() return df # 测试计算 def test_factor_calculation(): 因子计算测试函数 # 生成测试数据 dates pd.date_range(20230101, 20231231, freqD) test_data pd.DataFrame({ date: dates, close: np.cumsum(np.random.randn(len(dates)) * 0.01) 100 }) result calculate_momentum(test_data) print(动量因子计算完成前5行结果) print(result.head())3.2 波动率因子计算波动率因子需要考虑不同计算方法对结果的影响def calculate_volatility(df, window20): 计算波动率因子 df df.copy() returns df[close].pct_change() # 方法1简单历史波动率 df[volatility_std] returns.rolling(window).std() # 方法2年化波动率 df[volatility_annual] returns.rolling(window).std() * np.sqrt(252) # 方法3下行波动率只考虑负收益 downside_returns returns[returns 0] if len(downside_returns) 0: df[volatility_downside] returns.rolling(window).apply( lambda x: x[x 0].std() if len(x[x 0]) 0 else 0 ) return df3.3 因子标准化与组合单个因子需要标准化后才能组合使用def normalize_factors(factor_df): 因子标准化处理 normalized_df factor_df.copy() # 对每个因子进行横截面标准化 factor_columns [col for col in factor_df.columns if momentum in col or volatility in col] for factor in factor_columns: # 去除极端值3个标准差以外的值 mean_val factor_df[factor].mean() std_val factor_df[factor].std() cap_value mean_val 3 * std_val floor_value mean_val - 3 * std_val factor_df[factor] factor_df[factor].clip(floor_value, cap_value) # Z-score标准化 normalized_df[f{factor}_zscore] ( (factor_df[factor] - factor_df[factor].mean()) / factor_df[factor].std() ) return normalized_df4. 轮动策略的完整实现策略逻辑的实现需要注意调仓时机和交易成本的影响。4.1 基础轮动逻辑class ETFDoubleFactorStrategy: def __init__(self, momentum_weight0.6, volatility_weight0.4): self.momentum_weight momentum_weight self.volatility_weight volatility_weight self.positions {} # 当前持仓 self.trade_log [] # 交易记录 def calculate_composite_score(self, factor_data): 计算综合得分 # 确保因子方向正确动量越高越好波动率适中为好 factor_data[composite_score] ( factor_data[momentum_ret_zscore] * self.momentum_weight - abs(factor_data[volatility_std_zscore]) * self.volatility_weight ) return factor_data def generate_signal(self, etf_data, top_n3): 生成交易信号 # 计算每个ETF的综合得分 scored_etfs [] for code, data in etf_data.items(): if len(data) 0: latest_data data.iloc[-1].copy() latest_data[code] code scored_etfs.append(latest_data) scored_df pd.DataFrame(scored_etfs) scored_df self.calculate_composite_score(scored_df) # 选择得分最高的top_n个ETF top_etfs scored_df.nlargest(top_n, composite_score) return top_etfs[code].tolist()4.2 调仓执行与交易成本def execute_rebalancing(current_holdings, target_holdings, trade_date): 执行调仓操作 trades [] # 计算需要调整的仓位 current_codes set(current_holdings.keys()) target_codes set(target_holdings.keys()) # 需要卖出的代码 sell_codes current_codes - target_codes for code in sell_codes: trades.append({ date: trade_date, code: code, action: sell, weight: current_holdings[code] }) # 需要买入的代码 buy_codes target_codes - current_codes for code in buy_codes: trades.append({ date: trade_date, code: code, action: buy, weight: target_holdings[code] }) # 需要调整权重的代码 adjust_codes current_codes target_codes for code in adjust_codes: if abs(current_holdings[code] - target_holdings[code]) 0.01: # 权重变化超过1%才调整 trades.append({ date: trade_date, code: code, action: adjust, old_weight: current_holdings[code], new_weight: target_holdings[code] }) return trades # 考虑交易成本的收益计算 def calculate_returns_with_costs(portfolio_values, turnover_rates, cost_rate0.001): 考虑交易成本的收益计算 cost_rate: 单边交易成本率默认千分之一 raw_returns portfolio_values.pct_change().fillna(0) # 计算成本调整后的收益 cost_adjusted_returns raw_returns - turnover_rates * cost_rate return cost_adjusted_returns5. 回测系统的关键实现回测不是简单的收益计算需要充分考虑现实约束。5.1 完整的回测框架class BacktestEngine: def __init__(self, initial_capital1000000): self.initial_capital initial_capital self.results {} def run_backtest(self, strategy, etf_data, start_date, end_date, rebalance_freqM): 运行回测 current_date start_date portfolio_value self.initial_capital portfolio_history [] while current_date end_date: # 获取当前可用的ETF数据 available_data self.get_available_data(etf_data, current_date) if len(available_data) 0: # 生成交易信号 signals strategy.generate_signal(available_data) # 执行调仓 trades self.execute_trades(signals, current_date) # 更新组合价值 portfolio_value self.update_portfolio_value( portfolio_value, trades, available_data, current_date ) # 记录组合状态 portfolio_history.append({ date: current_date, value: portfolio_value, holdings: strategy.positions.copy() }) # 移动到下一个调仓日 current_date self.get_next_rebalance_date(current_date, rebalance_freq) return pd.DataFrame(portfolio_history) def calculate_metrics(self, portfolio_values): 计算回测指标 returns portfolio_values.pct_change().fillna(0) metrics { total_return: portfolio_values.iloc[-1] / portfolio_values.iloc[0] - 1, annual_return: returns.mean() * 252, annual_volatility: returns.std() * np.sqrt(252), sharpe_ratio: returns.mean() / returns.std() * np.sqrt(252), max_drawdown: self.calculate_max_drawdown(portfolio_values), win_rate: (returns 0).mean() } return metrics def calculate_max_drawdown(self, values): 计算最大回撤 peak values.expanding().max() drawdown (values - peak) / peak return drawdown.min()5.2 回测结果分析def analyze_backtest_results(portfolio_history, benchmark_dataNone): 分析回测结果 results {} # 基础收益分析 portfolio_values pd.Series( [x[value] for x in portfolio_history], index[x[date] for x in portfolio_history] ) # 计算各类指标 returns portfolio_values.pct_change().fillna(0) print( 回测结果分析 ) print(f初始资金: {portfolio_values.iloc[0]:,.2f}) print(f最终资金: {portfolio_values.iloc[-1]:,.2f}) print(f总收益率: {portfolio_values.iloc[-1]/portfolio_values.iloc[0]-1:.2%}) print(f年化收益率: {returns.mean()*252:.2%}) print(f年化波动率: {returns.std()*np.sqrt(252):.2%}) print(f夏普比率: {returns.mean()/returns.std()*np.sqrt(252):.2f}) print(f最大回撤: {calculate_max_drawdown(portfolio_values):.2%}) # 如果有基准数据计算超额收益 if benchmark_data is not None: benchmark_returns benchmark_data.pct_change().fillna(0) excess_returns returns - benchmark_returns print(f超额年化收益: {excess_returns.mean()*252:.2%}) print(f信息比率: {excess_returns.mean()/excess_returns.std()*np.sqrt(252):.2f}) return results6. 实盘注意事项与优化建议回测通过不代表实盘就能赚钱这几个点需要特别注意6.1 实盘过渡的关键检查点我一般会按这个顺序验证策略的实盘可行性数据延迟验证实盘数据与回测数据的时间匹配交易成本验证考虑佣金、滑点、冲击成本流动性验证确保调仓时能按理想价格成交极端行情验证股灾、熔断等特殊时期的处理# 实盘前的最终验证函数 def validate_strategy_realworld(strategy, historical_data, stress_periods): 实盘前验证 validation_results {} # 1. 检查数据质量 data_quality check_data_quality(historical_data) validation_results[data_quality] data_quality # 2. 压力测试 stress_test_results stress_test(strategy, stress_periods) validation_results[stress_test] stress_test_results # 3. 交易可行性检查 trade_feasibility check_trade_feasibility(strategy) validation_results[trade_feasibility] trade_feasibility return validation_results6.2 策略优化方向根据我的经验双因子轮动策略可以从这些方向优化动态权重调整根据市场状态调整动量因子和波动率因子的权重因子择时在动量失效时暂时关闭动量因子风险控制加入最大回撤止损机制多时间框架结合短期和长期动量信号# 动态权重调整示例 def dynamic_weight_adjustment(market_regime): 根据市场状态调整因子权重 if market_regime bull: # 牛市中加大动量权重 return {momentum: 0.7, volatility: 0.3} elif market_regime bear: # 熊市中加大低波动权重 return {momentum: 0.3, volatility: 0.7} else: # 震荡市中均衡配置 return {momentum: 0.5, volatility: 0.5}7. 常见问题排查清单实际运行中遇到问题按这个顺序排查7.1 数据问题排查检查数据完整性是否有缺失日期或异常价格验证复权价格除权除息是否正确处理确认交易时间避免在非交易日产生信号7.2 策略逻辑排查因子计算验证手动验证几个时间点的因子值信号生成检查确认选出的ETF符合预期调仓逻辑验证持仓变化是否合理7.3 回测结果排查收益曲线分析检查是否有异常收益点交易记录验证每笔交易是否可追溯参数敏感性小幅调整参数看结果稳定性这套代码框架提供了ETF双因子轮动的完整实现路径但真正落地时还需要根据具体的交易品种和市场环境进行调整。建议先用小资金实盘验证确认策略有效性后再逐步加大投入。