pylint

Author: jmafoster1

causal_testing/json_front/json_class.py

@@ -56,13 +56,15 @@ def __init__(self, output_path: str, output_overwrite: bool = False):
         self.output_path = Path(output_path)
         self.check_file_exists(self.output_path, output_overwrite)
 
-    def set_paths(self, json_path: str, dag_path: str, data_paths: str = []):
+    def set_paths(self, json_path: str, dag_path: str, data_paths: list[str] = None):
         """
         Takes a path of the directory containing all scenario specific files and creates individual paths for each file
         :param json_path: string path representation to .json file containing test specifications
         :param dag_path: string path representation to the .dot file containing the Causal DAG
         :param data_paths: string path representation to the data files
         """
+        if data_paths is None:
+            data_paths = []
         self.input_paths = JsonClassPaths(json_path=json_path, dag_path=dag_path, data_paths=data_paths)
 
     def setup(self, scenario: Scenario):

causal_testing/testing/causal_test_outcome.py

@@ -32,7 +32,7 @@ def apply(self, res: CausalTestResult) -> bool:
         if res.test_value.type == "coefficient":
             ci_low = res.ci_low() if isinstance(res.ci_low(), Iterable) else [res.ci_low()]
             ci_high = res.ci_high() if isinstance(res.ci_high(), Iterable) else [res.ci_high()]
-            return any([0 < ci_low < ci_high or ci_low < ci_high < 0 for ci_low, ci_high in zip(ci_low, ci_high)])
+            return any(0 < ci_low < ci_high or ci_low < ci_high < 0 for ci_low, ci_high in zip(ci_low, ci_high))
         if res.test_value.type == "risk_ratio":
             return (1 < res.ci_low() < res.ci_high()) or (res.ci_low() < res.ci_high() < 1)
         raise ValueError(f"Test Value type {res.test_value.type} is not valid for this TestOutcome")
@@ -41,18 +41,18 @@ def apply(self, res: CausalTestResult) -> bool:
 class NoEffect(CausalTestOutcome):
     """An extension of TestOutcome representing that the expected causal effect should be zero."""
 
-    def apply(self, res: CausalTestResult, threshold: float = 1e-10) -> bool:
+    def apply(self, res: CausalTestResult, atol: float = 1e-10) -> bool:
         if res.test_value.type == "ate":
-            return (res.ci_low() < 0 < res.ci_high()) or (abs(res.test_value.value) < 1e-10)
+            return (res.ci_low() < 0 < res.ci_high()) or (abs(res.test_value.value) < atol)
         if res.test_value.type == "coefficient":
             ci_low = res.ci_low() if isinstance(res.ci_low(), Iterable) else [res.ci_low()]
             ci_high = res.ci_high() if isinstance(res.ci_high(), Iterable) else [res.ci_high()]
             value = res.test_value.value if isinstance(res.ci_high(), Iterable) else [res.test_value.value]
-            return all([ci_low < 0 < ci_high for ci_low, ci_high in zip(ci_low, ci_high)]) or all(
-                [abs(v) < 1e-10 for v in value]
+            return all(ci_low < 0 < ci_high for ci_low, ci_high in zip(ci_low, ci_high)) or all(
+                abs(v) < 1e-10 for v in value
             )
         if 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)
+            return (res.ci_low() < 1 < res.ci_high()) or np.isclose(res.test_value.value, 1.0, atol=atol)
         raise ValueError(f"Test Value type {res.test_value.type} is not valid for this TestOutcome")
 
 
@@ -85,7 +85,8 @@ def apply(self, res: CausalTestResult) -> bool:
             return res.test_value.value > 0
         if res.test_value.type == "risk_ratio":
             return res.test_value.value > 1
-        raise ValueError(f"Test Value type {res.test_value.type} is not valid for this TestOutcome")
+        # Dead code?
+        # raise ValueError(f"Test Value type {res.test_value.type} is not valid for this TestOutcome")
 
 
 class Negative(SomeEffect):
