`statspai.dag`¶

dag ¶

DAG (Directed Acyclic Graph) module for causal reasoning.

Declare causal graphs, compute adjustment sets, check for collider bias, enumerate paths, detect bad controls, and visualize causal structures — the Python equivalent of R's dagitty and ggdag.

import statspai as sp g = sp.dag('X -> Y; Z -> X; Z -> Y') g.adjustment_sets('X', 'Y') [{'Z'}] g.backdoor_paths('X', 'Y') [['X', 'Z', 'Y']] g.bad_controls('X', 'Y') {} print(g.summary('X', 'Y').splitlines()[0]) DAG Summary: effect of X on Y g.do('X') # interventional graph DAG(3 nodes, 2 edges) sp.dag_example('discrimination') # classic textbook DAG DAG(4 nodes, 5 edges)

BayesNet ¶

A fitted or hand-specified discrete causal Bayesian network.

Build one with :func:statspai.bayes_net.

Attributes:

Name	Type	Description
`dag`	`DAG`	The graph.
`states`	`dict`	`{node: list of its states}`, in table order.
`nodes`	`list of str`	Nodes in a topological order.
`unseen`	`dict`	`{node: number of parent configurations with no data}` for the nodes that have any. The table rows for those configurations are uniform placeholders, and a query whose answer depends on them warns.

Examples:

>>> import statspai as sp
>>> net = sp.bayes_net(
...     "C -> X; C -> Y; X -> Y",
...     cpts={
...         "C": {"bear": 0.5, "bull": 0.5},
...         "X": {"bear": {"debt": 0.8, "equity": 0.2},
...               "bull": {"debt": 0.2, "equity": 0.8}},
...         "Y": {("bear", "debt"): {"fail": 0.3, "success": 0.7},
...               ("bull", "debt"): {"fail": 0.9, "success": 0.1},
...               ("bear", "equity"): {"fail": 0.7, "success": 0.3},
...               ("bull", "equity"): {"fail": 0.6, "success": 0.4}},
...     },
... )
>>> round(net.prob({"Y": "success"}, evidence={"X": "debt"}), 4)
0.58
>>> round(net.prob({"Y": "success"}, do={"X": "debt"}), 4)
0.4

parents ¶

parents(node: str) -> Tuple[str, ...]

Parents of node, in the order of its table's leading axes.

cpt ¶

cpt(node: str) -> DataFrame

The conditional probability table of node.

Returns:

Type	Description
`DataFrame`	One column per state of node; one row per configuration of its parents (a single row labelled `''` for a root).

Examples:

>>> import statspai as sp
>>> net = sp.bayes_net("A -> B", cpts={
...     "A": {"no": 0.7, "yes": 0.3},
...     "B": {"no": {"lo": 0.9, "hi": 0.1}, "yes": {"lo": 0.4, "hi": 0.6}}})
>>> float(net.cpt("B").loc["yes", "hi"])
0.6

summary ¶

summary() -> str

Nodes, their states and parents, and the size of the tables.

Examples:

>>> import statspai as sp
>>> net = sp.bayes_net("A -> B", cpts={
...     "A": {"no": 0.7, "yes": 0.3},
...     "B": {"no": {"lo": 0.9, "hi": 0.1}, "yes": {"lo": 0.4, "hi": 0.6}}})
>>> print(net.summary().splitlines()[0])
Discrete causal Bayesian network: 2 nodes, 3 free parameters

do ¶

do(interventions: Optional[Mapping[str, Any]] = None, **kw: Any) -> 'BayesNet'

The network after an ideal intervention.

Each intervened node loses its parents and takes its value with probability one; every other kernel is unchanged.

Parameters:

Name	Type	Description	Default
`interventions`	`mapping`	`{node: state}`. Keyword arguments are accepted for nodes whose names are identifiers: `net.do(X="debt")`.	`None`

Returns:

Type	Description
`BayesNet`

Examples:

>>> import statspai as sp
>>> net = sp.bayes_net("Z -> X; Z -> Y; X -> Y", cpts={
...     "Z": {0: 0.5, 1: 0.5},
...     "X": {0: {0: 0.9, 1: 0.1}, 1: {0: 0.1, 1: 0.9}},
...     "Y": {(0, 0): {0: 0.9, 1: 0.1}, (0, 1): {0: 0.5, 1: 0.5},
...           (1, 0): {0: 0.5, 1: 0.5}, (1, 1): {0: 0.1, 1: 0.9}}})
>>> net.do(X=1).parents("X")
()

query ¶

query(variables: Union[str, Sequence[str]], evidence: Optional[Mapping[str, Any]] = None, do: Optional[Mapping[str, Any]] = None) -> DataFrame

The joint distribution of variables given evidence, under an intervention if one is named.

Parameters:

Name	Type	Description	Default
`variables`	`str or sequence of str`		required
`evidence`	`mapping`	`{node: observed state}`.	`None`
`do`	`mapping`	`{node: state it is set to}`. Evidence is then evidence about the intervened system: `P(variables \| do(...), evidence)`. For evidence about the world as it was, with the intervention hypothetical, use :meth:`counterfactual`.	`None`

Returns:

Type	Description
`DataFrame`	One row per combination of states, with a `prob` column.

Examples:

>>> import statspai as sp
>>> net = sp.bayes_net("Z -> X; Z -> Y; X -> Y", cpts={
...     "Z": {0: 0.5, 1: 0.5},
...     "X": {0: {0: 0.9, 1: 0.1}, 1: {0: 0.1, 1: 0.9}},
...     "Y": {(0, 0): {0: 0.9, 1: 0.1}, (0, 1): {0: 0.5, 1: 0.5},
...           (1, 0): {0: 0.5, 1: 0.5}, (1, 1): {0: 0.1, 1: 0.9}}})
>>> seen = net.query("Y", evidence={"X": 1})
>>> done = net.query("Y", do={"X": 1})
>>> round(seen.loc[seen.Y == 1, "prob"].item(), 2)   # association
0.86
>>> round(done.loc[done.Y == 1, "prob"].item(), 2)   # effect
0.7

prob ¶

prob(event: Mapping[str, Any], evidence: Optional[Mapping[str, Any]] = None, do: Optional[Mapping[str, Any]] = None) -> float

Probability of one joint event, {node: state, ...}.

Examples:

>>> import statspai as sp
>>> net = sp.bayes_net("A -> B", cpts={
...     "A": {"no": 0.7, "yes": 0.3},
...     "B": {"no": {"lo": 0.9, "hi": 0.1}, "yes": {"lo": 0.4, "hi": 0.6}}})
>>> round(net.prob({"A": "yes"}, evidence={"B": "hi"}), 4)
0.72

expectation ¶

expectation(variable: str, evidence: Optional[Mapping[str, Any]] = None, do: Optional[Mapping[str, Any]] = None, values: Optional[Mapping[Any, float]] = None) -> float

Expected value of a node.

Parameters:

Name	Type	Description	Default
`variable`	`str`		required
`evidence`	`mapping`	As in :meth:`query`.	`None`
`do`	`mapping`	As in :meth:`query`.	`None`
`values`	`mapping`	`{state: number}`, a utility or payoff for each state. Without it the states themselves must be numbers.	`None`

Examples:

>>> import statspai as sp
>>> net = sp.bayes_net("X -> Y", cpts={
...     "X": {"debt": 0.5, "equity": 0.5},
...     "Y": {"debt": {"fail": 0.6, "success": 0.4},
...           "equity": {"fail": 0.66, "success": 0.34}}})
>>> payoff = {"fail": -1000, "success": 99000}
>>> round(net.expectation("Y", do={"X": "debt"}, values=payoff))
39000

counterfactual ¶

