Buckets:
| """cwm3/tests/test_alert_pair_common.py — the shared reporter helpers of the 2026-09-18 owner-question lanes (eval/alert_pair_common.py; pre-registration docs/decisions.md | |
| 2026-09-18T03:34:28Z): (a) pair_summary applies the >= ceil(2n/3) seed rule to per-seed verdicts (identical / WORSE / BETTER / n.s.); (b) timing_block reads a synthetic | |
| alert_timing paired.csv / metrics.csv and produces one summary row per endpoint with the same rule; (c) append_table / append_json are append-only; (d) no forbidden token. | |
| Run: cd cwm3 && python -m pytest tests/test_alert_pair_common.py -q | |
| """ | |
| from __future__ import annotations | |
| import json | |
| import os | |
| import re | |
| import sys | |
| from pathlib import Path | |
| os.environ.setdefault("OMP_NUM_THREADS", "4") | |
| import pandas as pd # noqa: E402 | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| for p in (CWM3 / "eval", CWM3 / "tests"): | |
| sys.path.insert(0, str(p)) | |
| import alert_pair_common as C # noqa: E402 | |
| from test_no_dms_inputs import FORBIDDEN # noqa: E402 | |
| def _pairs_df(verdicts, deltas): | |
| rows = [] | |
| for s, (v, d) in enumerate(zip(verdicts, deltas)): | |
| lo, hi = {"WORSE": (d - 0.01, d - 0.001), "BETTER": (d + 0.001, d + 0.01), "n.s.": (d - 0.05, d + 0.05), "identical": (0.0, 0.0)}[v] | |
| rows.append({"pair": "a − b", "a": "a", "b": "b", "label": "L", "prereg": True, "seed": s, "stratum": "pooled", "metric": "auroc", "value_a": 0.7 + d, "value_b": 0.7, "delta": d, | |
| "ci_lo": lo, "ci_hi": hi, "frac_boot_d_gt0": 0.5, "n": 512, "n_pos": 63, "n_dev": 60, "verdict": v, "n_boot": 10}) | |
| return pd.DataFrame(rows) | |
| def test_pair_summary_seed_rule(): | |
| S = C.pair_summary(_pairs_df(["WORSE", "WORSE", "n.s."], [-0.03, -0.02, -0.01])) | |
| assert len(S) == 1 and S.rule.iloc[0].startswith("SIGNIFICANTLY WORSE (2/3)") and abs(S.mean_delta.iloc[0] + 0.02) < 1e-9 | |
| S = C.pair_summary(_pairs_df(["BETTER", "BETTER", "BETTER"], [0.03, 0.02, 0.01])) | |
| assert S.rule.iloc[0].startswith("better (3/3") | |
| S = C.pair_summary(_pairs_df(["identical", "identical", "identical"], [0.0, 0.0, 0.0])) | |
| assert S.rule.iloc[0].startswith("identical") | |
| S = C.pair_summary(_pairs_df(["WORSE", "n.s.", "n.s."], [-0.03, 0.0, 0.01])) | |
| assert S.rule.iloc[0].startswith("not worse (1/3 worse") | |
| def test_timing_block(tmp_path): | |
| P, M = [], [] | |
| for s in range(3): | |
| for d in (0.5, 1.0, 2.0): | |
| for reg in ("manual", "engaged", "pooled"): | |
| P.append({"seed": s, "thr_fit": "val", "family": "A", "budget": 20.0, "regime": reg, "delta_s": d, "a": "x", "b": "y", "value_a": 0.05, "value_b": 0.06, "diff": -0.01, "ci_lo": -0.02, "ci_hi": -0.005, | |
| "frac_boot_gt0": 0.0, "n_events": 100}) | |
| for sc in ("x", "y"): | |
| M.append({"seed": s, "thr_fit": "val", "family": "A", "score": sc, "budget": 20.0, "regime": "pooled", "alerts_per_h_test": 19.9, "fa_per_h_test": 18.0, "recall_early_0.5s": 0.06, "recall_early_1s": 0.05, | |
| "recall_early_2s": 0.04, "detected": 0.2, "lead0_mean": 0.3, "n_events": 100}) | |
| pd.DataFrame(P).to_csv(tmp_path / "paired.csv", index=False) | |
| pd.DataFrame(M).to_csv(tmp_path / "metrics.csv", index=False) | |
| Td, TS, TL = C.timing_block(tmp_path, [("x", "y", True)], log=lambda m: None) | |
| assert len(Td) == 27 and len(TS) == 9 and TS.rule.str.startswith("SIGNIFICANTLY WORSE (3/3)").all() | |
| assert len(TL) == 2 and TL.iloc[0].recall_early_05s == [0.06, 0.06, 0.06] | |
| md = C.md_timing_tables(TS, TL, "x − y", ["x", "y"]) | |
| assert any("| 20 | 0.5 | engaged |" in l for l in md) and any(l.startswith("| x | 20 |") for l in md) | |
| def test_append_only(tmp_path): | |
| p, j = tmp_path / "t.md", tmp_path / "t.json" | |
| C.append_table(p, "T", "sec1\n"); C.append_table(p, "T", "sec2\n") | |
| assert p.read_text().startswith("# T\n") and "sec1" in p.read_text() and p.read_text().index("sec2") > p.read_text().index("sec1") | |
| C.append_json(j, {"a": 1}); C.append_json(j, {"b": 2}) | |
| assert json.loads(j.read_text()) == [{"a": 1}, {"b": 2}] | |
| def test_no_forbidden_tokens(): | |
| pat = re.compile("|".join(re.escape(t) for t in FORBIDDEN)) | |
| for f in ("eval/alert_pair_common.py", "eval/report_alert_nocan_ue.py", "eval/report_warning_trunk_heads.py", "tests/test_alert_pair_common.py"): | |
| assert not pat.search((CWM3 / f).read_text()), f | |
| def test_verdict_signed_far(): | |
| """(e) a lower-is-better metric (FAR): Δ < 0 with the CI below 0 reads BETTER, Δ > 0 with the CI above 0 reads WORSE; higher-is-better metrics unchanged.""" | |
| assert C.verdict_signed(-0.065, -0.109, -0.032, lower_better=True) == "BETTER" | |
| assert C.verdict_signed(+0.065, +0.032, +0.109, lower_better=True) == "WORSE" | |
| assert C.verdict_signed(-0.065, -0.109, -0.032, lower_better=False) == "WORSE" | |
| assert C.verdict_signed(-0.01, -0.05, +0.03, lower_better=True) == "n.s." | |
| assert C.verdict_signed(0.0, 0.0, 0.0, lower_better=True) == "identical" | |
Xet Storage Details
- Size:
- 5.04 kB
- Xet hash:
- 49afcf8ca18d7188c6524d2324198f732448f26d3b666bdb759f0719c2c43fdb
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.