|
| 1 | +"""Tier D P2 known-truth upgrades — d-separation & do-calculus on canonical DAGs. |
| 2 | +
|
| 3 | +Part of the P1/P2 "Tier D analytic special-cases" campaign (see |
| 4 | +``.tierd_campaign/CAMPAIGN.md``). The Tier D probe surfaced a HIGH-severity |
| 5 | +correctness bug: ``_d_separated`` moralised the ancestral graph by marrying |
| 6 | +*siblings* (children of a common parent) instead of *co-parents* (parents of a |
| 7 | +common child), so it was wrong on every fork and collider — see |
| 8 | +``.tierd_campaign/BUG_d_separation_moralization.md``. Fixed per CLAUDE.md §12 |
| 9 | +(CHANGELOG + MIGRATION, ``⚠️ Correctness fix``). |
| 10 | +
|
| 11 | +These tests are the regression guard: the three canonical d-separation |
| 12 | +structures (chain / fork / collider) plus the back-door adjustment set and |
| 13 | +Rule 2 of Pearl's do-calculus. |
| 14 | +
|
| 15 | +Purely additive test file (the estimator fix is logged separately). |
| 16 | +""" |
| 17 | + |
| 18 | +import statspai as sp |
| 19 | + |
| 20 | + |
| 21 | +# --------------------------------------------------------------------------- |
| 22 | +# d-separation: chain / fork / collider canonical truths |
| 23 | +# --------------------------------------------------------------------------- |
| 24 | +class TestDSeparationAnalytic: |
| 25 | + |
| 26 | + def test_chain(self): |
| 27 | + # A -> B -> C: open unconditionally, blocked by the mediator B. |
| 28 | + g = sp.dag("A -> B; B -> C") |
| 29 | + assert g.d_separated("A", "C") is False |
| 30 | + assert g.d_separated("A", "C", {"B"}) is True |
| 31 | + |
| 32 | + def test_fork(self): |
| 33 | + # A <- M -> C: open unconditionally, blocked by the common cause M. |
| 34 | + g = sp.dag("M -> A; M -> C") |
| 35 | + assert g.d_separated("A", "C") is False |
| 36 | + assert g.d_separated("A", "C", {"M"}) is True |
| 37 | + |
| 38 | + def test_collider(self): |
| 39 | + # A -> K <- C: blocked unconditionally, OPENED by conditioning on the |
| 40 | + # collider K (this is the case the pre-fix moralisation got backwards). |
| 41 | + g = sp.dag("A -> K; C -> K") |
| 42 | + assert g.d_separated("A", "C") is True |
| 43 | + assert g.d_separated("A", "C", {"K"}) is False |
| 44 | + |
| 45 | + def test_collider_descendant_also_opens(self): |
| 46 | + # Conditioning on a descendant of a collider also opens the path. |
| 47 | + g = sp.dag("A -> K; C -> K; K -> D") |
| 48 | + assert g.d_separated("A", "C") is True |
| 49 | + assert g.d_separated("A", "C", {"D"}) is False |
| 50 | + |
| 51 | + |
| 52 | +# --------------------------------------------------------------------------- |
| 53 | +# back-door adjustment set + do-calculus Rule 2 |
| 54 | +# --------------------------------------------------------------------------- |
| 55 | +class TestBackdoorAndDoCalculus: |
| 56 | + |
| 57 | + @staticmethod |
| 58 | + def _confounded(): |
| 59 | + # W confounds X and Y; X -> Y is the effect of interest. |
| 60 | + return sp.dag("W -> X; W -> Y; X -> Y") |
| 61 | + |
| 62 | + def test_backdoor_adjustment_set_is_the_confounder(self): |
| 63 | + g = self._confounded() |
| 64 | + sets = g.adjustment_sets("X", "Y") |
| 65 | + assert {"W"} in sets |
| 66 | + assert g.backdoor_paths("X", "Y") # the X <- W -> Y path exists |
| 67 | + |
| 68 | + def test_rule2_swap_requires_adjustment(self): |
| 69 | + # do(X) can be swapped for observing X iff the back-door path is |
| 70 | + # blocked: applicable given {W}, not applicable with no adjustment. |
| 71 | + g = self._confounded() |
| 72 | + assert sp.do_rule2(g, Y="Y", X=[], Z="X", W=["W"]).applicable is True |
| 73 | + assert sp.do_rule2(g, Y="Y", X=[], Z="X", W=[]).applicable is False |
| 74 | + |
| 75 | + def test_unconfounded_needs_no_adjustment(self): |
| 76 | + # With no common cause, the empty set already identifies X -> Y. |
| 77 | + g = sp.dag("X -> Y") |
| 78 | + assert g.d_separated("X", "Y") is False # X causes Y |
| 79 | + assert set() in g.adjustment_sets("X", "Y") |
0 commit comments