Apply constructive back-door criterion to minimal separators

Author: AndrewC19

causal_testing/specification/causal_dag.py

@@ -160,9 +160,9 @@ def is_acyclic(self) -> bool:
     def get_proper_backdoor_graph(self, treatments: list[str], outcomes: list[str]) -> CausalDAG:
         """Convert the causal DAG to a proper back-door graph.
 
-        A proper back-door graph of a causal DAG is obtained by
-        removing the first edge of every proper causal path from treatments to outcomes. A proper causal path from
-        X to Y is a path of directed edges that starts from X and ends in Y.
+        A proper back-door graph of a causal DAG is obtained by removing the first edge of every proper causal path from
+        treatments to outcomes. A proper causal path from X to Y is a path of directed edges that starts from X and ends
+        in Y.
 
         Reference: (Separators and adjustment sets in causal graphs: Complete criteria and an algorithmic framework,
         Zander et al.,  2019, Definition 3, p.15)
@@ -311,7 +311,10 @@ def enumerate_minimal_adjustment_sets(self, treatments: list[str], outcomes: lis
                 outcome_node_set,
             )
         )
-        return minimum_adjustment_sets
+        valid_minimum_adjustment_sets = [adj for adj in minimum_adjustment_sets
+                                         if self.constructive_backdoor_criterion(proper_backdoor_graph, treatments,
+                                                                                 outcomes, adj)]
+        return valid_minimum_adjustment_sets
 
     def adjustment_set_is_minimal(self, treatments: list[str], outcomes: list[str], adjustment_set: set[str]) -> bool:
         """Given a list of treatments X, a list of outcomes Y, and an adjustment set Z, determine whether Z is the
@@ -375,22 +378,23 @@ def constructive_backdoor_criterion(
         """
         # Condition (1)
         proper_causal_path_vars = self.proper_causal_pathway(treatments, outcomes)
-        descendents_of_proper_casual_paths = set.union(
-            *[
-                set.union(
-                    nx.descendants(self.graph, proper_causal_path_var),
-                    {proper_causal_path_var},
-                )
-                for proper_causal_path_var in proper_causal_path_vars
-            ]
-        )
-
-        if not set(covariates).issubset(set(self.graph.nodes).difference(descendents_of_proper_casual_paths)):
-            logger.info(
-                f"Failed Condition 1: Z={covariates} **is** a descendent of some variable on a proper causal "
-                f"path between X={treatments} and Y={outcomes}."
+        if proper_causal_path_vars:
+            descendents_of_proper_casual_paths = set.union(
+                *[
+                    set.union(
+                        nx.descendants(self.graph, proper_causal_path_var),
+                        {proper_causal_path_var},
+                    )
+                    for proper_causal_path_var in proper_causal_path_vars
+                ]
             )
-            return False
+
+            if not set(covariates).issubset(set(self.graph.nodes).difference(descendents_of_proper_casual_paths)):
+                logger.info(
+                    f"Failed Condition 1: Z={covariates} **is** a descendent of some variable on a proper causal "
+                    f"path between X={treatments} and Y={outcomes}."
+                )
+                return False
 
         # Condition (2)
         if not nx.d_separated(proper_backdoor_graph.graph, set(treatments), set(outcomes), set(covariates)):

tests/specification_tests/test_causal_dag.py

@@ -4,6 +4,7 @@
 from causal_testing.specification.causal_dag import CausalDAG, close_separator, list_all_min_sep
 from tests.test_helpers import create_temp_dir_if_non_existent, remove_temp_dir_if_existent
 
+
 class TestCausalDAGIssue90(unittest.TestCase):
 
     """
@@ -13,16 +14,16 @@ class TestCausalDAGIssue90(unittest.TestCase):
     def setUp(self) -> None:
         temp_dir_path = create_temp_dir_if_non_existent()
         self.dag_dot_path = os.path.join(temp_dir_path, "dag.dot")
-        dag_dot = """digraph DAG { rankdir=LR; content -> weight; weight -> S3; country -> S3; country -> distance; content -> S3; plane_transport -> S1; plane_transport -> S2; S1 -> alarm; S2 -> alarm; S3 -> alarm; }"""
+        dag_dot = """digraph DAG { rankdir=LR; Z -> X; X -> M; M -> Y; Z -> M; }"""
         with open(self.dag_dot_path, "w") as f:
             f.write(dag_dot)
 
     def test_enumerate_minimal_adjustment_sets(self):
         """Test whether enumerate_minimal_adjustment_sets lists all possible minimum sized adjustment sets."""
         causal_dag = CausalDAG(self.dag_dot_path)
-        xs, ys = ["weight"], ["alarm"]
+        xs, ys = ["X"], ["Y"]
         adjustment_sets = causal_dag.enumerate_minimal_adjustment_sets(xs, ys)
-        self.assertEqual([{"content"}], adjustment_sets)
+        self.assertEqual([{"Z"}], adjustment_sets)
 
     def tearDown(self) -> None:
         remove_temp_dir_if_existent()