More tests to make codecov happy

Author: jmafoster1

causal_testing/causal_testing_framework.py

@@ -7,6 +7,7 @@
 from importlib.metadata import entry_points
 from pathlib import Path
 
+import numpy as np
 import pandas as pd
 from tqdm import tqdm
 
@@ -43,8 +44,6 @@ def read_dataframe(file_path: str, **kwargs: dict) -> pd.DataFrame:
     suffix = Path(file_path).suffix.lower()
 
     if suffix in readers:
-        print("READING FROM", file_path, kwargs)
-        print(readers[suffix](file_path, **kwargs))
         return readers[suffix](file_path, **kwargs)
     raise ValueError(f"Unsupported file extension: '{suffix}'")
 
@@ -242,16 +241,18 @@ def evaluate_dag(self, bootstrap_size: bool = 100, alpha: float = 0.05) -> pd.Se
             for test_case in self.test_cases:
                 if test_case.skip:
                     continue
-                effect_estimate = test_case.estimate_effect(
-                    df=self.df.sample(len(self.df), replace=True, random_state=sample_index)
-                )
+                try:
+                    effect_estimate = test_case.estimate_effect(
+                        df=self.df.sample(len(self.df), replace=True, random_state=sample_index)
+                    )
+                except (np.linalg.LinAlgError, ValueError):
+                    test_outcomes[TestOutcome.INESTIMABLE] += 1
+
                 if effect_estimate:
                     if test_case.expected_causal_effect.apply(effect_estimate):
                         test_outcomes[TestOutcome.PASS] += 1
                     else:
                         test_outcomes[TestOutcome.FAIL] += 1
-                else:
-                    test_outcomes[TestOutcome.INESTIMABLE] += 1
             sample_results.append(test_outcomes)
 
         sample_results = pd.DataFrame(sample_results)
@@ -264,7 +265,6 @@ def evaluate_dag(self, bootstrap_size: bool = 100, alpha: float = 0.05) -> pd.Se
             results[f"{outcome.name}_ci_low"] = data[ci_low_inx]
             results[f"{outcome.name}_ci_high"] = data[ci_high_inx]
 
-        print(results)
         return pd.Series(results).sort_index()
 
     def save_results(self, output_path) -> list:

causal_testing/specification/causal_dag.py

