Merge branch 'main' into json_concrete_param

# Conflicts:
#	causal_testing/json_front/json_class.py

Author: christopher-wild

.github/workflows/ci-tests.yaml

@@ -24,7 +24,7 @@ jobs:
         run: |
           conda install -c conda-forge pygraphviz
           python --version
-          pip install -e .
+          pip install -e . --no-cache-dir
           pip install -e .[test]
           pip install pytest pytest-cov
         shell: bash -l {0}

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,17 @@ 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])
+        if len(self.data) == 0:
+            raise ValueError(
+                "No data found, either provide a path to a file containing data or manually populate the .data "
+                "attribute with a dataframe before calling .setup()"
+            )
         self._populate_metas()
 
     def _create_abstract_test_case(self, test, mutates, effects):
@@ -146,6 +158,7 @@ def run_json_tests(self, effects: dict, estimators: dict, f_flag: bool = False,
                     + f"control value = {test['control_value']}, treatment value = {test['treatment_value']} \n"
                     + f"Result: {'FAILED' if failed else 'Passed'}"
                 )
+                print(msg)
                 self._append_to_file(msg, logging.INFO)
 
     def _run_coefficient_test(self, test: dict, f_flag: bool, effects: dict):
@@ -225,15 +238,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
@@ -257,13 +261,11 @@ def _execute_test_case(
         causal_test_engine, estimation_model = self._setup_test(
             causal_test_case, test, test["conditions"] if "conditions" in test else None
         )
-        causal_test_result = causal_test_engine.execute_test(
-            estimation_model, causal_test_case, estimate_type=causal_test_case.estimate_type
-        )
+        causal_test_result = causal_test_engine.execute_test(estimation_model, causal_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()}"
@@ -378,7 +380,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_engine.py

@@ -81,7 +81,6 @@ def execute_test_suite(self, test_suite: CausalTestSuite) -> list[CausalTestResu
 
             estimators = test_suite[edge]["estimators"]
             tests = test_suite[edge]["tests"]
-            estimate_type = test_suite[edge]["estimate_type"]
             results = {}
             for estimator_class in estimators:
                 causal_test_results = []
@@ -96,16 +95,14 @@ def execute_test_suite(self, test_suite: CausalTestSuite) -> list[CausalTestResu
                     )
                     if estimator.df is None:
                         estimator.df = self.scenario_execution_data_df
-                    causal_test_result = self._return_causal_test_results(estimate_type, estimator, test)
+                    causal_test_result = self._return_causal_test_results(estimator, test)
                     causal_test_results.append(causal_test_result)
 
                 results[estimator_class.__name__] = causal_test_results
             test_suite_results[edge] = results
         return test_suite_results
 
-    def execute_test(
-        self, estimator: type(Estimator), causal_test_case: CausalTestCase, estimate_type: str = "ate"
-    ) -> CausalTestResult:
+    def execute_test(self, estimator: type(Estimator), causal_test_case: CausalTestCase) -> CausalTestResult:
         """Execute a causal test case and return the causal test result.
 
         Test case execution proceeds with the following steps:
@@ -120,7 +117,6 @@ def execute_test(
 
         :param estimator: A reference to an Estimator class.
         :param causal_test_case: The CausalTestCase object to be tested
-        :param estimate_type: A string which denotes the type of estimate to return, ATE or CATE.
         :return causal_test_result: A CausalTestResult for the executed causal test case.
         """
         if self.scenario_execution_data_df.empty:
@@ -142,18 +138,17 @@ def execute_test(
         if self._check_positivity_violation(variables_for_positivity):
             raise ValueError("POSITIVITY VIOLATION -- Cannot proceed.")
 
-        causal_test_result = self._return_causal_test_results(estimate_type, estimator, causal_test_case)
+        causal_test_result = self._return_causal_test_results(estimator, causal_test_case)
         return causal_test_result
 
-    def _return_causal_test_results(self, estimate_type, estimator, causal_test_case):
+    def _return_causal_test_results(self, estimator, causal_test_case):
         """Depending on the estimator used, calculate the 95% confidence intervals and return in a causal_test_result
 
-        :param estimate_type: A string which denotes the type of estimate to return
         :param estimator: An Estimator class object
         :param causal_test_case: The concrete test case to be executed
         :return: a CausalTestResult object containing the confidence intervals
         """
-        if estimate_type == "cate":
+        if causal_test_case.estimate_type == "cate":
             logger.debug("calculating cate")
             if not hasattr(estimator, "estimate_cates"):
                 raise NotImplementedError(f"{estimator.__class__} has no CATE method.")
@@ -165,7 +160,7 @@ def _return_causal_test_results(self, estimate_type, estimator, causal_test_case
                 effect_modifier_configuration=causal_test_case.effect_modifier_configuration,
                 confidence_intervals=confidence_intervals,
             )
-        elif estimate_type == "risk_ratio":
+        elif causal_test_case.estimate_type == "risk_ratio":
             logger.debug("calculating risk_ratio")
             risk_ratio, confidence_intervals = estimator.estimate_risk_ratio()
             causal_test_result = CausalTestResult(
@@ -174,7 +169,7 @@ def _return_causal_test_results(self, estimate_type, estimator, causal_test_case
                 effect_modifier_configuration=causal_test_case.effect_modifier_configuration,
                 confidence_intervals=confidence_intervals,
             )
-        elif estimate_type == "coefficient":
+        elif causal_test_case.estimate_type == "coefficient":
             logger.debug("calculating coefficient")
             coefficient, confidence_intervals = estimator.estimate_unit_ate()
             causal_test_result = CausalTestResult(
@@ -183,7 +178,7 @@ def _return_causal_test_results(self, estimate_type, estimator, causal_test_case
                 effect_modifier_configuration=causal_test_case.effect_modifier_configuration,
                 confidence_intervals=confidence_intervals,
             )
-        elif estimate_type == "ate":
+        elif causal_test_case.estimate_type == "ate":
             logger.debug("calculating ate")
             ate, confidence_intervals = estimator.estimate_ate()
             causal_test_result = CausalTestResult(
@@ -194,7 +189,7 @@ def _return_causal_test_results(self, estimate_type, estimator, causal_test_case
             )
             # causal_test_result = CausalTestResult(minimal_adjustment_set, ate, confidence_intervals)
             # causal_test_result.apply_test_oracle_procedure(self.causal_test_case.expected_causal_effect)
-        elif estimate_type == "ate_calculated":
+        elif causal_test_case.estimate_type == "ate_calculated":
             logger.debug("calculating ate")
             ate, confidence_intervals = estimator.estimate_ate_calculated()
             causal_test_result = CausalTestResult(
@@ -206,7 +201,9 @@ def _return_causal_test_results(self, estimate_type, estimator, causal_test_case
             # causal_test_result = CausalTestResult(minimal_adjustment_set, ate, confidence_intervals)
             # causal_test_result.apply_test_oracle_procedure(self.causal_test_case.expected_causal_effect)
         else:
-            raise ValueError(f"Invalid estimate type {estimate_type}, expected 'ate', 'cate', or 'risk_ratio'")
+            raise ValueError(
+                f"Invalid estimate type {causal_test_case.estimate_type}, expected 'ate', 'cate', or 'risk_ratio'"
+            )
         return causal_test_result
 
     def _check_positivity_violation(self, variables_list):

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")