Python实现ETF双因子轮动策略：动量与波动率因子实战指南

在实际量化投资领域，券商研报提供了很多有价值的策略思路，但往往只给出理论框架，缺少可直接运行的代码实现。对于想验证策略效果或学习量化开发的开发者来说，能够亲手复现一个完整策略流程，比单纯阅读研报更有实际价值。

ETF双因子轮动策略结合了动量因子和波动率因子，通过定期评估不同ETF的表现，动态调整持仓，力求在市场轮动中捕捉收益。本文将基于Python，完整复现从数据获取、因子计算、标的筛选到回测分析的ETF双因子轮动全流程，并提供可直接运行的全套源码。

文章适合有一定Python基础，对量化投资感兴趣，希望从理论到实践完整掌握一个策略实现的开发者。通过本文，你将学会如何构建一个可运行的量化策略框架，理解因子计算和组合调仓的核心逻辑，并掌握策略回测和绩效分析的基本方法。

1. 理解ETF双因子轮动的核心逻辑

1.1 什么是ETF轮动策略

ETF轮动策略的核心思想是"强者恒强"——在一组不同类型的ETF中，定期选择近期表现较好的品种进行投资，卖出表现较差的品种。这种策略基于市场动量效应，即过去一段时间表现好的资产，在未来短期内更有可能继续表现良好。

与单一资产长期持有相比，轮动策略试图通过动态调整持仓，在不同市场环境下都能抓住表现最好的资产类别。常见的轮动标的包括股票ETF、债券ETF、商品ETF等大类资产。

1.2 双因子模型的优势

单纯使用动量因子容易在市场转折点时产生较大回撤。双因子模型通过引入波动率因子进行优化：

• 动量因子：衡量资产近期价格表现，常用过去N日的累计收益率
• 波动率因子：衡量资产价格波动程度，常用过去N日的收益率标准差

双因子组合可以在追求收益的同时控制风险，避免过度追逐高波动但风险较大的资产。典型的做法是对两个因子分别排序，然后综合评分选择最优标的。

1.3 策略执行流程

一个完整的轮动策略包含以下关键步骤：

1. 确定标的池：选择一组具有代表性的ETF作为轮动候选
2. 数据获取与处理：获取历史价格数据，计算复权价格
3. 因子计算：定期计算每个ETF的动量得分和波动率得分
4. 标的筛选：根据因子综合评分选择要投资的ETF
5. 调仓执行：在调仓日卖出原有持仓，买入新选中的ETF
6. 绩效评估：回测策略表现，分析收益、风险等指标

2. 环境准备与数据源配置

2.1 Python环境与依赖库

策略实现需要以下核心库，建议使用Python 3.8+版本：

pip install pandas>=1.4.0 pip install numpy>=1.21.0 pip install matplotlib>=3.5.0 pip install seaborn>=0.11.0 pip install tushare>=1.2.0 pip install backtrader>=1.9.0

主要依赖库的作用：

• pandas：数据处理和分析
• numpy：数值计算
• matplotlib&seaborn：可视化展示
• tushare：金融数据获取
• backtrader：回测框架

2.2 数据源配置与API设置

本文使用Tushare作为数据源，需要先注册获取token：

import tushare as ts import pandas as pd import numpy as np import matplotlib.pyplot as plt import seaborn as sns from datetime import datetime, timedelta import warnings warnings.filterwarnings('ignore') # Tushare token配置（需要先注册获取） ts.set_token('你的Tushare_token') pro = ts.pro_api() # 初始化回测相关参数 START_DATE = '20180101' END_DATE = '20231231' INITIAL_CAPITAL = 1000000 # 初始资金100万 TRANSACTION_COST = 0.001 # 交易成本千分之一

注意：Tushare免费版有一定权限限制，实际项目中可根据需要选择付费版本或其他数据源。生产环境建议使用Wind、聚宽等专业金融数据接口。

2.3 ETF标的池选择

选择具有代表性的大类资产ETF构建标的池：

