json logging

Author: jmafoster1

causal_testing/json_front/json_class.py

@@ -129,40 +129,19 @@ def run_json_tests(self, effects: dict, estimators: dict, f_flag: bool = False,
             if "skip" in test and test["skip"]:
                 continue
             test["estimator"] = estimators[test["estimator"]]
-            if "mutations" in test:
+            # If we have specified concrete control and treatment value
+            if "mutations" not in test:
+                failed, msg = self._run_concrete_metamorphic_test()
+            # If we have a variable to mutate
+            else:
                 if test["estimate_type"] == "coefficient":
                     failed, msg = self._run_coefficient_test(test=test, f_flag=f_flag, effects=effects)
                 else:
-                    failed, msg = self._run_ate_test(test=test, f_flag=f_flag, effects=effects, mutates=mutates)
-                self._append_to_file(msg, logging.INFO)
-            else:
-                outcome_variable = next(
-                    iter(test["expected_effect"])
-                )  # Take first key from dictionary of expected effect
-                base_test_case = BaseTestCase(
-                    treatment_variable=self.variables["inputs"][test["treatment_variable"]],
-                    outcome_variable=self.variables["outputs"][outcome_variable],
-                )
-
-                causal_test_case = CausalTestCase(
-                    base_test_case=base_test_case,
-                    expected_causal_effect=effects[test["expected_effect"][outcome_variable]],
-                    control_value=test["control_value"],
-                    treatment_value=test["treatment_value"],
-                    estimate_type=test["estimate_type"],
-                )
-                failed, _ = self._execute_test_case(causal_test_case=causal_test_case, test=test, f_flag=f_flag)
-
-                msg = (
-                    f"Executing concrete test: {test['name']} \n"
-                    + f"treatment variable: {test['treatment_variable']} \n"
-                    + f"outcome_variable = {outcome_variable} \n"
-                    + f"control value = {test['control_value']}, treatment value = {test['treatment_value']} \n"
-                    + f"Result: {'FAILED' if failed else 'Passed'}"
-                )
-                self._append_to_file(msg, logging.INFO)
+                    failed, msg = self._run_metamorphic_tests(
+                        test=test, f_flag=f_flag, effects=effects, mutates=mutates
+                    )
             test["failed"] = failed
-            # print(msg)
+            test["result"] = msg
         return self.test_plan["tests"]
 
     def _run_coefficient_test(self, test: dict, f_flag: bool, effects: dict):
@@ -194,9 +173,36 @@ def _run_coefficient_test(self, test: dict, f_flag: bool, effects: dict):
             + "==============\n"
             + f"  Result: {'FAILED' if failed else 'Passed'}"
         )
+        self._append_to_file(msg, logging.INFO)
+        return failed, result
+
+    def _run_concrete_metamorphic_test(self, test: dict, f_flag: bool, effects: dict, mutates: dict):
+        outcome_variable = next(iter(test["expected_effect"]))  # Take first key from dictionary of expected effect
+        base_test_case = BaseTestCase(
+            treatment_variable=self.variables["inputs"][test["treatment_variable"]],
+            outcome_variable=self.variables["outputs"][outcome_variable],
+        )
+
+        causal_test_case = CausalTestCase(
+            base_test_case=base_test_case,
+            expected_causal_effect=effects[test["expected_effect"][outcome_variable]],
+            control_value=test["control_value"],
+            treatment_value=test["treatment_value"],
+            estimate_type=test["estimate_type"],
+        )
+        failed, msg = self._execute_test_case(causal_test_case=causal_test_case, test=test, f_flag=f_flag)
+
+        msg = (
+            f"Executing concrete test: {test['name']} \n"
+            + f"treatment variable: {test['treatment_variable']} \n"
+            + f"outcome_variable = {outcome_variable} \n"
+            + f"control value = {test['control_value']}, treatment value = {test['treatment_value']} \n"
+            + f"Result: {'FAILED' if failed else 'Passed'}"
+        )
+        self._append_to_file(msg, logging.INFO)
         return failed, msg
 