counterfactual(variables: Union[str, Sequence[str]], evidence: Mapping[str, Any], do: Mapping[str, Any]) -> DataFrame

P(variables_{do} | evidence): what the variables would have been under the intervention, for units on which the evidence was observed without it.

Abduction, action and prediction are done in one pass on a twin network: the descendants of the intervened nodes are duplicated, the copies get the intervention, and both worlds share every other node and so the same roots.

This is defined only when every node that is not a root is a deterministic function of its parents, so that the network is a structural causal model with its exogenous variables as roots. Conditional probability tables alone do not fix a counterfactual: two structural models can imply the same tables and disagree about it.

Parameters:

Name	Type	Description	Default
`variables`	`str or sequence of str`		required
`evidence`	`mapping`	What was observed in the factual world.	required
`do`	`mapping`	The hypothetical intervention.	required

Returns:

Type	Description
`DataFrame`	As :meth:`query`.

Examples:

Monty Hall as a structural model. For a player who stayed and lost, switching would certainly have won:

>>> import statspai as sp
>>> doors = [1, 2, 3]
>>> def host(car, first, coin):
...     free = [d for d in doors if d not in (car, first)]
...     return free[0] if len(free) == 1 or coin == "tails" else free[1]
>>> def second(first, host, strategy):
...     if strategy == "stay":
...         return first
...     return next(d for d in doors if d not in (first, host))
>>> net = sp.bayes_net(
...     "car -> host; first -> host; coin -> host; first -> second; "
...     "host -> second; strategy -> second; second -> win; car -> win",
...     cpts={"car": {d: 1 / 3 for d in doors},
...           "first": {d: 1 / 3 for d in doors},
...           "coin": {"tails": 0.5, "heads": 0.5},
...           "strategy": {"stay": 0.5, "switch": 0.5},
...           "host": host, "second": second,
...           "win": lambda second, car: second == car})
>>> cf = net.counterfactual("win", evidence={"strategy": "stay", "win": False},
...                         do={"strategy": "switch"})
>>> float(cf.loc[cf.win == True, "prob"].item())
1.0

simulate ¶

simulate(n: int, seed: Optional[int] = None, do: Optional[Mapping[str, Any]] = None) -> DataFrame

Draw n rows from the network, or from the network under an intervention.

Examples:

>>> import statspai as sp
>>> net = sp.bayes_net("A -> B", cpts={
...     "A": {"no": 0.7, "yes": 0.3},
...     "B": {"no": {"lo": 0.9, "hi": 0.1}, "yes": {"lo": 0.4, "hi": 0.6}}})
>>> net.simulate(4, seed=0, do={"A": "yes"})["A"].tolist()
['yes', 'yes', 'yes', 'yes']

SCM `dataclass` ¶

Structural Causal Model.

Each node has: - parents: iterable of parent node names - equation: callable(parents_dict, noise) -> value - noise: callable() -> float (a draw from the exogenous noise distribution; defaults to standard normal)

Examples:

>>> import statspai as sp
>>> scm = sp.SCM()
>>> scm.add(
...     "X", [], lambda pa, u: u, lambda rng: rng.normal()
... )
SCM(...)
>>> scm.add(
...     "Y", ["X"], lambda pa, u: 2 * pa["X"] + u
... )
SCM(...)
>>> sim = scm.simulate(n=100, seed=0)
>>> bool(sim["Y"].shape == (100,))
True

counterfactual ¶

counterfactual(evidence: Mapping[str, float], intervention: Mapping[str, float], n_samples: int = 2000, seed: int | None = None, tol: float = 0.01) -> SampleMap

Compute E[Y(intervention) | evidence] via abduction-action.

Parameters:

Name	Type	Description	Default
`evidence`	`dict`	Observed values of some subset of nodes (factual world).	required
`intervention`	`dict`	Values to set for do-intervened variables.	required
`n_samples`	`int`	Number of accepted noise draws for rejection-sampling.	`2000`
`tol`	`float`	Tolerance for matching continuous evidence.	`1e-2`

Returns:

Type	Description
`dict[str, ndarray]`	Counterfactual samples for every node.

CounterfactualIdentification `dataclass` ¶

Bases: ResultProtocolMixin

Outcome of :func:identify_counterfactual.

Attributes:

Name	Type	Description
`identifiable`	`bool`	Whether the query is a function of interventional distributions.
`estimand`	`str`	The formula, in terms `P(v \| do(x))`; a bare variable name stands for the value a summation ranges over.
`query`	`str`	The query as understood.
`experiments`	`list of str`	The distinct interventional distributions the formula uses.
`observational`	`dict`	For each of those, its formula in the observational distribution, or `None` when it needs an experiment.
`from_observational_data`	`bool`	True when every term has an observational formula, so that :meth:`estimate` can run on observational data.
`explanation`	`str`

Examples:

>>> import statspai as sp
>>> g = sp.dag("Z -> X; Z -> Y; X -> Y")
>>> res = sp.identify_counterfactual(
...     g, [("Y", 1, {"X": 0})], given=[("X", 1)])
>>> isinstance(res, sp.CounterfactualIdentification), res.identifiable
(True, True)

summary ¶

summary() -> str

A formatted account of the result.

Examples:

>>> import statspai as sp
>>> res = sp.identify_counterfactual(
...     sp.dag("X -> Y; X <-> Y"), [("Y", 1, {"X": 0})], given=[("X", 1)])
>>> "NOT IDENTIFIABLE" in res.summary()
True

evaluate ¶

evaluate(net: Any) -> float

Value of the estimand in a fully specified model.

Parameters:

Name	Type	Description	Default
`net`	`BayesNet`	A :func:`statspai.bayes_net` over the graph, latent variables included. Its interventional queries supply every term.	required

Returns:

Type	Description
`float`

Examples:

>>> import statspai as sp
>>> g = sp.dag("Z -> X; Z -> Y; X -> Y")
>>> net = sp.bayes_net(g, cpts={
...     "Z": {0: 0.5, 1: 0.5},
...     "X": {0: {0: 0.8, 1: 0.2}, 1: {0: 0.3, 1: 0.7}},
...     "Y": {(0, 0): {0: 0.9, 1: 0.1}, (0, 1): {0: 0.6, 1: 0.4},
...           (1, 0): {0: 0.5, 1: 0.5}, (1, 1): {0: 0.2, 1: 0.8}}})
>>> res = sp.identify_counterfactual(
...     g, [("Y", 1, {"X": 0})], given=[("X", 1)])
>>> round(res.evaluate(net), 4)   # sum_z P(z | X=1) P(Y=1 | X=0, z)
0.3333

estimate ¶

estimate(data: Any) -> float

Value of the estimand on categorical observational data.

Every interventional term is replaced by its observational formula (:func:statspai.identify) evaluated on the data's frequencies.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`	One categorical column per variable the formulas use.	required

Returns:

Type	Description
`float`

Raises:

Type	Description
`IdentificationFailure`	When a term needs an experiment. `experiments` and `observational` say which.

Examples:

>>> import numpy as np, pandas as pd, statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 40000
>>> z = rng.integers(0, 2, n)
>>> x = (rng.random(n) < np.where(z == 1, 0.7, 0.2)).astype(int)
>>> y = (rng.random(n) < 0.1 + 0.3 * x + 0.4 * z).astype(int)
>>> df = pd.DataFrame({"Z": z, "X": x, "Y": y})
>>> g = sp.dag("Z -> X; Z -> Y; X -> Y")
>>> ett0 = sp.identify_counterfactual(
...     g, [("Y", 1, {"X": 0})], given=[("X", 1)]).estimate(df)
>>> seen = df.loc[df.X == 1, "Y"].mean()
>>> round(float(seen - ett0), 1)   # effect of treatment on the treated
0.3

