Merge branch 'main' into dev_docs

Author: christopher-wild

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):
@@ -73,7 +75,12 @@ def setup(self, scenario: Scenario):
         self.causal_specification = CausalSpecification(
             scenario=self.scenario, causal_dag=CausalDAG(self.input_paths.dag_path)
         )
-        self._json_parse()
+        # Parse the JSON test plan
+        with open(self.input_paths.json_path, encoding="utf-8") as f:
+            self.test_plan = json.load(f)
+        # Populate the data
+        if self.input_paths.data_paths:
+            self.data = pd.concat([pd.read_csv(data_file, header=0) for data_file in self.input_paths.data_paths])
         self._populate_metas()
 
     def _create_abstract_test_case(self, test, mutates, effects):
@@ -144,6 +151,7 @@ def run_json_tests(self, effects: dict, estimators: dict, f_flag: bool = False,
                         + "==============\n"
                         + f"  Result: {'FAILED' if result[0] else 'Passed'}"
                     )
+                    print(msg)
                 else:
                     abstract_test = self._create_abstract_test_case(test, mutates, effects)
                     concrete_tests, _ = abstract_test.generate_concrete_tests(5, 0.05)
@@ -198,15 +206,6 @@ def _execute_tests(self, concrete_tests, test, f_flag):
                 failures += 1
         return failures, details
 
-    def _json_parse(self):
-        """Parse a JSON input file into inputs, outputs, metas and a test plan"""
-        with open(self.input_paths.json_path, encoding="utf-8") as f:
-            self.test_plan = json.load(f)
-        for data_file in self.input_paths.data_paths:
-            df = pd.read_csv(data_file, header=0)
-            self.data.append(df)
-        self.data = pd.concat(self.data)
-
     def _populate_metas(self):
         """
         Populate data with meta-variable values and add distributions to Causal Testing Framework Variables
@@ -236,7 +235,7 @@ def _execute_test_case(
 
         test_passes = causal_test_case.expected_causal_effect.apply(causal_test_result)
 
-        if causal_test_result.ci_low() and causal_test_result.ci_high():
+        if causal_test_result.ci_low() is not None and causal_test_result.ci_high() is not None:
             result_string = (
                 f"{causal_test_result.ci_low()} < {causal_test_result.test_value.value} <  "
                 f"{causal_test_result.ci_high()}"
@@ -351,7 +350,6 @@ def get_args(test_args=None) -> argparse.Namespace:
         parser.add_argument(
             "--data_path",
             help="Specify path to file containing runtime data",
-            required=True,
             nargs="+",
         )
         parser.add_argument(

causal_testing/specification/metamorphic_relation.py

@@ -7,13 +7,18 @@
 from abc import abstractmethod
 from typing import Iterable
 from itertools import combinations
-import numpy as np
-import pandas as pd
+import argparse
+import logging
+import json
 import networkx as nx
+import pandas as pd
+import numpy as np
 
 from causal_testing.specification.causal_specification import CausalDAG, Node
 from causal_testing.data_collection.data_collector import ExperimentalDataCollector
 
+logger = logging.getLogger(__name__)
+
 
 @dataclass(order=True)
 class MetamorphicRelation:
@@ -142,6 +147,7 @@ def to_json_stub(self, skip=True) -> dict:
             "effect": "direct",
             "mutations": [self.treatment_var],
             "expected_effect": {self.output_var: "SomeEffect"},
+            "formula": f"{self.output_var} ~ {' + '.join([self.treatment_var] + self.adjustment_vars)}",
             "skip": skip,
         }
 
@@ -174,6 +180,7 @@ def to_json_stub(self, skip=True) -> dict:
             "effect": "direct",
             "mutations": [self.treatment_var],
             "expected_effect": {self.output_var: "NoEffect"},
+            "formula": f"{self.output_var} ~ {' + '.join([self.treatment_var] + self.adjustment_vars)}",
             "skip": skip,
         }
 
@@ -244,3 +251,35 @@ def generate_metamorphic_relations(dag: CausalDAG) -> list[MetamorphicRelation]:
             metamorphic_relations.append(ShouldCause(v, u, adj_set, dag))
 
     return metamorphic_relations
+
+
+if __name__ == "__main__": # pragma: no cover
+    logging.basicConfig(format="%(levelname)s: %(message)s", level=logging.INFO)
+    parser = argparse.ArgumentParser(
+        description="A script for generating metamorphic relations to test the causal relationships in a given DAG."
+    )
+    parser.add_argument(
+        "--dag_path",
+        "-d",
+        help="Specify path to file containing the DAG, normally a .dot file.",
+        required=True,
+    )
+    parser.add_argument(
+        "--output_path",
+        "-o",
+        help="Specify path where tests should be saved, normally a .json file.",
+        required=True,
+    )
+    args = parser.parse_args()
+
+    causal_dag = CausalDAG(args.dag_path)
+    relations = generate_metamorphic_relations(causal_dag)
+    tests = [
+        relation.to_json_stub(skip=False)
+        for relation in relations
+        if len(list(causal_dag.graph.predecessors(relation.output_var))) > 0
+    ]
+
+    logger.info(f"Generated {len(tests)} tests. Saving to {args.output_path}.")
+    with open(args.output_path, "w", encoding="utf-8") as f:
+        json.dump({"tests": tests}, f, indent=2)

causal_testing/testing/causal_test_outcome.py

@@ -3,6 +3,7 @@
 ExactValue, Positive, Negative, SomeEffect, NoEffect"""
 
 from abc import ABC, abstractmethod
