The MuLaConf Package#

Package Wrapper#

class mulaconf.icp_wrapper.ICPWrapper(classification_strategy, measure: str = 'mahalanobis', weight_hamming: float = 0.0, weight_cardinality: float = 0.0, device: str | torch.device = 'cpu')[source]#

A wrapper for Inductive Conformal Prediction with Structural Penalties (Scikit-Learn compatible).

This class manages the lifecycle of the underlying multi-label classifier and the conformal predictor. It handles model training, calibration, and the efficient integration of the PyTorch mathematical engine without unnecessary retraining.

Note

Switching Strategies: You can switch the classification strategy or update its parameters. If the wrapper detects a change (via fingerprinting) during calibrate, it will automatically retrain the new model on the cached proper-training data.

Note

On-the-Fly Updates: The wrapper itself acts strictly as a bridge. If you want to perform lazy-evaluation updates on the distance measure or penalty weights without passing calibration data again, you can do so directly via wrapper (e.g., wrapper.measure='norm', wrapper.weight_hamming=1.0 and wrapper.weight_cardinality=0.5).

Parameters:

classification_strategy (sklearn.base.BaseEstimator) – The underlying multi-label classification model (e.g., RandomForest, ClassifierChain). Must support fit and predict_proba.
measure (str, optional, default='mahalanobis') – The distance metric used to score predictions. Supported options: 'mahalanobis' (accounts for correlations) or 'norm' (standard Euclidean).
weight_hamming (float, optional, default=0.0) – Initial weight for the Hamming distance penalty.
weight_cardinality (float, optional, default=0.0) – Initial weight for the Cardinality penalty.
device (str or torch.device, optional, default='cpu') – The device to use for tensor computations ('cpu' or 'cuda').

Fits the underlying multi-label classification model.

This method trains the classification_strategy on the provided features and labels, and caches the training data to enable auto-retraining if hyper-parameters change.

Parameters:

train_features (array-like) – The training features. Shape: (n_samples, w_features).
train_labels (array-like) – The training labels (binary multi-hot). Shape: (n_samples, c_classes).
**kwargs (dict) – Optional arguments to pass to the classifier’s parameters or fit method.

Returns:

self – The fitted wrapper instance.

Return type:

object

Example

>>> import numpy as np
>>> from sklearn.ensemble import RandomForestClassifier
>>> from sklearn.multioutput import MultiOutputClassifier
>>>
>>> # 1. Generate Dummy Training Data
>>> # 100 samples, 5 features, 3 target labels
>>> X_train = np.random.rand(100, 5)
>>> y_train = np.random.randint(0, 2, (100, 3))
>>>
>>> # 2. Initialize Wrapper
>>> base_model = MultiOutputClassifier(RandomForestClassifier())
>>> wrapper = ICPWrapper(base_model)
>>>
>>> # 3. Fit the Model (Standard)
>>> wrapper.fit(X_train, y_train)

Note

Change hyperparameters : You can update the classifier’s hyperparameters during the fit call. Note the estimator__ prefix for wrapped sklearn models.

>>> args = {'estimator__n_neighbors': 5}
>>> wrapper.fit(X_train, y_train, **args)

Calibrates the conformal predictor using a dedicated calibration set.

This step calculates the nonconformity scores and determines the thresholds required to guarantee coverage.

Note

If called without arguments after an initial calibration, it will manually apply any pending updates (measure or penalty weights) and recalibrate using the cached data.

Parameters:

calib_features (array-like) – Features of the calibration set. Shape: (q_samples, w_features).
calib_labels (array-like) – Labels of the calibration set. Shape: (q_samples, c_classes).

Returns:

self – The calibrated wrapper instance.

Return type:

object

Raises:

RuntimeError – If calibration features and labels are not provided.
RuntimeError – If fit() has not been called before running calibration.
RuntimeError – If retraining the underlying classifier fails.

Example

>>> # 1. Initialize & Fit (See `fit()` function documentation for details)
>>>
>>> # 2. Generate Dummy Calibration Data
>>> # 100 samples, 5 features, 3 target labels
>>> X_calib = np.random.rand(100, 5)
>>> y_calib = np.random.randint(0, 2, (100, 3))
>>>
>>> # Calibrate
>>> wrapper.calibrate(X_calib, y_calib)