# ETF标的池配置 ETF_POOL = { '510300.SH': '沪深300ETF', # 大盘股 '159915.SZ': '创业板ETF', # 成长股 '510500.SH': '中证500ETF', # 中小盘 '512100.SH': '中证1000ETF', # 小盘股 '511260.SH': '10年国债ETF', # 利率债 '511010.SH': '5年国债ETF', # 中短期债券 '518880.SH': '黄金ETF', # 商品 '159937.SZ': '货币ETF' # 现金替代 } # 策略参数配置 FACTOR_LOOKBACK = 20 # 因子计算回溯期20日 REBALANCE_FREQ = 10 # 调仓频率10个交易日 HOLDING_NUM = 3 # 持仓ETF数量

3. 数据获取与预处理模块

3.1 历史数据下载函数

实现数据获取的核心函数：

def get_etf_daily_data(etf_list, start_date, end_date): """ 获取ETF日线行情数据 """ all_data = {} for etf_code in etf_list: try: # 获取复权因子 df_adj = pro.adj_factor(ts_code=etf_code, start_date=start_date, end_date=end_date) df_adj['trade_date'] = pd.to_datetime(df_adj['trade_date']) df_adj = df_adj.set_index('trade_date') # 获取日线数据 df_daily = pro.daily(ts_code=etf_code, start_date=start_date, end_date=end_date) df_daily['trade_date'] = pd.to_datetime(df_daily['trade_date']) df_daily = df_daily.set_index('trade_date').sort_index() # 计算复权价格 df_merged = df_daily.join(df_adj[['adj_factor']]) df_merged['adj_close'] = df_merged['close'] * df_merged['adj_factor'] / df_merged['adj_factor'].iloc[-1] # 保留关键字段 df_merged = df_merged[['open', 'high', 'low', 'close', 'vol', 'amount', 'adj_close']] all_data[etf_code] = df_merged print(f"成功获取 {etf_code} 数据，共 {len(df_merged)} 个交易日") except Exception as e: print(f"获取 {etf_code} 数据失败: {e}") return all_data # 执行数据下载 etf_data = get_etf_daily_data(list(ETF_POOL.keys()), START_DATE, END_DATE)

3.2 数据质量检查与清洗

数据质量是量化策略的基础，需要系统性的检查：

def validate_etf_data(etf_data): """ 验证ETF数据质量 """ validation_results = {} for etf_code, df in etf_data.items(): issues = [] # 检查数据完整性 if len(df) < 100: issues.append(f"数据量不足，仅有{len(df)}条记录") # 检查缺失值 missing_ratio = df.isnull().sum() / len(df) for col, ratio in missing_ratio.items(): if ratio > 0.05: # 缺失率超过5% issues.append(f"{col}列缺失率{ratio:.2%}") # 检查价格合理性 price_issues = df[(df['adj_close'] <= 0) | (df['adj_close'] > 100)] if len(price_issues) > 0: issues.append(f"发现{len(price_issues)}条异常价格记录") # 检查成交量 zero_volume = df[df['vol'] == 0] if len(zero_volume) > 0: issues.append(f"发现{len(zero_volume)}条零成交量记录") validation_results[etf_code] = { 'status': 'OK' if len(issues) == 0 else 'ISSUES', 'issues': issues, 'data_points': len(df), 'date_range': f"{df.index.min().date()} 至 {df.index.max().date()}" } return validation_results # 执行数据验证 validation_results = validate_etf_data(etf_data) # 输出验证结果 for code, result in validation_results.items(): print(f"{ETF_POOL[code]}({code}): {result['status']}") if result['issues']: for issue in result['issues']: print(f" - {issue}")

3.3 数据对齐与重采样

确保不同ETF的数据在时间轴上对齐：

def align_etf_data(etf_data): """ 对齐不同ETF的交易日期 """ # 获取所有ETF的共同交易日期 common_dates = None for df in etf_data.values(): if common_dates is None: common_dates = set(df.index) else: common_dates = common_dates.intersection(set(df.index)) common_dates = sorted(common_dates) print(f"共同交易日期数量: {len(common_dates)}") # 按共同日期对齐数据 aligned_data = {} for etf_code, df in etf_data.items(): aligned_df = df.loc[common_dates].copy() aligned_data[etf_code] = aligned_df return aligned_data, common_dates # 执行数据对齐 aligned_etf_data, trading_dates = align_etf_data(etf_data)