-    def _run_ate_test(self, test: dict, f_flag: bool, effects: dict, mutates: dict):
+    def _run_metamorphic_tests(self, test: dict, f_flag: bool, effects: dict, mutates: dict):
         """Builds structures and runs test case for tests with an estimate_type of 'ate'.
 
         :param test: Single JSON test definition stored in a mapping (dict)
@@ -228,6 +234,7 @@ def _run_ate_test(self, test: dict, f_flag: bool, effects: dict, mutates: dict):
             + f"  Number of concrete tests for test case: {str(len(concrete_tests))} \n"
             + f"  {failures}/{len(concrete_tests)} failed for {test['name']}"
         )
+        self._append_to_file(msg, logging.INFO)
         return failures, msg
 
     def _execute_tests(self, concrete_tests, test, f_flag):
@@ -269,10 +276,11 @@ def _execute_test_case(
         test_passes = causal_test_case.expected_causal_effect.apply(causal_test_result)
 
         if "coverage" in test and test["coverage"]:
-            adequacy = DataAdequacy(causal_test_case, causal_test_engine, estimation_model)
-            effect_estimate, ci_low, ci_high, outcomes = adequacy.measure_adequacy(100)
-            self._append_to_file(f"KURTOSIS: {effect_estimate.mean()}", logging.INFO)
-            self._append_to_file(f"PASSING: {sum(outcomes)}/{len(outcomes)}", logging.INFO)
+            adequacy_metric = DataAdequacy(causal_test_case, causal_test_engine, estimation_model)
+            adequacy_metric.measure_adequacy()
+            # self._append_to_file(f"KURTOSIS: {effect_estimate.mean()}", logging.INFO)
+            # self._append_to_file(f"PASSING: {sum(outcomes)}/{len(outcomes)}", logging.INFO)
+            causal_test_result.adequacy = adequacy_metric
 
         if causal_test_result.ci_low() is not None and causal_test_result.ci_high() is not None:
             result_string = (

causal_testing/testing/causal_test_adequacy.py

@@ -34,14 +34,19 @@ def measure_adequacy(self):
 
 
 class DataAdequacy:
-    def __init__(self, test_case: CausalTestCase, test_engine: CausalTestEngine, estimator: Estimator):
+    def __init__(
+        self, test_case: CausalTestCase, test_engine: CausalTestEngine, estimator: Estimator, bootstrap_size: int = 100
+    ):
         self.test_case = test_case
         self.test_engine = test_engine
         self.estimator = estimator
+        self.kurtosis = None
+        self.outcomes = None
+        self.bootstrap_size = bootstrap_size
 
-    def measure_adequacy(self, bootstrap_size: int = 100):
+    def measure_adequacy(self):
         results = []
-        for i in range(bootstrap_size):
+        for i in range(self.bootstrap_size):
             estimator = deepcopy(self.estimator)
             estimator.df = estimator.df.sample(len(estimator.df), replace=True, random_state=i)
             try:
@@ -66,6 +71,8 @@ def convert_to_df(field):
             results[field] = convert_to_df(field)
 
         effect_estimate = pd.concat(results["effect_estimate"].tolist(), axis=1).transpose().reset_index(drop=True)
-        ci_low = pd.concat(results["ci_low"].tolist(), axis=1).transpose()
-        ci_high = pd.concat(results["ci_high"].tolist(), axis=1).transpose()
-        return effect_estimate.kurtosis(), ci_low.kurtosis(), ci_high.kurtosis(), outcomes
+        self.kurtosis = effect_estimate.kurtosis()
+        self.outcomes = sum(outcomes)
+
+    def to_dict(self):
+        return {"kurtosis": self.kurtosis.to_dict(), "bootstrap_size": self.bootstrap_size, "passing": self.outcomes}

causal_testing/testing/causal_test_result.py

@@ -28,8 +28,10 @@ def __init__(
         test_value: TestValue,
         confidence_intervals: [float, float] = None,
         effect_modifier_configuration: {Variable: Any} = None,
+        adequacy=None,
     ):
         self.estimator = estimator
+        self.adequacy = adequacy
         if estimator.adjustment_set:
             self.adjustment_set = estimator.adjustment_set
         else:
@@ -66,23 +68,29 @@ def push(s, inc="  "):
                 ci_str = " " + push(pd.DataFrame(self.confidence_intervals).transpose().to_string(header=False))
             confidence_str += f"Confidence intervals:{ci_str}\n"
             confidence_str += f"Alpha:{self.estimator.alpha}\n"
-        return base_str + confidence_str
+        adequacy_str = ""
+        if self.adequacy:
+            adequacy_str = str(self.adequacy)
+        return base_str + confidence_str + adequacy_str
 
-    def to_dict(self):
+    def to_dict(self, json=False):
         """Return result contents as a dictionary
         :return: Dictionary containing contents of causal_test_result
         """
-        return {
+        base_dict = {
             "treatment": self.estimator.treatment,
             "control_value": self.estimator.control_value,
             "treatment_value": self.estimator.treatment_value,
             "outcome": self.estimator.outcome,
-            "adjustment_set": self.adjustment_set,
+            "adjustment_set": list(self.adjustment_set) if json else self.adjustment_set,
             "effect_measure": self.test_value.type,
             "effect_estimate": self.test_value.value,
             "ci_low": self.ci_low(),
             "ci_high": self.ci_high(),
         }
+        if self.adequacy:
+            base_dict["adequacy"] = self.adequacy.to_dict()
+        return base_dict
 
     def ci_low(self):
         """Return the lower bracket of the confidence intervals."""