adding liep
This commit is contained in:
parent
cb73c78cf3
commit
3ecbadb921
|
|
@ -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) <manuals/methods:Regularized Learning under Label Shift (RLLS)>`
|
||||
and {ref}`Black Box Shift Estimation (BBSE) <manuals/methods: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.
|
||||
|
|
@ -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)
|
||||
|
|
@ -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
|
||||
<https://doi.org/10.1137/1.9781611978520.12>`_ (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
|
||||
|
|
@ -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()
|
||||
Loading…
Reference in New Issue