4. 双因子计算引擎实现

4.1 动量因子计算

动量因子反映资产的价格趋势强度：

def calculate_momentum_factor(etf_data, lookback_period): """ 计算动量因子：过去lookback_per期的累计收益率 """ momentum_factors = {} for etf_code, df in etf_data.items(): prices = df['adj_close'] # 计算每日收益率 returns = prices.pct_change().fillna(0) # 计算滚动累计收益率（动量因子） momentum = (prices / prices.shift(lookback_period) - 1).fillna(0) momentum_factors[etf_code] = momentum return momentum_factors def calculate_volatility_factor(etf_data, lookback_period): """ 计算波动率因子：过去lookback_per期的收益率标准差 """ volatility_factors = {} for etf_code, df in etf_data.items(): prices = df['adj_close'] # 计算每日收益率 returns = prices.pct_change().fillna(0) # 计算滚动波动率（年化） volatility = returns.rolling(lookback_period).std() * np.sqrt(252) volatility = volatility.fillna(volatility.mean()) volatility_factors[etf_code] = volatility return volatility_factors # 计算双因子 momentum_factors = calculate_momentum_factor(aligned_etf_data, FACTOR_LOOKBACK) volatility_factors = calculate_volatility_factor(aligned_etf_data, FACTOR_LOOKBACK)

4.2 因子标准化与综合评分

不同因子的量纲不同，需要进行标准化处理：

def normalize_factors(factors_dict, trading_dates): """ 对因子进行横截面标准化 """ # 创建因子DataFrame factors_df = pd.DataFrame(index=trading_dates) for etf_code, factor_series in factors_dict.items(): factors_df[etf_code] = factor_series # 横截面标准化：每日对所有ETF的因子值进行z-score标准化 normalized_df = factors_df.copy() for date in trading_dates: daily_factors = factors_df.loc[date] # 去除无限值和空值 valid_factors = daily_factors[np.isfinite(daily_factors)] if len(valid_factors) > 0: mean_val = valid_factors.mean() std_val = valid_factors.std() if std_val > 0: normalized_df.loc[date] = (daily_factors - mean_val) / std_val else: normalized_df.loc[date] = 0 else: normalized_df.loc[date] = 0 # 转换回字典格式 normalized_factors = {} for etf_code in factors_dict.keys(): normalized_factors[etf_code] = normalized_df[etf_code] return normalized_factors def calculate_composite_score(momentum_factors, volatility_factors, momentum_weight=0.7): """ 计算综合评分：动量因子正向，波动率因子负向 """ # 标准化因子 norm_momentum = normalize_factors(momentum_factors, trading_dates) norm_volatility = normalize_factors(volatility_factors, trading_dates) # 计算综合得分 composite_scores = {} for etf_code in momentum_factors.keys(): # 动量因子正向，波动率因子负向（波动率越低越好） composite_score = (momentum_weight * norm_momentum[etf_code] - (1 - momentum_weight) * norm_volatility[etf_code]) composite_scores[etf_code] = composite_score return composite_scores, norm_momentum, norm_volatility # 计算综合评分 composite_scores, norm_momentum, norm_volatility = calculate_composite_score( momentum_factors, volatility_factors, momentum_weight=0.7 )

4.3 因子有效性检验

在实盘前验证因子的预测能力：