@@ -512,11 +512,12 @@ def identification(
         :return: The smallest set of variables which can be adjusted for to obtain a causal
                  estimate as opposed to a purely associational estimate.
         """
-        # Naive method to guarantee termination when we have cycles
+        nodes_to_ignore = set(nodes_to_ignore) if nodes_to_ignore is not None else set()
+
         if self.ignore_cycles:
-            return set(self.predecessors(treatment_variable))
-        minimal_adjustment_sets = []
-        if effect_type == "total":
+            # Naive method to guarantee termination when we have cycles
+            minimal_adjustment_sets = [set(self.predecessors(treatment_variable))]
+        elif effect_type == "total":
             minimal_adjustment_sets = self.enumerate_minimal_adjustment_sets([treatment_variable], [outcome_variable])
         elif effect_type == "direct":
             minimal_adjustment_sets = self.direct_effect_adjustment_sets(
@@ -525,14 +526,20 @@ def identification(
                 nodes_to_ignore=nodes_to_ignore,
             )
         else:
-            raise ValueError("Causal effect should be 'total' or 'direct'")
+            raise ValueError(f"Causal effect should be 'total' or 'direct', not '{effect_type}'")
 
         if nodes_to_ignore is not None:
             minimal_adjustment_sets = [
-                adj for adj in minimal_adjustment_sets if not {x.name for x in nodes_to_ignore}.intersection(adj)
+                adj for adj in minimal_adjustment_sets if not set(nodes_to_ignore).intersection(adj)
             ]
 
-        minimal_adjustment_set = min(minimal_adjustment_sets, key=len, default=set())
+        if not minimal_adjustment_sets:
+            raise ValueError(
+                f"Could not find a suitable adjustment set for the {effect_type} effect of {treatment_variable} on "
+                f"{outcome_variable} while avoiding nodes in set {nodes_to_ignore}."
+            )
+
+        minimal_adjustment_set = min(minimal_adjustment_sets, key=len)
         return set(minimal_adjustment_set)
 
     def to_dot_string(self) -> str:
@@ -601,8 +608,7 @@ def generate_causal_test(  # pylint: disable=R0912
         :return: A list containing ShouldCause and ShouldNotCause metamorphic relations.
         """
 
-        if nodes_to_ignore is None:
-            nodes_to_ignore = set()
+        nodes_to_ignore = set(nodes_to_ignore) if nodes_to_ignore is not None else set()
 
         causal_tests = []
 
@@ -690,8 +696,7 @@ def generate_causal_tests(
         :return: A list containing ShouldCause and ShouldNotCause metamorphic relations.
         """
 
-        if nodes_to_ignore is None:
-            nodes_to_ignore = set()
+        nodes_to_ignore = set(nodes_to_ignore) if nodes_to_ignore is not None else set()
         nodes_to_ignore = nodes_to_ignore.union(set(self.cycle_nodes()))
 
         if nodes_to_test is None:

causal_testing/testing/causal_effect.py

@@ -87,7 +87,7 @@ class ExactValue(CausalEffect):
     """An extension of CausalEffect representing that the expected causal effect should be a specific value."""
 
     def __init__(
-        self, value: float, effect_type: str = "direct", atol: float = None, ci_low: float = None, ci_high: float = None
+        self, value: float, effect_type: str = "direct", atol: float = 0, ci_low: float = None, ci_high: float = None
     ):
         super().__init__(effect_type=effect_type)
         if (ci_low is not None) ^ (ci_high is not None):
@@ -98,7 +98,7 @@ def __init__(
         self.value = value
         self.ci_low = ci_low
         self.ci_high = ci_high
-        self.atol = atol if atol is not None else abs(value * 0.05)
+        self.atol = atol
 
         if self.ci_low is not None and self.ci_high is not None:
             if not self.ci_low <= self.value <= self.ci_high:
@@ -128,13 +128,12 @@ def to_dict(self):
 
         :returns: A JSON serialisable dict representing the expected effect.
         """
-        effect = {"value": self.value}
+        effect = {"value": self.value, "atol": self.atol}
         if self.ci_low:
             effect["ci_low"] = self.ci_low
         if self.ci_low:
             effect["ci_high"] = self.ci_high
-        if self.atol:
-            effect["atol"] = self.atol
+
         return super().to_dict() | effect
 
 

causal_testing/testing/causal_test_case.py

@@ -94,13 +94,20 @@ def measure_adequacy(
             except Exception:  # pylint: disable=W0718
                 outcomes.append(None)
 
-        results = pd.concat(results)
+        if results:
+            results = pd.concat(results)
 
-        results["var"] = results.index
+            results["var"] = results.index
 
+            return DataAdequacy(
+                results=results,
+                kurtosis=results.groupby("var")["effect_estimate"].apply(lambda x: x.kurtosis()),
+                passing=int(sum(filter(lambda x: x is not None, outcomes))),
+                successful=int(sum(x is not None for x in outcomes)),
+            )
         return DataAdequacy(
             results=results,
-            kurtosis=results.groupby("var")["effect_estimate"].apply(lambda x: x.kurtosis()),
+            kurtosis=None,
             passing=int(sum(filter(lambda x: x is not None, outcomes))),
             successful=int(sum(x is not None for x in outcomes)),
         )

tests/main_tests/test_ctf.py

@@ -5,6 +5,7 @@
 import pandas as pd
 
 from causal_testing.causal_testing_framework import CausalTestingFramework
+from causal_testing.specification.causal_dag import CausalDAG
 
 
 class TestCausalTestingFramework(unittest.TestCase):
@@ -184,3 +185,27 @@ def test_ctf_evaluate_dag(self):
             }
         ).sort_index()
         pd.testing.assert_series_equal(results, expected)
+
+    def test_ctf_evaluate_dag_inestimable(self):
+        framework = CausalTestingFramework()
+        framework.df = pd.read_csv("tests/resources/data/scarf_data.csv", index_col=0).query("length_in > 60")
+        framework.dag = CausalDAG(datatypes=framework.df.dtypes)
+        framework.dag.add_nodes_from(framework.df.columns)
+        framework.test_cases = framework.dag.generate_causal_tests()
+
+        results = framework.evaluate_dag()
+        expected = pd.Series(
+            {
+                "FAIL": 1,
+                "FAIL_ci_high": 2,
+                "FAIL_ci_low": 0,
+                "INESTIMABLE": 1,
+                "INESTIMABLE_ci_high": 1,
+                "INESTIMABLE_ci_low": 0,
+                "PASS": 4,
+                "PASS_ci_high": 4,
+                "PASS_ci_low": 0,
+            }
+        ).sort_index()
+        print(results)
+        pd.testing.assert_series_equal(results, expected)

tests/specification_tests/test_causal_dag.py

@@ -291,6 +291,21 @@ def test_enumerate_minimal_adjustment_sets(self):
         adjustment_sets = causal_dag.enumerate_minimal_adjustment_sets(xs, ys)
         self.assertEqual([{"Z"}], list(adjustment_sets))
 
+    def test_identification_total_effect(self):
+        """Test whether identification works for total effect."""
+        causal_dag = CausalDAG()
+        causal_dag.add_edges_from([("X", "M"), ("M", "Y")])
+
+        self.assertEqual(
+            set(), causal_dag.identification(treatment_variable="X", outcome_variable="Y", effect_type="total")
+        )
+
+    def test_identification_invalid_effect(self):
+        causal_dag = CausalDAG()
+        with self.assertRaises(ValueError) as e:
+            causal_dag.identification(treatment_variable="X", outcome_variable="Y", effect_type="invalid")
+            self.assertEqual(e.exception, f"Causal effect should be 'total' or 'direct', not 'invalid'.")
+
     def test_enumerate_minimal_adjustment_sets_multiple(self):
         """Test whether enumerate_minimal_adjustment_sets lists all minimum adjustment sets if multiple are possible."""
         causal_dag = CausalDAG()
@@ -403,32 +418,48 @@ def test_list_all_min_sep(self):
         min_separators = set(frozenset(min_separator) for min_separator in min_separators)
         self.assertEqual({frozenset({2, 3}), frozenset({3, 4}), frozenset({4, 5})}, min_separators)
 
+    def test_close_separator_exception(self):
+        g = nx.Graph()
+        g.add_edges_from([("X", "Y")])
+
+        with self.assertRaises(ValueError) as e:
+            close_separator(
+                graph=g,
+                treatment_node="X",
+                outcome_node="X",
+                treatment_node_set={"Y"},
+            )
+            self.assertEqual(e.exception, "No X-Y separator in the graph.")
+
 
 class TestHiddenVariableDAG(unittest.TestCase):
     """
     Test the CausalDAG identification for the exclusion of hidden variables.
     """
 
-    def setUp(self) -> None:
-        self.temp_dir_path = tempfile.mkdtemp()
-        self.dag_dot_path = os.path.join(self.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_ignore_varaible_adjustment_sets(self):
+    def test_impossible_identification(self):
         """Test whether identification produces different adjustment sets if nodes_to_ignore is set."""
-        causal_dag = CausalDAG(self.dag_dot_path)
-        adjustment_sets = causal_dag.identification(treatment_variable="X", outcome_variable="M")
+        causal_dag = CausalDAG()
+        causal_dag.add_edges_from([("X", "M"), ("M", "Y"), ("X", "Y")])
 
-        adjustment_sets_with_hidden = causal_dag.identification(
-            treatment_variable="X", outcome_variable="M", nodes_to_ignore=["Z"]
-        )
+        self.assertEqual(causal_dag.identification(treatment_variable="X", outcome_variable="Y"), {"M"})
 
-        self.assertNotEqual(adjustment_sets, adjustment_sets_with_hidden)
+        with self.assertRaises(ValueError) as e:
+            causal_dag.identification(treatment_variable="X", outcome_variable="Y", nodes_to_ignore=["M"])
+            self.assertEqual(
+                e.exception,
+                "Could not find a suitable adjustment set for the direct effect of X on Y while avoiding nodes in set {M}.",
+            )
 
-    def tearDown(self) -> None:
-        shutil.rmtree(self.temp_dir_path)
+    def test_adjustment_set_nodes_to_ignore(self):
+        """Test whether identification produces different adjustment sets if nodes_to_ignore is set."""
+        causal_dag = CausalDAG()
+        causal_dag.add_edges_from([("L", "V"), ("V", "X"), ("X", "Y"), ("L", "C"), ("C", "Y")])
+
+        self.assertEqual(causal_dag.identification(treatment_variable="X", outcome_variable="Y"), {"C"})
+        self.assertEqual(
+            causal_dag.identification(treatment_variable="X", outcome_variable="Y", nodes_to_ignore={"C"}), {"L"}
+        )
 
 
 def time_it(label, func, *args, **kwargs):

…ting_tests/test_metamorphic_relations.py …tion_tests/test_generate_causal_tests.pytests/testing_tests/test_metamorphic_relations.py renamed to tests/specification_tests/test_generate_causal_tests.py tests/testing_tests/test_metamorphic_relations.py renamed to tests/specification_tests/test_generate_causal_tests.py

@@ -66,9 +66,17 @@ def test_exactValue_pass(self):
         effect_estimate = EffectEstimate(type="ate", value=pd.Series(5.05))
         self.assertTrue(ExactValue(value=5, atol=0.1).apply(effect_estimate))
 
+    def test_exactValue_to_dict(self):
+        self.assertTrue(
+            ExactValue(value=5.01, ci_low=5.0, ci_high=5.08).to_dict(),
+            {"name": "ExactValue", "effect_type": "direct", "value": 5, "ci_high": 5.0, "ci_low": 5.08},
+        )
+
     def test_exactValue_categorical_pass(self):
         effect_estimate = EffectEstimate(type="ate", value=pd.Series({"color[T.red]": 5.05, "color[T.blue]": 4.03}))
-        self.assertTrue(ExactValue(pd.Series({"color[T.red]": 5, "color[T.blue]": 4}), 0.1).apply(effect_estimate))
+        self.assertTrue(
+            ExactValue(value=pd.Series({"color[T.red]": 5, "color[T.blue]": 4}), atol=0.1).apply(effect_estimate)
+        )
 
     def test_exactValue_pass_ci(self):
         effect_estimate = EffectEstimate(type="ate", value=pd.Series(5.05), ci_low=pd.Series(4), ci_high=pd.Series(6))
@@ -78,13 +86,13 @@ def test_exactValue_ci_pass_ci(self):
         effect_estimate = EffectEstimate(
             type="ate", value=pd.Series(5.05), ci_low=pd.Series(4.1), ci_high=pd.Series(5.9)
         )
-        self.assertTrue(ExactValue(value=5, ci_low=4, ci_high=6).apply(effect_estimate))
+        self.assertTrue(ExactValue(value=5, atol=0.05, ci_low=4, ci_high=6).apply(effect_estimate))
 
     def test_exactValue_ci_fail_ci(self):
         effect_estimate = EffectEstimate(
-            type="ate", value=pd.Series(5.05), ci_low=pd.Series(3.9), ci_high=pd.Series(6.1)
+            type="ate", value=pd.Series(5.05), ci_low=pd.Series(4.1), ci_high=pd.Series(5.9)
         )
-        self.assertFalse(ExactValue(value=5, ci_low=4, ci_high=6).apply(effect_estimate))
+        self.assertFalse(ExactValue(value=5, atol=0.04, ci_low=4, ci_high=6).apply(effect_estimate))
 
     def test_exactValue_fail(self):
         effect_estimate = EffectEstimate(type="ate", value=pd.Series(0))

tests/testing_tests/test_causal_effect.py

@@ -63,6 +63,37 @@ def test_data_adequacy_categorical(self):
         self.assertEqual(adequacy_metric.passing, 100, f"Expected passing 100 not {adequacy_metric.passing}")
         self.assertEqual(adequacy_metric.successful, 100, f"Expected successful 100 not {adequacy_metric.successful}")
 
+    def test_data_adequacy_categorical_inestimable(self):
+        df = pd.read_csv("tests/resources/data/scarf_data.csv")
+        causal_test_case = CausalTestCase(
+            expected_causal_effect=NoEffect(atol=1e-10),
+            effect_measure="coefficient",
+            estimator=LinearRegressionEstimator(
+                treatment_variable="color", outcome_variable="completed", adjustment_set=set()
+            ),
+        )
+        adequacy_metric = causal_test_case.measure_adequacy(df.loc[df["color"] == "grey"])
+
+        self.assertEqual(adequacy_metric.kurtosis, None, f"Expected passing None not {adequacy_metric.kurtosis}")
+        self.assertEqual(adequacy_metric.passing, 0, f"Expected passing 0 not {adequacy_metric.passing}")
+        self.assertEqual(adequacy_metric.successful, 0, f"Expected successful 0 not {adequacy_metric.successful}")
+        self.assertEqual(adequacy_metric.results, [])
+
+    def test_data_adequacy_categorical_partly_inestimable(self):
+        df = pd.read_csv("tests/resources/data/scarf_data.csv")
+        causal_test_case = CausalTestCase(
+            expected_causal_effect=NoEffect(atol=1e-10),
+            effect_measure="coefficient",
+            estimator=LinearRegressionEstimator(
+                treatment_variable="color", outcome_variable="completed", adjustment_set=set()
+            ),
+        )
+        adequacy_metric = causal_test_case.measure_adequacy(df.loc[df["length_in"] == 55])
+
+        self.assertEqual(adequacy_metric.kurtosis.values, [0], f"Expected [0] not {adequacy_metric.kurtosis.values}")
+        self.assertEqual(adequacy_metric.passing, 63, f"Expected passing 63 not {adequacy_metric.passing}")
+        self.assertEqual(adequacy_metric.successful, 63, f"Expected successful 63 not {adequacy_metric.successful}")
+
     def test_data_adequacy_group_by(self):
         timesteps_per_intervention = 1
         control_strategy = [[t, "t", 0] for t in range(1, 4, timesteps_per_intervention)]