Buckets:

AsianPlayer
/

TriDrive

Files

xet

AsianPlayer/TriDrive / code /eval /alert_pair_common.py

AsianPlayer

16 days ago

download

raw

13.7 kB

"""cwm3/eval/alert_pair_common.py — shared helpers of the 2026-09-18 owner-question reporters (eval/report_alert_nocan_ue.py, eval/report_warning_trunk_heads.py;
pre-registration docs/decisions.md 2026-09-18T03:34:28Z).  CPU only; read-only on every run dir; imports the helpers of eval/warning_ecan_vs_original.py unchanged.
  gold_block(seeds, specs, pairs, n_boot)  -> (levels, pairs, identity) DataFrames: per seed / stratum (pooled / manual / engaged) clean-test AUROC and PR-AUC of every
      spec {key: (W layout pattern with {seed}, column)} read at the gold rows (warning_gold.build_gold legacy_first via calibration_report.gold_for_seed), and the paired
      Δ of every (a, b) pair on IDENTICAL device-cluster resamples (warning_gold.cluster_boot, n_boot, seed 0); per-seed verdict identical / WORSE / BETTER / n.s. and the
      pre-registered >= ceil(2n/3) seed rule (warning_ecan_vs_original.verdict1 / seed_rule); layout identity = max |Δ| over every test / val row.
  timing_block(tdir, pairs) -> per-endpoint rows (paired.csv of an alert_timing_eval.py run) + the seed-rule summary + alert-load levels (metrics.csv).
  find_timing_dir(mark) -> newest results/alert_timing/<stamp> whose run.json mentions `mark`.
No openpilot DMS field is read.
"""
from __future__ import annotations

import json
import sys
from pathlib import Path

import numpy as np
import pandas as pd

CWM3 = Path(__file__).resolve().parents[1]
sys.path.insert(0, str(CWM3 / "eval"))
import warning_ecan_vs_original as EV                       # noqa: E402  helpers only (main() never called)

CR, OP, WG, LB = EV.CR, EV.OP, EV.WG, EV.LB
f3, ci_s, verdict1, seed_rule, load_col, at_gold, strata = EV.f3, EV.ci_s, EV.verdict1, EV.seed_rule, EV.load_col, EV.at_gold, EV.strata
REGIMES = ("manual", "engaged")
RUNS = CWM3 / "results" / "runs"
AT_ROOT = CWM3 / "results" / "alert_timing"
STRATA = ("pooled", "manual", "engaged")


def verdict_signed(d, lo, hi, lower_better=False):
    """per-seed verdict of a paired Δ = a − b; lower_better=True (e.g. a false-alarm rate) flips the sign so that a NEGATIVE Δ with its CI below 0 reads BETTER."""
    if lower_better:
        return verdict1(-d if np.isfinite(d) else d, -hi if np.isfinite(hi) else hi, -lo if np.isfinite(lo) else lo)
    return verdict1(d, lo, hi)


def find_timing_dir(mark, explicit=""):
    if explicit:
        return Path(explicit)
    cands = []
    for rj in sorted(AT_ROOT.glob("*/run.json")):
        try:
            if mark in rj.read_text():
                cands.append(rj.parent)
        except OSError:
            continue
    return cands[-1] if cands else None


def layouts_present(pattern, seed):
    w = pattern.format(seed=seed)
    return all((RUNS / w / f"pred_layout_{r}_{s}.npz").exists() for r in REGIMES for s in ("test", "val"))