Note

Strategy Switching: Change the underlying model, retrain and recalibrate automatically.

>>> from sklearn.neighbors import KNeighborsClassifier
>>> from sklearn.multioutput import ClassifierChain
>>>
>>> wrapper.strategy = ClassifierChain(KNeighborsClassifier(n_neighbors=5))
>>>
>>> # Calling `calibrate()` again detects the change and retrains automatically
>>> wrapper.calibrate(X_calib, y_calib)

Note

On-the-fly Updates: You can easily update the distance measure and penalty weights after the calibration process without passing your data again. Calling calibrate() without arguments will automatically apply all pending updates simultaneously using the cached calibration data.

>>> # Optional: The `calibrate()` method recalculates the covariance matrix and scores after a measure update.
>>> wrapper.measure = 'norm'
>>> wrapper.calibrate()

>>> # Optional: The `calibrate()` method recalculates calibration scores after penalty weight update.
>>> wrapper.weight_hamming = 1.0
>>> wrapper.weight_cardinality = 0.5
>>> wrapper.calibrate()

>>> # Optional: The `calibrate()` method applies both distance measure and penalty weight updates at once.
>>> wrapper.measure = 'norm'
>>> wrapper.weight_hamming = 1.0
>>> wrapper.weight_cardinality = 0.5
>>> wrapper.calibrate()

Generates conformal prediction regions for the input features.

This method calculates p-values for all test samples based on the calibrated scores.

Parameters:

test_features (array-like) – The test features. Shape: (t_samples, w_features).
non_empty_prediction_regions (bool, optional) – If True (default), the combination with the highest p-value is returned to ensure non-empty predictions.

Returns:

A callable object that returns prediction regions. You must call this object with a specific significance_level to retrieve the final prediction sets.

Return type:

PredictionRegions

Raises:

RuntimeError – If calibrate() has not been called, and the ICP engine does not exist.
RuntimeError – If the classifier has not been fitted yet (fit() must be called first).
RuntimeError – If the classifier model changed. Run the fit and calibration procedure.

Example

>>> # ... Assume wrapper is already fitted and calibrated (see fit() for details) ...
>>> X_test = np.random.rand(10, 5)
>>>
>>> # 1. Get prediction regions object with non empty prediction regions
>>> prediction_region_obj = wrapper.predict(X_test)
>>>
>>> # 2. Extract Prediction Sets (e.g., at 10% significance / 90% confidence)
>>> # Returns a list of Tensors, where each Tensor contains the indices of predicted labels.
>>> prediction_sets = prediction_region_obj(significance_level=0.1)

Note

Equivalent Syntax: Because the predict method returns a callable PredictionRegions object, you can chain the operations to evaluate the test features and extract prediction sets in a single line of code:

>>> prediction_sets = wrapper.predict(test_features)(significance_level=0.1)

Note

On-the-fly Update: You can update the distance measure and penalty weights on the fly and predict again immediately.

>>> wrapper.measure = 'norm'
>>> wrapper.weight_hamming = 2.0
>>> wrapper.weight_cardinality = 1.5
>>> updated_obj = wrapper.predict(X_test)
>>> updated_sets = updated_obj(significance_level=0.1)

Inductive Conformal Predictor#

Inductive Conformal Predictor with Structural Penalties.

This class implements Inductive Conformal Prediction (ICP) for multi-label classification, extended with structural penalties (Hamming and Cardinality). It uses a generalized distance metric (e.g., Mahalanobis or Euclidean Norm) in the error vector space to score predictions. Additionally, it allows for on-the-fly updating of the distance measure and penalty weights without retraining the underlying model or requiring the calibration data to be passed again.

Note

The predictor calculates and caches the structural penalty vectors for all possible label combinations during initialization.

Note

On-the-Fly Updates: You can update the distance measure ('norm' or 'mahalanobis'), weight_hamming, and weight_cardinality at any time after the calibration process.

