|
| 1 | +"""Tier D P2 known-truth upgrades — conformal prediction coverage guarantees. |
| 2 | +
|
| 3 | +Part of the P1/P2 "Tier D analytic special-cases" campaign (see |
| 4 | +``.tierd_campaign/CAMPAIGN.md``). Both were graded ``weak`` by |
| 5 | +``scripts/tierd_classify.py``. Each anchors to the conformal coverage guarantee |
| 6 | +— a finite-sample known truth: |
| 7 | +
|
| 8 | + sp.weighted_conformal_prediction split-conformal intervals attain marginal |
| 9 | + coverage ~ 1-alpha on exchangeable test |
| 10 | + points; coverage falls as alpha rises. |
| 11 | + sp.conformal_ite_interval the ITE interval covers a known constant |
| 12 | + treatment effect at >= 1-alpha, and the |
| 13 | + interval narrows as alpha rises. |
| 14 | +
|
| 15 | +Purely additive — no estimator numerics changed (campaign red line). |
| 16 | +""" |
| 17 | + |
| 18 | +import numpy as np |
| 19 | +import pandas as pd |
| 20 | +import pytest |
| 21 | + |
| 22 | +import statspai as sp |
| 23 | + |
| 24 | + |
| 25 | +# --------------------------------------------------------------------------- |
| 26 | +# sp.weighted_conformal_prediction — split conformal coverage |
| 27 | +# --------------------------------------------------------------------------- |
| 28 | +class TestWeightedConformalAnalytic: |
| 29 | + |
| 30 | + @staticmethod |
| 31 | + def _split(seed=1, n_tr=2000, n_ca=1500, n_te=5000): |
| 32 | + rng = np.random.default_rng(seed) |
| 33 | + beta = np.array([1.0, -0.5, 0.3]) |
| 34 | + |
| 35 | + def gen(n): |
| 36 | + X = rng.normal(0, 1, (n, 3)) |
| 37 | + return X, X @ beta + rng.normal(0, 1, n) |
| 38 | + |
| 39 | + return gen(n_tr), gen(n_ca), gen(n_te) |
| 40 | + |
| 41 | + def test_marginal_coverage_near_nominal(self): |
| 42 | + (Xtr, ytr), (Xca, yca), (Xte, yte) = self._split() |
| 43 | + lo, hi, _ = sp.weighted_conformal_prediction(Xtr, ytr, Xca, yca, Xte, alpha=0.1) |
| 44 | + cov = np.mean((yte >= lo) & (yte <= hi)) |
| 45 | + # Marginal guarantee ~ 1 - alpha = 0.90 (finite-sample band). |
| 46 | + assert 0.87 <= cov <= 0.95 |
| 47 | + assert np.all(hi > lo) |
| 48 | + |
| 49 | + def test_coverage_decreases_with_alpha(self): |
| 50 | + (Xtr, ytr), (Xca, yca), (Xte, yte) = self._split() |
| 51 | + lo1, hi1, _ = sp.weighted_conformal_prediction( |
| 52 | + Xtr, ytr, Xca, yca, Xte, alpha=0.1 |
| 53 | + ) |
| 54 | + lo2, hi2, _ = sp.weighted_conformal_prediction( |
| 55 | + Xtr, ytr, Xca, yca, Xte, alpha=0.2 |
| 56 | + ) |
| 57 | + cov1 = np.mean((yte >= lo1) & (yte <= hi1)) |
| 58 | + cov2 = np.mean((yte >= lo2) & (yte <= hi2)) |
| 59 | + assert cov1 > cov2 |
| 60 | + assert 0.74 <= cov2 <= 0.86 # ~ 1 - 0.2 |
| 61 | + |
| 62 | + |
| 63 | +# --------------------------------------------------------------------------- |
| 64 | +# sp.conformal_ite_interval — ITE interval coverage |
| 65 | +# --------------------------------------------------------------------------- |
| 66 | +class TestConformalITEAnalytic: |
| 67 | + |
| 68 | + @staticmethod |
| 69 | + def _constant_effect_dgp(seed=0, n=4000, tau=3.0): |
| 70 | + rng = np.random.default_rng(seed) |
| 71 | + X = rng.normal(0, 1, (n, 4)) |
| 72 | + d = rng.integers(0, 2, n) |
| 73 | + y = X @ np.array([1, 0.5, -0.5, 0.2]) + tau * d + rng.normal(0, 1, n) |
| 74 | + df = pd.DataFrame(X, columns=[f"x{i}" for i in range(4)]) |
| 75 | + df["y"] = y |
| 76 | + df["d"] = d |
| 77 | + return df, [f"x{i}" for i in range(4)], tau |
| 78 | + |
| 79 | + def test_covers_known_constant_ite(self): |
| 80 | + df, covs, tau = self._constant_effect_dgp() |
| 81 | + res = sp.conformal_ite_interval( |
| 82 | + df, y="y", treat="d", covariates=covs, alpha=0.1, random_state=0 |
| 83 | + ) |
| 84 | + lo, hi = np.asarray(res.lower), np.asarray(res.upper) |
| 85 | + coverage = np.mean((lo <= tau) & (tau <= hi)) |
| 86 | + assert coverage >= 0.9 # ITE intervals are (conservatively) valid |
| 87 | + |
| 88 | + def test_interval_narrows_with_larger_alpha(self): |
| 89 | + df, covs, _ = self._constant_effect_dgp() |
| 90 | + w_tight = sp.conformal_ite_interval( |
| 91 | + df, y="y", treat="d", covariates=covs, alpha=0.1, random_state=0 |
| 92 | + ) |
| 93 | + w_loose = sp.conformal_ite_interval( |
| 94 | + df, y="y", treat="d", covariates=covs, alpha=0.2, random_state=0 |
| 95 | + ) |
| 96 | + width_tight = np.mean(np.asarray(w_tight.upper) - np.asarray(w_tight.lower)) |
| 97 | + width_loose = np.mean(np.asarray(w_loose.upper) - np.asarray(w_loose.lower)) |
| 98 | + assert width_tight > width_loose # smaller alpha -> wider interval |
| 99 | + |
| 100 | + def test_point_estimate_recovers_constant_cate(self): |
| 101 | + df, covs, tau = self._constant_effect_dgp() |
| 102 | + res = sp.conformal_ite_interval( |
| 103 | + df, y="y", treat="d", covariates=covs, alpha=0.1, random_state=0 |
| 104 | + ) |
| 105 | + assert np.mean(np.asarray(res.point)) == pytest.approx(tau, abs=0.6) |
0 commit comments