|
8 | 8 |
|
9 | 9 | from tracer.fit.pipeline import ( |
10 | 10 | build_global, _split_buffer, _accept_features, _fit_acceptor, _cp_lower, |
| 11 | + _calibrate_threshold, |
11 | 12 | ) |
12 | 13 |
|
13 | 14 |
|
@@ -41,6 +42,52 @@ def test_large_separable_certifies_with_lower_bound(): |
41 | 42 | assert g["summary"]["teacher_agreement_lower_cal_total"] >= 0.9 |
42 | 43 |
|
43 | 44 |
|
| 45 | +def _max_qualifying_coverage(scores, preds, y, target, alpha, min_accept): |
| 46 | + """Highest coverage among thresholds whose CP lower bound clears target.""" |
| 47 | + for t in np.unique(np.sort(scores)): |
| 48 | + acc = scores >= t |
| 49 | + n_acc = int(acc.sum()) |
| 50 | + if n_acc < min_accept: |
| 51 | + continue |
| 52 | + k_acc = int((preds[acc] == y[acc]).sum()) |
| 53 | + if _cp_lower(k_acc, n_acc, alpha) >= target: |
| 54 | + return float(acc.mean()) |
| 55 | + return None |
| 56 | + |
| 57 | + |
| 58 | +def test_single_set_gate_deploys_max_coverage_threshold(): |
| 59 | + # n < 40 -> single-set fallback branch. Among the thresholds that clear the |
| 60 | + # same CP bound, the gate must deploy the highest-coverage one, as the |
| 61 | + # holdout branch does. It previously kept the last (lowest-coverage) one. |
| 62 | + n = 36 |
| 63 | + scores = np.linspace(0.50, 0.99, n) |
| 64 | + y = np.zeros(n, dtype=int) |
| 65 | + preds = y.copy() |
| 66 | + preds[:6] = 1 # the 6 lowest-scoring rows disagree with the teacher |
| 67 | + |
| 68 | + out = _calibrate_threshold(scores, preds, y, 0.80, alpha=0.1, min_accept=10) |
| 69 | + assert out is not None and out["holdout"] is False |
| 70 | + |
| 71 | + best = _max_qualifying_coverage(scores, preds, y, 0.80, 0.1, 10) |
| 72 | + assert best is not None |
| 73 | + assert out["coverage"] == best |
| 74 | + # the guarantee still holds at the deployed threshold |
| 75 | + assert out["teacher_agreement_lower"] >= 0.80 |
| 76 | + |
| 77 | + |
| 78 | +def test_single_set_gate_coverage_does_not_shrink_with_more_data(): |
| 79 | + # On cleanly-separated calibration data the accepted set must grow with n, |
| 80 | + # not stay pinned at min_accept. |
| 81 | + covs = [] |
| 82 | + for n in (20, 30, 39): |
| 83 | + scores = np.linspace(0.50, 0.99, n) |
| 84 | + y = np.zeros(n, dtype=int) |
| 85 | + out = _calibrate_threshold(scores, y.copy(), y, 0.80, alpha=0.1, min_accept=10) |
| 86 | + assert out is not None |
| 87 | + covs.append(out["coverage"]) |
| 88 | + assert all(c == 1.0 for c in covs), covs |
| 89 | + |
| 90 | + |
44 | 91 | def test_nan_probs_do_not_crash_acceptor(): |
45 | 92 | probs = np.array([[0.7, 0.3], [np.nan, np.nan], [0.6, 0.4], [0.2, 0.8]]) |
46 | 93 | feats = _accept_features(probs) |
|
0 commit comments