def gold_block(seeds, specs, pairs, n_boot=2000, log=print):
    """specs: {key: (pattern, column)}; pairs: [(a_key, b_key, label, prereg)] -> (levels df, pairs df, identity dict)."""
    lab = LB.episode_labels()
    lab = lab[lab.label != "unsure"].reset_index(drop=True)
    packs = {(r, s): WG.load_pack(r, s) for r in REGIMES for s in ("test", "val")}
    levels, prs, ident = [], [], {}
    for seed in seeds:
        G = CR.gold_for_seed(lab, packs, seed)
        yg, dong = G.y.to_numpy().astype(int), G.dongle.to_numpy()
        cols, gsc = {}, {}
        for key, (pat, col) in specs.items():
            if not layouts_present(pat, seed):
                log(f"  seed {seed}: layouts of {key} ({pat.format(seed=seed)}) missing -> skipped")
                continue
            cols[key] = load_col(pat.format(seed=seed), col, packs)
            gsc[key] = at_gold(G, cols[key])
        for a, b, label, prereg in pairs:
            if a in cols and b in cols:
                maxabs, nan_mis, n_rows = 0.0, 0, 0
                for k in cols[a]:
                    both = np.isfinite(cols[a][k]) & np.isfinite(cols[b][k])
                    nan_mis += int((np.isfinite(cols[a][k]) != np.isfinite(cols[b][k])).sum())
                    n_rows += len(cols[a][k])
                    if both.any():
                        maxabs = max(maxabs, float(np.abs(cols[a][k][both] - cols[b][k][both]).max()))
                ident[(a, b, seed)] = {"identical": maxabs == 0.0 and nan_mis == 0, "max_abs_diff": maxabs, "nan_mismatch_rows": nan_mis, "n_rows": n_rows}
        for st, m0 in strata(G).items():
            fin = m0.copy()
            for key in gsc:
                fin &= np.isfinite(gsc[key])
            y, dg = yg[fin], dong[fin]
            if len(y) == 0 or y.min() == y.max():
                continue
            S = {key: gsc[key][fin] for key in gsc}
            bs = WG.cluster_boot(y, S, dg, n_boot, 0)
            for key in gsc:
                for met, fn, kb in (("auroc", WG.fast_auroc, "auroc"), ("pr_auc", WG.fast_ap, "prauc")):
                    lo, hi = WG.pct(bs[key][kb])
                    levels.append({"key": key, "seed": seed, "stratum": st, "metric": met, "value": float(fn(y, S[key])), "ci_lo": lo, "ci_hi": hi, "n": int(fin.sum()), "n_pos": int(y.sum()),
                                   "n_dev": int(len(np.unique(dg))), "n_boot": n_boot})
            for a, b, label, prereg in pairs:
                if a not in gsc or b not in gsc:
                    continue
                for met, fn, kb in (("auroc", WG.fast_auroc, "auroc"), ("pr_auc", WG.fast_ap, "prauc")):
                    va, vb = float(fn(y, S[a])), float(fn(y, S[b]))
                    d = bs[a][kb] - bs[b][kb]
                    lo, hi = WG.pct(d)
                    v = verdict1(va - vb, lo, hi)
                    prs.append({"pair": f"{a} − {b}", "a": a, "b": b, "label": label, "prereg": prereg, "seed": seed, "stratum": st, "metric": met, "value_a": va, "value_b": vb, "delta": va - vb,
                                "ci_lo": lo, "ci_hi": hi, "frac_boot_d_gt0": float(np.mean(d[np.isfinite(d)] > 0)) if np.isfinite(d).any() else np.nan, "n": int(fin.sum()), "n_pos": int(y.sum()),
                                "n_dev": int(len(np.unique(dg))), "verdict": v, "n_boot": n_boot})
                    log(f"  s{seed} {st:8s} {met:6s} {a} − {b}: {va:.4f} − {vb:.4f} = {va - vb:+.4f} {ci_s(lo, hi, 4)} -> {v}")
    return pd.DataFrame(levels), pd.DataFrame(prs), ident


def pair_summary(P):
    """per (pair, stratum, metric): the seed rule over the per-seed verdicts."""
    rows = []
    if not len(P):
        return pd.DataFrame(rows)
    for (pair, st, met), g in P.groupby(["pair", "stratum", "metric"], sort=False):
        g = g.sort_values("seed")
        rule, nw, nb, n = seed_rule(g.verdict.tolist())
        rows.append({"pair": pair, "a": g.a.iloc[0], "b": g.b.iloc[0], "label": g.label.iloc[0], "prereg": bool(g.prereg.iloc[0]), "stratum": st, "metric": met, "seeds": n, "n_worse": nw, "n_better": nb,
                     "rule": rule, "mean_delta": float(g.delta.mean()), "mean_a": float(g.value_a.mean()), "mean_b": float(g.value_b.mean()), "n": int(g.n.iloc[0]), "n_pos": int(g.n_pos.iloc[0]),
                     "deltas": g.delta.round(4).tolist(), "ci": [[round(x, 4), round(y, 4)] for x, y in zip(g.ci_lo, g.ci_hi)]})
    return pd.DataFrame(rows)


