Merge pull request #119 from CITCOM-project/somers/test_value

causal_test_result + new dataclass for test values

Author: rsomers1998

causal_testing/json_front/json_class.py

@@ -177,10 +177,10 @@ def _execute_test_case(self, causal_test_case: CausalTestCase, estimator: Estima
         result_string = str()
         if causal_test_result.ci_low() and causal_test_result.ci_high():
             result_string = (
-                f"{causal_test_result.ci_low()} < {causal_test_result.ate} <  {causal_test_result.ci_high()}"
+                f"{causal_test_result.ci_low()} < {causal_test_result.test_value.value} <  {causal_test_result.ci_high()}"
             )
         else:
-            result_string = causal_test_result.ate
+            result_string = causal_test_result.test_value.value
         if f_flag:
             assert test_passes, (
                 f"{causal_test_case}\n    FAILED - expected {causal_test_case.expected_causal_effect}, "
@@ -191,7 +191,7 @@ def _execute_test_case(self, causal_test_case: CausalTestCase, estimator: Estima
             logger.warning(
                 "   FAILED- expected %s, got %s",
                 causal_test_case.expected_causal_effect,
-                causal_test_result.ate,
+                causal_test_result.test_value.value,
             )
         return failed
 

causal_testing/testing/causal_test_engine.py

@@ -5,7 +5,7 @@
 from causal_testing.data_collection.data_collector import DataCollector
 from causal_testing.specification.causal_specification import CausalSpecification
 from causal_testing.testing.causal_test_case import CausalTestCase
-from causal_testing.testing.causal_test_outcome import CausalTestResult
+from causal_testing.testing.causal_test_result import CausalTestResult, TestValue
 from causal_testing.testing.estimators import Estimator
 
 logger = logging.getLogger(__name__)
@@ -138,7 +138,7 @@ def execute_test(
                     treatment_value=estimator.treatment_values,
                     control_value=estimator.control_values,
                     adjustment_set=estimator.adjustment_set,
-                    ate=cates_df,
+                    test_value=TestValue("ate", cates_df),
                     effect_modifier_configuration=causal_test_case.effect_modifier_configuration,
                     confidence_intervals=confidence_intervals,
                 )
@@ -151,7 +151,7 @@ def execute_test(
                 treatment_value=estimator.treatment_values,
                 control_value=estimator.control_values,
                 adjustment_set=estimator.adjustment_set,
-                ate=risk_ratio,
+                test_value=TestValue("risk_ratio", risk_ratio),
                 effect_modifier_configuration=causal_test_case.effect_modifier_configuration,
                 confidence_intervals=confidence_intervals,
             )
@@ -164,7 +164,7 @@ def execute_test(
                 treatment_value=estimator.treatment_values,
                 control_value=estimator.control_values,
                 adjustment_set=estimator.adjustment_set,
-                ate=ate,
+                test_value=TestValue("ate", ate),
                 effect_modifier_configuration=causal_test_case.effect_modifier_configuration,
                 confidence_intervals=confidence_intervals,
             )
@@ -179,7 +179,7 @@ def execute_test(
                 treatment_value=estimator.treatment_values,
                 control_value=estimator.control_values,
                 adjustment_set=estimator.adjustment_set,
-                ate=ate,
+                test_value=TestValue("ate", ate),
                 effect_modifier_configuration=causal_test_case.effect_modifier_configuration,
                 confidence_intervals=confidence_intervals,
             )

causal_testing/testing/causal_test_outcome.py

@@ -1,92 +1,7 @@
 from abc import ABC, abstractmethod
-from typing import Any, Union
-
+from causal_testing.testing.causal_test_result import CausalTestResult
 import numpy as np
 
-from causal_testing.specification.variable import Variable
-
-
-class CausalTestResult:
-    """A container to hold the results of a causal test case. Every causal test case provides a point estimate of
-    the ATE, given a particular treatment, outcome, and adjustment set. Some but not all estimators can provide
-    confidence intervals."""
-
-    def __init__(
-        self,
-        treatment: tuple,
-        outcome: tuple,
-        treatment_value: Union[int, float, str],
-        control_value: Union[int, float, str],
-        adjustment_set: set,
-        ate: float,
-        confidence_intervals: [float, float] = None,
-        effect_modifier_configuration: {Variable: Any} = None,
-    ):
-        self.treatment = treatment
-        self.outcome = outcome
-        self.treatment_value = treatment_value
-        self.control_value = control_value
-        if adjustment_set:
-            self.adjustment_set = adjustment_set
-        else:
-            self.adjustment_set = set()
-        self.ate = ate
-        self.confidence_intervals = confidence_intervals
-
-        if effect_modifier_configuration is not None:
-            self.effect_modifier_configuration = effect_modifier_configuration
-        else:
-            self.effect_modifier_configuration = {}
-
-    def __str__(self):
-        base_str = (
-            f"Causal Test Result\n==============\n"
-            f"Treatment: {self.treatment[0]}\n"
-            f"Control value: {self.control_value}\n"
-            f"Treatment value: {self.treatment_value}\n"
-            f"Outcome: {self.outcome[0]}\n"
-            f"Adjustment set: {self.adjustment_set}\n"
-            f"ATE: {self.ate}\n"
-        )
-        confidence_str = ""
-        if self.confidence_intervals:
-            confidence_str += f"Confidence intervals: {self.confidence_intervals}\n"
-        return base_str + confidence_str
-
-    def to_dict(self):
-        base_dict = {
-            "treatment": self.treatment[0],
-            "control_value": self.control_value,
-            "treatment_value": self.treatment_value,
-            "outcome": self.outcome[0],
-            "adjustment_set": self.adjustment_set,
-            "ate": self.ate,
-        }
-        if self.confidence_intervals:
-            base_dict["ci_low"] = min(self.confidence_intervals)
-            base_dict["ci_high"] = max(self.confidence_intervals)
-        return base_dict
-
-    def ci_low(self):
-        """Return the lower bracket of the confidence intervals."""
-        if not self.confidence_intervals:
-            return None
-        return min(self.confidence_intervals)
-
-    def ci_high(self):
-        """Return the higher bracket of the confidence intervals."""
-        if not self.confidence_intervals:
-            return None
-        return max(self.confidence_intervals)
-
-    def summary(self):
-        """Summarise the causal test result as an intuitive sentence."""
-        print(
-            f"The causal effect of changing {self.treatment[0]} = {self.control_value} to "
-            f"{self.treatment[0]}' = {self.treatment_value} is {self.ate} (95% confidence intervals: "
-            f"{self.confidence_intervals})."
-        )
-
 
 class CausalTestOutcome(ABC):
     """An abstract class representing an expected causal effect."""
@@ -110,7 +25,7 @@ def __init__(self, value: float, tolerance: float = None):
             self.tolerance = tolerance
 
     def apply(self, res: CausalTestResult) -> bool:
-        return np.isclose(res.ate, self.value, atol=self.tolerance)
+        return np.isclose(res.test_value.value, self.value, atol=self.tolerance)
 
     def __str__(self):
         return f"ExactValue: {self.value}±{self.tolerance}"
@@ -121,22 +36,31 @@ class Positive(CausalTestOutcome):
 
     def apply(self, res: CausalTestResult) -> bool:
         # TODO: confidence intervals?
-        return res.ate > 0
+        if res.test_value.type == "ate":
+            return res.test_value.value > 0
+        elif res.test_value.type == "risk_ratio":
+            return res.test_value.value > 1
 
 
 class Negative(CausalTestOutcome):
     """An extension of TestOutcome representing that the expected causal effect should be negative."""
 
     def apply(self, res: CausalTestResult) -> bool:
         # TODO: confidence intervals?
-        return res.ate < 0
+        if res.test_value.type == "ate":
+            return res.test_value.value < 0
+        elif res.test_value.type == "risk_ratio":
+            return res.test_value.value < 1
 
 
 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)
+        if res.test_value.type == "ate":
+            return (0 < res.ci_low() < res.ci_high()) or (res.ci_low() < res.ci_high() < 0)
+        elif res.test_value.type == "risk_ratio":
+            return (1 < res.ci_low() < res.ci_high()) or (res.ci_low() < res.ci_high() < 1)
 
     def __str__(self):
         return "Changed"
@@ -146,7 +70,10 @@ class NoEffect(CausalTestOutcome):
     """An extension of TestOutcome representing that the expected causal effect should be zero."""
 
     def apply(self, res: CausalTestResult) -> bool:
-        return (res.ci_low() < 0 < res.ci_high()) or (abs(res.ate) < 1e-10)
+        if res.test_value.type == "ate":
+            return (res.ci_low() < 0 < res.ci_high()) or (abs(res.test_value.value) < 1e-10)
+        elif res.test_value.type == "risk_ratio":
+            return (res.ci_low() < 1 < res.ci_high()) or np.isclose(res.test_value.value, 1.0, atol=1e-10)
 
     def __str__(self):
         return "Unchanged"

causal_testing/testing/causal_test_result.py

@@ -0,0 +1,94 @@
+from typing import Any, Union
+from dataclasses import dataclass
+
+from causal_testing.specification.variable import Variable
+
+
+@dataclass
+class TestValue:
+    """A dataclass to hold both the type and value of a causal test result"""
+
+    type: str
+    value: float
+
+
+class CausalTestResult:
+    """A container to hold the results of a causal test case. Every causal test case provides a point estimate of
+    the ATE, given a particular treatment, outcome, and adjustment set. Some but not all estimators can provide
+    confidence intervals."""
+
+    def __init__(
+        self,
+        treatment: tuple,
+        outcome: tuple,
+        treatment_value: Union[int, float, str],
+        control_value: Union[int, float, str],
+        adjustment_set: set,
+        test_value: TestValue,
+        confidence_intervals: [float, float] = None,
+        effect_modifier_configuration: {Variable: Any} = None,
+    ):
+        self.treatment = treatment
+        self.outcome = outcome
+        self.treatment_value = treatment_value
+        self.control_value = control_value
+        if adjustment_set:
+            self.adjustment_set = adjustment_set
+        else:
+            self.adjustment_set = set()
+        self.test_value = test_value
+        self.confidence_intervals = confidence_intervals
+
+        if effect_modifier_configuration is not None:
+            self.effect_modifier_configuration = effect_modifier_configuration
+        else:
+            self.effect_modifier_configuration = {}
+
+    def __str__(self):
+        base_str = (
+            f"Causal Test Result\n==============\n"
+            f"Treatment: {self.treatment[0]}\n"
+            f"Control value: {self.control_value}\n"
+            f"Treatment value: {self.treatment_value}\n"
+            f"Outcome: {self.outcome[0]}\n"
+            f"Adjustment set: {self.adjustment_set}\n"
+            f"{self.test_value.type}: {self.test_value.value}\n"
+        )
+        confidence_str = ""
+        if self.confidence_intervals:
+            confidence_str += f"Confidence intervals: {self.confidence_intervals}\n"
+        return base_str + confidence_str
+
+    def to_dict(self):
+        base_dict = {
+            "treatment": self.treatment[0],
+            "control_value": self.control_value,
+            "treatment_value": self.treatment_value,
+            "outcome": self.outcome[0],
+            "adjustment_set": self.adjustment_set,
+            "test_value": self.test_value,
+        }
+        if self.confidence_intervals:
+            base_dict["ci_low"] = min(self.confidence_intervals)
+            base_dict["ci_high"] = max(self.confidence_intervals)
+        return base_dict
+
+    def ci_low(self):
+        """Return the lower bracket of the confidence intervals."""
+        if not self.confidence_intervals:
+            return None
+        return min(self.confidence_intervals)
+
+    def ci_high(self):
+        """Return the higher bracket of the confidence intervals."""
+        if not self.confidence_intervals:
+            return None
+        return max(self.confidence_intervals)
+
+    def summary(self):
+        """Summarise the causal test result as an intuitive sentence."""
+        print(
+            f"The causal effect of changing {self.treatment[0]} = {self.control_value} to "
+            f"{self.treatment[0]}' = {self.treatment_value} is {self.test_value.value} (95% confidence intervals: "
+            f"{self.confidence_intervals})."
+        )

examples/covasim_/doubling_beta/causal_test_beta.py

@@ -64,10 +64,10 @@ def doubling_beta_CATE_on_csv(observational_data_path: str, simulate_counterfact
     association_test_result = causal_test_engine.execute_test(no_adjustment_linear_regression_estimator, causal_test_case, 'ate')
 
     # Store results for plotting
-    results_dict['association'] = {'ate': association_test_result.ate,
+    results_dict['association'] = {'ate': association_test_result.test_value.value,
                                    'cis': association_test_result.confidence_intervals,
                                    'df': past_execution_df}
-    results_dict['causation'] = {'ate': causal_test_result.ate,
+    results_dict['causation'] = {'ate': causal_test_result.test_value.value,
                                  'cis': causal_test_result.confidence_intervals,
                                  'df': past_execution_df}
 
@@ -84,7 +84,7 @@ def doubling_beta_CATE_on_csv(observational_data_path: str, simulate_counterfact
                                                                                df=counterfactual_past_execution_df)
         counterfactual_linear_regression_estimator.add_squared_term_to_df('beta')
         counterfactual_causal_test_result = causal_test_engine.execute_test(linear_regression_estimator, causal_test_case, 'ate')
-        results_dict['counterfactual'] = {'ate': counterfactual_causal_test_result.ate,
+        results_dict['counterfactual'] = {'ate': counterfactual_causal_test_result.test_value.value,
                                           'cis': counterfactual_causal_test_result.confidence_intervals,
                                           'df': counterfactual_past_execution_df}
         if verbose:

examples/covasim_/vaccinating_elderly/causal_test_vaccine.py

@@ -91,7 +91,7 @@ def experimental_causal_test_vaccinate_elderly(runs_per_test_per_config: int = 3
         causal_test_result = causal_test_engine.execute_test(linear_regression_estimator, causal_test_case, 'ate')
         if verbose:
             print(f"Causation:\n{causal_test_result}")
-        results_dict[outcome_variable.name]['ate'] = causal_test_result.ate
+        results_dict[outcome_variable.name]['ate'] = causal_test_result.test_value.value
         results_dict[outcome_variable.name]['cis'] = causal_test_result.confidence_intervals
         results_dict[outcome_variable.name]['test_passes'] = causal_test_case.expected_causal_effect.apply(
             causal_test_result)

examples/lr91/causal_test_max_conductances.py

@@ -132,7 +132,7 @@ def effects_on_APD90(observational_data_path, treatment_var, control_val, treatm
     # 10. Run the causal test and print results
     causal_test_result = causal_test_engine.execute_test(linear_regression_estimator, causal_test_case, 'ate')
     print(causal_test_result)
-    return causal_test_result.ate, causal_test_result.confidence_intervals
+    return causal_test_result.test_value.value, causal_test_result.confidence_intervals
 
 
 def plot_ates_with_cis(results_dict: dict, xs: list, save: bool = True):

examples/poisson-line-process/causal_test_poisson.py

@@ -182,8 +182,8 @@ def test_intensity_num_shapes(
             "height": wh,
             "control": control_value,
             "treatment": treatment_value,
-            "smt_risk_ratio": smt_causal_test_result.ate,
-            "obs_risk_ratio": obs_causal_test_result.ate,
+            "smt_risk_ratio": smt_causal_test_result.test_value.value,
+            "obs_risk_ratio": obs_causal_test_result.test_value.value,
         }
         intensity_num_shapes_results.append(results)
 
@@ -218,7 +218,7 @@ def test_intensity_num_shapes(
             "control": control_value,
             "treatment": treatment_value,
             "intensity": i,
-            "ate": causal_test_result.ate,
+            "ate": causal_test_result.test_value.value,
             "ci_low": min(causal_test_result.confidence_intervals),
             "ci_high": max(causal_test_result.confidence_intervals),
         }

examples/poisson/run_causal_tests.py

@@ -3,8 +3,8 @@
 import scipy
 
 from causal_testing.testing.estimators import LinearRegressionEstimator, CausalForestEstimator
-from causal_testing.testing.causal_test_outcome import ExactValue, Positive, Negative, NoEffect, CausalTestOutcome, \
-    CausalTestResult
+from causal_testing.testing.causal_test_outcome import ExactValue, Positive, Negative, NoEffect, CausalTestOutcome
+from causal_testing.testing.causal_test_result import CausalTestResult
 from causal_testing.json_front.json_class import JsonUtility
 from causal_testing.testing.estimators import Estimator
 from causal_testing.specification.scenario import Scenario
@@ -70,8 +70,8 @@ def apply(self, res: CausalTestResult) -> bool:
         i = effect_modifier_configuration['intensity']
         self.i2c = i * 2 * c
         print("2ic:", f"2*{i}*{c}={self.i2c}")
-        print("ate:", res.ate)
-        return np.isclose(res.ate, self.i2c, atol=self.tolerance)
+        print("ate:", res.test_value.value)
+        return np.isclose(res.test_value.value, self.i2c, atol=self.tolerance)
 
     def __str__(self):
         if self.i2c is None: