feat(factor): 改造③size中性化变体+④分数加权rank^γ——③register_szneu_variants生产冠军挂composite_quant12_v2a_szneu(cs_neutralize((v2a),close*share_capital),残差算子线性⇒包裹式≡各源中性化再加权,引擎端到端测试逐点实证;端到端测试显式register_fast_ops钉生产算子语义——fast下alpha16小截面合成数据每日NaN部分股,5股有效对<3会假绿,20股修复)④FactorTopNConfig.weight_gamma目标市值∝(N+1−rank)^γ缺省0退化等权,透传四处+json测试扩六参 [vps]

docs/factor_research/factor-system-design.md

@@ -142,8 +142,9 @@
 1. **调仓缓冲带（no-trade band）——✅机制已落地（09-14 夜，TDD）**：只换出跌出前 X% 的持仓（如 top200 持有、跌出 top400 才卖），预期砍换手 11x→3-5x，成本拖累降一个量级——**实盘化的第一前提**。
    - 实现：`FactorTopNConfig.band_ratio`（缺省 1.0=硬切，**与原行为逐字节等价**的回归锚）；rebalance 3b 步——持仓截面 rank ≤ top_n×band_ratio 则不卖，跌出带才让位，TopN 照常补买；被过滤链剔出的持仓不在 rank 内不受带保护（照卖，与原行为一致）。透传四处：argparse `--band-ratio` + `_build_strategy`/`run_backtest_json`/结果回显（坑 1 纪律，json 测试扩五参钉死）。矩阵待跑：top200 × band{1.5, 2.0} vs base 对照（同批同快照）。
 2. **基准配置改 top200**：矩阵内各维度全面最优。零代码（top_n 参数现成），跑批矩阵默认即改。
-3. **行业/市值中性化**：治 2024 微盘雪崩失血；四域数据已备（sw_industry 09-14 落位、市值 valuation 域）。
-4. 分数加权替代硬切换（进出缓冲的连续版本）。
+3. **行业/市值中性化——✅size 变体已落地（09-14 夜，TDD）**：治 2024 微盘雪崩失血；四域数据已备（sw_industry 09-14 落位、市值 valuation 域）。
+   - 实现：`composite_weighting.register_szneu_variants`——生产冠军自动挂 `composite_quant12_v2a_szneu` = `cs_neutralize((v2a原式), close*share_capital)`，category=composite（评估批自动 join 财务域，share_capital 走 fund_*_neu 已验证链路）。**残差算子线性 ⇒ 包裹式中性化 ≡ 各源中性化再加权**（引擎端到端测试实证：szneu 值逐点 == v2a 值对 size 逐日 OLS 残差）。待跑：szneu 截面导出+top200 回测 vs 本批 band_base 对照。行业中性化（x=行业哑元组）待 sw_industry 数据验收后同式扩展。
+4. **分数加权替代硬切换——✅参数已落地（09-14 夜，TDD）**：`FactorTopNConfig.weight_gamma`（缺省 0=等权回归锚）：目标市值 ∝ (N+1−rank)^γ（rank∈1..N 为 TopN 内名次），γ>0 头重脚轻——硬切换「进出/留任」二元决策的连续版本。透传四处同 band_ratio（json 测试扩六参钉死）。矩阵待跑：top200 γ=1 vs 等权对照。
 
 ---
 

sanguo_factor/composite_weighting.py

