Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
22 changes: 22 additions & 0 deletions README.md
Original file line number Diff line number Diff line change
Expand Up @@ -370,6 +370,28 @@ sound but complete only under faithfulness, it is a warning rather than a proof
`admissible` means only that neither failure mode was detected. It is not a certificate
that adjusting for the covariate yields an unbiased estimate.

There is a third case, and getting it wrong is how a check becomes decoration. Path opening
asks whether conditioning *breaks* a separation that held — so when the target and the
outcome are already d-connected in the back-door graph, which is what any unmeasured common
cause produces, there is no separation to break and the question has no answer. The report
says so rather than reporting silence as a clean bill of health:

```text
Conditioning around do(knockdown) with outcome 'IFNG':

'knockdown' and 'IFNG' are d-connected in the back-door graph given nothing:
there is an open back-door path, so the effect is confounded before any covariate is chosen.
Closes it: ['donor']
For the rest, no path-level verdict is reachable -- reported as undecided, not as clean.
```

`report.back_door_open` carries the graph-level finding, `report.blocks_path` names the
covariates that *close* the open path — the reason adjustment sets exist, and something a
check that only ever asks "does this open a path?" cannot express — and `report.undecided`
names the rest. A covariate whose verdict was not reachable is never reported as admissible.
A covariate that is not in `observed_set` is never admissible either, however well it does
in the graph: the graph does not know what the assay measured.

Both findings are checked against exact ground truth over 120 generated models — 2,596
verdicts. Post-treatment is structural, so it is verified in *both* directions against a
reachability walk written independently of the library's. Path-opening is verified in the
Expand Down
137 changes: 133 additions & 4 deletions src/causal_hypergraphs/identification/covariates.py
Original file line number Diff line number Diff line change
Expand Up @@ -42,6 +42,30 @@ class CovariateVerdict:
post_treatment: bool
opens_path: bool
reason: str
blocks_path: bool = False
"""Conditioning on it *closes* a back-door path that was open.

The positive finding, and the reason adjustment sets exist. Only reachable when the
back-door path is open to begin with, which is why the old code -- which asked solely
whether a covariate opens a path -- could not express it.
"""
observed: bool = True
"""Whether the covariate is in the model's observed set.

Unobserved is a *structural* refusal, not an undecided one -- you cannot condition on
what was not measured, and no further evidence would change that. It shares
`path_test_applicable=False` with the undecided case because neither reached a path
verdict, which is what that flag is for; the two are separated here so the report does
not file a hard fact under a heading that says the question was unanswerable.
"""
path_test_applicable: bool = True
"""Whether a path-level verdict was actually reached for this covariate.

False means the question could not be decided here, not that it was decided
favourably, and `admissible` is never True alongside it. Distinguishing the two is the
whole repair: a check that runs, passes, and cannot report is worse than no check,
because it is read as a clean bill of health.
"""

def __str__(self) -> str:
mark = "ok " if self.admissible else "!! "
Expand All @@ -55,11 +79,32 @@ class CovariateReport:
target: str
outcome: str
verdicts: tuple[CovariateVerdict, ...]
back_door_open: bool = False
"""Whether target and outcome are d-connected in the back-door graph given nothing.

A fact about the *query*, not about any covariate, and the dominant one when true:
some non-causal path is open, so the effect is confounded until something closes it.
It is reported on the report rather than repeated per covariate because that is where
it belongs, and because leaving it unsaid is what made the old output read as clean.
"""

@property
def admissible(self) -> tuple[str, ...]:
return tuple(v.covariate for v in self.verdicts if v.admissible)

@property
def blocks_path(self) -> tuple[str, ...]:
return tuple(v.covariate for v in self.verdicts if v.blocks_path)

@property
def undecided(self) -> tuple[str, ...]:
"""Covariates for which no path-level verdict was reached."""
return tuple(
v.covariate
for v in self.verdicts
if not v.path_test_applicable and not v.post_treatment and v.observed
)

@property
def post_treatment(self) -> tuple[str, ...]:
return tuple(v.covariate for v in self.verdicts if v.post_treatment)
Expand All @@ -71,11 +116,31 @@ def opens_path(self) -> tuple[str, ...]:
def summary(self) -> str:
lines = [
f"Conditioning around do({self.target}) with outcome {self.outcome!r}:",
]
if self.back_door_open:
closers = list(self.blocks_path)
lines.extend([
"",
f" {self.target!r} and {self.outcome!r} are d-connected in the back-door "
"graph given nothing:",
" there is an open back-door path, so the effect is confounded before any "
"covariate is chosen.",
f" Closes it: {closers}" if closers
else " No candidate offered closes it.",
" For the rest, no path-level verdict is reachable -- reported as "
"undecided, not as clean.",
])
lines.extend([
"",
" Structural -- post-treatment, no distributional assumption involved:",
]
])
blocked = [v for v in self.verdicts if v.post_treatment or not v.admissible]
structural = [v for v in blocked if v.post_treatment or not v.opens_path]
undecided = set(self.undecided)
structural = [
v
for v in blocked
if (v.post_treatment or not v.opens_path) and v.covariate not in undecided
]
lines.extend(f" {v}" for v in structural) if structural else lines.append(
" (none)"
)
Expand All @@ -86,6 +151,13 @@ def summary(self) -> str:
)
warnings = [v for v in self.verdicts if v.opens_path and not v.post_treatment]
lines.extend(f" {v}" for v in warnings) if warnings else lines.append(" (none)")
if self.undecided:
lines.append("")
lines.append(
" Undecided -- the path test could not reach a verdict for these; this is "
"not a clean result:"
)
lines.extend(f" {v}" for v in self.verdicts if v.covariate in self.undecided)
lines.append("")
lines.append(f" Admissible: {list(self.admissible) or '(none)'}")
return "\n".join(lines)
Expand Down Expand Up @@ -207,7 +279,59 @@ def check_covariates(
)
continue

