API Reference

This page provides the complete API documentation for CausalBoundingEngine.

Core Classes

Scenario

class causalboundingengine.scenario.Scenario(X, Y, Z=None)[source]

Bases: object

Base class for scenarios in the Causal Bounding Engine.

Each subclass must define the AVAILABLE_ALGORITHMS dictionary, which maps query types (‘ATE’, ‘PNS’) to supported algorithm classes.

Upon initialization, this class creates AlgorithmDispatcher instances for both ATE and PNS queries using the provided data.

AVAILABLE_ALGORITHMS = {}

get_algorithms(query_type)[source]

Returns a list of available algorithm names for a given query type.

Parameters:: query_type (str) – Either ‘ATE’ or ‘PNS’.
Returns:: Names of supported algorithms.
Return type:: list[str]

AlgorithmDispatcher

class causalboundingengine.scenario.AlgorithmDispatcher(scenario, query_type)[source]

Bases: object

A dynamic dispatcher that exposes algorithms as methods via __getattr__.

scenario

The scenario instance this dispatcher is bound to.

Type:: Scenario

query_type

The type of causal query (‘ATE’ or ‘PNS’).

Type:: str

Data

class causalboundingengine.utils.data.Data(X, Y, Z=None)[source]

Bases: object

unpack()[source]

Algorithm Base Class

Algorithm

class causalboundingengine.algorithms.algorithm.Algorithm[source]

Bases: ABC

bound_ATE(*args, **kwargs) → tuple[float, float][source]

bound_PNS(*args, **kwargs) → tuple[float, float][source]

Scenarios

BinaryConf

class causalboundingengine.scenarios.BinaryConf(X, Y, Z=None)[source]

Bases: Scenario

AVAILABLE_ALGORITHMS = {'ATE': {'autobound': <class 'causalboundingengine.algorithms.autobound.Autobound'>, 'causaloptim': <class 'causalboundingengine.algorithms.causaloptim.CausalOptim'>, 'entropybounds': <class 'causalboundingengine.algorithms.entropybounds.Entropybounds'>, 'manski': <class 'causalboundingengine.algorithms.manski.Manski'>, 'zaffalonbounds': <class 'causalboundingengine.algorithms.zaffalonbounds.Zaffalonbounds'>}, 'PNS': {'autobound': <class 'causalboundingengine.algorithms.autobound.Autobound'>, 'causaloptim': <class 'causalboundingengine.algorithms.causaloptim.CausalOptim'>, 'entropybounds': <class 'causalboundingengine.algorithms.entropybounds.Entropybounds'>, 'tianpearl': <class 'causalboundingengine.algorithms.tianpearl.TianPearl'>, 'zaffalonbounds': <class 'causalboundingengine.algorithms.zaffalonbounds.Zaffalonbounds'>}}

BinaryIV

class causalboundingengine.scenarios.BinaryIV(X, Y, Z=None)[source]

Bases: Scenario

AVAILABLE_ALGORITHMS = {'ATE': {'autobound': <class 'causalboundingengine.algorithms.autobound.Autobound'>, 'causaloptim': <class 'causalboundingengine.algorithms.causaloptim.CausalOptim'>, 'zaffalonbounds': <class 'causalboundingengine.algorithms.zaffalonbounds.Zaffalonbounds'>}, 'PNS': {'autobound': <class 'causalboundingengine.algorithms.autobound.Autobound'>, 'causaloptim': <class 'causalboundingengine.algorithms.causaloptim.CausalOptim'>, 'zaffalonbounds': <class 'causalboundingengine.algorithms.zaffalonbounds.Zaffalonbounds'>}}

ContIV

class causalboundingengine.scenarios.ContIV(X, Y, Z=None)[source]

Bases: Scenario

AVAILABLE_ALGORITHMS = {'ATE': {'zhangbareinboim': <class 'causalboundingengine.algorithms.zhangbareinboim.ZhangBareinboim'>}}

Core Algorithms

Manski

class causalboundingengine.algorithms.manski.Manski[source]: Bases: Algorithm

TianPearl

class causalboundingengine.algorithms.tianpearl.TianPearl[source]: Bases: Algorithm

Autobound

class causalboundingengine.algorithms.autobound.Autobound[source]

Bases: Algorithm

static run_experiment(query, dagstring, unob, joint_probs)[source]

Run the AutoBound experiment. :param dag: The directed acyclic graph representing the causal structure. :type dag: DAG :param df: DataFrame containing the data for the experiment. :type df: pd.DataFrame

Returns:: (lower_bound, upper_bound) from AutoBound
Return type:: tuple

EntropyBounds

class causalboundingengine.algorithms.entropybounds.Entropybounds[source]: Bases: Algorithm

ZhangBareinboim

class causalboundingengine.algorithms.zhangbareinboim.ZhangBareinboim[source]: Bases: Algorithm

External Engine Algorithms

Causaloptim

Zaffalonbounds

class causalboundingengine.algorithms.zaffalonbounds.Zaffalonbounds[source]

Bases: Algorithm

static run_experiment(query, df, isConf=False)[source]

Utility Classes

AlgUtil

class causalboundingengine.utils.alg_util.AlgUtil[source]

Bases: object

Utility class for algorithms.

static get_trivial_Ceils(query)[source]