@@ -41,6 +41,9 @@ _DEFAULT_COMPOSITE_NAME = "composite_quant12_v2a"
 _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 = {          # 已显式注册(专用名),扫描跳过防双名
@@ -194,6 +197,7 @@ def _register_all() -> None:
     if _V2B_PROFILE_PATH.exists():
         register_composite_from_profile(load_profile(str(_V2B_PROFILE_PATH)),
                                         _V2B_COMPOSITE_NAME)
+    register_szneu_variants([_DEFAULT_COMPOSITE_NAME])  # ③ 改造:生产冠军挂 _szneu 变体
     register_scanned_profiles()
 
 
@@ -202,6 +206,31 @@ def composite_name_for(stem: str) -> str:
     return f"composite_{stem}"
 
 
+def register_szneu_variants(names: list[str]) -> list[str]:
+    """③ 市值中性化变体(改造③): 对已注册合成因子挂 <名>_szneu.
+
+    表达式 = cs_neutralize((原式), close*share_capital)——残差算子线性,
+    中性化加权和 ≡ 各源中性化再加权,一处包裹即数学同构;category=composite
+    使评估批自动 join 财务域(share_capital 可用性走 fund_*_neu 已验证链路).
+    幂等(已注册跳过);源未注册/未合成类则跳过不炸(返回实际新注册名表).
+    """
+    from .fundamental_neutralize import SIZE_EXPRESSION
+    registered: list[str] = []
+    for name in names:
+        target = name + _SZNEU_SUFFIX
+        if target in _REGISTRY:
+            continue
+        base = get_factor(name)
+        if base is None or base.get("category") != "composite":
+            continue
+        register_factor(
+            target,
+            f"cs_neutralize(({base['expression']}), {SIZE_EXPRESSION})",
+            category="composite")
+        registered.append(target)
+    return registered
+
+
 def register_scanned_profiles(profile_dir: "Path | str | None" = None) -> list[str]:
     """扫描档案目录,把可独立评估的档案注册成常量权重合成因子(返回注册名表).
 

sanguo_portfolio/runner_backtest.py

@@ -86,6 +86,10 @@ def parse_args() -> argparse.Namespace:
         "--band-ratio", type=float, default=1.0,
         help="factor_topn: 调仓缓冲带(改造①),>1 时持仓 rank≤top_n×ratio 不卖;1.0=硬切",
     )
+    p.add_argument(
+        "--weight-gamma", type=float, default=0.0,
+        help="factor_topn: 分数加权(改造④),目标市值∝(N+1−rank)^γ;0=等权",
+    )
     p.add_argument(
         "--provider", default="local", choices=["local", "baostock", "miniqmt", "unified"],
         help="数据 provider:local(parquet,旧) / baostock(Mac 跨平台) / miniqmt(VPS 实盘) / unified(方案A 权威层)",
@@ -216,6 +220,7 @@ def _build_strategy(args: argparse.Namespace, provider: Any) -> Any:
                 top_n=getattr(args, "top_n", 100),
                 rebalance_days=getattr(args, "rebalance_days", 21),
                 band_ratio=getattr(args, "band_ratio", 1.0),
+                weight_gamma=getattr(args, "weight_gamma", 0.0),
             ),
         )
     # TET Phase 2 验证副本(issue #19):取数走 _ex strict 接口,逻辑与原策略同源 copy
@@ -513,6 +518,7 @@ def run_backtest_json(params: Dict[str, Any]) -> Dict[str, Any]:
         top_n=int(params.get("top_n", 100)),
         rebalance_days=int(params.get("rebalance_days", 21)),
         band_ratio=float(params.get("band_ratio", 1.0)),
+        weight_gamma=float(params.get("weight_gamma", 0.0)),
         initial_positions=json.dumps(params["initial_positions"])
         if params.get("initial_positions") else "",
     )
@@ -860,6 +866,7 @@ def main() -> None:
             "top_n": getattr(args, "top_n", 100),
             "rebalance_days": getattr(args, "rebalance_days", 21),
             "band_ratio": getattr(args, "band_ratio", 1.0),
+            "weight_gamma": getattr(args, "weight_gamma", 0.0),
         })
         print(json.dumps(result, ensure_ascii=False, default=str))
     else:

sanguo_portfolio/strategies/factor_topn.py

@@ -10,7 +10,8 @@
 - 过滤链 = 科创北交剔除(实盘约束) + ST + 次新 + 涨跌停/停牌(复用 filters);
 - 调仓 = 每 ``rebalance_days`` 个交易日(5/10/21=周/双周/月),T-1 日截面
   在 T 日 9:30 开盘调仓(截面值由 T-1 收盘及更早 bar 算出,无前视);
-- 持仓 = TopN 等权(order_target_value 等额,与 value_selection 的 buy 同款)。
+- 持仓 = TopN 等权(order_target_value 等额,与 value_selection 的 buy 同款);
+  可选 band_ratio 缓冲带(改造①)/weight_gamma 分数加权(改造④),缺省全退化等权。
 
 方向约定: 因子值大 = 未来收益高(IC 为正的合成因子),TopN 取最大 N 只。
 """
@@ -62,6 +63,7 @@ class FactorTopNConfig:
     top_n: int = 100                    # 持仓只数
     rebalance_days: int = 21            # 每 K 个交易日调仓(5/10/21=周/双周/月)
     band_ratio: float = 1.0             # 缓冲带(改造①):>1 时持仓 rank≤top_n×ratio 不卖;1.0=硬切
+    weight_gamma: float = 0.0           # 分数加权(改造④):目标市值∝(N+1−rank)^γ;0=等权
     benchmark: str = "000300.XSHG"
     new_stock_days: int = 375           # 次新股过滤阈值(与现有策略同款)
 
@@ -149,13 +151,13 @@ class FactorTopNStrategy:
         # 3b) 缓冲带(改造①): band_ratio>1 时,持仓跌出 TopN 但仍在带内
         # (rank ≤ top_n×band_ratio)不卖,跌出带才让位——砍「边缘股反复进出」
         # 的换手;被过滤链剔出的持仓不在 rank 内,不受带保护(照卖).
+        rank_of = {sym: r for r, sym in enumerate(ranked.index, start=1)}
         band_keep: List[str] = []
         if cfg.band_ratio > 1.0:
             band = int(cfg.top_n * cfg.band_ratio)
-            rank_of = {sym: r for r, sym in enumerate(ranked.index, start=1)}
             band_keep = [s for s in positions if rank_of.get(s, band + 1) <= band]
 
-        # 4) 调仓: 卖出不在名单且不在带内 → 等额买入(value_selection 同款)
+        # 4) 调仓: 卖出不在名单且不在带内 → 按权重建仓(value_selection 同款)
         for stock in list(positions.keys()):
             if stock in buy_list or stock in band_keep:
                 continue
@@ -165,11 +167,21 @@ class FactorTopNStrategy:
         total = _total_value(context)
         if total <= 0:
             return
-        per_value = total / len(buy_list)
+        # ④ 分数加权(改造④): γ>0 时目标市值 ∝ (N+1−rank)^γ(头重脚轻连续版,
+        # rank∈1..N 为 TopN 内名次);γ=0 退化等权 total/N(原行为).
+        targets: dict[str, float] = {}
+        if cfg.weight_gamma > 0.0:
+            s_w = {s: (len(buy_list) + 1 - rank_of[s]) ** cfg.weight_gamma
+                   for s in buy_list if s in rank_of}
+            wsum = sum(s_w.values())
+            if wsum > 0 and len(s_w) == len(buy_list):
+                targets = {s: total * w / wsum for s, w in s_w.items()}
+        if not targets:
+            targets = {s: total / len(buy_list) for s in buy_list}
         for stock in buy_list:
             if stock in positions:
                 continue
-            if self.broker.order_target_value(stock, per_value):
+            if self.broker.order_target_value(stock, targets[stock]):
                 positions = _get_positions(context, self.broker)
 
     # =================== 截面读取 ===================

tests/factor/test_composite_weighting.py

@@ -356,3 +356,79 @@ def test_get_profile_payload_returns_raw_and_resolved(profile_dir):
     assert sum(payload["resolved_weights"].values()) == pytest.approx(1.0)
     with pytest.raises(ValueError, match="找不到"):
         composite_weighting.get_profile_payload("ghost")
+
+
+# ==================== ③ size 中性化变体(改造③) ====================
+def test_szneu_variant_registered_for_v2a():
+    """改造③: v2a 注册后自动挂 <名>_szneu——表达式=cs_neutralize((v2a原式), size),
+    category=composite(评估批自动 join share_capital;TopN 选股直接复用同管线)."""
+    from sanguo_factor.fundamental_neutralize import SIZE_EXPRESSION
+    fac = get_factor("composite_quant12_v2a_szneu")
+    assert fac is not None and fac["category"] == "composite"
+    v2a = get_factor("composite_quant12_v2a")
+    assert fac["expression"] == \
+        f"cs_neutralize(({v2a['expression']}), {SIZE_EXPRESSION})"
+
+
+def test_szneu_registration_idempotent():
+    """幂等: 重复 _register_all() 不重复注册不漂移(autouse fixture 已跑多次)."""
+    composite_weighting._register_all()
+    from sanguo_factor.registry import _REGISTRY
+    assert "composite_quant12_v2a_szneu" in _REGISTRY
+    composite_weighting._register_all()
+    fac = get_factor("composite_quant12_v2a_szneu")
+    assert fac is not None
+
+
+def test_szneu_expression_evaluates_end_to_end():
+    """引擎端到端(生产算子语义): szneu 全表达式可求值,且数值 == v2a 值对
+    size 逐日 OLS 残差(残差算子线性 ⇒ 包裹式中性化 ≡ 先合成再残差化实证).
+
+    20 股合成:fast ops 下 alpha16 在小截面合成数据上每日 NaN 掉部分股
+    (真实数据无此退化,生产 parquet 全密),5 股会让有效对 <3(_MIN_SAMPLES)
+    使 cs_neutralize 全空——20 股保足逐日 ≥3 有效对.
+    显式 register_fast_ops(): batch_eval 生产路径必注册,不依赖测试顺序
+    (suite 内其它批评估测试注册过 fast 算子,单跑则慢算子——两种语义
+    NaN 域不同,不钉死会顺序性飘绿飘红).
+    """
+    import numpy as np
+    import pandas as pd
+    from sanguo_factor.fast_ops import register_fast_ops
+    from vnpy.alpha.dataset.utility import calculate_by_expression
+    register_fast_ops()
+
+    rng = np.random.default_rng(42)
+    stocks = [f"s{i:03d}.SSE" for i in range(20)]
+    days = pd.bdate_range("2023-01-02", periods=130)
+    rows, px = [], {s: 8.0 * (i + 1) for i, s in enumerate(stocks)}
+    for day in days:
+        for s in stocks:
+            px[s] = max(px[s] * (1 + rng.normal(0, 0.02)), 0.5)
+            o = px[s] * (1 + rng.normal(0, 0.005))
+            h = max(o, px[s]) * (1 + abs(rng.normal(0, 0.004)))
+            lo = min(o, px[s]) * (1 - abs(rng.normal(0, 0.004)))
+            rows.append({"datetime": day, "vt_symbol": s, "open": o, "high": h,
+                         "low": lo, "close": px[s],
+                         "volume": float(np.exp(rng.normal(10, 0.4))),
+                         "vwap": (h + lo + px[s]) / 3.0,
+                         "share_capital": float(rng.normal(10, 1))})
+    import polars as pl
+    df = pl.DataFrame(rows).with_columns(pl.col("datetime").cast(pl.Datetime("us")))
+
+    def _eval(expr: str):
+        out = calculate_by_expression(df, expr).to_pandas()
+        return out.sort_values(["datetime", "vt_symbol"]).reset_index(drop=True)
+
+    got = _eval(get_factor("composite_quant12_v2a_szneu")["expression"])
+    raw = _eval(get_factor("composite_quant12_v2a")["expression"])
+    base = df.select(["datetime", "vt_symbol", "close", "share_capital"]).to_pandas()
+    m = got.merge(raw, on=["datetime", "vt_symbol"], suffixes=("_n", "_r"))
+    m = m.merge(base, on=["datetime", "vt_symbol"])
+    m = m.dropna(subset=["data_n", "data_r"])
+    assert len(m) > 100  # 130 日 × 20 股的主流日全覆盖
+    m["size"] = m["close"] * m["share_capital"]
+    for _dt, g in m.groupby("datetime"):
+        x, y = g["size"].to_numpy(), g["data_r"].to_numpy()
+        b = ((x - x.mean()) * (y - y.mean())).sum() / ((x - x.mean()) ** 2).sum()
+        resid = y - (y.mean() + b * (x - x.mean()))
+        np.testing.assert_allclose(g["data_n"].to_numpy(), resid, atol=1e-9)

tests/portfolio/test_factor_topn.py

@@ -214,7 +214,7 @@ def test_run_backtest_json_passes_factor_topn_params(monkeypatch):
         captured.update({
             k: getattr(args, k, "<MISSING>")
             for k in ("factor_name", "factor_dir", "top_n",
-                      "rebalance_days", "band_ratio")
+                      "rebalance_days", "band_ratio", "weight_gamma")
         })
         return {}
 
