176 lines
8.2 KiB
Python
176 lines
8.2 KiB
Python
"""合成层 v2a(Weighting): 三层权重档案——L1 模型权重 + L2 人工覆盖 + L3 表达式装配.
|
|
|
|
权重专题拍板的 v2a 落地: 在 v1.1 等权 baseline 之上,把 12 量价源权重从
|
|
模型拟合(L1)与人工干预(L2)两轨合成,装配成带权合成表达式因子
|
|
composite_quant12_v2a;方向/包裹口径完全复用 composite_library v1.1
|
|
(rank_term/_embed_term,不复制实现).
|
|
|
|
三层语义:
|
|
- L1 模型权重层: weight_profiles/*.json 档案(sources.{src}.weight =
|
|
fit_icir 拟合权重,非负、sum≈1),档案是唯一事实源,代码零硬编码权重;
|
|
- L2 覆盖层(overrides): 人工干预双轨验盘——每项 {source, weight, reason}
|
|
替换对应源权重后**全组重归一化** sum=1(reason 必填=干预留痕;重归一而非
|
|
局部缩放: 人工降权某源,其余源按 L1 相对结构分享余量);
|
|
- L3 表达式层: build_weighted_expression 拼接 (Σ w_i·项_i) / Σw_i,
|
|
权重 6 位定点落表达式(可复现、可 diff).
|
|
|
|
与 v1.1 关系: composite_quant12/fund6/all18 等权式零改动(v1.1 测试锁死);
|
|
v2a 为独立注册名.档案 sources 方向表须与三源表之一(QUANT_SOURCES /
|
|
FUND_NEU_SOURCES / 两者合并18源)逐对一致(硬校验——档案只给权重,方向唯一
|
|
权威在代码源表,防两轨漂移).幂等: 注册名已存在直接跳过(同 _register_all 语义);
|
|
v2b 档案(all18_v2b_v1.json)由数据拟合产出,存在才注册,缺省静默跳过.
|
|
"""
|
|
import json
|
|
from dataclasses import dataclass
|
|
from pathlib import Path
|
|
|
|
from . import composite_library
|
|
from .fundamental_neutralize import FUND_NEU_SOURCES
|
|
from .registry import register_factor, get_factor, _REGISTRY
|
|
|
|
# 默认 L1 档案: 相对本文件定位(与代码同仓同版本,不依赖 cwd)
|
|
_DEFAULT_PROFILE_PATH = Path(__file__).parent / "weight_profiles" / "quant12_icirfit_v1.json"
|
|
_DEFAULT_COMPOSITE_NAME = "composite_quant12_v2a"
|
|
|
|
# v2b(财务源市值中性化)档案: 由后续数据拟合产出,存在才注册(graceful,不 ImportError)
|
|
_V2B_PROFILE_PATH = Path(__file__).parent / "weight_profiles" / "all18_v2b_v1.json"
|
|
_V2B_COMPOSITE_NAME = "composite_all18_v2b"
|
|
|
|
|
|
@dataclass(frozen=True)
|
|
class WeightProfile:
|
|
"""三层权重档案(不可变值对象)."""
|
|
|
|
profile_id: str
|
|
name: str
|
|
method: str
|
|
sources: dict[str, tuple[str, float]] # 源名 → (方向 "+"/"-", L1 权重)
|
|
overrides: tuple[dict, ...] # L2 覆盖项 {source, weight, reason}
|
|
metadata: dict # 档案其余字段(fit_window/baseline 等)
|
|
|
|
|
|
def load_profile(path: str) -> WeightProfile:
|
|
"""读 JSON 档案并校验,违例 raise ValueError(带清晰信息,fail-fast).
|
|
|
|
校验: 权重全 ≥0 且和 >0 / override 源 ∈ sources / override 权重 ≥0 /
|
|
override reason 非空(人工干预必须留痕).
|
|
"""
|
|
with open(path, encoding="utf-8") as f:
|
|
raw = json.load(f)
|
|
for key in ("profile_id", "name", "method", "sources"):
|
|
if key not in raw:
|
|
raise ValueError(f"权重档案缺必填字段 {key!r}: {path}")
|
|
|
|
sources: dict[str, tuple[str, float]] = {}
|
|
for src, info in raw["sources"].items():
|
|
direction, weight = info.get("direction"), info.get("weight")
|
|
if direction not in ("+", "-"):
|
|
raise ValueError(f"源 {src} 方向非法(须 '+'/'-'): {direction!r}")
|
|
if not isinstance(weight, (int, float)) or weight < 0:
|
|
raise ValueError(f"源 {src} 权重非法(须为 ≥0 数值): {weight!r}")
|
|
sources[src] = (direction, float(weight))
|
|
if sum(w for _d, w in sources.values()) <= 0:
|
|
raise ValueError(f"全部源权重和须 > 0: {path}")
|
|
|
|
overrides = []
|
|
for ov in raw.get("overrides", []):
|
|
src, weight, reason = ov.get("source"), ov.get("weight"), ov.get("reason")
|
|
if src not in sources:
|
|
raise ValueError(f"override 引用 sources 之外的源: {src!r}")
|
|
if not isinstance(weight, (int, float)) or weight < 0:
|
|
raise ValueError(f"override[{src}] 权重非法(须为 ≥0 数值): {weight!r}")
|
|
if not isinstance(reason, str) or not reason.strip():
|
|
raise ValueError(f"override[{src}] 缺非空 reason(人工干预必须留痕)")
|
|
overrides.append(dict(ov))
|
|
|
|
metadata = {k: v for k, v in raw.items()
|
|
if k not in ("profile_id", "name", "method", "sources", "overrides")}
|
|
return WeightProfile(raw["profile_id"], raw["name"], raw["method"],
|
|
sources, tuple(overrides), metadata)
|
|
|
|
|
|
def resolve_weights(profile: WeightProfile) -> dict[str, float]:
|
|
"""L2 覆盖层语义: 从 L1 权重出发,逐 override 替换对应源权重,再全组重归一化 sum=1.
|
|
|
|
无 override 时 = L1 原样(L1 档案本身已归一,重除仅消浮点归一误差).
|
|
"""
|
|
weights = {src: w for src, (_d, w) in profile.sources.items()}
|
|
for ov in profile.overrides:
|
|
weights[ov["source"]] = float(ov["weight"])
|
|
total = sum(weights.values())
|
|
return {src: w / total for src, w in weights.items()}
|
|
|
|
|
|
def build_weighted_expression(terms: list[tuple[str, float]]) -> str:
|
|
"""加权和拼接: (w1 * (项1) + ... ) / Σw,权重与和均 6 位定点(可 diff)."""
|
|
sum_w = sum(w for _expr, w in terms)
|
|
return "(" + " + ".join(f"{w:.6f} * ({expr})" for expr, w in terms) \
|
|
+ f") / {sum_w:.6f}"
|
|
|
|
|
|
def register_composite_from_profile(profile: WeightProfile, name: str) -> None:
|
|
"""按档案权重注册合成因子(幂等,名字已注册跳过).
|
|
|
|
档案 sources 方向表须与三源表之一逐对一致(v2b 起): QUANT_SOURCES /
|
|
FUND_NEU_SOURCES / 两者拼接 18 源(名字集合 + direction 逐对相等,不一致
|
|
raise ValueError——先于幂等跳过,漂移不被静默吞掉);项形态按源 category 走
|
|
rank_term/_embed_term(复用 v1.1 两函数,口径不漂移);源因子必须已注册
|
|
(get_factor,缺则 raise).
|
|
"""
|
|
declared = {src: d for src, (d, _w) in profile.sources.items()}
|
|
matched: list[tuple[str, str]] | None = None
|
|
for table in (composite_library.QUANT_SOURCES,
|
|
FUND_NEU_SOURCES,
|
|
composite_library.QUANT_SOURCES + FUND_NEU_SOURCES):
|
|
if declared == dict(table):
|
|
matched = table
|
|
break
|
|
if matched is None:
|
|
raise ValueError(
|
|
"档案源方向表与三源表(QUANT_SOURCES/FUND_NEU_SOURCES/两者合并18源)均不一致"
|
|
"(名字集合+方向须逐对相等): "
|
|
f"档案={sorted(declared.items())}"
|
|
)
|
|
if name in _REGISTRY:
|
|
return
|
|
|
|
weights = resolve_weights(profile)
|
|
terms = []
|
|
for src_name, direction in matched: # 按匹配源表顺序拼装
|
|
src = get_factor(src_name)
|
|
if src is None:
|
|
raise ValueError(f"合成源未注册: {src_name}")
|
|
if src["category"] == "fundamental":
|
|
term = composite_library._embed_term(src["expression"], direction)
|
|
else:
|
|
term = composite_library.rank_term(src["expression"], direction)
|
|
terms.append((term, weights[src_name]))
|
|
register_factor(name, build_weighted_expression(terms), category="composite")
|
|
|
|
|
|
def _register_all() -> None:
|
|
"""幂等注册默认 v2a 合成因子(先幂等重挂 v1.1 源,registry 被清后可独立重建).
|
|
|
|
v2b(all18_v2b)档案存在才注册;不存在静默跳过(档案由后续数据拟合产出,
|
|
文件到位即自动注册,不 ImportError).
|
|
"""
|
|
composite_library._register_all()
|
|
# 幂等重挂 fund_*_neu(reload 场景 registry 被清后,neu 模块不会随之重执行,
|
|
# v2b 档案存在时其源必须可重建——v2b 档案上线前此缺口被 graceful 分支掩盖)
|
|
from .fundamental_neutralize import _register_all as _neu_register_all
|
|
_neu_register_all()
|
|
register_composite_from_profile(load_profile(str(_DEFAULT_PROFILE_PATH)),
|
|
_DEFAULT_COMPOSITE_NAME)
|
|
if _V2B_PROFILE_PATH.exists():
|
|
register_composite_from_profile(load_profile(str(_V2B_PROFILE_PATH)),
|
|
_V2B_COMPOSITE_NAME)
|
|
|
|
|
|
# 模块导入时自动注册(与 composite_library 同模式);失败给可读错误(fail-fast)
|
|
try:
|
|
_register_all()
|
|
except Exception as e:
|
|
raise ImportError(
|
|
f"合成层 v2a 默认权重档案加载/注册失败({_DEFAULT_PROFILE_PATH}): {e}"
|
|
) from e
|