RuleCheck `dataclass` ¶

Result of checking one do-calculus rule on a DAG.

Attributes:

Name	Type	Description
`applicable`	`bool`	Whether the rule's d-separation condition holds.
`rule`	`int`	Which rule was checked (1, 2, or 3).
`reason`	`str`	Human-readable statement of the rule's d-separation condition.
`transformed`	`str`	The resulting interventional expression; unchanged if not applicable.

Examples:

>>> import statspai as sp
>>> rc = sp.RuleCheck(applicable=True, rule=1, reason="(Y ⊥ Z | X,W)",
...                   transformed="P(Y | do(X))")
>>> bool(rc.applicable), rc.rule
(True, 1)
>>> rc.transformed
'P(Y | do(X))'

DAG ¶

A directed acyclic graph for causal reasoning.

Parameters:

Name	Type	Description	Default
`spec`	`str`	Edge specification. Supported formats: `"X -> Y; Z -> X; Z -> Y"` (semicolon-separated) `"X -> Y\n Z -> X\n Z -> Y"` (newline-separated) `"X -> Y, Z -> X, Z -> Y"` (comma-separated) Arrows may point either way and chain: `"X <- Z -> Y"`. Bidirected (latent common cause): `"X <-> Y"` adds a latent node `_L_X_Y` with edges to both X and Y. A named node is declared unobserved with `"F [latent]"` (the dagitty spelling) or through `latent=`. A name on its own declares a node with no edges. Names may be quoted, which is how a name containing `;` or `,` is written: `'"Won Items" -> "In-game Purchases"'`. A Graphviz or dagitty wrapper is read as its body: `'digraph { "A" -> "B"; }'`, `"dag { A -> B }"`; so are groups, `"{A B} -> C"`. Attribute lists other than `[latent]` are ignored. A statement that is none of these, an undirected edge `X -- Y` or a cycle raises instead of being skipped.	`''`
`latent`	`iterable of str`	Names of unobserved nodes. They take part in every path, but are never offered as adjustment variables, and identification treats them as unmeasured.	`None`

Examples:

>>> import statspai as sp
>>> g = sp.DAG('Z -> X; Z -> Y; X -> Y')
>>> g.adjustment_sets('X', 'Y')
[{'Z'}]

>>> g = sp.DAG('X -> M -> Y; X -> Y; U <-> Y')
>>> sorted(g.nodes)
['M', 'U', 'X', 'Y', '_L_U_Y']

latent_nodes `property` ¶

latent_nodes: Set[str]

Unobserved nodes: declared ones and the _L_* of <->.

observed_nodes `property` ¶

observed_nodes: Set[str]

Nodes that are not latent.

add_edge ¶

add_edge(parent: str, child: str) -> 'DAG'

Add parent -> child; an edge that closes a cycle raises.

add_bidirected ¶

add_bidirected(a: str, b: str) -> 'DAG'

Add a latent common cause of a and b.

The latent is named _L_<a>_<b>. When that name is already taken by a node that is not the latent of this same pair (A_B <-> C and A <-> B_C both spell _L_A_B_C), a numeric suffix keeps the two latents apart.

set_latent ¶

set_latent(*names: str) -> 'DAG'

Declare nodes as unobserved.

ancestors ¶

ancestors(node: str) -> Set[str]

All ancestors of node (not including node itself).

descendants ¶

descendants(node: str) -> Set[str]

All descendants of node.

is_collider ¶

is_collider(node: str, path: List[str]) -> bool

Check if node is a collider on path.

A node is a collider on a path if both its neighbours on the path are parents of it (arrows point into it: → node ←).

all_paths ¶

all_paths(x: str, y: str, *, directed_only: bool = False) -> List[List[str]]

Enumerate all simple paths between x and y.

Parameters:

Name	Type	Description	Default
`x`	`str`	Start and end nodes.	required
`y`	`str`	Start and end nodes.	required
`directed_only`	`bool`	If True, only follow directed edges parent→child. If False (default), traverse edges in either direction (needed for finding backdoor paths).	`False`

Returns:

Type	Description
`list of list of str`	Each inner list is an ordered path from x to y.

causal_paths ¶

causal_paths(exposure: str, outcome: str) -> List[List[str]]

All directed (causal) paths from exposure to outcome.

These are the paths through which the treatment actually causes changes in the outcome.

backdoor_paths ¶

backdoor_paths(exposure: str, outcome: str) -> List[List[str]]

All backdoor paths from exposure to outcome.

A backdoor path is a path that starts with an arrow into the exposure (exposure <- ...). A path that leaves the exposure along an arrow and is not directed all the way, such as X -> M -> C <- Y, is neither causal nor backdoor: it is closed at its collider until something on or below the collider is conditioned on. :meth:noncausal_paths returns both kinds.

Examples:

>>> import statspai as sp
>>> g = sp.dag('Z -> X; Z -> Y; X -> M -> C; Y -> C')
>>> g.backdoor_paths('X', 'Y')
[['X', 'Z', 'Y']]
>>> g.noncausal_paths('X', 'Y') == [['X', 'Z', 'Y'], ['X', 'M', 'C', 'Y']]
True

noncausal_paths ¶

noncausal_paths(exposure: str, outcome: str) -> List[List[str]]

Every path from exposure to outcome that is not directed from one to the other: the backdoor paths, then the paths that leave the exposure along an arrow and meet a collider.

format_path ¶

format_path(path: List[str]) -> str

Write a path with its arrows: 'X <- Z -> Y'.

Examples:

>>> import statspai as sp
>>> sp.dag('Z -> X; Z -> Y').format_path(['X', 'Z', 'Y'])
'X <- Z -> Y'

is_path_open ¶

is_path_open(path: List[str], conditioned: Optional[Set[str]] = None) -> bool

Check if a path is open (active) given a conditioning set.

Rules (Pearl 2009): - A non-collider on the path blocks if conditioned on. - A collider on the path blocks unless it or a descendant is conditioned on.

path_status ¶

path_status(exposure: str, outcome: str, conditioned: Optional[Set[str]] = None) -> List[dict]

Classify every path between exposure and outcome.

Returns:

Type	Description
`list of dict`	Each dict has keys `'path'`, `'type'`, `'open'` (bool given the conditioning set) and `'arrows'` (the path written with its arrows). `'type'` is `'causal'` for a directed path, `'backdoor'` for a path that enters the exposure through an arrowhead, and `'noncausal'` for the rest: paths that leave the exposure along an arrow and meet a collider.

Examples:

>>> import statspai as sp
>>> g = sp.dag('Z -> X; Z -> Y; X -> Y')
>>> status = g.path_status('X', 'Y')
>>> sorted((s['type'], s['path'], s['open']) for s in status)
[('backdoor', ['X', 'Z', 'Y'], True), ('causal', ['X', 'Y'], True)]
>>> status = g.path_status('X', 'Y', conditioned={'Z'})
>>> sorted((s['type'], s['path'], s['open']) for s in status)
[('backdoor', ['X', 'Z', 'Y'], False), ('causal', ['X', 'Y'], True)]

classify_variable ¶

classify_variable(node: str, exposure: str, outcome: str) -> Set[str]

Classify the role(s) of node relative to exposure → outcome.

Returns a set that may include: 'confounder', 'mediator', 'collider', 'instrument', 'ancestor_of_treatment', 'ancestor_of_outcome'.

A 'confounder' lies on a backdoor path that has no collider, so the path carries association until something on it is adjusted for. A 'collider' is one on any path that is not causal. A mediator is not a confounder for sitting on a path such as X -> M -> C <- Y, which does not enter the exposure.

Examples:

>>> import statspai as sp
>>> g = sp.dag('Z -> X; Z -> Y; X -> Y')
>>> sorted(g.classify_variable('Z', 'X', 'Y'))
['ancestor_of_outcome', 'ancestor_of_treatment', 'confounder']

bad_controls ¶

bad_controls(exposure: str, outcome: str) -> dict

Identify variables that should not be conditioned on.

Returns a dict mapping variable names to the reason they are bad controls. Based on Cinelli, Forney & Pearl (2022) and the "bad controls" discussion in Cunningham (2021, ch. 3).

Categories of bad controls:

descendant_of_treatment: conditioning on a descendant of exposure blocks part of the causal effect (over-control bias).
collider: conditioning opens a previously closed backdoor path (collider bias / selection bias).
mediator: conditioning on a mediator blocks the indirect causal effect (over-control / mediation bias).
M-bias: conditioning on a pre-treatment variable that is a collider on a backdoor path, opening a non-causal path.

Examples:

>>> import statspai as sp
>>> g = sp.dag('D -> O -> Y; A -> O; A -> Y; D -> Y')
>>> bad = g.bad_controls('D', 'Y')
>>> list(bad)
['O']
>>> [reason.split(' — ')[0] for reason in bad['O']]
['mediator', 'collider']

do ¶

do(intervention: Union[str, Set[str]]) -> 'DAG'

Return the interventional graph G_{\overline{X}}: the graph with all incoming edges to the intervention node(s) removed.

This implements Pearl's do-operator at the graphical level.

Parameters:

Name	Type	Description	Default
`intervention`	`str or set of str`	The node(s) being intervened on.	required

Returns:

Type	Description
`DAG`	A new DAG with incoming edges to intervention removed.

Examples:

>>> import statspai as sp
>>> g = sp.dag('Z -> X -> Y; Z -> Y')
>>> g_do = g.do('X')
>>> sorted(g_do.edges)  # Z -> X edge is removed
[('X', 'Y'), ('Z', 'Y')]

instrumental_variables ¶

instrumental_variables(exposure: str, outcome: str, ancestral: bool = True) -> List[Dict[str, Any]]

Instruments for the total effect of exposure on outcome.

Each entry is {'instrument': z, 'conditioning': [w, ...]}: z is d-connected to the exposure given W and d-separated from the outcome given W once the first edges of the causal paths are removed, with W free of descendants of the causal paths. By default only ancestral instruments (W within the ancestors of the outcome or the instrument), as R dagitty::instrumentalVariables lists them; ancestral=False adds conditional instruments that need a collider in W. See :mod:statspai.dag.instruments.

Examples:

>>> import statspai as sp
>>> g = sp.dag("i -> x; x -> y; x <-> y; z -> y; z -> i")
>>> g.instrumental_variables("x", "y")
[{'instrument': 'i', 'conditioning': ['z']}]

frontdoor_sets ¶

frontdoor_sets(exposure: str, outcome: str) -> List[Set[str]]

Find sets satisfying Pearl's frontdoor criterion.

A set M satisfies the frontdoor criterion relative to (X, Y) if:

M intercepts all directed paths from X to Y.
There is no unblocked backdoor path from X to M.
All backdoor paths from M to Y are blocked by X.

Returns:

Type	Description
`list of set`	Valid frontdoor adjustment sets (possibly empty).

Examples:

>>> import statspai as sp
>>> g = sp.dag('U <-> X; U <-> Y; X -> M -> Y')
>>> g.frontdoor_sets('X', 'Y')
[{'M'}]

d_separated ¶

d_separated(x: str, y: str, conditioned: Optional[Set[str]] = None) -> bool

Test if x and y are d-separated given conditioned.

Uses the Bayes-Ball algorithm (Shachter 1998).

adjustment_sets ¶

adjustment_sets(exposure: str, outcome: str, method: str = 'backdoor', minimal: bool = True) -> List[Set[str]]

Find valid adjustment sets for estimating the causal effect of exposure on outcome.

Parameters:

Name	Type	Description	Default
`exposure`	`str`	Treatment and outcome nodes.	required
`outcome`	`str`	Treatment and outcome nodes.	required
`method`	`str`	`'backdoor'` — Pearl's backdoor criterion (default).	`'backdoor'`
`minimal`	`bool`	If True, return the minimal sufficient adjustment sets: valid sets none of whose proper subsets is valid. They need not all be the same size; they are the sets R `dagitty::adjustmentSets()` returns. If False, return every valid backdoor set, which excludes all descendants of the exposure. R `dagitty::adjustmentSets(type = "all")` applies the adjustment criterion instead and also lists sets holding descendants of the exposure that lie off every causal path (`X -> W` with `W` not on a path to the outcome).	`True`

Returns:

Type	Description
`list of set`	Each set is a valid adjustment set (possibly empty). An empty list means no set of observed variables satisfies the criterion.

Notes

Whether any valid set exists is decided exactly, at any size: if one does, the observed non-descendants of the exposure that are ancestors of the exposure or the outcome form one [@vanderzander2019separators]. Sets of up to six variables are then enumerated. When none that small exists, that ancestral set is pruned one variable at a time and the result returned, so a graph that needs seven or more adjusters gets one irreducible set rather than all of them.

Examples:

>>> import statspai as sp
>>> spec = '; '.join(f'Z{i} -> X; Z{i} -> Y' for i in range(8))
>>> g = sp.dag(spec + '; X -> Y')
>>> [sorted(s) for s in g.adjustment_sets('X', 'Y')]
[['Z0', 'Z1', 'Z2', 'Z3', 'Z4', 'Z5', 'Z6', 'Z7']]

latent_projection ¶

latent_projection() -> 'DAG'

Project the unobserved nodes out of the graph.