opens = separated_alone and not d_separated(back_door, target, outcome, (name,))
if name not in graph.observed_set:
# A latent variable can be the graphically ideal covariate -- W below closes
# the back-door path perfectly -- and you still cannot condition on it. The
# graph does not know what the assay measured, so this is checked separately
# rather than inferred.
verdicts.append(
CovariateVerdict(
covariate=name,
admissible=False,
post_treatment=False,
opens_path=False,
observed=False,
path_test_applicable=False,
reason=(
"not observed in this model, so it cannot be conditioned on "
"whatever the graph says about it."
),
)
)
continue

given = d_separated(back_door, target, outcome, (name,))
if not separated_alone:
# The back-door path is already open, so "does conditioning OPEN one" has no
# verdict to give -- the old code silently answered "no" here and called the
# covariate admissible. The answerable question is whether it CLOSES the path.
closes = given
verdicts.append(
CovariateVerdict(
covariate=name,
admissible=closes,
post_treatment=False,
opens_path=False,
blocks_path=closes,
path_test_applicable=closes,
reason=(
f"closes an open back-door path: conditioning on it d-separates "
f"{target!r} from {outcome!r} in the graph with {target!r}'s "
"outgoing edges severed. Rests on faithfulness in the same way the "
"opening warning does."
)
if closes
else (
f"undecided: {target!r} and {outcome!r} are already d-connected in "
"the back-door graph, so no covariate can be cleared of opening a "
"path here, and this one does not close the open path either. Not "
"a clean result -- the question was not answerable, not answered."
),
)
)
continue

