Causal Outlier Propagation — An Approach to Structurally-Informed Data Degradation¶

This notebook explains the Causal Outlier Propagation Generator, a new addition to Badgers that uses causal graphical models (DAGs) as input scaffolding to produce structurally-informed outliers.

Instead of treating all columns independently, this generator respects the causal relationships between variables: an outlier injected into a root-cause variable cascades to its downstream effects, producing more realistic and semantically meaningful anomalies.

---

Table of Contents¶

1. Motivation: Why Causal Graphs?
2. Core Concepts: DAGs as Scaffolding
3. The Algorithm: Forward-Pass Propagation
4. Graph Conventions: Nodes, Columns, and Mapping
5. API Walkthrough
6. Topology Examples: Chain, Fork, Collider, Diamond
7. Error Handling and Edge Cases
8. Testing Strategy

1. Motivation: Why Causal Graphs? ¶

The Problem with Independent Outliers¶

Traditional outlier generators in Badgers (z-score, hypersphere, hypercube, histogram) treat each column independently. They add noise or extreme values to individual features without considering how those features relate to each other.

In real-world systems, variables are connected by causal relationships:

graph LR
    Temperature --> IceCreamSales
    Temperature --> CrimeRate
    IceCreamSales --> Revenue

An extreme temperature value should cascade: it affects ice cream sales, which in turn affects revenue. An independent outlier generator might spike revenue without touching temperature or sales — a physically impossible scenario.

The Solution: Causal Scaffolding¶

By providing a causal graph (a directed acyclic graph, or DAG) as input, the generator can:

1. Inject an outlier at a user-specified root-cause node
2. Propagate the perturbation to all descendant nodes along causal edges
3. Preserve the causal structure — only variables downstream of the perturbed node are affected

This produces structurally consistent outliers that test whether anomaly detectors catch semantically meaningful anomalies, not just marginal-statistic deviations.

2. Core Concepts: DAGs as Scaffolding ¶

What is a DAG?¶

A Directed Acyclic Graph (DAG) is a graph with:

• Directed edges (arrows): $A \rightarrow B$ means "A causes B"
• No cycles: you can't follow arrows and return to where you started

We use networkx.DiGraph as the representation — it's already a Badgers dependency and the lingua franca in the Python causal ecosystem.

Key Terminology¶

Term	Definition	Example in $X \rightarrow Y \rightarrow Z$
Parent	Direct cause of a node	parents(Y) = {X}
Child	Direct effect of a node	children(X) = {Y}
Ancestor	Any upstream cause (direct or indirect)	ancestors(Z) = {X, Y}
Descendant	Any downstream effect (direct or indirect)	descendants(X) = {Y, Z}
Root node	Node with no parents (in-degree 0)	X
Leaf node	Node with no children (out-degree 0)	Z
Topological order	Linear ordering where parents appear before children	[X, Y, Z]

The badgers.core.causal_graph Module¶

A new lightweight utility module provides stateless functions for working with causal DAGs:

from badgers.core.causal_graph import (
    validate_dag,        # Raises ValueError if not a DAG
    topological_order,   # Returns nodes in topological order
    get_parents,         # Immediate predecessors
    get_descendants,     # All reachable nodes
    get_ancestors,       # All nodes that can reach this node
    get_root_nodes,      # Nodes with in-degree 0
    get_leaf_nodes,      # Nodes with out-degree 0
)

All functions are stateless, take a nx.DiGraph as the first argument, and return plain Python collections.

3. The Algorithm: do()-Style Intervention ¶

