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

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

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

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

NetworkExposureResult `dataclass` ¶

Bases: ResultProtocolMixin

Container for :func:network_exposure estimates.

Attributes:

Name	Type	Description
`estimates`	`DataFrame`	One row per exposure level: mean outcome, standard error, confidence interval, number of units realised at that level, smallest exposure probability and effective sample size.
`contrasts`	`DataFrame`	Differences between exposure levels with standard errors that account for the dependence between the two estimated means.
`exposure_levels`	`list of str`	The exposure categories of the mapping.
`n_obs`	`int`	Number of units supplied.
`n_eligible`	`int`	Number of units for which every exposure level has positive probability; all averages are over these units.

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)
>>> isinstance(res, sp.NetworkExposureResult)
True
>>> res.exposure_levels
['c00', 'c01', 'c10', 'c11']

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

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

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

InterferenceTestResult `dataclass` ¶

Bases: ResultProtocolMixin

Result of :func:interference_test.

Attributes:

Name	Type	Description
`statistic`	`float`	Test statistic on the realised assignment.
`pvalue`	`float`	Permutation p-value `(1 + #{T_b at least as extreme}) / (1 + B)`, or the exact proportion when every assignment was enumerated.
`null`	`str`	The hypothesis tested.
`n_focal`	`int`	Number of focal units (all units for `null="no_effect"`).
`n_perm`	`int`	Number of alternative assignments evaluated.
`exact`	`bool`	Whether all admissible assignments were enumerated.

Examples:

>>> import numpy as np
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> Z = np.array([1, 0] * 6)
>>> Y = rng.normal(size=12) + 2.0 * Z
>>> r = sp.interference_test(Y, Z, null="no_effect")
>>> isinstance(r, sp.InterferenceTestResult), r.exact
(True, True)

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.

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). Event time is counted in observed periods, so unevenly spaced times (2000, 2002, ...) work; a cohort treated at a time that is not observed starts at the next observed period.	required
`cluster`	`str`	first_treat = first calendar time the cluster is treated (0 / NaN for never-treated). Event time is counted in observed periods, so unevenly spaced times (2000, 2002, ...) work; a cohort treated at a time that is not observed starts at the next observed period.	required
`time`	`str`	first_treat = first calendar time the cluster is treated (0 / NaN for never-treated). Event time is counted in observed periods, so unevenly spaced times (2000, 2002, ...) work; a cohort treated at a time that is not observed starts at the next observed period.	required
`first_treat`	`str`	first_treat = first calendar time the cluster is treated (0 / NaN for never-treated). Event time is counted in observed periods, so unevenly spaced times (2000, 2002, ...) work; a cohort treated at a time that is not observed starts at the next observed period.	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_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("spillover" in designs and "network_exposure" in designs)
True
>>> designs == sorted(designs)  # returned sorted alphabetically
True

interference ¶

interference(design: str = 'spillover', /, **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"` and `"hudgens_halloran"` route to :func:`sp.two_stage_interference` (the Hudgens-Halloran two-stage estimator, which needs a `group` and a `saturation` column), the same as `"two_stage"`. .. deprecated:: 1.42 `"parmigiani"` (an alias of `"network_hte"`) warns and will be removed in 1.43. .. versionchanged:: 1.41 Through 1.40 these two names routed to :func:`sp.spillover`, which is not the Hudgens-Halloran estimator. A call that passes :func:`sp.spillover`'s arguments (`cluster=`, `n_bootstrap=` ...) under either name raises `MethodIncompatibility`; use `"spillover"` for that estimator.	``"spillover"``
`**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("spillover", data=df, y="y", treat="d",
...                     cluster="household", n_bootstrap=100)
>>> type(r).__name__
'CausalResult'

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]

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). Standard errors are the heteroskedasticity-robust sandwich (HC0) of that regression.

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]

interference_test ¶

interference_test(Y: Sequence[float], Z: Sequence[int], adjacency: Any = None, *, null: str = 'no_spillover', statistic: Optional[Statistic] = None, focal: Union[None, float, Sequence[int], ndarray] = None, alternative: str = 'two-sided', n_perm: int = 2000, seed: Optional[int] = 0) -> InterferenceTestResult

Randomization test for treatment effects or spillovers on a network.

Parameters:

Name	Type	Description	Default
`Y`	`array - like(n)`	Observed outcomes.	required
`Z`	`array-like (n,) of {0,1}`	Realised treatment assignment, drawn independently across units or by fixing the number treated.	required
`adjacency`	`ndarray, sparse matrix, DataFrame, or list of edges`	Row `i` marks the units whose treatment may affect unit `i`. Required for every hypothesis except `"no_effect"`.	`None`
`null`	`('no_spillover', 'no_effect', 'anonymous', 'no_higher_order')`	The hypothesis tested, from most to least restrictive: `"no_effect"`: treatment affects no outcome at all. `"no_spillover"` (default): an outcome depends only on the unit's own treatment. `"anonymous"`: an outcome depends only on the unit's own treatment and on the share of its neighbours that are treated, not on which ones. `"no_higher_order"`: an outcome depends only on the treatments of the unit and of its neighbours, not on units further away.	`"no_spillover"`
`statistic`	`('neighbor_share', 'any_neighbor', 'difference_in_means')`	`"neighbor_share"` (default for `"no_spillover"`) is the coefficient on the share of treated neighbours in a regression of focal outcomes on own treatment and that share; `"any_neighbor"` uses an indicator for at least one treated neighbour instead. `"difference_in_means"` (default for `"no_effect"`) is the treated-minus-control mean. A callable `T(Y, Z, A, focal)` returning a number is also accepted; it must use the outcomes of the focal units only. For `"anonymous"` the default is `"weighted_share"`, the coefficient on the share of treated neighbours weighted by each neighbour's own number of neighbours, given own treatment and the unweighted share. For `"no_higher_order"` it is `"second_order_share"`, the coefficient on the share treated among neighbours of neighbours, given own treatment and the share of treated neighbours.	`"neighbor_share"`
`focal`	`float or array - like`	The share of units drawn at random (from `seed`, never from `Z`) as focal units, or the indices of the focal units (or a boolean mask with one entry per unit), which must have been chosen without looking at the realised assignment. The default share is 0.5 for `"no_spillover"` and 0.15 for `"no_higher_order"`. For `"anonymous"` focal units are picked in random order so that no two share a neighbour or are neighbours themselves, up to that share (default: as many as fit).	`None`
`alternative`	`('two-sided', 'greater', 'less')`		`"two-sided"`
`n_perm`	`int`	Number of alternative assignments. When the number of admissible assignments does not exceed `n_perm` they are all enumerated and the p-value is exact.	`2000`
`seed`	`int`	Seed for the focal draw and the permutations.	`0`

Returns:

Type	Description
`InterferenceTestResult`

Notes

Alternative assignments permute the treatments of the units the hypothesis leaves free: all units for "no_effect"; the non-focal units for "no_spillover"; the units that are neither focal nor neighbours of a focal unit for "no_higher_order"; and, for "anonymous", the neighbours of each focal unit among themselves, which keeps every focal unit's share of treated neighbours fixed. Each permutation keeps the number treated fixed. Under independent assignment the conditional distribution given that number is uniform over such permutations, so the test is valid for both Bernoulli and completely randomized designs. It is not valid for designs with unequal assignment probabilities or with clustering.

The hypotheses are nested, so testing them from the most restrictive down and stopping at the first that is not rejected needs no multiplicity correction.

A larger focal set makes the statistic more precise but leaves fewer treatments to permute; the defaults are a compromise. The tests of "anonymous" and "no_higher_order" hold fixed far more of the assignment than the test of "no_spillover" and have correspondingly less power: failing to reject them is weak evidence.

References

[@aronow2012general], [@athey2018exact], [@basse2019randomization]

Examples:

>>> import numpy as np
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> A = np.zeros((n, n), dtype=int)
>>> for i in range(n):
...     A[i, (i + 1) % n] = A[(i + 1) % n, i] = 1
>>> Z = (rng.random(n) < 0.5).astype(int)
>>> Y = 1.0 * Z + 2.0 * (A @ Z > 0) + rng.normal(size=n)
>>> r = sp.interference_test(Y, Z, A, n_perm=999, seed=1)
>>> bool(r.pvalue < 0.05)
True

two_stage_interference ¶

two_stage_interference(data: DataFrame, y: str, treat: str, group: str, saturation: str, reference: Optional[float] = None, weighting: str = 'group', alpha: float = 0.05) -> CausalResult

Direct, indirect, total and overall effects of a two-stage randomized (saturation) design under partial interference.

Groups are randomized to saturations, then units within each group are randomized with that share treated (complete randomization at both stages). Units may affect others in their own group but not in other groups.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`	One row per unit.	required
`y`	`str`	Outcome.	required
`treat`	`str`	Unit-level treatment, 0/1.	required
`group`	`str`	Group (cluster) identifier: the unit of the first-stage randomization and the boundary of interference.	required
`saturation`	`str`	The saturation each group was randomized to, the share of its units to be treated, constant within the group (0 for pure-control groups).	required
`reference`	`float`	The saturation the cross-saturation effects are measured against. Default: the lowest saturation in the data.	`None`
`weighting`	`('group', 'individual')`	`'group'` averages over groups (Hudgens and Halloran's estimands); `'individual'` over units, through the outcomes rescaled by `n_i / nbar`.	`'group'`
`alpha`	`float`	Level of the normal confidence intervals.	`0.05`

Returns:

Type Description

CausalResult

Headline: the total effect of the highest saturation, Ybar(1; high) - Ybar(0; reference). detail has one row per effect: direct at each saturation strictly between 0 and 1 (Ybar(1; a) - Ybar(0; a)), and for each other saturation a indirect (Ybar(0; a) - Ybar(0; reference)), total (Ybar(1; a) - Ybar(0; reference)) and overall (Ybar(a) - Ybar(reference)), with the conservative standard error, normal interval and p-value. model_info['means'] has each Ybar(z; a) and Ybar(a) with its standard error; model_info['count_mismatch'] lists groups whose treated count is one or more units away from saturation * size.

Notes

Signs are treated minus untreated and higher minus reference saturation; Hudgens and Halloran (2008) write each effect the other way round (untreated minus treated, low minus high coverage). The identity direct(a) + indirect(a) = total(a) holds exactly for the estimates.

Each standard error is the sample standard deviation across groups of the group-level estimate over the square root of the number of groups, with one such term per saturation in a cross-saturation contrast. It overstates the randomization variance by the between-group variance of the group-level effect divided by the number of groups, and is exact when the effect is constant across groups (Basse and Feller 2018, Theorem 5.1). The intervals rely on the number of groups being large.

Unbiasedness requires that the second stage fixes the number treated in each group; with Bernoulli assignment inside groups the group means are ratio estimators. A count far from saturation * size is reported (and warned about), not corrected.

Examples:

>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> rows = []
>>> for g, a in enumerate(np.repeat([0.0, 0.25, 0.75], 20)):
...     d = np.zeros(8, int); d[rng.choice(8, int(a * 8), replace=False)] = 1
...     y = 1 + 2 * d + 1.5 * a + rng.normal(size=8)
...     rows.append(pd.DataFrame({"g": g, "a": a, "d": d, "y": y}))
>>> df = pd.concat(rows, ignore_index=True)
>>> r = sp.two_stage_interference(df, "y", "d", "g", "a")
>>> sorted(set(r.detail["effect"]))
['direct', 'indirect', 'overall', 'total']

References

[@hudgens2008toward], [@basse2018analyzing]

statspai.interference¶

interference ¶

CrossClusterRCTResult dataclass ¶

MatchedPairResult dataclass ¶

StaggeredClusterRCTResult dataclass ¶

DNCGNNDiDResult dataclass ¶

NetworkExposureResult dataclass ¶

InwardOutwardResult dataclass ¶

to_dict ¶

NetworkHTEResult dataclass ¶

to_dict ¶

PeerEffectsResult dataclass ¶

InterferenceTestResult dataclass ¶

SpilloverEstimator ¶

fit ¶

cluster_cross_interference ¶

cluster_staggered_rollout ¶

interference_available_designs ¶

interference ¶

inward_outward_spillover ¶

network_hte ¶

interference_test ¶

two_stage_interference ¶

`statspai.interference`¶

CrossClusterRCTResult `dataclass` ¶

MatchedPairResult `dataclass` ¶

StaggeredClusterRCTResult `dataclass` ¶

DNCGNNDiDResult `dataclass` ¶

NetworkExposureResult `dataclass` ¶

InwardOutwardResult `dataclass` ¶

NetworkHTEResult `dataclass` ¶

PeerEffectsResult `dataclass` ¶

InterferenceTestResult `dataclass` ¶