The predictor utilizes lazy evaluation for automatic recalibration. This means you do not need to manually pass your calibration data again or explicitly call calibrate(). Simply assign new values to the properties (e.g., icp.measure = 'norm', icp.weight_hamming = 1.0, icp.weight_cardinality = 0.5) and immediately call predict(). It will automatically reform the underlying covariance matrix and recalibrate the scores on the fly before generating predictions.

Note

Memory Management: This class uses optimized batching, tensor expansion (compressed loading) to prevent GPU/CPU memory overflow when processing exponential powerset combinations. Even with these optimizations, Powerset Scoring prediction scales at O(2^C), where C is the number of labels. For standard systems with 16GB of RAM/VRAM, we recommend limiting tasks to a maximum of ~20 labels.

The default batching limits are approximated based on PyTorch’s float32 data type (which consumes 4 bytes per element) and the overhead required to hold multiple massive intermediate tensors simultaneously in memory during calculation.

Users can manually tune the hardware limits by modifying the module-level configuration variables (located in constants.py) to optimize for their specific CPU/GPU memory constraints:

_CPU_MAX_COMBINATIONS: Caps the maximum number of combinations processed at once during heavy matrix math on the CPU to protect system RAM (default: 2,000,000).

_GPU_MAX_COMBINATIONS: Caps the maximum number of combinations processed at once during heavy matrix math on the GPU to protect VRAM (default: 5,000,000).

_REGION_BATCH_SIZE: Caps memory usage (System RAM) when extracting the final prediction sets. Because this phase relies on lightweight boolean filtering rather than heavy matrix operations, this threshold is safely set much higher (default: 100,000,000).

_EMPTY_CUDA_CACHE: Controls whether the engine aggressively clears the CUDA memory cache after heavy tensor operations. Keep this set to True (default) to prevent VRAM fragmentation and Out-Of-Memory crashes. Set to False only for strict performance benchmarking to bypass the ~300ms OS-level synchronization delay.

Increasing these values speeds up the time required to generate predictions, but risks memory overflow and system instability. Decreasing these values results in slower predictions but guarantees safety from system crashes. In the documentation of constants.py, you can find a hardware cheat sheet for memory requirements.

Parameters:

measure (str, optional, default='mahalanobis') – The distance metric used to score predictions. Supported options: ‘mahalanobis’ (accounts for correlations) or ‘norm’ (standard Euclidean).
predicted_probabilities (Union[torch.Tensor, np.ndarray, list, pd.DataFrame, pd.Series]) – The predicted probabilities for the proper training set. Shape: (n_samples, c_classes).
true_labels (Union[torch.Tensor, np.ndarray, list, pd.DataFrame, pd.Series]) – The ground truth binary labels for the proper training set. Shape: (n_samples, c_classes).
weight_hamming (float, optional, default=0.0) – The weight for the Hamming distance penalty. Higher values penalize predictions that are structurally different from observed training labels.
weight_cardinality (float, optional, default=0.0) – The weight for the Cardinality penalty. Higher values penalize prediction set sizes that are infrequent in the training data.
device (str or torch.device, optional, default='cpu') – The device to use for computations (‘cpu’ or ‘cuda’).

Example

>>> import torch
>>>
>>> # 1. Generate dummy training data (500 samples, 5 classes)
>>> train_probs = torch.rand(500, 5)
>>> train_labels = torch.randint(0, 2, (500, 5)).float()
>>>
>>> # 2. Initialize the predictor
>>> icp = InductiveConformalPredictor(
...     predicted_probabilities=train_probs,
...     true_labels=train_labels,
...     measure='mahalanobis',
...     weight_hamming=2.0,
...     weight_cardinality=1.5,
... )

Calibrates the predictor using a dedicated calibration set.

This method computes nonconformity scores for the calibration data and sorts them to determine thresholds for future predictions.

Note

If called without arguments, it recalculates the calibration scores using the current distance measure and set penalty weights on the existing calibration data.

Parameters:

probabilities (Union[torch.Tensor, np.ndarray, list, pd.DataFrame, pd.Series]) – Predicted probabilities for the calibration set. Shape: (q_samples, c_classes).
labels (Union[torch.Tensor, np.ndarray, list, pd.DataFrame, pd.Series]) – Ground truth labels for the calibration set. Shape: (q_samples, c_classes).

