Merge pull request #72 from CITCOM-project/total-effect-id

Direct effect

Author: jmafoster1

causal_testing/specification/causal_dag.py

@@ -11,7 +11,6 @@
 from .scenario import Scenario
 from .variable import Output
 
-
 def list_all_min_sep(graph: nx.Graph, treatment_node: Node, outcome_node: Node,
                      treatment_node_set: set[Node], outcome_node_set: set[Node]):
     """ A backtracking algorithm for listing all minimal treatment-outcome separators in an undirected graph.
@@ -219,6 +218,56 @@ def get_ancestor_graph(self, treatments: list[str], outcomes: list[str]) -> Caus
         ancestor_graph.graph.remove_nodes_from(variables_to_remove)
         return ancestor_graph
 
+    def get_indirect_graph(self, treatments:list[str], outcomes:list[str]) -> CausalDAG:
+        """
+        This is the counterpart of the back-door graph for direct effects. We remove only edges pointing from X to Y.
+        It is a Python implementation of the indirectGraph function from Dagitty.
+
+        :param list[str] treatments: List of treatment names.
+        :param list[str] outcomes: List of outcome names.
+        :return: The indirect graph with edges pointing from X to Y removed.
+        :rtype: CausalDAG
+        """
+        gback = self.copy()
+        ee = []
+        for s in treatments:
+            for t in outcomes:
+                if (s, t) in gback.graph.edges:
+                    ee.append((s, t))
+        for v1, v2 in ee:
+            gback.graph.remove_edge(v1,v2)
+        return gback
+
+    def direct_effect_adjustment_sets(self, treatments:list[str], outcomes:list[str]) -> list[set[str]]:
+        """
+        Get the smallest possible set of variables that blocks all back-door paths between all pairs of treatments
+        and outcomes for DIRECT causal effect.
+
+        This is an Python implementation of the listMsasTotalEffect function from Dagitty using Algorithms presented in
+        Adjustment Criteria in Causal Diagrams: An Algorithmic Perspective, Textor and Lískiewicz, 2012 and extended in
+        Separators and adjustment sets in causal graphs: Complete criteria and an algorithmic framework, Zander et al.,
+        2019. These works use the algorithm presented by Takata et al. in their work entitled: Space-optimal,
+        backtracking algorithms to list the minimal vertex separators of a graph, 2013.
+
+        :param list[str] treatments: List of treatment names.
+        :param list[str] outcomes: List of outcome names.
+        :return: A list of possible adjustment sets.
+        :rtype: list[set[str]]
+        """
+
+        indirect_graph = self.get_indirect_graph(treatments, outcomes)
+        ancestor_graph = indirect_graph.get_ancestor_graph(treatments, outcomes)
+        gam =  nx.moral_graph(ancestor_graph.graph)
+
+        edges_to_add = [("TREATMENT", treatment) for treatment in treatments]
+        edges_to_add += [("OUTCOME", outcome) for outcome in outcomes]
+        gam.add_edges_from(edges_to_add)
+
+        min_seps = list(list_all_min_sep(gam, "TREATMENT", "OUTCOME", set(treatments), set(outcomes)))
+        # min_seps.remove(set(outcomes))
+        return min_seps
+
+
     def enumerate_minimal_adjustment_sets(self, treatments: list[str], outcomes: list[str]) -> list[set[str]]:
         """ Get the smallest possible set of variables that blocks all back-door paths between all pairs of treatments
         and outcomes.

causal_testing/testing/causal_test_case.py