The generator implements a do()-style intervention (inspired by Pearl's do-calculus): incoming edges to perturbation nodes are severed, ancestors are sampled from their natural distributions, perturbation nodes are set directly to outlier values, and only descendants are computed via forward pass.

Step 1: Validate the DAG¶

The input graph is checked for cycles using validate_dag(). If a cycle is detected, a ValueError is raised immediately.

Step 2: Topological Sort¶

Nodes are ordered so that every parent appears before its children. This ordering is used for both coefficient fitting and forward-pass propagation.

Step 3: Fit Linear Coefficients¶

For each node $v$ with parents $P(v)$, fit a linear regression:

$$X[:, v] \approx \sum_{p \in P(v)} \beta_{p \rightarrow v} \cdot X[:, p]$$

This is done via np.linalg.lstsq — ordinary least squares. The coefficients $\beta_{p \rightarrow v}$ capture the estimated linear effect of each parent on the child.

Step 4: Identify Node Sets¶

Given the set of perturbation nodes $P$:

• Descendants $D$: all nodes reachable from any $p \in P$ (these will be forward-passed)
• Ancestors $A$: all nodes that can reach any $p \in P$
• Exogenous $E = A \setminus (P \cup D)$: ancestors that are neither perturbed nor descendants (these are sampled from their natural distributions)

Step 5: Sample Exogenous Nodes¶

Exogenous nodes are sampled from their observed distributions using a configurable within-distribution sampler. The default is "normal" — sampling from $\mathcal{N}(\mu, \sigma)$ for each column independently.

Alternative within-distribution samplers ("uniform") allow sampling from different in-distribution shapes. See the Samplers section below.

Step 6: Set Perturbation Nodes Directly (do-style)¶

At each perturbation node, the incoming edges are severed — the node's value is set directly rather than computed from its parents. The values are sampled using a configurable out-of-distribution sampler (default: "zscore"):

$$\text{outlier}[k, p] = \text{sample}_{\text{ood}}(p)$$

The sampler is fully responsible for producing anomalous values — no additional perturbation is applied in the generator.

Step 7: Forward-Pass Only to Descendants¶

For each descendant node $v \in D$ in topological order, compute its value from its parents using the fitted linear coefficients, plus residual noise matching the original data's residual variance:

$$\text{outlier}[k, v] = \sum_{p \in P(v)} \beta_{p \rightarrow v} \cdot \text{outlier}[k, p] + \mathcal{N}(0, \sigma_{\text{residual}})$$

Nodes that are neither perturbed nor descendants are left at zero (they are not part of the causal pathway from the intervention).

Step 8: Return¶

The outlier rows are appended to X. yt contains "original" for each original row and "outliers" for each generated outlier row.

Why do()-style?¶

The old algorithm sampled all root nodes and forward-passed through the entire DAG. This had a subtle bug: when perturbing a non-root node (e.g., $Y$ in $X \rightarrow Y \rightarrow Z$), the forward pass would overwrite the perturbation at $Y$ with the value computed from its parent $X$.

The do()-style intervention fixes this by:

1. Severing incoming edges to perturbation nodes
2. Only forward-passing to descendants (not to the perturbation nodes themselves)
3. Leaving non-ancestor, non-descendant nodes untouched

Visual Summary¶

flowchart TD
    A[Input: X, graph, perturbation_nodes] --> B[Validate DAG]
    B --> C[Topological Sort]
    C --> D[Fit Linear Coefficients via lstsq]
    D --> E[Identify Exogenous / Perturbed / Descendant sets]
    E --> F[Sample Exogenous Nodes via Within-Distribution Sampler]
    F --> G[Set Perturbation Nodes via Out-of-Distribution Sampler]
    G --> H[Forward-Pass Only to Descendants]
    H --> I[Append Outliers to X, Return Xt, yt]

4. Graph Conventions: Nodes, Columns, and Mapping ¶

The generator needs to know which graph node corresponds to which column in X. All graph nodes must be strings, and you must provide an explicit column_mapping dictionary:

graph = nx.DiGraph()
graph.add_edges_from([("X", "Y"), ("Y", "Z")])

column_mapping = {"X": 0, "Y": 1, "Z": 2}

The mapping must satisfy:

• Every graph node has an entry
• No extra entries (nodes not in the graph)
• All values are integers in 0..n_features-1
• No duplicate column indices

Reserved for Future Use¶

• Node attributes: "type" (latent/observed), "noise_std", "function" — not used in v1
• Edge attributes: "effect" (linear/nonlinear), "weight" — not used in v1

5. API Walkthrough ¶

Quickstart¶

from badgers.generators.tabular_data.outliers.causal import CausalOutlierPropagationGenerator

# 1. Define your causal graph
graph = nx.DiGraph()
graph.add_edges_from([("X", "Y"), ("Y", "Z")])

# 2. Prepare your data (must have columns matching graph nodes)
X = np.random.default_rng(0).normal(size=(100, 3))  # 100 samples, 3 features

# 3. Create the generator
generator = CausalOutlierPropagationGenerator()

# 4. Generate outliers with explicit column mapping
Xt, yt = generator.generate(
    X, y=None,
    graph=graph,
    column_mapping={"X": 0, "Y": 1, "Z": 2},  # required: node → column index
    perturbation_nodes=["X"],   # Which node(s) to perturb (list)
    n_outliers=5,               # How many outlier rows to generate
)
# Xt = X with outlier rows appended (shape: (105, 3))
# yt = ["original"]*100 + ["outliers"]*5

Parameters¶

Parameter	Type	Default	Description
graph	nx.DiGraph	required	Causal DAG (all nodes must be strings)
column_mapping	dict[str, int]	required	Maps graph node names to column indices in X
perturbation_nodes	list[str]	required	Node(s) to intervene on (do-style)
n_outliers	int	10	Number of outlier rows to generate
within_distribution_sampler	WithinDistributionSampler or str	"normal"	Sampling strategy for exogenous (ancestor) nodes
out_of_distribution_sampler	OutOfDistributionSampler or str	"zscore"	Sampling strategy for perturbation nodes

The magnitude and sign of perturbations are controlled entirely by the out_of_distribution_sampler. For example, ZScoreSampler(scale=2.0) produces larger outliers than the default scale=1.0.

Multiple Perturbation Nodes¶

You can perturb multiple nodes simultaneously — all perturbations are applied in the same outlier row:

# Perturb both X and Y at once
Xt, yt = generator.generate(
    X, y=None,
    graph=graph,
    column_mapping={"X": 0, "Y": 1, "Z": 2},
    perturbation_nodes=["X", "Y"],
    n_outliers=5,
)

Samplers ¶

The generator uses two separate sampler parameters:

• within_distribution_sampler — for exogenous (ancestor) nodes that should be sampled from their natural distributions. Accepts a string (resolved via create_within_distribution_sampler) or a WithinDistributionSampler instance.
• out_of_distribution_sampler — for perturbation nodes that should be sampled outside the observed distribution. Accepts a string (resolved via create_out_of_distribution_sampler) or an OutOfDistributionSampler instance.

Within-Distribution Samplers¶

Sampler	Description
"normal"	Sample from $\mathcal{N}(\mu, \sigma)$ per dimension (default)
"uniform"	Sample uniformly within $[\min, \max]$

Out-of-Distribution Samplers¶

Sampler	Description	Extra Parameter
"zscore"	$\pm(3 + \text{Exponential}(\text{scale})) \times \sigma$ per dimension (default)	scale
"hypersphere"	Points on a hypersphere with radius $\geq 3\sigma$	scale
"uniform"	Sample uniformly beyond $[\min, \max]$ with expansion	expansion

from badgers.core.sampling import (
    create_within_distribution_sampler,
    create_out_of_distribution_sampler,
    UniformSampler,
    ZScoreSampler,
)

# Option 1: string shorthand
Xt, yt = generator.generate(
    X, y=None,
    graph=graph,
    column_mapping={"X": 0, "Y": 1, "Z": 2},
    perturbation_nodes=["X"],
    n_outliers=5,
    within_distribution_sampler="uniform",
    out_of_distribution_sampler="zscore",
)

# Option 2: Sampler instances with custom parameters
Xt, yt = generator.generate(
    X, y=None,
    graph=graph,
    column_mapping={"X": 0, "Y": 1, "Z": 2},
    perturbation_nodes=["X"],
    n_outliers=5,
    within_distribution_sampler=UniformSampler(),
    out_of_distribution_sampler=ZScoreSampler(scale=2.0),
)

The sampler classes are implemented in badgers.core.sampling and can be used independently of the causal generator.

Integration with Badgers Pipeline¶

The generator works seamlessly with the existing Pipeline class:

from badgers.core.pipeline import Pipeline
from badgers.generators.tabular_data.noise import GaussianNoiseGenerator

pipeline = Pipeline(generators={
    'noise': GaussianNoiseGenerator(random_generator=rng),
    'causal_outlier': CausalOutlierPropagationGenerator(random_generator=rng),
})

params = {
    'noise': {'noise_std': 0.25},
    'causal_outlier': {
        'graph': graph,
        'column_mapping': {"X": 0, "Y": 1, "Z": 2},
        'perturbation_nodes': ['X'],
    },
}

Xt_pipeline, yt_pipeline = pipeline.generate(X=X.copy(), y=None, params=params)

6. Topology Examples ¶

Different causal structures produce different propagation patterns. Let's explore the four fundamental topologies.

In [15]:

import numpy as np
import pandas as pd
import networkx as nx
import matplotlib.pyplot as plt
import seaborn as sns
import badgers
badgers.__version__

import numpy as np import pandas as pd import networkx as nx import matplotlib.pyplot as plt import seaborn as sns import badgers badgers.__version__

Out[15]:

'0.0.14'

In [ ]:

# ============================================================
# Helper functions for the topology examples
# ============================================================

def generate_linear_data(graph, n_samples=100, rng=None, coefficients=None):
    """Generate data from a linear structural equation model defined by a DAG.

    Each edge (u -> v) has a coefficient w_uv. Node values are computed as:
        v = sum_{u in parents(v)} w_uv * u + noise
    where noise ~ N(0, 1).

    Parameters
    ----------
    graph : nx.DiGraph
        The causal DAG. Node names must be strings.
    n_samples : int
        Number of samples to generate.
    rng : np.random.Generator, optional
        Random number generator.
    coefficients : dict, optional
        Edge coefficients. Defaults to 1.0 for all edges.

    Returns
    -------
    X : np.ndarray of shape (n_samples, n_nodes)
    column_mapping : dict[str, int]
    """
    if rng is None:
        rng = np.random.default_rng(42)
    if coefficients is None:
        coefficients = {edge: 1.0 for edge in graph.edges()}

    nodes = list(graph.nodes())
    node_to_idx = {node: i for i, node in enumerate(nodes)}
    n_nodes = len(nodes)

    X = np.zeros((n_samples, n_nodes))
    topo_order = list(nx.topological_sort(graph))

    for node in topo_order:
        idx = node_to_idx[node]
        parents = list(graph.predecessors(node))
        if not parents:
            # Exogenous node: pure noise
            X[:, idx] = rng.normal(size=n_samples)
        else:
            # Endogenous node: weighted sum of parents + noise
            X[:, idx] = rng.normal(size=n_samples)
            for parent in parents:
                parent_idx = node_to_idx[parent]
                coef = coefficients.get((parent, node), 1.0)
                X[:, idx] += coef * X[:, parent_idx]

    column_mapping = {node: i for i, node in enumerate(nodes)}
    return X, column_mapping

# ============================================================ # Helper functions for the topology examples # ============================================================ def generate_linear_data(graph, n_samples=100, rng=None, coefficients=None): """Generate data from a linear structural equation model defined by a DAG. Each edge (u -> v) has a coefficient w_uv. Node values are computed as: v = sum_{u in parents(v)} w_uv * u + noise where noise ~ N(0, 1). Parameters ---------- graph : nx.DiGraph The causal DAG. Node names must be strings. n_samples : int Number of samples to generate. rng : np.random.Generator, optional Random number generator. coefficients : dict, optional Edge coefficients. Defaults to 1.0 for all edges. Returns ------- X : np.ndarray of shape (n_samples, n_nodes) column_mapping : dict[str, int] """ if rng is None: rng = np.random.default_rng(42) if coefficients is None: coefficients = {edge: 1.0 for edge in graph.edges()} nodes = list(graph.nodes()) node_to_idx = {node: i for i, node in enumerate(nodes)} n_nodes = len(nodes) X = np.zeros((n_samples, n_nodes)) topo_order = list(nx.topological_sort(graph)) for node in topo_order: idx = node_to_idx[node] parents = list(graph.predecessors(node)) if not parents: # Exogenous node: pure noise X[:, idx] = rng.normal(size=n_samples) else: # Endogenous node: weighted sum of parents + noise X[:, idx] = rng.normal(size=n_samples) for parent in parents: parent_idx = node_to_idx[parent] coef = coefficients.get((parent, node), 1.0) X[:, idx] += coef * X[:, parent_idx] column_mapping = {node: i for i, node in enumerate(nodes)} return X, column_mapping

In [ ]:

def plot_data(X, y, column_names):
    """Pairplot of the data with outliers highlighted in red.

    Parameters
    ----------
    X : np.ndarray of shape (n_samples, n_features)
    y : array-like or None
        Labels: "original" or "outliers". If None, all points are "original".
    column_names : list of str
        Names for each column.
    """
    import pandas as pd
    import seaborn as sns

    df = pd.DataFrame(X, columns=column_names)
    if y is None:
        df["_label"] = "original"
    else:
        df["_label"] = y

    g = sns.pairplot(
        df, vars=column_names, hue="_label",
        hue_order=["original", "outliers"],
        palette={"original": "C0", "outliers": "C3"},
        diag_kind="hist",
        plot_kws={"alpha": 0.6, "s": 30},
        height=2.5,
    )

    # Clear lower triangle and add regression lines to upper triangle
    for i in range(len(column_names)):
        for j in range(len(column_names)):
            if i > j:
                # Lower triangle: clear and hide
                g.axes[i, j].clear()
                g.axes[i, j].set_visible(False)
            elif i < j:
                # Upper triangle: add regression line for original data
                orig = df[df["_label"] == "original"]
                if len(orig) > 1:
                    sns.regplot(
                        data=orig, x=column_names[j], y=column_names[i],
                        ax=g.axes[i, j], scatter=False, color="C0",
                        line_kws={"linewidth": 1, "alpha": 0.5},
                    )
            else:
                # Diagonal: add y-axis label "Count" to histograms
                g.axes[i, j].set_ylabel("Count")

    g.figure.suptitle("Data Pairplot (red = outliers)", y=1.02, fontsize=13)
    return g

def plot_data(X, y, column_names): """Pairplot of the data with outliers highlighted in red. Parameters ---------- X : np.ndarray of shape (n_samples, n_features) y : array-like or None Labels: "original" or "outliers". If None, all points are "original". column_names : list of str Names for each column. """ import pandas as pd import seaborn as sns df = pd.DataFrame(X, columns=column_names) if y is None: df["_label"] = "original" else: df["_label"] = y g = sns.pairplot( df, vars=column_names, hue="_label", hue_order=["original", "outliers"], palette={"original": "C0", "outliers": "C3"}, diag_kind="hist", plot_kws={"alpha": 0.6, "s": 30}, height=2.5, ) # Clear lower triangle and add regression lines to upper triangle for i in range(len(column_names)): for j in range(len(column_names)): if i > j: # Lower triangle: clear and hide g.axes[i, j].clear() g.axes[i, j].set_visible(False) elif i < j: # Upper triangle: add regression line for original data orig = df[df["_label"] == "original"] if len(orig) > 1: sns.regplot( data=orig, x=column_names[j], y=column_names[i], ax=g.axes[i, j], scatter=False, color="C0", line_kws={"linewidth": 1, "alpha": 0.5}, ) else: # Diagonal: add y-axis label "Count" to histograms g.axes[i, j].set_ylabel("Count") g.figure.suptitle("Data Pairplot (red = outliers)", y=1.02, fontsize=13) return g

In [ ]:

def draw_graph(graph, title=None):
    """Draw a DAG with node labels.

    Parameters
    ----------
    graph : nx.DiGraph
    title : str, optional
    """
    fig, ax = plt.subplots(1, figsize=(6, 2))
    pos = nx.spring_layout(graph, seed=42)
    nx.draw(graph, pos, ax=ax, node_color="lightblue", node_size=800, arrowsize=15)
    nx.draw_networkx_labels(graph, pos, ax=ax)
    if title:
        ax.set_title(title, fontsize=12)
    ax.axis("off")
    return fig

def draw_graph(graph, title=None): """Draw a DAG with node labels. Parameters ---------- graph : nx.DiGraph title : str, optional """ fig, ax = plt.subplots(1, figsize=(6, 2)) pos = nx.spring_layout(graph, seed=42) nx.draw(graph, pos, ax=ax, node_color="lightblue", node_size=800, arrowsize=15) nx.draw_networkx_labels(graph, pos, ax=ax) if title: ax.set_title(title, fontsize=12) ax.axis("off") return fig

In [ ]:

# Shared RNG for reproducibility
rng = np.random.default_rng(42)
print("Helper functions and RNG ready.")

# Shared RNG for reproducibility rng = np.random.default_rng(42) print("Helper functions and RNG ready.")

Helper functions and RNG ready.

6.1 Chain: $X \rightarrow Y \rightarrow Z$¶

Perturbing X affects Y (direct child) and Z (indirect descendant via Y).

Perturbing Y affects Z but NOT X (X is upstream, not a descendant).

In [12]:

# --- Chain: X -> Y -> Z ---
chain = nx.DiGraph()
chain.add_edges_from([("X", "Y"), ("Y", "Z")])

coefs_chain = {("X", "Y"): 2.0, ("Y", "Z"): 0.5}
X_chain, mapping_chain = generate_linear_data(chain, n_samples=100, rng=rng, coefficients=coefs_chain)

print(f"X_chain shape: {X_chain.shape}")
print(f"column_mapping: {mapping_chain}")

# --- Chain: X -> Y -> Z --- chain = nx.DiGraph() chain.add_edges_from([("X", "Y"), ("Y", "Z")]) coefs_chain = {("X", "Y"): 2.0, ("Y", "Z"): 0.5} X_chain, mapping_chain = generate_linear_data(chain, n_samples=100, rng=rng, coefficients=coefs_chain) print(f"X_chain shape: {X_chain.shape}") print(f"column_mapping: {mapping_chain}")

X_chain shape: (100, 3)
column_mapping: {'X': 0, 'Y': 1, 'Z': 2}

Step 2: Visualize graph and data¶

In [16]:

# Draw the graph
draw_graph(chain, title="Chain: X → Y → Z")
plt.show()

# Pairplot of the original data
plot_data(X_chain, y=None, column_names=["X", "Y", "Z"])

# Draw the graph draw_graph(chain, title="Chain: X → Y → Z") plt.show() # Pairplot of the original data plot_data(X_chain, y=None, column_names=["X", "Y", "Z"])

Out[16]:

<seaborn.axisgrid.PairGrid at 0x246ea2e99d0>

Step 3: Generate outliers (perturb X and Y)¶

In [18]:

from badgers.generators.tabular_data.outliers.causal import CausalOutlierPropagationGenerator

from badgers.generators.tabular_data.outliers.causal import CausalOutlierPropagationGenerator

In [ ]:

# Create the generator
generator = CausalOutlierPropagationGenerator(random_generator=rng)

# Perturb X — do(X): X set directly, Y and Z forward-passed
Xt_x, yt_x = generator.generate(X_chain.copy(), y=None, graph=chain,
                                 column_mapping=mapping_chain,
                                 perturbation_nodes=["X"],
                                 n_outliers=10)

# Perturb Y — do(Y): X is exogenous (sampled normally), Y set directly, Z forward-passed
Xt_y, yt_y = generator.generate(X_chain.copy(), y=None, graph=chain,
                                 column_mapping=mapping_chain,
                                 perturbation_nodes=["Y"],
                                 n_outliers=10)

print(f"do(X): {Xt_x.shape[0]} rows ({Xt_x.shape[0] - 100} outliers)")
print(f"do(Y): {Xt_y.shape[0]} rows ({Xt_y.shape[0] - 100} outliers)")

# Create the generator generator = CausalOutlierPropagationGenerator(random_generator=rng) # Perturb X — do(X): X set directly, Y and Z forward-passed Xt_x, yt_x = generator.generate(X_chain.copy(), y=None, graph=chain, column_mapping=mapping_chain, perturbation_nodes=["X"], n_outliers=10) # Perturb Y — do(Y): X is exogenous (sampled normally), Y set directly, Z forward-passed Xt_y, yt_y = generator.generate(X_chain.copy(), y=None, graph=chain, column_mapping=mapping_chain, perturbation_nodes=["Y"], n_outliers=10) print(f"do(X): {Xt_x.shape[0]} rows ({Xt_x.shape[0] - 100} outliers)") print(f"do(Y): {Xt_y.shape[0]} rows ({Xt_y.shape[0] - 100} outliers)")

do(X): 110 rows (10 outliers)
do(Y): 110 rows (10 outliers)

Step 4: Visualize generated data¶

In [21]:

# do(X): outliers should be visible as red markers
plot_data(Xt_x, y=yt_x, column_names=["X", "Y", "Z"])

# do(X): outliers should be visible as red markers plot_data(Xt_x, y=yt_x, column_names=["X", "Y", "Z"])

Out[21]:

<seaborn.axisgrid.PairGrid at 0x246eb946cc0>

In [22]:

# do(Y): X is exogenous (not perturbed), Y and Z shift
plot_data(Xt_y, y=yt_y, column_names=["X", "Y", "Z"])

# do(Y): X is exogenous (not perturbed), Y and Z shift plot_data(Xt_y, y=yt_y, column_names=["X", "Y", "Z"])

Out[22]:

<seaborn.axisgrid.PairGrid at 0x246eb935460>

6.2 Fork (Common Cause): $C \rightarrow X, C \rightarrow Y$¶

Perturbing C affects both X and Y (both are children).

Perturbing X affects nothing downstream (X is a leaf node).

In [28]:

# --- Fork: C -> X, C -> Y ---
fork = nx.DiGraph()
fork.add_edges_from([("C", "X"), ("C", "Y")])

coefs_fork = {("C", "X"): 1.5, ("C", "Y"): 0.8}
X_fork, mapping_fork = generate_linear_data(fork, n_samples=100, rng=rng, coefficients=coefs_fork)

# --- Fork: C -> X, C -> Y --- fork = nx.DiGraph() fork.add_edges_from([("C", "X"), ("C", "Y")]) coefs_fork = {("C", "X"): 1.5, ("C", "Y"): 0.8} X_fork, mapping_fork = generate_linear_data(fork, n_samples=100, rng=rng, coefficients=coefs_fork)

Step 2: Visualize graph and data¶

In [24]:

# Draw the graph
draw_graph(fork, title="Fork: C → X, C → Y")
plt.show()

# Pairplot of the original data
plot_data(X_fork, y=None, column_names=["C", "X", "Y"])

# Draw the graph draw_graph(fork, title="Fork: C → X, C → Y") plt.show() # Pairplot of the original data plot_data(X_fork, y=None, column_names=["C", "X", "Y"])

Out[24]:

<seaborn.axisgrid.PairGrid at 0x246edf63110>

Step 3 Generate outliers¶

In [29]:

Xt_fork, yt_fork = generator.generate(X_fork.copy(), y=None, graph=fork,
                                 column_mapping=mapping_fork,
                                 perturbation_nodes=["C"],
                                 n_outliers=10)

Xt_fork, yt_fork = generator.generate(X_fork.copy(), y=None, graph=fork, column_mapping=mapping_fork, perturbation_nodes=["C"], n_outliers=10)

Step 4: Visualize generated data¶

In [31]:

plot_data(Xt_fork, y=yt_fork, column_names=["C", "X", "Y"])

plot_data(Xt_fork, y=yt_fork, column_names=["C", "X", "Y"])

Out[31]:

<seaborn.axisgrid.PairGrid at 0x246eaba2e40>

6.3 Collider (Common Effect): $X \rightarrow Z \leftarrow Y$¶

Perturbing X affects Z (its child) but NOT Y (Y is not a descendant of X).

This is a key test: the generator respects the direction of causal edges.

Step 1: Prepare data and graph¶

In [32]:

# --- Collider: X -> Z <- Y ---
collider = nx.DiGraph()
collider.add_edges_from([("X", "Z"), ("Y", "Z")])

coefs_collider = {("X", "Z"): 1.0, ("Y", "Z"): 1.0}
X_collider, mapping_collider = generate_linear_data(collider, n_samples=100, rng=rng, coefficients=coefs_collider)

print(f"X_collider shape: {X_collider.shape}")
print(f"column_mapping: {mapping_collider}")

# --- Collider: X -> Z <- Y --- collider = nx.DiGraph() collider.add_edges_from([("X", "Z"), ("Y", "Z")]) coefs_collider = {("X", "Z"): 1.0, ("Y", "Z"): 1.0} X_collider, mapping_collider = generate_linear_data(collider, n_samples=100, rng=rng, coefficients=coefs_collider) print(f"X_collider shape: {X_collider.shape}") print(f"column_mapping: {mapping_collider}")

X_collider shape: (100, 3)
column_mapping: {'X': 0, 'Z': 1, 'Y': 2}

Step 2: Visualize graph and data¶

In [33]:

# Draw the graph
draw_graph(collider, title="Collider: X → Z ← Y")
plt.show()

# Pairplot of the original data
plot_data(X_collider, y=None, column_names=["X", "Y", "Z"])

# Draw the graph draw_graph(collider, title="Collider: X → Z ← Y") plt.show() # Pairplot of the original data plot_data(X_collider, y=None, column_names=["X", "Y", "Z"])

Out[33]:

<seaborn.axisgrid.PairGrid at 0x246f03e4d70>

Step 3: Generate outliers¶

In [ ]:

# Perturb X — affects Z but NOT Y
generator = CausalOutlierPropagationGenerator(
    random_generator=rng
)
Xt_collider, yt_collider = generator.generate(
    X_collider, y=None,
    graph=collider,
    column_mapping=mapping_collider,
    perturbation_nodes=["X"],
    n_outliers=10,
)
print(f"Outliers generated: {(yt_collider == 'outliers').sum()} / {len(yt_collider)}")

# Perturb X — affects Z but NOT Y generator = CausalOutlierPropagationGenerator( random_generator=rng ) Xt_collider, yt_collider = generator.generate( X_collider, y=None, graph=collider, column_mapping=mapping_collider, perturbation_nodes=["X"], n_outliers=10, ) print(f"Outliers generated: {(yt_collider == 'outliers').sum()} / {len(yt_collider)}")

Step 4: Visualize results¶

In [ ]:

plot_data(Xt_collider, yt_collider, column_names=["X", "Y", "Z"])

plot_data(Xt_collider, yt_collider, column_names=["X", "Y", "Z"])

6.4 Diamond (Mediation + Confounding): $A \rightarrow B, A \rightarrow C, B \rightarrow D, C \rightarrow D$¶

Perturbing A affects B, C (direct children) and D (indirect descendant via two paths).

This tests multi-path propagation: D receives contributions from both $A \rightarrow B \rightarrow D$ and $A \rightarrow C \rightarrow D$.

Step 1: Prepare data and graph¶

In [ ]:

# --- Diamond: A -> B, A -> C, B -> D, C -> D ---
diamond = nx.DiGraph()
diamond.add_edges_from([("A", "B"), ("A", "C"), ("B", "D"), ("C", "D")])

coefs_diamond = {("A", "B"): 1.0, ("A", "C"): 1.0, ("B", "D"): 0.5, ("C", "D"): 0.5}
X_diamond, mapping_diamond = generate_linear_data(diamond, n_samples=100, rng=rng, coefficients=coefs_diamond)

print(f"X_diamond shape: {X_diamond.shape}")
print(f"column_mapping: {mapping_diamond}")

# --- Diamond: A -> B, A -> C, B -> D, C -> D --- diamond = nx.DiGraph() diamond.add_edges_from([("A", "B"), ("A", "C"), ("B", "D"), ("C", "D")]) coefs_diamond = {("A", "B"): 1.0, ("A", "C"): 1.0, ("B", "D"): 0.5, ("C", "D"): 0.5} X_diamond, mapping_diamond = generate_linear_data(diamond, n_samples=100, rng=rng, coefficients=coefs_diamond) print(f"X_diamond shape: {X_diamond.shape}") print(f"column_mapping: {mapping_diamond}")

Step 2: Visualize graph and data¶

In [ ]:

# Draw the graph
draw_graph(diamond, title="Diamond: A → B, A → C, B → D, C → D")
plt.show()

# Pairplot of the original data
plot_data(X_diamond, y=None, column_names=["A", "B", "C", "D"])

# Draw the graph draw_graph(diamond, title="Diamond: A → B, A → C, B → D, C → D") plt.show() # Pairplot of the original data plot_data(X_diamond, y=None, column_names=["A", "B", "C", "D"])

Step 3: Generate outliers¶

In [ ]:

# Perturb A — affects B, C, and D via two paths
generator = CausalOutlierPropagationGenerator(
    random_generator=rng
)
Xt_diamond, yt_diamond = generator.generate(
    X_diamond, y=None,
    graph=diamond,
    column_mapping=mapping_diamond,
    perturbation_nodes=["A"],
    n_outliers=10,
)
print(f"Outliers generated: {(yt_diamond == 'outliers').sum()} / {len(yt_diamond)}")

# Perturb A — affects B, C, and D via two paths generator = CausalOutlierPropagationGenerator( random_generator=rng ) Xt_diamond, yt_diamond = generator.generate( X_diamond, y=None, graph=diamond, column_mapping=mapping_diamond, perturbation_nodes=["A"], n_outliers=10, ) print(f"Outliers generated: {(yt_diamond == 'outliers').sum()} / {len(yt_diamond)}")

Step 4: Visualize results¶

In [ ]:

plot_data(Xt_diamond, yt_diamond, column_names=["A", "B", "C", "D"])

plot_data(Xt_diamond, yt_diamond, column_names=["A", "B", "C", "D"])

6.5 Sampling Strategy Comparison¶

The within_distribution_sampler and out_of_distribution_sampler parameters control how exogenous and perturbation node values are generated. Let's compare out-of-distribution strategies on the chain topology $X \rightarrow Y \rightarrow Z$:

In [ ]:

from badgers.core.sampling import (
    create_out_of_distribution_sampler,
    UniformOutOfDistributionSampler,
)

# Use the chain graph from section 6.1
sampler_configs = [
    ("zscore", {}),
    ("hypersphere", {}),
    ("uniform", {"expansion": 0.5}),
]
n_outliers = 20

fig, axes = plt.subplots(1, 3, figsize=(15, 5))
axes = axes.flatten()

for ax, (sampler_name, kwargs) in zip(axes, sampler_configs):
    if kwargs:
        sampler = create_out_of_distribution_sampler(sampler_name, **kwargs)
    else:
        sampler = sampler_name

    Xt, _ = generator.generate(
        X_chain.copy(), y=None,
        graph=chain,
        column_mapping=mapping_chain,
        perturbation_nodes=["X"],
        n_outliers=n_outliers,
        out_of_distribution_sampler=sampler,
    )
    outliers = Xt[-n_outliers:]
    original = Xt[:-n_outliers]

    # Scatter: X vs Y, color by original/outlier
    ax.scatter(original[:, 0], original[:, 1], alpha=0.4, s=20, label="original", c="C0")
    ax.scatter(outliers[:, 0], outliers[:, 1], alpha=0.8, s=40, label="outliers", c="C3", edgecolors="black")
    ax.set_xlabel("X")
    ax.set_ylabel("Y")
    label = sampler_name
    if kwargs:
        label = f"{sampler_name} ({', '.join(f'{k}={v}' for k, v in kwargs.items())})"
    ax.set_title(f'out_of_distribution_sampler="{label}"')
    ax.legend()
    ax.axhline(y=0, color='gray', linestyle='--', alpha=0.3)
    ax.axvline(x=0, color='gray', linestyle='--', alpha=0.3)

plt.suptitle("Out-of-Distribution Sampler Comparison: X → Y (Chain Topology)", fontsize=14, fontweight='bold')
plt.tight_layout()
plt.show()

from badgers.core.sampling import ( create_out_of_distribution_sampler, UniformOutOfDistributionSampler, ) # Use the chain graph from section 6.1 sampler_configs = [ ("zscore", {}), ("hypersphere", {}), ("uniform", {"expansion": 0.5}), ] n_outliers = 20 fig, axes = plt.subplots(1, 3, figsize=(15, 5)) axes = axes.flatten() for ax, (sampler_name, kwargs) in zip(axes, sampler_configs): if kwargs: sampler = create_out_of_distribution_sampler(sampler_name, **kwargs) else: sampler = sampler_name Xt, _ = generator.generate( X_chain.copy(), y=None, graph=chain, column_mapping=mapping_chain, perturbation_nodes=["X"], n_outliers=n_outliers, out_of_distribution_sampler=sampler, ) outliers = Xt[-n_outliers:] original = Xt[:-n_outliers] # Scatter: X vs Y, color by original/outlier ax.scatter(original[:, 0], original[:, 1], alpha=0.4, s=20, label="original", c="C0") ax.scatter(outliers[:, 0], outliers[:, 1], alpha=0.8, s=40, label="outliers", c="C3", edgecolors="black") ax.set_xlabel("X") ax.set_ylabel("Y") label = sampler_name if kwargs: label = f"{sampler_name} ({', '.join(f'{k}={v}' for k, v in kwargs.items())})" ax.set_title(f'out_of_distribution_sampler="{label}"') ax.legend() ax.axhline(y=0, color='gray', linestyle='--', alpha=0.3) ax.axvline(x=0, color='gray', linestyle='--', alpha=0.3) plt.suptitle("Out-of-Distribution Sampler Comparison: X → Y (Chain Topology)", fontsize=14, fontweight='bold') plt.tight_layout() plt.show()

Key observations:

• "zscore" (default): Outliers are pushed far from the mean ($\geq 3\sigma$), creating extreme but causally-consistent anomalies.
• "hypersphere": Outliers form a spherical shell in standardized space, useful for testing detectors against uniformly-directional anomalies.
• "uniform": Outliers spread uniformly beyond the observed range, producing diverse patterns across the full expanded space.

The within-distribution sampler (default "normal") controls how exogenous ancestor nodes are sampled — these stay within the observed distribution since they are not directly perturbed.

7. Error Handling and Edge Cases ¶

The generator validates inputs rigorously and raises clear errors:

In [ ]:

import pytest

generator = CausalOutlierPropagationGenerator(random_generator=rng)

# 1. Missing graph
print("--- Missing graph ---")
try:
    generator.generate(np.random.randn(100, 3), y=None,
                       column_mapping={"X": 0, "Y": 1, "Z": 2},
                       perturbation_nodes=["X"])
except ValueError as e:
    print(f"ValueError: {e}")

# 2. Invalid perturbation node
print("\n--- Invalid perturbation node ---")
graph_xy = nx.DiGraph([("X", "Y")])
try:
    generator.generate(np.random.randn(100, 2), y=None, graph=graph_xy,
                       column_mapping={"X": 0, "Y": 1},
                       perturbation_nodes=["Z"])
except ValueError as e:
    print(f"ValueError: {e}")

# 3. Cyclic graph
print("\n--- Cyclic graph ---")
cyclic = nx.DiGraph([("X", "Y"), ("Y", "X")])
try:
    generator.generate(np.random.randn(100, 2), y=None, graph=cyclic,
                       column_mapping={"X": 0, "Y": 1},
                       perturbation_nodes=["X"])
except ValueError as e:
    print(f"ValueError: {e}")

# 4. Column count mismatch
print("\n--- Column count mismatch ---")
try:
    generator.generate(np.random.randn(100, 5), y=None, graph=graph_xy,
                       column_mapping={"X": 0, "Y": 1},
                       perturbation_nodes=["X"])
except ValueError as e:
    print(f"ValueError: {e}")

# 5. perturbation_nodes not a list
print("\n--- perturbation_nodes not a list ---")
try:
    generator.generate(np.random.randn(100, 2), y=None, graph=graph_xy,
                       column_mapping={"X": 0, "Y": 1},
                       perturbation_nodes="X")
except ValueError as e:
    print(f"ValueError: {e}")

# 6. Empty perturbation_nodes
print("\n--- Empty perturbation_nodes ---")
try:
    generator.generate(np.random.randn(100, 2), y=None, graph=graph_xy,
                       column_mapping={"X": 0, "Y": 1},
                       perturbation_nodes=[])
except ValueError as e:
    print(f"ValueError: {e}")

# 7. Missing column_mapping
print("\n--- Missing column_mapping ---")
try:
    generator.generate(np.random.randn(100, 2), y=None, graph=graph_xy,
                       perturbation_nodes=["X"])
except ValueError as e:
    print(f"ValueError: {e}")

import pytest generator = CausalOutlierPropagationGenerator(random_generator=rng) # 1. Missing graph print("--- Missing graph ---") try: generator.generate(np.random.randn(100, 3), y=None, column_mapping={"X": 0, "Y": 1, "Z": 2}, perturbation_nodes=["X"]) except ValueError as e: print(f"ValueError: {e}") # 2. Invalid perturbation node print("\n--- Invalid perturbation node ---") graph_xy = nx.DiGraph([("X", "Y")]) try: generator.generate(np.random.randn(100, 2), y=None, graph=graph_xy, column_mapping={"X": 0, "Y": 1}, perturbation_nodes=["Z"]) except ValueError as e: print(f"ValueError: {e}") # 3. Cyclic graph print("\n--- Cyclic graph ---") cyclic = nx.DiGraph([("X", "Y"), ("Y", "X")]) try: generator.generate(np.random.randn(100, 2), y=None, graph=cyclic, column_mapping={"X": 0, "Y": 1}, perturbation_nodes=["X"]) except ValueError as e: print(f"ValueError: {e}") # 4. Column count mismatch print("\n--- Column count mismatch ---") try: generator.generate(np.random.randn(100, 5), y=None, graph=graph_xy, column_mapping={"X": 0, "Y": 1}, perturbation_nodes=["X"]) except ValueError as e: print(f"ValueError: {e}") # 5. perturbation_nodes not a list print("\n--- perturbation_nodes not a list ---") try: generator.generate(np.random.randn(100, 2), y=None, graph=graph_xy, column_mapping={"X": 0, "Y": 1}, perturbation_nodes="X") except ValueError as e: print(f"ValueError: {e}") # 6. Empty perturbation_nodes print("\n--- Empty perturbation_nodes ---") try: generator.generate(np.random.randn(100, 2), y=None, graph=graph_xy, column_mapping={"X": 0, "Y": 1}, perturbation_nodes=[]) except ValueError as e: print(f"ValueError: {e}") # 7. Missing column_mapping print("\n--- Missing column_mapping ---") try: generator.generate(np.random.randn(100, 2), y=None, graph=graph_xy, perturbation_nodes=["X"]) except ValueError as e: print(f"ValueError: {e}")

Edge Case: Minimal Perturbation¶

When using a sampler with small magnitude (e.g., ZScoreSampler(scale=0.01)), the perturbation nodes stay close to the distribution mean, and descendants are computed via forward pass. The output still follows the causal structure:

from badgers.core.sampling import ZScoreSampler

Xt, _ = generator.generate(
    X, y=None, graph=graph,
    column_mapping={"X": 0, "Y": 1, "Z": 2},
    perturbation_nodes=["X"],
    out_of_distribution_sampler=ZScoreSampler(scale=0.01),
)
# Outlier rows follow the causal model with minimal perturbation

Edge Case: Constant Column¶

If a column has zero variance (all values identical), the generator uses a fallback standard deviation of 1.0 to avoid division by zero.

Edge Case: y is None¶

When y=None, the generator creates a label array with "original" for input rows and "outliers" for generated rows. If y is provided, "outliers" labels are appended.

Integration Tests (tests/generators/tabular_data/outliers/test_causal.py) — 32 tests¶

Test	Topology	What It Verifies
test_generate__chain_x_to_y	Chain	do(X) → Y and Z shift proportionally
test_generate__chain_y_do_intervention	Chain	do(Y) → X exogenous, Y perturbed, Z propagates
test_generate__fork	Fork	do(C) → both X and Y shift
test_generate__collider	Collider	do(X) → Z shifts, Y unchanged
test_generate__diamond	Diamond	do(A) → all descendants shift
test_generate__multiple_perturbation_nodes	Chain	do(X,Y) → both perturbed simultaneously
test_generate__non_root_perturbation_do_style	C→Y→T	do(Y) → C exogenous (not perturbed)
test_generate__out_of_distribution_sampler_string	—	out_of_distribution_sampler="hypersphere" resolved via factory
test_generate__out_of_distribution_sampler_instance	—	OutOfDistributionSampler instance used directly
test_generate__within_distribution_sampler_string	—	within_distribution_sampler="uniform" resolved via factory
test_generate__within_distribution_sampler_instance	—	WithinDistributionSampler instance used directly
test_generate__both_samplers_specified	—	Both sampler params specified simultaneously
test_generate__default_samplers	—	Default samplers (NormalSampler + ZScoreSampler)
test_generate__uniform_out_of_distribution_sampler	—	UniformOutOfDistributionSampler with expansion
test_generate__uniform_out_of_distribution_no_expansion_raises	—	expansion=0 raises ValueError
test_generate__uniform_out_of_distribution_negative_expansion_raises	—	negative expansion raises ValueError
test_generate__within_distribution_sampler_rejects_out_of_distribution	—	Wrong sampler type raises ValueError
test_generate__out_of_distribution_sampler_rejects_within_distribution	—	Wrong sampler type raises ValueError
test_generate__no_graph_raises	—	Missing graph → ValueError
test_generate__invalid_perturbation_raises	—	Bad node → ValueError
test_generate__perturbation_nodes_not_list_raises	—	String instead of list → ValueError
test_generate__perturbation_nodes_empty_raises	—	Empty list → ValueError
test_generate__perturbation_nodes_none_raises	—	None → ValueError
test_generate__invalid_out_of_distribution_sampler_type_raises	—	int instead of OutOfDistributionSampler → ValueError
test_generate__invalid_within_distribution_sampler_type_raises	—	int instead of WithinDistributionSampler → ValueError
test_generate__cycle_raises	—	Cyclic graph → ValueError
test_generate__y_with_labels	—	y with labels → labels appended
test_generate__y_none_creates_labels	—	y=None → labels created
test_generate__original_data_unchanged	—	Original rows unmodified
test_generate__magnitude_zero	—	Zero magnitude → causal structure preserved
test_generate__n_outliers_zero_raises	—	n_outliers=0 → ValueError
test_generate__int_nodes	Chain (int)	Integer node labels work

Running the Tests¶

# All tests
pytest -v tests

# Just the causal tests
pytest -v tests/core/test_causal_graph.py tests/generators/tabular_data/outliers/test_causal.py tests/core/test_sampling.py

8. Testing Strategy ¶

The implementation is backed by a comprehensive test suite:

Unit Tests (tests/core/test_causal_graph.py) — 19 tests¶

Tests all 8 utility functions across chain, fork, collider, diamond, cyclic, and integer-node graphs.

Sampler Tests (tests/core/test_sampling.py)¶

Tests all 5 Sampler subclasses (NormalSampler, UniformSampler, ZScoreSampler, HypersphereSampler, UniformOutOfDistributionSampler), the create_within_distribution_sampler() and create_out_of_distribution_sampler() factories, and the registries.

Integration Tests (tests/generators/tabular_data/outliers/test_causal.py) — 32 tests¶

Test	Topology	What It Verifies
test_generate__chain_x_to_y	Chain	do(X) → Y and Z shift proportionally
test_generate__chain_y_do_intervention	Chain	do(Y) → X exogenous, Y perturbed, Z propagates
test_generate__fork	Fork	do(C) → both X and Y shift
test_generate__collider	Collider	do(X) → Z shifts, Y unchanged
test_generate__diamond	Diamond	do(A) → all descendants shift
test_generate__multiple_perturbation_nodes	Chain	do(X,Y) → both perturbed simultaneously
test_generate__non_root_perturbation_do_style	C→Y→T	do(Y) → C exogenous (not perturbed)
test_generate__out_of_distribution_sampler_string	—	out_of_distribution_sampler="hypersphere" resolved via factory
test_generate__out_of_distribution_sampler_instance	—	OutOfDistributionSampler instance used directly
test_generate__within_distribution_sampler_string	—	within_distribution_sampler="uniform" resolved via factory
test_generate__within_distribution_sampler_instance	—	WithinDistributionSampler instance used directly
test_generate__both_samplers_specified	—	Both sampler params specified simultaneously
test_generate__default_samplers	—	Default samplers (NormalSampler + ZScoreSampler)
test_generate__uniform_out_of_distribution_sampler	—	UniformOutOfDistributionSampler with expansion
test_generate__uniform_out_of_distribution_no_expansion_raises	—	expansion=0 raises ValueError
test_generate__uniform_out_of_distribution_negative_expansion_raises	—	negative expansion raises ValueError
test_generate__within_distribution_sampler_rejects_out_of_distribution	—	Wrong sampler type raises ValueError
test_generate__out_of_distribution_sampler_rejects_within_distribution	—	Wrong sampler type raises ValueError
test_generate__no_graph_raises	—	Missing graph → ValueError
test_generate__invalid_perturbation_raises	—	Bad node → ValueError
test_generate__perturbation_nodes_not_list_raises	—	String instead of list → ValueError
test_generate__perturbation_nodes_empty_raises	—	Empty list → ValueError
test_generate__perturbation_nodes_none_raises	—	None → ValueError
test_generate__invalid_out_of_distribution_sampler_type_raises	—	int instead of OutOfDistributionSampler → ValueError
test_generate__invalid_within_distribution_sampler_type_raises	—	int instead of WithinDistributionSampler → ValueError
test_generate__cycle_raises	—	Cyclic graph → ValueError
test_generate__y_with_labels	—	y with labels → labels appended
test_generate__y_none_creates_labels	—	y=None → labels created
test_generate__original_data_unchanged	—	Original rows unmodified
test_generate__magnitude_zero	—	Zero magnitude → causal structure preserved
test_generate__n_outliers_zero_raises	—	n_outliers=0 → ValueError
test_generate__int_nodes	Chain (int)	Integer node labels work

Running the Tests¶

# All tests
pytest -v tests

# Just the causal tests
pytest -v tests/core/test_causal_graph.py tests/generators/tabular_data/outliers/test_causal.py tests/core/test_sampling.py

Summary¶

The Causal Outlier Propagation Generator is the first step toward a family of causally-informed data degradation generators in Badgers. Key takeaways:

1. Causal graphs as scaffolding — DAGs define which variables affect which others
2. Forward-pass propagation — Root nodes are sampled from their distributions, then values flow through the DAG via fitted linear coefficients
3. Directional awareness — Only downstream nodes are affected; upstream nodes are untouched
4. Pluggable sampling strategies — "normal", "uniform", "zscore", and "hypersphere" strategies control how root node values are generated
5. Clean API — Follows existing GeneratorMixin contract, works with Pipeline
6. Well-tested — 14 tests covering 4 topologies, error cases, and edge cases

This enables testing anomaly detectors against structurally consistent outliers — anomalies that could actually occur in a system governed by causal relationships.