Returns:

self – The initialized and calibrated predictor object.

Return type:

object

Raises:

RuntimeError – If one of probabilities or labels is provided but not the other.
RuntimeError – If the provided calibration set is empty.
RuntimeError – If labels shape does not match the number of classes.
RuntimeError – If probabilities shape does not match the number of classes.

Example

>>> # 1. Generate dummy calibration data (100 samples, 5 classes)
>>> calib_probs = torch.rand(100, 5)
>>> calib_labels = torch.randint(0, 2, (100, 5)).float()
>>>
>>> # 2. Calibrate
>>> icp.calibrate(calib_probs, calib_labels)

Note

Update the distance measure and penalty weights before calling calibrate().

>>> # Optional: The `calibrate()` method recalculates the distance matrix and scores after a measure update.
>>> icp.measure = 'norm'
>>> icp.calibrate()

>>> # Optional: The `calibrate()` method recalculates calibration scores after penalty weight update.
>>> icp.weight_hamming = 1.0
>>> icp.weight_cardinality = 0.5
>>> icp.calibrate()

>>> # Optional: The `calibrate()` method applies both distance measure and penalty weight updates at once.
>>> icp.measure = 'norm'
>>> icp.weight_hamming = 1.0
>>> icp.weight_cardinality = 0.5
>>> icp.calibrate()

predict(probabilities: torch.Tensor | numpy.ndarray | list | pandas.DataFrame | pandas.Series, non_empty_prediction_regions: bool = True) → PredictionRegions#

Computes p-values for the test samples.

This method calculates the p-value for every possible label combination based on the calibrated scores.

Parameters:

probabilities (Union[torch.Tensor, np.ndarray, list, pd.DataFrame, pd.Series]) – Predicted probabilities for the test set. Shape: (t_samples, c_classes).
non_empty_prediction_regions (bool, optional) – If True (default), the combination with the highest p-value is returned to ensure non-empty predictions.

Returns:

A callable object that wraps the p-values and combinations. You must call this object with a significance level to get the actual prediction sets.

Return type:

PredictionRegions

Raises:

RuntimeError – If a distance measure was changed, but no calibration data is cached to perform the auto-recalculation.
RuntimeError – If calibrate() has not been called before predict().
RuntimeError – If probabilities shape does not match the number of classes.

Example

>>> # Generate dummy test probabilities
>>> test_probs = torch.rand(30, 5)
>>>
>>> # Get prediction regions object with non empty prediction regions
>>> prediction_obj = icp.predict(test_probs)
>>>
>>> # Extract prediction sets for significance level 0.1 (90% confidence)
>>> prediction_sets = prediction_obj(significance_level=0.1)

Note

Equivalent Syntax: Because the predictor itself is callable and it returns a callable PredictionRegions object, you can chain the operations to extract prediction sets in a single line of code:

>>> prediction_sets = icp.predict(test_probs)(significance_level=0.1)

Note

Update distance measure and penalty weights on-the-fly and predict again. The predictor will automatically apply the pending updates and recalibrate the scores before generating the new predictions.

>>> icp.measure = 'norm'
>>> icp.weight_hamming = 1.0
>>> icp.weight_cardinality = 0.5
>>> new_prediction_obj = icp.predict(test_probs)
>>> new_prediction_sets = new_prediction_obj(significance_level=0.1)

Prediction Regions#

class mulaconf.prediction_regions.PredictionRegions(p_values: torch.Tensor, combinations: torch.Tensor, non_empty_prediction_regions: bool = True)[source]#

A container class for conformal prediction results.

This class wraps the p-values generated by the Inductive Conformal Predictor and provides methods to extract prediction sets at specific significance levels and evaluate performance metrics.

Parameters:

p_values (torch.Tensor) – The p-values associated with every possible label combination for each test sample. Shape: (t_samples, 2^(c_classes)).
combinations (torch.Tensor) – The matrix of all possible label combinations corresponding to the columns of p_values. Shape: (2^(c_classes), c_classes).
non_empty_prediction_regions (bool, optional) – If True (default), the combination with the highest p-value is returned to ensure non-empty predictions.

