`statspai.transport`¶

transport ¶

Transportability (sp.transport): generalize causal effects across populations. Combines Pearl-Bareinboim identification (selection diagrams) with Dahabreh-Stuart-style density-ratio weighting.

Quick start

import statspai as sp g = sp.dag("S -> X; X -> Y; S -> Y") sp.transport.identify_transport(g, treatment="X", outcome="Y", ... selection_nodes={"S"}) sp.transport.weights(source=rct, target=target_df, ... features=["age", "sex"], ... treatment="treat", outcome="y")

TransportWeightResult `dataclass` ¶

Bases: ResultProtocolMixin

Result of density-ratio (inverse-odds-of-sampling) transport weighting.

Attributes:

Name	Type	Description
`weights`	`ndarray`	Per-source-unit transport weights.
`ess`	`float`	Effective sample size after weighting.
`max_weight`	`float`	Largest weight (after optional truncation).
`effect_source`	`float`	Unweighted difference-in-means effect in the source population.
`effect_transported`	`float`	Weighted effect transported to the target population.
`se_transported`	`float`	Standard error of the transported effect.

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> ns, nt = 500, 500
>>> source = pd.DataFrame({
...     "x": rng.normal(0, 1, ns),
...     "a": rng.integers(0, 2, ns),
... })
>>> source["y"] = source["a"] + 0.5 * source["x"] + rng.normal(0, 1, ns)
>>> target = pd.DataFrame({"x": rng.normal(0.5, 1, nt)})
>>> res = sp.transport_weights_fn(
...     source, target, features=["x"], treatment="a", outcome="y")
>>> type(res).__name__
'TransportWeightResult'
>>> bool(res.ess <= ns)
True
>>> bool(np.isfinite(res.effect_transported))
True

TransportIdentificationResult `dataclass` ¶

Bases: ResultProtocolMixin

Result of a Pearl-Bareinboim transportability identification check.

Attributes:

Name	Type	Description
`transportable`	`bool`	Whether `P(Y \| do(X))` in the target is identifiable from source distributions and target covariate margins.
`formula`	`str`	The transport formula (or `"NOT IDENTIFIABLE"`).
`admissible_set`	`frozenset`	The s-admissible conditioning set `Z` found (empty if the effect transports directly).
`reason`	`str`	Human-readable justification.

Examples:

>>> import statspai as sp
>>> g = sp.dag("X -> Y; W -> Y; W -> X; S -> W")
>>> res = sp.identify_transport(
...     g, treatment="X", outcome="Y", selection_nodes="S")
>>> type(res).__name__
'TransportIdentificationResult'
>>> res.transportable
True
>>> sorted(res.admissible_set)
['W']

EvidenceSynthesisResult `dataclass` ¶

Bases: ResultProtocolMixin

Output of :func:synthesise_evidence.

HeterogeneityResult `dataclass` ¶

Bases: ResultProtocolMixin

Effect-heterogeneity diagnostic.

ConcordanceResult `dataclass` ¶

Bases: ResultProtocolMixin

RCT-vs-RWD concordance report.

weights ¶

weights(source: DataFrame, target: DataFrame, features: Sequence[str], treatment: str, outcome: str, truncate: tuple[float, float] | None = (0.01, 0.99)) -> TransportWeightResult

Compute transport weights and a transported ATE.

Parameters:

Name	Type	Description	Default
`source`	`DataFrame`	Source (must contain `features`, `treatment`, `outcome`) and target (must contain `features`).	required
`target`	`DataFrame`	Source (must contain `features`, `treatment`, `outcome`) and target (must contain `features`).	required
`features`	`list[str]`	Baseline covariates shared by both populations that identify target-conditional exchangeability.	required
`treatment`	`str`	Column names in `source`.	required
`outcome`	`str`	Column names in `source`.	required
`truncate`	`tuple[float, float] \| None`	Weight quantiles for truncation.	`(0.01, 0.99)`

Returns:

Type	Description
`TransportWeightResult`

Examples:

Transport an effect from a source sample to a target population whose covariate distribution is shifted:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> ns, nt = 500, 500
>>> source = pd.DataFrame({
...     "x": rng.normal(0, 1, ns),
...     "a": rng.integers(0, 2, ns),
... })
>>> source["y"] = source["a"] + 0.5 * source["x"] + rng.normal(0, 1, ns)
>>> target = pd.DataFrame({"x": rng.normal(0.5, 1, nt)})
>>> res = sp.transport_weights_fn(
...     source, target, features=["x"], treatment="a", outcome="y")
>>> bool(abs(res.effect_source - 1.0) < 0.4)  # true source ATE is 1.0
True
>>> bool(res.ess <= ns)
True

References

stuart2011propensity dahabreh2020extending

identify_transport ¶

identify_transport(dag: Any, treatment: str | Iterable[str], outcome: str | Iterable[str], selection_nodes: str | Iterable[str]) -> TransportIdentificationResult

Test s-admissibility of some conditioning set.