def validate_factor_effectiveness(factors_dict, etf_data, holding_period=5): """ 检验因子对未来收益的预测能力 """ ic_results = {} for etf_code, factor_series in factors_dict.items(): prices = etf_data[etf_code]['adj_close'] future_returns = prices.pct_change(holding_period).shift(-holding_period) # 计算IC（信息系数） valid_data = pd.DataFrame({ 'factor': factor_series, 'future_return': future_returns }).dropna() if len(valid_data) > 0: ic = valid_data['factor'].corr(valid_data['future_return']) ic_results[etf_code] = { 'IC': ic, 'IC_abs': abs(ic), 'observations': len(valid_data) } # 整体IC分析 ic_values = [result['IC'] for result in ic_results.values()] mean_ic = np.mean(ic_values) ic_ir = mean_ic / np.std(ic_values) if np.std(ic_values) > 0 else 0 print(f"因子IC分析结果:") print(f"平均IC: {mean_ic:.4f}") print(f"ICIR: {ic_ir:.4f}") print(f"IC>0的比例: {sum(ic > 0 for ic in ic_values)/len(ic_values):.2%}") return ic_results # 检验动量因子有效性 momentum_ic = validate_factor_effectiveness(momentum_factors, aligned_etf_data)

5. 轮动策略核心逻辑

5.1 标的筛选与权重分配

基于综合评分选择最优ETF并分配权重：

def select_etf_by_score(composite_scores, trading_dates, holding_num=3): """ 根据综合评分选择持仓ETF """ selection_results = {} # 创建综合评分DataFrame score_df = pd.DataFrame(index=trading_dates) for etf_code, score_series in composite_scores.items(): score_df[etf_code] = score_series for date in trading_dates: daily_scores = score_df.loc[date] # 选择评分最高的holding_num只ETF selected_etfs = daily_scores.nlargest(holding_num).index.tolist() # 等权重分配 weights = {etf: 1.0/holding_num for etf in selected_etfs} selection_results[date] = { 'selected_etfs': selected_etfs, 'weights': weights, 'scores': daily_scores[selected_etfs].to_dict() } return selection_results def generate_rebalance_dates(trading_dates, rebalance_freq): """ 生成调仓日期序列 """ rebalance_dates = [] for i, date in enumerate(trading_dates): if i % rebalance_freq == 0: rebalance_dates.append(date) print(f"共生成 {len(rebalance_dates)} 个调仓时点") return rebalance_dates # 生成调仓计划 rebalance_dates = generate_rebalance_dates(trading_dates, REBALANCE_FREQ) selection_results = select_etf_by_score(composite_scores, trading_dates, HOLDING_NUM)

5.2 策略回测引擎

实现完整的回测逻辑：

class ETFStrategyBacktest: def __init__(self, initial_capital, transaction_cost): self.initial_capital = initial_capital self.transaction_cost = transaction_cost self.portfolio_records = [] def run_backtest(self, etf_data, selection_results, rebalance_dates): """ 执行回测 """ # 初始化持仓和资金 capital = self.initial_capital current_holdings = {} # {etf_code: shares} portfolio_value = capital trade_records = [] prev_rebalance_date = None for i, date in enumerate(trading_dates): # 获取当日价格 daily_prices = {} for etf_code, df in etf_data.items(): if date in df.index: daily_prices[etf_code] = df.loc[date, 'adj_close'] # 计算当前持仓价值 holdings_value = 0 for etf_code, shares in current_holdings.items(): if etf_code in daily_prices: holdings_value += shares * daily_prices[etf_code] portfolio_value = capital + holdings_value # 记录每日净值 self.portfolio_records.append({ 'date': date, 'portfolio_value': portfolio_value, 'cash': capital, 'holdings_value': holdings_value, 'holdings': current_holdings.copy() }) # 调仓日执行调仓 if date in rebalance_dates: if prev_rebalance_date is not None: # 卖出原有持仓 for etf_code, shares in current_holdings.items(): if etf_code in daily_prices: sell_amount = shares * daily_prices[etf_code] capital += sell_amount * (1 - self.transaction_cost) current_holdings = {} # 买入新持仓 if date in selection_results: target_etfs = selection_results[date]['selected_etfs'] weights = selection_results[date]['weights'] for etf_code in target_etfs: if etf_code in daily_prices: target_value = portfolio_value * weights[etf_code] price = daily_prices[etf_code] shares = target_value / price # 考虑交易成本 cost = target_value * self.transaction_cost actual_investment = target_value - cost actual_shares = actual_investment / price current_holdings[etf_code] = actual_shares capital -= actual_investment trade_records.append({ 'date': date, 'etf': etf_code, 'action': 'BUY', 'shares': actual_shares, 'price': price, 'amount': actual_investment }) prev_rebalance_date = date return trade_records # 执行回测 backtester = ETFStrategyBacktest(INITIAL_CAPITAL, TRANSACTION_COST) trade_records = backtester.run_backtest(aligned_etf_data, selection_results, rebalance_dates) # 转换回测结果为DataFrame portfolio_df = pd.DataFrame(backtester.portfolio_records) portfolio_df['return'] = portfolio_df['portfolio_value'].pct_change().fillna(0) portfolio_df['cum_return'] = (1 + portfolio_df['return']).cumprod() portfolio_df['nav'] = portfolio_df['portfolio_value'] / INITIAL_CAPITAL

