`statspai.proximal`¶

proximal ¶

Proximal Causal Inference (Tchetgen Tchetgen et al. 2020).

Identifies the ATE in the presence of an unmeasured confounder :math:U, using two proxies of :math:U:

:math:Z — "treatment-inducing confounding proxy" (independent of Y | D, U)
:math:W — "outcome-inducing confounding proxy" (independent of D | U)

plus measured covariates :math:X.

ProximalCausalInference ¶

Class wrapper for :func:proximal.

Construct with the same keyword arguments as :func:proximal (minus data), then call :meth:fit with a DataFrame. The fitted :class:~statspai.core.results.CausalResult is stored on result_.

Examples:

>>> import numpy as np
>>> import pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 300
>>> u = rng.normal(size=n)
>>> smoker = (rng.normal(0.6 * u, 1, n) > 0).astype(float)
>>> df = pd.DataFrame({
...     "lung_cancer": 0.4 * smoker + 0.8 * u + rng.normal(0, 1, n),
...     "smoker": smoker,
...     "occupation": u + rng.normal(0, 1, n),
...     "shs_exposure": u + rng.normal(0, 1, n),
... })
>>> model = sp.ProximalCausalInference(
...     y="lung_cancer", treat="smoker",
...     proxy_z=["occupation"], proxy_w=["shs_exposure"],
... ).fit(df)
>>> model.result_.estimand
'ATE'

NegativeControlResult `dataclass` ¶

Bases: ResultProtocolMixin

Unified result for negative-control procedures.

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> U = rng.normal(size=n)
>>> D = (0.5 * U + rng.normal(size=n) > 0).astype(int)
>>> df = pd.DataFrame({"D": D, "NCO": 0.8 * U + rng.normal(size=n)})
>>> res = sp.negative_control_outcome(df, nco="NCO", treat="D")
>>> est = float(res.estimate)

to_dict ¶

to_dict() -> dict

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

ProximalRegResult `dataclass` ¶

Bases: ResultProtocolMixin

Result of the regression-based proximal causal inference estimator.

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 300
>>> U = rng.normal(size=n)
>>> D = (0.5 * U + rng.normal(size=n) > 0).astype(int)
>>> df = pd.DataFrame({
...     "Y": 1.5 * D + 1.0 * U + rng.normal(size=n),
...     "D": D,
...     "Z": 0.7 * U + rng.normal(size=n) * 0.5,
...     "W": 0.7 * U + rng.normal(size=n) * 0.5,
... })
>>> res = sp.proximal_regression(df, y="Y", treat="D", z_proxy="Z",
...                              w_proxy="W")
>>> ate = float(res.ate)

to_dict ¶

to_dict() -> dict

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

ProxyScoreResult `dataclass` ¶

Bases: ResultProtocolMixin

Per-candidate proxy score for PCI.

Returned by :func:select_pci_proxies. Holds the ranked Z- and W-side candidate tables plus the recommended top-k proxy names for each role.

Examples:

>>> import statspai as sp
>>> import numpy as np
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 300
>>> u = rng.normal(size=n)
>>> z = u + rng.normal(scale=0.5, size=n)
>>> w = u + rng.normal(scale=0.5, size=n)
>>> d = (z + rng.normal(size=n) > 0).astype(float)
>>> y = 1.0 + 0.5 * d + w + rng.normal(size=n)
>>> df = pd.DataFrame({"y": y, "d": d, "z": z, "w": w})
>>> res = sp.select_pci_proxies(df, y="y", treat="d",
...                             candidates=["z", "w"])
>>> isinstance(res, sp.ProxyScoreResult)
True
>>> isinstance(res.summary(), str)
True

proximal ¶

proximal(data: DataFrame, y: str, treat: str, proxy_z: List[str], proxy_w: List[str], covariates: Optional[List[str]] = None, bridge: str = 'linear', n_boot: int = 0, alpha: float = 0.05, seed: Optional[int] = None) -> CausalResult