@@ -223,6 +223,21 @@ def test_run_backtest_json_passes_factor_topn_params(monkeypatch):
         "strategy": "factor_topn",
         "factor_name": "composite_test", "factor_dir": "/x/y",
         "top_n": 50, "rebalance_days": 10, "band_ratio": 1.5,
+        "weight_gamma": 1.0,
     })
     assert captured == {"factor_name": "composite_test", "factor_dir": "/x/y",
-                        "top_n": 50, "rebalance_days": 10, "band_ratio": 1.5}
+                        "top_n": 50, "rebalance_days": 10, "band_ratio": 1.5,
+                        "weight_gamma": 1.0}
+
+
+def test_rebalance_weight_gamma_tilts_to_top(factor_dir):
+    """④ 分数加权: γ>0 时目标市值 ∝ (N+1−rank)^γ(N=TopN 只数);
+    γ=1 → rank1/2/3 权重 = 3:2:1 = 0.5/1/3/1/6(头重脚轻连续版,缺省 γ=0 等权
+    由 test_rebalance_topn_and_universe 钉死)."""
+    st, otv = make_strategy(factor_dir, top_n=3, weight_gamma=1.0)
+    st.rebalance(_ctx("2024-10-08"))
+    vals = {c.args[0]: c.args[1] for c in otv.call_args_list}
+    # 09-30 截面 rank: 300002(6)=1, 300001(5)=2, 000002(4)=3
+    assert vals["300002.XSHE"] == pytest.approx(1e6 * 3 / 6)
+    assert vals["300001.XSHE"] == pytest.approx(1e6 * 2 / 6)
+    assert vals["000002.XSHE"] == pytest.approx(1e6 * 1 / 6)