135 lines
6.0 KiB
Python
135 lines
6.0 KiB
Python
"""合成层 v2b: 财务源市值中性化 —— cs_neutralize 截面 OLS 残差算子 + fund_*_neu 六源.
|
||
|
||
治「财务源 2021 后沦为微盘风格代理稀释合成」: 财务源因子对总市值(size)截面
|
||
中性化取残差,剥离 size 因子暴露后仅留财务信息本身.中性化变量只用 size
|
||
(close × share_capital 总市值,表达式层两列现成可得);BP 不进中性化变量——
|
||
fund_bp 自身即 BP,对自身回归残差恒零.
|
||
|
||
cs_neutralize(y, x): 逐日截面 OLS 残差 resid = y − a − b·x
|
||
(b=cov(x,y)/var(x), a=mean(y)−b·mean(x)),纯 polars over("datetime") 聚合代数
|
||
(cs_rank 同款窗口广播,无 python 逐组循环).边界: 当日有效配对样本 <3 或
|
||
var(x)=0 → 该日残差 null;x/y 任一 null 的行残差 null(配对掩码,缺一侧的行
|
||
不进回归统计).注册进 vnpy EXPRESSION_FUNCTIONS(fast_ops 官方扩展点,导入即注册,
|
||
表达式字符串求值可达).
|
||
|
||
FUND_NEU_SOURCES: 6 项 = composite_library.FUND_SOURCES 同名源加 _neu 后缀,
|
||
方向全 "+"(中性化因子保持「高=好」语义,负号在表达式内定向,同原源模式).
|
||
neu 表达式 = cs_rank((±)cs_neutralize(<原指标式>, size)),其中原指标式与负号
|
||
严格从注册表原源表达式剥掉外层 cs_rank 后代码取得(不手抄,防两处漂移);
|
||
原内层负号(如 '-nsi')移到 cs_neutralize 外——OLS 线性 resid(−y)=−resid(y),
|
||
统一「中性化正指标+外置符号」形态.
|
||
"""
|
||
import os
|
||
import sys
|
||
|
||
import polars as pl
|
||
|
||
# Inject vnpy source path (follow fast_ops / tests conftest pattern)
|
||
_VNPY_SRC = os.path.abspath(os.path.join(os.path.dirname(__file__), "..", "vnpy_v4.4.0"))
|
||
if _VNPY_SRC not in sys.path:
|
||
sys.path.insert(0, _VNPY_SRC)
|
||
|
||
from vnpy.alpha.dataset.utility import DataProxy, EXPRESSION_FUNCTIONS
|
||
|
||
from . import composite_library
|
||
from .registry import register_factor, get_factor, _REGISTRY
|
||
|
||
# size 中性化变量 = 总市值(close × share_capital,与 fundamental_library 估值族同式)
|
||
SIZE_EXPRESSION = "close * share_capital"
|
||
_NEU_SUFFIX = "_neu"
|
||
_MIN_SAMPLES = 3 # 当日有效配对样本 <3 无截面回归意义 → 残差 null
|
||
|
||
|
||
def cs_neutralize(feature_y: DataProxy, feature_x: DataProxy) -> DataProxy:
|
||
"""逐日截面 OLS 残差: resid = y − a − b·x(b=cov/var, a=mean_y−b·mean_x).
|
||
|
||
纯 polars over("datetime") 聚合代数(组统计量广播回行,无逐组循环);
|
||
配对掩码: x 或 y null 的行不进回归统计且自身残差 null.
|
||
"""
|
||
# 两操作数源自同一父 df 且全算子保长保序(fast_ts_corr 同前提)——
|
||
# 按位拼接等价于按键 join,免大表 hash join
|
||
h_y, h_x = feature_y.df.height, feature_x.df.height
|
||
if h_y != h_x:
|
||
raise ValueError(f"cs_neutralize 操作数长度不一致: {h_y} vs {h_x}(行序对齐前提被破坏)")
|
||
df = feature_y.df.with_columns(feature_x.df["data"].alias("data_x"))
|
||
|
||
y, x = pl.col("data"), pl.col("data_x")
|
||
pair = y.is_not_null() & x.is_not_null()
|
||
ym = pl.when(pair).then(y) # 配对掩码后仅有效对进统计(null 自动被 mean 跳过)
|
||
xm = pl.when(pair).then(x)
|
||
n = xm.count().over("datetime")
|
||
mean_y = ym.mean().over("datetime")
|
||
mean_x = xm.mean().over("datetime")
|
||
var_x = (xm * xm).mean().over("datetime") - mean_x * mean_x
|
||
cov_xy = (xm * ym).mean().over("datetime") - mean_x * mean_y
|
||
slope = cov_xy / var_x
|
||
resid = y - (mean_y + slope * (x - mean_x)) # = y − a − b·x
|
||
result = pl.when((n >= _MIN_SAMPLES) & (var_x > 0)).then(resid).otherwise(None)
|
||
|
||
out = df.select(pl.col("datetime"), pl.col("vt_symbol"), result.alias("data"))
|
||
return DataProxy(out)
|
||
|
||
|
||
def register_neutralize_ops() -> list[str]:
|
||
"""注册 cs_neutralize 进 vnpy EXPRESSION_FUNCTIONS(幂等),返回算子名清单."""
|
||
EXPRESSION_FUNCTIONS["cs_neutralize"] = cs_neutralize
|
||
return ["cs_neutralize"]
|
||
|
||
|
||
# 财务 6 源中性化版源表(方向全 "+": cs_neutralize 线性保持定向,cs_rank 后仍高=好)
|
||
FUND_NEU_SOURCES: list[tuple[str, str]] = [
|
||
(f"{name}{_NEU_SUFFIX}", "+") for name, _direction in composite_library.FUND_SOURCES
|
||
]
|
||
|
||
|
||
def _strip_cs_rank(expression: str) -> str:
|
||
"""剥掉外层 cs_rank(...): 'cs_rank(-nsi)' → '-nsi';非 cs_rank 包裹 raise."""
|
||
prefix = "cs_rank("
|
||
if not (expression.startswith(prefix) and expression.endswith(")")):
|
||
raise ValueError(f"财务源表达式非 cs_rank 包裹,无法剥壳: {expression!r}")
|
||
return expression[len(prefix):-1]
|
||
|
||
|
||
def build_neutralized_expression(source_expression: str) -> str:
|
||
"""原源注册表达式 → 市值中性化版: cs_rank((±)cs_neutralize(<原指标式>, size)).
|
||
|
||
原内层负号(如 '-nsi')移到 cs_neutralize 外成 '(-1) *'(resid(−y)=−resid(y)
|
||
数学等价,形态统一);正内层直接中性化.不手抄指标式,符号从入参表达式派生.
|
||
"""
|
||
inner = _strip_cs_rank(source_expression)
|
||
if inner.startswith("-"):
|
||
signed = f"(-1) * cs_neutralize({inner[1:]}, {SIZE_EXPRESSION})"
|
||
else:
|
||
signed = f"cs_neutralize({inner}, {SIZE_EXPRESSION})"
|
||
return f"cs_rank({signed})"
|
||
|
||
|
||
def _register_all() -> None:
|
||
"""注册 6 个 fund_*_neu 因子(幂等,category=fundamental→合成层内嵌项口径;
|
||
先幂等重挂 v1.1 源,registry 被清后可独立重建)."""
|
||
composite_library._register_all()
|
||
register_neutralize_ops()
|
||
for src_name, _direction in composite_library.FUND_SOURCES:
|
||
neu_name = f"{src_name}{_NEU_SUFFIX}"
|
||
if neu_name in _REGISTRY:
|
||
continue
|
||
src = get_factor(src_name)
|
||
if src is None:
|
||
raise ValueError(f"中性化源因子未注册: {src_name}")
|
||
register_factor(neu_name, build_neutralized_expression(src["expression"]),
|
||
category="fundamental")
|
||
|
||
|
||
# 模块导入时自动注册(与 fundamental_library / composite_library 同模式);
|
||
# cs_neutralize 同步进 EXPRESSION_FUNCTIONS,表达式字符串求值可达
|
||
_register_all()
|
||
|
||
|
||
__all__ = [
|
||
"cs_neutralize",
|
||
"register_neutralize_ops",
|
||
"FUND_NEU_SOURCES",
|
||
"SIZE_EXPRESSION",
|
||
"build_neutralized_expression",
|
||
]
|