def timing_block(tdir, pairs, log=print):
    """pairs: [(a, b, prereg)] -> (rows per endpoint / seed, summary per endpoint, levels)."""
    P, M = pd.read_csv(tdir / "paired.csv"), pd.read_csv(tdir / "metrics.csv")
    rows, summ, levels = [], [], []
    for na, nb, prereg in pairs:
        sub = P[(P.a == na) & (P.b == nb)]
        if not len(sub):
            log(f"  timing pair {na} − {nb}: no rows")
            continue
        for (fit, fam, B, dl, reg), g in sub.groupby(["thr_fit", "family", "budget", "delta_s", "regime"]):
            g = g.sort_values("seed")
            vs = [verdict1(r.diff, r.ci_lo, r.ci_hi) for r in g.itertuples()]
            for r, v in zip(g.itertuples(), vs):
                rows.append({"pair": f"{na} − {nb}", "a": na, "b": nb, "prereg": prereg, "thr_fit": fit, "family": fam, "budget": B, "delta_s": dl, "regime": reg, "seed": int(r.seed), "value_a": r.value_a,
                             "value_b": r.value_b, "delta": r.diff, "ci_lo": r.ci_lo, "ci_hi": r.ci_hi, "n_events": int(r.n_events), "verdict": v})
            rule, nw, nbet, n = seed_rule(vs)
            summ.append({"pair": f"{na} − {nb}", "a": na, "b": nb, "prereg": prereg, "thr_fit": fit, "family": fam, "budget": B, "delta_s": dl, "regime": reg, "seeds": n, "n_worse": nw, "n_better": nbet,
                         "rule": rule, "mean_delta": float(g["diff"].mean()), "mean_a": float(g.value_a.mean()), "mean_b": float(g.value_b.mean()), "n_events": int(g.n_events.iloc[0]),
                         "deltas": g["diff"].round(4).tolist()})
    names = sorted({x for a, b, _ in pairs for x in (a, b)})
    for sc in names:
        for (fit, fam, B, reg), g in M[(M.score == sc)].groupby(["thr_fit", "family", "budget", "regime"]):
            if "|" in str(reg):
                continue
            g = g.sort_values("seed")
            levels.append({"score": sc, "thr_fit": fit, "family": fam, "budget": B, "regime": reg, "alerts_per_h_test": g.alerts_per_h_test.round(2).tolist(), "fa_per_h_test": g.fa_per_h_test.round(2).tolist(),
                           "recall_early_05s": g["recall_early_0.5s"].round(4).tolist(), "recall_early_1s": g.recall_early_1s.round(4).tolist(), "recall_early_2s": g.recall_early_2s.round(4).tolist(),
                           "detected": g.detected.round(4).tolist(), "lead0_mean": g.lead0_mean.round(3).tolist(), "n_events": int(g.n_events.iloc[0])})
    return pd.DataFrame(rows), pd.DataFrame(summ), pd.DataFrame(levels)


def md_gold_table(L, S, keys_order, labels):
    """levels + pair summary -> markdown lines: per arm x metric x stratum the per-seed values, and per pair the Δ rule."""
    out = ["| arm | metric | stratum | s0 / s1 / s2 (CI of s0) | n / pos |", "|---|---|---|---|---|"]
    for key in keys_order:
        for met in ("auroc", "pr_auc"):
            for st in STRATA:
                g = L[(L.key == key) & (L.metric == met) & (L.stratum == st)].sort_values("seed") if len(L) else L
                if not len(g):
                    continue
                r0 = g.iloc[0]
                out.append(f"| {labels.get(key, key)} | {met} | {st} | {' / '.join(f3(x) for x in g.value)} {ci_s(r0.ci_lo, r0.ci_hi)} | {int(r0.n)} / {int(r0.n_pos)} |")
    out += ["", "| pair (Δ = a − b) | metric | stratum | Δ s0 / s1 / s2 | CIs | mean Δ | rule |", "|---|---|---|---|---|---|---|"]
    for r in S.itertuples():
        rule = r.rule if r.prereg else "descriptive: " + r.rule
        out.append(f"| {r.label} | {r.metric} | {r.stratum} | {' / '.join(f3(x, 3, True) for x in r.deltas)} | {'; '.join(ci_s(x, y) for x, y in r.ci)} | {f3(r.mean_delta, 3, True)} | {rule} |")
    return out


def md_timing_tables(TS, TL, primary_pair, level_scores):
    out = []
    if not len(TS):
        return ["timing run not available"]
    A = TS[(TS.pair == primary_pair) & (TS.thr_fit == "val") & (TS.family == "A")]
    if len(A):
        out += [f"**Family A, VAL-fitted thresholds (PRIMARY), pair `{primary_pair}` — every endpoint:**", "", "| K/h | δ s | regime | a mean | b mean | Δ s0 / s1 / s2 | mean Δ | events | rule |", "|---|---|---|---|---|---|---|---|---|"]
        for r in A.sort_values(["budget", "delta_s", "regime"]).itertuples():
            out.append(f"| {r.budget:g} | {r.delta_s:g} | {r.regime} | {f3(r.mean_a)} | {f3(r.mean_b)} | {' / '.join(f3(x, 3, True) for x in r.deltas)} | {f3(r.mean_delta, 3, True)} | {r.n_events} | {r.rule} |")
    out += ["", "**All pairs — endpoint counts per (pair, fit, family, K); 9 endpoints each = 3 δ × 3 regimes:**", "",
            "| pair | prereg | thr fit | family | K/h | endpoints | identical | significantly worse | better | n.s. | mean Δ |", "|---|---|---|---|---|---|---|---|---|---|---|"]
    for (pair, prereg, fit, fam, B), g in TS.groupby(["pair", "prereg", "thr_fit", "family", "budget"], sort=False):
        ni = int(g.rule.str.startswith("identical").sum()); nw = int(g.rule.str.startswith("SIGNIFICANTLY").sum()); nb = int(g.rule.str.startswith("better").sum())
        out.append(f"| {pair} | {'yes' if prereg else 'descr.'} | {fit} | {fam} | {B:g} | {len(g)} | {ni} | {nw} | {nb} | {len(g) - ni - nw - nb} | {f3(g.mean_delta.mean(), 4, True)} |")
    if len(TL):
        out += ["", "**Alert load and lead levels at the VAL-fitted thresholds (pooled, family A; per seed s0 / s1 / s2):**", "",
                "| score | K/h | alerts/h test | FA/h | Recall ≥ 0.5 s | Recall ≥ 1 s | Recall ≥ 2 s | detected | lead0 mean s | events |", "|---|---|---|---|---|---|---|---|---|---|"]
        j = lambda xs, nd=3: " / ".join(f3(x, nd) for x in xs)         # noqa: E731
        for r in TL[(TL.thr_fit == "val") & (TL.family == "A") & (TL.regime == "pooled") & (TL.score.isin(level_scores))].sort_values(["budget", "score"]).itertuples():
            out.append(f"| {r.score} | {r.budget:g} | {j(r.alerts_per_h_test, 1)} | {j(r.fa_per_h_test, 1)} | {j(r.recall_early_05s)} | {j(r.recall_early_1s)} | {j(r.recall_early_2s)} | {j(r.detected)} | {j(r.lead0_mean, 2)} | {r.n_events} |")
    return out


def append_table(path, title, md):
    path = Path(path)
    if not path.exists():
        path.write_text(f"# {title}\n\n")
    with open(path, "a") as f:
        f.write("\n" + md)


def append_json(path, rec):
    path = Path(path)
    arr = json.loads(path.read_text()) if path.exists() else []
    arr.append(rec)
    path.write_text(json.dumps(arr, indent=1, default=str))

Xet Storage Details