6. 绩效评估与可视化分析

6.1 关键绩效指标计算

def calculate_performance_metrics(portfolio_df, risk_free_rate=0.03): """ 计算策略绩效指标 """ total_return = portfolio_df['nav'].iloc[-1] - 1 annual_return = (1 + total_return) ** (252/len(portfolio_df)) - 1 # 年化波动率 annual_volatility = portfolio_df['return'].std() * np.sqrt(252) # 夏普比率 sharpe_ratio = (annual_return - risk_free_rate) / annual_volatility if annual_volatility > 0 else 0 # 最大回撤 portfolio_df['peak'] = portfolio_df['nav'].cummax() portfolio_df['drawdown'] = (portfolio_df['nav'] - portfolio_df['peak']) / portfolio_df['peak'] max_drawdown = portfolio_df['drawdown'].min() # 卡玛比率 calmar_ratio = annual_return / abs(max_drawdown) if max_drawdown != 0 else 0 metrics = { '总收益率': f"{total_return:.2%}", '年化收益率': f"{annual_return:.2%}", '年化波动率': f"{annual_volatility:.2%}", '夏普比率': f"{sharpe_ratio:.2f}", '最大回撤': f"{max_drawdown:.2%}", '卡玛比率': f"{calmar_ratio:.2f}", '交易次数': len(trade_records) } return metrics # 计算绩效指标 performance_metrics = calculate_performance_metrics(portfolio_df) print("策略绩效指标:") for metric, value in performance_metrics.items(): print(f"{metric}: {value}")

6.2 策略净值可视化

def plot_strategy_performance(portfolio_df, etf_data): """ 绘制策略绩效图表 """ fig, ((ax1, ax2), (ax3, ax4)) = plt.subplots(2, 2, figsize=(15, 10)) # 净值曲线 ax1.plot(portfolio_df['date'], portfolio_df['nav'], label='双因子轮动策略', linewidth=2) # 对比基准（等权持有） benchmark_nav = pd.Series(index=portfolio_df['date'], data=1.0) for etf_code, df in etf_data.items(): etf_returns = df['adj_close'].pct_change().fillna(0) aligned_returns = etf_returns.reindex(portfolio_df['date']).fillna(0) benchmark_nav *= (1 + aligned_returns / len(etf_data)) ax1.plot(portfolio_df['date'], benchmark_nav, label='等权基准', linestyle='--') ax1.set_title('策略净值曲线') ax1.set_ylabel('累计净值') ax1.legend() ax1.grid(True) # 回撤曲线 ax2.fill_between(portfolio_df['date'], portfolio_df['drawdown'], 0, alpha=0.3, color='red') ax2.plot(portfolio_df['date'], portfolio_df['drawdown'], color='red', linewidth=1) ax2.set_title('策略回撤曲线') ax2.set_ylabel('回撤比例') ax2.grid(True) # 月度收益热力图 portfolio_df['year_month'] = portfolio_df['date'].dt.to_period('M') monthly_returns = portfolio_df.groupby('year_month')['nav'].last().pct_change().fillna(0) # 创建月度收益矩阵 years = sorted(set([period.year for period in monthly_returns.index])) months = range(1, 13) returns_matrix = pd.DataFrame(index=years, columns=months, data=np.nan) for period, ret in monthly_returns.items(): returns_matrix.loc[period.year, period.month] = ret im = ax3.imshow(returns_matrix, cmap='RdYlGn', aspect='auto', vmin=-0.1, vmax=0.1) ax3.set_title('月度收益热力图') ax3.set_xlabel('月份') ax3.set_ylabel('年份') plt.colorbar(im, ax=ax3) # 持仓分布 holding_counts = {} for record in backtester.portfolio_records: for etf_code in record['holdings']: if etf_code in holding_counts: holding_counts[etf_code] += 1 else: holding_counts[etf_code] = 1 etf_names = [ETF_POOL[code] for code in holding_counts.keys()] ax4.bar(etf_names, list(holding_counts.values())) ax4.set_title('ETF持仓频率分布') ax4.set_ylabel('持仓天数') plt.xticks(rotation=45) plt.tight_layout() plt.show() # 绘制绩效图表 plot_strategy_performance(portfolio_df, aligned_etf_data)

