claude_dev
9431a50969
09-24 用户裁决(等权 vs ICIR 辩论, issue #69 / 流水线 spec 8147224 §10):
ICIR v2a=在位生产(带考场履历), 等权 v1=挑战者, 影子 A/B 4 周+数据判。
注册名表扩一项: composite_weighting._register_all 的
register_szneu_variants 加等权基线 composite_quant12(v1.1)——产出挑战者
composite_quant12_szneul, 与在位者 composite_quant12_v2a_szneul 只差
权重(源表/中性化包裹/策略参数全同构,残差算子线性保证数学同构)。
设计档 §7.2 加决议 K 节(裁决链+对照口径)。起跑挂 szneul_reb63 同参
影子属 VPS 生产动作, 等口令(件③④)。
Co-Authored-By: Claude Code <noreply@anthropic.com>
402 lines
19 KiB
Python
402 lines
19 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 copy
|
|
import json
|
|
import os
|
|
import re
|
|
import sys
|
|
from dataclasses import dataclass
|
|
from datetime import date
|
|
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"
|
|
|
|
# ③ size 中性化变体后缀(改造③,§7.3): 对合成总分 cs_neutralize(原式, size)
|
|
_SZNEU_SUFFIX = "_szneu"
|
|
|
|
# ---- ④ L2 覆盖 UI 化(2026-09-14): 档案目录扫描注册 + 建档写路径 ----
|
|
_PROFILE_DIR = Path(__file__).parent / "weight_profiles"
|
|
_EXPLICIT_COMPOSITE_NAMES = { # 已显式注册(专用名),扫描跳过防双名
|
|
_DEFAULT_PROFILE_PATH.name: _DEFAULT_COMPOSITE_NAME,
|
|
_V2B_PROFILE_PATH.name: _V2B_COMPOSITE_NAME,
|
|
}
|
|
_ROLL_MARK = "_roll_" # 滚动年度档案属 composite_rolling 域,不作常量权重注册
|
|
_UI_MARK = "_ui_" # UI 建档命名规约: <基准stem>_ui_<日期>
|
|
_NEW_ID_PATTERN = re.compile(r"[a-z0-9_]{4,64}")
|
|
|
|
|
|
@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 源 / fund6+P1-C 拼接 8 源(09-22 fund8 扩展
|
|
入白名单;名字集合 + direction 逐对相等,不一致
|
|
raise ValueError——先于幂等跳过,漂移不被静默吞掉);项形态按源 category 走
|
|
rank_term/_embed_term(复用 v1.1 两函数,口径不漂移);源因子必须已注册
|
|
(get_factor,缺则 raise).
|
|
|
|
resolved 后权重为 0 的源项跳过不入式(v2c 滚动档案存在 neu 源全 0 的
|
|
制度期,0*w 项纯耗求值、cs_neutralize OLS 白算;全零档案 raise ValueError).
|
|
"""
|
|
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,
|
|
composite_library.FUND_SOURCES + composite_library.FUND_P1C_SOURCES):
|
|
if declared == dict(table):
|
|
matched = table
|
|
break
|
|
if matched is None:
|
|
raise ValueError(
|
|
"档案源方向表与四源表(QUANT_SOURCES/FUND_NEU_SOURCES/两者合并18源/"
|
|
"FUND_SOURCES+FUND_P1C_SOURCES 合并8源)均不一致"
|
|
"(名字集合+方向须逐对相等): "
|
|
f"档案={sorted(declared.items())}"
|
|
)
|
|
if name in _REGISTRY:
|
|
return
|
|
|
|
# 全零档案非法: resolve 归一会除零,先给可读 ValueError(L2 override 也可能压成全 0)
|
|
effective = {src: w for src, (_d, w) in profile.sources.items()}
|
|
for ov in profile.overrides:
|
|
effective[ov["source"]] = float(ov["weight"])
|
|
if sum(effective.values()) <= 0:
|
|
raise ValueError(f"档案全部源有效权重为 0(全零档案非法): {name}")
|
|
|
|
weights = resolve_weights(profile)
|
|
terms = []
|
|
for src_name, direction in matched: # 按匹配源表顺序拼装
|
|
if weights[src_name] <= 0: # 0 权项跳过不入式(见 docstring)
|
|
continue
|
|
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]))
|
|
if not terms:
|
|
raise ValueError(f"档案过滤 0 权后无有效项: {name}")
|
|
register_factor(name, build_weighted_expression(terms), category="composite")
|
|
|
|
|
|
def _register_all() -> None:
|
|
"""幂等注册默认 v2a 合成因子(先幂等重挂 v1.1 源,registry 被清后可独立重建).
|
|
|
|
v2b(all18_v2b)档案存在才注册;不存在静默跳过(档案由后续数据拟合产出,
|
|
文件到位即自动注册,不 ImportError). 末尾目录扫描: ui 建档等新档案
|
|
注册成 composite_<stem>(跳过显式/滚动/坏档案).
|
|
"""
|
|
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)
|
|
# ③ 改造:生产冠军挂 _szneu 变体; 等权基线 composite_quant12(v1.1)同挂
|
|
# (决议 K/issue #69): _szneul 变体=影子 A/B 挑战者,与在位者 v2a_szneul
|
|
# 只差权重——源表/中性化包裹/策略参数全同构
|
|
register_szneu_variants([_DEFAULT_COMPOSITE_NAME, "composite_quant12"])
|
|
register_scanned_profiles()
|
|
|
|
|
|
def composite_name_for(stem: str) -> str:
|
|
"""档案文件名 stem → 合成因子注册名(composite_<stem>)."""
|
|
return f"composite_{stem}"
|
|
|
|
|
|
def register_szneu_variant(name: str, suffix: str, size_expression: str) -> str | None:
|
|
"""③ 市值中性化变体(改造③): 对已注册合成因子挂 <名><suffix>.
|
|
|
|
表达式 = cs_neutralize((原式), size_expression)——残差算子线性,
|
|
中性化加权和 ≡ 各源中性化再加权,一处包裹即数学同构;category=composite
|
|
使评估批自动 join 财务域(share_capital 可用性走 fund_*_neu 已验证链路).
|
|
幂等(已注册跳过);源未注册/未合成类则跳过返 None.
|
|
size 双式(09-15 晨定): 原式 _szneu=close*share_capital(线性,OLS 被
|
|
右尾绑架对主体股修正≈0) / _szneul=log(·)(Barra 同款,真移除秩倾斜).
|
|
"""
|
|
target = name + suffix
|
|
if target in _REGISTRY:
|
|
return target
|
|
base = get_factor(name)
|
|
if base is None or base.get("category") != "composite":
|
|
return None
|
|
register_factor(
|
|
target,
|
|
f"cs_neutralize(({base['expression']}), {size_expression})",
|
|
category="composite")
|
|
return target
|
|
|
|
|
|
def register_szneu_variants(names: list[str]) -> list[str]:
|
|
"""③ 变体批量注册(兼容旧调用形): raw 与 log 双式各挂一版."""
|
|
from .fundamental_neutralize import SIZE_EXPRESSION, SIZE_LOG_EXPRESSION
|
|
registered: list[str] = []
|
|
for name in names:
|
|
for suffix, size_expr in (( _SZNEU_SUFFIX, SIZE_EXPRESSION),
|
|
(_SZNEU_SUFFIX + "l", SIZE_LOG_EXPRESSION)):
|
|
got = register_szneu_variant(name, suffix, size_expr)
|
|
if got is not None:
|
|
registered.append(got)
|
|
return registered
|
|
|
|
|
|
def register_scanned_profiles(profile_dir: "Path | str | None" = None) -> list[str]:
|
|
"""扫描档案目录,把可独立评估的档案注册成常量权重合成因子(返回注册名表).
|
|
|
|
④ L2 覆盖建档即生效的机制底座: UI 落一档,下次模块加载自动可用.
|
|
跳过: 显式注册档案(_EXPLICIT_COMPOSITE_NAMES,防 v2a 双名) / 滚动年度
|
|
档案(_roll_,属 composite_rolling 域) / 加载违例档案(stderr 留痕不炸注册链
|
|
——坏档案不该让整个因子注册 import 失败).
|
|
"""
|
|
d = Path(profile_dir) if profile_dir else _PROFILE_DIR
|
|
registered: list[str] = []
|
|
if not d.is_dir():
|
|
return registered
|
|
for p in sorted(d.glob("*.json")):
|
|
if p.name in _EXPLICIT_COMPOSITE_NAMES or _ROLL_MARK in p.stem:
|
|
continue
|
|
name = composite_name_for(p.stem)
|
|
try:
|
|
register_composite_from_profile(load_profile(str(p)), name)
|
|
except ValueError as e: # 含 json.JSONDecodeError: 坏档不炸链
|
|
print(f"[composite_weighting] 跳过档案 {p.name}: {e}", file=sys.stderr)
|
|
continue
|
|
registered.append(name)
|
|
return registered
|
|
|
|
|
|
def _find_profile_path(profile_key: str) -> Path:
|
|
"""按文件 stem 或档案内部 profile_id 定位档案(兼容连字符 id 寻址)."""
|
|
p = _PROFILE_DIR / f"{profile_key}.json"
|
|
if p.exists():
|
|
return p
|
|
for q in sorted(_PROFILE_DIR.glob("*.json")):
|
|
try:
|
|
if load_profile(str(q)).profile_id == profile_key:
|
|
return q
|
|
except ValueError:
|
|
continue
|
|
raise ValueError(f"找不到基准档案 {profile_key!r} (weight_profiles/*.json)")
|
|
|
|
|
|
def list_weight_profiles(profile_dir: "Path | str | None" = None) -> list[dict]:
|
|
"""枚举档案元信息(L2 覆盖 UI 选基准/列档用; 读目录不注册)."""
|
|
d = Path(profile_dir) if profile_dir else _PROFILE_DIR
|
|
out: list[dict] = []
|
|
if not d.is_dir():
|
|
return out
|
|
for p in sorted(d.glob("*.json")):
|
|
try:
|
|
prof = load_profile(str(p))
|
|
except ValueError as e:
|
|
out.append({"file": p.name, "valid": False, "error": str(e)})
|
|
continue
|
|
name = _EXPLICIT_COMPOSITE_NAMES.get(p.name) or composite_name_for(p.stem)
|
|
out.append({
|
|
"file": p.name,
|
|
"valid": True,
|
|
"profile_id": prof.profile_id,
|
|
"name": prof.name,
|
|
"method": prof.method,
|
|
"sources": len(prof.sources),
|
|
"overrides": len(prof.overrides),
|
|
"created": prof.metadata.get("created"),
|
|
"is_ui": _UI_MARK in p.stem,
|
|
"is_rolling": _ROLL_MARK in p.stem,
|
|
"explicit_name": _EXPLICIT_COMPOSITE_NAMES.get(p.name),
|
|
"composite_name": name,
|
|
"registered": get_factor(name) is not None,
|
|
})
|
|
return out
|
|
|
|
|
|
def get_profile_payload(profile_key: str) -> dict:
|
|
"""档案详情载荷: 原样 json + resolved_weights(sum=1,免前端复算口径)."""
|
|
path = _find_profile_path(profile_key)
|
|
with open(path, encoding="utf-8") as f:
|
|
raw = json.load(f)
|
|
raw["resolved_weights"] = resolve_weights(load_profile(str(path)))
|
|
return raw
|
|
|
|
|
|
def create_override_profile(base_profile: str, profile_id: str,
|
|
overrides: list[dict], reason: str) -> dict:
|
|
"""L2 覆盖建档(④ UI 唯一写路径): 基准 L1 原样拷贝 + overrides + ui 溯源块.
|
|
|
|
铁律: 只写新文件(重名即拒,绝不覆盖/触碰既有档案——生产冠军定义不动);
|
|
命名规约 <基准stem>_ui_<日期>; 总说明与每项 reason 必填(人工干预留痕);
|
|
落档原子(tmp+os.replace,并发双击不产半档)后重载复读校验并立即注册
|
|
(批量评估免重启可用). 档案存 L1 原值+overrides(非 resolved 值)——
|
|
存 resolved 会双重应用,重放 resolve 不幂等(test 钉死).
|
|
"""
|
|
if not _NEW_ID_PATTERN.fullmatch(profile_id):
|
|
raise ValueError(f"档案名非法(小写字母/数字/下划线 4-64 位): {profile_id!r}")
|
|
if _UI_MARK not in profile_id:
|
|
raise ValueError(f"命名规约: <基准档案名>_ui_<日期>(须含 {_UI_MARK}): {profile_id!r}")
|
|
if not isinstance(reason, str) or not reason.strip():
|
|
raise ValueError("总调整说明 reason 必填(人工干预留痕铁律)")
|
|
if not overrides:
|
|
raise ValueError("至少一项覆盖(无覆盖直接用基准档案即可)")
|
|
|
|
base_path = _find_profile_path(base_profile)
|
|
prefix = base_path.stem + _UI_MARK
|
|
if not profile_id.startswith(prefix):
|
|
raise ValueError(f"命名规约: 须以基准档案名开头 {prefix!r}: {profile_id!r}")
|
|
target = _PROFILE_DIR / f"{profile_id}.json"
|
|
if target.exists():
|
|
raise ValueError(f"档案已存在(只写新文件,不覆盖): {target.name}")
|
|
|
|
with open(base_path, encoding="utf-8") as f:
|
|
base_raw = json.load(f)
|
|
base = load_profile(str(base_path)) # 复用 L1 全部校验
|
|
|
|
clean = []
|
|
for ov in overrides:
|
|
src, w, r = ov.get("source"), ov.get("weight"), ov.get("reason")
|
|
if src not in base.sources:
|
|
raise ValueError(f"覆盖引用基准 sources 之外的源: {src!r}")
|
|
if not isinstance(w, (int, float)) or w < 0:
|
|
raise ValueError(f"覆盖[{src}] 权重非法(须为 ≥0 数值): {w!r}")
|
|
if not isinstance(r, str) or not r.strip():
|
|
raise ValueError(f"覆盖[{src}] 缺非空 reason(人工干预必须留痕)")
|
|
clean.append({"source": src, "weight": float(w), "reason": r.strip()})
|
|
effective = {s: w for s, (_d, w) in base.sources.items()}
|
|
for ov in clean:
|
|
effective[ov["source"]] = ov["weight"]
|
|
if sum(effective.values()) <= 0:
|
|
raise ValueError("全部源有效权重为 0(全零档案非法)")
|
|
|
|
doc = copy.deepcopy(base_raw) # L1 原样(含 fit_window/constraints/baseline)
|
|
doc["profile_id"] = profile_id
|
|
doc["name"] = f"{base_raw.get('name', base_path.stem)} · UI覆盖({len(clean)}项)"
|
|
doc["overrides"] = clean
|
|
doc["created"] = date.today().isoformat()
|
|
doc["ui"] = {"base_profile": base_path.stem, "reason": reason.strip(), "actor": "web"}
|
|
|
|
tmp = target.with_suffix(".json.tmp")
|
|
tmp.write_text(json.dumps(doc, ensure_ascii=False, indent=2), encoding="utf-8")
|
|
os.replace(tmp, target)
|
|
|
|
prof = load_profile(str(target)) # 落档重载 = 复读校验(写坏即炸,不留半档)
|
|
name = composite_name_for(profile_id)
|
|
register_composite_from_profile(prof, name)
|
|
return {"profile_id": profile_id, "composite_name": name,
|
|
"path": str(target), "resolved_weights": resolve_weights(prof)}
|
|
|
|
|
|
# 模块导入时自动注册(与 composite_library 同模式);失败给可读错误(fail-fast)
|
|
try:
|
|
_register_all()
|
|
except Exception as e:
|
|
raise ImportError(
|
|
f"合成层 v2a 默认权重档案加载/注册失败({_DEFAULT_PROFILE_PATH}): {e}"
|
|
) from e
|