在实际量化投资领域券商研报提供了很多有价值的策略思路但往往只给出理论框架缺少可直接运行的代码实现。对于想验证策略效果或学习量化开发的开发者来说能够亲手复现一个完整策略流程比单纯阅读研报更有实际价值。ETF双因子轮动策略结合了动量因子和波动率因子通过定期评估不同ETF的表现动态调整持仓力求在市场轮动中捕捉收益。本文将基于Python完整复现从数据获取、因子计算、标的筛选到回测分析的ETF双因子轮动全流程并提供可直接运行的全套源码。文章适合有一定Python基础对量化投资感兴趣希望从理论到实践完整掌握一个策略实现的开发者。通过本文你将学会如何构建一个可运行的量化策略框架理解因子计算和组合调仓的核心逻辑并掌握策略回测和绩效分析的基本方法。1. 理解ETF双因子轮动的核心逻辑1.1 什么是ETF轮动策略ETF轮动策略的核心思想是强者恒强——在一组不同类型的ETF中定期选择近期表现较好的品种进行投资卖出表现较差的品种。这种策略基于市场动量效应即过去一段时间表现好的资产在未来短期内更有可能继续表现良好。与单一资产长期持有相比轮动策略试图通过动态调整持仓在不同市场环境下都能抓住表现最好的资产类别。常见的轮动标的包括股票ETF、债券ETF、商品ETF等大类资产。1.2 双因子模型的优势单纯使用动量因子容易在市场转折点时产生较大回撤。双因子模型通过引入波动率因子进行优化动量因子衡量资产近期价格表现常用过去N日的累计收益率波动率因子衡量资产价格波动程度常用过去N日的收益率标准差双因子组合可以在追求收益的同时控制风险避免过度追逐高波动但风险较大的资产。典型的做法是对两个因子分别排序然后综合评分选择最优标的。1.3 策略执行流程一个完整的轮动策略包含以下关键步骤确定标的池选择一组具有代表性的ETF作为轮动候选数据获取与处理获取历史价格数据计算复权价格因子计算定期计算每个ETF的动量得分和波动率得分标的筛选根据因子综合评分选择要投资的ETF调仓执行在调仓日卖出原有持仓买入新选中的ETF绩效评估回测策略表现分析收益、风险等指标2. 环境准备与数据源配置2.1 Python环境与依赖库策略实现需要以下核心库建议使用Python 3.8版本pip install pandas1.4.0 pip install numpy1.21.0 pip install matplotlib3.5.0 pip install seaborn0.11.0 pip install tushare1.2.0 pip install backtrader1.9.0主要依赖库的作用pandas数据处理和分析numpy数值计算matplotlibseaborn可视化展示tushare金融数据获取backtrader回测框架2.2 数据源配置与API设置本文使用Tushare作为数据源需要先注册获取tokenimport 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_codeetf_code, start_datestart_date, end_dateend_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_codeetf_code, start_datestart_date, end_dateend_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(indextrading_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_weight0.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_weight0.7 )4.3 因子有效性检验在实盘前验证因子的预测能力def validate_factor_effectiveness(factors_dict, etf_data, holding_period5): 检验因子对未来收益的预测能力 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(fICIR: {ic_ir:.4f}) print(fIC0的比例: {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_num3): 根据综合评分选择持仓ETF selection_results {} # 创建综合评分DataFrame score_df pd.DataFrame(indextrading_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_CAPITAL6. 绩效评估与可视化分析6.1 关键绩效指标计算def calculate_performance_metrics(portfolio_df, risk_free_rate0.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双因子轮动策略, linewidth2) # 对比基准等权持有 benchmark_nav pd.Series(indexportfolio_df[date], data1.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, alpha0.3, colorred) ax2.plot(portfolio_df[date], portfolio_df[drawdown], colorred, linewidth1) 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(indexyears, columnsmonths, datanp.nan) for period, ret in monthly_returns.items(): returns_matrix.loc[period.year, period.month] ret im ax3.imshow(returns_matrix, cmapRdYlGn, aspectauto, vmin-0.1, vmax0.1) ax3.set_title(月度收益热力图) ax3.set_xlabel(月份) ax3.set_ylabel(年份) plt.colorbar(im, axax3) # 持仓分布 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(rotation45) 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 数据质量问题处理问题现象可能原因解决方案部分ETF数据缺失上市时间不同、停牌等使用共同交易日期剔除数据不完整的ETF价格异常跳动除权除息、数据错误使用复权价格设置价格变动阈值过滤成交量为零非交易日、数据错误检查日期是否为实际交易日剔除异常数据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_data7.2 策略过拟合防范回测中常见的过拟合问题及应对措施参数优化过拟合避免在过多参数组合中寻找最优解数据窥探偏差使用样本外数据验证策略效果生存者偏差考虑已退市ETF的影响def out_of_sample_test(etf_data, trading_dates, train_ratio0.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 实盘注意事项从回测到实盘需要额外考虑的因素交易冲击成本大额交易对价格的影响流动性限制避免在低流动性ETF上配置过大权重实时数据质量实盘数据与历史数据的差异策略监控建立策略失效预警机制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_selection8. 策略扩展与优化方向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