`statspai.interference`¶

interference ¶

Interference and Spillover Effects.

Estimates direct and spillover treatment effects when SUTVA (Stable Unit Treatment Value Assumption) is violated, i.e., one unit's treatment affects another unit's outcome.

References

Hudgens, M. G. & Halloran, M. E. (2008). Toward Causal Inference with Interference. JASA, 103(482), 832-842. [@hudgens2008toward]

Aronow, P. M. & Samii, C. (2017). Estimating Average Causal Effects Under General Interference. Annals of Applied Statistics, 11(4), 1912-1947. [@aronow2017estimating]

SpilloverEstimator ¶

Spillover / interference effects estimator.

The engine behind :func:sp.spillover. Construct it with the data and column names, then call :meth:fit to obtain a CausalResult whose model_info carries direct_effect, spillover_effect and total_effect.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> rows = []
>>> for c in range(40):
...     treated = rng.integers(0, 2, size=5)
...     frac = treated.mean()
...     for i in range(5):
...         y = 1.0 + 0.8 * treated[i] + 0.5 * frac + rng.normal()
...         rows.append({"hh": c, "vaccinated": int(treated[i]),
...                      "outcome": y})
>>> df = pd.DataFrame(rows)
>>> est = sp.SpilloverEstimator(
...     data=df, y="outcome", treat="vaccinated", cluster="hh",
...     n_bootstrap=50, random_state=0,
... )
>>> res = est.fit()
>>> sorted(k for k in res.model_info if k.endswith("_effect"))
['direct_effect', 'spillover_effect', 'total_effect']

fit ¶

fit() -> CausalResult

Estimate direct and spillover effects.

NetworkExposureResult `dataclass` ¶

Bases: ResultProtocolMixin

Container for :func:network_exposure Horvitz-Thompson estimates.

Attributes:

Name	Type	Description
`estimates`	`DataFrame`	One row per exposure level with HT mean, SE and CI.
`contrasts`	`DataFrame`	Pairwise contrasts (direct / spillover / composite) for the AS4 map.
`exposure_levels`	`list of str`	The exposure categories realised in the data.

Examples:

>>> import statspai as sp
>>> import numpy as np
>>> rng = np.random.default_rng(0)
>>> n = 60
>>> A = np.zeros((n, n), dtype=int)
>>> for i in range(n):
...     A[i, (i + 1) % n] = 1
...     A[i, (i - 1) % n] = 1
>>> Z = (rng.random(n) < 0.5).astype(int)
>>> Y = 1.0 + 2.0 * Z + 0.5 * (A @ Z) + rng.normal(size=n)
>>> res = sp.network_exposure(Y, Z, A, p_treat=0.5, n_sim=500, seed=0)
>>> isinstance(res, sp.NetworkExposureResult)
True
>>> res.exposure_levels
['c00', 'c01', 'c10', 'c11']

PeerEffectsResult `dataclass` ¶

Bases: ResultProtocolMixin

Container for :func:peer_effects linear-in-means estimates.

Attributes:

Name	Type	Description
`endogenous_peer`	`float`	The estimated endogenous peer effect `beta` (coefficient on `WY`).
`contextual_peer`	`dict`	Contextual peer effects (`gamma`) per covariate.
`direct`	`dict`	Own-covariate (`delta`) coefficients.
`coefficients`	`DataFrame`	Full coefficient table with `variable`, `coef` and `se`.

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 80
>>> W = np.zeros((n, n))
>>> for i in range(n - 1):
...     W[i, i + 1] = 1
...     W[i + 1, i] = 1
>>> x = rng.normal(size=n)
>>> df = pd.DataFrame({"y": 0.8 * x + rng.normal(size=n), "x": x})
>>> res = sp.peer_effects(df, y="y", covariates=["x"], W=W)
>>> isinstance(res, sp.PeerEffectsResult)
True
>>> isinstance(res.endogenous_peer, float)
True

NetworkHTEResult `dataclass` ¶

Bases: ResultProtocolMixin

Output of :func:network_hte.

Examples:

>>> import statspai as sp
>>> import numpy as np
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> d = (rng.uniform(size=n) < 0.5).astype(float)
>>> e = rng.uniform(size=n)
>>> x1 = rng.normal(size=n)
>>> y = 1.0 + 0.8 * d + 0.5 * e + 0.3 * x1 + rng.normal(scale=0.5, size=n)
>>> df = pd.DataFrame({"y": y, "d": d, "e": e, "x1": x1})
>>> res = sp.network_hte(
...     df, y="y", treatment="d", neighbor_exposure="e",
...     covariates=["x1"], n_folds=3, random_state=0,
... )
>>> bool(np.isfinite(res.direct_effect))
True

to_dict ¶

to_dict() -> dict

JSON-safe dict of every field (agent-native serialization).

InwardOutwardResult `dataclass` ¶

Bases: ResultProtocolMixin

Output of :func:inward_outward_spillover.

Examples:

>>> import statspai as sp
>>> import numpy as np
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> d = (rng.uniform(size=n) < 0.5).astype(float)
>>> e_in = rng.uniform(size=n)
>>> e_out = rng.uniform(size=n)
>>> y = 1.0 + 0.6 * d + 0.4 * e_in + 0.2 * e_out + rng.normal(scale=0.5, size=n)
>>> df = pd.DataFrame({"y": y, "d": d, "e_in": e_in, "e_out": e_out})
>>> res = sp.inward_outward_spillover(
...     df, y="y", treatment="d",
...     inward_exposure="e_in", outward_exposure="e_out",
... )
>>> bool(res.inward_se > 0)
True

to_dict ¶

to_dict() -> dict

JSON-safe dict of every field (agent-native serialization).

MatchedPairResult `dataclass` ¶

Bases: ResultProtocolMixin

Output of matched-pair cluster RCT estimation.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(7)
>>> rows = []
>>> for p in range(20):                       # 20 matched pairs
...     base = rng.normal(0, 1)               # shared pair-level baseline
...     for arm in (0, 1):                    # one control, one treated cluster
...         cid = p * 2 + arm
...         for _ in range(15):
...             y = base + 0.5 * arm + rng.normal(0, 0.3)
...             rows.append({"y": y, "cluster": cid, "treat": arm, "pair": p})
>>> df = pd.DataFrame(rows)
>>> res = sp.cluster_matched_pair(df, y="y", cluster="cluster",
...                               treat="treat", pair="pair")
>>> isinstance(res, sp.MatchedPairResult)
True
>>> res.n_pairs
20
>>> res.n_clusters
40

CrossClusterRCTResult `dataclass` ¶

Bases: ResultProtocolMixin

Output of cross-cluster RCT with interference correction.

Produced by :func:cluster_cross_interference. Holds the estimated direct_effect and spillover_effect (with standard errors), the number of clusters, and a formatted .summary().

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> rows = []
>>> for c in range(40):
...     d = int(rng.integers(0, 2))
...     nshare = float(rng.random())
...     base = 1.0 * d + 0.8 * nshare
...     for _ in range(int(rng.integers(5, 15))):
...         rows.append({"cluster": c, "treat": d,
...                      "nshare": nshare, "y": base + rng.normal()})
>>> df = pd.DataFrame(rows)
>>> res = sp.cluster_cross_interference(df, y="y", cluster="cluster",
...                                     treat="treat",
...                                     neighbour_treat_share="nshare")
>>> type(res).__name__
'CrossClusterRCTResult'
>>> res.n_clusters
40
>>> isinstance(res.summary(), str)
True

StaggeredClusterRCTResult `dataclass` ¶

Bases: ResultProtocolMixin

Staggered-rollout cluster RCT output.

Attributes:

Name	Type	Description
`overall_att`	`float`	Mean post-treatment dynamic ATT across event times.
`overall_se`	`float`	Cluster-bootstrap standard error of `overall_att`.
`event_study`	`DataFrame`	Per relative-time ATTs with `se`, `ci_low` and `ci_high`.
`n_clusters`	`int`	Number of clusters in the design.

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> rows = []
>>> for c in range(12):
...     ft = 4 if c < 4 else (6 if c < 8 else 0)  # 0 = never-treated
...     fe = rng.normal()
...     for t in range(8):
...         d = 1 if (ft > 0 and t >= ft) else 0
...         y = 1.0 + fe + 0.2 * t + 1.5 * d + rng.normal(scale=0.3)
...         rows.append({"cluster": c, "time": t, "first_treat": ft, "y": y})
>>> df = pd.DataFrame(rows)
>>> res = sp.cluster_staggered_rollout(
...     df, y="y", cluster="cluster", time="time", first_treat="first_treat")
>>> isinstance(res, sp.StaggeredClusterRCTResult)
True
>>> res.n_clusters
12

DNCGNNDiDResult `dataclass` ¶

Bases: ResultProtocolMixin

Output of DNC + GNN + DiD.

Examples:

>>> import statspai as sp
>>> import numpy as np
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> rows = []
>>> for uid in range(40):
...     treat_time = 3 if uid < 20 else 0
...     u = rng.normal()
...     for t in range(1, 6):
...         post = 1 if (treat_time > 0 and t >= treat_time) else 0
...         rows.append({
...             "id": uid, "time": t, "treat": treat_time,
...             "y": u + 0.5 * t + 1.2 * post + rng.normal(0, 0.3),
...             "nc_y": u + 0.4 * t + rng.normal(0, 0.3),
...             "nc_d": u + rng.normal(0, 0.3),
...         })
>>> df = pd.DataFrame(rows)
>>> res = sp.dnc_gnn_did(
...     df, y="y", treat="treat", time="time", id="id",
...     nc_outcome=["nc_y"], nc_exposure=["nc_d"], n_boot=50, seed=1)
>>> isinstance(res, sp.DNCGNNDiDResult)
True
>>> bool(res.se > 0 and res.ci[0] < res.estimate < res.ci[1])
True

network_hte ¶

network_hte(data: DataFrame, *, y: str, treatment: str, neighbor_exposure: str, covariates: Sequence[str], n_folds: int = 5, alpha: float = 0.05, random_state: int = 0) -> NetworkHTEResult

Orthogonal learning of direct + spillover effects on networks.

Estimates the partially-linear model

Y_i = alpha(X_i) + tau_d * D_i + tau_s * E_i + eps_i,
E[eps_i | X_i, D_i, E_i] = 0,

where D_i is unit i's own treatment and E_i is a scalar summary of neighbourhood exposure (e.g. share treated). Uses Chernozhukov-style double orthogonalisation: cross-fit nuisance models for E[Y|X], E[D|X] and E[E|X], then regress residualised Y on residualised (D, E).

Parameters:

Name	Type	Default
`data`	`DataFrame`	required
`y`	`str`	required
`treatment`	`str`	required
`neighbor_exposure`	`str`	required
`covariates`	`sequence of str`	required
`n_folds`	`int`	`5`
`alpha`	`float`	`0.05`
`random_state`	`int`	`0`

Returns:

Type	Description
`NetworkHTEResult`

Examples:

>>> import statspai as sp
>>> import numpy as np
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> d = (rng.uniform(size=n) < 0.5).astype(float)
>>> e = rng.uniform(size=n)  # neighbour exposure share
>>> x1 = rng.normal(size=n)
>>> x2 = rng.normal(size=n)
>>> y = 1.0 + 0.8 * d + 0.5 * e + 0.3 * x1 + rng.normal(scale=0.5, size=n)
>>> df = pd.DataFrame({"y": y, "d": d, "e": e, "x1": x1, "x2": x2})
>>> res = sp.network_hte(
...     df, y="y", treatment="d", neighbor_exposure="e",
...     covariates=["x1", "x2"], n_folds=3, random_state=0,
... )
>>> bool(res.direct_effect > res.spillover_effect)
True

References

Wu & Yuan (arXiv:2509.18484, 2025). [@wu2025estimating]

inward_outward_spillover ¶

inward_outward_spillover(data: DataFrame, *, y: str, treatment: str, inward_exposure: str, outward_exposure: str, covariates: Optional[Sequence[str]] = None, alpha: float = 0.05) -> InwardOutwardResult

Decompose spillover into 'inward' (incoming edges to unit i) and 'outward' (outgoing edges from unit i to neighbours).

The model is

Y_i = alpha + tau*D_i + tau_in * E_in_i + tau_out * E_out_i + X'beta + eps

where E_in_i and E_out_i are user-constructed inward / outward exposure summaries (e.g. in a directed network: share of incoming neighbours treated vs. share of outgoing).

Parameters:

Name	Type	Default
`data`	`DataFrame`	required
`y`	`str`	required
`treatment`	`str`	required
`inward_exposure`	`str`	required
`outward_exposure`	`str`	required
`covariates`	`sequence of str`	`None`
`alpha`	`float`	`0.05`

Returns:

Type	Description
`InwardOutwardResult`

Examples:

>>> import statspai as sp
>>> import numpy as np
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> d = (rng.uniform(size=n) < 0.5).astype(float)
>>> e_in = rng.uniform(size=n)
>>> e_out = rng.uniform(size=n)
>>> y = 1.0 + 0.6 * d + 0.4 * e_in + 0.2 * e_out + rng.normal(scale=0.5, size=n)
>>> df = pd.DataFrame({"y": y, "d": d, "e_in": e_in, "e_out": e_out})
>>> res = sp.inward_outward_spillover(
...     df, y="y", treatment="d",
...     inward_exposure="e_in", outward_exposure="e_out",
... )
>>> bool(res.inward_se > 0)
True

References

Fang, Airoldi & Forastiere (arXiv:2506.06615, 2025). [@fang2025inward]

cluster_cross_interference ¶

cluster_cross_interference(data: DataFrame, y: str, cluster: str, treat: str, neighbour_treat_share: str, alpha: float = 0.05) -> CrossClusterRCTResult

Cluster RCT with explicit interference adjustment.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`		required
`y`	`str`	Individual-level outcome.	required
`cluster`	`str`	Cluster identifier.	required
`treat`	`str`	Cluster-level binary treatment.	required
`neighbour_treat_share`	`str`	Share of treated neighbours per cluster (precomputed by user from spatial / network adjacency).	required
`alpha`	`float`		`0.05`

Returns:

Type	Description
`CrossClusterRCTResult`

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> rows = []
>>> for c in range(40):
...     d = int(rng.integers(0, 2))
...     nshare = float(rng.random())
...     base = 1.0 * d + 0.8 * nshare
...     for _ in range(int(rng.integers(5, 15))):
...         rows.append({"cluster": c, "treat": d,
...                      "nshare": nshare, "y": base + rng.normal()})
>>> df = pd.DataFrame(rows)
>>> res = sp.cluster_cross_interference(df, y="y", cluster="cluster",
...                                     treat="treat",
...                                     neighbour_treat_share="nshare")
>>> res.n_clusters
40
>>> isinstance(res.direct_effect, float)
True
>>> isinstance(res.summary(), str)
True

cluster_staggered_rollout ¶

cluster_staggered_rollout(data: DataFrame, y: str, cluster: str, time: str, first_treat: str, leads: int = 2, lags: int = 4, alpha: float = 0.05) -> StaggeredClusterRCTResult

Staggered-rollout cluster RCT estimator.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`	Panel: cluster × time × outcome.	required
`y`	`str`	first_treat = first calendar time the cluster is treated (0 / NaN for never-treated).	required
`cluster`	`str`	first_treat = first calendar time the cluster is treated (0 / NaN for never-treated).	required
`time`	`str`	first_treat = first calendar time the cluster is treated (0 / NaN for never-treated).	required
`first_treat`	`str`	first_treat = first calendar time the cluster is treated (0 / NaN for never-treated).	required
`leads`	`int`		`2`
`lags`	`int`		`2`
`alpha`	`float`		`0.05`