Proximal causal inference via linear 2SLS on the outcome bridge.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`		required
`y`	`str`	Outcome variable.	required
`treat`	`str`	Treatment variable (binary or continuous).	required
`proxy_z`	`list of str`	Treatment-inducing confounding proxy variable(s) (Z). These serve as instruments for the outcome proxy W.	required
`proxy_w`	`list of str`	Outcome-inducing confounding proxy variable(s) (W). Endogenous regressors in the linear bridge.	required
`covariates`	`list of str`	Measured baseline covariates X (exogenous controls).	`None`
`bridge`	`linear`	Functional form of the outcome-confounding bridge. Only `'linear'` is currently implemented — the linear 2SLS estimator described in Cui et al. (2024, §4). Kernel-based bridges (Mastouri et al. 2021) and sieve/RKHS non-parametric bridges (Deaner 2018) are planned for a future release and will be accepted values of this argument. Passing any other string raises `NotImplementedError` — we prefer to fail loudly now over silently falling back to the linear bridge and mis-attributing results.	`'linear'`
`n_boot`	`int`	If > 0, nonparametric bootstrap SE (rows, not cluster-robust). If 0, use closed-form 2SLS sandwich SE (homoskedastic).	`0`
`alpha`	`float`		`0.05`
`seed`	`int`		`None`

Returns:

Type	Description
`CausalResult`	`estimate` is the coefficient on the treatment in the linear bridge — the proximal ATE under correct specification.

Examples:

ATE of smoking on lung cancer, with occupation (Z) and secondhand-smoke exposure (W) as proxies for an unmeasured confounder u (health behaviour / genetics):

>>> import numpy as np
>>> import pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 300
>>> u = rng.normal(size=n)                       # unmeasured confounder
>>> smoker = (rng.normal(0.6 * u, 1, n) > 0).astype(float)
>>> occupation = u + rng.normal(0, 1, n)         # proxy Z
>>> shs_exposure = u + rng.normal(0, 1, n)       # proxy W
>>> age = rng.normal(50, 8, n)
>>> lung_cancer = 0.4 * smoker + 0.8 * u + 0.01 * age + rng.normal(0, 1, n)
>>> df = pd.DataFrame({
...     "lung_cancer": lung_cancer, "smoker": smoker,
...     "occupation": occupation, "shs_exposure": shs_exposure, "age": age,
... })
>>> res = sp.proximal(df, y="lung_cancer", treat="smoker",
...                   proxy_z=["occupation"], proxy_w=["shs_exposure"],
...                   covariates=["age"])
>>> res.estimand
'ATE'

negative_control_outcome ¶

negative_control_outcome(data: DataFrame, nco: str, treat: str, covariates: Optional[Sequence[str]] = None, alpha: float = 0.05) -> NegativeControlResult

Lipsitch-style NCO calibration.

Fit an OLS of the negative-control outcome nco on treat and optional covariates. A coefficient significantly different from zero signals residual confounding that the measured covariates failed to control for.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`		required
`nco`	`str`	Negative-control outcome — a variable plausibly unaffected by the true treatment but sharing confounders with the real Y.	required
`treat`	`str`	Treatment indicator or exposure variable.	required
`covariates`	`sequence of str`	Measured confounders to condition on.	`None`
`alpha`	`float`		`0.05`

Returns:

Type	Description
`NegativeControlResult`

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> U = rng.normal(size=n)
>>> D = (0.5 * U + rng.normal(size=n) > 0).astype(int)
>>> df = pd.DataFrame({
...     "D": D,
...     "X": rng.normal(size=n),
...     "NCO": 0.8 * U + rng.normal(size=n),
... })
>>> res = sp.negative_control_outcome(df, nco="NCO", treat="D",
...                                   covariates=["X"])
>>> est = float(res.estimate)

negative_control_exposure ¶

negative_control_exposure(data: DataFrame, y: str, nce: str, covariates: Optional[Sequence[str]] = None, alpha: float = 0.05) -> NegativeControlResult

Regress outcome on a negative-control exposure.

A significant coefficient on nce — which by design is assumed to not causally affect y — indicates residual confounding along the exposure axis (selection, measurement error, etc.).

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 200
>>> U = rng.normal(size=n)
>>> df = pd.DataFrame({
...     "Y": 1.0 * U + rng.normal(size=n),
...     "NCE": 0.8 * U + rng.normal(size=n),
... })
>>> res = sp.negative_control_exposure(df, y="Y", nce="NCE")
>>> est = float(res.estimate)