Size:: 13.7 kB
Xet hash:: 264ca241926846b203621cfb59564b721972d75970d48d98c58ccd446c52a5b6

Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.

	"""cwm3/eval/alert_pair_common.py — shared helpers of the 2026-09-18 owner-question reporters (eval/report_alert_nocan_ue.py, eval/report_warning_trunk_heads.py;
	pre-registration docs/decisions.md 2026-09-18T03:34:28Z). CPU only; read-only on every run dir; imports the helpers of eval/warning_ecan_vs_original.py unchanged.

	gold_block(seeds, specs, pairs, n_boot) -> (levels, pairs, identity) DataFrames: per seed / stratum (pooled / manual / engaged) clean-test AUROC and PR-AUC of every
	spec {key: (W layout pattern with {seed}, column)} read at the gold rows (warning_gold.build_gold legacy_first via calibration_report.gold_for_seed), and the paired
	Δ of every (a, b) pair on IDENTICAL device-cluster resamples (warning_gold.cluster_boot, n_boot, seed 0); per-seed verdict identical / WORSE / BETTER / n.s. and the
	pre-registered >= ceil(2n/3) seed rule (warning_ecan_vs_original.verdict1 / seed_rule); layout identity = max \|Δ\| over every test / val row.
	timing_block(tdir, pairs) -> per-endpoint rows (paired.csv of an alert_timing_eval.py run) + the seed-rule summary + alert-load levels (metrics.csv).
	find_timing_dir(mark) -> newest results/alert_timing/<stamp> whose run.json mentions `mark`.
	No openpilot DMS field is read.
	"""
	from __future__ import annotations

	import json
	import sys
	from pathlib import Path

	import numpy as np
	import pandas as pd

	CWM3 = Path(__file__).resolve().parents[1]
	sys.path.insert(0, str(CWM3 / "eval"))
	import warning_ecan_vs_original as EV # noqa: E402 helpers only (main() never called)

	CR, OP, WG, LB = EV.CR, EV.OP, EV.WG, EV.LB
	f3, ci_s, verdict1, seed_rule, load_col, at_gold, strata = EV.f3, EV.ci_s, EV.verdict1, EV.seed_rule, EV.load_col, EV.at_gold, EV.strata
	REGIMES = ("manual", "engaged")
	RUNS = CWM3 / "results" / "runs"
	AT_ROOT = CWM3 / "results" / "alert_timing"
	STRATA = ("pooled", "manual", "engaged")


	def verdict_signed(d, lo, hi, lower_better=False):
	"""per-seed verdict of a paired Δ = a − b; lower_better=True (e.g. a false-alarm rate) flips the sign so that a NEGATIVE Δ with its CI below 0 reads BETTER."""
	if lower_better:
	return verdict1(-d if np.isfinite(d) else d, -hi if np.isfinite(hi) else hi, -lo if np.isfinite(lo) else lo)
	return verdict1(d, lo, hi)


	def find_timing_dir(mark, explicit=""):
	if explicit:
	return Path(explicit)
	cands = []
	for rj in sorted(AT_ROOT.glob("*/run.json")):
	try:
	if mark in rj.read_text():
	cands.append(rj.parent)
	except OSError:
	continue
	return cands[-1] if cands else None


	def layouts_present(pattern, seed):
	w = pattern.format(seed=seed)
	return all((RUNS / w / f"pred_layout_{r}_{s}.npz").exists() for r in REGIMES for s in ("test", "val"))


	def gold_block(seeds, specs, pairs, n_boot=2000, log=print):
	"""specs: {key: (pattern, column)}; pairs: [(a_key, b_key, label, prereg)] -> (levels df, pairs df, identity dict)."""
	lab = LB.episode_labels()
	lab = lab[lab.label != "unsure"].reset_index(drop=True)
	packs = {(r, s): WG.load_pack(r, s) for r in REGIMES for s in ("test", "val")}
	levels, prs, ident = [], [], {}
	for seed in seeds:
	G = CR.gold_for_seed(lab, packs, seed)
	yg, dong = G.y.to_numpy().astype(int), G.dongle.to_numpy()
	cols, gsc = {}, {}
	for key, (pat, col) in specs.items():
	if not layouts_present(pat, seed):
	log(f" seed {seed}: layouts of {key} ({pat.format(seed=seed)}) missing -> skipped")
	continue
	cols[key] = load_col(pat.format(seed=seed), col, packs)
	gsc[key] = at_gold(G, cols[key])
	for a, b, label, prereg in pairs:
	if a in cols and b in cols:
	maxabs, nan_mis, n_rows = 0.0, 0, 0
	for k in cols[a]:
	both = np.isfinite(cols[a][k]) & np.isfinite(cols[b][k])
	nan_mis += int((np.isfinite(cols[a][k]) != np.isfinite(cols[b][k])).sum())
	n_rows += len(cols[a][k])
	if both.any():
	maxabs = max(maxabs, float(np.abs(cols[a][k][both] - cols[b][k][both]).max()))
	ident[(a, b, seed)] = {"identical": maxabs == 0.0 and nan_mis == 0, "max_abs_diff": maxabs, "nan_mismatch_rows": nan_mis, "n_rows": n_rows}
	for st, m0 in strata(G).items():
	fin = m0.copy()
	for key in gsc:
	fin &= np.isfinite(gsc[key])
	y, dg = yg[fin], dong[fin]
	if len(y) == 0 or y.min() == y.max():
	continue
	S = {key: gsc[key][fin] for key in gsc}
	bs = WG.cluster_boot(y, S, dg, n_boot, 0)
	for key in gsc:
	for met, fn, kb in (("auroc", WG.fast_auroc, "auroc"), ("pr_auc", WG.fast_ap, "prauc")):
	lo, hi = WG.pct(bs[key][kb])
	levels.append({"key": key, "seed": seed, "stratum": st, "metric": met, "value": float(fn(y, S[key])), "ci_lo": lo, "ci_hi": hi, "n": int(fin.sum()), "n_pos": int(y.sum()),
	"n_dev": int(len(np.unique(dg))), "n_boot": n_boot})
	for a, b, label, prereg in pairs:
	if a not in gsc or b not in gsc:
	continue
	for met, fn, kb in (("auroc", WG.fast_auroc, "auroc"), ("pr_auc", WG.fast_ap, "prauc")):
	va, vb = float(fn(y, S[a])), float(fn(y, S[b]))
	d = bs[a][kb] - bs[b][kb]
	lo, hi = WG.pct(d)
	v = verdict1(va - vb, lo, hi)
	prs.append({"pair": f"{a} − {b}", "a": a, "b": b, "label": label, "prereg": prereg, "seed": seed, "stratum": st, "metric": met, "value_a": va, "value_b": vb, "delta": va - vb,
	"ci_lo": lo, "ci_hi": hi, "frac_boot_d_gt0": float(np.mean(d[np.isfinite(d)] > 0)) if np.isfinite(d).any() else np.nan, "n": int(fin.sum()), "n_pos": int(y.sum()),
	"n_dev": int(len(np.unique(dg))), "verdict": v, "n_boot": n_boot})
	log(f" s{seed} {st:8s} {met:6s} {a} − {b}: {va:.4f} − {vb:.4f} = {va - vb:+.4f} {ci_s(lo, hi, 4)} -> {v}")
	return pd.DataFrame(levels), pd.DataFrame(prs), ident


	def pair_summary(P):
	"""per (pair, stratum, metric): the seed rule over the per-seed verdicts."""
	rows = []
	if not len(P):
	return pd.DataFrame(rows)
	for (pair, st, met), g in P.groupby(["pair", "stratum", "metric"], sort=False):
	g = g.sort_values("seed")
	rule, nw, nb, n = seed_rule(g.verdict.tolist())
	rows.append({"pair": pair, "a": g.a.iloc[0], "b": g.b.iloc[0], "label": g.label.iloc[0], "prereg": bool(g.prereg.iloc[0]), "stratum": st, "metric": met, "seeds": n, "n_worse": nw, "n_better": nb,
	"rule": rule, "mean_delta": float(g.delta.mean()), "mean_a": float(g.value_a.mean()), "mean_b": float(g.value_b.mean()), "n": int(g.n.iloc[0]), "n_pos": int(g.n_pos.iloc[0]),
	"deltas": g.delta.round(4).tolist(), "ci": [[round(x, 4), round(y, 4)] for x, y in zip(g.ci_lo, g.ci_hi)]})
	return pd.DataFrame(rows)


	def timing_block(tdir, pairs, log=print):
	"""pairs: [(a, b, prereg)] -> (rows per endpoint / seed, summary per endpoint, levels)."""
	P, M = pd.read_csv(tdir / "paired.csv"), pd.read_csv(tdir / "metrics.csv")
	rows, summ, levels = [], [], []
	for na, nb, prereg in pairs:
	sub = P[(P.a == na) & (P.b == nb)]
	if not len(sub):
	log(f" timing pair {na} − {nb}: no rows")
	continue
	for (fit, fam, B, dl, reg), g in sub.groupby(["thr_fit", "family", "budget", "delta_s", "regime"]):
	g = g.sort_values("seed")
	vs = [verdict1(r.diff, r.ci_lo, r.ci_hi) for r in g.itertuples()]
	for r, v in zip(g.itertuples(), vs):
	rows.append({"pair": f"{na} − {nb}", "a": na, "b": nb, "prereg": prereg, "thr_fit": fit, "family": fam, "budget": B, "delta_s": dl, "regime": reg, "seed": int(r.seed), "value_a": r.value_a,
	"value_b": r.value_b, "delta": r.diff, "ci_lo": r.ci_lo, "ci_hi": r.ci_hi, "n_events": int(r.n_events), "verdict": v})
	rule, nw, nbet, n = seed_rule(vs)
	summ.append({"pair": f"{na} − {nb}", "a": na, "b": nb, "prereg": prereg, "thr_fit": fit, "family": fam, "budget": B, "delta_s": dl, "regime": reg, "seeds": n, "n_worse": nw, "n_better": nbet,
	"rule": rule, "mean_delta": float(g["diff"].mean()), "mean_a": float(g.value_a.mean()), "mean_b": float(g.value_b.mean()), "n_events": int(g.n_events.iloc[0]),
	"deltas": g["diff"].round(4).tolist()})
	names = sorted({x for a, b, _ in pairs for x in (a, b)})
	for sc in names:
	for (fit, fam, B, reg), g in M[(M.score == sc)].groupby(["thr_fit", "family", "budget", "regime"]):
	if "\|" in str(reg):
	continue
	g = g.sort_values("seed")
	levels.append({"score": sc, "thr_fit": fit, "family": fam, "budget": B, "regime": reg, "alerts_per_h_test": g.alerts_per_h_test.round(2).tolist(), "fa_per_h_test": g.fa_per_h_test.round(2).tolist(),
	"recall_early_05s": g["recall_early_0.5s"].round(4).tolist(), "recall_early_1s": g.recall_early_1s.round(4).tolist(), "recall_early_2s": g.recall_early_2s.round(4).tolist(),
	"detected": g.detected.round(4).tolist(), "lead0_mean": g.lead0_mean.round(3).tolist(), "n_events": int(g.n_events.iloc[0])})
	return pd.DataFrame(rows), pd.DataFrame(summ), pd.DataFrame(levels)


	def md_gold_table(L, S, keys_order, labels):
	"""levels + pair summary -> markdown lines: per arm x metric x stratum the per-seed values, and per pair the Δ rule."""
	out = ["\| arm \| metric \| stratum \| s0 / s1 / s2 (CI of s0) \| n / pos \|", "\|---\|---\|---\|---\|---\|"]
	for key in keys_order:
	for met in ("auroc", "pr_auc"):
	for st in STRATA:
	g = L[(L.key == key) & (L.metric == met) & (L.stratum == st)].sort_values("seed") if len(L) else L
	if not len(g):
	continue
	r0 = g.iloc[0]
	out.append(f"\| {labels.get(key, key)} \| {met} \| {st} \| {' / '.join(f3(x) for x in g.value)} {ci_s(r0.ci_lo, r0.ci_hi)} \| {int(r0.n)} / {int(r0.n_pos)} \|")
	out += ["", "\| pair (Δ = a − b) \| metric \| stratum \| Δ s0 / s1 / s2 \| CIs \| mean Δ \| rule \|", "\|---\|---\|---\|---\|---\|---\|---\|"]
	for r in S.itertuples():
	rule = r.rule if r.prereg else "descriptive: " + r.rule
	out.append(f"\| {r.label} \| {r.metric} \| {r.stratum} \| {' / '.join(f3(x, 3, True) for x in r.deltas)} \| {'; '.join(ci_s(x, y) for x, y in r.ci)} \| {f3(r.mean_delta, 3, True)} \| {rule} \|")
	return out


	def md_timing_tables(TS, TL, primary_pair, level_scores):
	out = []
	if not len(TS):
	return ["timing run not available"]
	A = TS[(TS.pair == primary_pair) & (TS.thr_fit == "val") & (TS.family == "A")]
	if len(A):
	out += [f"Family A, VAL-fitted thresholds (PRIMARY), pair `{primary_pair}` — every endpoint:", "", "\| K/h \| δ s \| regime \| a mean \| b mean \| Δ s0 / s1 / s2 \| mean Δ \| events \| rule \|", "\|---\|---\|---\|---\|---\|---\|---\|---\|---\|"]
	for r in A.sort_values(["budget", "delta_s", "regime"]).itertuples():
	out.append(f"\| {r.budget:g} \| {r.delta_s:g} \| {r.regime} \| {f3(r.mean_a)} \| {f3(r.mean_b)} \| {' / '.join(f3(x, 3, True) for x in r.deltas)} \| {f3(r.mean_delta, 3, True)} \| {r.n_events} \| {r.rule} \|")
	out += ["", "All pairs — endpoint counts per (pair, fit, family, K); 9 endpoints each = 3 δ × 3 regimes:", "",
	"\| pair \| prereg \| thr fit \| family \| K/h \| endpoints \| identical \| significantly worse \| better \| n.s. \| mean Δ \|", "\|---\|---\|---\|---\|---\|---\|---\|---\|---\|---\|---\|"]
	for (pair, prereg, fit, fam, B), g in TS.groupby(["pair", "prereg", "thr_fit", "family", "budget"], sort=False):
	ni = int(g.rule.str.startswith("identical").sum()); nw = int(g.rule.str.startswith("SIGNIFICANTLY").sum()); nb = int(g.rule.str.startswith("better").sum())
	out.append(f"\| {pair} \| {'yes' if prereg else 'descr.'} \| {fit} \| {fam} \| {B:g} \| {len(g)} \| {ni} \| {nw} \| {nb} \| {len(g) - ni - nw - nb} \| {f3(g.mean_delta.mean(), 4, True)} \|")
	if len(TL):
	out += ["", "Alert load and lead levels at the VAL-fitted thresholds (pooled, family A; per seed s0 / s1 / s2):", "",
	"\| score \| K/h \| alerts/h test \| FA/h \| Recall ≥ 0.5 s \| Recall ≥ 1 s \| Recall ≥ 2 s \| detected \| lead0 mean s \| events \|", "\|---\|---\|---\|---\|---\|---\|---\|---\|---\|---\|"]
	j = lambda xs, nd=3: " / ".join(f3(x, nd) for x in xs) # noqa: E731
	for r in TL[(TL.thr_fit == "val") & (TL.family == "A") & (TL.regime == "pooled") & (TL.score.isin(level_scores))].sort_values(["budget", "score"]).itertuples():
	out.append(f"\| {r.score} \| {r.budget:g} \| {j(r.alerts_per_h_test, 1)} \| {j(r.fa_per_h_test, 1)} \| {j(r.recall_early_05s)} \| {j(r.recall_early_1s)} \| {j(r.recall_early_2s)} \| {j(r.detected)} \| {j(r.lead0_mean, 2)} \| {r.n_events} \|")
	return out


	def append_table(path, title, md):
	path = Path(path)
	if not path.exists():
	path.write_text(f"# {title}\n\n")
	with open(path, "a") as f:
	f.write("\n" + md)


	def append_json(path, rec):
	path = Path(path)
	arr = json.loads(path.read_text()) if path.exists() else []
	arr.append(rec)
	path.write_text(json.dumps(arr, indent=1, default=str))