`statspai.ope`¶

ope ¶

Off-Policy Evaluation (sp.ope): estimate the value of a target policy from data collected under a different behaviour policy. Covers contextual bandits and off-policy reinforcement learning evaluation.

Implemented: DM, IPS, SNIPS, DR, Switch-DR, sharp OPE under unobserved confounding (Heß, Frauen, Melnychuk & Feuerriegel 2025, arXiv:2502.13022) [@hess2025efficient], causal-policy forest (Kato 2025, arXiv:2512.22846).

OPEResult `dataclass` ¶

Bases: ResultProtocolMixin

Canonical Off-Policy Evaluation result.

All sp.ope.* and sp.direct_method/ips/snips/doubly_robust estimators return this class (the policy_learning OPEResult subclass is a thin alias that adds an estimator attribute for back-compat). isinstance(res, sp.OPEResult) therefore holds for results from either entry point.

Examples:

>>> import numpy as np
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n, K = 500, 3
>>> pi_b = np.full((n, K), 1.0 / K)          # uniform behaviour policy
>>> pi_e = np.tile([0.6, 0.3, 0.1], (n, 1))  # evaluation policy to score
>>> actions = np.array([rng.choice(K, p=pi_b[i]) for i in range(n)])
>>> rewards = (actions == 0).astype(float) + rng.normal(0, 0.1, n)
>>> res = sp.ope.ips(actions, rewards, pi_b, pi_e)
>>> type(res).__name__
'OPEResult'
>>> res.method
'IPS'
>>> bool(isinstance(res, sp.OPEResult))
True

estimator `property` ¶

estimator: str

Backwards-compatible alias for :attr:method.

n_obs `property` ¶

n_obs: int

Return diagnostics['n'] or diagnostics['n_obs'] if present.

MarginalPolicyEffectResult `dataclass` ¶

Bases: ResultProtocolMixin

Result of :func:marginal_policy_effect.

Attributes:

Name	Type	Description
`estimate`	`float`	`theta`: removing a fraction `eps` of the treatments that occur lowers the average outcome per period by `eps * theta`.
`se`	`float`
`ci`	`tuple of float`
`per_treatment`	`float`	`theta` divided by the share of treated periods: the average effect of one treatment on the sum of the outcomes of the current and the next `horizon` periods.

Examples:

>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> w = rng.binomial(1, 0.3, 500)
>>> yv = 1.0 * w + rng.normal(size=500)
>>> df = pd.DataFrame({"y": yv, "w": w, "x": rng.normal(size=500)})
>>> r = sp.marginal_policy_effect(df, "y", "w", ["x"], horizon=2)
>>> isinstance(r, sp.MarginalPolicyEffectResult)
True

MDPPolicyValueResult `dataclass` ¶

Bases: ResultProtocolMixin

Result of :func:mdp_policy_value.

Attributes:

Name	Type	Description
`value`	`float`	Estimated long-run average outcome under the target policy, or its difference from the baseline policy when one was given.
`se`	`float`
`ci`	`tuple of float`
`value_policy, value_baseline`	`float`	The two long-run values (`value_baseline` is `nan` without a baseline).
`n_obs`	`int`	Number of transitions used.

Examples:

>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> s, rows = 0, []
>>> for t in range(600):
...     w = int(rng.random() < 0.5)
...     rows.append((s, w, s + w + rng.normal()))
...     s = int(rng.random() < (0.7 if w else 0.3))
>>> df = pd.DataFrame(rows, columns=["s", "w", "y"])
>>> r = sp.mdp_policy_value(df, "y", "w", ["s"], policy=1, propensity=0.5)
>>> isinstance(r, sp.MDPPolicyValueResult)
True

direct_method ¶

direct_method(reward_model: RewardModel, X: ndarray, pi_e: ndarray) -> OPEResult

Plug-in estimator using a fitted reward model Q(x, a).

Parameters:

Name	Type	Description	Default
`reward_model`	`callable`	`reward_model(X, a)` → vector of length n predicting E[R \| X, a].	required
`X`	`(n, d) array`		required
`pi_e`	`(n, K) array`	Evaluation policy probabilities over K actions at each X_i.	required

doubly_robust ¶

doubly_robust(X: ndarray, actions: ndarray, rewards: ndarray, pi_b: ndarray, pi_e: ndarray, reward_model: RewardModel, clip: float | None = 1000.0) -> OPEResult

Doubly Robust estimator (Dudik, Langford, Li 2011).

evaluate ¶

