`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') g.bad_controls('X', 'Y') g.summary('X', 'Y') g.do('X') # interventional graph sp.dag_example('discrimination') # classic textbook DAG

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) Bidirected (latent common cause): `"X <-> Y"` adds a latent node `_L_X_Y` with edges to both X and Y.	`''`

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']

observed_nodes `property` ¶

observed_nodes: Set[str]

Nodes that are not latent (latents start with _L_).

add_bidirected ¶

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

Add a latent common cause of a and b.

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 (non-causal) paths from exposure to outcome.

A backdoor path is any path that starts with an arrow into the exposure (← exposure), creating spurious association.

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'` (`'causal'` or `'backdoor'`), and `'open'` (bool given conditioning set).

Examples:

>>> g = sp.dag('Z -> X; Z -> Y; X -> Y')
>>> g.path_status('X', 'Y')
[{'path': ['X', 'Y'], 'type': 'causal', 'open': True},
 {'path': ['X', 'Z', 'Y'], 'type': 'backdoor', 'open': True}]
>>> g.path_status('X', 'Y', conditioned={'Z'})
[{'path': ['X', 'Y'], 'type': 'causal', 'open': True},
 {'path': ['X', 'Z', 'Y'], 'type': 'backdoor', 'open': False}]

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'.

Examples:

>>> g = sp.dag('Z -> X; Z -> Y; X -> Y')
>>> g.classify_variable('Z', 'X', 'Y')
{'confounder', 'ancestor_of_treatment', 'ancestor_of_outcome'}

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:

>>> g = sp.dag('D -> O -> Y; A -> O; A -> Y; D -> Y')
>>> g.bad_controls('D', 'Y')
{'O': ['collider — conditioning opens D→O←A→Y']}

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:

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

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:

>>> 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 only minimal sufficient adjustment sets.	`True`

Returns:

Type	Description
`list of set`	Each set is a valid adjustment set (possibly empty).

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:

>>> g = sp.dag('Z -> X; Z -> Y; X -> Y')
>>> print(g.summary('X', 'Y'))

IdentificationResult `dataclass` ¶

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

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))'

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.

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.

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())

dag ¶

dag(spec: str = '') -> 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)	`''`

Returns:

Type	Description
`DAG`

Examples:

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

>>> g = sp.dag('X -> Y; X <-> Y')  # unobserved confounder
>>> g.adjustment_sets('X', 'Y')
[]  # no valid adjustment set exists

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:

>>> g = sp.dag_example('discrimination')
>>> print(g.summary('D', 'Y'))
>>> g.plot('D', 'Y')

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

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_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_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:

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

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).	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

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

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

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 ¶

DAG ¶

observed_nodes property ¶

add_bidirected ¶

ancestors ¶

descendants ¶

is_collider ¶

all_paths ¶

causal_paths ¶

backdoor_paths ¶

is_path_open ¶

path_status ¶

classify_variable ¶

bad_controls ¶

do ¶

frontdoor_sets ¶

d_separated ¶

adjustment_sets ¶

to_ascii ¶

plot ¶

summary ¶

IdentificationResult dataclass ¶

RuleCheck dataclass ¶

SWIGGraph ¶

counterfactual_nodes ¶

ascii ¶

SCM dataclass ¶

counterfactual ¶

LLMDAGResult dataclass ¶

LLMCausalAssessResult dataclass ¶

PairwiseBenchmarkResult dataclass ¶

dag ¶

dag_example ¶

dag_examples ¶

dag_example_positions ¶

dag_simulate ¶

identify ¶

rule1 ¶

rule2 ¶

rule3 ¶

apply_rules ¶

llm_causal_assess ¶

pairwise_causal_benchmark ¶

recommend_estimator ¶

`statspai.dag`¶

observed_nodes `property` ¶

IdentificationResult `dataclass` ¶

RuleCheck `dataclass` ¶

SCM `dataclass` ¶

LLMDAGResult `dataclass` ¶

LLMCausalAssessResult `dataclass` ¶

PairwiseBenchmarkResult `dataclass` ¶