6.3 策略敏感性分析

检验策略对关键参数的敏感性：

def parameter_sensitivity_analysis(etf_data, trading_dates): """ 参数敏感性分析 """ results = [] # 测试不同的参数组合 lookback_periods = [10, 20, 30] rebalance_freqs = [5, 10, 20] momentum_weights = [0.5, 0.7, 0.9] for lookback in lookback_periods: for freq in rebalance_freqs: for weight in momentum_weights: # 计算因子 momentum = calculate_momentum_factor(etf_data, lookback) volatility = calculate_volatility_factor(etf_data, lookback) scores, _, _ = calculate_composite_score(momentum, volatility, weight) # 生成调仓计划 rebalance_dates = generate_rebalance_dates(trading_dates, freq) selection = select_etf_by_score(scores, trading_dates, HOLDING_NUM) # 回测 backtester = ETFStrategyBacktest(INITIAL_CAPITAL, TRANSACTION_COST) backtester.run_backtest(etf_data, selection, rebalance_dates) portfolio_df = pd.DataFrame(backtester.portfolio_records) portfolio_df['nav'] = portfolio_df['portfolio_value'] / INITIAL_CAPITAL total_return = portfolio_df['nav'].iloc[-1] - 1 results.append({ 'lookback': lookback, 'rebalance_freq': freq, 'momentum_weight': weight, 'total_return': total_return }) results_df = pd.DataFrame(results) # 分析参数敏感性 sensitivity = results_df.groupby('lookback')['total_return'].mean() print("回溯期敏感性:") for lookback, ret in sensitivity.items(): print(f" 回溯期{lookback}天: 平均收益{ret:.2%}") return results_df # 执行敏感性分析 sensitivity_results = parameter_sensitivity_analysis(aligned_etf_data, trading_dates)

7. 常见问题与实战优化

7.1 数据质量问题处理

def handle_data_issues(etf_data): """ 处理常见数据问题 """ cleaned_data = {} for etf_code, df in etf_data.items(): cleaned_df = df.copy() # 处理价格异常值 price_changes = cleaned_df['adj_close'].pct_change().abs() abnormal_price = price_changes > 0.2 # 单日涨跌幅超过20%视为异常 if abnormal_price.any(): print(f"发现 {etf_code} 价格异常，进行平滑处理") # 使用前后价格平均进行平滑 for idx in cleaned_df[abnormal_price].index: prev_idx = cleaned_df.index.get_loc(idx) - 1 next_idx = cleaned_df.index.get_loc(idx) + 1 if prev_idx >= 0 and next_idx < len(cleaned_df): avg_price = (cleaned_df.iloc[prev_idx]['adj_close'] + cleaned_df.iloc[next_idx]['adj_close']) / 2 cleaned_df.loc[idx, 'adj_close'] = avg_price cleaned_data[etf_code] = cleaned_df return cleaned_data

7.2 策略过拟合防范

回测中常见的过拟合问题及应对措施：

