Testing IV assumptions

Author: jmafoster1

causal_testing/specification/causal_dag.py

@@ -138,6 +138,36 @@ def __init__(self, dot_path: str = None, **attr):
         if not self.is_acyclic():
             raise nx.HasACycle("Invalid Causal DAG: contains a cycle.")
 
+    def check_iv_assumptions(self, treatment, outcome, instrument) -> bool:
+        """
+        Checks the three instrumental variable assumptions, raising a
+        ValueError if any are violated.
+
+        :return Boolean True if the three IV assumptions hold.
+        """
+        # (i) Instrument is associated with treatment
+        if nx.d_separated(self.graph, {instrument}, {treatment}, set()):
+            raise ValueError(f"Instrument {instrument} is not associated with treatment {treatment} in the DAG")
+
+        # (ii) Instrument does not affect outcome except through its potential effect on treatment
+        if not all([treatment in path for path in nx.all_simple_paths(self.graph, source=instrument, target=outcome)]):
+            raise ValueError(
+                f"Instrument {instrument} affects the outcome {outcome} other than through the treatment {treatment}"
+            )
+
+        # (iii) Instrument and outcome do not share causes
+        if any(
+            [
+                cause
+                for cause in self.graph.nodes
+                if list(nx.all_simple_paths(self.graph, source=cause, target=instrument))
+                and list(nx.all_simple_paths(self.graph, source=cause, target=outcome))
+            ]
+        ):
+            raise ValueError(f"Instrument {instrument} and outcome {outcome} share common causes")
+
+        return True
+
     def add_edge(self, u_of_edge: Node, v_of_edge: Node, **attr):
         """Add an edge to the causal DAG.
 

tests/specification_tests/test_causal_dag.py

@@ -29,6 +29,38 @@ def tearDown(self) -> None:
         remove_temp_dir_if_existent()
 
 
+class TestIVAssumptions(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 G { I -> X; X -> Y; U -> X; U -> Y;}"""
+        f = open(self.dag_dot_path, "w")
+        f.write(dag_dot)
+        f.close()
+
+    def test_valid_iv(self):
+        causal_dag = CausalDAG(self.dag_dot_path)
+        self.assertTrue(causal_dag.check_iv_assumptions("X", "Y", "I"))
+
+    def test_unrelated_instrument(self):
+        causal_dag = CausalDAG(self.dag_dot_path)
+        causal_dag.graph.remove_edge("I", "X")
+        with self.assertRaises(ValueError):
+            causal_dag.check_iv_assumptions("X", "Y", "I")
+
+    def test_direct_cause(self):
+        causal_dag = CausalDAG(self.dag_dot_path)
+        causal_dag.graph.add_edge("I", "Y")
+        with self.assertRaises(ValueError):
+            causal_dag.check_iv_assumptions("X", "Y", "I")
+
+    def test_common_cause(self):
+        causal_dag = CausalDAG(self.dag_dot_path)
+        causal_dag.graph.add_edge("U", "I")
+        with self.assertRaises(ValueError):
+            causal_dag.check_iv_assumptions("X", "Y", "I")
+
+
 class TestCausalDAG(unittest.TestCase):
 
     """