Buckets:
| """cwm3/eval/overlap_eval.py — results/tables/overlap_head_f8.md (+ .json): F8 joint overlap head vs the product baseline (R2 WP4 batch 2, 2026-09-17; | |
| pre-registration docs/decisions.md 2026-09-17T19:32:07Z, endpoints (1) and (2)). CPU, numpy only. | |
| python eval/overlap_eval.py [--seeds 0,1,2] [--split test|val] [--n-boot 1000] [--out results/tables/overlap_head_f8.md] | |
| [--pairs joint:prod,joint:persistence,joint:margonly,margonly:prod,prod:persistence] | |
| Inputs: results/runs/OV_<arm>_s<k>/pred_overlap_{regime}_{split}.npz (arms joint / margonly / prod: q [N,15] paper-facing, q_hands, q_head, J), the target cache | |
| results/runs/OV_cache/targets_{regime}_{split}.npz (S / row_ok / M / P / Pvalid / P0 / P0valid; uid / t witness against every arm) -> models/overlap_head.build_targets | |
| (strict future: horizon k -> index k−1). `persistence` = the current-state read-out q_k = 1[U_t] 1[D_t] for every k (U_t from the anchor states S0 / P0, invalid | |
| component = met), J = 0.2 · 15 · q. | |
| Endpoints per regime, identical rows, shared-resample device-cluster bootstrap (n_boot, seed 0; ties in AUROC handled by mid-ranks; PR-AUC = paired_compare.ap_weighted | |
| convention, descending stable order), Δ = A − B per seed, verdict = kill_checks.beats over the 3 seeds (report_joint.verdict; lower better for Brier): | |
| (1) q_k at k = 3 (0.6 s) / 5 / 10 / 15: Brier and AUROC of P(U_{t+k} ∧ D_{t+k}); rows = row_ok ∧ the U components and D observed at k | |
| (manual U = hands_off ∨ head_turned; engaged U = head_turned only — engaged hand cells masked from every paper-facing endpoint); | |
| (2) J_t ranking: PR-AUC and AUROC against the realised overlap any_{k ≤ 15}(U ∧ D) on rows whose 15-step window is fully observed. | |
| Descriptive extras in the json: the hands-inclusive engaged reading (q_hands), calibration ECE (10 bins) of q at 3 s, base rates, n rows / positives / devices. | |
| Append-only: a new dated section per invocation. Nothing here reads an openpilot DMS field. | |
| """ | |
| from __future__ import annotations | |
| import argparse | |
| import json | |
| import sys | |
| import time | |
| from pathlib import Path | |
| import numpy as np | |
| import pandas as pd | |
| CWM3 = Path(__file__).resolve().parents[1] | |
| for p in (CWM3, CWM3 / "eval", CWM3.parent / "cwm2" / "eval"): | |
| if str(p) not in sys.path: | |
| sys.path.insert(0, str(p)) | |
| import report_joint as RJ # noqa: E402 | |
| from models.overlap_head import build_targets, realised_overlap, REPORT_STEPS, DT, K # noqa: E402 | |
| RUNS = CWM3 / "results" / "runs" | |
| CACHE = RUNS / "OV_cache" | |
| REGIMES = ("manual", "engaged") | |
| PAIRS_DEFAULT = "joint:prod,joint:persistence,joint:margonly,margonly:prod,prod:persistence" | |
| # ----------------------------------------------------------------------------- paired device-cluster bootstrap (fixed orders, O(N) per resample) | |
| class Boot: | |
| def __init__(self, dongle, n_boot, seed=0): | |
| self.uniq, self.dev = np.unique(dongle, return_inverse=True) | |
| self.n_boot, self.seed = int(n_boot), int(seed) | |
| def weights(self): | |
| rng = np.random.default_rng(self.seed) | |
| for _ in range(self.n_boot): | |
| cnt = np.bincount(rng.integers(0, len(self.uniq), len(self.uniq)), minlength=len(self.uniq)).astype(np.float64) | |
| yield cnt[self.dev] | |
| def _auroc_prep(y, s): | |
| s = np.asarray(s, np.float64) | |
| order = np.argsort(s, kind="mergesort") | |
| ss, ys = s[order], np.asarray(y, bool)[order] | |
| _, grp = np.unique(ss, return_inverse=True) | |
| return order, ys, grp | |
| def _auroc_w(prep, w): | |
| order, ys, grp = prep | |
| ws = w[order] | |
| neg = ws * ~ys | |
| pos = ws * ys | |
| n_pos, n_neg = pos.sum(), neg.sum() | |
| if n_pos <= 0 or n_neg <= 0: | |
| return np.nan | |
| neg_g = np.bincount(grp, neg, minlength=grp.max() + 1) | |
| cum_before = np.cumsum(neg_g) - neg_g | |
| return float((pos * (cum_before[grp] + 0.5 * neg_g[grp])).sum() / (n_pos * n_neg)) | |
| def _ap_prep(y, s): | |
| order = np.argsort(-np.asarray(s, np.float64), kind="mergesort") | |
| return np.asarray(y, bool)[order], order | |
| def _ap_w(prep, w): | |
| ys, order = prep | |
| ws = w[order] | |
| tp, fp = np.cumsum(ws * ys), np.cumsum(ws * ~ys) | |
| if tp[-1] <= 0: | |
| return np.nan | |
| prec = tp / np.maximum(tp + fp, 1e-12) | |
| rec = tp / tp[-1] | |
| return float((prec * np.diff(np.r_[0.0, rec])).sum()) | |
| def paired(y, sa, sb, boot, metric): | |
| """-> dict(value_a, value_b, ci_a, ci_b, delta, ci_d, frac_gt0, n_dev, n, n_pos).""" | |
| y = np.asarray(y, bool) | |
| N = len(y) | |
| ones = np.ones(N) | |
| if metric == "brier": | |
| ea, eb = (np.asarray(sa, np.float64) - y) ** 2, (np.asarray(sb, np.float64) - y) ** 2 | |
| f = lambda w: ((w * ea).sum() / w.sum(), (w * eb).sum() / w.sum()) | |
| elif metric == "auroc": | |
| pa_, pb_ = _auroc_prep(y, sa), _auroc_prep(y, sb) | |
| f = lambda w: (_auroc_w(pa_, w), _auroc_w(pb_, w)) | |
| elif metric == "pr_auc": | |
| pa_, pb_ = _ap_prep(y, sa), _ap_prep(y, sb) | |
| f = lambda w: (_ap_w(pa_, w), _ap_w(pb_, w)) | |
| else: | |
| raise ValueError(metric) | |
| pa, pb = f(ones) | |
| da, db = np.full(boot.n_boot, np.nan), np.full(boot.n_boot, np.nan) | |
| for i, w in enumerate(boot.weights()): | |
| if (w * y).sum() <= 0 or (w * y).sum() >= w.sum(): | |
| continue | |
| da[i], db[i] = f(w) | |
| d = da - db | |
| ok = np.isfinite(d) | |
| pct = lambda v: (float(np.nanpercentile(v, 2.5)), float(np.nanpercentile(v, 97.5))) | |
| return {"value_a": float(pa), "value_b": float(pb), "ci_a": pct(da), "ci_b": pct(db), "delta": float(pa - pb), "ci_d": pct(d), | |
| "frac_gt0": float((d[ok] > 0).mean()) if ok.any() else np.nan, "n_dev": int(len(boot.uniq)), "n": int(N), "n_pos": int(y.sum())} | |
| def ece(p, y, bins=10): | |
| p = np.asarray(p, np.float64) | |
| e, edges = 0.0, np.linspace(0, 1, bins + 1) | |
| idx = np.clip(np.digitize(p, edges[1:-1]), 0, bins - 1) | |
| for b in range(bins): | |
| m = idx == b | |
| if m.any(): | |
| e += m.mean() * abs(p[m].mean() - y[m].mean()) | |
| return float(e) | |
| # ----------------------------------------------------------------------------- data | |
| def load_arm(arm, seed, regime, split): | |
| p = RUNS / f"OV_{arm}_s{seed}" / f"pred_overlap_{regime}_{split}.npz" | |
| if not p.exists(): | |
| return None | |
| with np.load(p, allow_pickle=True) as z: | |
| return {k: z[k] for k in z.files} | |
| def load_targets(regime, split): | |
| with np.load(CACHE / f"targets_{regime}_{split}.npz", allow_pickle=True) as z: | |
| tg = {k: z[k] for k in z.files} | |
| y3, obs3 = build_targets(tg["S"], tg["P"], tg["Pvalid"], tg["M"], tg["row_ok"]) | |
| tg["y3"], tg["obs3"] = y3, obs3 | |
| S0 = tg["S0"].astype(np.float32) | |
| hands_off0 = np.isfinite(S0[:, 0]) & (S0[:, 0] < 0.5) | |
| head0 = np.isfinite(S0[:, 1]) & (S0[:, 1] > 0.5) | |
| D0 = (tg["P0"] & tg["P0valid"]).any(1) | |
| tg["pers_hands"] = ((hands_off0 | head0) & D0).astype(np.float32) | |
| tg["pers_head"] = (head0 & D0).astype(np.float32) | |
| return tg | |
| def arm_info(arm, seed): | |
| c = RUNS / f"OV_{arm}_s{seed}" / "config.json" | |
| if not c.exists(): | |
| return {} | |
| j = json.loads(c.read_text()) | |
| return {"params": j.get("params"), "best_epoch": j.get("best_epoch"), "val_joint_nll": j.get("val_joint_nll"), "frozen": j.get("frozen"), | |
| "witness": (j.get("meta") or {}).get("witness"), "model_id": (j.get("meta") or {}).get("model_id")} | |
| def main(): | |
| ap = argparse.ArgumentParser() | |
| ap.add_argument("--seeds", default="0,1,2") | |
| ap.add_argument("--split", default="test", choices=["test", "val"]) | |
| ap.add_argument("--n-boot", type=int, default=1000) | |
| ap.add_argument("--pairs", default=PAIRS_DEFAULT) | |
| ap.add_argument("--out", default=str(CWM3 / "results" / "tables" / "overlap_head_f8.md")) | |
| a = ap.parse_args() | |
| t0 = time.time() | |
| stamp = time.strftime("%Y%m%dT%H%M%SZ", time.gmtime()) | |
| seeds = [int(s) for s in a.seeds.split(",")] | |
| pairs = [tuple(p.split(":")) for p in a.pairs.split(",") if p] | |
| arms = sorted({x for p in pairs for x in p if x != "persistence"}) | |
| rows, extras = [], {} | |
| for regime in REGIMES: | |
| use_hands = regime == "manual" | |
| tg = load_targets(regime, a.split) | |
| y3, obs3, ok = tg["y3"], tg["obs3"], tg["row_ok"].astype(bool) | |
| need = (obs3[:, :, 0] & obs3[:, :, 1] & obs3[:, :, 2]) if use_hands else (obs3[:, :, 1] & obs3[:, :, 2]) | |
| U = ((y3[:, :, 0] > 0) | (y3[:, :, 1] > 0)) if use_hands else (y3[:, :, 1] > 0) | |
| yk = U & (y3[:, :, 2] > 0) # [N,K] | |
| yJ, full = realised_overlap(y3, obs3, use_hands) | |
| rowsJ = ok & full | |
| pers_q = tg["pers_hands"] if use_hands else tg["pers_head"] | |
| for seed in seeds: | |
| S = {} | |
| for arm in arms: | |
| z = load_arm(arm, seed, regime, a.split) | |
| if z is None: | |
| print(f"[{regime} s{seed}] arm {arm} missing"); continue | |
| assert np.array_equal(z["uid"], tg["uid"]) and np.allclose(z["t"], tg["t"]), f"{arm} s{seed} {regime}: rows differ from the target cache" | |
| S[arm] = {"q": z["q"].astype(np.float32), "J": z["J"].astype(np.float32), "q_hands": z["q_hands"].astype(np.float32), "J_hands": z["J_hands"].astype(np.float32)} | |
| S["persistence"] = {"q": np.repeat(pers_q[:, None], K, 1), "J": DT * K * pers_q, "q_hands": np.repeat(tg["pers_hands"][:, None], K, 1), "J_hands": DT * K * tg["pers_hands"]} | |
| boot_cache = {} | |
| for A, B in pairs: | |
| if A not in S or B not in S: | |
| continue | |
| for hname, k in REPORT_STEPS.items(): | |
| r = ok & need[:, k - 1] | |
| key = ("k", k) | |
| if key not in boot_cache: | |
| boot_cache[key] = Boot(tg["dongle"][r], a.n_boot, 0) | |
| y = yk[r, k - 1] | |
| for metric in ("brier", "auroc"): | |
| res = paired(y, S[A]["q"][r, k - 1], S[B]["q"][r, k - 1], boot_cache[key], metric) | |
| rows.append(dict(regime=regime, split=a.split, seed=seed, a=A, b=B, cell=f"q@{hname}", metric=metric, **{kk: v for kk, v in res.items() if kk not in ("ci_a", "ci_b", "ci_d")}, | |
| ci_a_lo=res["ci_a"][0], ci_a_hi=res["ci_a"][1], ci_b_lo=res["ci_b"][0], ci_b_hi=res["ci_b"][1], delta_ci_lo=res["ci_d"][0], delta_ci_hi=res["ci_d"][1], | |
| base_rate=float(y.mean()))) | |
| if ("J",) not in boot_cache: | |
| boot_cache[("J",)] = Boot(tg["dongle"][rowsJ], a.n_boot, 0) | |
| for metric in ("pr_auc", "auroc"): | |
| res = paired(yJ[rowsJ], S[A]["J"][rowsJ], S[B]["J"][rowsJ], boot_cache[("J",)], metric) | |
| rows.append(dict(regime=regime, split=a.split, seed=seed, a=A, b=B, cell="J_3s", metric=metric, **{kk: v for kk, v in res.items() if kk not in ("ci_a", "ci_b", "ci_d")}, | |
| ci_a_lo=res["ci_a"][0], ci_a_hi=res["ci_a"][1], ci_b_lo=res["ci_b"][0], ci_b_hi=res["ci_b"][1], delta_ci_lo=res["ci_d"][0], delta_ci_hi=res["ci_d"][1], | |
| base_rate=float(yJ[rowsJ].mean()))) | |
| print(f"[{time.strftime('%H:%M:%S')}] {regime} s{seed} {A} vs {B} done ({time.time() - t0:.0f}s)", flush=True) | |
| # descriptive extras (no bootstrap): ECE at 3 s, hands-inclusive engaged reading | |
| r15 = ok & need[:, 14] | |
| ex = {arm: {"ece_q3s": ece(S[arm]["q"][r15, 14], yk[r15, 14]), "mean_q3s": float(S[arm]["q"][r15, 14].mean())} for arm in S} | |
| if not use_hands: | |
| need_h = obs3[:, :, 0] & obs3[:, :, 1] & obs3[:, :, 2] | |
| yh = ((y3[:, :, 0] > 0) | (y3[:, :, 1] > 0)) & (y3[:, :, 2] > 0) | |
| rh = ok & need_h[:, 14] | |
| for arm in S: | |
| e = (S[arm]["q_hands"][rh, 14] - yh[rh, 14]) ** 2 | |
| ex[arm]["hands_inclusive_brier_3s_descriptive"] = float(e.mean()) | |
| ex["hands_inclusive_base_rate_3s"] = float(yh[rh, 14].mean()) | |
| extras[f"{regime}_s{seed}"] = dict(ex, base_rate_q3s=float(yk[r15, 14].mean()), n_rows_3s=int(r15.sum()), n_rows_J=int(rowsJ.sum()), base_rate_J=float(yJ[rowsJ].mean())) | |
| df = pd.DataFrame(rows) | |
| # ---- verdicts | |
| verd = {} | |
| for (regime, A, B, cell, metric), g in df.groupby(["regime", "a", "b", "cell", "metric"]): | |
| g = g.sort_values("seed") | |
| lb = metric == "brier" | |
| v = RJ.verdict(g, lower_better=lb) | |
| verd[(regime, A, B, cell, metric)] = v | |
| infos = {f"{arm}_s{s}": arm_info(arm, s) for arm in arms for s in seeds} | |
| out = Path(a.out) | |
| out.parent.mkdir(parents=True, exist_ok=True) | |
| L = [f"## F8 overlap head — split {a.split}, seeds {seeds}, n_boot {a.n_boot}, run {stamp}", "", | |
| f"Pre-registration docs/decisions.md 2026-09-17T19:32:07Z. Arms: {', '.join(arms)} (+ persistence = current-state read-out). Frozen model per seed: " | |
| f"{', '.join(sorted({str(v.get('frozen')) for v in infos.values() if v}))}. Params: " + ", ".join(f"{k} {v.get('params')}" for k, v in infos.items() if v) + ".", | |
| "Δ = A − B on identical rows, shared-resample device-cluster bootstrap; verdict = kill_checks.beats over the seeds (lower better for Brier). q@h = P(U_{t+h} ∧ D_{t+h}); " | |
| "J_3s = 0.2 Σ_{k≤15} q_k ranked against the realised overlap within 3 s (fully observed windows). Manual U = hands_off ∨ head_turned; engaged U = head_turned only " | |
| "(engaged hand cells are out of the paper).", ""] | |
| for regime in REGIMES: | |
| L += [f"### {regime}", "", "| A | B | cell | metric | " + " | ".join(f"A s{s}" for s in seeds) + " | " + " | ".join(f"B s{s}" for s in seeds) + " | " + " | ".join(f"Δ s{s} [CI]" for s in seeds) + | |
| " | mean Δ ± sd | k/3 | verdict | n / pos / dev |", "|---|---|---|---|" + "---|" * (3 * len(seeds)) + "---|---|---|---|"] | |
| for A, B in pairs: | |
| for cell in [f"q@{h}" for h in REPORT_STEPS] + ["J_3s"]: | |
| for metric in (("brier", "auroc") if cell != "J_3s" else ("pr_auc", "auroc")): | |
| g = df[(df.regime == regime) & (df.a == A) & (df.b == B) & (df.cell == cell) & (df.metric == metric)].sort_values("seed") | |
| if not len(g): | |
| continue | |
| v = verd[(regime, A, B, cell, metric)] | |
| by = {int(r.seed): r for r in g.itertuples()} | |
| Av = [f"{by[s].value_a:.4f}" if s in by else "—" for s in seeds] | |
| Bv = [f"{by[s].value_b:.4f}" if s in by else "—" for s in seeds] | |
| Dv = [f"{by[s].delta:+.4f} [{by[s].delta_ci_lo:+.4f},{by[s].delta_ci_hi:+.4f}]" if s in by else "—" for s in seeds] | |
| d = g.delta.astype(float) | |
| L.append(f"| {A} | {B} | {cell} | {metric} | " + " | ".join(Av) + " | " + " | ".join(Bv) + " | " + " | ".join(Dv) + | |
| f" | {d.mean():+.4f} ± {d.std(ddof=1) if len(d) > 1 else 0:.4f} | {v.get('ci_excl_zero', 0)}/{len(g)} | **{v['verdict']}** | {int(g.n.iloc[0]):,} / {int(g.n_pos.iloc[0]):,} / {int(g.n_dev.iloc[0])} |") | |
| L.append("") | |
| L += ["Descriptive (no bootstrap): " + json.dumps(extras, default=float), ""] | |
| with open(out, "a") as f: | |
| f.write("\n".join(L) + "\n") | |
| js = out.with_name(out.stem + f"_{stamp}.json") | |
| js.write_text(json.dumps({"stamp": stamp, "args": vars(a), "rows": rows, "verdicts": {"|".join(k): v for k, v in verd.items()}, "extras": extras, "arm_info": infos, | |
| "wall_s": round(time.time() - t0, 1)}, indent=1, default=float)) | |
| print(f"wrote {out} and {js} in {time.time() - t0:.0f}s") | |
| if __name__ == "__main__": | |
| main() | |
Xet Storage Details
- Size:
- 15.8 kB
- Xet hash:
- bcf1f48762ead9471ddb342ece0c19abe4b9a861167ce9b26e8016899f45c440
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.