evaluate(method: str, *, X: ndarray | None = None, actions: ndarray | None = None, rewards: ndarray | None = None, pi_b: ndarray | None = None, pi_e: ndarray | None = None, reward_model: RewardModel | None = None, **kw: Any) -> OPEResult

Dispatch-by-name for OPE methods.

method : {"DM", "IPS", "SNIPS", "DR", "Switch-DR"}

ips ¶

ips(actions: ndarray, rewards: ndarray, pi_b: ndarray, pi_e: ndarray, clip: float | None = 1000.0) -> OPEResult

Inverse Propensity Scoring (aka IPS / Horvitz-Thompson).

snips ¶

snips(actions: ndarray, rewards: ndarray, pi_b: ndarray, pi_e: ndarray, clip: float | None = 1000.0) -> OPEResult

Self-Normalized IPS -- reduces variance at small bias cost.

switch_dr ¶

switch_dr(X: ndarray, actions: ndarray, rewards: ndarray, pi_b: ndarray, pi_e: ndarray, reward_model: RewardModel, tau: float = 10.0) -> OPEResult

Switch-DR (Wang, Agarwal, Dudík 2017): fall back to the DM whenever the importance ratio exceeds tau.

marginal_policy_effect ¶

marginal_policy_effect(data: DataFrame, y: str, treat: str, covariates: Sequence[str], *, horizon: int, propensity: Union[None, str, float] = None, trajectory: Optional[str] = None, time: Optional[str] = None, hac_lags: Optional[int] = None, alpha: float = 0.05) -> MarginalPolicyEffectResult

