Merge branch 'main' of github.com:CITCOM-project/CausalTestingFramework into json-scenario

Author: jmafoster1

causal_testing/json_front/json_class.py

@@ -176,11 +176,9 @@ 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()}"
-            )
+            result_string = 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 +189,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})."
+        )