+from collections.abc import Iterable
 import numpy as np
 
 from causal_testing.testing.causal_test_result import CausalTestResult
@@ -26,8 +27,12 @@ class SomeEffect(CausalTestOutcome):
     """An extension of TestOutcome representing that the expected causal effect should not be zero."""
 
     def apply(self, res: CausalTestResult) -> bool:
-        if res.test_value.type in {"ate", "coefficient"}:
+        if res.test_value.type == "ate":
             return (0 < res.ci_low() < res.ci_high()) or (res.ci_low() < res.ci_high() < 0)
+        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))
         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")
@@ -36,32 +41,41 @@ def apply(self, res: CausalTestResult) -> bool:
 class NoEffect(CausalTestOutcome):
     """An extension of TestOutcome representing that the expected causal effect should be zero."""
 
+    def __init__(self, atol: float = 1e-10):
+        self.atol = atol
+
     def apply(self, res: CausalTestResult) -> bool:
-        print("RESULT", res)
-        if res.test_value.type in {"ate", "coefficient"}:
-            return (res.ci_low() < 0 < res.ci_high()) or (abs(res.test_value.value) < 1e-10)
+        if res.test_value.type == "ate":
+            return (res.ci_low() < 0 < res.ci_high()) or (abs(res.test_value.value) < self.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) < self.atol 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=self.atol)
         raise ValueError(f"Test Value type {res.test_value.type} is not valid for this TestOutcome")
 
 
 class ExactValue(SomeEffect):
     """An extension of TestOutcome representing that the expected causal effect should be a specific value."""
 
-    def __init__(self, value: float, tolerance: float = None):
+    def __init__(self, value: float, atol: float = None):
         self.value = value
-        if tolerance is None:
-            self.tolerance = value * 0.05
+        if atol is None:
+            self.atol = value * 0.05
         else:
-            self.tolerance = tolerance
+            self.atol = atol
 
     def apply(self, res: CausalTestResult) -> bool:
         if res.ci_valid():
-            return super().apply(res) and np.isclose(res.test_value.value, self.value, atol=self.tolerance)
-        return np.isclose(res.test_value.value, self.value, atol=self.tolerance)
+            return super().apply(res) and np.isclose(res.test_value.value, self.value, atol=self.atol)
+        return np.isclose(res.test_value.value, self.value, atol=self.atol)
 
     def __str__(self):
-        return f"ExactValue: {self.value}±{self.tolerance}"
+        return f"ExactValue: {self.value}±{self.atol}"
 
 
 class Positive(SomeEffect):
@@ -74,6 +88,7 @@ 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
+        # Dead code but necessary for pylint
         raise ValueError(f"Test Value type {res.test_value.type} is not valid for this TestOutcome")
 
 
@@ -87,4 +102,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
+        # Dead code but necessary for pylint
         raise ValueError(f"Test Value type {res.test_value.type} is not valid for this TestOutcome")

causal_testing/testing/causal_test_result.py