Returns:

Type	Description
`StaggeredClusterRCTResult`

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> rows = []
>>> for c in range(12):  # 4 cohorts at t=4, 4 at t=6, 4 never-treated
...     ft = 4 if c < 4 else (6 if c < 8 else 0)
...     fe = rng.normal()
...     for t in range(8):
...         d = 1 if (ft > 0 and t >= ft) else 0
...         y = 1.0 + fe + 0.2 * t + 1.5 * d + rng.normal(scale=0.3)
...         rows.append({"cluster": c, "time": t, "first_treat": ft, "y": y})
>>> df = pd.DataFrame(rows)
>>> res = sp.cluster_staggered_rollout(
...     df, y="y", cluster="cluster", time="time", first_treat="first_treat")
>>> res.event_study.columns.tolist()
['rel_time', 'att', 'se', 'ci_low', 'ci_high']
>>> res.n_clusters
12

interference ¶

interference(design: str = 'partial', /, **kwargs: Any) -> Any

Unified entry point for the interference / spillover family.

Parameters:

Name	Type	Description	Default
`design`	`str`	The interference design to estimate under. Supported values are listed by `sp.interference_available_designs()`.	``"partial"``
`**kwargs`	`Any`	Passed through unchanged to the target function.	`{}`

Returns:

Type	Description
`The underlying estimator's return object. See each underlying`
`function's docstring for details.`

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> cluster = np.repeat(np.arange(40), 8)
>>> treat = rng.integers(0, 2, size=cluster.size)
>>> frac = pd.Series(treat).groupby(cluster).transform("mean").to_numpy()
>>> y = 1.0 + 0.8 * treat + 0.5 * frac + rng.normal(size=cluster.size)
>>> df = pd.DataFrame({"y": y, "d": treat, "household": cluster})
>>> r = sp.interference("partial", data=df, y="y", treat="d",
...                     cluster="household", n_bootstrap=100)
>>> type(r).__name__
'CausalResult'

interference_available_designs ¶

interference_available_designs() -> list[str]

Return the full list of registered interference design names.

These are the strings accepted as the first positional argument of :func:statspai.interference.

Examples:

>>> import statspai as sp
>>> designs = sp.interference_available_designs()
>>> bool("partial" in designs and "network_exposure" in designs)
True
>>> designs == sorted(designs)  # returned sorted alphabetically
True

statspai.interference¶

interference ¶

SpilloverEstimator ¶

fit ¶

NetworkExposureResult dataclass ¶

PeerEffectsResult dataclass ¶

NetworkHTEResult dataclass ¶

to_dict ¶

InwardOutwardResult dataclass ¶

to_dict ¶

MatchedPairResult dataclass ¶

CrossClusterRCTResult dataclass ¶

StaggeredClusterRCTResult dataclass ¶

DNCGNNDiDResult dataclass ¶

network_hte ¶

inward_outward_spillover ¶

cluster_cross_interference ¶

cluster_staggered_rollout ¶

interference ¶

interference_available_designs ¶

`statspai.interference`¶

NetworkExposureResult `dataclass` ¶

PeerEffectsResult `dataclass` ¶

NetworkHTEResult `dataclass` ¶

InwardOutwardResult `dataclass` ¶

MatchedPairResult `dataclass` ¶

CrossClusterRCTResult `dataclass` ¶

StaggeredClusterRCTResult `dataclass` ¶

DNCGNNDiDResult `dataclass` ¶