From 4eefbfdd74c4385747530acdbdc325e88554e1ba Mon Sep 17 00:00:00 2001 From: iomgaa Date: Tue, 7 Jul 2026 09:38:21 -0400 Subject: [PATCH] =?UTF-8?q?feat(evolution):=20gate.py=20=E2=80=94=20CE-Gat?= =?UTF-8?q?e=20e-process=20=E7=BA=AF=E5=87=BD=E6=95=B0=20(#5=20=E7=AE=97?= =?UTF-8?q?=E6=B3=95=E4=BF=9D=E7=9C=9F)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 从 TRM4 core/harness/eprocess.py 逐行迁移，零逻辑变更: - compute_e_value: 截断 Beta 混合 e 值（log 空间 + betainc 对称性） - gate_decision: 四出口优先级链（confirmed→directional→futility→exhaustion→continue） - probation_verdict: 试用期非对称双向结算 - 常量保真: _WALD_WIN=ln1.4, _WALD_LOSS=ln0.6, _SHRINK_PSEUDO=4 仅变更: import 路径 + GateParams/GateVerdict 移至 types.py + 中文 docstring 14 tests 覆盖: e 值数学、边界校验、四出口路径、试用期三分支 Co-Authored-By: Claude Opus 4.6 (1M context) --- core/evolution/gate.py | 129 ++++++++++++++++++++++++++++++++++++++++ tests/unit/test_gate.py | 96 ++++++++++++++++++++++++++++++ 2 files changed, 225 insertions(+) create mode 100644 core/evolution/gate.py create mode 100644 tests/unit/test_gate.py diff --git a/core/evolution/gate.py b/core/evolution/gate.py new file mode 100644 index 0000000..c8625aa --- /dev/null +++ b/core/evolution/gate.py @@ -0,0 +1,129 @@ +"""CE-Gate 统计核心：截断 Beta 混合 e-process 的纯函数实现。 + +配对不一致检验：候选与基线跑同一题，只数翻转（基线错->候选对 = W； +基线对->候选错 = L）。H0（候选不优）下翻转方向精确五五开， +E = 2^(W+L+1)*B(W+1,L+1)*[1-I_1/2(W+1,L+1)] 为 H0 下非负上鞅， +Ville 不等式给出任意停时 P(E >= 1/alpha) <= alpha。 + +设计规格见 research-wiki/designs/2026-07-03-ce-gate-formal-design.md。 +仅依赖 scipy.special，无 I/O、无状态，便于单测与历史回放复用。 +""" + +from __future__ import annotations + +import math + +from scipy.special import betainc, betaln + +from core.evolution.types import GateParams, GateVerdict + +# Wald 方向游走步长（theta_1=0.70 固定设计常量，不入配置）： +# 胜 +ln(2*theta_1)=ln1.4，负 ln(2*(1-theta_1))=ln0.6。 +_WALD_WIN = math.log(1.4) +_WALD_LOSS = math.log(0.6) + +# delta_shrunk 的伪计数（Agresti-Coull 风格收缩，只作观测输出不进判据）。 +_SHRINK_PSEUDO = 4 + + +def compute_e_value(w: int, l: int) -> float: # noqa: E741 + """截断 Beta 混合 e 值：E = 2^(W+L+1)*B(W+1,L+1)*[1-I_1/2(W+1,L+1)]。 + + 参数: + w: 基线错->候选对的翻转数。 + l: 基线对->候选错的翻转数。 + + 返回: + e 值（W=L=0 时为 1）。 + + 异常: + ValueError: 翻转计数为负时抛出。 + + 关键实现细节: + log 空间计算在 n_max<=40 的设计工作区间内数值稳定（数百级计数 + 亦可）；极大计数（>1000）时最终 exp 仍可能溢出。用正则化不完全 + Beta 的对称性 1-I_1/2(a,b) = I_1/2(b,a) 避免 1-x 的灾难性精度损失。 + """ + if w < 0 or l < 0: + raise ValueError(f"翻转计数不能为负: w={w}, l={l}") + a, b = w + 1, l + 1 + tail = betainc(b, a, 0.5) # = 1 - I_1/2(a, b) + if tail <= 0.0: + return 0.0 + log_e = (w + l + 1) * math.log(2.0) + betaln(a, b) + math.log(tail) + return math.exp(log_e) + + +def gate_decision( + w: int, + l: int, # noqa: E741 + n_used: int, + n_remaining: int, + *, + params: GateParams, +) -> GateVerdict: + """块间四出口判定（每块结束时调用一次）。 + + 出口优先级：CONFIRMED（有证书先走）-> 方向拒绝 -> futility 拒绝 -> + 题尽（provisional / inertia）-> continue。 + + 参数: + w: 累计 W。 + l: 累计 L。 + n_used: 已消费的阶梯题数（含一致题）。 + n_remaining: 阶梯剩余可用题数（min(阶梯长, n_max) - n_used）。 + params: 判据阈值组。 + + 返回: + GateVerdict（decision + e 值/游走/效应量诊断）。 + + 异常: + ValueError: n_used <= 0 或 n_remaining < 0 时抛出。 + """ + if n_used <= 0: + raise ValueError(f"gate_decision 须在至少消费一块后调用: n_used={n_used}") + if n_remaining < 0: + raise ValueError(f"n_remaining 不能为负: {n_remaining}") + e_value = compute_e_value(w, l) + wald = w * _WALD_WIN + l * _WALD_LOSS + delta_hat = (w - l) / n_used + delta_shrunk = (w - l) / (n_used + _SHRINK_PSEUDO) + + if e_value >= params.e_confirm and delta_hat >= params.delta_min: + decision = "accept_confirmed" + elif wald <= params.lambda_dir: + decision = "reject_directional" + elif n_remaining > 0 and compute_e_value(w + n_remaining, l) < params.e_provisional: + # futility 只在题未尽时有意义；题尽后的弱证据归 inertia 出口。 + decision = "reject_futility" + elif n_remaining <= 0: + if ( + e_value >= params.e_provisional + and (w - l) >= params.w_net_min + and delta_hat >= params.delta_min + ): + decision = "accept_provisional" + else: + decision = "reject_inertia" + else: + decision = "continue" + return GateVerdict(decision, e_value, wald, delta_hat, delta_shrunk) + + +def probation_verdict(w: int, l: int, *, params: GateParams) -> str: # noqa: E741 + """试用期一次性结算：固定样本 e 值双向检验。 + + 参数: + w: 结算配对的 W（锚快照错->候选重跑对）。 + l: 结算配对的 L（锚快照对->候选重跑错）。 + params: 判据阈值组（用 e_confirm / e_rollback）。 + + 返回: + "confirmed"（E>=e_confirm 转正）/ "rollback"（对称 E'>=e_rollback 回滚） + / "unverified"（证据不足，elitist 惯性转正）。 + """ + if compute_e_value(w, l) >= params.e_confirm: + return "confirmed" + if compute_e_value(l, w) >= params.e_rollback: + return "rollback" + return "unverified" diff --git a/tests/unit/test_gate.py b/tests/unit/test_gate.py new file mode 100644 index 0000000..5b4a53e --- /dev/null +++ b/tests/unit/test_gate.py @@ -0,0 +1,96 @@ +"""CE-Gate e-process 纯函数单元测试。 + +覆盖 compute_e_value、gate_decision、probation_verdict 三个公共函数 +的核心路径与边界条件。 +""" + +import math + +import pytest + +from core.evolution.gate import compute_e_value, gate_decision, probation_verdict +from core.evolution.types import GateParams, GateVerdict + +_PARAMS = GateParams( + e_confirm=20.0, + e_provisional=3.0, + w_net_min=2, + delta_min=0.02, + lambda_dir=-0.642, + e_rollback=10.0, +) + + +class TestComputeEValue: + """compute_e_value 的数学正确性与边界校验。""" + + def test_zero_zero_returns_one(self) -> None: + """W=L=0 时 e 值应为 1（无证据，中性）。""" + assert compute_e_value(0, 0) == pytest.approx(1.0) + + def test_negative_w_raises(self) -> None: + """负 W 应立即报错。""" + with pytest.raises(ValueError): + compute_e_value(-1, 0) + + def test_negative_l_raises(self) -> None: + """负 L 应立即报错。""" + with pytest.raises(ValueError): + compute_e_value(0, -1) + + def test_heavy_loss_returns_near_zero(self) -> None: + """重度失败时 e 值趋近于零。""" + assert compute_e_value(0, 20) < 0.05 + + def test_heavy_win_returns_large(self) -> None: + """重度胜利时 e 值远大于 100。""" + assert compute_e_value(10, 0) > 100 + + def test_symmetric(self) -> None: + """W>L 时 e 值应大于 W compute_e_value(3, 5) + + +class TestGateDecision: + """gate_decision 四出口优先级链测试。""" + + def test_confirmed_needs_both_e_and_delta(self) -> None: + """高 e 值 + 足够 delta → accept_confirmed。""" + v = gate_decision(10, 0, 10, 10, params=_PARAMS) + assert v.decision == "accept_confirmed" + + def test_continue_on_balanced(self) -> None: + """平衡局面且题未尽 → continue。""" + v = gate_decision(3, 3, 6, 20, params=_PARAMS) + assert v.decision == "continue" + + def test_reject_inertia_on_exhaustion(self) -> None: + """题尽且证据不足 → reject_inertia。""" + v = gate_decision(1, 1, 2, 0, params=_PARAMS) + assert v.decision == "reject_inertia" + + def test_n_used_zero_raises(self) -> None: + """n_used=0 应立即报错。""" + with pytest.raises(ValueError): + gate_decision(0, 0, 0, 10, params=_PARAMS) + + def test_n_remaining_negative_raises(self) -> None: + """n_remaining<0 应立即报错。""" + with pytest.raises(ValueError): + gate_decision(1, 0, 1, -1, params=_PARAMS) + + +class TestProbationVerdict: + """probation_verdict 试用期结算测试。""" + + def test_strong_win_confirmed(self) -> None: + """强烈胜利 → confirmed 转正。""" + assert probation_verdict(10, 0, params=_PARAMS) == "confirmed" + + def test_strong_loss_rollback(self) -> None: + """强烈失败 → rollback 回滚。""" + assert probation_verdict(0, 10, params=_PARAMS) == "rollback" + + def test_balanced_unverified(self) -> None: + """平衡局面 → unverified 惯性转正。""" + assert probation_verdict(3, 3, params=_PARAMS) == "unverified"