opens = not given
verdicts.append(
CovariateVerdict(
covariate=name,
Expand All @@ -228,4 +352,9 @@ def check_covariates(
)
)

return CovariateReport(target=target, outcome=outcome, verdicts=tuple(verdicts))
return CovariateReport(
target=target,
outcome=outcome,
verdicts=tuple(verdicts),
back_door_open=not separated_alone,
)
185 changes: 185 additions & 0 deletions tests/test_covariate_vacuity.py
Original file line number Diff line number Diff line change
@@ -0,0 +1,185 @@
"""A check that runs, passes, and cannot report is the failure this module names.

`check_covariates` decided path-opening as

separated_alone = d_separated(back_door, target, outcome) # computed once
opens = separated_alone and not d_separated(back_door, target, outcome, (name,))

so whenever the target and outcome were *already* d-connected in the back-door graph --
which is what an unmeasured common cause produces, and therefore the normal case in any
real system -- `opens` was False for every covariate and the report read

Admissible: ['A', 'B', 'C', 'W']

with both finding sections empty. Three things were wrong at once, and the second is the
one that makes it more than cosmetic:

1. Nothing distinguished "tested and clean" from "the test could not fire".
2. A collider the module flags correctly on a clean graph goes **unflagged** the moment a
confounder exists. The finding is not merely unstated, it is lost.
3. `W` is *unobserved*. Covariates were validated against `graph.variable_set`, never
`graph.observed_set`, so the report advised conditioning on a latent variable.

The module's own docstring warned about exactly this shape for the name-clash case --
"skipping the path-opening test would leave a check that runs, passes, and cannot report
-- every covariate would come back admissible for the wrong reason" -- and the guard was
installed there and not here.
"""

from __future__ import annotations

import pytest

from causal_hypergraphs import DeleteMechanism, MechanismGraph, check_covariates

QUERY = DeleteMechanism("m_T", outcomes=("Y",))
BASE = {
"m_A": {"inputs": (), "outputs": ("A",)},
"m_B": {"inputs": (), "outputs": ("B",)},
"m_C": {"inputs": ("A", "B"), "outputs": ("C",)},
}


def clean_graph() -> MechanismGraph:
"""A -> m_T, and a collider C with parents A and B; no open back-door path."""
return MechanismGraph(
variables={"A", "B", "C", "T", "Y"},
mechanisms={
**BASE,
"m_T": {"inputs": ("A",), "outputs": ("T",)},
"m_Y": {"inputs": ("T", "B"), "outputs": ("Y",)},
},
)


def confounded_graph(*, w_observed: bool) -> MechanismGraph:
"""The same graph plus W, a common cause of the target and the outcome."""
observed = {"A", "B", "C", "T", "Y"} | ({"W"} if w_observed else set())
return MechanismGraph(
variables={"A", "B", "C", "T", "Y", "W"},
mechanisms={
**BASE,
"m_W": {"inputs": (), "outputs": ("W",)},
"m_T": {"inputs": ("A", "W"), "outputs": ("T",)},
"m_Y": {"inputs": ("T", "B", "W"), "outputs": ("Y",)},
},
observed_variables=observed,
)


def verdict(report, name: str):
(found,) = [v for v in report.verdicts if v.covariate == name]
return found


def test_the_collider_is_still_flagged_when_the_back_door_is_clean() -> None:
"""The behaviour that was already right, pinned so the repair cannot cost it."""
report = check_covariates(clean_graph(), QUERY, "Y", ("A", "B", "C"))

assert not report.back_door_open
assert report.opens_path == ("C",)
assert report.admissible == ("A", "B")


def test_a_confounder_no_longer_hides_the_collider() -> None:
"""Defect 2, and the reason this is not cosmetic: the finding was LOST, not unstated.

`C` is the same collider as above. Adding an unmeasured common cause elsewhere in the
graph must not silence a verdict about `C`.
"""
report = check_covariates(confounded_graph(w_observed=False), QUERY, "Y", ("A", "B", "C"))

assert report.back_door_open
assert "C" not in report.admissible


def test_an_open_back_door_is_reported_and_not_read_as_clean() -> None:
"""Defect 1. `admissible` must not be asserted on a test that could not run."""
report = check_covariates(confounded_graph(w_observed=False), QUERY, "Y", ("A", "B", "C"))

assert report.back_door_open
for name in ("A", "B", "C"):
found = verdict(report, name)
assert not found.path_test_applicable
assert not found.admissible
assert report.admissible == ()
text = report.summary()
assert "open back-door path" in text
assert "Admissible: (none)" in text or "Admissible: []" in text


def test_a_covariate_that_closes_the_back_door_is_named() -> None:
"""The previously-vacuous branch now answers the question a user actually has.

`W` is the confounder. Conditioning on it *closes* the open back-door path, which is
the entire purpose of covariate adjustment -- and the old code, which only ever asked
whether a covariate OPENS a path, could not express it.
"""
report = check_covariates(confounded_graph(w_observed=True), QUERY, "Y", ("A", "B", "C", "W"))

assert report.back_door_open
assert report.blocks_path == ("W",)
assert verdict(report, "W").admissible
assert not verdict(report, "A").admissible
assert "closes" in verdict(report, "W").reason


def test_an_unobserved_covariate_is_never_admissible() -> None:
"""Defect 3. `W` is latent here; you cannot condition on what you did not measure.

It closes the back-door path in the graph, so a purely graphical check calls it the
ideal covariate. That is precisely why observation has to be checked separately: the
graph cannot tell you what the assay measured.
"""
report = check_covariates(confounded_graph(w_observed=False), QUERY, "Y", ("W",))

found = verdict(report, "W")
assert not found.admissible
assert not found.path_test_applicable
assert "not observed" in found.reason
assert report.blocks_path == ()


def test_post_treatment_still_outranks_everything() -> None:
"""A descendant of the intervention is refused whatever the back-door graph says."""
graph = MechanismGraph(
variables={"A", "B", "C", "T", "Y", "W", "M"},
mechanisms={
**BASE,
"m_W": {"inputs": (), "outputs": ("W",)},
"m_T": {"inputs": ("A", "W"), "outputs": ("T",)},
"m_M": {"inputs": ("T",), "outputs": ("M",)},
"m_Y": {"inputs": ("T", "B", "W"), "outputs": ("Y",)},
},
observed_variables={"A", "B", "C", "T", "Y", "M"},
)
report = check_covariates(graph, DeleteMechanism("m_T", outcomes=("Y",)), "Y", ("M",))

found = verdict(report, "M")
assert found.post_treatment
assert not found.admissible
assert "Structural" in found.reason


def test_the_report_leads_with_the_graph_level_finding() -> None:
"""An open back-door is a fact about the query, not about any one covariate."""
confounded = check_covariates(confounded_graph(w_observed=True), QUERY, "Y", ("A", "W"))
clean = check_covariates(clean_graph(), QUERY, "Y", ("A", "B"))

assert "open back-door path" in confounded.summary()
assert "open back-door path" not in clean.summary()
# And the closing covariate is surfaced, not buried in the per-covariate reasons.
assert "W" in confounded.summary()


@pytest.mark.parametrize("w_observed", [True, False])
def test_admissible_never_includes_a_covariate_whose_test_could_not_run(
w_observed: bool,
) -> None:
"""The invariant the whole repair exists to establish, stated once, directly."""
report = check_covariates(
confounded_graph(w_observed=w_observed), QUERY, "Y", ("A", "B", "C", "W")
)
for found in report.verdicts:
if not found.path_test_applicable:
assert not found.admissible, found.covariate
Loading