double_negative_control ¶

double_negative_control(data: DataFrame, y: str, treat: str, nce: str, nco: str, covariates: Optional[Sequence[str]] = None, alpha: float = 0.05) -> NegativeControlResult

Double negative control estimator (Miao et al. 2018; Shi et al. 2020).

Under the linear / index model::

Y      = α0 + α_D D + α_U U + α_X X + ε_Y
NCO    = β0 + β_U U + β_X X + ε_W
E[U | NCE, X, D] linear in (NCE, X, D)

(plus standard independence/exclusion conditions), the ATE is point-identified by IV-regressing Y on (D, NCO, X) using (D, NCE, X) as instruments: NCE instruments for the proxy NCO, breaking the dependence on U. The coefficient on D is the de-biased ATE.

This is implemented as a just-identified 2SLS. The fitted ATE is asymptotically unbiased under the assumptions above and consistent with Shi et al. (2020, §3) closed-form.

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 300
>>> U = rng.normal(size=n)
>>> D = (0.5 * U + rng.normal(size=n) > 0).astype(int)
>>> df = pd.DataFrame({
...     "Y": 1.5 * D + 1.0 * U + rng.normal(size=n),
...     "D": D,
...     "NCE": 0.8 * U + rng.normal(size=n) * 0.5,
...     "NCO": 0.8 * U + rng.normal(size=n) * 0.5,
... })
>>> res = sp.double_negative_control(df, y="Y", treat="D", nce="NCE",
...                                  nco="NCO")
>>> ate = float(res.estimate)

proximal_regression ¶

proximal_regression(data: DataFrame, y: str, treat: str, z_proxy: str, w_proxy: str, covariates: Optional[Sequence[str]] = None, alpha: float = 0.05, propensity_bounds: tuple = (0.02, 0.98)) -> ProximalRegResult

Doubly-robust regression-based PCI estimator for the ATE.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`		required
`y`	`str`	Outcome column.	required
`treat`	`str`	Binary treatment column.	required
`z_proxy`	`str`	Treatment-inducing confounding proxy Z.	required
`w_proxy`	`str`	Outcome-inducing confounding proxy W.	required
`covariates`	`sequence of str`	Measured covariates X.	`None`
`alpha`	`float`		`0.05`
`propensity_bounds`	`(float, float)`		`(0.02, 0.98)`

Returns:

Type	Description
`ProximalRegResult`

Examples:

>>> import statspai as sp
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 300
>>> U = rng.normal(size=n)
>>> D = (0.5 * U + rng.normal(size=n) > 0).astype(int)
>>> df = pd.DataFrame({
...     "Y": 1.5 * D + 1.0 * U + rng.normal(size=n),
...     "D": D,
...     "X": rng.normal(size=n),
...     "Z": 0.7 * U + rng.normal(size=n) * 0.5,
...     "W": 0.7 * U + rng.normal(size=n) * 0.5,
... })
>>> res = sp.proximal_regression(df, y="Y", treat="D", z_proxy="Z",
...                              w_proxy="W", covariates=["X"])
>>> ate = float(res.ate)

Notes

If the treatment-bridge logistic regression fails, the propensity is set to the constant marginal :math:P(D=1) — which neutralises the doubly-robust correction term — and a :class:~statspai.exceptions.ConvergenceWarning is emitted with detail['propensity_fallback'] set to True.

fortified_pci ¶

fortified_pci(data: DataFrame, y: str, treat: str, proxy_z: List[str], proxy_w: List[str], covariates: Optional[List[str]] = None, alpha: float = 0.05, n_boot: int = 200, seed: int = 0) -> CausalResult

Fortified Proximal Causal Inference (doubly-robust PCI).

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`		required
`y`	`str`		required
`treat`	`str`		required
`proxy_z`	`list of str`	Treatment-side proxies (instruments for W).	required
`proxy_w`	`list of str`	Outcome-side proxies (endogenous bridge regressors).	required
`covariates`	`list of str`		`None`
`alpha`	`float`		`0.05`
`n_boot`	`int`	Bootstrap reps for SE.	`200`
`seed`	`int`		`0`

