Jmafoster1/dag evaluation

causal_testing/__main__.py

@@ -1,6 +1,7 @@
 """This module contains the main entrypoint functionality to the Causal Testing Framework."""
 
 import argparse
+import json
 import logging
 from enum import Enum
 from importlib.metadata import entry_points
@@ -10,7 +11,7 @@
 import pandas as pd
 
 from causal_testing.causal_testing_framework import CausalTestingFramework, read_dataframe
-from causal_testing.testing.metamorphic_relation import generate_causal_tests
+from causal_testing.specification.causal_dag import CausalDAG
 
 logger = logging.getLogger(__name__)
 
@@ -23,6 +24,7 @@ class Command(Enum):
     TEST = "test"
     GENERATE = "generate"
     DISCOVER = "discover"
+    EVALUATE = "evaluate"
 
 
 def setup_logging(level: str) -> None:
@@ -40,15 +42,6 @@ def parse_args(args: Optional[Sequence[str]] = None) -> argparse.Namespace:
         "A causal inference-driven framework for functional black-box testing of complex software.",
     )
 
-    main_parser.add_argument(
-        "-l",
-        "--log_level",
-        default="WARNING",
-        type=str.upper,
-        choices=["NONE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
-        help="Set the logging level (default: WARNING).",
-    )
-
     subparsers = main_parser.add_subparsers(
         help="The action you want to run - call `causal_testing {action} -h` for further details", dest="command"
     )
@@ -57,24 +50,6 @@ def parse_args(args: Optional[Sequence[str]] = None) -> argparse.Namespace:
     parser_generate = subparsers.add_parser(Command.GENERATE.value, help="Generate causal tests from a DAG")
     parser_generate.add_argument("-D", "--dag-path", help="Path to the DAG file (.dot)", required=True)
     parser_generate.add_argument("-o", "--output", help="Path for output file (.json)", required=True)
-    parser_generate.add_argument(
-        "-e",
-        "--estimator",
-        help="The name of the estimator class to use when evaluating tests (defaults to LinearRegressionEstimator)",
-        default="LinearRegressionEstimator",
-    )
-    parser_generate.add_argument(
-        "-T",
-        "--effect-type",
-        help="The effect type to estimate {direct, total}",
-        default="direct",
-    )
-    parser_generate.add_argument(
-        "-E",
-        "--estimate-type",
-        help="The estimate type to use when evaluating tests (defaults to coefficient)",
-        default="coefficient",
-    )
     parser_generate.add_argument(
         "-i", "--ignore-cycles", help="Ignore cycles in DAG", action="store_true", default=False
     )
@@ -87,11 +62,10 @@ def parse_args(args: Optional[Sequence[str]] = None) -> argparse.Namespace:
     parser_test.add_argument("-D", "--dag-path", help="Path to the DAG file (.dot)", required=True)
     parser_test.add_argument("-o", "--output", help="Path for output file (.json)", required=True)
     parser_test.add_argument("-i", "--ignore-cycles", help="Ignore cycles in DAG", action="store_true", default=False)
-    parser_test.add_argument("-d", "--data-paths", help="Paths to data files (.csv)", nargs="+", required=True)
     parser_test.add_argument("-t", "--test-config", help="Path to test configuration file (.json)", required=True)
     parser_test.add_argument("-q", "--query", help="Query string to filter data (e.g. 'age > 18')", type=str)
     parser_test.add_argument(
-        "-a", "--adequacy", help="Calculate causal test adequacy for each test case", action="store_true", default=False
+        "-A", "--adequacy", help="Calculate causal test adequacy for each test case", action="store_true", default=False
     )
     parser_test.add_argument(
         "-b",
@@ -108,18 +82,34 @@ def parse_args(args: Optional[Sequence[str]] = None) -> argparse.Namespace:
         default=False,
     )
 
+    # DAG evaluation
+    parser_evaluate = subparsers.add_parser(
+        Command.EVALUATE.value, help="Evaluate how well a causal DAG fits a dataset"
+    )
+    parser_evaluate.add_argument("-D", "--dag-path", help="Path to the DAG file (.dot)", required=True)
+    parser_evaluate.add_argument("-o", "--output", help="Path for output file (.csv)", required=True)
+    parser_evaluate.add_argument(
+        "-i", "--ignore-cycles", help="Ignore cycles in DAG", action="store_true", default=False
+    )
+    parser_evaluate.add_argument("-q", "--query", help="Query string to filter data (e.g. 'age > 18')", type=str)
+    parser_evaluate.add_argument(
+        "-b",
+        "--adequacy-bootstrap-size",
+        dest="bootstrap_size",
+        help="Number of bootstrap samples for causal test adequacy. Defaults to 100",
+        type=int,
+        default=100,
+    )
+    parser_evaluate.add_argument(
+        "-s",
+        "--silent",
+        action="store_true",
+        help="Do not crash on error. If set to true, errors are recorded as test results.",
+        default=False,
+    )
+
     # Discovery
     parser_discover = subparsers.add_parser(Command.DISCOVER.value, help="Discover causal structures from data")
-    parser_discover.add_argument("-d", "--data-paths", help="Paths to data files (.csv)", nargs="+", required=True)
-    parser_discover.add_argument(
-        "-a",
-        "--alpha",
-        help=(
-            "The significance level of the confidence intervals used to determine causality. "
-            "This should be a value between 0 and 1. Defaults to 0.05 for 95%% confidence intervals."
-        ),
-        default=0.05,
-    )
     parser_discover.add_argument(
         "-t",
         "--technique",
@@ -147,6 +137,26 @@ def parse_args(args: Optional[Sequence[str]] = None) -> argparse.Namespace:
         default=[],
     )
 
+    for parser in [parser_generate, parser_discover, parser_test, parser_evaluate]:
+        parser.add_argument(
+            "-l",
+            "--log_level",
+            default="WARNING",
+            type=str.upper,
+            choices=["NONE", "DEBUG", "INFO", "WARNING", "ERROR", "CRITICAL"],
+            help="Set the logging level (default: WARNING).",
+        )
+        parser.add_argument(
+            "-a",
+            "--alpha",
+            help=(
+                "The significance level of the confidence intervals used to determine causality. "
+                "This should be a value between 0 and 1. Defaults to 0.05 for 95%% confidence intervals."
+            ),
+            default=0.05,
+        )
+        parser.add_argument("-d", "--data-paths", help="Paths to data files (.csv)", nargs="+", required=True)
+
     args = main_parser.parse_args(args)
 
     # Assume the user wants test adequacy if they're setting bootstrap_size
@@ -175,16 +185,14 @@ def main() -> None:
     match args.command:
         case Command.GENERATE:
             logging.info("Generating causal tests")
-            generate_causal_tests(
-                args.dag_path,
-                args.output,
-                args.ignore_cycles,
-                args.threads,
-                effect_type=args.effect_type,
-                estimate_type=args.estimate_type,
-                estimator=args.estimator,
+            df = pd.concat(read_dataframe(path) for path in args.data_paths)
+            causal_dag = CausalDAG(args.dag_path, ignore_cycles=args.ignore_cycles, datatypes=df.dtypes)
+            causal_tests = causal_dag.generate_causal_tests(
+                threads=args.threads,
                 skip=False,
             )
+            with open(args.output, "w", encoding="utf-8") as f:
+                json.dump({"tests": [test.to_dict() for test in causal_tests]}, f)
             logging.info("Causal test generation completed successfully.")
 
         case Command.DISCOVER:
@@ -246,6 +254,23 @@ def main() -> None:
             framework.save_results(args.output)
 
             logging.info("Causal testing completed successfully.")
+        case Command.EVALUATE:
+            # Create and setup framework
+            framework = CausalTestingFramework()
+
+            framework.setup(
+                dag_path=args.dag_path,
+                data_paths=args.data_paths,
+                test_cases_path=args.test_config,
+                query=args.query,
+                ignore_cycles=args.ignore_cycles,
+            )
+
+            logging.info("Running tests on entire dataset")
+            results = framework.evaluate_dag(alpha=args.alpha, bootstrap_size=args.bootstrap_size)
+            logging.info("Causal testing completed successfully.")
+            logging.info("Running tests on bootstrap samples")
+            results.to_csv(args.output)
 
 
 if __name__ == "__main__":

causal_testing/causal_testing_framework.py

@@ -14,6 +14,7 @@
 from causal_testing.specification.variable import Input, Output
 from causal_testing.testing.base_test_case import BaseTestCase
 from causal_testing.testing.causal_test_case import CausalTestCase
+from causal_testing.testing.causal_test_result import TestOutcome
 
 logger = logging.getLogger(__name__)
 
@@ -265,6 +266,55 @@ def run_tests(self, silent: bool = False, adequacy: bool = False, bootstrap_size
                 self.df, suppress_estimation_errors=silent, adequacy=adequacy, bootstrap_size=bootstrap_size
             )
 
+    def evaluate_dag(self, bootstrap_size: bool = 100, alpha: float = 0.05) -> pd.Series:
+        """
+        Calculate confidence intervals for how well a causal DAG fits a dataset by repeatedly resampling the dataset
+        and executing the causal tests.
+        Confidence intervals are then calcualted for how many tests pass, fail, or are inestimable.
+
+        :param bootstrap_size: The number of bootstrap samples to use when calculating causal test adequacy
+                               (defaults to 100)
+        :param alpha: The significance level to use when calculating confidence intervals.
+                      (defaults to 0.05).
+        """
+        self.run_tests(silent=True, adequacy=False)
+        results = {
+            test_outcome.name: len(
+                [test for test in self.test_cases if test.result and test.result.outcome == test_outcome]
+            )
+            for test_outcome in TestOutcome
+        }
+
+        sample_results = []
+        for sample_index in range(bootstrap_size):
+            test_outcomes = {test_outcome: 0 for test_outcome in TestOutcome}
+            for test_case in self.test_cases:
+                if test_case.skip:
+                    continue
+                effect_estimate = test_case.estimate_effect(
+                    df=self.df.sample(len(self.df), replace=True, random_state=sample_index)
+                )
+                if effect_estimate:
+                    if test_case.expected_causal_effect.apply(effect_estimate):
+                        test_outcomes[TestOutcome.PASS] += 1
+                    else:
+                        test_outcomes[TestOutcome.FAIL] += 1
+                else:
+                    test_outcomes[TestOutcome.INESTIMABLE] += 1
+            sample_results.append(test_outcomes)
+
+        sample_results = pd.DataFrame(sample_results)
+
+        # Calculate the confidence interval of each column
+        ci_low_inx = round((alpha / 2) * bootstrap_size)
+        ci_high_inx = round(((1 - alpha) / 2) * bootstrap_size)
+        for outcome in TestOutcome:
+            data = sorted(sample_results[outcome])
+            results[f"{outcome.name}_ci_low"] = data[ci_low_inx]
+            results[f"{outcome.name}_ci_high"] = data[ci_high_inx]
+
+        return pd.Series(results).sort_index()
+
     def save_results(self, output_path) -> list:
         """Save test results to JSON file in the expected format."""
         logger.info(f"Saving results to {output_path}")
@@ -301,15 +351,11 @@ def save_results(self, output_path) -> list:
                 result = test_case.result
                 result_index += 1
 
-                test_passed = (
-                    test_case.expected_causal_effect.apply(result) if result.effect_estimate is not None else False
-                )
-
                 output = {
                     **base_output,
                     "formula": test_case.estimator.formula if hasattr(test_case.estimator, "formula") else None,
                     "skip": False,
-                    "passed": test_passed,
+                    "passed": test_case.result.outcome == TestOutcome.PASS,
                     "result": (
                         {
                             "treatment": test_case.estimator.base_test_case.treatment_variable.name,

causal_testing/discovery/abstract_discovery.py

@@ -6,7 +6,6 @@
 import re
 import warnings
 from abc import ABC, abstractmethod
-from enum import Enum
 from itertools import permutations
 
 import networkx as nx
@@ -19,9 +18,7 @@
 from causal_testing.specification.scenario import Scenario
 from causal_testing.testing.causal_effect import Negative, Positive
 from causal_testing.testing.causal_test_case import CausalTestCase
-from causal_testing.testing.metamorphic_relation import generate_metamorphic_relations
-
-TestResult = Enum("TestResult", [("PASS", 2), ("FAIL", 0), ("INESTIMABLE", 1)])
+from causal_testing.testing.causal_test_result import TestOutcome
 
 # Ignore warnings from statsmodels when we try to evaluate test cases
 warnings.simplefilter("ignore")
@@ -111,9 +108,9 @@ def effect_direction(self, test_case: CausalTestCase) -> str:
         if pd.api.types.is_numeric_dtype(
             self.df[test_case.base_test_case.treatment_variable.name]
         ) and pd.api.types.is_numeric_dtype(self.df[test_case.base_test_case.outcome_variable.name]):
-            if Negative().apply(test_case.result):
+            if Negative().apply(test_case.result.effect_estimate):
                 return "negative"
-            if Positive().apply(test_case.result):
+            if Positive().apply(test_case.result.effect_estimate):
                 return "positive"
         return None
 
@@ -148,15 +145,15 @@ def write_dot(self, individual: CausalDAG, output_file: str):
 
                     print(test)
 
-                    if test["result"] == TestResult.PASS:
+                    if test["result"] == TestOutcome.PASS:
                         print("  GREEN")
                         individual[test["treatment"]][test["outcome"]]["color"] = "green"
                         individual[test["treatment"]][test["outcome"]]["fontcolor"] = "green"
-                    elif test["result"] == TestResult.INESTIMABLE:
+                    elif test["result"] == TestOutcome.INESTIMABLE:
                         print("  ORANGE")
                         individual[test["treatment"]][test["outcome"]]["color"] = "orange"
                         individual[test["treatment"]][test["outcome"]]["fontcolor"] = "orange"
-                    elif test["result"] == TestResult.FAIL:
+                    elif test["result"] == TestOutcome.FAIL:
                         print("  RED")
                         individual[test["treatment"]][test["outcome"]]["color"] = "red"
                         individual[test["treatment"]][test["outcome"]]["fontcolor"] = "red"
@@ -166,13 +163,13 @@ def write_dot(self, individual: CausalDAG, output_file: str):
                     individual.add_edge(test["treatment"], test["outcome"], ignore_cycles=True)
                     individual[test["treatment"]][test["outcome"]]["style"] = "dashed"
                     individual[test["treatment"]][test["outcome"]]["label"] = test["effect"]
-                    if test["result"] == TestResult.PASS:
+                    if test["result"] == TestOutcome.PASS:
                         individual[test["treatment"]][test["outcome"]]["style"] = "invis"
                         individual[test["treatment"]][test["outcome"]]["constraint"] = False
-                    elif test["result"] == TestResult.INESTIMABLE:
+                    elif test["result"] == TestOutcome.INESTIMABLE:
                         individual[test["treatment"]][test["outcome"]]["color"] = "orange"
                         individual[test["treatment"]][test["outcome"]]["fontcolor"] = "orange"
-                    elif test["result"] == TestResult.FAIL:
+                    elif test["result"] == TestOutcome.FAIL:
                         individual[test["treatment"]][test["outcome"]]["color"] = "red"
                         individual[test["treatment"]][test["outcome"]]["fontcolor"] = "red"
                     else:
@@ -204,15 +201,8 @@ def evaluate_tests(self, causal_dag: CausalDAG) -> pd.DataFrame:
         ctf.create_variables()
         ctf.scenario = Scenario(list(ctf.variables["inputs"].values()) + list(ctf.variables["outputs"].values()))
 
-        ctf.test_cases = [
-            ctf.create_causal_test(
-                relation.to_json_stub(
-                    alpha=self.alpha,
-                    **self._json_stub_params(relation.base_test_case.outcome_variable),
-                )
-            )
-            for relation in generate_metamorphic_relations(causal_dag)
-        ]
+        causal_dag.datatypes = self.df.dtypes
+        ctf.test_cases = causal_dag.generate_causal_tests()
 
         results = []
 
@@ -222,9 +212,9 @@ def evaluate_tests(self, causal_dag: CausalDAG) -> pd.DataFrame:
                 results.append(
                     {
                         "result": (
-                            TestResult.PASS
-                            if test_case.expected_causal_effect.apply(test_case.result)
-                            else TestResult.FAIL
+                            TestOutcome.PASS
+                            if test_case.expected_causal_effect.apply(test_case.result.effect_estimate)
+                            else TestOutcome.FAIL
                         ),
                         "expected_effect": test_case.expected_causal_effect.__class__.__name__,
                         "treatment": test_case.base_test_case.treatment_variable.name,
@@ -235,7 +225,7 @@ def evaluate_tests(self, causal_dag: CausalDAG) -> pd.DataFrame:
             except np.linalg.LinAlgError:
                 results.append(
                     {
-                        "result": TestResult.INESTIMABLE,
+                        "result": TestOutcome.INESTIMABLE,
                         "expected_effect": test_case.expected_causal_effect.__class__.__name__,
                         "treatment": test_case.base_test_case.treatment_variable.name,
                         "outcome": test_case.base_test_case.outcome_variable.name,