Effect of removing a small share of treatments in a dynamic system.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`	One row per period, in time order (or sortable by `time`).	required
`y`	`str`	Outcome of the period.	required
`treat`	`str`	Binary (0/1) treatment of the period.	required
`covariates`	`sequence of str`	Observed part of the state, given which treatment is as good as random. May be empty for a randomized treatment. Categorical columns (`category` dtype, strings, `C(col)`) are expanded into indicators, the first level as the base; integer columns are taken as numeric.	required
`horizon`	`int`	Number of future periods over which a treatment can still affect outcomes. The effect is on the sum of the outcomes of the current period and the next `horizon` periods.	required
`propensity`	`str or float`	Probability of treatment given the covariates: a column or one number. If omitted, a logistic regression on the covariates.	`None`
`trajectory`	`str`	Trajectory identifier and time column.	`None`
`time`	`str`	Trajectory identifier and time column.	`None`
`hac_lags`	`int`	Lags of the Bartlett kernel for the standard error. Defaults to `horizon` plus `floor(4 (T / 100) ** (2 / 9))`. Must be at least `horizon`, because the forward sums of periods up to `horizon` apart share outcomes, and smaller than the number of periods used.	`None`
`alpha`	`float`		`0.05`

Returns:

Type	Description
`MarginalPolicyEffectResult`

Notes

The estimand is theta = mean_t E[e(X_t) * Delta_t(X_t)], where e is the treatment probability and Delta_t the effect of treating in period t on the outcomes of periods t to t + horizon, given the observed covariates. It is identified without observing the whole state, provided treatment depends on the past only through the covariates. It answers "what are the treatments that occur worth, at the margin", not "what if no one were treated".

The estimator is doubly robust. Writing G_t for the forward sum of outcomes and m0 for its linear regression on the covariates among untreated periods, it averages W_t (G_t - m0) - (1 - W_t) e / (1 - e) (G_t - m0). The forward sums of neighbouring periods overlap, so the standard error uses a heteroskedasticity and autocorrelation consistent estimate. The last horizon periods of each trajectory have no complete forward sum and are dropped. Rows with a missing value at the start or the end of a trajectory are dropped as well; a missing value inside a trajectory is refused, because the forward sums would run across the gap.

A horizon that is too short leaves out part of the effect; one that is too long adds noise. Report the estimate for several values.

Examples:

>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> n = 3000
>>> w = rng.binomial(1, 0.3, n)
>>> u = np.zeros(n)
>>> for t in range(1, n):
...     u[t] = 0.5 * u[t - 1] + w[t - 1] + rng.normal()
>>> yv = w + u + rng.normal(size=n)   # one treatment is worth 1 + 1 + 0.5
>>> df = pd.DataFrame({"y": yv, "w": w})
>>> r = sp.marginal_policy_effect(df, "y", "w", [], horizon=2, propensity=0.3)
>>> bool(r.ci[0] < 0.3 * 2.5 < r.ci[1])
True

mdp_policy_value ¶

mdp_policy_value(data: DataFrame, y: str, treat: str, state: Sequence[str], policy: PolicyLike, *, baseline: Optional[PolicyLike] = None, propensity: Union[None, str, float] = None, features: Any = None, trajectory: Optional[str] = None, time: Optional[str] = None, alpha: float = 0.05) -> MDPPolicyValueResult

Long-run average outcome under a policy in a Markov decision process.

Parameters:

Name	Type	Description	Default
`data`	`DataFrame`	One row per period, in time order (or sortable by `time`), from one or several trajectories.	required
`y`	`str`	Outcome (reward) of the period.	required
`treat`	`str`	Action taken in the period.	required
`state`	`sequence of str`	Columns of the state observed before the action. The state must carry everything through which past actions affect the future.	required
`policy`	`(callable, str or scalar)`	The target policy: a function of the state columns returning one action per row, the name of a column holding the action the policy would take, or a single action taken always.	required
`baseline`	`(callable, str or scalar)`	A second policy. If given, the result is the difference in long-run value between `policy` and `baseline`.	`None`
`propensity`	`str or float`	Probability with which the action actually taken was chosen, given the state: a column, or one number if every action had the same probability. If omitted it is estimated by the frequency of the action in each distinct state, which requires a state with few distinct values.	`None`
`features`	`"tabular", sequence of str or callable`	Basis for the excess-reward function and the stationary ratio. `"tabular"` (default) uses one indicator per distinct state and is exact for a finite state space. A list of numeric state columns, or a callable returning a matrix, gives a linear basis (a constant is added).	`None`
`trajectory`	`str`	Column identifying trajectories; transitions are not formed across them.	`None`
`time`	`str`	Column to sort by within a trajectory.	`None`
`alpha`	`float`		`0.05`

Returns:

Type	Description
`MDPPolicyValueResult`

Notes

The assumptions are those of a time-homogeneous Markov decision process observed in its stationary regime: the state is fully observed, the action depends on the past only through the state, and both the data and the target policy visit the same states. With the tabular basis a failure of this condition is detected exactly: if the policy leads to a state in which the data never contain the action the policy takes there, IdentificationFailure is raised naming the state (the value would otherwise be computed as if the policy never entered it). With a linear basis the same failure cannot be detected exactly and shows only in the estimated stationary ratio (detail["min_omega"] and detail["max_omega"] report the range of the estimated stationary ratio; a warning is issued when it is negative, which a linear basis can produce when overlap is poor).

The two nuisance functions are fitted on the same trajectory, by moment equations that are linear in the basis. With the tabular basis the estimate coincides with the plug-in value of the estimated transition model. The standard error uses the fact that the summands are martingale differences at the truth, so no correction for serial correlation is applied; it does not account for the estimation of propensity when that is omitted. Value and standard error depend on the basis only through its span: not on how the states are labelled (tabular basis), and not on the origin or the unit of a numeric feature (linear basis).

Rows with a missing value at the start or the end of a trajectory are dropped; a missing value inside a trajectory is refused, because the periods on either side would be read as consecutive.

References

[@liao2022batch], [@kallus2022efficiently]

Examples:

>>> import numpy as np, pandas as pd
>>> import statspai as sp
>>> rng = np.random.default_rng(0)
>>> s, rows = 0, []
>>> for t in range(2000):
...     w = int(rng.random() < 0.5)
...     rows.append((s, w, s + w + rng.normal()))
...     s = int(rng.random() < (0.7 if w else 0.3))
>>> df = pd.DataFrame(rows, columns=["s", "w", "y"])
>>> r = sp.mdp_policy_value(df, "y", "w", ["s"], policy=1, baseline=0,
...                         propensity=0.5)
>>> bool(r.ci[0] < 1.4 < r.ci[1])   # true difference: 1 + 0.7 - 0.3
True

statspai.ope¶

ope ¶

OPEResult dataclass ¶

estimator property ¶

n_obs property ¶

MarginalPolicyEffectResult dataclass ¶

MDPPolicyValueResult dataclass ¶

direct_method ¶

doubly_robust ¶

evaluate ¶

ips ¶

snips ¶

switch_dr ¶

marginal_policy_effect ¶

mdp_policy_value ¶

`statspai.ope`¶

OPEResult `dataclass` ¶

estimator `property` ¶

n_obs `property` ¶

MarginalPolicyEffectResult `dataclass` ¶

MDPPolicyValueResult `dataclass` ¶