因子投资实战(三):用Python挖掘Alpha因子

classFactorMomentumStrategy:"""    因子动量策略：根据因子近期表现动态调整权重    """def__init__(self, lookback_months=6, momentum_window=3):self.lookback_months = lookback_monthsself.momentum_window = momentum_windowself.factor_momentum_scores = {}defcalculate_factor_momentum(self, factor_performance, current_period):"""        计算因子动量得分        """        momentum_scores = {}for factor_name, perf in factor_performance.items():# 获取最近N期的收益            start_idx = max(0, current_period - self.lookback_months)            recent_returns = []for i inrange(start_idx, current_period):if i < len(perf['period_returns']):                    recent_returns.append(perf['period_returns'][i]['long_short_return'])iflen(recent_returns) >= self.momentum_window:# 计算动量得分（最近M期的平均收益）                momentum_score = np.mean(recent_returns[-self.momentum_window:])                momentum_scores[factor_name] = momentum_scorereturn momentum_scoresdefdynamic_factor_weighting(self, factor_performance, current_period):"""        动态因子加权        """# 计算因子动量        momentum_scores = self.calculate_factor_momentum(factor_performance, current_period)ifnot momentum_scores:# 如果没有动量数据，使用等权重            factor_names = list(factor_performance.keys())            weights = {factor: 1.0/len(factor_names) for factor in factor_names}return weights# 根据动量得分计算权重（动量越强，权重越高）        momentum_values = np.array(list(momentum_scores.values()))# 使用softmax函数将动量得分转换为权重        exp_scores = np.exp(momentum_values)        weights = exp_scores / exp_scores.sum()        weighted_weights = dict(zip(momentum_scores.keys(), weights))# 归一化确保权重和为1        total_weight = sum(weighted_weights.values())if total_weight > 0:            normalized_weights = {k: v/total_weight for k, v in weighted_weights.items()}else:            factor_names = list(factor_performance.keys())            normalized_weights = {factor: 1.0/len(factor_names) for factor in factor_names}return normalized_weights# 测试因子动量策略momentum_strategy = FactorMomentumStrategy(lookback_months=6, momentum_window=3)# 模拟动态调整过程print("\n因子动量策略动态权重调整示例:")for period inrange(5, 10):    dynamic_weights = momentum_strategy.dynamic_factor_weighting(tester.factor_performance, period)print(f"\n第{period}期动态权重:")for factor, weight in dynamic_weights.items():print(f"  {factor}: {weight:.3f}")

classMarketRegimeAdjustment:"""    市场状态自适应：根据不同市场状态调整因子暴露    """def__init__(self):self.market_regimes = {}  # 存储市场状态defdetect_market_regime(self, market_data, window=60):"""        检测市场状态        """# 计算市场指标        returns = market_data['returns']# 波动率状态        volatility = returns.rolling(window).std()        high_vol = volatility > volatility.quantile(0.7)        low_vol = volatility < volatility.quantile(0.3)# 趋势状态        trend = market_data['close'].rolling(window).mean()        price_above_ma = market_data['close'] > trend        price_below_ma = market_data['close'] < trend# 市场状态分类        regimes = pd.Series('Neutral', index=market_data.index)# 高波动熊市        regimes[(high_vol) & (price_below_ma)] = 'HighVol_Bear'# 高波动牛市        regimes[(high_vol) & (price_above_ma)] = 'HighVol_Bull'# 低波动熊市        regimes[(low_vol) & (price_below_ma)] = 'LowVol_Bear'# 低波动牛市        regimes[(low_vol) & (price_above_ma)] = 'LowVol_Bull'return regimesdefget_regime_specific_weights(self, regime, factor_performance_history):"""        获取特定市场状态的因子权重        """# 根据不同市场状态预设权重# 实际应用中需要基于历史数据优化        regime_weights = {'HighVol_Bear': {# 熊市中，价值因子和质量因子可能更有效'PE': 0.4, 'PB': 0.3, 'LOG_MV': 0.3            },'LowVol_Bull': {# 牛市中，动量因子可能更有效'REV_1M': 0.5, 'VOL_20D': 0.3, 'LOG_MV': 0.2            },'HighVol_Bull': {# 高波动牛市中，小盘和动量可能表现好'LOG_MV': 0.4, 'REV_1M': 0.4, 'VOL_20D': 0.2            },'LowVol_Bear': {# 低波动熊市中，防御性因子可能更有效'PE': 0.5, 'PB': 0.5            },'Neutral': {# 中性市场使用等权重'PE': 0.2, 'PB': 0.2, 'LOG_MV': 0.2, 'REV_1M': 0.2, 'VOL_20D': 0.2            }        }# 如果该状态没有预设权重，使用中性权重if regime notin regime_weights:            regime = 'Neutral'return regime_weights.get(regime, regime_weights['Neutral'])

deffactor_investing_best_practices():"""    因子投资最佳实践建议    """    practices = {'因子选择': ['选择经济逻辑清晰的因子','避免数据挖掘导致的过拟合','考虑因子的持久性和普适性','定期检查因子有效性'        ],'数据处理': ['处理极端值和缺失值','注意幸存者偏差','避免前视偏差','考虑交易成本和流动性'        ],'组合构建': ['适当分散因子暴露','控制换手率和交易成本','考虑现实约束（最小投资额、流动性等）','定期再平衡'        ],'风险管理': ['监控因子拥挤度','设置最大回撤止损','准备极端情况应对方案','保持策略透明度和可解释性'        ],'持续改进': ['定期回顾和优化模型','跟踪学术研究和行业实践','保持策略的适应性','文档化所有决策过程'        ]    }return practices# 打印最佳实践practices = factor_investing_best_practices()print("\n因子投资最佳实践:")for category, items in practices.items():print(f"\n{category}:")for item in items:print(f"  • {item}")

defcommon_factor_pitfalls():"""    因子投资常见陷阱及规避方法    """    pitfalls = pd.DataFrame({'陷阱': ['过拟合历史数据','忽略交易成本','前视偏差','幸存者偏差','因子拥挤','忽略市场状态','过度优化','黑箱模型'        ],'表现': ['回测表现优异，实盘表现糟糕','理论收益被成本侵蚀','使用未来信息导致虚高收益','回测包含已退市股票导致偏差','太多人使用同一因子导致失效','策略在某些市场环境下失效','对参数过度敏感，稳健性差','无法解释失败原因'        ],'规避方法': ['使用样本外测试、交叉验证','回测中计入完整交易成本','严格检查数据时间戳','使用生存偏差调整后的数据','监控因子拥挤度，适时调整','开发自适应策略','简化模型，重视经济逻辑','保持模型透明，定期审查'        ]    })return pitfalls# 显示常见陷阱print("\n因子投资常见陷阱及规避方法:")print(common_factor_pitfalls().to_string(index=False))