Example

>>> import torch
>>> from mulaconf.prediction_regions import PredictionRegions
>>>
>>> # 1. Generate dummy data (100 samples, 5 classes -> 32 combinations)
>>> combinations = torch.cartesian_prod(*[torch.tensor([0, 1])] * 5)
>>> p_values = torch.rand(100, 2**5)
>>>
>>> # 2. Initialize container
>>> prediction_regions_obj = PredictionRegions(p_values, combinations)

Extracts prediction sets for one or multiple significance levels.

Parameters:

significance_level (float or List[float]) – The significance level(s) (alpha) for which to return prediction regions. Must be strictly between 0 and 1.

Returns:

If significance_level is a scalar: Returns a list of valid tuples.
If significance_level is a list: Returns a dictionary mapping alpha values to lists of valid tuples.

Return type:

Union[List[torch.Tensor], Dict[float, List[torch.Tensor]]]

Raises:

ValueError – If significance_level is None or out of range [0, 1].

Example

>>> # Scalar alpha
>>> prediction_sets_lst = prediction_regions_obj(0.1)
>>>
>>> # Multiple alphas
>>> prediction_sets_dict = prediction_regions_obj([0.05, 0.1])

evaluate(return_true_label_p_value: bool = True, return_coverage: bool = True, return_n_criterion: bool = True, return_s_criterion: bool = True, return_observed_fuzziness: bool = True, return_observed_excess: bool = True, *, true_labelsets: numpy.ndarray | torch.Tensor | pandas.DataFrame | pandas.Series | List | None = None, significance_level: float | List[float] | None = None) → Dict | None#

Evaluates the conformal predictor using standard metrics.

Calculates validity (coverage) and efficiency (N-criterion, S-criterion, Observed fuzziness, Observed excess) metrics.

Parameters:

return_true_label_p_value (bool, default=True) – Whether to return the p-values of the true class (requires true labelsets).
return_coverage (bool, default=True) – Whether to calculate empirical coverage (requires true labelsets and significance_level).
return_n_criterion (bool, default=True) – Whether to calculate average set size (N-criterion) (requires true labelsets and significance_level).
return_s_criterion (bool, default=True) – Whether to calculate the S-criterion (sum of p-values), a threshold-independent efficiency metric.
return_observed_fuzziness (bool, default=True) – Whether to calculate the observed fuzziness (requires true labelsets).
return_observed_excess (bool, default=True) – Whether to calculate the observed excess (requires true labelsets and significance_level).
true_labelsets (torch.Tensor or array-like, optional) – The ground truth labels. Required for coverage and p-value metrics.
significance_level (float or List[float], optional) – The alpha level(s) to evaluate coverage and N-criterion.

Note

The true_labelsets and the significance_level must be passed as a keyword argument.

Returns:

A dictionary containing the requested evaluation metrics.

Depending on the boolean flags provided, the returned dictionary will contain the following specific string keys: - 'true_labels_p_values' : A 1D tensor of the exact p-values for the true labels. - 'coverage' : The empirical coverage (validity). - 'n_criterion' : The average prediction set size. - 's_criterion' : The average sum of all p-values. - 'observed_fuzziness' : The average sum of the p-values of the false labels. - 'observed_excess' : The average sum of false labels included in the prediction region.

If significance_level is a list of multiple alphas, it returns a nested dictionary where the top-level keys are the alpha values (floats), and the values are metric dictionaries containing the keys listed above.

Return type:

Dict

Raises:

ValueError – If true_labelsets` and the significance_level is not provided.
ValueError – If true_labelsets shape does not match the number of classes.

Example

>>> # Generate dummy test labels
>>> y_test = torch.rand(10, 5)
>>>
>>> # Evaluate specific alpha
>>> metrics = prediction_regions_obj.evaluate(true_labelsets=y_test, significance_level=0.1)
>>>
>>> # Evaluate multiple alphas
>>> metrics_multi = prediction_regions_obj.evaluate(true_labelsets=y_test, significance_level=[0.05, 0.1])