Returns:

Type	Description
`CausalResult`	ATE estimate that is doubly robust to bridge / outcome misspecification.

Notes

If the outcome-regression augmentation fails on the point-estimate sample, the fortified estimator degrades to the plain bridge 2SLS estimate (tau_outcome is set to tau_bridge); a :class:~statspai.exceptions.ConvergenceWarning is emitted and model_info['outcome_augmentation_fallback'] is set to True together with the error type.

Examples:

ATE of smoking on lung cancer, with occupation (Z) and secondhand-smoke exposure (W) as proxies for an unmeasured confounder u:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 300
>>> u = rng.normal(size=n)                       # unmeasured confounder
>>> smoker = (rng.normal(0.6 * u, 1, n) > 0).astype(float)
>>> occupation = u + rng.normal(0, 1, n)         # treatment-side proxy Z
>>> shs_exposure = u + rng.normal(0, 1, n)       # outcome-side proxy W
>>> age = rng.normal(50, 8, n)
>>> lung_cancer = 0.4 * smoker + 0.8 * u + 0.01 * age + rng.normal(0, 1, n)
>>> df = pd.DataFrame({
...     "lung_cancer": lung_cancer, "smoker": smoker,
...     "occupation": occupation, "shs_exposure": shs_exposure, "age": age,
... })
>>> res = sp.fortified_pci(df, y="lung_cancer", treat="smoker",
...                        proxy_z=["occupation"], proxy_w=["shs_exposure"],
...                        covariates=["age"], n_boot=50, seed=0)
>>> res.estimand
'ATE'
>>> bool(np.isfinite(res.estimate))
True

References

Yu, Shi & Tchetgen Tchetgen (2025). Fortified Proximal Causal Inference with Many Invalid Proxies. arXiv 2506.13152. [@yu2025fortified]

bidirectional_pci ¶

bidirectional_pci(data: DataFrame, y: str, treat: str, proxy_z: List[str], proxy_w: List[str], covariates: Optional[List[str]] = None, alpha: float = 0.05, n_boot: int = 200, seed: int = 0) -> CausalResult

Bidirectional PCI: simultaneous outcome + treatment bridge.

Parameters:

Name	Type	Description	Default
`data`	`same as`	:func:`statspai.proximal.proximal`.	required
`y`	`same as`	:func:`statspai.proximal.proximal`.	required
`treat`	`same as`	:func:`statspai.proximal.proximal`.	required
`proxy_z`	`same as`	:func:`statspai.proximal.proximal`.	required
`proxy_w`	`same as`	:func:`statspai.proximal.proximal`.	required
`covariates`	`same as`	:func:`statspai.proximal.proximal`.	required
`alpha`	`float`		`0.05`
`n_boot`	`int`		`200`
`seed`	`int`		`0`

Returns:

Type	Description
`CausalResult`	ATE estimate from the bidirectional moment condition.

Notes

If the treatment-bridge (Z-based logistic IPW) step fails on the point-estimate sample, the estimator degrades to the outcome bridge only (tau_treatment is set to tau_outcome); a :class:~statspai.exceptions.ConvergenceWarning is emitted and model_info['treatment_bridge_fallback'] is set to True together with the error type.

References

[@min2025regression]

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(42)
>>> n = 400
>>> U = rng.standard_normal(n)            # unobserved confounder
>>> Z = U + 0.5 * rng.standard_normal(n)  # treatment-side proxy
>>> W = U + 0.5 * rng.standard_normal(n)  # outcome-side proxy
>>> X = rng.standard_normal(n)            # observed covariate
>>> D = (1.0 * Z + 0.5 * X + rng.standard_normal(n) > 0).astype(int)
>>> Y = 2.0 * D + 1.0 * U + 0.3 * X + rng.standard_normal(n)
>>> df = pd.DataFrame({"y": Y, "treat": D, "z": Z, "w": W, "x": X})
>>> res = sp.bidirectional_pci(df, y="y", treat="treat",
...                            proxy_z=["z"], proxy_w=["w"],
...                            covariates=["x"], n_boot=50, seed=0)
>>> _ = res.summary()  # ATE recovered near the true value of 2.0