@@ -43,18 +43,27 @@ def __init__(
             self.effect_modifier_configuration = {}
 
     def __str__(self):
+        def push(s, inc="  "):
+            return inc + str(s).replace("\n", "\n" + inc)
+
+        result_str = str(self.test_value.value)
+        if "\n" in result_str:
+            result_str = "\n" + push(self.test_value.value)
         base_str = (
             f"Causal Test Result\n==============\n"
             f"Treatment: {self.estimator.treatment}\n"
             f"Control value: {self.estimator.control_value}\n"
             f"Treatment value: {self.estimator.treatment_value}\n"
             f"Outcome: {self.estimator.outcome}\n"
             f"Adjustment set: {self.adjustment_set}\n"
-            f"{self.test_value.type}: {self.test_value.value}\n"
+            f"{self.test_value.type}: {result_str}\n"
         )
         confidence_str = ""
         if self.confidence_intervals:
-            confidence_str += f"Confidence intervals: {self.confidence_intervals}\n"
+            ci_str = " " + str(self.confidence_intervals)
+            if "\n" in ci_str:
+                ci_str = " " + push(pd.DataFrame(self.confidence_intervals).transpose().to_string(header=False))
+            confidence_str += f"Confidence intervals:{ci_str}\n"
         return base_str + confidence_str
 
     def to_dict(self):
@@ -76,14 +85,14 @@ def to_dict(self):
 
     def ci_low(self):
         """Return the lower bracket of the confidence intervals."""
-        if self.confidence_intervals and all(self.confidence_intervals):
-            return min(self.confidence_intervals)
+        if self.confidence_intervals:
+            return self.confidence_intervals[0]
         return None
 
     def ci_high(self):
         """Return the higher bracket of the confidence intervals."""
-        if self.confidence_intervals and all(self.confidence_intervals):
-            return max(self.confidence_intervals)
+        if self.confidence_intervals:
+            return self.confidence_intervals[1]
         return None
 
     def ci_valid(self) -> bool:

causal_testing/testing/estimators.py

@@ -335,10 +335,21 @@ def estimate_unit_ate(self) -> float:
         :return: The unit average treatment effect and the 95% Wald confidence intervals.
         """
         model = self._run_linear_regression()
-        assert self.treatment in model.params, f"{self.treatment} not in {model.params}"
-        unit_effect = model.params[[self.treatment]].values[0]  # Unit effect is the coefficient of the treatment
-        [ci_low, ci_high] = self._get_confidence_intervals(model)
-
+        newline = "\n"
+        print(model.conf_int())
+        treatment = [self.treatment]
+        if str(self.df.dtypes[self.treatment]) == "object":
+            design_info = dmatrix(self.formula.split("~")[1], self.df).design_info
+            treatment = design_info.column_names[design_info.term_name_slices[self.treatment]]
+        assert set(treatment).issubset(
+            model.params.index.tolist()
+        ), f"{treatment} not in\n{'  '+str(model.params.index).replace(newline, newline+'  ')}"
+        unit_effect = model.params[treatment]  # Unit effect is the coefficient of the treatment
+        [ci_low, ci_high] = self._get_confidence_intervals(model, treatment)
+        if str(self.df.dtypes[self.treatment]) != "object":
+            unit_effect = unit_effect[0]
+            ci_low = ci_low[0]
+            ci_high = ci_high[0]
         return unit_effect, [ci_low, ci_high]
 
     def estimate_ate(self) -> tuple[float, list[float, float], float]:
@@ -353,12 +364,6 @@ def estimate_ate(self) -> tuple[float, list[float, float], float]:
         # Create an empty individual for the control and treated
         individuals = pd.DataFrame(1, index=["control", "treated"], columns=model.params.index)
 
-        # This is a temporary hack
-        # for t in self.square_terms:
-        #     individuals[t + "^2"] = individuals[t] ** 2
-        # for a, b in self.product_terms:
-        #     individuals[f"{a}*{b}"] = individuals[a] * individuals[b]
-
         # It is ABSOLUTELY CRITICAL that these go last, otherwise we can't index
         # the effect with "ate = t_test_results.effect[0]"
         individuals.loc["control", [self.treatment]] = self.control_value
@@ -424,35 +429,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.
 
@@ -472,22 +448,16 @@ 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"]]
-        for col in treatment_and_adjustments_cols:
-            if str(treatment_and_adjustments_cols.dtypes[col]) == "object":
-                treatment_and_adjustments_cols = pd.get_dummies(
-                    treatment_and_adjustments_cols, columns=[col], drop_first=True
-                )
         model = smf.ols(formula=self.formula, data=self.df).fit()
         return model
 
-    def _get_confidence_intervals(self, model):
+    def _get_confidence_intervals(self, model, treatment):
         confidence_intervals = model.conf_int(alpha=0.05, cols=None)
         ci_low, ci_high = (
-            confidence_intervals[0][[self.treatment]],
-            confidence_intervals[1][[self.treatment]],
+            confidence_intervals[0].loc[treatment],
+            confidence_intervals[1].loc[treatment],
         )
-        return [ci_low.values[0], ci_high.values[0]]
+        return [ci_low, ci_high]
 
 
 class InstrumentalVariableEstimator(Estimator):