From 3ecbadb921419d3287fefc3bc95354ef9090dda1 Mon Sep 17 00:00:00 2001 From: Alejandro Moreo Date: Fri, 18 Sep 2026 11:54:50 +0200 Subject: [PATCH] adding liep --- docs/source/manuals/label-shift-adaptation.md | 108 +++++++++ quapy/classification/labelshift.py | 110 +++++++++ quapy/method/_liep.py | 227 ++++++++++++++++++ quapy/tests/test_labelshift.py | 84 +++++++ 4 files changed, 529 insertions(+) create mode 100644 docs/source/manuals/label-shift-adaptation.md create mode 100644 quapy/classification/labelshift.py create mode 100644 quapy/method/_liep.py create mode 100644 quapy/tests/test_labelshift.py diff --git a/docs/source/manuals/label-shift-adaptation.md b/docs/source/manuals/label-shift-adaptation.md new file mode 100644 index 0000000..f71994f --- /dev/null +++ b/docs/source/manuals/label-shift-adaptation.md @@ -0,0 +1,108 @@ +# Label Shift Adaptation + +Quantification methods estimate the class prevalence values of a test sample, but a prevalence estimate is +not, by itself, a classifier that has been corrected for the shift it describes. In some scenarios you may +want the latter: a classifier whose decisions account for the (estimated) change in class priors between +the training and the test distributions, rather than a single prevalence vector. + +QuaPy provides two building blocks for this: + +* `qp.method.aggregative.ImportanceWeightQuantifier`, an interface implemented by the quantifiers that + natively compute a vector of importance weights as part of estimating the target prevalence. +* `qp.classification.labelshift.LabelShiftedClassifier`, a classifier wrapper that uses any quantifier + (whether or not it implements the interface above) to adapt a classifier to a specific, (potentially) + shifted batch of test instances. + +## The `ImportanceWeightQuantifier` interface + +Some quantifiers (currently, `RLLS`, `BBSEhard`, and `BBSEsoft`; see the +{ref}`Regularized Learning under Label Shift (RLLS) ` +and {ref}`Black Box Shift Estimation (BBSE) ` sections of +the methods manual) work by first estimating a vector of importance weights + +:::{math} +w_y = \frac{Q(y)}{P(y)} +::: + +with :math:`P` and :math:`Q` the training and target class distributions, and only then rescaling the +training prevalence by these weights to obtain the target prevalence estimate: +:math:`\hat{Q}(y) \propto w_y \cdot P(y)`. For these methods, the weight vector is not an afterthought +computed from the prevalence estimate; it is the primitive quantity from which the estimate itself is +derived. + +`ImportanceWeightQuantifier` exposes this quantity directly, in addition to the usual `predict` method +inherited from any quantifier: + +* `get_importance_weights(instances)` returns the weight vector $w$ estimated for a given batch of + (unlabelled) target instances. +* `quantify_and_weigh(instances)` returns a tuple `(prevalence, weights)`, computed from a single pass of + classifier predictions over the instances (avoiding the double classification that calling `predict` and + `get_importance_weights` separately would incur). + +Neither method mutates any internal state of the quantifier, so both are safe to call concurrently, on the +same fitted instance, for different batches of target instances. + +```python +import quapy as qp +from quapy.method.aggregative import BBSEhard +from sklearn.linear_model import LogisticRegression + +train, test = qp.datasets.fetch_UCIBinaryDataset('haberman').train_test + +model = BBSEhard(LogisticRegression(max_iter=2000), val_split=5) +model.fit(*train.Xy) + +prevalence, weights = model.quantify_and_weigh(test.X) +``` + +## Adapting a classifier: `LabelShiftedClassifier` + +Rescaling the classifier's posterior probabilities after the fact (as, e.g., `EMQ` does) is one way of +accounting for an estimated prevalence shift. Another common practice is instead to retrain the classifier +with a reweighted loss, via scikit-learn's `class_weight` constructor parameter; this changes the decision +function of the classifier itself, rather than only rescaling its outputs, which can matter for classifiers +whose decision boundary is not a simple function of the posterior probabilities alone. + +`LabelShiftedClassifier` automates this second strategy. Given a base classifier `h` and any `quapy` +quantifier `q`, it fits `q` on the labelled training data (keeping a copy of it), and, for a given batch of +(unlabelled) target instances, estimates the importance weights: directly, via `get_importance_weights`, if +`q` is an `ImportanceWeightQuantifier`; or otherwise by dividing `q`'s estimated target prevalence by the +training prevalence. It then retrains a *fresh copy* of `h`, with `class_weight` set according to these +weights, on the original training data, and hands off classification of the target instances to this +adapted classifier. Neither the classifier instance `h` passed at construction time, nor `q`'s own internal +classifier (if any), are ever mutated. + +```python +import quapy as qp +from quapy.classification.labelshift import LabelShiftedClassifier +from quapy.method.aggregative import ACC +from sklearn.linear_model import LogisticRegression + +train, test = qp.datasets.fetch_UCIBinaryDataset('haberman').train_test + +h = LogisticRegression(max_iter=2000) +q = ACC(LogisticRegression(max_iter=2000), val_split=5) + +adapter = LabelShiftedClassifier(h, q) +adapter.fit(*train.Xy) + +# label predictions and posterior probabilities from a classifier adapted to test's estimated distribution +y_pred = adapter.predict(test.X) +y_proba = adapter.predict_proba(test.X) + +# or, to obtain the adapted classifier itself +adapted_h = adapter.get_classifier(test.X) +``` + +`h` is required to accept a `class_weight` parameter in its constructor, as `LogisticRegression` and many +other scikit-learn classifiers do; `q` can be any (fitted or unfitted) `quapy` quantifier, and need not be an +`ImportanceWeightQuantifier`. + +Since the adaptation is specific to the batch of target instances given, `predict` and `predict_proba` +retrain `h` on every call. If you plan to issue several predictions against the same (suspected) shifted +distribution, call `get_classifier` once and reuse the classifier instance it returns, rather than calling +`predict`/`predict_proba` repeatedly on the same or related samples. + +A small positive floor, controlled by the `weight_epsilon` constructor parameter (default `1e-4`), is +applied to the estimated weights before they are used as `class_weight`; this keeps a class with training +support from being assigned a weight of exactly `0`, which would silently drop it from the retraining loss. diff --git a/quapy/classification/labelshift.py b/quapy/classification/labelshift.py new file mode 100644 index 0000000..6225f1e --- /dev/null +++ b/quapy/classification/labelshift.py @@ -0,0 +1,110 @@ +import numpy as np +from sklearn.base import BaseEstimator, clone + +import quapy.functional as F + + +class LabelShiftedClassifier(BaseEstimator): + """ + Adapts a classifier to a shifted (unlabelled) target distribution by means of a `quapy` quantifier. + + Given a base classifier `h` and a quantifier `q`, this wrapper: (i) at `fit` time, fits `q` on the labelled + training data and keeps a copy of this data; (ii) at prediction time, for a given batch of (unlabelled) + target instances, estimates the importance weights :math:`w_y=Q(y)/P(y)` that would explain the shift + between the training distribution and the target one (directly, if `q` is an + :class:`quapy.method.aggregative.ImportanceWeightQuantifier`, or otherwise by dividing `q`'s prevalence + estimate by the training prevalence); and (iii) retrains a fresh copy of `h`, reweighted according to `w` + via `class_weight`, on the original training data, before delegating the classification of the target + instances to this adapted classifier. + + Note that a fresh copy of `h` is retrained for every batch of target instances passed to `predict`, + `predict_proba`, or `get_classifier`, since the adaptation is specific to that batch's estimated shift; `q` + and its own internal classifier (if any) are otherwise left untouched. `h` itself is required to accept a + `class_weight` parameter in its constructor (e.g., `sklearn.linear_model.LogisticRegression`). + + :param classifier: a scikit-learn classifier that accepts a `class_weight` parameter in its constructor + :param quantifier: a `quapy` quantifier, used to estimate the target prevalence (or, if it is an + :class:`quapy.method.aggregative.ImportanceWeightQuantifier`, the importance weights directly) + :param weight_epsilon: float, a small positive floor applied to the estimated importance weights (default + 1e-4), so that a class with training support is never assigned a `class_weight` of exactly 0 (which + would silently drop it from the retraining loss) + """ + + def __init__(self, classifier, quantifier, weight_epsilon=1e-4): + self.classifier = classifier + self.quantifier = quantifier + self.weight_epsilon = weight_epsilon + + def fit(self, X, y): + """ + Fits the internal quantifier on the training data, and stores the training data for the subsequent, + per-target-batch, classifier adaptation. + + :param X: array-like of shape `(n_samples, n_features)` with the training instances + :param y: array-like of shape `(n_samples,)` with the class labels + :return: self + """ + self.quantifier.fit(X, y) + self.classes_ = self.quantifier.classes_ + self.train_prevalence_ = F.prevalence_from_labels(y, classes=self.classes_) + self.X_, self.y_ = X, y + return self + + def get_importance_weights(self, X): + """ + Estimates the vector of importance weights :math:`w_y=Q(y)/P(y)` that would explain the shift between + the training distribution and the distribution of the given target instances. This method does not + mutate any internal state, so it is safe to call concurrently for different batches of target + instances. + + :param X: array-like of shape `(n_samples, n_features)` with the (unlabelled) target instances + :return: np.ndarray of shape `(n_classes,)` + """ + from quapy.method.aggregative import ImportanceWeightQuantifier + if isinstance(self.quantifier, ImportanceWeightQuantifier): + weights = self.quantifier.get_importance_weights(X) + else: + test_prevalence = self.quantifier.predict(X) + weights = test_prevalence / np.maximum(self.train_prevalence_, self.weight_epsilon) + return np.clip(weights, self.weight_epsilon, None) + + def get_classifier(self, X): + """ + Returns a fresh classifier instance, sharing `classifier`'s hyperparameters, retrained on the original + training data with `class_weight` set according to the importance weights estimated for the given + target instances. The classifier passed at construction time, and the one used internally by the + quantifier, are left untouched. + + :param X: array-like of shape `(n_samples, n_features)` with the (unlabelled) target instances + :return: a fitted scikit-learn classifier + """ + weights = self.get_importance_weights(X) + class_weight = dict(zip(self.classes_, weights)) + adapted = clone(self.classifier) + adapted.set_params(class_weight=class_weight) + adapted.fit(self.X_, self.y_) + return adapted + + def predict(self, X): + """ + Adapts the classifier to the target instances in `X`, and returns its label predictions for `X`. + Adaptation implies retraining. If you intend to generate inferences over many samples from the same + suspected distribution, then call the `get_classifier` method once, and use the returned classifier + over such samples. + + :param X: array-like of shape `(n_samples, n_features)` with the (unlabelled) target instances + :return: array-like of shape `(n_samples,)` with the class label predictions + """ + return self.get_classifier(X).predict(X) + + def predict_proba(self, X): + """ + Adapts the classifier to the target instances in `X`, and returns its posterior probabilities for `X`. + Adaptation implies retraining. If you intend to generate inferences over many samples from the same + suspected distribution, then call the `get_classifier` method once, and use the returned classifier + over such samples. + + :param X: array-like of shape `(n_samples, n_features)` with the (unlabelled) target instances + :return: array-like of shape `(n_samples, n_classes)` with posterior probabilities + """ + return self.get_classifier(X).predict_proba(X) diff --git a/quapy/method/_liep.py b/quapy/method/_liep.py new file mode 100644 index 0000000..3c28203 --- /dev/null +++ b/quapy/method/_liep.py @@ -0,0 +1,227 @@ +import warnings + +import numpy as np +from sklearn.base import BaseEstimator + +import quapy.functional as F +from quapy.method._helper import _labels_to_indices +from quapy.method.aggregative import AggregativeSoftQuantifier + + +def _normalize(v, eps=1e-12): + v = np.maximum(np.asarray(v, dtype=float), eps) + return v / v.sum() + + +def _bayes_prior_update(probs, new_prior, source_prior, eps=1e-12): + """ + Applies the standard label-shift prior correction: + + p_new(y|x) \\propto p_old(y|x) * new_prior(y) / source_prior(y) + + :param probs: array of shape `(n_instances, n_classes)` with the (uncorrected) posterior probabilities + :param new_prior: array of shape `(n_classes,)`, the current estimate of the target prior + :param source_prior: array of shape `(n_classes,)`, the source (training) prior + :param eps: numerical stabilizer + :return: array of shape `(n_instances, n_classes)` with the corrected posterior probabilities + """ + weights = _normalize(new_prior, eps) / np.maximum(_normalize(source_prior, eps), eps) + updated = probs * weights[np.newaxis, :] + updated /= np.maximum(updated.sum(axis=1, keepdims=True), eps) + return updated + + +def _confusion_statistic(y_true, y_pred, n_classes, mode='recall', eps=1e-12): + """ + Computes the minimum diagonal statistic of the confusion matrix, used to choose the fraction of target + instances to be retained in the high-confidence set (LEIP, Section 4.1). + + :param y_true: array of true label indices in `[0, n_classes)` + :param y_pred: array of predicted label indices in `[0, n_classes)` + :param n_classes: number of classes + :param mode: "recall" (per-true-class recall, i.e., diag(C) / row sums) or "precision" (per-predicted-class + precision, i.e., diag(C) / column sums) + :param eps: numerical stabilizer + :return: float, the minimum, across classes with at least one valid instance, of the chosen statistic + """ + C = np.zeros((n_classes, n_classes), dtype=float) + for yt, yp in zip(y_true, y_pred): + C[yt, yp] += 1. + + denom = C.sum(axis=1) if mode == 'recall' else C.sum(axis=0) + diag = np.diag(C) + valid = denom > 0 + if not np.any(valid): + raise ValueError('no valid classes found in the confusion matrix') + + return float(np.min(diag[valid] / np.maximum(denom[valid], eps))) + + +class LEIP(AggregativeSoftQuantifier): + """ + `Label Shift Estimation With Incremental Prior update + `_ (LEIP). + + Zhang, Y., Batista, G., & Kanhere, S.S. (2025). Label Shift Estimation With Incremental Prior Update. + In Proceedings of the 2025 SIAM International Conference on Data Mining (SDM), pp. 134-142. + + LEIP is a variant of :class:`EMQ` (aka SLD) that estimates the target prior without resorting to an + iterative Expectation-Maximization procedure. The target instances are first split into a high-confidence + set (those instances whose maximum posterior probability is at or above a threshold `tau`) and a + low-confidence set. The high-confidence set is used to obtain an initial classify-and-count estimate of + the target prior. The low-confidence instances are then relabelled one at a time, in order of decreasing + confidence, applying at each step the standard label-shift Bayes correction (see the module-level function + :func:`_bayes_prior_update`) using the running target-prior estimate; each newly assigned pseudo-label + updates this running estimate before the next instance is processed. Finally, a single Bayes correction is + applied to the whole target set using the resulting prior, and a hard-label classify-and-count pass on the + corrected posteriors yields the returned prevalence estimate. + + :param classifier: a scikit-learn's BaseEstimator, or None, in which case the classifier is taken to be + the one indicated in `qp.environ['DEFAULT_CLS']` + + :param fit_classifier: whether to train the classifier (default is True). Set to False if the + given classifier has already been trained. + + :param val_split: specifies the data used, when `tau=None`, for estimating the fraction of target + instances to retain in the high-confidence set (see `threshold_mode`). This specification can be made + as a float in (0, 1) indicating the proportion of stratified held-out validation set to be extracted + from the training set; or as an integer (default 5), indicating that the predictions are to be + generated in a `k`-fold cross-validation manner (with this integer indicating the value for `k`); or + as a tuple `(X, y)` defining the specific set of data to use for validation. This parameter is ignored + when a fixed `tau` is provided. + + :param tau: float or None (default). A fixed confidence threshold used to split the target instances into + the high- and low-confidence sets. If None, the threshold is instead derived, independently for each + target sample, as the quantile of that sample's own posterior-confidence distribution corresponding to + a retention fraction estimated from `val_split` (see `threshold_mode`). + + :param threshold_mode: either "recall" (default) or "precision", the statistic computed on the validation + confusion matrix that determines the fraction of target instances to be retained in the high-confidence + set (see the module-level function :func:`_confusion_statistic`). Only used when `tau=None`. + + :param count_smoothing: float, additive smoothing applied to the pseudo-label counts used to compute the + running target-prior estimate (default 0.0, matching the original paper; a small positive value, e.g., + 1e-8, can instead be used for extra numerical robustness). + """ + + THRESHOLD_MODES = ['recall', 'precision'] + + def __init__(self, classifier: BaseEstimator = None, fit_classifier=True, val_split=5, tau=None, + threshold_mode='recall', count_smoothing=0.0): + + assert threshold_mode in LEIP.THRESHOLD_MODES, \ + f'invalid {threshold_mode=}; valid ones are {LEIP.THRESHOLD_MODES}' + assert tau is None or isinstance(tau, (int, float)), f'invalid {tau=}; must be None or a number' + assert isinstance(count_smoothing, (int, float)) and count_smoothing >= 0, \ + f'invalid {count_smoothing=}; must be a non-negative number' + + super().__init__(classifier, fit_classifier, val_split) + self.tau = tau + self.threshold_mode = threshold_mode + self.count_smoothing = count_smoothing + + def _check_init_parameters(self): + if self.tau is None: + if self.val_split is None: + raise ValueError( + 'LEIP requires validation predictions to estimate the retention threshold when tau=None; ' + 'please set val_split to an integer, float, or validation tuple, or provide a fixed tau.' + ) + elif self.val_split is not None: + warnings.warn(f'a fixed {self.tau=} was set; {self.val_split=} will be ignored, since LEIP only ' + f'needs validation data to estimate a retention threshold when tau is not fixed.') + + def classify(self, X): + """ + Provides the posterior probabilities for the given instances. + + :param X: array-like of shape `(n_instances, n_dimensions,)` + :return: np.ndarray of shape `(n_instances, n_classes,)` with posterior probabilities + """ + return self.classifier.predict_proba(X) + + def classifier_fit_predict(self, X, y): + classif_predictions = super().classifier_fit_predict(X, y) + self.train_prevalence = F.prevalence_from_labels(y, classes=self.classes_) + return classif_predictions + + def aggregation_fit(self, classif_predictions, labels): + """ + Trains the aggregation function of LEIP. When a fixed `tau` was not provided at construction time, + this estimates, from the validation predictions, the fraction of target instances that should be + retained in the high-confidence set (see :func:`_confusion_statistic`). + + :param classif_predictions: array-like with the posterior probabilities of the validation instances + :param labels: array-like with the true labels associated to each classifier prediction + """ + if self.tau is None: + n_classes = len(self.classes_) + y_val_idx = _labels_to_indices(labels, self.classes_) + y_val_pred_idx = classif_predictions.argmax(axis=1) + self.retain_fraction_ = _confusion_statistic( + y_val_idx, y_val_pred_idx, n_classes, mode=self.threshold_mode + ) + else: + self.retain_fraction_ = None + + def aggregate(self, classif_posteriors): + prevalence, _ = self._leip(classif_posteriors) + return prevalence + + def predict_proba(self, instances): + """ + Returns the posterior probabilities updated by the final Bayes correction applied by LEIP. + + :param instances: np.ndarray of shape `(n_instances, n_dimensions)` + :return: np.ndarray of shape `(n_instances, n_classes)` + """ + classif_posteriors = self.classify(instances) + _, corrected_posteriors = self._leip(classif_posteriors) + return corrected_posteriors + + def _leip(self, target_probs): + n_classes = len(self.classes_) + source_prior = self.train_prevalence + + target_conf = target_probs.max(axis=1) + target_top = target_probs.argmax(axis=1) + + if self.tau is not None: + tau = self.tau + else: + retain_fraction = float(np.clip(self.retain_fraction_, 0., 1.)) + if retain_fraction <= 0: + tau = np.inf + elif retain_fraction >= 1: + tau = -np.inf + else: + tau = np.quantile(target_conf, 1. - retain_fraction) + self.tau_ = tau + + # high-confidence set A: an initial classify-and-count estimate of the target prior + A_mask = target_conf >= tau + counts = np.full(n_classes, self.count_smoothing, dtype=float) + if np.any(A_mask): + counts += np.bincount(target_top[A_mask], minlength=n_classes) + current_prior = counts / counts.sum() + else: + current_prior = source_prior.copy() + + # low-confidence set B, relabelled incrementally in order of decreasing confidence + B_indices = np.where(~A_mask)[0] + B_indices = B_indices[np.argsort(-target_conf[B_indices])] + + for idx in B_indices: + corrected = _bayes_prior_update(target_probs[idx:idx + 1], current_prior, source_prior)[0] + pseudo_label = int(np.argmax(corrected)) + counts[pseudo_label] += 1. + current_prior = counts / counts.sum() + + self.intermediate_prior_ = current_prior.copy() + + # final full-batch Bayes correction and hard-label classify-and-count pass + corrected_posteriors = _bayes_prior_update(target_probs, current_prior, source_prior) + final_labels = corrected_posteriors.argmax(axis=1) + prevalence = F.prevalence_from_labels(final_labels, classes=np.arange(n_classes)) + + return prevalence, corrected_posteriors diff --git a/quapy/tests/test_labelshift.py b/quapy/tests/test_labelshift.py new file mode 100644 index 0000000..adaae03 --- /dev/null +++ b/quapy/tests/test_labelshift.py @@ -0,0 +1,84 @@ +import unittest + +import numpy as np +from sklearn.linear_model import LogisticRegression +from sklearn.exceptions import NotFittedError +from sklearn.utils.validation import check_is_fitted + +from quapy.classification.labelshift import LabelShiftedClassifier +from quapy.method.aggregative import ACC, BBSEhard, BBSEsoft, ImportanceWeightQuantifier, RLLS +from quapy.functional import check_prevalence_vector +from quapy.tests._synthetic import make_dataset + + +class TestLabelShift(unittest.TestCase): + + dataset = make_dataset(n_train=200, n_test=100, n_classes=3, n_features=10, random_state=7, name='labelshift') + + def test_importance_weight_quantifiers(self): + Xtr, ytr = self.dataset.training.Xy + Xte = self.dataset.test.X + for cls in [BBSEhard, BBSEsoft, RLLS]: + if cls is RLLS: + try: + import cvxpy # noqa: F401 + except ImportError: + continue + + q = cls(LogisticRegression(max_iter=2000), val_split=3) + q.fit(Xtr, ytr) + + weights = q.get_importance_weights(Xte) + prevalence, weights2 = q.quantify_and_weigh(Xte) + prevalence_direct = q.predict(Xte) + + self.assertTrue(check_prevalence_vector(prevalence)) + np.testing.assert_allclose(weights, weights2) + np.testing.assert_allclose(prevalence, prevalence_direct) + # no leftover mutable per-call state (safety against races under concurrent calls) + self.assertFalse(hasattr(q, 'weights')) + self.assertFalse(hasattr(q, 'last_w_')) + + def test_label_shifted_classifier_generic_quantifier(self): + Xtr, ytr = self.dataset.training.Xy + Xte = self.dataset.test.X + + base_classifier = LogisticRegression(max_iter=2000) + q = ACC(LogisticRegression(max_iter=2000), val_split=3) + self.assertNotIsInstance(q, ImportanceWeightQuantifier) + + lsc = LabelShiftedClassifier(base_classifier, q) + lsc.fit(Xtr, ytr) + + weights = lsc.get_importance_weights(Xte) + self.assertEqual(weights.shape, (self.dataset.training.n_classes,)) + self.assertTrue(np.all(weights >= lsc.weight_epsilon)) + + preds = lsc.predict(Xte) + probs = lsc.predict_proba(Xte) + self.assertEqual(len(preds), len(Xte)) + self.assertEqual(probs.shape, (len(Xte), self.dataset.training.n_classes)) + np.testing.assert_allclose(probs.sum(axis=1), 1.0, rtol=1e-5) + + # the classifier instance passed at construction time must be left untouched + with self.assertRaises(NotFittedError): + check_is_fitted(base_classifier) + + def test_label_shifted_classifier_importance_weight_quantifier(self): + Xtr, ytr = self.dataset.training.Xy + Xte = self.dataset.test.X + + q = BBSEhard(LogisticRegression(max_iter=2000), val_split=3) + self.assertIsInstance(q, ImportanceWeightQuantifier) + + lsc = LabelShiftedClassifier(LogisticRegression(max_iter=2000), q) + lsc.fit(Xtr, ytr) + + weights = lsc.get_importance_weights(Xte) + adapted = lsc.get_classifier(Xte) + check_is_fitted(adapted) # should not raise + self.assertEqual(set(adapted.get_params()['class_weight'].keys()), set(lsc.classes_)) + + +if __name__ == '__main__': + unittest.main()