Exchangeability Testing¶

Tests and monitoring tools for assessing exchangeability.

See Exchangeability Tests for conceptual background and guidance on selecting a test.

High-level Interfaces¶

mapie.exchangeability_testing.FixedDatasetExchangeabilityTest ¶

FixedDatasetExchangeabilityTest(
    method_names: Union[
        FixedDatasetTestMethods,
        Literal["all"],
        List[FixedDatasetTestMethods],
    ] = "all",
    method_params: Optional[MethodParams] = None,
    test_level: float = 0.05,
    warn: bool = False,
)

Run one or several exchangeability tests on a labeled dataset.

This wrapper provides a high-level interface around the exchangeability testing methods implemented in MAPIE. It can instantiate permutation-based tests as well as online martingale tests and run them through a shared API.

PARAMETER	DESCRIPTION
`method_names`	Name of the test method to run, a list of method names, or `"all"` to run every available fixed-dataset method. TYPE: `Union[FixedDatasetTestMethods, Literal['all'], List[FixedDatasetTestMethods]]` DEFAULT: `"all"`
`method_params`	Additional keyword arguments passed to each method constructor. Keys are method names and values are dictionaries of keyword arguments. TYPE: `Optional[MethodParams]` DEFAULT: `None`
`test_level`	Significance level passed to each underlying test. TYPE: `float` DEFAULT: `0.05`
`warn`	Whether underlying methods should raise warnings when they reject exchangeability. TYPE: `bool` DEFAULT: `False`

Examples:

>>> import numpy as np
>>> X = np.arange(20, dtype=float).reshape(-1, 1)
>>> y = 2 * X.ravel() + np.linspace(0.0, 0.1, X.shape[0])
>>> test = FixedDatasetExchangeabilityTest(
...     method_names="pvalue_permutation", warn=False
... )
>>> _ = test.run(X, y)

Source code in mapie/exchangeability_testing/exchangeability.py

def __init__(
    self,
    method_names: Union[
        FixedDatasetTestMethods, Literal["all"], List[FixedDatasetTestMethods]
    ] = "all",
    method_params: Optional[MethodParams] = None,
    test_level: float = 0.05,
    warn: bool = False,
) -> None:
    if method_names == "all":
        self.method_names = list(fixed_dataset_test_method_choice_map.keys())
    elif isinstance(method_names, str):
        self.method_names = [method_names]
    elif isinstance(method_names, list):
        self.method_names = cast(List[str], method_names)
    else:
        raise ValueError(
            f"Invalid method_names type: {type(method_names)}. Must be a string, list, or 'all'."
        )

    for method_name in self.method_names:
        if method_name not in fixed_dataset_test_method_choice_map:
            raise ValueError(
                f"Invalid method name: {method_name}. Valid methods are: {list(fixed_dataset_test_method_choice_map.keys())}"
            )

    self.test_level = test_level
    self.warn = warn
    self.method_params = method_params or {}
    self.test_methods = [
        self._init_test_method(method_name) for method_name in self.method_names
    ]

is_exchangeable `property` ¶

is_exchangeable: Dict[str, ExchangeabilityDecision]

Return the current exchangeability decision for each configured method.

RETURNS	DESCRIPTION
`Dict[str, Optional[bool]]`	A dictionary mapping each method name to its current decision. Values are typically `True`, `False`, or `None` when the underlying test is still inconclusive.

run ¶

run(
    X_test: NDArray, y_test: NDArray
) -> Dict[str, FixedDatasetTestMethod]

Run all configured exchangeability tests on the provided dataset.

PARAMETER	DESCRIPTION
`X_test`	Feature matrix of the labeled dataset. TYPE: `NDArray`
`y_test`	Labels or targets associated with `X_test`. TYPE: `NDArray`

RETURNS	DESCRIPTION
`Dict[str, FixedDatasetTestMethod]`	A dictionary mapping each method name to the updated underlying test instance.

RAISES	DESCRIPTION
`AttributeError`	If one of the configured test methods defines neither `update` nor `run`.

Source code in mapie/exchangeability_testing/exchangeability.py

def run(
    self,
    X_test: NDArray,
    y_test: NDArray,
) -> Dict[str, FixedDatasetTestMethod]:
    """
    Run all configured exchangeability tests on the provided dataset.

    Parameters
    ----------
    X_test : NDArray
        Feature matrix of the labeled dataset.

    y_test : NDArray
        Labels or targets associated with ``X_test``.

    Returns
    -------
    Dict[str, FixedDatasetTestMethod]
        A dictionary mapping each method name to the updated underlying
        test instance.

    Raises
    ------
    AttributeError
        If one of the configured test methods defines neither ``update``
        nor ``run``.
    """
    results = {}
    for test_method, method_name in zip(self.test_methods, self.method_names):
        if callable(getattr(test_method, "update", None)):
            results[method_name] = cast(
                UpdatableExchangeabilityTestProtocol, test_method
            ).update(X_test, y_test)
        elif callable(getattr(test_method, "run", None)):
            results[method_name] = cast(
                RunnableExchangeabilityTestProtocol, test_method
            ).run(X_test, y_test)
        else:
            raise AttributeError(
                f"Test method '{method_name}' must define either 'update' or 'run'."
            )

    return results

mapie.exchangeability_testing.OnlineExchangeabilityTest ¶

OnlineExchangeabilityTest(
    method_names: Union[
        OnlineTestMethods,
        Literal["all"],
        List[OnlineTestMethods],
    ] = "all",
    method_params: Optional[MethodParams] = None,
    test_level: float = 0.05,
    warn: bool = True,
)

Monitor exchangeability online with one or several martingale tests.

This wrapper exposes a shared interface for the online exchangeability testing methods available in MAPIE. Each configured method is updated on the same labeled stream, allowing side-by-side monitoring of different martingale constructions.

PARAMETER	DESCRIPTION
`method_names`	Name of the online method to use, a list of method names, or `"all"` to instantiate every available online method. TYPE: `Union[OnlineTestMethods, Literal['all'], List[OnlineTestMethods]]` DEFAULT: `"all"`
`method_params`	Additional keyword arguments passed to each method constructor. Keys are method names and values are dictionaries of keyword arguments. TYPE: `Optional[MethodParams]` DEFAULT: `None`
`test_level`	Significance level passed to each underlying online test. TYPE: `float` DEFAULT: `0.05`
`warn`	Whether underlying methods should raise warnings when they reject exchangeability. TYPE: `bool` DEFAULT: `True`

Examples:

>>> import numpy as np
>>> X = np.arange(120, dtype=float).reshape(-1, 1)
>>> y = 2 * X.ravel() + np.linspace(0.0, 0.1, X.shape[0])
>>> online_test = OnlineExchangeabilityTest(
...     method_names="plugin_martingale", warn=False
... )
>>> _ = online_test.update(X, y)

Source code in mapie/exchangeability_testing/exchangeability.py

def __init__(
    self,
    method_names: Union[
        OnlineTestMethods, Literal["all"], List[OnlineTestMethods]
    ] = "all",
    method_params: Optional[MethodParams] = None,
    test_level: float = 0.05,
    warn: bool = True,
) -> None:
    if method_names == "all":
        self.method_names = list(online_test_method_choice_map.keys())
    elif isinstance(method_names, str):
        self.method_names = [method_names]
    elif isinstance(method_names, list):
        self.method_names = cast(List[str], method_names)
    else:
        raise ValueError(
            f"Invalid method_names type: {type(method_names)}. Must be a string, list, or 'all'."
        )

    for method_name in self.method_names:
        if method_name not in online_test_method_choice_map:
            raise ValueError(
                f"Invalid method name: {method_name}. Valid methods are: {list(online_test_method_choice_map.keys())}"
            )

    self.test_level = test_level
    self.warn = warn
    self.method_params = method_params or {}
    self.test_methods = [
        self._init_test_method(method_name) for method_name in self.method_names
    ]

is_exchangeable `property` ¶

is_exchangeable: Dict[str, ExchangeabilityDecision]

Return the current exchangeability decision for each online method.

RETURNS	DESCRIPTION
`Dict[str, Optional[bool]]`	A dictionary mapping each method name to its current online decision. Values may be `None` during the burn-in phase.

update ¶

update(
    X_test: NDArray, y_test: NDArray
) -> Dict[str, OnlineTestMethod]

Update all configured online tests with newly labeled observations.

PARAMETER	DESCRIPTION
`X_test`	Feature matrix for the newly observed batch. TYPE: `NDArray`
`y_test`	Labels or targets associated with `X_test`. TYPE: `NDArray`

RETURNS	DESCRIPTION
`Dict[str, OnlineMartingaleTest]`	A dictionary mapping each method name to the updated underlying test instance.

RAISES	DESCRIPTION
`AttributeError`	If one of the configured methods does not define `update`.

Source code in mapie/exchangeability_testing/exchangeability.py

def update(
    self,
    X_test: NDArray,
    y_test: NDArray,
) -> Dict[str, OnlineTestMethod]:
    """
    Update all configured online tests with newly labeled observations.

    Parameters
    ----------
    X_test : NDArray
        Feature matrix for the newly observed batch.

    y_test : NDArray
        Labels or targets associated with ``X_test``.

    Returns
    -------
    Dict[str, OnlineMartingaleTest]
        A dictionary mapping each method name to the updated underlying
        test instance.

    Raises
    ------
    AttributeError
        If one of the configured methods does not define ``update``.
    """
    results = {}
    for test_method, method_name in zip(self.test_methods, self.method_names):
        if callable(getattr(test_method, "update", None)):
            results[method_name] = test_method.update(X_test, y_test)
        else:
            raise AttributeError(
                f"Test method '{method_name}' must define an 'update' method."
            )
    return results

mapie.exchangeability_testing.RiskMonitoring ¶

RiskMonitoring(
    risk: RiskLike,
    test_level: float = 0.05,
    tolerance: float = 0.05,
    tolerance_type: Literal[
        "absolute", "relative"
    ] = "absolute",
    threshold: Optional[float] = None,
    reference_data: Optional[
        Tuple[NDArray, NDArray]
    ] = None,
    warn: bool = True,
)

Monitor a risk on an online stream relative to a reference set.

The class first estimates an acceptable risk threshold on reference data and then updates a time-uniform lower confidence bound on the online risk as new observations arrive. A harmful shift is detected when the latest online lower bound exceeds the threshold.

PARAMETER	DESCRIPTION
`risk`	Risk to monitor. If a string is provided, it must be one of the keys in `mapie.risk_control.risks.binary_risk_choice_map` or `mapie.risk_control.risks.continuous_risk_choice_map`. TYPE: `RiskLike`
`test_level`	Level used to test the hypothesis that the online risk is greater than the reference risk. The probability that the test gives a false positive is at most test_level (type I error). TYPE: `float` DEFAULT: `0.05`
`tolerance`	Margin applied to the reference upper confidence bound to define the monitoring threshold. TYPE: `float` DEFAULT: `0.05`
`tolerance_type`	Whether `tolerance` is added to the reference upper bound or applied as a multiplicative factor. TYPE: `(absolute, relative)` DEFAULT: `"absolute"`
`threshold`	Precomputed monitoring threshold. If provided, `compute_threshold` must not be called. TYPE: `Optional[float]` DEFAULT: `None`
`reference_data`	Optional reference labels and predictions `(y_true, y_pred)` used to compute the threshold at initialization time. This is ignored if `threshold` is also provided. TYPE: `Optional[Tuple[NDArray, NDArray]]` DEFAULT: `None`
`warn`	Whether to emit a warning when a harmful shift is detected. TYPE: `bool` DEFAULT: `True`

ATTRIBUTE	DESCRIPTION
`risk`	Resolved risk object used internally. TYPE: `BinaryRisk`
`threshold`	Monitoring threshold used to flag harmful shifts. TYPE: `Optional[float]`
`reference_risk_upper_bound`	Upper confidence bound estimated on the reference risk, available after `compute_threshold`. TYPE: `Optional[float]`
`online_risk_sequence_history`	Concatenated sequence of observed online risk values. TYPE: `NDArray[float64]`
`online_risk_lower_bound_sequence_history`	History of online lower confidence bounds. TYPE: `NDArray[float64]`
`online_risk_lower_bound_latest`	Latest value of the online lower confidence bound. TYPE: `Optional[float]`

Examples:

>>> from sklearn.linear_model import LogisticRegression
>>> from sklearn.datasets import make_classification
>>> from sklearn.model_selection import train_test_split
>>> from mapie.exchangeability_testing import RiskMonitoring

>>> X, y = make_classification(n_samples=300, n_features=2, n_redundant=0, n_informative=2, random_state=42, class_sep=2.0)
>>> X_train, X_test, y_train, y_test = train_test_split(
...     X, y, test_size=0.4, random_state=42
... )
>>> clf = LogisticRegression().fit(X_train, y_train)
>>> monitor = RiskMonitoring(risk="accuracy", warn=False)
>>> y_pred = clf.predict(X_test)
>>> _ = monitor.compute_threshold(y_test, y_pred)
>>> X_online, y_online = make_classification(n_samples=200, n_features=2, n_redundant=0, n_informative=2, random_state=42, class_sep=0.3)
>>> y_pred_online = clf.predict(X_online)
>>> _ = monitor.update(y_online, y_pred_online)
>>> print(monitor.harmful_shift_detected)
True

References

[1] Aleksandr Podkopaev and Aaditya Ramdas. Tracking the risk of a deployed model and detecting harmful distribution shifts. International Conference on Learning Representations, 2022.

Source code in mapie/exchangeability_testing/risk_monitoring.py

def __init__(
    self,
    risk: RiskLike,
    test_level: float = 0.05,
    tolerance: float = 0.05,
    tolerance_type: Literal["absolute", "relative"] = "absolute",
    threshold: Optional[float] = None,
    reference_data: Optional[Tuple[NDArray, NDArray]] = None,
    warn: bool = True,
) -> None:
    try:
        resolved_risk = risk_choice_map[risk] if isinstance(risk, str) else risk
    except KeyError as e:
        raise ValueError(
            "When risk is provided as a string, it must be one of: "
            f"{list(risk_choice_map.keys())}"
        ) from e
    if not isinstance(resolved_risk, BinaryRisk):
        raise TypeError(
            "risk must be a single BinaryRisk instance or a supported risk name."
        )
    self.risk: BinaryRisk = resolved_risk
    self.tolerance = tolerance
    self.tolerance_type = tolerance_type
    self.warn = warn
    self.threshold = threshold

    if not (0.0 < test_level < 1.0):
        raise ValueError("test_level must be in (0, 1).")
    self.test_level_reference = test_level / 2
    self.test_level_online = test_level / 2

    self.reference_risk_upper_bound: Optional[float] = None
    self.online_risk_sequence_history: NDArray[np.float64] = np.array(
        [], dtype=float
    )
    self.online_risk_lower_bound_sequence_history: NDArray[np.float64] = np.array(
        [], dtype=float
    )
    self.online_risk_lower_bound_latest: Optional[float] = None

    if reference_data is not None and threshold is None:
        self.compute_threshold(reference_data[0], reference_data[1])
    elif reference_data is not None and threshold is not None:
        warnings.warn(
            "reference_data and threshold are both provided. The threshold will be used and the reference data will be ignored.",
            UserWarning,
        )

harmful_shift_detected `property` ¶

harmful_shift_detected: bool

Whether the latest online lower bound exceeds the threshold.

compute_threshold ¶

compute_threshold(
    y_true: NDArray, y_pred: NDArray
) -> RiskMonitoring

Estimate the monitoring threshold from reference predictions. Data can be the test set on which the model is evaluated before deployment.

PARAMETER	DESCRIPTION
`y_true`	Ground-truth binary labels for the reference data. TYPE: `NDArray`
`y_pred`	Predicted binary labels for the reference data. TYPE: `NDArray`

RETURNS	DESCRIPTION
`RiskMonitoring`	The fitted instance.

Source code in mapie/exchangeability_testing/risk_monitoring.py

def compute_threshold(self, y_true: NDArray, y_pred: NDArray) -> "RiskMonitoring":
    """
    Estimate the monitoring threshold from reference predictions.
    Data can be the test set on which the model is evaluated before deployment.

    Parameters
    ----------
    y_true : NDArray
        Ground-truth binary labels for the reference data.
    y_pred : NDArray
        Predicted binary labels for the reference data.

    Returns
    -------
    RiskMonitoring
        The fitted instance.
    """
    if self.threshold is not None:
        warnings.warn(
            "Threshold is already computed and will be replaced.",
            UserWarning,
        )

    reference_risk_sequence = self.risk.get_risk_sequence(y_true, y_pred)
    if reference_risk_sequence.size == 0:
        raise ValueError(
            "Reference risk is undefined because no samples satisfy the risk condition."
        )

    self.reference_risk_upper_bound = hoeffding_bound(
        reference_risk_sequence,
        self.test_level_reference,
        bound_side="upper",
    )

    if self.tolerance_type == "absolute":
        self.threshold = self.reference_risk_upper_bound + self.tolerance
    elif self.tolerance_type == "relative":
        self.threshold = self.reference_risk_upper_bound * (1 + self.tolerance)
    else:
        raise ValueError(
            "Invalid tolerance type. Must be 'absolute' or 'relative'."
        )

    return self

update ¶

update(y_true: NDArray, y_pred: NDArray) -> RiskMonitoring

Update the online risk history and its lower confidence bound. Raises a warning when a harmful shift is detected.

PARAMETER	DESCRIPTION
`y_true`	Ground-truth binary labels for the newly observed online data. TYPE: `NDArray`
`y_pred`	Predicted binary labels for the newly observed online data. TYPE: `NDArray`

RETURNS	DESCRIPTION
`RiskMonitoring`	The updated instance.

Source code in mapie/exchangeability_testing/risk_monitoring.py

def update(self, y_true: NDArray, y_pred: NDArray) -> "RiskMonitoring":
    """
    Update the online risk history and its lower confidence bound.
    Raises a warning when a harmful shift is detected.

    Parameters
    ----------
    y_true : NDArray
        Ground-truth binary labels for the newly observed online data.
    y_pred : NDArray
        Predicted binary labels for the newly observed online data.

    Returns
    -------
    RiskMonitoring
        The updated instance.
    """
    if self.threshold is None:
        raise ValueError(
            "Threshold must be computed with compute_threshold or set at initialization before updating the online risk"
        )

    new_risk_sequence = self.risk.get_risk_sequence(y_true, y_pred)
    if new_risk_sequence.size == 0:
        return self
    self.online_risk_sequence_history = np.concatenate(
        [self.online_risk_sequence_history, new_risk_sequence]
    )

    # in the current implementation, the bound is recomputed from scratch with the full history
    new_risk_lower_bound_sequence = conjugate_mixture_empirical_bernstein_bound(
        self.online_risk_sequence_history,
        v_opt=1,
        alpha=self.test_level_online,
        bound_side="lower",
    )

    self.online_risk_lower_bound_sequence_history = np.asarray(
        new_risk_lower_bound_sequence, dtype=float
    )
    self.online_risk_lower_bound_latest = (
        self.online_risk_lower_bound_sequence_history[-1]
    )

    if self.harmful_shift_detected and self.warn:
        warnings.warn(
            f"Harmful shift detected. The last value of the online risk lower bound ({self.online_risk_lower_bound_latest:.3f}) is greater than the threshold ({self.threshold:.3f})."
        )

    return self

summary ¶

summary() -> None

Placeholder for a future summary API.

Source code in mapie/exchangeability_testing/risk_monitoring.py

def summary(self) -> None:
    """Placeholder for a future summary API."""
    pass

Individual Tests¶

mapie.exchangeability_testing.OnlineMartingaleTest ¶

OnlineMartingaleTest(
    mapie_estimator: Optional[MapieEstimator] = None,
    task: Optional[
        Literal["classification", "regression"]
    ] = None,
    test_method: Literal[
        "jumper_martingale", "plugin_martingale"
    ] = "jumper_martingale",
    test_level: float = 0.05,
    warn: bool = True,
    jump_size: float = 0.01,
    burn_in: int = 100,
    random_state: Optional[int_] = None,
)

Online test of exchangeability based on conformal p-values and test martingales.

OnlineMartingaleTest sequentially monitors whether newly observed labeled data remain exchangeable with respect to a reference stream, using conformity scores, conformal p-values, and a martingale-based evidence process.

At each update, the class:

computes conformity scores from observed features and labels,
converts these scores into conformal p-values using past scores,
updates a martingale statistic from the p-values,
monitors whether the observed stream provides evidence against exchangeability.

Two martingale constructions are currently supported:

"jumper_martingale": a simple and robust betting martingale based on a finite set of experts.
"plugin_martingale": a plug-in martingale using an estimated density of past p-values.

The null hypothesis is that the sequence of observations is exchangeable. Large martingale values provide evidence against exchangeability.

PARAMETER	DESCRIPTION
`mapie_estimator`	MAPIE estimator used to compute predictions and non-conformity scores. Supported estimators are `SplitConformalClassifier`, and `SplitConformalRegressor`. If `None`, a default `SplitConformalClassifier` or `SplitConformalRegressor` is built when needed. If the estimator is not fitted or not provided, it will be fitted on a slice of the data in order to compute non-conformity scores. TYPE: `Optional[MapieEstimator]` DEFAULT: `None`
`task`	Task type. If `None`, the task is inferred from `y`. TYPE: `Optional[Literal['classification', 'regression']]` DEFAULT: `None`
`test_method`	Martingale construction used to aggregate evidence across p-values. To compare both methods in parallel, instantiate two `OnlineMartingaleTest` objects with different `test_method` values and update them on the same stream. TYPE: `('jumper_martingale', 'plugin_martingale')` DEFAULT: `"jumper_martingale"`
`test_level`	Level used to test the hypothesis that the dataset is exchangeable. The probability that the test gives a false positive is at most `test_level` (type I error). TYPE: `float` DEFAULT: `0.05`
`warn`	Whether to raise a warning when exchangeability is rejected. The warning is issued at most once per instance. TYPE: `bool` DEFAULT: `True`
`jump_size`	Mixing parameter used by the jumper martingale. Ignored when `test_method="plugin_martingale"`. TYPE: `float` DEFAULT: `0.01`
`burn_in`	Minimum sample size required before the `is_exchangeable` property is allowed to return a non-`None` decision. TYPE: `int` DEFAULT: `100`
`random_state`	Random seed used for random tie-breaking and density estimation. TYPE: `Optional[int]` DEFAULT: `None`

ATTRIBUTE	DESCRIPTION
`pvalue_history`	History of conformal p-values observed so far. TYPE: `list of float`
`conformity_score_history`	History of conformity scores observed so far. TYPE: `list of float`
`martingale_value_history`	History of martingale values after each update. TYPE: `list of float`
`current_martingale_value`	Current value of the martingale process. TYPE: `float`

Examples:

>>> RANDOM_STATE = 7
>>> mapie_estimator = SplitConformalRegressor(prefit=False)
>>> omt = OnlineMartingaleTest(
...     mapie_estimator=mapie_estimator,
...     task="regression",
...     random_state=0,
...     burn_in=1,
... )
>>> rng = np.random.default_rng(RANDOM_STATE)
>>> X = np.linspace(0.1, 0.9, 2400).reshape(-1, 1)
>>> y = 3.0 * X.ravel() + rng.normal(scale=0.1, size=X.shape[0])
>>> omt = omt.update(X, y)
>>> omt.is_exchangeable is True
True

Notes

The class is designed for sequential monitoring. It can be initialized on a reference labeled dataset using update and then updated online whenever new labels become available.

The martingale provides a valid sequential test against exchangeability when the p-values are valid under the null hypothesis.

References

.. [1] Angelopoulos, Barber, Bates (2026). "Theoretical Foundations of Conformal Prediction". arXiv preprint arXiv:2411.11824. .. [2] Vovk, Gammerman, Shafer (2005). "Algorithmic Learning in a Random World". Boston, MA: Springer US. Section 7.1, page 169. .. [3] Fedorova, Gammerman, Nouretdinov, Vovk (2012). "Plug-in Martingales for Testing Exchangeability on-line". In Proceedings of the 29^th ICML. Algorithm 1, page 3.

Initialize the online martingale test.

PARAMETER	DESCRIPTION
`mapie_estimator`	MAPIE estimator used to compute predictions and non-conformity scores. Supported estimators are `SplitConformalClassifier`, and `SplitConformalRegressor`. If `None`, a default `SplitConformalClassifier` or `SplitConformalRegressor` is built when needed. If the estimator is not fitted or not provided, it will be fitted on a slice of the data in order to compute non-conformity scores. TYPE: `Optional[MapieEstimator]` DEFAULT: `None`
`task`	Task type. If `None`, the task is inferred from `y`. TYPE: `Optional[Literal['classification', 'regression']]` DEFAULT: `None`
`test_method`	Martingale construction used to aggregate evidence from conformal p-values. "jumper_martingale" is more stable and less tuning-sensitive. "plugin_martingale" is more adaptive but sensitive to density estimation. To monitor both methods, use two instances and update both online. TYPE: `('jumper_martingale', 'plugin_martingale')` DEFAULT: `"jumper_martingale"`
`test_level`	Level used to test the hypothesis that the dataset is exchangeable. The probability that the test gives a false positive is at most `test_level` (type I error). TYPE: `float` DEFAULT: `0.05`
`warn`	Whether to raise a warning when exchangeability is rejected. TYPE: `bool` DEFAULT: `True`
`jump_size`	Mixing parameter for the jumper martingale, controlling expert diversity. Must lie in (0, 1). Ignored when test_method="plugin_martingale". TYPE: `float` DEFAULT: `0.01`
`burn_in`	Minimum number of observations required before is_exchangeable returns a non-None decision. TYPE: `int` DEFAULT: `100`
`random_state`	Random seed used for randomization (e.g., tie-breaking in p-value computation). TYPE: `Optional[int]` DEFAULT: `None`

RAISES	DESCRIPTION
`ValueError`	If test_level is not in (0, 1), if test_method is not supported, or if jump_size is not in (0, 1).

reject_threshold `property` ¶

reject_threshold: float

Return the martingale rejection threshold.

RETURNS	DESCRIPTION
`float`	Rejection threshold equal to `1 / test_level`. Exchangeability is rejected when the martingale exceeds this threshold.

is_exchangeable `property` ¶

is_exchangeable: Optional[bool]

Return the current exchangeability decision based on the martingale process.

The decision is based on the trajectory of martingale values compared to the rejection threshold (1 / test_level). The interpretation is:

False: Exchangeability is rejected when the martingale exceeds the rejection threshold at least once.
True: Failure to reject exchangeability when the martingale remains below the significance level (test_level) throughout the history.
None: The test is currently inconclusive because insufficient observations have been processed (fewer than burn_in).

RETURNS	DESCRIPTION
`Optional[bool]`	Exchangeability decision, or `None` if inconclusive.

Notes

This implementation uses a persistent stopping-rule interpretation: once the martingale has crossed the rejection threshold at any time, the decision remains False thereafter, even if the martingale later decreases.

Therefore, True should be interpreted as "no rejection so far", not as evidence in favor of exchangeability.

compute_p_value ¶

compute_p_value(
    current_conformity_score: float,
    conformity_score_history: NDArray,
) -> float

Compute the conformal p-value associated with a new conformity score.

The p-value is computed using only the past conformity scores, according to the empirical conformal formula:

.. math::

p_t = \frac{1 + \#\{i : s_i > s_t\} + U \cdot \#\{i : s_i = s_t\}}{n + 1}

where \(s_t\) is the current conformity score, \(s_i\) are past scores, \(U \sim \mathrm{Uniform}(0, 1)\) is a random tie-breaker, and \(n\) is the number of past observations.

PARAMETER	DESCRIPTION
`current_conformity_score`	Conformity score of the current observation. TYPE: `float`
`conformity_score_history`	Array of past conformity scores used as reference. TYPE: `NDArray`

RETURNS	DESCRIPTION
`float`	Conformal p-value in `[0, 1]` associated with the current score. Under the null hypothesis of exchangeability, this p-value is uniformly distributed on `[0, 1]`.

Notes

When no past observations are available, a uniform random p-value is returned. Tie-breaking via random uniform sampling ensures valid p-values even when conformity scores have ties.

References

.. [1] Angelopoulos, Barber, Bates (2026). "Theoretical Foundations of Conformal Prediction". arXiv preprint arXiv:2411.11824. .. [3] Fedorova, Gammerman, Nouretdinov, Vovk (2012). "Plug-in Martingales for Testing Exchangeability on-line". In Proceedings of the 29^th ICML. Algorithm 1, page 3.

Source code in mapie/exchangeability_testing/martingales.py

def compute_p_value(
    self,
    current_conformity_score: float,
    conformity_score_history: NDArray,
) -> float:
    r"""
    Compute the conformal p-value associated with a new conformity score.

    The p-value is computed using only the past conformity scores, according
    to the empirical conformal formula:

    .. math::

        p_t = \frac{1 + \#\{i : s_i > s_t\} + U \cdot \#\{i : s_i = s_t\}}{n + 1}

    where $s_t$ is the current conformity score, $s_i$ are past
    scores, $U \sim \mathrm{Uniform}(0, 1)$ is a random tie-breaker, and
    $n$ is the number of past observations.

    Parameters
    ----------
    current_conformity_score : float
        Conformity score of the current observation.

    conformity_score_history : NDArray
        Array of past conformity scores used as reference.

    Returns
    -------
    float
        Conformal p-value in ``[0, 1]`` associated with the current score.
        Under the null hypothesis of exchangeability, this p-value is uniformly
        distributed on ``[0, 1]``.

    Notes
    -----
    When no past observations are available, a uniform random p-value is returned.
    Tie-breaking via random uniform sampling ensures valid p-values even when
    conformity scores have ties.

    References
    ----------
    .. [1] Angelopoulos, Barber, Bates (2026).
        "Theoretical Foundations of Conformal Prediction".
        arXiv preprint arXiv:2411.11824.
    .. [3] Fedorova, Gammerman, Nouretdinov, Vovk (2012).
        "Plug-in Martingales for Testing Exchangeability on-line".
        In Proceedings of the 29th ICML. Algorithm 1, page 3.
    """
    history = np.asarray(conformity_score_history, dtype=float)
    n = len(history)
    u = self.rng.uniform()

    if n == 0:
        return float(self.rng.uniform())

    n_greater: int = int(np.sum(history > current_conformity_score))
    n_equal: int = int(np.sum(history == current_conformity_score))

    return float((1.0 + n_greater + u * n_equal) / (n + 1.0))

update_simple_jumper_martingale ¶

update_simple_jumper_martingale(pvalue: float) -> float

Update the simple jumper martingale with a new p-value.

The simple jumper martingale maintains a mixture of betting experts and updates their wealth sequentially according to the incoming p-values.

PARAMETER	DESCRIPTION
`pvalue`	New conformal p-value in `[0, 1]`. TYPE: `float`

RETURNS	DESCRIPTION
`float`	Updated martingale value.

RAISES	DESCRIPTION
`ValueError`	If `pvalue` does not lie in `[0, 1]`.

Notes

This martingale is generally more stable and less tuning-sensitive than the plug-in martingale, making it a suitable default choice in practice.

References

.. [2] Vovk, Gammerman, Shafer (2005). "Algorithmic Learning in a Random World". Boston, MA: Springer US. Section 7.1, page 169.

Source code in mapie/exchangeability_testing/martingales.py

def update_simple_jumper_martingale(self, pvalue: float) -> float:
    """
    Update the simple jumper martingale with a new p-value.

    The simple jumper martingale maintains a mixture of betting experts and
    updates their wealth sequentially according to the incoming p-values.

    Parameters
    ----------
    pvalue : float
        New conformal p-value in ``[0, 1]``.

    Returns
    -------
    float
        Updated martingale value.

    Raises
    ------
    ValueError
        If ``pvalue`` does not lie in ``[0, 1]``.

    Notes
    -----
    This martingale is generally more stable and less tuning-sensitive than the
    plug-in martingale, making it a suitable default choice in practice.

    References
    ----------
    .. [2] Vovk, Gammerman, Shafer (2005).
        "Algorithmic Learning in a Random World".
        Boston, MA: Springer US. Section 7.1, page 169.
    """
    if not (0.0 <= pvalue <= 1.0):
        raise ValueError("pvalue must lie in [0, 1].")

    m_prev = float(np.sum(self._jumper_wealth_by_expert))

    mixed_wealth = (1.0 - self.jump_size) * self._jumper_wealth_by_expert + (
        self.jump_size / 3.0
    ) * m_prev

    betting_multipliers = 1.0 + self._jumper_expert_grid * (pvalue - 0.5)
    self._jumper_wealth_by_expert = mixed_wealth * betting_multipliers

    self.current_martingale_value = float(np.sum(self._jumper_wealth_by_expert))
    self.martingale_value_history.append(self.current_martingale_value)

    return self.current_martingale_value

update_plugin_martingale ¶

update_plugin_martingale(pvalue: float) -> float

Update the plug-in martingale with a new p-value.

The plug-in martingale multiplies the current martingale value by an estimate of the p-value density evaluated at the new p-value.

PARAMETER	DESCRIPTION
`pvalue`	New conformal p-value in `[0, 1]`. TYPE: `float`

RETURNS	DESCRIPTION
`float`	Updated martingale value.

RAISES	DESCRIPTION
`ValueError`	If `pvalue` does not lie in `[0, 1]`.

Notes

The plug-in martingale can be more adaptive than the jumper martingale, but is also more sensitive to density estimation choices and warm-up size.

References

.. [3] Fedorova, Gammerman, Nouretdinov, Vovk (2012). "Plug-in Martingales for Testing Exchangeability on-line". In Proceedings of the 29^th ICML. Algorithm 1, page 3.

Source code in mapie/exchangeability_testing/martingales.py

def update_plugin_martingale(self, pvalue: float) -> float:
    """
    Update the plug-in martingale with a new p-value.

    The plug-in martingale multiplies the current martingale value by an estimate
    of the p-value density evaluated at the new p-value.

    Parameters
    ----------
    pvalue : float
        New conformal p-value in ``[0, 1]``.

    Returns
    -------
    float
        Updated martingale value.

    Raises
    ------
    ValueError
        If ``pvalue`` does not lie in ``[0, 1]``.

    Notes
    -----
    The plug-in martingale can be more adaptive than the jumper martingale,
    but is also more sensitive to density estimation choices and warm-up size.

    References
    ----------
    .. [3] Fedorova, Gammerman, Nouretdinov, Vovk (2012).
        "Plug-in Martingales for Testing Exchangeability on-line".
        In Proceedings of the 29th ICML. Algorithm 1, page 3.
    """
    rho_hat = self._estimate_pvalues_density(pvalue)
    self.current_martingale_value *= rho_hat
    self.martingale_value_history.append(self.current_martingale_value)
    return self.current_martingale_value

update ¶

update(X: NDArray, y: NDArray) -> OnlineMartingaleTest

Update the online martingale test with newly labeled observations.

This method computes conformity scores from the provided features and labels using the MAPIE conformalizer, converts them into conformal p-values using past history, updates the selected martingale, and appends the new observations to the internal state.

PARAMETER	DESCRIPTION
`X`	Feature matrix associated with the new observations. TYPE: `NDArray`
`y`	True labels associated with the new observations. TYPE: `NDArray`

RETURNS	DESCRIPTION
`OnlineMartingaleTest`	Updated instance.

WARNS	DESCRIPTION
`UserWarning`	If exchangeability is rejected and `warn=True`.

Notes

This method can be used both to initialize the test on a labeled reference set and to update it online as new labels become available.

Source code in mapie/exchangeability_testing/martingales.py

def update(
    self,
    X: NDArray,
    y: NDArray,
) -> OnlineMartingaleTest:
    """
    Update the online martingale test with newly labeled observations.

    This method computes conformity scores from the provided features and
    labels using the MAPIE conformalizer, converts them into conformal
    p-values using past history, updates the selected martingale, and
    appends the new observations to the internal state.

    Parameters
    ----------
    X : NDArray
        Feature matrix associated with the new observations.

    y : NDArray
        True labels associated with the new observations.

    Returns
    -------
    OnlineMartingaleTest
        Updated instance.

    Warns
    -----
    UserWarning
        If exchangeability is rejected and ``warn=True``.

    Notes
    -----
    This method can be used both to initialize the test on a labeled reference
    set and to update it online as new labels become available.
    """
    y = self._to_1d_array(y)
    X = np.asarray(X)
    if X.ndim == 1:
        X = X.reshape(-1, 1)

    if X.shape[0] != y.shape[0]:
        raise ValueError(
            "X and y must have the same number of rows. "
            f"Got X.shape[0]={X.shape[0]} and y.shape[0]={y.shape[0]}."
        )

    scores = self._compute_non_conformity_scores(X, y)
    scores = self._to_1d_array(scores).astype(float)

    for current_score in scores:
        pvalue = self.compute_p_value(
            current_conformity_score=current_score,
            conformity_score_history=np.asarray(
                self.conformity_score_history, dtype=float
            ),
        )

        if self.test_method == "jumper_martingale":
            self.update_simple_jumper_martingale(pvalue)
        elif self.test_method == "plugin_martingale":
            self.update_plugin_martingale(pvalue)
        else:
            raise ValueError(f"Unsupported test method: {self.test_method}")

        self.conformity_score_history.append(float(current_score))
        self.pvalue_history.append(float(pvalue))

    if (
        self.is_exchangeable is False
        and self.warn
        and not self._warning_already_raised
    ):
        n_crossings = int(
            np.sum(
                np.asarray(self.martingale_value_history) > self.reject_threshold
            )
        )
        warnings.warn(
            "The online martingale test has rejected exchangeability. "
            f"The martingale exceeded the rejection threshold "
            f"{n_crossings} time(s), with threshold = {self.reject_threshold:.3g}.",
            UserWarning,
        )
        self._warning_already_raised = True

    return self

summary ¶

summary() -> dict

Summarize the current state of the online martingale test.

RETURNS	DESCRIPTION
`dict`	Dictionary containing the current martingale value, exchangeability decision, rejection threshold, summary statistics of the martingale trajectory, and stopping-time information.

Notes

The returned summary is intended for diagnostics and monitoring. It does not modify the internal state of the test.

The reported stopping_time is the first index at which the martingale exceeds the rejection threshold. If the martingale never exceeds the threshold, stopping_time is the index of the last martingale value.

Source code in mapie/exchangeability_testing/martingales.py

def summary(self) -> dict:
    """
    Summarize the current state of the online martingale test.

    Returns
    -------
    dict
        Dictionary containing the current martingale value, exchangeability
        decision, rejection threshold, summary statistics of the martingale
        trajectory, and stopping-time information.

    Notes
    -----
    The returned summary is intended for diagnostics and monitoring.
    It does not modify the internal state of the test.

    The reported ``stopping_time`` is the first index at which the martingale
    exceeds the rejection threshold. If the martingale never exceeds the
    threshold, stopping_time is the index of the last martingale value.
    """
    martingale_values = np.asarray(self.martingale_value_history, dtype=float)

    if martingale_values.size == 0:
        return {
            "test_method": self.test_method,
            "burn_in": self.burn_in,
            "test_level": self.test_level,
            "is_exchangeable": self.is_exchangeable,
            "stopping_time": None,
            "martingale_value_at_decision": None,
            "last_martingale_value": float(self.current_martingale_value),
            "martingale_statistics": {
                "min": None,
                "q025": None,
                "q25": None,
                "median": None,
                "mean": None,
                "q75": None,
                "q975": None,
                "max": None,
            },
        }

    quantiles = np.asarray(
        np.quantile(
            martingale_values,
            [0.0, 0.025, 0.25, 0.5, 0.75, 0.975, 1.0],
        ),
        dtype=float,
    )

    above_threshold = martingale_values > self.reject_threshold

    # Find the first index where a value exceeds the rejection threshold
    threshold_crossing_indices = np.flatnonzero(above_threshold)

    if threshold_crossing_indices.size > 0:
        stopping_time = int(threshold_crossing_indices[0]) + 1
    else:
        stopping_time = int(martingale_values.size)

    return {
        "test_method": self.test_method,
        "burn_in": self.burn_in,
        "test_level": self.test_level,
        "is_exchangeable": self.is_exchangeable,
        "stopping_time": stopping_time,
        "martingale_value_at_decision": float(martingale_values[stopping_time - 1]),
        "last_martingale_value": float(self.current_martingale_value),
        "martingale_statistics": {
            "min": float(quantiles[0]),
            "q025": float(quantiles[1]),
            "q25": float(quantiles[2]),
            "median": float(quantiles[3]),
            "mean": float(np.mean(martingale_values)),
            "q75": float(quantiles[4]),
            "q975": float(quantiles[5]),
            "max": float(quantiles[6]),
        },
    }

mapie.exchangeability_testing.PValuePermutationTest ¶

PValuePermutationTest(
    test_level: float = 0.05,
    mapie_estimator: Optional[MapieEstimator] = None,
    task: Optional[
        Literal["classification", "regression"]
    ] = None,
    random_state: Optional[int] = None,
    num_permutations: int = 1000,
    warn: bool = True,
)

Bases: PermutationTest

Permutation test based on p-values computed from conformity scores.

PARAMETER	DESCRIPTION
`test_level`	Level used to test the hypothesis that the dataset is exchangeable. The probability that the test gives a false positive is at most `test_level` (type I error). TYPE: `float` DEFAULT: `0.05`
`mapie_estimator`	MAPIE estimator used to compute predictions and non-conformity scores. Supported estimators are `SplitConformalClassifier`, and `SplitConformalRegressor`. If `None`, a default `SplitConformalClassifier` or `SplitConformalRegressor` is built when needed. If the estimator is not fitted or not provided, it will be fitted on a slice of the data in order to compute non-conformity scores. TYPE: `Optional[MapieEstimator]` DEFAULT: `None`
`task`	Task type. If `None`, the task is inferred from `y`. TYPE: `Optional[Literal['classification', 'regression']]` DEFAULT: `None`
`random_state`	Seed controlling the randomness of permutations. TYPE: `Optional[int]` DEFAULT: `None`
`num_permutations`	Number of permutations used to estimate the p-value. TYPE: `int` DEFAULT: `1000`
`warn`	Whether to raise a warning when the exchangeability test fails at the end of `run`. TYPE: `bool` DEFAULT: `True`

Examples:

>>> import numpy as np
>>> from mapie.exchangeability_testing.permutations import (
...     PValuePermutationTest,
... )
>>> X = np.arange(100, dtype=float).reshape(-1, 1)
>>> y = 2 * X.ravel() + np.array([0.0, 0.1] * 50)
>>> test = PValuePermutationTest(
...     test_level=0.2,
... )
>>> test.run(X, y).is_exchangeable
True

Source code in mapie/exchangeability_testing/permutations.py

def __init__(
    self,
    test_level: float = 0.05,
    mapie_estimator: Optional[MapieEstimator] = None,
    task: Optional[Literal["classification", "regression"]] = None,
    random_state: Optional[int] = None,
    num_permutations: int = 1000,
    warn: bool = True,
) -> None:
    super().__init__(
        test_level=test_level,
        mapie_estimator=mapie_estimator,
        task=task,
        random_state=random_state,
        num_permutations=num_permutations,
        warn=warn,
    )

run ¶

run(X: NDArray, y: NDArray) -> PValuePermutationTest

Run a p-value permutation test.

PARAMETER	DESCRIPTION
`X`	Feature matrix. TYPE: `NDArray`
`y`	Target values. TYPE: `NDArray`

RETURNS	DESCRIPTION
`PValuePermutationTest`	Updated instance.

Source code in mapie/exchangeability_testing/permutations.py

def run(self, X: NDArray, y: NDArray) -> "PValuePermutationTest":
    """Run a p-value permutation test.

    Parameters
    ----------
    X : NDArray
        Feature matrix.
    y : NDArray
        Target values.
    Returns
    -------
    PValuePermutationTest
        Updated instance.
    """
    scores = self._compute_non_conformity_scores(X, y)

    test_statistic_reference = self.test_statistic(scores)

    rank = 1
    self.p_values = np.empty(self.num_permutations + 1)
    self.p_values[0] = 1.0
    n = len(scores)
    for t in range(1, self.num_permutations + 1):
        permuted = self.rng.permutation(n)
        scores_permuted = scores[permuted]
        test_statistic_permutation = self.test_statistic(scores_permuted)

        if test_statistic_permutation >= test_statistic_reference:
            rank += 1
        self.p_values[t] = rank / (t + 1)

    self._warn_if_not_exchangeable()
    return self

mapie.exchangeability_testing.PermutationTest ¶

PermutationTest(
    test_level: float = 0.05,
    mapie_estimator: Optional[MapieEstimator] = None,
    task: Optional[
        Literal["classification", "regression"]
    ] = None,
    random_state: Optional[int] = None,
    num_permutations: int = 1000,
    warn: bool = True,
)

Bases: ABC

Base class for exchangeability tests based on permutations.

PARAMETER	DESCRIPTION
`test_level`	Level used to test the hypothesis that the dataset is exchangeable. The probability that the test gives a false positive is at most `test_level` (type I error). TYPE: `float` DEFAULT: `0.05`
`mapie_estimator`	MAPIE estimator used to compute predictions and non-conformity scores. Supported estimators are `SplitConformalClassifier`, and `SplitConformalRegressor`. If `None`, a default `SplitConformalClassifier` or `SplitConformalRegressor` is built when needed. If the estimator is not fitted or not provided, it will be fitted on a slice of the data in order to compute non-conformity scores. TYPE: `Optional[MapieEstimator]` DEFAULT: `None`
`random_state`	Seed controlling the randomness of permutations. TYPE: `Optional[int]` DEFAULT: `None`
`num_permutations`	Number of permutations used by permutation-based tests. TYPE: `int` DEFAULT: `1000`
`warn`	Whether to raise a warning when the exchangeability test fails at the end of `run`. TYPE: `bool` DEFAULT: `True`

Source code in mapie/exchangeability_testing/permutations.py

def __init__(
    self,
    test_level: float = 0.05,
    mapie_estimator: Optional[MapieEstimator] = None,
    task: Optional[Literal["classification", "regression"]] = None,
    random_state: Optional[int] = None,
    num_permutations: int = 1000,
    warn: bool = True,
) -> None:
    if not (0.0 < test_level < 1.0):
        raise ValueError("test_level must be in (0, 1).")
    if num_permutations < 1:
        raise ValueError("num_permutations must be greater than or equal to 1.")
    self.test_level = test_level
    self.mapie_estimator = self._prepare_estimator(mapie_estimator)
    self.task = task
    self.rng = np.random.RandomState(random_state)
    self.num_permutations = num_permutations
    self.warn = warn
    self.p_values: NDArray = np.array([])
    self.test_statistic = MaxSplitMeanDifferenceTestStatistic()

is_exchangeable `property` ¶

is_exchangeable: Optional[bool]

Return the latest exchangeability decision.

RETURNS	DESCRIPTION
`Optional[bool]`	`None` if the test has not been run yet, otherwise whether the dataset is deemed exchangeable based on the last p-value.

run `abstractmethod` ¶

run(X: NDArray, y: NDArray) -> PermutationTest

Run a permutation-based exchangeability test.

Source code in mapie/exchangeability_testing/permutations.py

@abstractmethod
def run(self, X: NDArray, y: NDArray) -> "PermutationTest":
    """Run a permutation-based exchangeability test."""
    raise NotImplementedError  # pragma: no cover

mapie.exchangeability_testing.SequentialMonteCarloTest ¶

SequentialMonteCarloTest(
    strategy: Literal[
        "aggressive", "binomial", "binomial_mixture"
    ],
    test_level: float = 0.05,
    mapie_estimator: Optional[MapieEstimator] = None,
    task: Optional[
        Literal["classification", "regression"]
    ] = None,
    random_state: Optional[int] = None,
    num_permutations: int = 1000,
    warn: bool = True,
    burn_in: int = 100,
)

Bases: PermutationTest

Sequential Monte Carlo exchangeability test.

PARAMETER	DESCRIPTION
`strategy`	Wealth update strategy for the sequential test. TYPE: `(aggressive, binomial, binomial_mixture)` DEFAULT: `"aggressive"`
`test_level`	Level used to test the hypothesis that the dataset is exchangeable. The probability that the test gives a false positive is at most `test_level` (type I error). TYPE: `float` DEFAULT: `0.05`
`mapie_estimator`	MAPIE estimator used to compute predictions and non-conformity scores. Supported estimators are `SplitConformalClassifier`, and `SplitConformalRegressor`. If `None`, a default `SplitConformalClassifier` or `SplitConformalRegressor` is built when needed. If the estimator is not fitted or not provided, it will be fitted on a slice of the data in order to compute non-conformity scores. TYPE: `Optional[MapieEstimator]` DEFAULT: `None`
`task`	Task type. If `None`, the task is inferred from `y`. TYPE: `Optional[Literal['classification', 'regression']]` DEFAULT: `None`
`random_state`	Seed controlling the randomness of permutations. TYPE: `Optional[int]` DEFAULT: `None`
`num_permutations`	Maximum number of permutations. TYPE: `int` DEFAULT: `1000`
`warn`	Whether to raise a warning when the exchangeability test fails at the end of `run`. TYPE: `bool` DEFAULT: `True`
`burn_in`	Minimum number of permutations before considering early stopping. TYPE: `int` DEFAULT: `100`

Source code in mapie/exchangeability_testing/permutations.py

def __init__(
    self,
    strategy: Literal["aggressive", "binomial", "binomial_mixture"],
    test_level: float = 0.05,
    mapie_estimator: Optional[MapieEstimator] = None,
    task: Optional[Literal["classification", "regression"]] = None,
    random_state: Optional[int] = None,
    num_permutations: int = 1000,
    warn: bool = True,
    burn_in: int = 100,
) -> None:
    super().__init__(
        test_level=test_level,
        mapie_estimator=mapie_estimator,
        task=task,
        random_state=random_state,
        num_permutations=num_permutations,
        warn=warn,
    )
    self.strategy = strategy
    self.burn_in = burn_in

    valid_strategies = {"aggressive", "binomial", "binomial_mixture"}
    if self.strategy not in valid_strategies:
        raise ValueError(
            f"Unknown strategy '{self.strategy}'. Expected one of {valid_strategies}."
        )

run ¶

run(X: NDArray, y: NDArray) -> SequentialMonteCarloTest

Run a sequential Monte Carlo permutation test.

PARAMETER	DESCRIPTION
`X`	Feature matrix. TYPE: `NDArray`
`y`	Target values. TYPE: `NDArray`

RETURNS	DESCRIPTION
`SequentialMonteCarloTest`	Updated instance.

Source code in mapie/exchangeability_testing/permutations.py

def run(self, X: NDArray, y: NDArray) -> "SequentialMonteCarloTest":
    """Run a sequential Monte Carlo permutation test.

    Parameters
    ----------
    X : NDArray
        Feature matrix.
    y : NDArray
        Target values.
    Returns
    -------
    SequentialMonteCarloTest
        Updated instance.
    """
    scores = self._compute_non_conformity_scores(X, y)

    test_statistic_reference = self.test_statistic(scores)

    c = self.test_level * 0.90
    p_zero = 1 / np.ceil(np.sqrt(2 * np.pi * np.exp(1 / 6)) / self.test_level)

    rank = 1
    wealth_bin = np.array([1.0])
    wealth_agg = np.array([1.0])
    wealth_bm = np.array([1.0])
    n = len(scores)
    for i in range(1, self.num_permutations + 1):  # pragma: no branch
        permuted = self.rng.permutation(n)
        scores_permuted = scores[permuted]
        test_statistic_permutation = self.test_statistic(scores_permuted)

        if wealth_bin[-1] * p_zero * (i + 1) / rank < self.test_level:
            pt = 0
        else:
            pt = p_zero

        if test_statistic_permutation >= test_statistic_reference:
            bet_bin_i = pt * (i + 1) / rank
            bet_agg_i = 0.0
            rank += 1
        else:
            bet_bin_i = (1 - pt) * (i + 1) / (i - (rank - 1))
            bet_agg_i = (i + 1) / i
        wealth_bm_i = (1 - binom.cdf(rank - 1, i + 1, c)) / c

        wealth_bin = np.append(wealth_bin, wealth_bin[-1] * bet_bin_i)
        wealth_agg = np.append(wealth_agg, wealth_agg[-1] * bet_agg_i)
        wealth_bm = np.append(wealth_bm, wealth_bm_i)

        # early stopping if possible
        strategy_to_current_wealth = {
            "binomial": wealth_bin[-1],
            "aggressive": wealth_agg[-1],
            "binomial_mixture": wealth_bm[-1],
        }
        current_wealth = strategy_to_current_wealth[self.strategy]
        if (
            current_wealth < self.test_level
            or current_wealth >= 1 / self.test_level
        ) and i > self.burn_in:
            break

    strategy_to_wealth = {
        "binomial": wealth_bin,
        "aggressive": wealth_agg,
        "binomial_mixture": wealth_bm,
    }
    wealth_history = strategy_to_wealth[self.strategy]
    running_max_wealth = np.maximum.accumulate(wealth_history)
    self.p_values = np.minimum(1 / running_max_wealth, 1)

    self._warn_if_not_exchangeable()
    return self

Exchangeability Testing¶

High-level Interfaces¶

mapie.exchangeability_testing.FixedDatasetExchangeabilityTest ¶

is_exchangeable property ¶

run ¶

mapie.exchangeability_testing.OnlineExchangeabilityTest ¶

is_exchangeable property ¶

update ¶

mapie.exchangeability_testing.RiskMonitoring ¶

harmful_shift_detected property ¶

compute_threshold ¶

update ¶

summary ¶

Individual Tests¶

mapie.exchangeability_testing.OnlineMartingaleTest ¶

reject_threshold property ¶

is_exchangeable property ¶

compute_p_value ¶

update_simple_jumper_martingale ¶

update_plugin_martingale ¶

update ¶

summary ¶

mapie.exchangeability_testing.PValuePermutationTest ¶

run ¶

mapie.exchangeability_testing.PermutationTest ¶

is_exchangeable property ¶

run abstractmethod ¶

mapie.exchangeability_testing.SequentialMonteCarloTest ¶

run ¶

is_exchangeable `property` ¶

is_exchangeable `property` ¶

harmful_shift_detected `property` ¶

reject_threshold `property` ¶

is_exchangeable `property` ¶

is_exchangeable `property` ¶

run `abstractmethod` ¶