Parameters:

Name	Type	Description	Default
`dag`	`DAG`	Causal graph.	required
`treatment`	`str \| Iterable[str]`	Intervention set X.	required
`outcome`	`str \| Iterable[str]`	Outcome set Y.	required
`selection_nodes`	`Iterable[str]`	Nodes (`S1`, `S2`, ...) pointing to variables whose distributions differ across source and target populations.	required

Returns:

Type	Description
`TransportIdentificationResult`

Examples:

Conditioning on W blocks the S -> W -> Y selection path, so the effect transports:

>>> import statspai as sp
>>> g = sp.dag("X -> Y; W -> Y; W -> X; S -> W")
>>> res = sp.identify_transport(
...     g, treatment="X", outcome="Y", selection_nodes="S")
>>> res.transportable
True
>>> sorted(res.admissible_set)
['W']

synthesise_evidence ¶

synthesise_evidence(*, rct_estimate: float, rct_se: float, rwd_estimate: float, rwd_se: float, transport_shift: float = 0.0, transport_shift_se: float = 0.0, alpha: float = 0.05, weight_mode: str = 'inverse_variance') -> EvidenceSynthesisResult

Pool an RCT and an RWD estimate into a single evidence-synthesis.

Parameters:

Name	Type	Description	Default
`rct_estimate`	`float`	RCT effect estimate and its SE.	required
`rct_se`	`float`	RCT effect estimate and its SE.	required
`rwd_estimate`	`float`	RWD effect estimate and its SE (already transport-weighted if that is the design).	required
`rwd_se`	`float`	RWD effect estimate and its SE (already transport-weighted if that is the design).	required
`transport_shift`	`float`	Additive adjustment applied to the RCT estimate to transport it to the RWD target population (e.g. using density-ratio weights).	`0.0`
`transport_shift_se`	`float`	SE of the transport shift, added in quadrature to the RCT SE.	`0.0`
`alpha`	`float`		`0.05`
`weight_mode`	`('inverse_variance', 'rct_heavy')`		`'inverse_variance'`

Returns:

Type	Description
`EvidenceSynthesisResult`

Examples:

>>> import statspai as sp
>>> res = sp.synthesise_evidence(
...     rct_estimate=0.50, rct_se=0.20,
...     rwd_estimate=0.42, rwd_se=0.10,
...     transport_shift=-0.05, transport_shift_se=0.02,
... )
>>> round(res.pooled_estimate, 3)
0.426
>>> round(res.weights["rwd"], 3)  # RWD is more precise, so weighted higher
0.802

References

Dahabreh et al. (2020); Yang, Gamalo & Fu (arXiv:2511.19735, 2025). [@yang2025integrating]

heterogeneity_of_effect ¶

heterogeneity_of_effect(estimates: Sequence[float], ses: Sequence[float]) -> HeterogeneityResult

Effect-heterogeneity diagnostic (DerSimonian-Laird).

Parameters:

Name	Type	Description	Default
`estimates`	`sequence of float`		required
`ses`	`sequence of float`		required

Returns:

Type	Description
`HeterogeneityResult`

Examples:

>>> import statspai as sp
>>> het = sp.heterogeneity_of_effect(
...     estimates=[0.50, 0.42, 0.55], ses=[0.20, 0.10, 0.15],
... )
>>> round(het.i2, 3)  # no excess heterogeneity across the three studies
0.0
>>> round(het.q_stat, 4)
0.5541

rwd_rct_concordance ¶

rwd_rct_concordance(*, rct_estimate: float, rct_se: float, rwd_estimate: float, alpha: float = 0.05) -> ConcordanceResult

Report-card metric for whether an RWD estimate agrees with the RCT.

Parameters:

Name	Type	Default
`rct_estimate`	`float`	required
`rct_se`	`float`	required
`rwd_estimate`	`float`	required
`alpha`	`float`	`0.05`

Returns:

Type	Description
`ConcordanceResult`

Examples:

>>> import statspai as sp
>>> con = sp.rwd_rct_concordance(
...     rct_estimate=0.50, rct_se=0.20, rwd_estimate=0.42,
... )
>>> bool(con.rwd_inside_rct_ci)  # RWD point falls inside the RCT 95% CI
True
>>> round(con.zscore_difference, 1)
-0.4

statspai.transport¶

transport ¶

TransportWeightResult dataclass ¶

TransportIdentificationResult dataclass ¶

EvidenceSynthesisResult dataclass ¶

HeterogeneityResult dataclass ¶

ConcordanceResult dataclass ¶

weights ¶

identify_transport ¶

synthesise_evidence ¶

heterogeneity_of_effect ¶

rwd_rct_concordance ¶

`statspai.transport`¶

TransportWeightResult `dataclass` ¶

TransportIdentificationResult `dataclass` ¶

EvidenceSynthesisResult `dataclass` ¶

HeterogeneityResult `dataclass` ¶

ConcordanceResult `dataclass` ¶