Get trivial Ceils for the given query.

Parameters:: query (str) – The query type (e.g., ‘ATE’ or ‘PNS’).
Returns:: A tuple containing the lower and upper bounds.
Return type:: tuple

static flatten_bounds_to_trivial_ceils(query, bound_lower, bound_upper, failed)[source]

static flatten_bounds_to_trivial_ceils_vectorized(query, bound_lower, bound_upper, failed)[source]

Vectorized version: Flatten bounds to trivial ceils for pandas Series. :param query: The query type (e.g., ‘ATE’ or ‘PNS’). :type query: str :param bound_lower: Lower bounds. :type bound_lower: pd.Series :param bound_upper: Upper bounds. :type bound_upper: pd.Series :param failed: Boolean mask for failed bounds. :type failed: pd.Series

Returns:: (flattened_lower, flattened_upper) as pd.Series
Return type:: tuple

Constants and Types

Common Parameters

Many algorithms accept these standard parameters:

X (np.ndarray): Binary treatment array with values in {0, 1}
Y (np.ndarray): Binary outcome array with values in {0, 1}
Z (np.ndarray, optional): Binary instrument array with values in {0, 1}

Algorithm-Specific Parameters

EntropyBounds:

theta (float): Information constraint level for mutual information

Causaloptim:

r_path (str, optional): Custom path to R executable

Return Values

All bound computation methods return:

Tuple[float, float]: (lower_bound, upper_bound) where lower_bound <= upper_bound

For ATE bounds: -1.0 <= lower_bound <= upper_bound <= 1.0

For PNS bounds: 0.0 <= lower_bound <= upper_bound <= 1.0

Exception Handling

Algorithm Failures

When algorithms fail, they return trivial bounds instead of raising exceptions:

ATE: (-1.0, 1.0)
PNS: (0.0, 1.0)

Missing Dependencies

Algorithms with missing dependencies raise ImportError:

try:
    bounds = scenario.ATE.causaloptim()
except ImportError as e:
    print(f"R support not available: {e}")

Invalid Parameters

Invalid parameters raise ValueError or TypeError:

try:
    bounds = scenario.ATE.entropybounds(theta=-1.0)  # Invalid theta (must be in [0,1])
except ValueError as e:
    print(f"Invalid parameter: {e}")

Usage Patterns

Basic Pattern

from causalboundingengine.scenarios import BinaryConf
import numpy as np

# Data preparation
X = np.array([0, 1, 1, 0, 1])
Y = np.array([1, 0, 1, 0, 1])

# Scenario creation
scenario = BinaryConf(X, Y)

# Bound computation
bounds = scenario.ATE.manski()
print(f"Bounds: {bounds}")

Dynamic Algorithm Access

# Get available algorithms
algorithms = scenario.get_algorithms('ATE')

# Use algorithms dynamically
for alg_name in algorithms:
    try:
        alg_func = getattr(scenario.ATE, alg_name)
        bounds = alg_func()
        print(f"{alg_name}: {bounds}")
    except Exception as e:
        print(f"{alg_name} failed: {e}")

Error Handling Pattern

import logging

# Enable warnings for algorithm failures
logging.basicConfig(level=logging.WARNING)

# Robust computation with fallback
def compute_bounds_robust(scenario, preferred_alg='autobound', fallback_alg='manski'):
    try:
        return getattr(scenario.ATE, preferred_alg)()
    except Exception:
        return getattr(scenario.ATE, fallback_alg)()

bounds = compute_bounds_robust(scenario)

Extending the API

Custom Algorithms

Implement the Algorithm interface:

from causalboundingengine.algorithms.algorithm import Algorithm

class MyAlgorithm(Algorithm):
    def _compute_ATE(self, X, Y, **kwargs):
        # Your implementation
        return lower_bound, upper_bound

    def _compute_PNS(self, X, Y, **kwargs):
        # Your implementation
        return lower_bound, upper_bound

Custom Scenarios

Extend the Scenario class:

from causalboundingengine.scenario import Scenario

class MyScenario(Scenario):
    AVAILABLE_ALGORITHMS = {
        'ATE': {'my_algorithm': MyAlgorithm},
        'PNS': {'my_algorithm': MyAlgorithm},
    }

Version Information

Access version information:

import causalboundingengine
print(causalboundingengine.__version__)  # Package version

Module Structure

The package is organized as follows:

causalboundingengine/
├── __init__.py              # Package initialization
├── scenario.py              # Base scenario and dispatcher classes
├── scenarios.py             # Concrete scenario implementations
├── algorithms/              # Algorithm implementations
│   ├── __init__.py
│   ├── algorithm.py         # Base algorithm class
│   ├── manski.py           # Manski bounds
│   ├── tianpearl.py        # Tian-Pearl bounds
│   ├── autobound.py        # Autobound algorithm
│   ├── entropybounds.py    # Entropy-based bounds
│   ├── causaloptim.py      # R-based Causaloptim
│   ├── zaffalonbounds.py   # Java-based Zaffalonbounds
│   └── zhangbareinboim.py  # Zhang-Bareinboim bounds
└── utils/                   # Utility functions
    ├── __init__.py
    ├── data.py             # Data handling
    ├── alg_util.py         # Algorithm utilities
    └── r_utils.py          # R integration utilities