Download tests/benchmark_a_prime.py from logos42/headless: direct link, hf CLI and curl.
- Browser
- Download file 11.5 kB
-
https://huggingface.co/logos42/headless/resolve/main/tests/benchmark_a_prime.py
- Command line
-
hf download hf://logos42/headless/tests/benchmark_a_prime.py
-
curl -L -o benchmark_a_prime.py https://huggingface.co/logos42/headless/resolve/main/tests/benchmark_a_prime.py
11.5 kB
| #!/usr/bin/env python3 | |
| # -*- coding: utf-8 -*- | |
| """benchmark_a_prime.py — A′ 四象限可辨识性 + 检测指标(C4/C5)。 | |
| ## 要回答的唯一问题 | |
| **A1 能不能把 S2(新但稳定)与 S3(熟悉但变了)区分开?** | |
| 若 A1(S2) ≈ A1(S3) ⇒ A1 仍然只是 novelty detector(证伪成立) | |
| 若 A1(S3) >> A1(S2) ⇒ A1 真的捕捉到了 dynamics(可辨识) | |
| ## 四象限怎么"构造"出来, 而不是事后切窗 | |
| 每个 transition 发生的那一刻, 有两件**我们知道真值**的事: | |
| novelty : 这一步的 `n_before(s,a)` —— 更新**前**的计数 | |
| dynamics : 这一步是在 `shift_at` 之前还是之后 | |
| 于是四象限是**逐步标注**出来的 2x2: | |
| | | 动力学未变 (t<shift) | 动力学已变 (t>=shift) | | |
| | 熟悉 | S1 n>=FAM | S3 n>=FAM | | |
| | 新 | S2 n< NOV | S4 n< NOV | | |
| **S2 与 S3 是同一张表里的两个格子** —— 用 n 桶把 novelty 对齐之后, | |
| "新"与"变"才可比。这是本文件的全部要点。 | |
| """ | |
| import json | |
| import sys | |
| from pathlib import Path | |
| import numpy as np | |
| sys.path.insert(0, str(Path(__file__).resolve().parents[1])) # 仓库根 —— | |
| # ★ 必须是**仓库根**(`parents[1]`), 不是 `hibs_lnn/` 本身。插错成 | |
| # `.../headless/hibs_lnn` 会让 `import hibs_lnn.rr_agent` 找不到包 | |
| # (`ModuleNotFoundError: No module named 'hibs_lnn'`), 因为那正是包**内部**。 | |
| # 脚本目录会自动进 sys.path, 所以 `from benchmark_three_state import ...` | |
| # 仍然可用。 | |
| from hibs_lnn.rr_agent import RRSkillAgent # noqa: E402 | |
| from hibs_lnn.stochastic_keydoor import ( # noqa: E402 | |
| NonStationaryKeyDoor, StochasticKeyDoor) | |
| from benchmark_three_state import FROZEN # noqa: E402 | |
| NOV, FAM = 3, 10 # 新: n_before < 3 熟悉: n_before >= 10 | |
| CHANNELS = ("N", "A1", "A2", "A3") | |
| class _LogRRSkillAgent(RRSkillAgent): | |
| """逐步记录 (t, s, a, n_before, 各通道读数)。 | |
| `n_before` 必须是**更新前**的计数 —— 它就是那一步的 novelty 真值。 | |
| """ | |
| def __init__(self, *a, **kw): | |
| super().__init__(*a, **kw) | |
| self.rows = [] | |
| def _rr_dval(self, s, a, s2): # noqa: D102 | |
| d = self._rr_T.get((int(s), int(a))) | |
| n_before = sum(d.values()) if d else 0 | |
| v = super()._rr_dval(s, a, s2) | |
| self.rows.append({ | |
| "t": int(getattr(self.mdp, "t_total", len(self.rows))), | |
| "s": int(s), "a": int(a), "n": int(n_before), | |
| "v": float(v), | |
| }) | |
| return v | |
| def log_channel(channel, env, episodes=140, seed=7): | |
| """在一个环境上跑一遍, 返回该通道的逐步日志。""" | |
| ag = _LogRRSkillAgent(env, mode="rr", seed=seed, rr_v_main="learned", | |
| **dict(FROZEN, rr_three_state=True, rr_dyn=channel)) | |
| for _ in range(episodes): | |
| ag.run_episode() | |
| return ag.rows | |
| def quad_table(rows, shift_at): | |
| """把逐步日志切成四象限, 返回 {(cell, ch): mean} 的统计。""" | |
| out = {} | |
| for r in rows: | |
| cell = ("S1" if r["n"] >= FAM else | |
| "S2" if r["n"] < NOV else None) | |
| if cell is None: | |
| continue | |
| if r["t"] >= shift_at: | |
| cell = {"S1": "S3", "S2": "S4"}[cell] | |
| out.setdefault(cell, []).append(r["v"]) | |
| return {c: (float(np.mean(v)), len(v)) for c, v in out.items()} | |
| # ── 指标(A′-4)────────────────────────────────────────────────────── | |
| def detection_metrics(rows, shift_at, thresh=None, win=100): | |
| """在 shift 附近算 T_detect / FP / FN / AUC / discrimination。 | |
| * 触发 = 读数超过 `thresh`。默认 `pre_mean + 2·pre_std`。 | |
| ★ 阈值只能从"变之前"的数据里定 —— 用全段分位数定阈值等于偷看答案。 | |
| ★ **不能用 pre 的 95 分位**:`_rr_dval` 在 `n<2` 时返回满值 1.0, | |
| 热身期到处都是这种步 ⇒ 95 分位 = 1.0000 ⇒ 没有任何步能超过它 | |
| ⇒ `T_detect = None`、FN 恒 1(第一版的实测就是这个指纹: | |
| 四个通道的阈值全是 1.0000、T_detect 全是 None)。均值+2σ 不会被 | |
| 饱和步锁死在 1.0。 | |
| * `T_detect − T_shift` = 检测延迟(核心指标)。 | |
| * AUC = 把"shift 后的步"与"shift 前的步"分开的可分性(Mann-Whitney)。 | |
| """ | |
| pre = np.array([r["v"] for r in rows if r["t"] < shift_at], float) | |
| post = np.array([r["v"] for r in rows if r["t"] >= shift_at], float) | |
| if pre.size == 0 or post.size == 0: | |
| return None | |
| if thresh is None: | |
| # ★ 阈值估计必须**排除饱和步**(n<2 时 `_rr_dval` 返回满值 1.0)。 | |
| # 否则 μ+2σ 会被那些 1.0 撑到 **超过指标上限**(实测 A1 的阈值 | |
| # 算成 1.0680 > 1.0) ⇒ 永远没有步能超过它 ⇒ T_detect=None、 | |
| # FN 恒 1。那不是"检测不到", 是**阈值不可达**。 | |
| # A 类读数恒在 [0,1], 所以阈值必须落在 [0,1] 内才有意义。 | |
| pre_ok = np.array([r["v"] for r in rows | |
| if r["t"] < shift_at and r["n"] >= 2], float) | |
| base = pre_ok if pre_ok.size >= 10 else pre | |
| thr = float(base.mean() + 2.0 * base.std()) | |
| thr = float(min(max(thr, 1e-6), 1.0)) | |
| else: | |
| thr = float(thresh) | |
| # 检测延迟: shift 后第一个超过阈值、且随后 win 步里多数都超的时刻 | |
| t_det = None | |
| for i, r in enumerate(rows): | |
| if r["t"] < shift_at or r["v"] <= thr: | |
| continue | |
| nxt = [q["v"] for q in rows[i:i + win]] | |
| if sum(v > thr for v in nxt) > 0.5 * len(nxt): | |
| t_det = r["t"] | |
| break | |
| fp = float(np.mean(pre > thr)) if pre.size else float("nan") | |
| fn = 1.0 if t_det is None else 0.0 | |
| # AUC: P(读数(shift后) > 读数(shift前)) —— Mann-Whitney 形式 | |
| if pre.size and post.size: | |
| gt = (post[:, None] > pre[None, :]).mean() | |
| eq = (post[:, None] == pre[None, :]).mean() | |
| auc = float(gt + 0.5 * eq) | |
| else: | |
| auc = float("nan") | |
| return {"thr": thr, "t_det": t_det, | |
| "delay": (None if t_det is None else t_det - shift_at), | |
| "fp": fp, "fn": fn, "auc": auc, | |
| "pre_mean": float(pre.mean()), "post_mean": float(post.mean())} | |
| def main(): | |
| SH, EP = 400, 140 | |
| print("=" * 104) | |
| print("A′ 四象限可辨识性(C4):**A1 能不能把 S2(新但稳定)与 S3(熟悉但变了)分开?**") | |
| print(f" 熟悉 = n_before >= {FAM} 新 = n_before < {NOV} shift @ t={SH}") | |
| print("=" * 104) | |
| results = {} | |
| # 环境 A: 无 shift(所有格子都在"动力学未变"那一列) | |
| envs = { | |
| "E1 平稳随机(无 shift)": lambda: StochasticKeyDoor(slip=0.10, seed=7), | |
| "E2 非平稳(shiftt@400)": lambda: NonStationaryKeyDoor( | |
| slip0=0.10, slip1=0.60, shift_at=SH, seed=7), | |
| } | |
| for nm, mk in envs.items(): | |
| print(f"\n── {nm} " + "─" * (96 - len(nm))) | |
| print(f" {'通道':>4} {'S1熟/静':>10} {'S2新/静':>10} {'S3熟/变':>10} " | |
| f"{'S4新/变':>10} {'S3−S2':>9} {'S3/S2':>7} {'N(S2)':>7}") | |
| for ch in CHANNELS: | |
| rows = log_channel(ch, mk(), EP, 7) | |
| q = quad_table(rows, SH) | |
| g = lambda c: q.get(c, (float("nan"), 0))[0] # noqa: E731 | |
| s2, s3 = g("S2"), g("S3") | |
| disc = s3 - s2 | |
| ratio = (s3 / s2) if s2 and s2 > 1e-12 else float("inf") | |
| results.setdefault(nm, {})[ch] = { | |
| "S1": g("S1"), "S2": s2, "S3": s3, "S4": g("S4"), | |
| "disc": disc, "n_S2": q.get("S2", (0, 0))[1], | |
| "n_S3": q.get("S3", (0, 0))[1], | |
| } | |
| print(f" {ch:>4} {g('S1'):>10.4f} {s2:>10.4f} {s3:>10.4f} " | |
| f"{g('S4'):>10.4f} {disc:>+9.4f} {ratio:>7.2f} " | |
| f"{q.get('S2',(0,0))[1]:>7}") | |
| print() | |
| print("=" * 104) | |
| print("A′ 检测指标(C5): T_detect/T_shift · FP · FN · AUC(阈值只从 shift **前** 定)") | |
| print("=" * 104) | |
| print(f" {'通道':>4} {'阈值':>8} {'T_detect':>9} {'延迟':>6} {'FP':>7} " | |
| f"{'FN':>4} {'AUC':>7} {'变前均值':>10} {'变后均值':>10}") | |
| met = {} | |
| for ch in CHANNELS: | |
| rows = log_channel(ch, NonStationaryKeyDoor( | |
| slip0=0.10, slip1=0.60, shift_at=SH, seed=7), EP, 7) | |
| m = detection_metrics(rows, SH) | |
| met[ch] = m | |
| if m is None: | |
| print(f" {ch:>4} (样本不足)") | |
| continue | |
| print(f" {ch:>4} {m['thr']:>8.4f} {str(m['t_det']):>9} " | |
| f"{str(m['delay']):>6} {m['fp']:>7.2f} {m['fn']:>4.0f} " | |
| f"{m['auc']:>7.3f} {m['pre_mean']:>10.4f} {m['post_mean']:>10.4f}") | |
| # ── 判定 ── | |
| print() | |
| print("=" * 104) | |
| print("判定") | |
| print("=" * 104) | |
| ok = True | |
| e2 = results["E2 非平稳(shiftt@400)"] | |
| # ★ 主判据 = **n 对齐的 S1 vs S3**: 同为"熟悉"状态, 只有动力学变了。 | |
| # 不能用 S2 vs S3 —— S2 是首次访问步, 而 `_rr_dval` 在 `n<2` 时返回 | |
| # 满值 1.0, 于是 S2 在所有通道上都被顶到 0.7~0.8, 必然压过 S3。 | |
| # 那是**饱和分支**的指纹, 不是"A 类分不开新与变"的证据。 | |
| d1 = e2["A1"]["S3"] - e2["A1"]["S1"] | |
| dN = e2["N"]["S3"] - e2["N"]["S1"] | |
| a1_auc = met.get("A1", {}).get("auc", float("nan")) | |
| n_auc = met.get("N", {}).get("auc", float("nan")) | |
| print(" 【主判据】n 对齐(同为熟悉状态)的 S1 -> S3:") | |
| print(f" A1: {e2['A1']['S1']:.4f} -> {e2['A1']['S3']:.4f} ({d1:+.4f})") | |
| print(f" N : {e2['N']['S1']:.4f} -> {e2['N']['S3']:.4f} ({dN:+.4f})") | |
| print(f" A2: {e2['A2']['S1']:.4f} -> {e2['A2']['S3']:.4f} " | |
| f"({e2['A2']['S3'] - e2['A2']['S1']:+.4f})") | |
| print(f" AUC: A1 {a1_auc:.3f} vs N {n_auc:.3f}") | |
| print() | |
| print(" 【诊断, 非判据】原始 S2 vs S3(被 `n<2 -> 1.0` 饱和分支污染):") | |
| print(f" S2 各通道 {[round(e2[c]['S2'], 3) for c in CHANNELS]}") | |
| print(f" S3 各通道 {[round(e2[c]['S3'], 3) for c in CHANNELS]}") | |
| print(" ⇒ S2 全面高于 S3。这是「首次访问报满分」造成的, 不代表" | |
| "A 类分不开新与变。") | |
| print() | |
| checks = [ | |
| (f"A1 在熟悉状态上对'变'敏感 (S3−S1 > +0.10) [{d1:+.4f}]", d1 > 0.10), | |
| (f"N 在同样一对上保持平坦 (|S3−S1| < 0.05) [{dN:+.4f}]", | |
| abs(dN) < 0.05), | |
| (f"A1 的响应明显强于纯新颖度 (dA1 > dN + 0.10) " | |
| f"[{d1:+.4f} vs {dN:+.4f}]", d1 - dN > 0.10), | |
| (f"A1 的 AUC 明显高于 N 的 AUC (>0.10) " | |
| f"[{a1_auc:.3f} vs {n_auc:.3f}]", a1_auc - n_auc > 0.10), | |
| ] | |
| for name, good in checks: | |
| print(f" {'PASS' if good else 'FAIL'} {name}") | |
| ok = ok and good | |
| print(f"\n {'A 类在本环境上具有可辨识性' if ok else 'A 类仍不可辨识 —— 记录为 coverage-only'}") | |
| with open("results/a_prime.json", "w") as f: | |
| json.dump({"quadrants": results, "metrics": met, | |
| "shift_at": SH, "NOV": NOV, "FAM": FAM}, f, | |
| indent=2, ensure_ascii=False) | |
| print(" -> results/a_prime.json") | |
| return 0 if ok else 1 | |
| if __name__ == "__main__": | |
| sys.exit(main()) | |