statspai.longitudinal

longitudinal

Longitudinal causal inference (sp.longitudinal).

Unified entry for What If Layer-4 methods (time-varying treatments with time-varying confounders). Wraps MSM / g-formula ICE / IPW under a single dispatcher with a dynamic-regime DSL.

import statspai as sp r = sp.longitudinal.analyze( ... data=panel, ... id="pid", ... time="visit", ... treatment="drug", ... outcome="cd4", ... time_varying=["cd4_lag", "viral_load_lag"], ... baseline=["age", "sex"], ... regime="if cd4_lag < 200 then 1 else 0", ... ) r.summary()

diff = sp.longitudinal.contrast( ... data=panel, id="pid", time="visit", ... treatment="drug", outcome="cd4", ... regime_a="always_treat", ... regime_b="never_treat", ... time_varying=["cd4_lag"], ... )

Regime dataclass

A dynamic treatment regime.

Attributes:

Name	Type	Description
kind	{'static', 'dynamic'}
name	str
rule	Union[list, Callable]	Static regimes store a list/sequence; dynamic regimes store a callable (history_dict, t) -> treatment_value.

Examples:

>>> import statspai as sp
>>> reg = sp.Regime(kind="static", name="all1", rule=[1.0, 1.0])
>>> reg.treatment({}, t=0)
1.0
>>> reg.kind
'static'

treatment

treatment(history: dict, t: int = 0, K: int = 1) -> Any

Evaluate the regime at time t given history.

apply

apply(history_df: DataFrame) -> Series

Vectorized: apply the regime to each row of a history dataframe.

LongitudinalResult dataclass

Bases: ResultProtocolMixin

Result of a unified longitudinal analysis.

Attributes:

Name	Type	Description
method	str	Which estimator was used ("msm", "g-formula", "ipw", "ltmle").
regime_name	str	Name of the regime being evaluated.
estimate	float	E[Y(regime)] under the chosen estimator.
se	float
ci	tuple[float, float]
n	int	Panel sample size.
n_periods	int
diagnostics	dict	Weight quantiles, positivity flags, etc.
underlying_result	Any	Raw result object from the underlying estimator.

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> rows = []
>>> for i in range(80):
...     base = rng.normal()
...     for t in range(3):
...         a = int(rng.random() < 0.5)
...         y = 0.5 * a + 0.3 * base + rng.normal(0, 0.5)
...         rows.append({"id": i, "time": t, "A": a, "Y": y})
>>> df = pd.DataFrame(rows)
>>> res = sp.longitudinal_analyze(
...     df, id="id", time="time", treatment="A", outcome="Y",
...     regime="always_treat")
>>> isinstance(res, sp.LongitudinalResult)
True
>>> res.method
'ipw'
>>> res.n, res.n_periods
(80, 3)
>>> bool("Longitudinal analysis" in res.summary())
True

always_treat

always_treat(K: int = 1) -> Regime

Convenience: the always-treat regime over K periods.

Examples:

>>> import statspai as sp
>>> r = sp.always_treat(K=3)
>>> r.treatment({}, t=0), r.treatment({}, t=2)
(1.0, 1.0)

never_treat

never_treat(K: int = 1) -> Regime

Convenience: the never-treat regime over K periods.

Examples:

>>> import statspai as sp
>>> r = sp.never_treat(K=3)
>>> r.treatment({}, t=0), r.treatment({}, t=2)
(0.0, 0.0)

contrast

contrast(data: DataFrame, id: str, time: str, treatment: str, outcome: str, regime_a: Union[str, Sequence, Regime], regime_b: Union[str, Sequence, Regime], **kwargs: Any) -> dict

Estimate E[Y(regime_a)] - E[Y(regime_b)] using :func:analyze.

Returns:

Type	Description
dict	With keys regime_a, regime_b, contrast, a_result, b_result. The contrast value is the plug-in difference a.estimate - b.estimate; its SE uses the delta-method approximation sqrt(se_a^2 + se_b^2).

Examples:

>>> import statspai as sp
>>> import numpy as np, pandas as pd
>>> rng = np.random.default_rng(0)
>>> rows = []
>>> for i in range(80):
...     base = rng.normal()
...     for t in range(3):
...         a = int(rng.random() < 0.5)
...         y = 0.5 * a + 0.3 * base + rng.normal(0, 0.5)
...         rows.append({"id": i, "time": t, "A": a, "Y": y})
>>> df = pd.DataFrame(rows)
>>> res = sp.longitudinal_contrast(
...     df, id="id", time="time", treatment="A", outcome="Y",
...     regime_a="always_treat", regime_b="never_treat")
>>> res["regime_a"], res["regime_b"]
('always_treat', 'never_treat')
>>> sorted(res.keys())
['a_result', 'b_result', 'ci', 'contrast', 'regime_a', 'regime_b', 'se']
>>> isinstance(res["contrast"], float)
True