""" 因子研究模块 — 因子定义、计算、注册与 IC/分层分析 """ from abc import ABC, abstractmethod from dataclasses import dataclass, field from typing import Callable, Dict, List, Optional, Tuple import numpy as np import pandas as pd from scipy import stats # --------------------------------------------------------------------------- # 因子注册表 # --------------------------------------------------------------------------- class FactorRegistry: """因子注册表，用于管理和发现所有因子""" _factors: Dict[str, type] = {} @classmethod def register(cls, factor_cls: type): """注册因子类""" name = factor_cls.__name__ cls._factors[name] = factor_cls return factor_cls @classmethod def get(cls, name: str) -> Optional[type]: """根据名称获取因子类""" return cls._factors.get(name) @classmethod def list_all(cls) -> List[str]: """列出所有已注册因子""" return sorted(cls._factors.keys()) @classmethod def create(cls, name: str, **kwargs): """根据名称创建因子实例""" factor_cls = cls._factors.get(name) if factor_cls is None: raise KeyError(f"因子 '{name}' 未注册，可用: {cls.list_all()}") return factor_cls(**kwargs) # --------------------------------------------------------------------------- # 因子基类 # --------------------------------------------------------------------------- class BaseFactor(ABC): """因子抽象基类""" name: str = "BaseFactor" category: str = "unknown" # technical / fundamental / alternative def __init__(self, **params): self.params = params @abstractmethod def compute(self, data: pd.DataFrame) -> pd.Series: """计算因子值，返回 Series(index=timestamp) 或多索引 Series""" ... def __repr__(self): return f"{self.name}({self.params})" def neutralize(self, factor: pd.Series, group: pd.Series, market_cap: pd.Series) -> pd.Series: """ 因子中性化：对市值和行业做正交化处理。 factor : 原始因子值 group : 行业分组 market_cap : 市值 返回中性化后的因子值。 """ # 创建虚拟变量并拟合线性模型，取残差作为中性化因子 df = pd.DataFrame({ "factor": factor, "log_mcap": np.log(market_cap.clip(lower=1)), }) dummies = pd.get_dummies(group, prefix="group") X = pd.concat([df["log_mcap"], dummies], axis=1) # 对齐索引 common_idx = factor.index.intersection(X.dropna().index) X = X.loc[common_idx] y = factor.loc[common_idx] # OLS 回归取残差 beta = np.linalg.lstsq(X.values, y.values, rcond=None)[0] pred = X.values @ beta residual = y.values - pred neutralized = pd.Series(residual, index=common_idx, name=factor.name) return neutralized # --------------------------------------------------------------------------- # 技术面因子 # --------------------------------------------------------------------------- @FactorRegistry.register class TechnicalFactor(BaseFactor): """通用技术面因子 — 支持自定义参数组合""" name = "TechnicalFactor" category = "technical" def compute(self, data: pd.DataFrame) -> pd.Series: """ data 必须包含: ['close', 'volume', 'high', 'low'] 返回: 标准化后的多因子合成值 """ # 示例因子: # 1) 动量因子 (20日收益率) # 2) 波动率因子 (20日波动率倒数) close = data["close"] mom = close.pct_change(self.params.get("mom_window", 20)) vol = close.pct_change().rolling(self.params.get("vol_window", 20)).std() # 合成 (等权，可按需要改为 ICIR 加权) mom_z = (mom - mom.mean()) / mom.std() vol_z = -(vol - vol.mean()) / vol.std() # 低波为正 composite = 0.5 * mom_z + 0.5 * vol_z return composite.rename("TechnicalFactor") # --------------------------------------------------------------------------- # 基本面因子 # --------------------------------------------------------------------------- @FactorRegistry.register class FundamentalFactor(BaseFactor): """通用基本面因子 — ROE / PE / PB / 营收增速等""" name = "FundamentalFactor" category = "fundamental" def compute(self, data: pd.DataFrame) -> pd.Series: """ data 应包含: ['roe', 'pe', 'pb', 'revenue_growth'] 返回基本面复合因子 """ roe = data.get("roe", None) pe = data.get("pe", None) pb = data.get("pb", None) growth = data.get("revenue_growth", None) # 将各因子标准化后等权合成 scores = [] for series, inv in [(roe, False), (pe, True), (pb, True), (growth, False)]: if series is not None: z = (series - series.mean()) / series.std() scores.append(-z if inv else z) if not scores: raise ValueError("无可用基本面数据列") composite = sum(scores) / len(scores) return composite.rename("FundamentalFactor") # --------------------------------------------------------------------------- # IC / 分层分析工具 # --------------------------------------------------------------------------- class FactorAnalyzer: """因子分析器 — IC 计算、分层回测、换手率分析""" def __init__(self, factor_values: pd.Series, forward_returns: pd.Series): """ factor_values : 因子值 Series (MultiIndex: [date, code] 或 对齐 index) forward_returns : 未来一期收益率，与 factor 对齐 """ self.factor = factor_values self.forward = forward_returns self.ic_series: Optional[pd.Series] = None def compute_ic(self, method: str = "rank") -> pd.Series: """逐截面计算 IC (Rank IC 或 Pearson IC)""" if isinstance(self.factor.index, pd.MultiIndex): grouped = self.factor.groupby(level=0) fwd = self.forward.groupby(level=0) if method == "rank": ic_data = grouped.apply( lambda g: g.corr(fwd.get_group(g.name), method="spearman") ) else: ic_data = grouped.apply( lambda g: g.corr(fwd.get_group(g.name), method="pearson") ) else: # 单截面 if method == "rank": ic_data = pd.Series( stats.spearmanr(self.factor, self.forward)[0], index=[self.factor.index[0] if len(self.factor) > 0 else 0], ) else: ic_data = pd.Series( stats.pearsonr(self.factor, self.forward)[0], index=[self.factor.index[0] if len(self.factor) > 0 else 0], ) self.ic_series = ic_data.dropna() return self.ic_series def ic_summary(self) -> Dict[str, float]: """IC 汇总统计""" if self.ic_series is None: self.compute_ic() ic = self.ic_series return { "IC_Mean": ic.mean(), "IC_Std": ic.std(), "IR": ic.mean() / (ic.std() + 1e-12), "IC>0_Ratio": (ic > 0).mean(), "IC_Abs_Mean": ic.abs().mean(), } def ic_decay(self, forward_returns: Dict[int, pd.Series]) -> pd.Series: """IC 衰减: 计算不同前瞻期的 IC 均值""" decay = {} for horizon, fwd in forward_returns.items(): ratio = fwd.reindex(self.forward.index) valid = self.factor.notna() & ratio.notna() ic = stats.spearmanr( self.factor[valid].values, ratio[valid].values )[0] decay[horizon] = ic return pd.Series(decay, name="IC_Decay") def quantile_returns(self, n_quantiles: int = 5) -> pd.DataFrame: """分层回测：按因子值分 5 组，计算各组平均收益率""" df = pd.DataFrame({"factor": self.factor, "fwd": self.forward}).dropna() df["quantile"] = pd.qcut(df["factor"], n_quantiles, labels=False) + 1 result = df.groupby("quantile")["fwd"].mean().to_frame("avg_return") result.index.name = "quantile" result["cum_return"] = result["avg_return"].cumsum() return result def turnover(self, n_quantiles: int = 5) -> pd.Series: """因子换手率分析（相邻期分位数变化比例）""" df = pd.DataFrame({"factor": self.factor}) df["quantile"] = pd.qcut(df["factor"], n_quantiles, labels=False) + 1 if isinstance(df.index, pd.MultiIndex): df = df.reset_index() date_col = df.columns[0] turnover_list = [] for date, grp in df.groupby(date_col): pass # 此处需按股票计算 --- 简化处理 # 简化: 逐期计算因子自相关系数 if isinstance(self.ic_series, pd.Series): auto_corr = self.factor.autocorr(lag=1) return pd.Series({"auto_corr": auto_corr}) return pd.Series()