@@ -18,7 +18,7 @@ class CausalTestCase:
 
     def __init__(self, control_input_configuration: dict[Variable: Any], expected_causal_effect: CausalTestOutcome,
                  outcome_variables: dict[Variable], treatment_input_configuration: dict[Variable: Any] = None,
-                 estimate_type: str = "ate", effect_modifier_configuration: dict[Variable: Any] = None):
+                 estimate_type: str = "ate", effect_modifier_configuration: dict[Variable: Any] = None, effect: str = "total"):
         """
         When a CausalTestCase is initialised, it takes the intervention and applies it to the input configuration to
         create two distinct input configurations: a control input configuration and a treatment input configuration.
@@ -35,6 +35,7 @@ def __init__(self, control_input_configuration: dict[Variable: Any], expected_ca
         self.outcome_variables = outcome_variables
         self.treatment_input_configuration = treatment_input_configuration
         self.estimate_type = estimate_type
+        self.effect = effect
         if effect_modifier_configuration:
             self.effect_modifier_configuration = effect_modifier_configuration
         else:

causal_testing/testing/causal_test_engine.py

@@ -59,10 +59,20 @@ def load_data(self, **kwargs):
 
         self.scenario_execution_data_df = self.data_collector.collect_data(**kwargs)
 
-        minimal_adjustment_sets = self.casual_dag.enumerate_minimal_adjustment_sets(
-                [v.name for v in self.treatment_variables],
-                [v.name for v in self.causal_test_case.outcome_variables]
-            )
+        minimal_adjustment_sets = []
+        if self.causal_test_case.effect == "total":
+            minimal_adjustment_sets = self.casual_dag.enumerate_minimal_adjustment_sets(
+                    [v.name for v in self.treatment_variables],
+                    [v.name for v in self.causal_test_case.outcome_variables]
+                )
+        elif self.causal_test_case.effect == "direct":
+            minimal_adjustment_sets = self.casual_dag.direct_effect_adjustment_sets(
+                    [v.name for v in self.treatment_variables],
+                    [v.name for v in self.causal_test_case.outcome_variables]
+                )
+        else:
+            raise ValueError("Causal effect should be 'total' or 'direct'")
+
         minimal_adjustment_set = min(minimal_adjustment_sets, key=len)
         return minimal_adjustment_set
 

causal_testing/testing/causal_test_outcome.py

@@ -118,6 +118,14 @@ def apply(self, res: CausalTestResult) -> bool:
         # TODO: confidence intervals?
         return res.ate < 0
 
+class SomeEffect(CausalTestOutcome):
+    """An extension of TestOutcome representing that the expected causal effect should not be zero."""
+
+    def apply(self, res: CausalTestResult) -> bool:
+        return (0 < res.ci_low() < res.ci_high()) or (res.ci_low() < res.ci_high() < 0)
+
+    def __str__(self):
+        return "Changed"
 
 class NoEffect(CausalTestOutcome):
     """An extension of TestOutcome representing that the expected causal effect should be zero."""

tests/specification_tests/test_causal_dag.py

@@ -64,6 +64,32 @@ def tearDown(self) -> None:
         remove_temp_dir_if_existent()
 
 
+class TestDAGDirectEffectIdentification(unittest.TestCase):
+    """
+    Test the Causal DAG identification algorithms and supporting algorithms.
+    """
+
+    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 G { X1->X2;X2->V;X2->D1;X2->D2;D1->Y;D1->D2;Y->D3;Z->X2;Z->Y;}"""
+        )
+        f = open(self.dag_dot_path, "w")
+        f.write(dag_dot)
+        f.close()
+
+    def test_direct_effect_adjustment_sets(self):
+        causal_dag = CausalDAG(self.dag_dot_path)
+        adjustment_sets = causal_dag.direct_effect_adjustment_sets(["X1"], ["Y"])
+        self.assertEqual(list(adjustment_sets), [{"Y"}, {"D1", "Z"}, {"X2", "Z"}])
+
+    def test_direct_effect_adjustment_sets_no_adjustment(self):
+        causal_dag = CausalDAG(self.dag_dot_path)
+        adjustment_sets = causal_dag.direct_effect_adjustment_sets(["X2"], ["D1"])
+        self.assertEqual(list(adjustment_sets), [set()])
+
+
 class TestDAGIdentification(unittest.TestCase):
     """
     Test the Causal DAG identification algorithms and supporting algorithms.
@@ -79,6 +105,15 @@ def setUp(self) -> None:
         f.write(dag_dot)
         f.close()
 
+    def test_get_indirect_graph(self):
+        causal_dag = CausalDAG(self.dag_dot_path)
+        indirect_graph = causal_dag.get_indirect_graph(["D1"], ["Y"])
+        original_edges = list(causal_dag.graph.edges)
+        original_edges.remove(("D1", "Y"))
+        self.assertEqual(list(indirect_graph.graph.edges), original_edges)
+        self.assertEqual(indirect_graph.graph.nodes, causal_dag.graph.nodes)
+
+
     def test_proper_backdoor_graph(self):
         """Test whether converting a Causal DAG to a proper back-door graph works correctly."""
         causal_dag = CausalDAG(self.dag_dot_path)

tests/testing_tests/test_causal_test_engine.py

@@ -141,6 +141,26 @@ def test_execute_test_observational_causal_forest_estimator(self):
         causal_test_result = self.causal_test_engine.execute_test(estimation_model)
         self.assertAlmostEqual(causal_test_result.ate, 4, delta=1)
 
+
+    def test_invalid_causal_effect(self):
+        """ Check that executing the causal test case returns the correct results for dummy data using a linear
+        regression estimator. """
+        causal_test_case = CausalTestCase(
+            control_input_configuration={self.A: 0},
+            expected_causal_effect=self.expected_causal_effect,
+            treatment_input_configuration={self.A: 1},
+            outcome_variables={self.C},
+            effect="error")
+        # 5. Create causal test engine
+        causal_test_engine = CausalTestEngine(
+            causal_test_case,
+            self.causal_specification,
+            self.data_collector
+        )
+        with self.assertRaises(Exception):
+            causal_test_engine.load_data()
+
+
     def test_execute_test_observational_linear_regression_estimator(self):
         """ Check that executing the causal test case returns the correct results for dummy data using a linear
         regression estimator. """
@@ -153,6 +173,37 @@ def test_execute_test_observational_linear_regression_estimator(self):
         causal_test_result = self.causal_test_engine.execute_test(estimation_model)
         self.assertAlmostEqual(causal_test_result.ate, 4, delta=1e-10)
 
+
+    def test_execute_test_observational_linear_regression_estimator_direct_effect(self):
+        """ Check that executing the causal test case returns the correct results for dummy data using a linear
+        regression estimator. """
+        causal_test_case = CausalTestCase(
+            control_input_configuration={self.A: 0},
+            expected_causal_effect=self.expected_causal_effect,
+            treatment_input_configuration={self.A: 1},
+            outcome_variables={self.C},
+            effect="direct")
+
+        # 5. Create causal test engine
+        causal_test_engine = CausalTestEngine(
+            causal_test_case,
+            self.causal_specification,
+            self.data_collector
+        )
+        self.minimal_adjustment_set = causal_test_engine.load_data()
+
+        # 6. Easier to access treatment and outcome values
+        self.treatment_value = 1
+        self.control_value = 0
+        estimation_model = LinearRegressionEstimator(('A',),
+                                                     self.treatment_value,
+                                                     self.control_value,
+                                                     self.minimal_adjustment_set,
+                                                     ('C',),
+                                                     causal_test_engine.scenario_execution_data_df)
+        causal_test_result = causal_test_engine.execute_test(estimation_model)
+        self.assertAlmostEqual(causal_test_result.ate, 0, delta=1e-10)
+
     def test_execute_test_observational_linear_regression_estimator_risk_ratio(self):
         """ Check that executing the causal test case returns the correct results for dummy data using a linear
         regression estimator. """

tests/testing_tests/test_causal_test_outcome.py

@@ -1,5 +1,5 @@
 import unittest
-from causal_testing.testing.causal_test_outcome import CausalTestResult, ExactValue
+from causal_testing.testing.causal_test_outcome import CausalTestResult, ExactValue, SomeEffect
 
 class TestCausalTestOutcome(unittest.TestCase):
     """ Test the TestCausalTestOutcome basic methods.
@@ -28,3 +28,19 @@ def test_exactValue_fail(self):
                      confidence_intervals = None, effect_modifier_configuration = None)
         ev = ExactValue(5, 0.1)
         self.assertFalse(ev.apply(ctr))
+
+
+    def test_someEffect_pass(self):
+        ctr = CausalTestResult(treatment="A", outcome="A", treatment_value=1,
+                     control_value=0, adjustment_set={}, ate=5.05,
+                     confidence_intervals = [4.8, 6.7], effect_modifier_configuration = None)
+        ev = SomeEffect()
+        self.assertTrue(ev.apply(ctr))
+
+
+    def test_someEffect_fail(self):
+        ctr = CausalTestResult(treatment="A", outcome="A", treatment_value=1,
+                     control_value=0, adjustment_set={}, ate=0,
+                     confidence_intervals = [-0.1, 0.2], effect_modifier_configuration = None)
+        ev = SomeEffect()
+        self.assertFalse(ev.apply(ctr))