Merge branch 'main' into test-coverage-json

Author: christopher-wild

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,12 @@ 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 +380,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

@@ -5,6 +5,30 @@
 from tests.test_helpers import create_temp_dir_if_non_existent, remove_temp_dir_if_existent
 
 
+class TestCausalDAGIssue90(unittest.TestCase):
+
+    """
+    Test the CausalDAG class for the resolution of Issue 90.
+    """
+
+    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; 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 = ["X"], ["Y"]
+        adjustment_sets = causal_dag.enumerate_minimal_adjustment_sets(xs, ys)
+        self.assertEqual([{"Z"}], adjustment_sets)
+
+    def tearDown(self) -> None:
+        remove_temp_dir_if_existent()
+
+
 class TestCausalDAG(unittest.TestCase):
 
     """