Buckets:
| """tests/test_gold_event_recall.py — unit tests of eval/gold_event_recall.py (human gold-event Recall_{>= delta s early} @ K alerts/h): the threshold -> alerts/h | |
| inversion (exposure = all anchors, gate does not change it, floor flag), the 3 s per-route cool-down merging (against a brute-force reference), recall counting with | |
| delta / detected / late, bootstrap determinism and pairing, the project verdict rule, the by-product precision split by block, and a synthetic case where a perfect | |
| scorer reaches recall 1 while the chance scorer stays at the base rate. | |
| python -m pytest -q tests/test_gold_event_recall.py | |
| """ | |
| 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 gold_event_recall as GR # noqa: E402 | |
| import alert_timing_eval as AT # noqa: E402 | |
| def _grid(n_routes=2, hours_per_route=1.0, dt=0.2): | |
| n = int(round(hours_per_route * 3600 / dt)) | |
| uid = np.repeat(np.array([f"r{i}" for i in range(n_routes)]), n) | |
| dongle = np.repeat(np.array([f"d{i}" for i in range(n_routes)]), n) | |
| t = np.tile(np.round(np.arange(n) * dt, 1), n_routes) | |
| return uid, dongle, t, n_routes * hours_per_route | |
| def _brute_cooldown(uid, t, s, thr, cooldown): | |
| out, last = [], {} | |
| for j in range(len(t)): | |
| if np.isfinite(s[j]) and s[j] >= thr: | |
| if uid[j] not in last or t[j] - last[uid[j]] >= cooldown: | |
| out.append(j) | |
| last[uid[j]] = t[j] | |
| return np.asarray(out, np.int64) | |
| def test_cooldown_merging_per_route(): | |
| uid = np.array(["a"] * 10 + ["b"] * 3) | |
| t = np.array([0, 1, 2, 3.5, 4, 7, 8, 9, 10, 11] + [0, 1, 4.0]) | |
| s = np.ones(len(t), np.float32) | |
| ar = AT.alert_rows(uid, t, s, 0.5, GR.COOLDOWN) | |
| assert np.allclose(t[ar], [0, 3.5, 7, 10, 0, 4.0]), t[ar] # 1, 2 merged into 0; 4 into 3.5; 8, 9 into 7; 11 into 10; route b restarts the clock | |
| assert [uid[j] for j in ar] == ["a", "a", "a", "a", "b", "b"] | |
| s[3] = np.nan # a NaN (ineligible) row never fires and does not reset the clock | |
| ar = AT.alert_rows(uid, t, s, 0.5, GR.COOLDOWN) | |
| assert np.allclose(t[ar], [0, 4.0, 7, 10, 0, 4.0]), t[ar] | |
| rng = np.random.default_rng(0) | |
| uid, dongle, t, hours = _grid(3, 0.2) | |
| s = rng.random(len(t)).astype(np.float32) | |
| s[rng.random(len(t)) < 0.2] = np.nan | |
| for thr in (0.3, 0.8, 0.99): | |
| assert np.array_equal(AT.alert_rows(uid, t, s, thr, GR.COOLDOWN), _brute_cooldown(uid, t, s, thr, GR.COOLDOWN)) | |
| def test_threshold_alert_rate_inversion_and_floor(): | |
| rng = np.random.default_rng(1) | |
| uid, dongle, t, hours = _grid(2, 1.0) | |
| s = rng.random(len(t)).astype(np.float32) | |
| for K in (10.0, 20.0, 40.0): | |
| r = GR.thr_for_rate(uid, t, s, K, hours) | |
| assert r["reachable"] and r["achieved"] <= K + 1e-9 | |
| assert r["n_alerts"] >= int(K * hours) - 1, (K, r) # continuous scores: within one alert of the budget | |
| n_lo, rate_lo = GR.alerts_per_hour(uid, t, s, r["thr"] - 1e-3, hours) | |
| assert rate_lo > K - 1e-9 or n_lo == r["n_alerts"] # just below the threshold the budget is exceeded (or the score has no row in between) | |
| # the gate does not change the exposure: only 10 % of the rows eligible, the budget is still K per hour of ALL anchors | |
| el = rng.random(len(t)) < 0.1 | |
| r = GR.thr_for_rate(uid, t, GR.gated(s, el), 20.0, hours) | |
| assert r["reachable"] and abs(r["achieved"] - 20.0) <= 1.0 / hours + 1e-9 | |
| ar = AT.alert_rows(uid, t, GR.gated(s, el), r["thr"], GR.COOLDOWN) | |
| assert el[ar].all() # alerts only on eligible rows | |
| # floor: a binary score firing every 4 s (900 / h with a 3 s cool-down) cannot be brought down to 20 / h -> reported at the floor, flagged | |
| b = np.zeros(len(t), np.float32) | |
| b[np.isclose(np.mod(t, 4.0), 0.0)] = 1.0 | |
| r = GR.thr_for_rate(uid, t, b, 20.0, hours) | |
| assert not r["reachable"] and r["thr"] == 1.0 and r["achieved"] > 20.0 and np.isclose(r["floor"], r["achieved"]) | |
| # all-NaN score: nothing fires, trivially within budget | |
| r = GR.thr_for_rate(uid, t, np.full(len(t), np.nan, np.float32), 20.0, hours) | |
| assert r["n_alerts"] == 0 and r["achieved"] == 0.0 | |
| def test_recall_counting_with_delta_detected_late(): | |
| uid = np.array(["r"] * 501) | |
| t = np.round(np.arange(501) * 0.2, 1) # 0 .. 100 s | |
| gold = pd.DataFrame({"uid": ["r"] * 4, "t_star": [50.0, 80.0, 20.0, 60.0], "dongle": ["d"] * 4}) | |
| s = np.zeros(len(t), np.float32) | |
| for ta in (48.0, 79.6, 20.4, 56.5, 5.0): # lead 2.0; lead 0.4; late 0.4; 3.5 s early = outside the window; false alarm | |
| s[np.searchsorted(t, ta)] = 1.0 | |
| ar = AT.alert_rows(uid, t, s, 0.5, GR.COOLDOWN) | |
| lead = GR.match_gold(gold, uid, t, ar) | |
| assert np.allclose(np.nan_to_num(lead, nan=-99), [2.0, 0.4, -0.4, -99], atol=1e-6), lead | |
| ind = GR.recall_indicators(lead, (0.5, 1.0, 2.0)) | |
| assert ind.shape == (4, 3) | |
| assert np.allclose(ind.mean(0), [0.25, 0.25, 0.25]) # only the 2.0 s lead counts at every delta (>= 2 inclusive) | |
| ind = GR.recall_indicators(lead, (0.4, 0.3)) | |
| assert np.allclose(ind.mean(0), [0.5, 0.5]) | |
| st = GR.lead_stats(lead) | |
| assert np.isclose(st["detected"], 0.5) and np.isclose(st["late"], 0.25) and st["n_detected"] == 2 | |
| assert np.isclose(st["lead_median"], 1.2) and np.isclose(st["lead_mean"], 1.2) | |
| # an alert exactly at t* - delta counts (inclusive window), 0.2 s later does not | |
| s2 = np.zeros(len(t), np.float32) | |
| s2[np.searchsorted(t, 49.0)] = 1.0 | |
| lead2 = GR.match_gold(gold, uid, t, AT.alert_rows(uid, t, s2, 0.5, GR.COOLDOWN)) | |
| assert np.isclose(lead2[0], 1.0) and GR.recall_indicators(lead2, (1.0,))[0, 0] == 1.0 and GR.recall_indicators(lead2, (1.2,))[0, 0] == 0.0 | |
| # the FIRST alert in the window defines the lead: an early alert followed by a late one keeps the early lead | |
| s3 = np.zeros(len(t), np.float32) | |
| s3[np.searchsorted(t, 47.4)] = 1.0 | |
| s3[np.searchsorted(t, 50.4)] = 1.0 | |
| lead3 = GR.match_gold(gold, uid, t, AT.alert_rows(uid, t, s3, 0.5, GR.COOLDOWN)) | |
| assert np.isclose(lead3[0], 2.6) | |
| def test_bootstrap_determinism_and_pairing(): | |
| rng = np.random.default_rng(2) | |
| dongle = np.array([f"d{i % 5}" for i in range(60)]) | |
| ind = (rng.random((60, 4)) < 0.3).astype(float) | |
| ind[:, 2] = ind[:, 0] # identical systems -> paired difference exactly 0 on every resample | |
| b1 = GR.paired_boot(dongle, ind, 200, 0) | |
| b2 = GR.paired_boot(dongle, ind, 200, 0) | |
| assert b1.shape == (200, 4) and np.array_equal(b1, b2) | |
| assert not np.array_equal(b1, GR.paired_boot(dongle, ind, 200, 1)) | |
| assert np.all(b1[:, 0] - b1[:, 2] == 0.0) | |
| lo, hi = AT.pct(b1[:, 0] - b1[:, 1]) | |
| assert lo <= ind[:, 0].mean() - ind[:, 1].mean() <= hi | |
| def test_verdict_rule(): | |
| mk = lambda diffs, half, va, vb: [dict(seed=k, diff=d, ci_lo=d - half, ci_hi=d + half, value_a=a, value_b=b) for k, (d, a, b) in enumerate(zip(diffs, va, vb))] # noqa: E731 | |
| v = GR.verdict(mk([0.10, 0.12, 0.11], 0.05, [0.5, 0.52, 0.51], [0.4, 0.4, 0.4])) | |
| assert v["verdict"] == "BEATS" and v["ci_excl_zero"] == 3 | |
| v = GR.verdict(mk([0.10, 0.12, 0.03], 0.05, [0.5, 0.52, 0.43], [0.4, 0.4, 0.4])) | |
| assert v["verdict"] == "BEATS" and v["ci_excl_zero"] == 2 # >= 2/3 seeds with CI excl. 0, same sign | |
| v = GR.verdict(mk([-0.10, -0.12, -0.11], 0.05, [0.4, 0.4, 0.4], [0.5, 0.52, 0.51])) | |
| assert v["verdict"] == "WORSE" and v["ci_excl_zero"] == 3 | |
| v = GR.verdict(mk([0.002, -0.003, 0.001], 0.05, [0.4, 0.4, 0.4], [0.4, 0.4, 0.4])) | |
| assert v["verdict"] == "NOT_WORSE" and v["ci_excl_zero"] == 0 | |
| v = GR.verdict(mk([0.10, -0.12, 0.11], 0.05, [0.5, 0.3, 0.51], [0.4, 0.42, 0.4])) | |
| assert v["verdict"] == "INCONCLUSIVE" | |
| assert GR.verdict([])["verdict"] == "INSUFFICIENT" | |
| assert GR.verdict(mk([0.1, 0.1], 0.01, [0.5, 0.5], [0.4, 0.4]))["verdict"] == "INSUFFICIENT" # < 3 seeds | |
| def test_precision_by_block_splits_R_and_RE(): | |
| items = [{"id": "G_0", "meta": {"block": "R200", "event": "hand_return"}}, {"id": "G_1", "meta": {"block": "R200", "event": "hand_return"}}, | |
| {"id": "G_2", "meta": {"block": "R100", "event": "hand_return"}}, {"id": "G_3", "meta": {"block": "R100", "event": "hand_return"}}, | |
| {"id": "GE_0", "meta": {"block": "RE", "event": "hand_return"}}, {"id": "GE_1", "meta": {"block": "RE", "event": "hand_return"}}, | |
| {"id": "G_9", "meta": {"block": "L100", "event": "hand_leave"}}, {"id": "GW_0", "meta": {"block": "W", "event": "recall_window"}}, | |
| {"id": "GM_0", "meta": {"block": "MR", "event": "hand_return"}}] | |
| export = {"G_0": {"label": "at_event"}, "G_1": {"label": "ret_dt=0.3"}, "G_2": {"label": "never"}, "G_3": {"label": "unsure"}, "GE_0": {"label": "always"}, "GE_1": {"label": "never"}, | |
| "GW_0": {"label": "yes"}, "GM_0": {"label": "at_event"}} | |
| p = GR.precision_by_block(items, export) | |
| R, RE, L, MR = p["R (R200+R100, random)"], p["RE (VLM-ranked)"], p["L (L100+L50)"], p["MR (manual)"] | |
| assert R["n"] == 4 and R["labelled"] == 4 and R["gold"] == 2 and R["n_eff"] == 3 and np.isclose(R["precision"], 2 / 3) and R["ci"][0] < 2 / 3 < R["ci"][1] | |
| assert R["at_event"] == 1 and R["shifted"] == 1 and R["never"] == 1 and R["unsure"] == 1 | |
| assert RE["n"] == 2 and RE["gold"] == 0 and RE["n_eff"] == 2 and RE["precision"] == 0.0 # RE never pooled into R | |
| assert L["n"] == 1 and L["labelled"] == 0 and L["unlabelled"] == 1 and not np.isfinite(L["precision"]) | |
| assert MR["gold"] == 1 and MR["precision"] == 1.0 | |
| assert sum(v["n"] for v in p.values()) == 8 # the W window is not a candidate | |
| def test_load_gold_maps_dongle_and_drops_unknown_routes(tmp_path): | |
| uid = np.array(["r0"] * 5 + ["r1"] * 5) | |
| grid = {"uid": uid, "dongle": np.array(["dA"] * 5 + ["dB"] * 5)} | |
| G = {"gold_events": [{"id": "G_0", "uid": "r0", "car": "c", "regime": "engaged", "event": "hand_return", "t_anchor": 1.0, "t_event_candidate": 1.2, "t_star": 1.5000000001, "dt_star": 0.3, "label": "ret_dt=0.3", "block": "R200"}, | |
| {"id": "G_1", "uid": "rX", "car": "c", "regime": "engaged", "event": "hand_return", "t_anchor": 1.0, "t_event_candidate": 1.2, "t_star": 1.2, "dt_star": 0.0, "label": "at_event", "block": "R200"}, | |
| {"id": "G_2", "uid": "r1", "car": "c", "regime": "engaged", "event": "hand_leave", "t_anchor": 1.0, "t_event_candidate": 1.2, "t_star": 1.2, "dt_star": 0.0, "label": "at_event", "block": "L100"}]} | |
| p = tmp_path / "gold.json" | |
| p.write_text(json.dumps(G)) | |
| df, n_drop = GR.load_gold(p, "engaged", "hand_return", grid) | |
| assert n_drop == 1 and len(df) == 1 and df.dongle.iloc[0] == "dA" and df.t_star.iloc[0] == 1.5 | |
| df, n_drop = GR.load_gold(p, "engaged", "hand_leave", grid) | |
| assert n_drop == 0 and len(df) == 1 and df.dongle.iloc[0] == "dB" | |
| df, _ = GR.load_gold(p, "manual", "hand_return", grid) | |
| assert len(df) == 0 | |
| def test_eligible_gate(): | |
| grid = {"t": np.arange(6.0), "hands_on": {"v3": np.array([0, 1, np.nan, 0, 1, 1], np.float32), "v2": np.array([1, 1, 1, 0, 0, 0], np.float32)}} | |
| assert GR.eligible(grid, "hand_return", "v3").tolist() == [True, False, False, True, False, False] | |
| assert GR.eligible(grid, "hand_leave", "v3").tolist() == [False, True, False, False, True, True] | |
| assert GR.eligible(grid, "hand_return", "v2").tolist() == [False, False, False, True, True, True] | |
| assert GR.eligible(grid, "hand_leave", "none").all() | |
| g = GR.gated(np.arange(6, dtype=np.float32), GR.eligible(grid, "hand_return", "v3")) | |
| assert np.isnan(g[[1, 2, 4, 5]]).all() and g[0] == 0.0 and g[3] == 3.0 | |
| def test_perfect_scorer_recall_one_chance_at_base_rate(): | |
| """5 h of exposure on 4 routes, one gold event every 6 min (10 / h): at K = 20 / h a scorer that fires only in [t* - 2, t* - 1] reaches Recall 1 at every delta; | |
| the chance scorer (seeded uniform) stays near the analytic base rate 1 - exp(-K (3 - delta) / 3600) (~1 % at delta = 1 s).""" | |
| uid, dongle, t, hours = _grid(4, 1.25) | |
| n_per = len(t) // 4 | |
| gold_rows = [] | |
| for r in range(4): | |
| for k in range(12): | |
| gold_rows.append({"uid": f"r{r}", "dongle": f"d{r}", "t_star": 300.0 + 360.0 * k}) | |
| gold = pd.DataFrame(gold_rows) | |
| assert len(gold) == 48 | |
| perfect = np.zeros(len(t), np.float32) | |
| for g in gold.itertuples(): | |
| m = (uid == g.uid) & (t >= g.t_star - 2.0 - 1e-6) & (t <= g.t_star - 1.0 + 1e-6) | |
| perfect[m] = 1.0 | |
| K = 20.0 | |
| r = GR.thr_for_rate(uid, t, perfect, K, hours) | |
| assert r["reachable"] and r["achieved"] <= K | |
| lead = GR.match_gold(gold, uid, t, AT.alert_rows(uid, t, perfect, r["thr"], GR.COOLDOWN)) | |
| ind = GR.recall_indicators(lead, (0.5, 1.0, 2.0)) | |
| assert np.allclose(ind.mean(0), [1.0, 1.0, 1.0]), ind.mean(0) | |
| assert np.allclose(lead, 2.0) | |
| chance = np.random.default_rng(GR.CHANCE_SEED).random(len(t)).astype(np.float32) | |
| r = GR.thr_for_rate(uid, t, chance, K, hours) | |
| assert r["reachable"] and abs(r["achieved"] - K) <= 1.0 / hours + 1e-9 | |
| lead_c = GR.match_gold(gold, uid, t, AT.alert_rows(uid, t, chance, r["thr"], GR.COOLDOWN)) | |
| rc = GR.recall_indicators(lead_c, (0.5, 1.0, 2.0)).mean(0) | |
| base = 1.0 - np.exp(-K * (3.0 - np.array([0.5, 1.0, 2.0])) / 3600.0) # ~0.014 / 0.011 / 0.006 | |
| assert (rc <= 0.15).all() and (rc < 1.0).all(), rc | |
| assert np.all(rc[0] >= rc[1]) and np.all(rc[1] >= rc[2]) # monotone in delta | |
| assert abs(rc[1] - base[1]) < 0.1 | |
Xet Storage Details
- Size:
- 13.8 kB
- Xet hash:
- 85eca253fcabc8f974a63aca5ec4f379ad86792e7315cd1cee039fa66fe9facb
·
Xet efficiently stores files, intelligently splitting them into unique chunks and accelerating uploads and downloads. More info.