1. 参数优化过拟合：避免在过多参数组合中寻找最优解
2. 数据窥探偏差：使用样本外数据验证策略效果
3. 生存者偏差：考虑已退市ETF的影响

def out_of_sample_test(etf_data, trading_dates, train_ratio=0.7): """ 样本外测试 """ # 划分训练集和测试集 split_idx = int(len(trading_dates) * train_ratio) train_dates = trading_dates[:split_idx] test_dates = trading_dates[split_idx:] # 在训练集上确定最优参数 train_data = {} for etf_code, df in etf_data.items(): train_data[etf_code] = df.loc[train_dates] # 使用默认参数（避免过度优化） momentum = calculate_momentum_factor(train_data, FACTOR_LOOKBACK) volatility = calculate_volatility_factor(train_data, FACTOR_LOOKBACK) scores, _, _ = calculate_composite_score(momentum, volatility, 0.7) # 在测试集上验证 test_rebalance_dates = [date for date in rebalance_dates if date in test_dates] test_selection = select_etf_by_score(scores, test_dates, HOLDING_NUM) test_data = {} for etf_code, df in etf_data.items(): test_data[etf_code] = df.loc[test_dates] backtester = ETFStrategyBacktest(INITIAL_CAPITAL, TRANSACTION_COST) backtester.run_backtest(test_data, test_selection, test_rebalance_dates) portfolio_df = pd.DataFrame(backtester.portfolio_records) test_return = portfolio_df['portfolio_value'].iloc[-1] / INITIAL_CAPITAL - 1 print(f"样本外测试收益: {test_return:.2%}") return test_return # 执行样本外测试 oos_return = out_of_sample_test(aligned_etf_data, trading_dates)

7.3 实盘注意事项

从回测到实盘需要额外考虑的因素：

1. 交易冲击成本：大额交易对价格的影响
2. 流动性限制：避免在低流动性ETF上配置过大权重
3. 实时数据质量：实盘数据与历史数据的差异
4. 策略监控：建立策略失效预警机制

def real_trading_enhancements(selection_results, etf_data): """ 实盘增强功能 """ enhanced_selection = {} for date, selection in selection_results.items(): enhanced_selection[date] = selection.copy() # 流动性过滤：剔除日均成交额过低的ETF liquid_etfs = [] for etf_code in selection['selected_etfs']: if date in etf_data[etf_code].index: # 检查近期平均成交额 recent_data = etf_data[etf_code].loc[:date].tail(20) avg_turnover = recent_data['amount'].mean() if avg_turnover > 1e7: # 日均成交额大于1000万 liquid_etfs.append(etf_code) else: print(f"{date}: {etf_code} 流动性不足，日均成交额{avg_turnover:.0f}") if len(liquid_etfs) > 0: enhanced_selection[date]['selected_etfs'] = liquid_etfs # 重新分配权重 new_weights = {etf: 1.0/len(liquid_etfs) for etf in liquid_etfs} enhanced_selection[date]['weights'] = new_weights else: # 所有ETF流动性都不足，持有现金 enhanced_selection[date]['selected_etfs'] = [] enhanced_selection[date]['weights'] = {} return enhanced_selection

8. 策略扩展与优化方向

8.1 多因子模型扩展

除了动量和波动率，可以考虑加入更多因子：

• 估值因子：PE、PB等估值指标
• 质量因子：ROE、营收增长率等基本面指标
• 技术因子：均线排列、MACD等技术指标

def multi_factor_integration(etf_data, additional_factors): """ 多因子整合框架 """ # 计算基础因子 momentum = calculate_momentum_factor(etf_data, 20) volatility = calculate_volatility_factor(etf_data, 20) # 标准化所有因子 all_factors = {'momentum': momentum, 'volatility': volatility} all_factors.update(additional_factors) normalized_factors = {} for factor_name, factor_dict in all_factors.items(): normalized_factors[factor_name] = normalize_factors(factor_dict, trading_dates) # 因子加权（可根据IC值动态调整权重） factor_weights = { 'momentum': 0.4