256 lines
9.2 KiB
Python
256 lines
9.2 KiB
Python
"""
|
|
因子研究模块 — 因子定义、计算、注册与 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() |