pci_mtp ¶

pci_mtp(data: DataFrame, y: str, treat: str, proxy_z: List[str], proxy_w: List[str], delta: float, covariates: Optional[List[str]] = None, alpha: float = 0.05, n_boot: int = 200, seed: int = 0) -> CausalResult

PCI for Modified Treatment Policies (continuous-shift effect).

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`		required
`y`	`str`	Outcome and continuous treatment.	required
`treat`	`str`	Outcome and continuous treatment.	required
`proxy_z`	`list of str`	Standard PCI proxies.	required
`proxy_w`	`list of str`	Standard PCI proxies.	required
`delta`	`float`	MTP shift; estimand is E[Y(D + δ)] - E[Y(D)].	required
`covariates`	`list of str`		`None`
`alpha`	`float`		`0.05`
`n_boot`	`int`		`200`
`seed`	`int`		`0`

Returns:

Type	Description
`CausalResult`

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> u = rng.normal(0, 1, 300)               # unmeasured confounder
>>> d = 0.5 * u + rng.normal(0, 1, 300)     # continuous treatment
>>> z = 0.8 * u + rng.normal(0, 1, 300)     # treatment-side proxy
>>> w = 0.8 * u + rng.normal(0, 1, 300)     # outcome-side proxy
>>> y = 1.0 * d + 1.5 * u + rng.normal(0, 1, 300)
>>> df = pd.DataFrame({"y": y, "d": d, "z": z, "w": w})
>>> res = sp.pci_mtp(df, y="y", treat="d", proxy_z=["z"], proxy_w=["w"],
...                  delta=1.0, n_boot=100, seed=0)
>>> bool(np.isfinite(res.estimate))
True

References

olivas2025proximal

select_pci_proxies ¶

select_pci_proxies(data: DataFrame, y: str, treat: str, candidates: List[str], covariates: Optional[List[str]] = None, top_k: int = 2) -> ProxyScoreResult

Score and rank candidate proxies for PCI.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`		required
`y`	`str`		required
`treat`	`str`		required
`candidates`	`list of str`	All variables that could plausibly serve as proxies.	required
`covariates`	`list of str`		`None`
`top_k`	`int`	Number of top candidates to recommend per side.	`2`

Returns:

Type	Description
`ProxyScoreResult`

Notes

If the partial-correlation / residualisation step fails for a candidate, its score falls back to the marginal correlation (with p_indep=1.0 on the Z-side, so the conditional-independence test is uninformative for that candidate) and a :class:~statspai.exceptions.ConvergenceWarning names the candidate and side. Scores for unaffected candidates are unchanged.

Examples:

>>> import statspai as sp
>>> import numpy as np
>>> import pandas as pd
>>> rng = np.random.default_rng(0)
>>> n = 300
>>> u = rng.normal(size=n)                 # unmeasured confounder
>>> z = u + rng.normal(scale=0.5, size=n)  # treatment-side proxy
>>> w = u + rng.normal(scale=0.5, size=n)  # outcome-side proxy
>>> d = (z + rng.normal(size=n) > 0).astype(float)
>>> y = 1.0 + 0.5 * d + w + rng.normal(size=n)
>>> df = pd.DataFrame({"y": y, "d": d, "z": z, "w": w})
>>> res = sp.select_pci_proxies(df, y="y", treat="d",
...                             candidates=["z", "w"])
>>> type(res).__name__
'ProxyScoreResult'
>>> sorted(res.z_candidates["name"])
['w', 'z']
>>> bool(len(res.recommended_z) <= 2)
True

statspai.proximal¶

proximal ¶

ProximalCausalInference ¶

NegativeControlResult dataclass ¶

to_dict ¶

ProximalRegResult dataclass ¶

to_dict ¶

ProxyScoreResult dataclass ¶

proximal ¶

negative_control_outcome ¶

negative_control_exposure ¶

double_negative_control ¶

proximal_regression ¶

fortified_pci ¶

bidirectional_pci ¶

pci_mtp ¶

select_pci_proxies ¶

`statspai.proximal`¶

NegativeControlResult `dataclass` ¶

ProximalRegResult `dataclass` ¶

ProxyScoreResult `dataclass` ¶