@@ -98,4 +99,5 @@ def apply(self, res: CausalTestResult) -> bool:
             return res.test_value.value < 0
         if res.test_value.type == "risk_ratio":
             return res.test_value.value < 1
-        raise ValueError(f"Test Value type {res.test_value.type} is not valid for this TestOutcome")
+        # Dead code?
+        # raise ValueError(f"Test Value type {res.test_value.type} is not valid for this TestOutcome")

causal_testing/testing/estimators.py

@@ -434,35 +434,6 @@ def estimate_ate_calculated(self, adjustment_config: dict = None) -> tuple[float
 
         return (treatment_outcome["mean"] - control_outcome["mean"]), [ci_low, ci_high]
 
-    def estimate_cates(self) -> tuple[float, list[float, float]]:
-        """Estimate the conditional average treatment effect of the treatment on the outcome. That is, the change
-        in outcome caused by changing the treatment variable from the control value to the treatment value.
-
-        :return: The conditional average treatment effect and the 95% Wald confidence intervals.
-        """
-        assert (
-            self.effect_modifiers
-        ), f"Must have at least one effect modifier to compute CATE - {self.effect_modifiers}."
-        x = pd.DataFrame()
-        x[self.treatment] = [self.treatment_value, self.control_value]
-        x["Intercept"] = 1  # self.intercept
-        for k, v in self.effect_modifiers.items():
-            self.adjustment_set.add(k)
-            x[k] = v
-        if hasattr(self, "square_terms"):
-            for t in self.square_terms:
-                x[t + "^2"] = x[t] ** 2
-        if hasattr(self, "product_terms"):
-            for a, b in self.product_terms:
-                x[f"{a}*{b}"] = x[a] * x[b]
-
-        model = self._run_linear_regression()
-        y = model.predict(x)
-        treatment_outcome = y.iloc[0]
-        control_outcome = y.iloc[1]
-
-        return treatment_outcome - control_outcome, None
-
     def _run_linear_regression(self) -> RegressionResultsWrapper:
         """Run linear regression of the treatment and adjustment set against the outcome and return the model.
 
@@ -482,7 +453,6 @@ def _run_linear_regression(self) -> RegressionResultsWrapper:
         # 3. Estimate the unit difference in outcome caused by unit difference in treatment
         cols = [self.treatment]
         cols += [x for x in self.adjustment_set if x not in cols]
-        treatment_and_adjustments_cols = reduced_df[cols + ["Intercept"]]
         model = smf.ols(formula=self.formula, data=self.df).fit()
         return model
 

tests/json_front_tests/test_json_class.py

@@ -43,6 +43,13 @@ def setUp(self) -> None:
         self.json_class.set_paths(self.json_path, self.dag_path, self.data_path)
         self.json_class.setup(self.scenario)
 
+    def test_setting_no_path(self):
+        json_class = JsonUtility("temp_out.txt", True)
+        json_class.set_paths(self.json_path, self.dag_path, None)
+        self.assertEqual(json_class.input_paths.data_paths, [])  # Needs to be list of Paths
+
+
+
     def test_setting_paths(self):
         self.assertEqual(self.json_class.input_paths.json_path, Path(self.json_path))
         self.assertEqual(self.json_class.input_paths.dag_path, Path(self.dag_path))

tests/testing_tests/test_causal_test_outcome.py

@@ -64,7 +64,7 @@ def test_empty_adjustment_set(self):
             ),
         )
 
-    def test_Positive_pass(self):
+    def test_Positive_ate_pass(self):
         test_value = TestValue(type="ate", value=5.05)
         ctr = CausalTestResult(
             estimator=self.estimator,
@@ -75,6 +75,17 @@ def test_Positive_pass(self):
         ev = Positive()
         self.assertTrue(ev.apply(ctr))
 
+    def test_Positive_risk_ratio_pass(self):
+        test_value = TestValue(type="risk_ratio", value=2)
+        ctr = CausalTestResult(
+            estimator=self.estimator,
+            test_value=test_value,
+            confidence_intervals=None,
+            effect_modifier_configuration=None,
+        )
+        ev = Positive()
+        self.assertTrue(ev.apply(ctr))
+
     def test_Positive_fail(self):
         test_value = TestValue(type="ate", value=0)
         ctr = CausalTestResult(
@@ -97,7 +108,7 @@ def test_Positive_fail_ci(self):
         ev = Positive()
         self.assertFalse(ev.apply(ctr))
 
-    def test_Negative_pass(self):
+    def test_Negative_ate_pass(self):
         test_value = TestValue(type="ate", value=-5.05)
         ctr = CausalTestResult(
             estimator=self.estimator,
@@ -108,6 +119,17 @@ def test_Negative_pass(self):
         ev = Negative()
         self.assertTrue(ev.apply(ctr))
 
+    def test_Negative_risk_ratio_pass(self):
+        test_value = TestValue(type="risk_ratio", value=0.2)
+        ctr = CausalTestResult(
+            estimator=self.estimator,
+            test_value=test_value,
+            confidence_intervals=None,
+            effect_modifier_configuration=None,
+        )
+        ev = Negative()
+        self.assertTrue(ev.apply(ctr))
+
     def test_Negative_fail(self):
         test_value = TestValue(type="ate", value=0)
         ctr = CausalTestResult(
@@ -163,17 +185,33 @@ def test_exactValue_fail(self):
         ev = ExactValue(5, 0.1)
         self.assertFalse(ev.apply(ctr))
 
-    def test_someEffect_invalid(self):
+    def test_invalid(self):
         test_value = TestValue(type="invalid", value=5.05)
         ctr = CausalTestResult(
             estimator=self.estimator,
             test_value=test_value,
             confidence_intervals=[4.8, 6.7],
             effect_modifier_configuration=None,
         )
-        ev = SomeEffect()
         with self.assertRaises(ValueError):
-            ev.apply(ctr)
+            SomeEffect().apply(ctr)
+        with self.assertRaises(ValueError):
+            NoEffect().apply(ctr)
+        with self.assertRaises(ValueError):
+            Positive().apply(ctr)
+        with self.assertRaises(ValueError):
+            Negative().apply(ctr)
+
+    def test_someEffect_pass_coefficient(self):
+        test_value = TestValue(type="coefficient", value=5.05)
+        ctr = CausalTestResult(
+            estimator=self.estimator,
+            test_value=test_value,
+            confidence_intervals=[4.8, 6.7],
+            effect_modifier_configuration=None,
+        )
+        self.assertTrue(SomeEffect().apply(ctr))
+        self.assertFalse(NoEffect().apply(ctr))
 
     def test_someEffect_pass_ate(self):
         test_value = TestValue(type="ate", value=5.05)

tests/testing_tests/test_estimators.py

@@ -93,6 +93,14 @@ def setUpClass(cls) -> None:
             ]
         )
 
+    # Yes, this probably shouldn't be in here, but it uses the scarf data so it makes more sense to put it
+    # here than duplicating the scarf data for a single test
+    def test_linear_regression_categorical_ate(self):
+        df = self.scarf_df.copy()
+        logistic_regression_estimator = LinearRegressionEstimator("color", None, None, set(), "completed", df)
+        ate, confidence = logistic_regression_estimator.estimate_unit_ate()
+        self.assertTrue(all([ci_low < 0 < ci_high for ci_low, ci_high in zip(confidence[0], confidence[1])]))
+
     def test_ate(self):
         df = self.scarf_df.copy()
         logistic_regression_estimator = LogisticRegressionEstimator("length_in", 65, 55, set(), "completed", df)
@@ -212,11 +220,9 @@ def test_program_11_2(self):
         model = linear_regression_estimator._run_linear_regression()
         ate, _ = linear_regression_estimator.estimate_unit_ate()
 
-        print(model.summary())
         self.assertEqual(round(model.params["Intercept"] + 90 * model.params["treatments"], 1), 216.9)
 
         # Increasing treatments from 90 to 100 should be the same as 10 times the unit ATE
-        print("ATE", ate)
         self.assertEqual(round(model.params["treatments"], 1), round(ate, 1))
 
     def test_program_11_3(self):