causal_testing.main

This module contains the classes that executes the Causal Testing Framework.

Module Contents

Classes

Command	Enum for supported CTF commands.
CausalTestingPaths	Class for managing paths for causal testing inputs and outputs.
CausalTestingFramework	Main class for running causal tests.

Functions

setup_logging(→ None)	Set up logging configuration.
parse_args(→ argparse.Namespace)	Parse command line arguments.

Attributes

logger

causal_testing.main.logger

class causal_testing.main.Command(*args, **kwds)

Bases: enum.Enum

Enum for supported CTF commands.

TEST = 'test'

GENERATE = 'generate'

class causal_testing.main.CausalTestingPaths(dag_path: str | pathlib.Path, data_paths: List[str | pathlib.Path], test_config_path: str | pathlib.Path, output_path: str | pathlib.Path)

Class for managing paths for causal testing inputs and outputs.

Parameters:

• dag_path – Path to the DAG definition file

• data_paths – List of paths to input data files

• test_config_path – Path to the test configuration file

• output_path – Path where test results will be written

dag_path: pathlib.Path

data_paths: List[pathlib.Path]

test_config_path: pathlib.Path

output_path: pathlib.Path

validate_paths() → None

Validate existence of all input paths and writability of output path.

Raises:

FileNotFoundError if any required input file is missing.

class causal_testing.main.CausalTestingFramework(paths: CausalTestingPaths, ignore_cycles: bool = False, query: str | None = None)

Main class for running causal tests.

Parameters:

• paths – CausalTestingPaths object containing required file paths

• ignore_cycles – Flag to ignore cycles in the DAG

• query – Optional query string to filter the input dataframe

setup() → None

Set up the framework by loading DAG, runtime csv data, creating the scenario and causal specification.

Raises:

FileNotFoundError if required files are missing

load_dag() → causal_testing.specification.causal_dag.CausalDAG

Load the causal DAG from the specified file path.

_read_dataframe(data_path)

load_data(query: str | None = None) → pandas.DataFrame

Load and combine all data sources with optional filtering.

Parameters:

query – Optional pandas query string to filter the loaded data

Returns:

Combined pandas DataFrame containing all loaded and filtered data

create_variables() → None

Create variable objects from DAG nodes based on their connectivity.

load_tests() → None

Load and prepare test configurations from file.

create_base_test(test: dict) → causal_testing.testing.base_test_case.BaseTestCase

Create base test case from test configuration.

Parameters:

test – Dictionary containing test configuration parameters

Returns:

BaseTestCase object

Raises:

KeyError if required variables are not found in inputs or outputs

create_test_cases(test_configs: dict) → List[causal_testing.testing.causal_test_case.CausalTestCase]

Create test case objects from configuration dictionary.

Parameters:

test_configs – Dictionary containing test configurations

Returns:

List of CausalTestCase objects containing the initialised test cases

Raises:

KeyError if required variables are not found

Raises:

ValueError if invalid test configuration is provided

create_causal_test(test: dict, base_test: causal_testing.testing.base_test_case.BaseTestCase) → causal_testing.testing.causal_test_case.CausalTestCase

Create causal test case from test configuration and base test.

Parameters:

• test – Dictionary containing test configuration parameters

• base_test – BaseTestCase object

Returns:

CausalTestCase object

Raises:

ValueError if invalid estimator or configuration is provided

run_tests_in_batches(batch_size: int = 100, silent: bool = False, adequacy: bool = False, bootstrap_size: int = 100) → List[causal_testing.testing.causal_test_result.CausalTestResult]

Run tests in batches to reduce memory usage.

Parameters:

• batch_size – Number of tests to run in each batch

• silent – Whether to suppress errors

• adequacy – Whether to calculate causal test adequacy (defaults to False)

• bootstrap_size – The number of bootstrap samples to use when calculating causal test adequacy (defaults to 100)

Returns:

List of all test results

Raises:

ValueError if no tests are loaded

run_tests(silent: bool = False, adequacy: bool = False, bootstrap_size: int = 100) → List[causal_testing.testing.causal_test_result.CausalTestResult]

Run all test cases and return their results.

Parameters:

• silent – Whether to suppress errors

• adequacy – Whether to calculate causal test adequacy (defaults to False)

• bootstrap_size – The number of bootstrap samples to use when calculating causal test adequacy (defaults to 100)

Returns:

List of CausalTestResult objects

Raises:

ValueError if no tests are loaded

Raises:

Exception if test execution fails

save_results(results: List[causal_testing.testing.causal_test_result.CausalTestResult], output_path: str = None) → list

Save test results to JSON file in the expected format.

causal_testing.main.setup_logging(verbose: bool = False) → None

Set up logging configuration.

causal_testing.main.parse_args(args: Sequence[str] | None = None) → argparse.Namespace

Parse command line arguments.