The result has the observed nodes only. It keeps a -> b when the graph has a directed path from a to b whose interior is unobserved, and adds a <-> b when a and b descend from a common unobserved node along paths whose interiors are unobserved (Verma and Pearl's projection). Every causal query about the observed variables has the same answer in the two graphs, which is how :func:statspai.identify handles a named latent with parents of its own.

Examples:

>>> import statspai as sp
>>> g = sp.dag('X -> F; F -> D; F -> M; D -> Y', latent=['F'])
>>> p = g.latent_projection()
>>> sorted(e for e in p.edges if not e[0].startswith('_L_'))
[('D', 'Y'), ('X', 'D'), ('X', 'M')]
>>> sorted(p.latent_nodes)
['_L_D_M']

implied_independencies ¶

implied_independencies(max_cond: Optional[int] = None) -> List[Tuple[str, str, Set[str]]]

Conditional independencies among observed variables that the graph implies.

For every pair of observed, non-adjacent nodes, lists each minimal set of observed variables that d-separates them (no proper subset does). These are the testable implications of the graph: if one fails in the data, the graph is wrong. Same list as dagitty::impliedConditionalIndependencies.

Parameters:

Name	Type	Description	Default
`max_cond`	`int`	Largest conditioning set to consider. The search is exponential in the number of observed nodes; set this on graphs with more than about 15 of them.	`None`

Returns:

Type	Description
`list of (x, y, set)`	`x` is independent of `y` given the set; sorted.

Examples:

>>> import statspai as sp
>>> g = sp.dag('Z -> X -> Y')
>>> g.implied_independencies()
[('Y', 'Z', {'X'})]

test_implications ¶

test_implications(data: 'pd.DataFrame', max_cond: Optional[int] = None, test: str = 'auto') -> 'pd.DataFrame'

Test the implied conditional independencies on data.

A small p-value is evidence against the graph; a large one is not evidence for it. With enough data every such test rejects a graph that is only approximately right, so read the size of the departure next to the p-value.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`	One column per observed node that appears in an implication.	required
`max_cond`	`int`	Passed to :meth:`implied_independencies`.	`None`
`test`	`('auto', 'fisher-z', 'chi-square', 'g-test')`	`'fisher-z'`: the partial correlation of `x` and `y` given `Z`, zero under the independence if the relations are linear; `atanh(r) * sqrt(n - \|Z\| - 3)` is standard normal under the null, as in `dagitty::localTests(type = "cis")`. Numeric columns. `'chi-square'`: for categorical columns. Pearson's statistic is computed in each stratum of `Z` and summed; the degrees of freedom are summed likewise, counting in each stratum only the levels that occur there. Same as `dagitty::localTests(type = "cis.chisq")` and `bnlearn::ci.test(test = "x2-adf")`. `'g-test'`: the likelihood-ratio statistic in place of Pearson's (`bnlearn::ci.test(test = "mi-adf")`). `'auto'`: `'chi-square'` when any column involved is not numeric, `'fisher-z'` otherwise.	`'auto'`

Returns:

Type Description

DataFrame

One row per implication, with x, y, given, p_value and p_holm (Holm-adjusted over the implications tested). The Fisher z test adds partial_corr; the categorical tests add statistic, df and cramers_v (the square root of the statistic over n times the smaller table dimension less one, a measure of the departure that does not grow with n). An implication no stratum can test has df zero and a missing p-value.

Examples:

>>> import numpy as np, pandas as pd, statspai as sp
>>> rng = np.random.default_rng(0)
>>> z = rng.normal(size=500)
>>> x = z + rng.normal(size=500)
>>> y = x + rng.normal(size=500)
>>> df = pd.DataFrame({'Z': z, 'X': x, 'Y': y})
>>> out = sp.dag('Z -> X -> Y').test_implications(df)
>>> bool(out['p_value'].iloc[0] > 0.05)  # Y _||_ Z | X holds
True

Categorical data are tested with the chi-square statistic:

>>> lab = pd.DataFrame({
...     'Z': np.where(z > 0, 'hi', 'lo'),
...     'X': np.where(x > 0, 'hi', 'lo'),
...     'Y': np.where(y > 0, 'hi', 'lo'),
... })
>>> out = sp.dag('Z -> X -> Y').test_implications(lab)
>>> list(out.columns)
['x', 'y', 'given', 'statistic', 'df', 'cramers_v', 'p_value', 'p_holm']

equivalent_dags ¶

equivalent_dags(max_edges: int = 20) -> List['DAG']

Every DAG in the Markov equivalence class of this one.

Two DAGs imply the same conditional independencies exactly when they have the same adjacencies and the same unshielded colliders. No data set, however large, can tell the members of the class apart, so a causal claim that depends on the direction of an edge that flips inside the class rests on background knowledge alone.

Parameters:

Name	Type	Description	Default
`max_edges`	`int`	The search enumerates the `2 ** n_edges` orientations of the skeleton; graphs with more edges are refused.	`20`

Returns:

Type	Description
`list of DAG`	The members of the class, this graph included, in a deterministic order. Latent declarations are carried over.

Examples:

>>> import statspai as sp
>>> g = sp.dag("T -> X; T -> Y; X -> Y; Z -> Y")
>>> len(g.equivalent_dags())
2
>>> g.equivalence_class()["undirected"]
[('T', 'X')]

equivalence_class ¶

equivalence_class(max_edges: int = 20) -> Dict[str, Any]

The Markov equivalence class as a partially directed graph.

Returns:

Type	Description
`dict`	`'directed'`: edges `(parent, child)` that point the same way in every member of the class; `'undirected'`: edges `(a, b)` that point either way in some member, i.e. the ones the data cannot orient; `'n_dags'`: size of the class.

Examples:

>>> import statspai as sp
>>> cls = sp.dag("A -> C; B -> C").equivalence_class()
>>> cls["n_dags"], cls["undirected"]
(1, [])

to_ascii ¶

to_ascii() -> str

Simple text representation of the DAG.

plot ¶

plot(exposure: Optional[str] = None, outcome: Optional[str] = None, conditioned: Optional[Set[str]] = None, positions: Optional[Dict[str, Tuple[float, float]]] = None, figsize: Tuple[float, float] = (8, 6), seed: int = 42, title: Optional[str] = None, style: str = 'ggdag', node_size: float = 0.22, font_size: int = 12, ax: Any = None) -> Tuple[Any, Any]

Plot the DAG with publication-quality styling.

When exposure and outcome are provided, nodes are colour-coded by causal role (like R's ggdag):

Exposure: green
Outcome: blue
Confounder: orange
Mediator: purple
Collider / bad control: red
Unobserved (latent): grey dashed outline
Adjusted / conditioned: hatched fill

Bidirected edges (latent common causes) are rendered as curved dashed arcs rather than routing through hidden nodes.

Parameters:

Name	Type	Description	Default
`exposure`	`str`	Treatment and outcome nodes for role colouring.	`None`
`outcome`	`str`	Treatment and outcome nodes for role colouring.	`None`
`conditioned`	`set of str`	Nodes being conditioned on (shown with hatched fill).	`None`
`positions`	`dict`	`{node_name: (x, y)}` for custom layout. If `None`, an automatic topological layout is used.	`None`
`figsize`	`tuple`	Figure size (width, height) in inches.	`(8, 6)`
`seed`	`int`	Random seed for layout jitter.	`42`
`title`	`str`	Plot title. Auto-generated if exposure/outcome given.	`None`
`style`	`str`	`'ggdag'` (default, clean white) or `'classic'` (grey background, smaller nodes).	`'ggdag'`
`node_size`	`float`	Radius of node circles in data coordinates.	`0.22`
`font_size`	`int`	Font size for node labels.	`12`
`ax`	`matplotlib Axes`	Axes to draw on. If `None`, creates a new figure.	`None`

Returns:

Type	Description
`(fig, ax) : matplotlib figure and axes`

summary ¶

summary(exposure: str, outcome: str) -> str

Print a rich text summary of the DAG for identification analysis.

Shows all paths, their status, adjustment sets, bad controls, and variable roles — a one-stop diagnostic like dagitty's adjustmentSets + paths.

Examples:

>>> import statspai as sp
>>> g = sp.dag('Z -> X; Z -> Y; X -> Y')
>>> text = g.summary('X', 'Y')
>>> print(text.splitlines()[0])
DAG Summary: effect of X on Y

IdentificationResult `dataclass` ¶

Bases: ResultProtocolMixin

Outcome of an identification query.

Attributes:

Name	Type	Description
`identifiable`	`bool`	True iff P(Y \| do(X)) is identifiable from the observed distribution.
`estimand`	`str`	Do-free formula when identifiable; a structured hedge otherwise.
`c_components`	`list[set[str]]`	The c-components of the ancestral semi-Markovian graph G[An(Y)].
`hedge`	`tuple[frozenset, frozenset] \| None`	Witness C-forest pair (F, F') that proves non-identifiability.
`explanation`	`str`	Human-readable proof / refutation.

Examples:

>>> import statspai as sp
>>> g = sp.dag("Z -> X; Z -> Y; X -> Y")
>>> res = sp.identify(g, treatment="X", outcome="Y")
>>> isinstance(res, sp.IdentificationResult)
True
>>> bool(res.identifiable)
True
>>> "Y" in res.estimand
True

summary ¶

summary() -> str

Return a formatted multi-line summary of the ID query.

Every StatsPAI result object exposes .summary() (CLAUDE.md §3.3); this one renders the identifiability verdict, the do-free estimand or the hedge witnessing non-identifiability, and the c-components of the ancestral semi-Markovian graph.

Examples:

>>> import statspai as sp
>>> g = sp.dag("Z -> X; Z -> Y; X -> Y")
>>> res = sp.identify(g, treatment="X", outcome="Y")
>>> "IDENTIFIABLE" in res.summary()
True

estimate ¶

estimate(data: Any, *, n_boot: int = 0, alpha: float = 0.05, seed: int | None = None) -> Any

Evaluate the estimand on categorical data: P(Y | do(X)).

Each probability in the formula is replaced by the matching frequency in the data and the sums are carried out. Only the observed variables are used, so this works where a latent confounder rules out fitting the whole graph (the front door, for one).

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`	One column for every variable in the estimand, each treated as categorical.	required
`n_boot`	`int`	Nonparametric bootstrap replications. When positive, `se`, `ci_lower` and `ci_upper` (percentile) are added.	`0`
`alpha`	`float`	Level of the bootstrap interval.	`0.05`
`seed`	`int`	Seed for the bootstrap.	`None`

Returns:

Type	Description
`DataFrame`	One row per combination of treatment and outcome values, with `prob`, the estimate of `P(outcome \| do(treatment))`.

Notes

A term such as P(y | x', m) is undefined where (x', m) never occurs. When the answer for a treatment value depends on such a term, the data do not determine it: those rows are returned as missing and an :class:~statspai.exceptions.AssumptionWarning says which. This is the positivity condition of the estimand, checked on the data.

When the formula keeps a variable free that is neither treatment nor outcome (the algorithm may intervene on variables that cannot affect the outcome, and the result does not depend on their value in the population), it is averaged over that variable's observed distribution.

Examples:

>>> import numpy as np, pandas as pd, statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 20000
>>> u = rng.random(n) < 0.5                         # unobserved
>>> x = rng.random(n) < np.where(u, 0.8, 0.2)
>>> m = rng.random(n) < np.where(x, 0.9, 0.2)
>>> y = rng.random(n) < 0.1 + 0.5 * m + 0.3 * u
>>> df = pd.DataFrame({"X": x, "M": m, "Y": y})
>>> g = sp.dag("X -> M -> Y; U -> X; U -> Y", latent=["U"])
>>> res = sp.identify(g, "X", "Y")
>>> est = res.estimate(df)
>>> p1 = est.query("X == True and Y == True")["prob"].item()
>>> p0 = est.query("X == False and Y == True")["prob"].item()
>>> round(p1 - p0, 2)        # true effect 0.5 * (0.9 - 0.2)
0.35

LLMDAGResult `dataclass` ¶

Bases: ResultProtocolMixin

Merged LLM-oracle / CI-test DAG returned by :func:llm_dag.

Attributes:

Name	Type	Description
`edges`	`list of (str, str)`	Final merged edge set after `merge_strategy` is applied.
`oracle_edges`	`list of (str, str)`	Raw `(from, to)` edges proposed by the LLM oracle.
`ci_rejects, ci_asserts`	`list of (str, str)`	Variable pairs the CI-test skeleton rejected / asserted.
`disagreements`	`list of (str, str, str)`	`(from, to, reason)` where oracle and CI test conflict.
`provenance`	`dict`	Bookkeeping: oracle error (if any), merge strategy, ci_test, alpha.

Examples:

>>> import statspai as sp
>>> # An oracle is any callable: (variables, descriptions) -> edges.
>>> def domain_oracle(variables, descriptions):
...     return [("smoking", "cancer")]  # e.g. an LLM API call
>>> res = sp.llm_dag(
...     variables=["smoking", "cancer", "tar"],
...     oracle=domain_oracle,
...     merge_strategy="oracle_only",
... )
>>> res.edges          # final merged edges
>>> res.disagreements  # oracle vs CI-test conflicts

LLMCausalAssessResult `dataclass` ¶

Bases: ResultProtocolMixin

Output of :func:llm_causal_assess.

Attributes:

Name	Type	Description
`level1_accuracy, level2_accuracy`	`float or None`	Per-level accuracy (`None` when that level was not assessed).
`per_item`	`DataFrame`	One row per question: `id, level, question, truth, pred, correct`.
`llm_identifier`	`str`	Label of the assessed model.

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> # llm_client wraps any model API: prompt (str) -> response (str).
>>> def llm_client(question):
...     return "the answer is Y"  # e.g. an Anthropic API call
>>> level1 = pd.DataFrame({
...     "question": ["Does X cause Y?"],
...     "answer": ["Y"],
... })
>>> res = sp.llm_causal_assess(
...     level1_items=level1, llm_client=llm_client,
...     llm_identifier="my-model",
... )
>>> res.level1_accuracy
>>> print(res.summary())

PairwiseBenchmarkResult `dataclass` ¶

Bases: ResultProtocolMixin

Output of :func:pairwise_causal_benchmark.

Attributes:

Name	Type	Description
`accuracy`	`float`	Fraction of pairs whose predicted direction matches the truth.
`precision_forward, recall_forward`	`float`	Precision / recall for the `A -> B` (forward) class.
`per_pair`	`DataFrame`	One row per pair: `A, B, truth, pred, correct, raw_response`.

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> # llm_client wraps any model API: prompt (str) -> response (str).
>>> def llm_client(prompt):
...     return "yes"  # e.g. an Anthropic API call
>>> gt = pd.DataFrame({
...     "A": ["smoking"], "B": ["cancer"], "a_causes_b": [True],
... })
>>> res = sp.pairwise_causal_benchmark(
...     gt, llm_client=llm_client,
... )
>>> res.accuracy, res.precision_forward
>>> print(res.summary())

SWIGGraph ¶

Single World Intervention Graph for a DAG under do(X=x).

Attributes:

Name	Type	Description
`parent`	`DAG`	Source DAG.
`intervention`	`dict[str, str]`	Variable → value label.
`nodes`	`set[str]`	All SWIG nodes (split halves + potential outcomes).
`edges`	`dict[str, set[str]]`	Adjacency map on SWIG nodes.

Examples:

>>> import statspai as sp
>>> g = sp.dag("L -> X; L -> Y; X -> Y")
>>> sw = sp.swig(g, {"X": "x"})
>>> isinstance(sw, sp.SWIGGraph)
True
>>> len(sw.nodes)
4
>>> print(sw.ascii())
SWIG under do({'X': 'x'}):
  L(X=x) -> X
  L(X=x) -> Y(X=x)
  X(x) -> Y(X=x)

counterfactual_nodes ¶

counterfactual_nodes() -> set[str]

Return only the potential-outcome / action-half labels.

ascii ¶

ascii() -> str

Compact edge-list representation.

identify_counterfactual ¶

identify_counterfactual(dag: Any, event: Any, given: Any = None) -> CounterfactualIdentification

Decide whether a counterfactual probability is identifiable, and from what (ID* and IDC*, Shpitser and Pearl).

Parameters:

Name	Type	Description	Default
`dag`	`DAG`	The causal graph. Latent variables are declared with `latent=[...]` or as bidirected edges `A <-> B`.	required
`event`	`list of tuple`	The counterfactual events whose joint probability is wanted. Each is `(variable, value)` for the world as it is, or `(variable, value, {intervened: value, ...})` for the world in which those variables are set. `[("Y", 1, {"X": 0})]` is `Y_{X=0} = 1`.	required
`given`	`list of tuple`	Events to condition on, written the same way. They may refer to other worlds than `event` does.	`None`

Returns:

Type	Description
`CounterfactualIdentification`	`identifiable`, the `estimand` in interventional distributions, which of its terms reduce to observational data, and `evaluate(net)` / `estimate(data)` for a number.

Notes

Identifiable here means: computable from the collection of all interventional distributions, that is from experiments. That is the most one can ask of a counterfactual without stating the structural functions. from_observational_data then says whether those experiments are themselves replaceable by observational data in this graph.

Examples:

The effect of treatment on the treated is the standard example. P(Y_{X=0} = y | X = 1) is identified whenever the effect of X on Y has no unblocked confounding, and not otherwise:

>>> import statspai as sp
>>> ok = sp.identify_counterfactual(
...     sp.dag("Z -> X; Z -> Y; X -> Y"), [("Y", 1, {"X": 0})], given=[("X", 1)])
>>> ok.identifiable, ok.from_observational_data
(True, True)
>>> print(ok.estimand.split(" / ")[0])
[sum_{Z} [P(X=1 | do(Z)) * P(Y=1 | do(X=0, Z)) * P(Z)]]
>>> bow = sp.identify_counterfactual(
...     sp.dag("X -> Y; X <-> Y"), [("Y", 1, {"X": 0})], given=[("X", 1)])
>>> bow.identifiable
False

A probability of necessity, P(Y_{X=0} = 0 | X = 1, Y = 1), is not identified even without confounding: it needs the joint of Y in two worlds, which no experiment reveals.

>>> pn = sp.identify_counterfactual(
...     sp.dag("X -> Y"), [("Y", 0, {"X": 0})], given=[("X", 1), ("Y", 1)])
>>> pn.identifiable
False

References

[@shpitser2008complete]

apply_rules ¶

apply_rules(dag: Any, Y: NodeInput, X: NodeInput, Z: NodeInput, W: NodeInput = None) -> list[RuleCheck]

Try all three rules and return every applicable simplification.

Examples:

>>> import statspai as sp
>>> g = sp.dag('Z -> X -> Y')
>>> checks = sp.do_calculus_apply(g, Y='Y', X='X', Z='Z')
>>> [c.rule for c in checks]
[1, 2, 3]
>>> bool(checks[0].applicable)  # Rule 1 fires on this chain
True

rule1 ¶

rule1(dag: Any, Y: NodeInput, X: NodeInput, Z: NodeInput, W: NodeInput = None) -> RuleCheck

Check Rule 1: can we insert or delete observation of Z?

Rule 1 licenses P(y | do(x), z, w) = P(y | do(x), w) when (Y ⊥ Z | X, W) holds in the graph with edges into X deleted.

Examples:

>>> import statspai as sp
>>> g = sp.dag('Z -> X -> Y')
>>> chk = sp.do_rule1(g, Y='Y', X='X', Z='Z')
>>> bool(chk.applicable)  # observing Z is irrelevant once we do(X)
True
>>> chk.rule
1

rule2 ¶

rule2(dag: Any, Y: NodeInput, X: NodeInput, Z: NodeInput, W: NodeInput = None) -> RuleCheck

Check Rule 2: can do(Z) be swapped for observing Z?

Rule 2 licenses P(y | do(x), do(z), w) = P(y | do(x), z, w) when (Y ⊥ Z | X, W) holds in the graph with edges into X and edges out of Z deleted (action/observation exchange).

Examples:

>>> import statspai as sp
>>> g = sp.dag('X -> Y; Z -> Y')
>>> chk = sp.do_rule2(g, Y='Y', X='X', Z='Z')
>>> bool(chk.applicable)  # no back-door from Z to Y, so do(Z) == observe Z
True
>>> chk.rule
2

rule3 ¶

rule3(dag: Any, Y: NodeInput, X: NodeInput, Z: NodeInput, W: NodeInput = None) -> RuleCheck

Check Rule 3: can we delete do(Z)?

Rule 3 licenses P(y | do(x), do(z), w) = P(y | do(x), w) when (Y ⊥ Z | X, W) holds in the graph with edges into X and into Z(W) deleted (insertion/deletion of actions).

Examples:

>>> import statspai as sp
>>> g = sp.dag('X -> Y; Z -> W')
>>> chk = sp.do_rule3(g, Y='Y', X='X', Z='Z')
>>> bool(chk.applicable)  # Z has no effect on Y, so do(Z) drops out
True
>>> chk.rule
3

dag ¶

dag(spec: str = '', latent: Optional[Iterable[str]] = None) -> DAG

Create a causal DAG from a string specification.

Parameters:

Name	Type	Description	Default
`spec`	`str`	Edge specification. Examples: `"X -> Y; Z -> X; Z -> Y"` `"X -> M -> Y; X -> Y"` `"X <-> Y; Z -> X; Z -> Y"` (bidirected = latent common cause) `"F -> X; F -> D; F [latent]"` (a named unobserved node)	`''`
`latent`	`iterable of str`	Names of unobserved nodes, as an alternative to `[latent]` in the specification.	`None`

Returns:

Type	Description
`DAG`

Examples:

>>> import statspai as sp
>>> g = sp.dag('Z -> X -> Y; Z -> Y')
>>> g.adjustment_sets('X', 'Y')
[{'Z'}]

No valid adjustment set exists with an unobserved confounder:

>>> g = sp.dag('X -> Y; X <-> Y')
>>> g.adjustment_sets('X', 'Y')
[]

A named latent is never offered as an adjustment variable:

>>> g = sp.dag('D -> Y; X -> D; X -> Y; F -> X; F -> D', latent=['F'])
>>> g.adjustment_sets('D', 'Y')
[{'X'}]

dag_example ¶

dag_example(name: str) -> DAG

Load a classic textbook DAG by name.

Parameters:

Name	Type	Description	Default
`name`	`str`	One of: `'confounding'`, `'collider'`, `'mediation'`, `'discrimination'`, `'movie_star'`, `'police'`, `'frontdoor'`, `'bad_control_earnings'`, `'m_bias'`.	required

Returns:

Type	Description
`DAG`	The example DAG. Call `.summary(exposure, outcome)` for analysis, or `.plot(exposure, outcome)` for a ggdag-style visualisation with role-coloured nodes.

Examples:

>>> import statspai as sp
>>> g = sp.dag_example('discrimination')
>>> sorted(g.nodes)
['A', 'D', 'O', 'Y']
>>> print(g.summary('D', 'Y').splitlines()[0])
DAG Summary: effect of D on Y
>>> fig, ax = g.plot('D', 'Y')

>>> g = sp.dag_example('frontdoor')
>>> pos = sp.dag_example_positions('frontdoor')
>>> fig, ax = g.plot('X', 'Y', positions=pos)

dag_example_positions ¶

dag_example_positions(name: str) -> Dict[str, Tuple[float, float]]

Return hand-tuned node positions for a named example DAG.

The returned {node: (x, y)} mapping can be passed to DAG.plot(..., positions=...) for a reproducible layout.

Examples:

>>> import statspai as sp
>>> pos = sp.dag_example_positions('frontdoor')
>>> sorted(pos.keys())
['M', 'X', 'Y']
>>> pos['X']
(-1.5, 0)

dag_examples ¶

dag_examples() -> List[str]

List available classic DAG examples.

Examples:

>>> import statspai as sp
>>> names = sp.dag_examples()
>>> 'confounding' in names and 'frontdoor' in names
True

dag_simulate ¶

dag_simulate(name: str, n: int = 10000, seed: int = 42) -> 'pd.DataFrame'

Run a classic DAG simulation from Cunningham (2021, ch. 3).

Available simulations:

'discrimination' — Gender discrimination / occupational sorting. True effect of discrimination on wage is -1. Conditioning on occupation alone flips the sign (collider bias).
'movie_star' — Beauty–Talent collider. Beauty and Talent are independent in the population, but conditioning on Star status induces a spurious negative correlation.

Parameters:

Name	Type	Description	Default
`name`	`str`	`'discrimination'` or `'movie_star'`.	required
`n`	`int`	Number of observations (default 10000).	`10000`
`seed`	`int`	Random seed for reproducibility.	`42`

Returns:

Type	Description
`DataFrame`	Simulated dataset.

Examples:

>>> import statspai as sp
>>> df = sp.dag_simulate('discrimination')
>>> df.shape
(10000, 5)
>>> import statsmodels.formula.api as smf
>>> # Biased: wrong sign due to collider
>>> biased = smf.ols('wage ~ female + occupation', data=df).fit()
>>> bool(biased.params['female'] > 0)
True
>>> # Correct: includes ability
>>> fit = smf.ols('wage ~ female + occupation + ability', data=df).fit()
>>> bool(fit.params['female'] < 0)
True

identify ¶

identify(dag: Any, treatment: NodeInput, outcome: NodeInput) -> IdentificationResult

Run Shpitser-Pearl ID algorithm on dag.

Parameters:

Name	Type	Description	Default
`dag`	`DAG`	`statspai.dag.DAG` instance, possibly with latent nodes `_L_*` (representing bidirected edges) or named latents declared with `sp.dag(..., latent=[...])`; the latter are projected out with :meth:`DAG.latent_projection` first.	required
`treatment`	`str \| Iterable[str]`	Set of variables X being intervened on.	required
`outcome`	`str \| Iterable[str]`	Set of outcome variables Y.	required

Returns:

Type	Description
`IdentificationResult`

Examples:

>>> import statspai as sp
>>> # Backdoor confounder -- P(Y | do(X)) is identifiable
>>> g = sp.dag("Z -> X; Z -> Y; X -> Y")
>>> res = sp.identify(g, treatment="X", outcome="Y")
>>> bool(res.identifiable)
True
>>> # Bow arc (X <-> Y latent confounding) -- NOT identifiable
>>> g2 = sp.dag("X -> Y; X <-> Y")
>>> res2 = sp.identify(g2, treatment="X", outcome="Y")
>>> bool(res2.identifiable)
False
>>> "hedge" in res2.estimand
True

llm_causal_assess ¶

llm_causal_assess(level1_items: Optional[DataFrame] = None, level2_items: Optional[DataFrame] = None, *, llm_client: Callable[[str], str], llm_identifier: str = 'llm') -> LLMCausalAssessResult

Combined Level-1 + Level-2 LLM causal-reasoning assessment.

Parameters:

Name	Type	Description	Default
`level1_items`	`DataFrame`	Columns: `question`, `answer`. The LLM's response is marked correct if the target `answer` appears (case-insensitive) in the response.	`None`
`level2_items`	`DataFrame`	Columns: `question`, `answer`. Level-2 questions ask the LLM to reason about a DAG fragment; answer checking uses substring match.	`None`
`llm_client`	`Callable[[str], str]`		required
`llm_identifier`	`Callable[[str], str]`		required

Returns:

Type	Description
`LLMCausalAssessResult`

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> def stub_client(question):
...     # echo the last token so substring matching scores it correct
...     return "The cause is " + question.split()[-1]
>>> level1 = pd.DataFrame({
...     "question": ["Does X cause Y", "Does A cause B"],
...     "answer": ["Y", "B"],
... })
>>> res = sp.llm_causal_assess(
...     level1_items=level1, llm_client=stub_client, llm_identifier="stub",
... )
>>> res.level1_accuracy
1.0

pairwise_causal_benchmark ¶

pairwise_causal_benchmark(ground_truth: DataFrame, *, llm_client: Callable[[str], str], llm_identifier: str = 'llm', pair_a_col: str = 'A', pair_b_col: str = 'B', truth_col: str = 'a_causes_b', prompt_template: str = "Does variable {a} causally influence variable {b}? Answer 'yes' or 'no'.") -> PairwiseBenchmarkResult

Benchmark an LLM on pairwise causal-direction identification.

Parameters:

Name	Type	Description	Default
`ground_truth`	`DataFrame`	One row per pair with columns `pair_a_col`, `pair_b_col`, and a boolean `truth_col` indicating whether A causally influences B.	required
`llm_client`	`callable(str) -> str`	Function taking a prompt and returning a string.	required
`llm_identifier`	`str`		`'llm'`
`prompt_template`	`str`		``"Does variable {a} causally ..."``

Returns:

Type	Description
`PairwiseBenchmarkResult`

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> def stub_client(prompt):
...     # toy oracle: only "smoking" prompts get a "yes"
...     return "yes" if "smoking" in prompt.lower() else "no"
>>> gt = pd.DataFrame({
...     "A": ["smoking", "ice_cream"],
...     "B": ["cancer", "drowning"],
...     "a_causes_b": [True, False],
... })
>>> res = sp.pairwise_causal_benchmark(gt, llm_client=stub_client)
>>> res.accuracy
1.0

recommend_estimator ¶

recommend_estimator(dag: Any, exposure: str, outcome: str, candidate_instruments: Optional[Sequence[str]] = None) -> EstimatorRecommendation

Inspect a DAG and recommend a statspai estimator.

Parameters:

Name	Type	Description	Default
`dag`	`DAG`		required
`exposure`	`str`		required
`outcome`	`str`		required
`candidate_instruments`	`list of str`	Variable names to check as potential IVs. If omitted, all observed nodes other than exposure/outcome are considered.	`None`

Examples:

>>> import statspai as sp
>>> g = sp.dag("Z -> X; Z -> Y; X -> Y")  # Z confounds X -> Y
>>> rec = sp.dag_recommend_estimator(g, exposure="X", outcome="Y")
>>> rec.estimator
'regress'
>>> "Z" in rec.adjustment_set  # backdoor path blocked by conditioning on Z
True

statspai.dag¶

dag ¶

BayesNet ¶

parents ¶

cpt ¶

summary ¶

do ¶

query ¶

prob ¶

expectation ¶

counterfactual ¶

simulate ¶

SCM dataclass ¶

counterfactual ¶

CounterfactualIdentification dataclass ¶

summary ¶

evaluate ¶

estimate ¶

RuleCheck dataclass ¶

DAG ¶

latent_nodes property ¶

observed_nodes property ¶

add_edge ¶

add_bidirected ¶

set_latent ¶

ancestors ¶

descendants ¶

is_collider ¶

all_paths ¶

causal_paths ¶

backdoor_paths ¶

noncausal_paths ¶

format_path ¶

is_path_open ¶

path_status ¶

classify_variable ¶

bad_controls ¶

do ¶

instrumental_variables ¶

frontdoor_sets ¶

d_separated ¶

adjustment_sets ¶

latent_projection ¶

implied_independencies ¶

test_implications ¶

equivalent_dags ¶

equivalence_class ¶

to_ascii ¶

plot ¶

summary ¶

IdentificationResult dataclass ¶

summary ¶

estimate ¶

LLMDAGResult dataclass ¶

LLMCausalAssessResult dataclass ¶

PairwiseBenchmarkResult dataclass ¶

SWIGGraph ¶

counterfactual_nodes ¶

ascii ¶

identify_counterfactual ¶

apply_rules ¶

rule1 ¶

rule2 ¶

rule3 ¶

dag ¶

dag_example ¶

dag_example_positions ¶

dag_examples ¶

dag_simulate ¶

identify ¶

llm_causal_assess ¶

pairwise_causal_benchmark ¶

recommend_estimator ¶

`statspai.dag`¶

SCM `dataclass` ¶

CounterfactualIdentification `dataclass` ¶

RuleCheck `dataclass` ¶

latent_nodes `property` ¶

observed_nodes `property` ¶

IdentificationResult `dataclass` ¶

LLMDAGResult `dataclass` ¶

LLMCausalAssessResult `dataclass` ¶

PairwiseBenchmarkResult `dataclass` ¶