Classify & Count (CC)

quack.quantifiers._baselines.CC

Bases: BaseQuantifier

Classify and Count (CC) quantifier.

The CC method is the simplest baseline in quantification. It works by classifying all unlabeled instances in the test bag using a hard (crisp) classifier, and then computing the relative frequency (prevalence) of each class based on those predictions.

Parameters:

Name Type Description Default
classifier estimator object

The classifier to be used as the base for quantification. If None, an instance of LogisticRegression() will be created.

= None

Attributes:

Name Type Description
classes_ ndarray of shape (n_classes,)

The distinct class labels found during training, sorted ascending (as returned by np.unique).

n_classes_ int

The number of distinct classes.

train_prevalence_ ndarray of shape (n_classes,)

The prevalence of each class in the training dataset.

y_prevs_ ndarray of shape (n_classes,)

Alias for train_prevalence_ kept for backward compatibility.

classifier_ estimator object

The fitted base classifier trained on the entire dataset.

Notes

The Classify and Count method does not adjust for misclassifications made by the base classifier (false positives and false negatives). Therefore, its performance is highly dependent on the classification accuracy.

The estimated prevalence for class 'c' is given by the formula:

p_hat(c) = (1 / |X|) * sum( I( f(x) == c ) )

where f(x) is the crisp prediction of the classifier for instance x, and I() is the indicator function.

References

George Forman. Counting positives accurately despite inaccurate classification. In Proceedings of the 16th European Conference on Machine Learning, pages 564-575, Porto, Portugal, 2005.

Examples:

>>> from sklearn.datasets import make_classification
>>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
>>> quantifier = CC()
>>> quantifier.fit(X, y)
>>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
>>> prevalences = quantifier.predict(X_test)
>>> print(prevalences)
Source code in quack/quantifiers/_baselines.py
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
class CC(BaseQuantifier):
  """Classify and Count (CC) quantifier.

  The CC method is the simplest baseline in quantification. It works by
  classifying all unlabeled instances in the test bag using a hard (crisp)
  classifier, and then computing the relative frequency (prevalence) of
  each class based on those predictions.

  Parameters
  ----------
  classifier : estimator object, default = None
    The classifier to be used as the base for quantification.
    If None, an instance of `LogisticRegression()` will be created.

  Attributes
  ----------
  classes_ : ndarray of shape (n_classes,)
    The distinct class labels found during training, sorted ascending
    (as returned by `np.unique`).

  n_classes_ : int
    The number of distinct classes.

  train_prevalence_ : ndarray of shape (n_classes,)
    The prevalence of each class in the training dataset.

  y_prevs_ : ndarray of shape (n_classes,)
    Alias for train_prevalence_ kept for backward compatibility.

  classifier_ : estimator object
    The fitted base classifier trained on the entire dataset.

  Notes
  -----
  The Classify and Count method does not adjust for misclassifications
  made by the base classifier (false positives and false negatives). Therefore,
  its performance is highly dependent on the classification accuracy. 

  The estimated prevalence for class 'c' is given by the formula:

    p_hat(c) = (1 / |X|) * sum( I( f(x) == c ) )

  where f(x) is the crisp prediction of the classifier for instance x,
  and I() is the indicator function.

  References
  ----------
  George Forman. Counting positives accurately despite inaccurate classification.
  In Proceedings of the 16th European Conference on Machine Learning, pages 564-575,
  Porto, Portugal, 2005.

  Examples
  --------
  >>> from sklearn.datasets import make_classification
  >>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
  >>> quantifier = CC()
  >>> quantifier.fit(X, y)
  >>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
  >>> prevalences = quantifier.predict(X_test)
  >>> print(prevalences)
  """

  def __init__(self, classifier: BaseEstimator = None):
    super().__init__(classifier)

  def fit(self, X: np.ndarray, y: np.ndarray) -> "CC":
    """Adjusts the CC quantifier by fitting the base classifier.

    Parameters
    ----------
    X : {array-like, sparse matrix} of shape (n_samples, n_features)
      Training data.
    y : array-like of shape (n_samples,)
      Labels for the corresponding classes.

    Returns
    -------
    self : object
        Returns the fitted estimator instance itself.
    """
    X, y = check_X_y(X, y, accept_sparse=True)

    self.classes_, counts = np.unique(y, return_counts=True)
    self.n_classes_ = len(self.classes_)
    self.train_prevalence_ = counts / len(y)
    self.y_prevs_ = self.train_prevalence_ # Compatibility purposes
    # lazy validation of the classifier
    base_classifier = self.classifier if self.classifier is not None else LogisticRegression()
    self.classifier_ = clone(base_classifier)
    self.classifier_.fit(X, y) # fit the classifier with all training data

    return self

  def predict(self, X: np.ndarray) -> np.ndarray:
    """Estimate the class prevalences for the test bag X.

    Parameters
    ----------
    X : {array-like, sparse matrix} of shape (n_samples, n_features)
      The test bag with unlabelled instances.

    Returns
    -------
    p_adjusted : ndarray of shape (n_classes,)
      An array with the estimated prevalences for each class,
      normalized to sum up to 1.0.
    """
    check_is_fitted(self)
    X = check_array(X, accept_sparse=True)

    y_pred = self.classifier_.predict(X)

    # fully vectorized counting: since self.classes_ is sorted ascending
    # (np.unique's guarantee), searchsorted maps every prediction to its
    # position in self.classes_ and bincount tallies them in one pass,
    # avoiding the per-call Python dict construction/lookup this replaces
    class_indices = np.searchsorted(self.classes_, y_pred)
    counts = np.bincount(class_indices, minlength=self.n_classes_).astype(float)

    return normalize_prevalence(counts, self.n_classes_)

fit(X, y)

Adjusts the CC quantifier by fitting the base classifier.

Parameters:

Name Type Description Default
X array-like, sparse matrix

Training data.

array-like
y array-like of shape (n_samples,)

Labels for the corresponding classes.

required

Returns:

Name Type Description
self object

Returns the fitted estimator instance itself.

Source code in quack/quantifiers/_baselines.py
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
def fit(self, X: np.ndarray, y: np.ndarray) -> "CC":
  """Adjusts the CC quantifier by fitting the base classifier.

  Parameters
  ----------
  X : {array-like, sparse matrix} of shape (n_samples, n_features)
    Training data.
  y : array-like of shape (n_samples,)
    Labels for the corresponding classes.

  Returns
  -------
  self : object
      Returns the fitted estimator instance itself.
  """
  X, y = check_X_y(X, y, accept_sparse=True)

  self.classes_, counts = np.unique(y, return_counts=True)
  self.n_classes_ = len(self.classes_)
  self.train_prevalence_ = counts / len(y)
  self.y_prevs_ = self.train_prevalence_ # Compatibility purposes
  # lazy validation of the classifier
  base_classifier = self.classifier if self.classifier is not None else LogisticRegression()
  self.classifier_ = clone(base_classifier)
  self.classifier_.fit(X, y) # fit the classifier with all training data

  return self

predict(X)

Estimate the class prevalences for the test bag X.

Parameters:

Name Type Description Default
X array-like, sparse matrix

The test bag with unlabelled instances.

array-like

Returns:

Name Type Description
p_adjusted ndarray of shape (n_classes,)

An array with the estimated prevalences for each class, normalized to sum up to 1.0.

Source code in quack/quantifiers/_baselines.py
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
def predict(self, X: np.ndarray) -> np.ndarray:
  """Estimate the class prevalences for the test bag X.

  Parameters
  ----------
  X : {array-like, sparse matrix} of shape (n_samples, n_features)
    The test bag with unlabelled instances.

  Returns
  -------
  p_adjusted : ndarray of shape (n_classes,)
    An array with the estimated prevalences for each class,
    normalized to sum up to 1.0.
  """
  check_is_fitted(self)
  X = check_array(X, accept_sparse=True)

  y_pred = self.classifier_.predict(X)

  # fully vectorized counting: since self.classes_ is sorted ascending
  # (np.unique's guarantee), searchsorted maps every prediction to its
  # position in self.classes_ and bincount tallies them in one pass,
  # avoiding the per-call Python dict construction/lookup this replaces
  class_indices = np.searchsorted(self.classes_, y_pred)
  counts = np.bincount(class_indices, minlength=self.n_classes_).astype(float)

  return normalize_prevalence(counts, self.n_classes_)

Adjusted Classify & Count (ACC)

quack.quantifiers._baselines.ACC

Bases: BaseCalibratedQuantifier

Adjusted Classify and Count (ACC) quantifier for binary problems.

ACC is a calibrated quantification method that adjusts the simple Classify and Count (CC) estimates. It uses Out-of-Fold (OOF) predictions to calculate the True Positive Rate (TPR) and False Positive Rate (FPR) of the base classifier. Then, it mathematically corrects the final prevalence using the classical binary adjustment formula.

Parameters:

Name Type Description Default
classifier estimator object

The classifier to be used as the base for quantification. If None, an instance of LogisticRegression() will be created.

None
cv int, cross-validation generator or an iterable

Determines the cross-validation splitting strategy to generate the Out-of-Fold predictions used for calibration.

= 10
n_jobs int

Number of jobs to run in parallel while fitting the cv folds (plus the final full-data classifier refit). See BaseCalibratedQuantifier.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs. See BaseCalibratedQuantifier.

= "loky"

Attributes:

Name Type Description
classes_ ndarray of shape (2,)

The distinct class labels found during training.

n_classes_ int

The number of distinct classes (expected to be 2).

train_prevalence_ ndarray of shape (2,)

The prevalence of each class in the training dataset.

tpr_ float

The True Positive Rate (Sensitivity) estimated out-of-fold.

fpr_ float

The False Positive Rate (1 - Specificity) estimated out-of-fold.

classifier_ estimator object

The fitted base classifier trained on the entire dataset.

Notes

The ACC method adjusts the observed crisp prevalence (p_pred) using the following analytical equation for the positive class:

p_adjusted = (p_pred - fpr) / (tpr - fpr)

The prevalence for the negative class is then derived as 1 - p_corrected. If TPR equals FPR, the denominator becomes zero, and the method falls back to the unadjusted p_pred.

References

George Forman. Counting positives accurately despite inaccurate classification. In Proceedings of the 16th European Conference on Machine Learning, pages 564-575, Porto, Portugal, 2005.

Examples:

>>> from sklearn.datasets import make_classification
>>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
>>> quantifier = ACC()
>>> quantifier.fit(X, y)
>>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
>>> prevalences = quantifier.predict(X_test)
>>> print(prevalences)
Source code in quack/quantifiers/_baselines.py
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
class ACC(BaseCalibratedQuantifier):
  """Adjusted Classify and Count (ACC) quantifier for binary problems.

  ACC is a calibrated quantification method that adjusts the simple Classify 
  and Count (CC) estimates. It uses Out-of-Fold (OOF) predictions to calculate 
  the True Positive Rate (TPR) and False Positive Rate (FPR) of the base classifier. 
  Then, it mathematically corrects the final prevalence using the classical 
  binary adjustment formula.

  Parameters
  ----------
  classifier : estimator object, default=None
    The classifier to be used as the base for quantification.
    If None, an instance of `LogisticRegression()` will be created.
  cv : int, cross-validation generator or an iterable, default = 10
    Determines the cross-validation splitting strategy to generate the 
    Out-of-Fold predictions used for calibration.
  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds (plus
    the final full-data classifier refit). See `BaseCalibratedQuantifier`.
  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs. See
    `BaseCalibratedQuantifier`.

  Attributes
  ----------
  classes_ : ndarray of shape (2,)
    The distinct class labels found during training.

  n_classes_ : int
    The number of distinct classes (expected to be 2).

  train_prevalence_ : ndarray of shape (2,)
    The prevalence of each class in the training dataset.

  tpr_ : float
    The True Positive Rate (Sensitivity) estimated out-of-fold.

  fpr_ : float
    The False Positive Rate (1 - Specificity) estimated out-of-fold.

  classifier_ : estimator object
    The fitted base classifier trained on the entire dataset.

  Notes
  -----
  The ACC method adjusts the observed crisp prevalence (p_pred) using the 
  following analytical equation for the positive class:

    p_adjusted = (p_pred - fpr) / (tpr - fpr)

  The prevalence for the negative class is then derived as 1 - p_corrected.
  If TPR equals FPR, the denominator becomes zero, and the method falls back 
  to the unadjusted p_pred.

  References
  ----------
  George Forman. Counting positives accurately despite inaccurate classification.
  In Proceedings of the 16th European Conference on Machine Learning, pages 564-575,
  Porto, Portugal, 2005.

  Examples
  --------
  >>> from sklearn.datasets import make_classification
  >>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
  >>> quantifier = ACC()
  >>> quantifier.fit(X, y)
  >>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
  >>> prevalences = quantifier.predict(X_test)
  >>> print(prevalences)
  """

  def __init__(self, classifier: BaseEstimator = None, cv: int = 10,
               n_jobs: int = None, parallel_backend: str = "loky"):
    super().__init__(classifier=classifier, cv=cv, n_jobs=n_jobs, parallel_backend=parallel_backend)

  def _get_oof_method(self) -> str:
    return "predict"

  def _calibrate(self, y_true_oof: np.ndarray, y_pred_oof: np.ndarray):
    """Calculates TPR and FPR using the Out-of-Fold predictions.

    Parameters
    ----------
    y_true_oof : ndarray of shape (n_samples,)
      True labels collected out-of-fold.
    y_pred_oof : ndarray of shape (n_samples,)
      Crisp predictions generated out-of-fold.
    """
    # maps each class using the scikit-learn pattern (asc format)
    neg_label = self.classes_[0]
    pos_label = self.classes_[1]

    is_true_pos = (y_true_oof == pos_label)
    is_true_neg = (y_true_oof == neg_label)

    tp = np.sum(is_true_pos & (y_pred_oof == pos_label))
    fn = np.sum(is_true_pos & (y_pred_oof == neg_label))
    fp = np.sum(is_true_neg & (y_pred_oof == pos_label))
    tn = np.sum(is_true_neg & (y_pred_oof == neg_label))

    self.tpr_ = tp / (tp + fn) if (tp + fn) > 0 else 1.0
    self.fpr_ = fp / (fp + tn) if (fp + tn) > 0 else 0.0

  def _quantify(self, X: np.ndarray) -> np.ndarray:
    """Applies the binary ACC equation to adjust the predicted prevalence.

    Parameters
    ----------
    X : ndarray of shape (n_samples, n_features)
      Raw data from test bags.

    Returns
    -------
    p_acc : ndarray of shape (2,)
      The adjusted prevalences for [negative_class, positive_class].
    """
    y_pred = self.classifier_.predict(X)
    pos_label = self.classes_[1]

    p_pred = np.mean(y_pred == pos_label)

    denominator = self.tpr_ - self.fpr_
    if denominator == 0:
      p_pos_adjusted = p_pred
    else:
      p_pos_adjusted = (p_pred - self.fpr_) / denominator

    # guardrails: clipping
    p_pos_adjusted = np.clip(p_pos_adjusted, 0.0, 1.0)
    p_neg_adjusted = 1.0 - p_pos_adjusted

    return np.array([p_neg_adjusted, p_pos_adjusted])

  def fit(self, X: np.ndarray, y: np.ndarray) -> "ACC":
    """Fits the ACC quantifier, enforcing the binary-only constraint.

    Raises
    ------
    ValueError
      If `y` contains more than 2 distinct classes.
    """
    if len(np.unique(y)) > 2:
      raise ValueError(
        "ACC method only works for binary quantification. Multiclass "
        "quantification is possible via OVR strategies, but not recommended due to "
        "theoretical issues with that approach."
      )

    return super().fit(X, y)

fit(X, y)

Fits the ACC quantifier, enforcing the binary-only constraint.

Raises:

Type Description
ValueError

If y contains more than 2 distinct classes.

Source code in quack/quantifiers/_baselines.py
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
def fit(self, X: np.ndarray, y: np.ndarray) -> "ACC":
  """Fits the ACC quantifier, enforcing the binary-only constraint.

  Raises
  ------
  ValueError
    If `y` contains more than 2 distinct classes.
  """
  if len(np.unique(y)) > 2:
    raise ValueError(
      "ACC method only works for binary quantification. Multiclass "
      "quantification is possible via OVR strategies, but not recommended due to "
      "theoretical issues with that approach."
    )

  return super().fit(X, y)

Probabilistic Classify & Count (PCC)

quack.quantifiers._baselines.PCC

Bases: BaseQuantifier

Probabilistic Classify and Count (PCC) quantifier.

The PCC method extends the classical Classify and Count (CC) by using the posterior probabilities generated by a classifier instead of its crisp (hard) predictions. The estimated prevalence for each class is computed as the expected value (average) of the predicted probabilities across all instances in the test bag.

Parameters:

Name Type Description Default
classifier estimator object

The classifier to be used as the base for quantification. Must implement the predict_proba method. If None, an instance of LogisticRegression() will be created.

= None

Attributes:

Name Type Description
classes_ ndarray of shape (n_classes,)

The distinct class labels found during training.

n_classes_ int

The number of distinct classes.

train_prevalence_ ndarray of shape (n_classes,)

The prevalence of each class in the training dataset.

y_prevs_ ndarray of shape (n_classes,)

Alias for train_prevalence_ kept for backward compatibility.

classifier_ estimator object

The fitted base classifier trained on the entire dataset.

Notes

Unlike CC, PCC accounts for the classifier's confidence in its predictions, which often leads to better quantification performance, especially when the classifier is well-calibrated. However, like CC, it is an "unadjusted" method, meaning it does not explicitly correct for systematic classification errors via a confusion matrix.

The estimated prevalence for class 'c' is given by the formula:

p_hat(c) = (1 / |X|) * sum( P(c | x) )

where P(c | x) is the probability that instance x belongs to class c, estimated by the base classifier's predict_proba method.

References

Antonio Bella, Cesar Ferri, José Hernández-Orallo, and María José Ramírez-Quintana. Quantification via probability estimators. In 2010 IEEE International Conference on Data Mining, pages 737-742, Sydney, Australia, 2010.

Examples:

>>> from sklearn.datasets import make_classification
>>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
>>> quantifier = PCC()
>>> quantifier.fit(X, y)
>>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
>>> prevalences = quantifier.predict(X_test)
>>> print(prevalences)
Source code in quack/quantifiers/_baselines.py
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
class PCC(BaseQuantifier):
  """Probabilistic Classify and Count (PCC) quantifier.

  The PCC method extends the classical Classify and Count (CC) by using the
  posterior probabilities generated by a classifier instead of its crisp (hard) 
  predictions. The estimated prevalence for each class is computed as the 
  expected value (average) of the predicted probabilities across all instances 
  in the test bag.

  Parameters
  ----------
  classifier : estimator object, default = None
    The classifier to be used as the base for quantification.
    Must implement the `predict_proba` method. If None, an instance of 
    `LogisticRegression()` will be created.

  Attributes
  ----------
  classes_ : ndarray of shape (n_classes,)
    The distinct class labels found during training.

  n_classes_ : int
    The number of distinct classes.

  train_prevalence_ : ndarray of shape (n_classes,)
    The prevalence of each class in the training dataset.

  y_prevs_ : ndarray of shape (n_classes,)
    Alias for train_prevalence_ kept for backward compatibility.

  classifier_ : estimator object
    The fitted base classifier trained on the entire dataset.

  Notes
  -----
  Unlike CC, PCC accounts for the classifier's confidence in its predictions, 
  which often leads to better quantification performance, especially when 
  the classifier is well-calibrated. However, like CC, it is an "unadjusted" 
  method, meaning it does not explicitly correct for systematic classification 
  errors via a confusion matrix.

  The estimated prevalence for class 'c' is given by the formula:

    p_hat(c) = (1 / |X|) * sum( P(c | x) )

  where P(c | x) is the probability that instance x belongs to class c, 
  estimated by the base classifier's `predict_proba` method.

  References
  ----------
  Antonio Bella, Cesar Ferri, José Hernández-Orallo, and María José Ramírez-Quintana.
  Quantification via probability estimators. In 2010 IEEE International Conference on
  Data Mining, pages 737-742, Sydney, Australia, 2010.

  Examples
  --------
  >>> from sklearn.datasets import make_classification
  >>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
  >>> quantifier = PCC()
  >>> quantifier.fit(X, y)
  >>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
  >>> prevalences = quantifier.predict(X_test)
  >>> print(prevalences)
  """
  def __init__(self, classifier: BaseEstimator = None):
    super().__init__(classifier=classifier)

  def fit(self, X: np.ndarray, y: np.ndarray) -> "PCC":
    """Adjusts the PCC quantifier by fitting the probabilistic base classifier.

    Parameters
    ----------
    X : {array-like, sparse matrix} of shape (n_samples, n_features)
      Training data.
    y : array-like of shape (n_samples,)
      Labels for the corresponding classes.

    Returns
    -------
    self : object
      Returns the fitted estimator instance itself.

    Raises
    ------
    TypeError
      If the provided classifier does not support probability estimation 
      (i.e., lacks a `predict_proba` method).
    """
    X, y = check_X_y(X, y, accept_sparse=True)

    self.classes_, counts = np.unique(y, return_counts=True)
    self.n_classes_ = len(self.classes_)
    self.train_prevalence_ = counts / len(y)
    self.y_prevs_ = self.train_prevalence_ # Compatibility purposes

    base_classifier = self.classifier if self.classifier is not None else LogisticRegression()

    if not hasattr(base_classifier, "predict_proba"):
      raise TypeError(
        f"The classifier {base_classifier.__class__.__name__} does not "
        "support probability estimation. PCC requires 'predict_proba'."
      )

    self.classifier_ = clone(base_classifier)
    self.classifier_.fit(X, y)     

    return self

  def predict(self, X: np.ndarray) -> np.ndarray:
    """Estimate the class prevalences for the test bag X using soft probabilities.

    Parameters
    ----------
    X : {array-like, sparse matrix} of shape (n_samples, n_features)
      The test bag with unlabelled instances.

    Returns
    -------
    p_adjusted : ndarray of shape (n_classes,)
      An array with the estimated prevalences for each class,
      normalized to sum up to 1.0.
    """
    check_is_fitted(self)
    X = check_array(X, accept_sparse=True)
    # obtain the soft probability matrix (n_samples, n_classes)
    y_probas = self.classifier_.predict_proba(X)

    p_pred = np.mean(y_probas, axis=0)
    return normalize_prevalence(p_pred, self.n_classes_)

fit(X, y)

Adjusts the PCC quantifier by fitting the probabilistic base classifier.

Parameters:

Name Type Description Default
X array-like, sparse matrix

Training data.

array-like
y array-like of shape (n_samples,)

Labels for the corresponding classes.

required

Returns:

Name Type Description
self object

Returns the fitted estimator instance itself.

Raises:

Type Description
TypeError

If the provided classifier does not support probability estimation (i.e., lacks a predict_proba method).

Source code in quack/quantifiers/_baselines.py
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
def fit(self, X: np.ndarray, y: np.ndarray) -> "PCC":
  """Adjusts the PCC quantifier by fitting the probabilistic base classifier.

  Parameters
  ----------
  X : {array-like, sparse matrix} of shape (n_samples, n_features)
    Training data.
  y : array-like of shape (n_samples,)
    Labels for the corresponding classes.

  Returns
  -------
  self : object
    Returns the fitted estimator instance itself.

  Raises
  ------
  TypeError
    If the provided classifier does not support probability estimation 
    (i.e., lacks a `predict_proba` method).
  """
  X, y = check_X_y(X, y, accept_sparse=True)

  self.classes_, counts = np.unique(y, return_counts=True)
  self.n_classes_ = len(self.classes_)
  self.train_prevalence_ = counts / len(y)
  self.y_prevs_ = self.train_prevalence_ # Compatibility purposes

  base_classifier = self.classifier if self.classifier is not None else LogisticRegression()

  if not hasattr(base_classifier, "predict_proba"):
    raise TypeError(
      f"The classifier {base_classifier.__class__.__name__} does not "
      "support probability estimation. PCC requires 'predict_proba'."
    )

  self.classifier_ = clone(base_classifier)
  self.classifier_.fit(X, y)     

  return self

predict(X)

Estimate the class prevalences for the test bag X using soft probabilities.

Parameters:

Name Type Description Default
X array-like, sparse matrix

The test bag with unlabelled instances.

array-like

Returns:

Name Type Description
p_adjusted ndarray of shape (n_classes,)

An array with the estimated prevalences for each class, normalized to sum up to 1.0.

Source code in quack/quantifiers/_baselines.py
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
def predict(self, X: np.ndarray) -> np.ndarray:
  """Estimate the class prevalences for the test bag X using soft probabilities.

  Parameters
  ----------
  X : {array-like, sparse matrix} of shape (n_samples, n_features)
    The test bag with unlabelled instances.

  Returns
  -------
  p_adjusted : ndarray of shape (n_classes,)
    An array with the estimated prevalences for each class,
    normalized to sum up to 1.0.
  """
  check_is_fitted(self)
  X = check_array(X, accept_sparse=True)
  # obtain the soft probability matrix (n_samples, n_classes)
  y_probas = self.classifier_.predict_proba(X)

  p_pred = np.mean(y_probas, axis=0)
  return normalize_prevalence(p_pred, self.n_classes_)

Probabilistic Adjusted Classify & Count (PACC)

quack.quantifiers._baselines.PACC

Bases: BaseCalibratedQuantifier

Probabilistic Adjusted Classify and Count (PACC) quantifier for binary problems.

PACC is a calibrated quantification method that refines the Probabilistic Classify and Count (PCC) by correcting for the base classifier's systematic probabilistic bias. Using Out-of-Fold (OOF) predictions, it computes the expected predicted probability for the positive class given the true class labels. It then applies an analytical correction formula similar to ACC but entirely based on continuous probability scores.

Parameters:

Name Type Description Default
classifier estimator object

The classifier to be used as the base for quantification. Must implement the predict_proba method. If None, an instance of LogisticRegression() will be created.

= None
cv int, cross-validation generator or an iterable

Determines the cross-validation splitting strategy to generate the Out-of-Fold predictions used for calibration.

= 10
n_jobs int

Number of jobs to run in parallel while fitting the cv folds (plus the final full-data classifier refit). See BaseCalibratedQuantifier.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs. See BaseCalibratedQuantifier.

= "loky"

Attributes:

Name Type Description
classes_ ndarray of shape (2,)

The distinct class labels found during training.

n_classes_ int

The number of distinct classes (expected to be 2).

train_prevalence_ ndarray of shape (2,)

The prevalence of each class in the training dataset.

y_prevs_ ndarray of shape (2,)

Alias for train_prevalence_ kept for backward compatibility.

mu_pos_pos_ float

The mean probability assigned to the positive class for instances that are truly positive, estimated out-of-fold. Analogue to TPR.

mu_neg_pos_ float

The mean probability assigned to the positive class for instances that are truly negative, estimated out-of-fold. Analogue to FPR.

classifier_ estimator object

The fitted base classifier trained on the entire dataset.

Notes

The PACC method adjusts the observed probabilistic prevalence (p_pcc) using the following analytical equation for the positive class:

p_corrected = (p_pcc - mu_neg_pos) / (mu_pos_pos - mu_neg_pos)

where 'p_pcc' is the average predicted probability of the positive class in the test bag. The prevalence for the negative class is then derived as 1 - p_corrected. If the denominator evaluates to zero, the method falls back to the unadjusted p_pcc.

References

Antonio Bella, Cesar Ferri, José Hernández-Orallo, and María José Ramírez-Quintana. Quantification via probability estimators. In 2010 IEEE International Conference on Data Mining, pages 737-742, Sydney, Australia, 2010.

Examples:

>>> from sklearn.datasets import make_classification
>>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
>>> quantifier = PACC()
>>> quantifier.fit(X, y)
>>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
>>> prevalences = quantifier.predict(X_test)
>>> print(prevalences)
Source code in quack/quantifiers/_baselines.py
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
439
440
441
442
443
444
445
446
447
448
449
450
451
452
453
454
455
456
457
458
459
460
461
462
463
464
465
466
467
468
469
470
471
472
473
474
475
476
477
478
479
480
481
482
483
484
485
486
487
488
489
490
491
492
493
494
495
496
497
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
524
525
526
527
528
529
530
531
532
533
534
535
536
537
538
539
540
541
542
543
544
545
546
547
548
549
550
551
552
553
554
555
556
557
558
559
560
561
562
563
564
565
566
567
568
569
570
571
572
573
class PACC(BaseCalibratedQuantifier):
  """Probabilistic Adjusted Classify and Count (PACC) quantifier for binary problems.

  PACC is a calibrated quantification method that refines the Probabilistic 
  Classify and Count (PCC) by correcting for the base classifier's systematic 
  probabilistic bias. Using Out-of-Fold (OOF) predictions, it computes the 
  expected predicted probability for the positive class given the true class labels. 
  It then applies an analytical correction formula similar to ACC but entirely 
  based on continuous probability scores.

  Parameters
  ----------
  classifier : estimator object, default = None
    The classifier to be used as the base for quantification.
    Must implement the `predict_proba` method. If None, an instance of 
    `LogisticRegression()` will be created.
  cv : int, cross-validation generator or an iterable, default = 10
    Determines the cross-validation splitting strategy to generate the 
    Out-of-Fold predictions used for calibration.  
  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds (plus
    the final full-data classifier refit). See `BaseCalibratedQuantifier`.
  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs. See
    `BaseCalibratedQuantifier`.

  Attributes
  ----------
  classes_ : ndarray of shape (2,)
    The distinct class labels found during training.

  n_classes_ : int
    The number of distinct classes (expected to be 2).

  train_prevalence_ : ndarray of shape (2,)
    The prevalence of each class in the training dataset.

  y_prevs_ : ndarray of shape (2,)
    Alias for train_prevalence_ kept for backward compatibility.

  mu_pos_pos_ : float
    The mean probability assigned to the positive class for instances that 
    are truly positive, estimated out-of-fold. Analogue to TPR.

  mu_neg_pos_ : float
    The mean probability assigned to the positive class for instances that 
    are truly negative, estimated out-of-fold. Analogue to FPR.

  classifier_ : estimator object
    The fitted base classifier trained on the entire dataset.

  Notes
  -----
  The PACC method adjusts the observed probabilistic prevalence (p_pcc) using 
  the following analytical equation for the positive class:

    p_corrected = (p_pcc - mu_neg_pos) / (mu_pos_pos - mu_neg_pos)

  where 'p_pcc' is the average predicted probability of the positive class 
  in the test bag. The prevalence for the negative class is then derived as 
  1 - p_corrected. If the denominator evaluates to zero, the method falls 
  back to the unadjusted p_pcc.

  References
  ----------
  Antonio Bella, Cesar Ferri, José Hernández-Orallo, and María José Ramírez-Quintana.
  Quantification via probability estimators. In 2010 IEEE International Conference on
  Data Mining, pages 737-742, Sydney, Australia, 2010.

  Examples
  --------
  >>> from sklearn.datasets import make_classification
  >>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
  >>> quantifier = PACC()
  >>> quantifier.fit(X, y)
  >>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
  >>> prevalences = quantifier.predict(X_test)
  >>> print(prevalences)
  """

  def __init__(self, classifier: BaseEstimator = None, cv: int = 10,
               n_jobs: int = None, parallel_backend: str = "loky"):
    super().__init__(classifier=classifier, cv=cv, n_jobs=n_jobs, parallel_backend=parallel_backend)

  def fit(self, X: np.ndarray, y: np.ndarray) -> "PACC":
    """Fits the PACC quantifier, validating classifier and class-count constraints.

    Raises
    ------
    TypeError
      If the provided classifier does not support probability estimation.
    ValueError
      If `y` contains more than 2 distinct classes.
    """
    base_clf = self.classifier if self.classifier is not None else LogisticRegression()

    if not hasattr(base_clf, "predict_proba"):
      raise TypeError(
        f"The classifier {base_clf.__class__.__name__} does not "
        "support probability estimation. PACC requires 'predict_proba'."
      )

    if len(np.unique(y)) > 2:
      raise ValueError(
        "PACC method only works for binary quantification. Multiclass "
        "quantification is possible via OVR strategies, but not recommended due to "
        "theoretical issues with that approach."
      )

    return super().fit(X, y)

  def _get_oof_method(self) -> str:
    return "predict_proba"

  def _calibrate(self, y_true_oof: np.ndarray, y_pred_oof: np.ndarray):
    """Calculates the expected continuous scores (mu) using Out-of-Fold probabilities.

    Parameters
    ----------
    y_true_oof : ndarray of shape (n_samples,)
      True labels collected out-of-fold.
    y_pred_oof : ndarray of shape (n_samples, 2)
      Continuous probability matrices generated out-of-fold.
    """
    neg_label = self.classes_[0]
    pos_label = self.classes_[1]

    prob_pos_oof = y_pred_oof[:, 1]

    is_truly_pos = (y_true_oof == pos_label)
    is_truly_neg = (y_true_oof == neg_label)

    # mu_neg: probabilities mean attributed to the negative class
    self.mu_pos_pos_ = np.mean(prob_pos_oof[is_truly_pos]) if np.any(is_truly_pos) else 1.0
    self.mu_neg_pos_ = np.mean(prob_pos_oof[is_truly_neg]) if np.any(is_truly_neg) else 0.0

  def _quantify(self, X: np.ndarray) -> np.ndarray:
    """Applies the analytical continuous equation to adjust the probabilistic prevalence.

    Parameters
    ----------
    X : ndarray of shape (n_samples, n_features)
      Raw data from test bags.

    Returns
    -------
    p_pacc : ndarray of shape (2,)
      The adjusted prevalences for [negative_class, positive_class].
    """
    y_probas = self.classifier_.predict_proba(X)
    p_pcc = np.mean(y_probas[:, 1])

    denominator = self.mu_pos_pos_ - self.mu_neg_pos_

    if denominator == 0:
      p_pos_adjusted = p_pcc
    else:
      p_pos_adjusted = (p_pcc - self.mu_neg_pos_) / denominator

    return np.array([1.0 - p_pos_adjusted, p_pos_adjusted])

fit(X, y)

Fits the PACC quantifier, validating classifier and class-count constraints.

Raises:

Type Description
TypeError

If the provided classifier does not support probability estimation.

ValueError

If y contains more than 2 distinct classes.

Source code in quack/quantifiers/_baselines.py
498
499
500
501
502
503
504
505
506
507
508
509
510
511
512
513
514
515
516
517
518
519
520
521
522
523
def fit(self, X: np.ndarray, y: np.ndarray) -> "PACC":
  """Fits the PACC quantifier, validating classifier and class-count constraints.

  Raises
  ------
  TypeError
    If the provided classifier does not support probability estimation.
  ValueError
    If `y` contains more than 2 distinct classes.
  """
  base_clf = self.classifier if self.classifier is not None else LogisticRegression()

  if not hasattr(base_clf, "predict_proba"):
    raise TypeError(
      f"The classifier {base_clf.__class__.__name__} does not "
      "support probability estimation. PACC requires 'predict_proba'."
    )

  if len(np.unique(y)) > 2:
    raise ValueError(
      "PACC method only works for binary quantification. Multiclass "
      "quantification is possible via OVR strategies, but not recommended due to "
      "theoretical issues with that approach."
    )

  return super().fit(X, y)

Generalized Adjusted Classify & Count (GAC)

quack.quantifiers._dmm.GAC

Bases: BaseCalibratedQuantifier, BaseMixtureQuantifier

Generalized Adjusting Confusion Matrix (GAC) Quantifier.

A distance-minimizing multi-class generalization of the Adjusting Count (AC) algorithm that optimizes target distributions across the discrete labels confusion matrix profile.

Parameters:

Name Type Description Default
classifier estimator object

The underlying base classifier. Defaults to LogisticRegression().

= LogisticRegression
distance_metric str

The distance metric minimized.

= 'L2'
cv int

The number of cross-validation folds.

= 10
use_convex_solver bool

If True, optimizes via CVXPY.

= True
n_jobs int

Number of jobs to run in parallel while fitting the cv folds.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"
References

Aykut Firat. Unified framework for quantification. arXiv preprint arXiv:1606.00868, 2016.

Source code in quack/quantifiers/_dmm.py
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
271
272
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
class GAC(BaseCalibratedQuantifier, BaseMixtureQuantifier):
  """Generalized Adjusting Confusion Matrix (GAC) Quantifier.

  A distance-minimizing multi-class generalization of the Adjusting Count (AC) 
  algorithm that optimizes target distributions across the discrete labels 
  confusion matrix profile.

  Parameters
  ----------
  classifier : estimator object, default = LogisticRegression
    The underlying base classifier. Defaults to `LogisticRegression()`.

  distance_metric : str, default = 'L2'
    The distance metric minimized.

  cv : int, default = 10
    The number of cross-validation folds.

  use_convex_solver : bool, default = True
    If True, optimizes via CVXPY.

  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds.

  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  References
  ----------
  Aykut Firat. Unified framework for quantification. arXiv preprint arXiv:1606.00868, 2016.
  """

  def __init__(self,
               classifier: BaseEstimator = LogisticRegression(),
               distance_metric: str = "L2", 
               cv: int = 10,
               use_convex_solver: bool = True,
               n_jobs: int = None,
               parallel_backend: str = "loky"):
    BaseCalibratedQuantifier.__init__(self, classifier=classifier, cv=cv,
                                      n_jobs=n_jobs, parallel_backend=parallel_backend)
    BaseMixtureQuantifier.__init__(self, classifier=classifier, distance_metric=distance_metric, 
                                  use_convex_solver=use_convex_solver)

  def _get_oof_method(self) -> str:
    return "predict"

  def _calibrate(self, y_true_oof: np.ndarray, y_pred_oof: np.ndarray):
    # fully vectorized confusion matrix: one-hot encode both true and
    # predicted labels against the sorted classes_ grid, then a single
    # matmul (n_classes, n_samples) @ (n_samples, n_classes) tallies
    # every (pred, true) pair at once, replacing the double Python loop
    # this used to run over classes_ x classes_
    true_idx = np.searchsorted(self.classes_, y_true_oof)
    pred_idx = np.searchsorted(self.classes_, y_pred_oof)
    one_hot_true = np.eye(self.n_classes_)[true_idx]
    one_hot_pred = np.eye(self.n_classes_)[pred_idx]
    confusion_matrix = one_hot_pred.T @ one_hot_true

    _, class_counts = np.unique(y_true_oof, return_counts=True)
    self.conditional_matrix_ = confusion_matrix / class_counts

  def _compute_score(self, X: np.ndarray) -> np.ndarray:
    y_predictions = self.classifier_.predict(X)
    return np.array([np.mean(y_predictions == class_label) for class_label in self.classes_])

  def _quantify(self, X: np.ndarray) -> np.ndarray:
    test_frequencies = self._compute_score(X)
    return self._solve_mixture(test_frequencies)

Generalized Probabilistic Adjusted Classify & Count (GPAC)

quack.quantifiers._dmm.GPAC

Bases: BaseCalibratedQuantifier, BaseMixtureQuantifier

Generalized Probabilistic Adjusting Confusion Matrix (GPAC) Quantifier.

A distance-minimizing multi-class generalization of the Probabilistic Adjusting Count (PAC) algorithm that optimizes target distributions using soft-probability profiles.

Parameters:

Name Type Description Default
classifier estimator object

The underlying base classifier. Defaults to LogisticRegression().

= LogisticRegression
distance_metric str

The distance metric minimized.

= 'L2'
cv int

The number of cross-validation folds.

= 10
use_convex_solver bool

If True, optimizes via CVXPY.

= True
n_jobs int

Number of jobs to run in parallel while fitting the cv folds.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"
References

Aykut Firat. Unified framework for quantification. arXiv preprint arXiv:1606.00868, 2016.

Source code in quack/quantifiers/_dmm.py
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
class GPAC(BaseCalibratedQuantifier, BaseMixtureQuantifier):
  """Generalized Probabilistic Adjusting Confusion Matrix (GPAC) Quantifier.

  A distance-minimizing multi-class generalization of the Probabilistic Adjusting 
  Count (PAC) algorithm that optimizes target distributions using soft-probability profiles.

  Parameters
  ----------
  classifier : estimator object, default = LogisticRegression
    The underlying base classifier. Defaults to `LogisticRegression()`.

  distance_metric : str, default = 'L2'
    The distance metric minimized.

  cv : int, default = 10
    The number of cross-validation folds.

  use_convex_solver : bool, default = True
    If True, optimizes via CVXPY.

  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds.

  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  References
  ----------
  Aykut Firat. Unified framework for quantification. arXiv preprint arXiv:1606.00868, 2016.
  """

  def __init__(self,
               classifier: BaseEstimator = LogisticRegression(),
               distance_metric: str = "L2", 
               cv: int = 10,
               use_convex_solver: bool = True,
               n_jobs: int = None,
               parallel_backend: str = "loky"):
    BaseCalibratedQuantifier.__init__(self, classifier=classifier, cv=cv,
                                      n_jobs=n_jobs, parallel_backend=parallel_backend)
    BaseMixtureQuantifier.__init__(self, classifier=classifier, distance_metric=distance_metric, 
                                  use_convex_solver=use_convex_solver)

  def _get_oof_method(self) -> str:
    return "predict_proba"

  def _calibrate(self, y_true_oof: np.ndarray, y_pred_oof: np.ndarray):
    # single matmul replaces the loop over classes_: column l of the
    # result is the sum of predicted-probability rows whose true label is
    # class l, i.e. y_pred_oof.T @ one_hot(true_labels)
    true_idx = np.searchsorted(self.classes_, y_true_oof)
    one_hot_true = np.eye(self.n_classes_)[true_idx]
    probabilistic_matrix = y_pred_oof.T @ one_hot_true

    _, class_counts = np.unique(y_true_oof, return_counts=True)
    self.conditional_matrix_ = probabilistic_matrix / class_counts

  def _compute_score(self, X: np.ndarray) -> np.ndarray:
    return self.classifier_.predict_proba(X).sum(axis=0) / X.shape[0]

  def _quantify(self, X: np.ndarray) -> np.ndarray:
    test_frequencies = self._compute_score(X)
    return self._solve_mixture(test_frequencies)

Friedman's Method (FM)

quack.quantifiers._dmm.FM

Bases: BaseCalibratedQuantifier, BaseMixtureQuantifier

Friedman's Method (FM) Quantifier.

An adjusting prediction mixture model that maps soft classifier probabilities into binary indicator matrices by comparing them against baseline training priors.

Parameters:

Name Type Description Default
classifier estimator object

The underlying base classifier. Defaults to LogisticRegression().

= LogisticRegression
distance_metric str

The distance metric minimized.

= 'L2'
cv int

The number of cross-validation folds.

= 10
use_convex_solver bool

If True, optimizes via CVXPY.

= True
n_jobs int

Number of jobs to run in parallel while fitting the cv folds.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"
References

Jerome H. Friedman. Class counts in future unlabeled samples, 2014. Presentation at MIT CSAIL Big Data Event.

Source code in quack/quantifiers/_dmm.py
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
434
435
436
437
438
class FM(BaseCalibratedQuantifier, BaseMixtureQuantifier):
  """Friedman's Method (FM) Quantifier.

  An adjusting prediction mixture model that maps soft classifier probabilities 
  into binary indicator matrices by comparing them against baseline training priors.

  Parameters
  ----------
  classifier : estimator object, default = LogisticRegression
    The underlying base classifier. Defaults to `LogisticRegression()`.

  distance_metric : str, default = 'L2'
    The distance metric minimized.

  cv : int, default = 10
    The number of cross-validation folds.

  use_convex_solver : bool, default = True
    If True, optimizes via CVXPY.

  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds.

  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  References
  ----------
  Jerome H. Friedman. Class counts in future unlabeled samples, 2014.
  Presentation at MIT CSAIL Big Data Event.
  """

  def __init__(self,
               classifier: BaseEstimator = LogisticRegression(),
               distance_metric: str = "L2",
               cv: int = 10,
               use_convex_solver: bool = True,
               n_jobs: int = None,
               parallel_backend: str = "loky"):
    BaseCalibratedQuantifier.__init__(self, classifier=classifier, cv=cv,
                                      n_jobs=n_jobs, parallel_backend=parallel_backend)
    BaseMixtureQuantifier.__init__(self, classifier=classifier, distance_metric=distance_metric, 
                                  use_convex_solver=use_convex_solver)

  def _get_oof_method(self) -> str:
    return "predict_proba"

  def _calibrate(self, y_true_oof: np.ndarray, y_pred_oof: np.ndarray):
    # single matmul replaces the loop over classes_, mirroring GPAC's
    # vectorization but comparing against train_prevalence_ first
    true_idx = np.searchsorted(self.classes_, y_true_oof)
    one_hot_true = np.eye(self.n_classes_)[true_idx]
    above_prior = (y_pred_oof > self.train_prevalence_).astype(float)
    threshold_matrix = above_prior.T @ one_hot_true

    _, class_counts = np.unique(y_true_oof, return_counts=True)
    self.conditional_matrix_ = threshold_matrix / class_counts

  def _compute_score(self, X: np.ndarray) -> np.ndarray:
    return np.sum(self.classifier_.predict_proba(X) > self.train_prevalence_, axis=0) / X.shape[0]

  def _quantify(self, X: np.ndarray) -> np.ndarray:
    test_frequencies = self._compute_score(X)
    return self._solve_mixture(test_frequencies)

Threshold Selector X (X)

quack.quantifiers._threshold.X

Bases: BaseThresholdQuantifier

Forman's X threshold selection quantifier matching QFY's TSX strategy.

This method selects the optimal threshold from the dynamic grid that minimizes the absolute distance |TPR - (1 - FPR)|, searching for the intersection point in the ROC curve where TPR + FPR ~= 1.0.

Parameters:

Name Type Description Default
classifier estimator object

The classifier to be used as the base for quantification.

= None
cv int

Determines the cross-validation splitting strategy.

= 10
precision int

The decimal precision used to round the Out-of-Fold probabilities.

= 3
n_jobs int

Number of jobs to run in parallel while fitting the cv folds. See BaseCalibratedQuantifier.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"
References

Forman, G. (2008). Quantifying counts and costs via classification. Data Mining and Knowledge Discovery, 17(2):164-206.

Examples:

>>> from sklearn.datasets import make_classification
>>> X_, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
>>> quantifier = X()
>>> quantifier.fit(X_, y)
>>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
>>> prevalences = quantifier.predict(X_test)
>>> print(prevalences)
Source code in quack/quantifiers/_threshold.py
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
class X(BaseThresholdQuantifier):
  """Forman's X threshold selection quantifier matching QFY's TSX strategy.

  This method selects the optimal threshold from the dynamic grid that 
  minimizes the absolute distance |TPR - (1 - FPR)|, searching for the 
  intersection point in the ROC curve where TPR + FPR ~= 1.0.

  Parameters
  ----------
  classifier : estimator object, default = None
    The classifier to be used as the base for quantification.
  cv : int, default = 10
    Determines the cross-validation splitting strategy.
  precision : int, default = 3
    The decimal precision used to round the Out-of-Fold probabilities.
  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds. See
    `BaseCalibratedQuantifier`.
  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  References
  ----------
  Forman, G. (2008). Quantifying counts and costs via classification. 
  Data Mining and Knowledge Discovery, 17(2):164-206.

  Examples
  --------
  >>> from sklearn.datasets import make_classification
  >>> X_, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
  >>> quantifier = X()
  >>> quantifier.fit(X_, y)
  >>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
  >>> prevalences = quantifier.predict(X_test)
  >>> print(prevalences)
  """

  def _quantify(self, X_test: np.ndarray) -> np.ndarray:
    # minimizes the absolute distance |TPR - (1 - FPR)|
    idx = np.argmin(np.abs(self.tpr_by_thresh_ - (1.0 - self.fpr_by_thresh_)))

    t_chosen = self.thresholds_[idx]
    tpr = self.tpr_by_thresh_[idx]
    fpr = self.fpr_by_thresh_[idx]

    pos_probs_test = self.classifier_.predict_proba(X_test)[:, 1]
    p_raw_pos = np.mean(pos_probs_test >= t_chosen)

    p_adj_pos = self._apply_acc_formula(p_raw_pos, tpr, fpr)
    return np.array([1.0 - p_adj_pos, p_adj_pos])

Threshold Selector MAX (Max)

quack.quantifiers._threshold.Max

Bases: BaseThresholdQuantifier

Forman's Max threshold selection quantifier matching QFY's TSMax strategy.

This method selects the threshold that maximizes the separation split between classes, maximizing the absolute difference |TPR - FPR| to achieve the highest denominator stability during shift adjustment.

Parameters:

Name Type Description Default
classifier estimator object

The classifier to be used as the base for quantification.

= None
cv int

Determines the cross-validation splitting strategy.

= 10
precision int

The decimal precision used to round the Out-of-Fold probabilities.

= 3
n_jobs int

Number of jobs to run in parallel while fitting the cv folds. See BaseCalibratedQuantifier.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"
References

Forman, G. (2008). Quantifying counts and costs via classification. Data Mining and Knowledge Discovery, 17(2):164-206.

Examples:

>>> from sklearn.datasets import make_classification
>>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
>>> quantifier = Max()
>>> quantifier.fit(X, y)
>>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
>>> prevalences = quantifier.predict(X_test)
>>> print(prevalences)
Source code in quack/quantifiers/_threshold.py
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
class Max(BaseThresholdQuantifier):
  """Forman's Max threshold selection quantifier matching QFY's TSMax strategy.

  This method selects the threshold that maximizes the separation split 
  between classes, maximizing the absolute difference |TPR - FPR| to achieve 
  the highest denominator stability during shift adjustment.

  Parameters
  ----------
  classifier : estimator object, default = None
    The classifier to be used as the base for quantification.
  cv : int, default = 10
    Determines the cross-validation splitting strategy.
  precision : int, default = 3
    The decimal precision used to round the Out-of-Fold probabilities.
  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds. See
    `BaseCalibratedQuantifier`.
  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  References
  ----------
  Forman, G. (2008). Quantifying counts and costs via classification. 
  Data Mining and Knowledge Discovery, 17(2):164-206.

  Examples
  --------
  >>> from sklearn.datasets import make_classification
  >>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
  >>> quantifier = Max()
  >>> quantifier.fit(X, y)
  >>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
  >>> prevalences = quantifier.predict(X_test)
  >>> print(prevalences)
  """

  def _quantify(self, X_test: np.ndarray) -> np.ndarray:
    # maximizes the separation split |TPR - FPR|
    idx = np.argmax(np.abs(self.tpr_by_thresh_ - self.fpr_by_thresh_))

    t_chosen = self.thresholds_[idx]
    tpr = self.tpr_by_thresh_[idx]
    fpr = self.fpr_by_thresh_[idx]

    pos_probs_test = self.classifier_.predict_proba(X_test)[:, 1]
    p_raw_pos = np.mean(pos_probs_test >= t_chosen)

    p_adj_pos = self._apply_acc_formula(p_raw_pos, tpr, fpr)
    return np.array([1.0 - p_adj_pos, p_adj_pos])

Threshold Selector 50 (T50)

quack.quantifiers._threshold.T50

Bases: BaseThresholdQuantifier

Forman's Threshold 50 quantifier matching QFY's TS50 strategy.

This method selects the threshold that is closest to achieving a True Positive Rate (TPR) of exactly 0.5, minimizing the cost function |TPR - 0.5| across the candidate grid.

Parameters:

Name Type Description Default
classifier estimator object

The classifier to be used as the base for quantification.

= None
cv int

Determines the cross-validation splitting strategy.

= 10
precision int

The decimal precision used to round the Out-of-Fold probabilities.

= 3
n_jobs int

Number of jobs to run in parallel while fitting the cv folds. See BaseCalibratedQuantifier.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"
References

Forman, G. (2008). Quantifying counts and costs via classification. Data Mining and Knowledge Discovery, 17(2):164-206.

Examples:

>>> from sklearn.datasets import make_classification
>>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
>>> quantifier = T50()
>>> quantifier.fit(X, y)
>>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
>>> prevalences = quantifier.predict(X_test)
>>> print(prevalences)
Source code in quack/quantifiers/_threshold.py
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
class T50(BaseThresholdQuantifier):
  """Forman's Threshold 50 quantifier matching QFY's TS50 strategy.

  This method selects the threshold that is closest to achieving a 
  True Positive Rate (TPR) of exactly 0.5, minimizing the cost function 
  |TPR - 0.5| across the candidate grid.

  Parameters
  ----------
  classifier : estimator object, default = None
    The classifier to be used as the base for quantification.
  cv : int, default = 10
    Determines the cross-validation splitting strategy.
  precision : int, default = 3
    The decimal precision used to round the Out-of-Fold probabilities.
  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds. See
    `BaseCalibratedQuantifier`.
  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  References
  ----------
  Forman, G. (2008). Quantifying counts and costs via classification. 
  Data Mining and Knowledge Discovery, 17(2):164-206.

  Examples
  --------
  >>> from sklearn.datasets import make_classification
  >>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
  >>> quantifier = T50()
  >>> quantifier.fit(X, y)
  >>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
  >>> prevalences = quantifier.predict(X_test)
  >>> print(prevalences)
  """

  def _quantify(self, X_test: np.ndarray) -> np.ndarray:
    # minimizes abs(tpr - 0.5)
    idx = np.argmin(np.abs(self.tpr_by_thresh_ - 0.5))

    t_chosen = self.thresholds_[idx]
    tpr = self.tpr_by_thresh_[idx]
    fpr = self.fpr_by_thresh_[idx]

    pos_probs_test = self.classifier_.predict_proba(X_test)[:, 1]
    p_raw_pos = np.mean(pos_probs_test >= t_chosen)

    p_adj_pos = self._apply_acc_formula(p_raw_pos, tpr, fpr)
    return np.array([1.0 - p_adj_pos, p_adj_pos])

MedianSweep

quack.quantifiers._threshold.MedianSweep

Bases: BaseThresholdQuantifier

Forman's Median Sweep (MS) quantifier faithful to QFY's implementation.

Median Sweep evaluates individual ACC adjustments across all candidates in the dynamic grid. It enforces a strict filter where thresholds with a denominator (TPR - FPR) smaller than delta_min are ignored. If no thresholds satisfy the filter, it falls back to the estimate with the largest available separation. Otherwise, it returns the robust median of all valid predictions.

Parameters:

Name Type Description Default
classifier estimator object

The classifier to be used as the base for quantification.

None
cv int

Determines the cross-validation splitting strategy.

10
precision int

The decimal precision used to round the Out-of-Fold probabilities.

3
delta_min float

The minimum required threshold separation (TPR - FPR) to accept an individual ACC adaptation, avoiding denominator instability.

0.25
n_jobs int

Number of jobs to run in parallel while fitting the cv folds. See BaseCalibratedQuantifier.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"
Notes

The fallback path (used only when no threshold satisfies delta > delta_min) picks the threshold with the largest delta among all candidates: as long as no threshold is ever accepted as valid, every candidate is a fallback contender, so the largest-delta candidate found while sweeping the whole grid is exactly the one a sequential scan would have kept until the end.

References

Forman, G. (2008). Quantifying counts and costs via classification. Data Mining and Knowledge Discovery, 17(2):164-206.

Examples:

>>> from sklearn.datasets import make_classification
>>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
>>> quantifier = MedianSweep()
>>> quantifier.fit(X, y)
>>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
>>> prevalences = quantifier.predict(X_test)
>>> print(prevalences)
Source code in quack/quantifiers/_threshold.py
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
class MedianSweep(BaseThresholdQuantifier):
  """Forman's Median Sweep (MS) quantifier faithful to QFY's implementation.

  Median Sweep evaluates individual ACC adjustments across all candidates in the 
  dynamic grid. It enforces a strict filter where thresholds with a denominator
  (TPR - FPR) smaller than `delta_min` are ignored.
  If no thresholds satisfy the filter, it falls back to the estimate with the 
  largest available separation. Otherwise, it returns the robust median of all 
  valid predictions.

  Parameters
  ----------
  classifier : estimator object, default=None
    The classifier to be used as the base for quantification.
  cv : int, default=10
    Determines the cross-validation splitting strategy.
  precision : int, default=3
    The decimal precision used to round the Out-of-Fold probabilities.
  delta_min : float, default=0.25
    The minimum required threshold separation (TPR - FPR) to accept an 
    individual ACC adaptation, avoiding denominator instability.
  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds. See
    `BaseCalibratedQuantifier`.
  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  Notes
  -----
  The fallback path (used only when **no** threshold satisfies
  `delta > delta_min`) picks the threshold with the largest `delta` among
  *all* candidates: as long as no threshold is ever accepted as valid,
  every candidate is a fallback contender, so the largest-`delta`
  candidate found while sweeping the whole grid is exactly the one a
  sequential scan would have kept until the end.

  References
  ----------
  Forman, G. (2008). Quantifying counts and costs via classification. 
  Data Mining and Knowledge Discovery, 17(2):164-206.

  Examples
  --------
  >>> from sklearn.datasets import make_classification
  >>> X, y = make_classification(n_samples=1000, n_classes=2, random_state=42)
  >>> quantifier = MedianSweep()
  >>> quantifier.fit(X, y)
  >>> X_test, _ = make_classification(n_samples=200, n_classes=2, random_state=7)
  >>> prevalences = quantifier.predict(X_test)
  >>> print(prevalences)
  """

  def __init__(self, classifier: BaseEstimator = None, cv: int = 10, precision: int = 3,
               delta_min: float = 0.25, n_jobs: int = None, parallel_backend: str = "loky"):
    super().__init__(classifier=classifier, cv=cv, precision=precision,
                     n_jobs=n_jobs, parallel_backend=parallel_backend)
    self.delta_min = delta_min

  def _quantify(self, X_test: np.ndarray) -> np.ndarray:
    pos_probs_test = self.classifier_.predict_proba(X_test)[:, 1]
    delta = self.tpr_by_thresh_ - self.fpr_by_thresh_

    # vectorized ACC adjustment for every threshold at once: broadcasting
    # (n_test, n_thresholds) comparisons, then averaging each column gives
    # the raw positive rate at every threshold in one shot
    p_raw_pos_by_thresh = np.mean(
      pos_probs_test[:, np.newaxis] >= self.thresholds_[np.newaxis, :], axis=0
    )

    valid_mask = delta > self.delta_min

    if np.any(valid_mask):
      p_adj_valid = (p_raw_pos_by_thresh[valid_mask] - self.fpr_by_thresh_[valid_mask]) / delta[valid_mask]
      p_adj_pos = np.median(np.clip(p_adj_valid, 0.0, 1.0))
    else:
      # no threshold cleared delta_min: fall back to the single threshold
      # with the largest available separation across the whole grid
      # (equivalent to the sequential scan's behavior in this scenario,
      # see the class-level Notes section)
      idx = np.argmax(delta)
      if delta[idx] == 0:
        p_max_fallback = self.tpr_by_thresh_[idx]
      else:
        p_max_fallback = (p_raw_pos_by_thresh[idx] - self.fpr_by_thresh_[idx]) / delta[idx]
      p_adj_pos = np.clip(p_max_fallback, 0.0, 1.0)

    return np.array([1.0 - p_adj_pos, p_adj_pos])

HDx

quack.quantifiers._features.HDx

Bases: BaseMixtureQuantifier

Hellinger Distance x (HDx) quantifier.

HDx is a non-parametric feature-space mixture model that operates directly on categorical or discretized continuous features without training an underlying classifier. It projects each feature column independently, constructs a global marginal conditional probability matrix during training, and minimizes the Hellinger Distance to estimate the test class prevalences.

Parameters:

Name Type Description Default
use_convex_solver bool

If True, attempts to solve the statistical mixture distribution using cvxpy. If False, falls back to the Golden Section Search numerical solver.

True

Attributes:

Name Type Description
classes_ ndarray of shape (n_classes,)

The distinct class labels found during the training phase.

n_classes_ int

The total number of unique classes.

train_prevalence_ ndarray of shape (n_classes,)

The baseline prevalence proportion of each class observed in the training data.

feature_spaces_ list of ndarray

A list of length n_features, where each element contains the unique sorted values observed for that specific feature column during training.

conditional_matrix_ ndarray of shape (n_total_unique_bins, n_classes)

The stacked conditional probability matrix built during the fit phase. Represents the marginal distribution profiles for each class.

References

Víctor González-Castro, Rocío Alaiz-Rodríguez, and Enrique Alegre. Class distribution estimation based on the Hellinger distance. Information Sciences, 218(1):146-164, 2013

Source code in quack/quantifiers/_features.py
  9
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
class HDx(BaseMixtureQuantifier):
  """Hellinger Distance x (HDx) quantifier.

  HDx is a non-parametric feature-space mixture model that operates directly on 
  categorical or discretized continuous features without training an underlying 
  classifier. It projects each feature column independently, constructs a global 
  marginal conditional probability matrix during training, and minimizes the 
  Hellinger Distance to estimate the test class prevalences.

  Parameters
  ----------
  use_convex_solver : bool, default=True
    If True, attempts to solve the statistical mixture distribution using `cvxpy`.
    If False, falls back to the Golden Section Search numerical solver.

  Attributes
  ----------
  classes_ : ndarray of shape (n_classes,)
    The distinct class labels found during the training phase.

  n_classes_ : int
    The total number of unique classes.

  train_prevalence_ : ndarray of shape (n_classes,)
    The baseline prevalence proportion of each class observed in the training data.

  feature_spaces_ : list of ndarray
    A list of length `n_features`, where each element contains the unique sorted 
    values observed for that specific feature column during training.

  conditional_matrix_ : ndarray of shape (n_total_unique_bins, n_classes)
    The stacked conditional probability matrix built during the `fit` phase.
    Represents the marginal distribution profiles for each class.

  References
  ----------
  Víctor González-Castro, Rocío Alaiz-Rodríguez, and Enrique Alegre. Class distribution
  estimation based on the Hellinger distance. Information Sciences, 218(1):146-164, 2013
  """
  def __init__(self, use_convex_solver: bool = True):
    # HDx operates on features directly (classifier=None) and strictly uses Hellinger Distance ("HD")
    super().__init__(classifier=None,
                     distance_metric="HD",
                     use_convex_solver=use_convex_solver)
    self.feature_spaces_ = None

  def fit(self, X: np.ndarray, y: np.ndarray) -> 'HDx':
    """Fits the HDx mixture model by building the marginal conditional matrix.

    Parameters
    ----------
    X : ndarray of shape (n_samples, n_features)
      The training feature matrix.
    y : ndarray of shape (n_samples,)
      The target class labels.

    Returns
    -------
    self : object
      Returns the instance itself.
    """
    X, y = check_X_y(X, y, accept_sparse=False)

    self.classes_, class_counts = np.unique(y, return_counts=True)
    self.n_classes_ = len(self.classes_)
    self.train_prevalence_ = class_counts / len(y)

    if self.n_classes_ < 2:
      raise ValueError("HDx requires at least 2 distinct classes to fit.")

    n_features = X.shape[1]
    class_idx = np.searchsorted(self.classes_, y)

    # map and store the unique token space for each individual feature column
    self.feature_spaces_ = [np.unique(X[:, j]) for j in range(n_features)]

    # build the conditional matrix (CM): the loop over features is
    # unavoidable (each column has a different number of unique values,
    # so the blocks can't be stacked into a single rectangular operation),
    # but within each feature the value x class crosstab is fully
    # vectorized via a combined-index bincount instead of the previous
    # nested (class x unique_value) Python loop
    conditional_blocks = []
    for j in range(n_features):
      unique_values = self.feature_spaces_[j]
      val_idx = np.searchsorted(unique_values, X[:, j])

      combined_idx = val_idx * self.n_classes_ + class_idx
      counts_flat = np.bincount(combined_idx, minlength=len(unique_values) * self.n_classes_)
      crosstab_counts = counts_flat.reshape(len(unique_values), self.n_classes_)

      # normalize counts by each class size to form conditional probabilities
      conditional_blocks.append(crosstab_counts / class_counts)

    # vertically stack all independent column representations into a single global system matrix
    self.conditional_matrix_ = np.vstack(conditional_blocks)

    return self

  def _compute_score(self, X: np.ndarray) -> np.ndarray:
    """Extracts the empirical marginal test frequencies across all features.

    Calculates the relative sample distribution frequency over the saved 
    training feature spaces for the incoming test batch.

    Parameters
    ----------
    X : ndarray of shape (n_samples, n_features)
      The testing feature matrix.

    Returns
    -------
    test_frequencies : ndarray of shape (n_total_unique_bins,)
      The stacked vector representing empirical marginal frequencies of the test batch.
    """
    n_samples = X.shape[0]
    n_features = X.shape[1]
    frequencies_list = []

    # compute marginal frequencies matching the exact bins established
    # during training; the per-feature loop remains for the same reason
    # as in fit (ragged bin counts across columns), but the inner
    # per-unique-value counting is now a single searchsorted + bincount
    # pass instead of one np.count_nonzero comparison per unique value.
    # Test values absent from the trained feature space (i.e. not an
    # exact match to any trained unique value) contribute 0, exactly
    # matching the original np.count_nonzero(X[:, j] == val) semantics.
    for j in range(n_features):
      unique_values = self.feature_spaces_[j]
      col = X[:, j]

      idx = np.searchsorted(unique_values, col)
      idx_clipped = np.clip(idx, 0, len(unique_values) - 1)
      exact_match = unique_values[idx_clipped] == col

      counts = np.bincount(idx_clipped[exact_match], minlength=len(unique_values))
      frequencies_list.append(counts / n_samples)

    # combine the individual frequency vectors into the stacked global vector
    return np.hstack(frequencies_list)

fit(X, y)

Fits the HDx mixture model by building the marginal conditional matrix.

Parameters:

Name Type Description Default
X ndarray of shape (n_samples, n_features)

The training feature matrix.

required
y ndarray of shape (n_samples,)

The target class labels.

required

Returns:

Name Type Description
self object

Returns the instance itself.

Source code in quack/quantifiers/_features.py
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def fit(self, X: np.ndarray, y: np.ndarray) -> 'HDx':
  """Fits the HDx mixture model by building the marginal conditional matrix.

  Parameters
  ----------
  X : ndarray of shape (n_samples, n_features)
    The training feature matrix.
  y : ndarray of shape (n_samples,)
    The target class labels.

  Returns
  -------
  self : object
    Returns the instance itself.
  """
  X, y = check_X_y(X, y, accept_sparse=False)

  self.classes_, class_counts = np.unique(y, return_counts=True)
  self.n_classes_ = len(self.classes_)
  self.train_prevalence_ = class_counts / len(y)

  if self.n_classes_ < 2:
    raise ValueError("HDx requires at least 2 distinct classes to fit.")

  n_features = X.shape[1]
  class_idx = np.searchsorted(self.classes_, y)

  # map and store the unique token space for each individual feature column
  self.feature_spaces_ = [np.unique(X[:, j]) for j in range(n_features)]

  # build the conditional matrix (CM): the loop over features is
  # unavoidable (each column has a different number of unique values,
  # so the blocks can't be stacked into a single rectangular operation),
  # but within each feature the value x class crosstab is fully
  # vectorized via a combined-index bincount instead of the previous
  # nested (class x unique_value) Python loop
  conditional_blocks = []
  for j in range(n_features):
    unique_values = self.feature_spaces_[j]
    val_idx = np.searchsorted(unique_values, X[:, j])

    combined_idx = val_idx * self.n_classes_ + class_idx
    counts_flat = np.bincount(combined_idx, minlength=len(unique_values) * self.n_classes_)
    crosstab_counts = counts_flat.reshape(len(unique_values), self.n_classes_)

    # normalize counts by each class size to form conditional probabilities
    conditional_blocks.append(crosstab_counts / class_counts)

  # vertically stack all independent column representations into a single global system matrix
  self.conditional_matrix_ = np.vstack(conditional_blocks)

  return self

ReadMe

quack.quantifiers._features.ReadMe

Bases: BaseQuantifier

ReadMe Ensemble Quantifier.

ReadMe is an ensemble mixture model specifically designed for high-dimensional categorical data or short text analysis (e.g., Bag-of-Words). It circumvents the curse of dimensionality by training multiple independent sub-space mixture models over randomized feature subsets, obtaining final test prevalences by averaging individual predictions.

Parameters:

Name Type Description Default
distance_metric str

The distance metric minimized by internal sub-quantifiers ('L1', 'L2', 'HD', 'TS').

'L2'
use_convex_solver bool

If True, internal sub-quantifiers utilize cvxpy optimization.

True
n_features int

Number of random features selected per subset. If None, it automatically defaults to max(int(D/5), 2) or bit length depending on dataset dimensionality.

None
n_subsets int

The total number of random subspace sub-quantifiers to ensemble.

100
n_jobs int

Number of jobs to run in parallel while fitting/predicting the n_subsets independent sub-quantifiers, since none of them depend on each other. None means sequential (matching the previous behavior); -1 uses all available processors. See joblib.Parallel.

= None
parallel_backend str

joblib.Parallel backend used for the subspace jobs ("loky" for process-based parallelism, "threading" for thread-based).

= "loky"
random_state int, RandomState instance or None

Controls the randomness of the per-subset feature selection. Pass an int for reproducible subspaces across repeated fit calls.

= None

Attributes:

Name Type Description
classes_ ndarray of shape (n_classes,)

The distinct class labels found during the training phase.

n_classes_ int

The total number of unique classes.

train_prevalence_ ndarray of shape (n_classes,)

The baseline prevalence proportion of each class observed in the training data.

feature_subsets_ list of ndarray

A list containing the chosen feature column indices for each random subset.

sub_quantifiers_ list of _RawSubspaceMixture

The collection of fitted internal mixture models making up the ensemble.

References

Hopkins, D., & King, G. (2010). A method of automated nonparametric content analysis for social science. American Journal of Political Science, 54(1), 229-247.

Source code in quack/quantifiers/_features.py
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
302
303
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
321
322
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
346
347
348
349
350
351
352
353
354
355
356
357
358
359
360
361
362
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
409
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
class ReadMe(BaseQuantifier):
  """ReadMe Ensemble Quantifier.

  ReadMe is an ensemble mixture model specifically designed for high-dimensional 
  categorical data or short text analysis (e.g., Bag-of-Words). It circumvents the 
  curse of dimensionality by training multiple independent sub-space mixture models 
  over randomized feature subsets, obtaining final test prevalences by averaging 
  individual predictions.

  Parameters
  ----------
  distance_metric : str, default='L2'
    The distance metric minimized by internal sub-quantifiers ('L1', 'L2', 'HD', 'TS').

  use_convex_solver : bool, default=True
    If True, internal sub-quantifiers utilize `cvxpy` optimization.

  n_features : int, default=None
    Number of random features selected per subset. If None, it automatically 
    defaults to `max(int(D/5), 2)` or bit length depending on dataset dimensionality.

  n_subsets : int, default=100
    The total number of random subspace sub-quantifiers to ensemble.

  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting/predicting the
    `n_subsets` independent sub-quantifiers, since none of them depend
    on each other. `None` means sequential (matching the previous
    behavior); `-1` uses all available processors. See `joblib.Parallel`.

  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the subspace jobs (`"loky"` for
    process-based parallelism, `"threading"` for thread-based).

  random_state : int, RandomState instance or None, default = None
    Controls the randomness of the per-subset feature selection. Pass an
    int for reproducible subspaces across repeated `fit` calls.

  Attributes
  ----------
  classes_ : ndarray of shape (n_classes,)
    The distinct class labels found during the training phase.

  n_classes_ : int
    The total number of unique classes.

  train_prevalence_ : ndarray of shape (n_classes,)
    The baseline prevalence proportion of each class observed in the training data.

  feature_subsets_ : list of ndarray
    A list containing the chosen feature column indices for each random subset.

  sub_quantifiers_ : list of _RawSubspaceMixture
    The collection of fitted internal mixture models making up the ensemble.

  References
  ----------
  Hopkins, D., & King, G. (2010). A method of automated nonparametric content 
  analysis for social science. American Journal of Political Science, 54(1), 229-247.
  """

  def __init__(self,
               distance_metric: str = "L2",
               use_convex_solver: bool = True, 
               n_features: int = None,
               n_subsets: int = 100,
               n_jobs: int = None,
               parallel_backend: str = "loky",
               random_state=None):
    # ReadMe manages an internal collection of sub-quantifiers, bypassing a single core classifier
    super().__init__(classifier=None)
    self.distance_metric = distance_metric
    self.use_convex_solver = use_convex_solver
    self.n_features = n_features
    self.n_subsets = n_subsets
    self.n_jobs = n_jobs
    self.parallel_backend = parallel_backend
    self.random_state = random_state
    self.feature_subsets_ = []
    self.sub_quantifiers_ = []

  def fit(self, X: np.ndarray, y: np.ndarray) -> 'ReadMe':
    """Fits the ReadMe ensemble by training multiple subspace mixture models.

    The `n_subsets` sub-quantifiers are mutually independent, so they are
    dispatched as independent `joblib` jobs (see `n_jobs`/`parallel_backend`)
    instead of a sequential Python loop.

    Parameters
    ----------
    X : ndarray of shape (n_samples, n_features)
      The training feature matrix.
    y : ndarray of shape (n_samples,)
      The target class labels.

    Returns
    -------
    self : object
      Returns the instance itself.
    """
    X, y = check_X_y(X, y, accept_sparse=False)
    self.classes_, class_counts = np.unique(y, return_counts=True)
    self.n_classes_ = len(self.classes_)
    self.train_prevalence_ = class_counts / len(y)

    total_features = X.shape[1]

    # dynamically determine the subspace feature size if not explicitly provided
    if self.n_features is None:
      if total_features > 25:
        self.n_features = total_features.bit_length()
      else:
        self.n_features = max(int(total_features / 5), 2)

    rng = check_random_state(self.random_state)
    self.feature_subsets_ = [
      rng.choice(total_features, self.n_features, replace=False) for _ in range(self.n_subsets)
    ]

    jobs = [
      delayed(_fit_subspace_job)(X, y, self.classes_, class_counts, feature_indices,
                                 self.distance_metric, self.use_convex_solver)
      for feature_indices in self.feature_subsets_
    ]
    self.sub_quantifiers_ = Parallel(n_jobs=self.n_jobs, backend=self.parallel_backend)(jobs)

    return self

  def predict(self, X: np.ndarray) -> np.ndarray:
    """Estimates class prevalences by averaging sub-quantifier predictions.

    Parameters
    ----------
    X : ndarray of shape (n_samples, n_features)
      The testing feature matrix.

    Returns
    -------
    ensemble_prevalences : ndarray of shape (n_classes,)
      The final aggregated and normalized prevalence estimation vector.
    """
    check_is_fitted(self)
    X = check_array(X, accept_sparse=False)

    jobs = [
      delayed(_predict_subspace_job)(sub_quantifier, X[:, feature_indices])
      for sub_quantifier, feature_indices in zip(self.sub_quantifiers_, self.feature_subsets_)
    ]
    subspace_predictions = Parallel(n_jobs=self.n_jobs, backend=self.parallel_backend)(jobs)

    # compute the final ensemble mean distribution
    return np.mean(subspace_predictions, axis=0)

fit(X, y)

Fits the ReadMe ensemble by training multiple subspace mixture models.

The n_subsets sub-quantifiers are mutually independent, so they are dispatched as independent joblib jobs (see n_jobs/parallel_backend) instead of a sequential Python loop.

Parameters:

Name Type Description Default
X ndarray of shape (n_samples, n_features)

The training feature matrix.

required
y ndarray of shape (n_samples,)

The target class labels.

required

Returns:

Name Type Description
self object

Returns the instance itself.

Source code in quack/quantifiers/_features.py
363
364
365
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
385
386
387
388
389
390
391
392
393
394
395
396
397
398
399
400
401
402
403
404
405
406
407
408
def fit(self, X: np.ndarray, y: np.ndarray) -> 'ReadMe':
  """Fits the ReadMe ensemble by training multiple subspace mixture models.

  The `n_subsets` sub-quantifiers are mutually independent, so they are
  dispatched as independent `joblib` jobs (see `n_jobs`/`parallel_backend`)
  instead of a sequential Python loop.

  Parameters
  ----------
  X : ndarray of shape (n_samples, n_features)
    The training feature matrix.
  y : ndarray of shape (n_samples,)
    The target class labels.

  Returns
  -------
  self : object
    Returns the instance itself.
  """
  X, y = check_X_y(X, y, accept_sparse=False)
  self.classes_, class_counts = np.unique(y, return_counts=True)
  self.n_classes_ = len(self.classes_)
  self.train_prevalence_ = class_counts / len(y)

  total_features = X.shape[1]

  # dynamically determine the subspace feature size if not explicitly provided
  if self.n_features is None:
    if total_features > 25:
      self.n_features = total_features.bit_length()
    else:
      self.n_features = max(int(total_features / 5), 2)

  rng = check_random_state(self.random_state)
  self.feature_subsets_ = [
    rng.choice(total_features, self.n_features, replace=False) for _ in range(self.n_subsets)
  ]

  jobs = [
    delayed(_fit_subspace_job)(X, y, self.classes_, class_counts, feature_indices,
                               self.distance_metric, self.use_convex_solver)
    for feature_indices in self.feature_subsets_
  ]
  self.sub_quantifiers_ = Parallel(n_jobs=self.n_jobs, backend=self.parallel_backend)(jobs)

  return self

predict(X)

Estimates class prevalences by averaging sub-quantifier predictions.

Parameters:

Name Type Description Default
X ndarray of shape (n_samples, n_features)

The testing feature matrix.

required

Returns:

Name Type Description
ensemble_prevalences ndarray of shape (n_classes,)

The final aggregated and normalized prevalence estimation vector.

Source code in quack/quantifiers/_features.py
410
411
412
413
414
415
416
417
418
419
420
421
422
423
424
425
426
427
428
429
430
431
432
433
def predict(self, X: np.ndarray) -> np.ndarray:
  """Estimates class prevalences by averaging sub-quantifier predictions.

  Parameters
  ----------
  X : ndarray of shape (n_samples, n_features)
    The testing feature matrix.

  Returns
  -------
  ensemble_prevalences : ndarray of shape (n_classes,)
    The final aggregated and normalized prevalence estimation vector.
  """
  check_is_fitted(self)
  X = check_array(X, accept_sparse=False)

  jobs = [
    delayed(_predict_subspace_job)(sub_quantifier, X[:, feature_indices])
    for sub_quantifier, feature_indices in zip(self.sub_quantifiers_, self.feature_subsets_)
  ]
  subspace_predictions = Parallel(n_jobs=self.n_jobs, backend=self.parallel_backend)(jobs)

  # compute the final ensemble mean distribution
  return np.mean(subspace_predictions, axis=0)

Energy Distance (ED)

quack.quantifiers._probabilities.ED

Bases: BaseQuantifier

Energy Distance Minimization (ED) Quantifier.

A non-parametric, feature-space mixture model that estimates target class prevalences by minimizing the Energy Distance divergence between the joint training distributions and the unlabelled test batch. It uses an exact analytical solution for binary settings and a quadratic programming solver (via CVXPY) for multiclass problems.

Parameters:

Name Type Description Default
n_jobs int

Number of jobs to run in parallel while computing the pairwise-distance sums that make up class_distances_matrix_ (during fit) and test_cross_distances (during predict). Each (class_i, class_j) pair (fit) or (class_i, test_bag) pair (predict) is mutually independent, so they are dispatched as independent joblib jobs. None means sequential (matching the previous behavior); -1 uses all available processors.

= None
parallel_backend str

joblib.Parallel backend used for the distance-sum jobs.

= "loky"

Attributes:

Name Type Description
classes_ ndarray of shape (n_classes,)

The distinct class labels found during the training phase.

n_classes_ int

The total number of unique classes.

train_class_samples_ list of ndarray

A list of length n_classes_ where each element stores a subset of the training feature matrix belonging strictly to that specific class index.

class_distances_matrix_ ndarray of shape (n_classes, n_classes)

Matrix 'A' representing the expected cross-class average pairwise distances calculated across the training data subsets.

quadratic_matrix_ ndarray of shape (n_classes - 1, n_classes - 1)

Matrix 'B' storing the transformed quadratic form coefficients used to solve multiclass optimization steps. Only populated if n_classes_ > 2.

References

Hideko Kawakubo, Marthinus Christoffel du Plessis, and Masashi Sugiyama. Computationally efficient class-prior estimation under class balance change using energy distance. IEICE Transactions on Information and Systems, 99(1):176-186, 2016.

Examples:

>>> from sklearn.datasets import make_classification
>>> X, y = make_classification(n_samples=500, n_classes=2, random_state=0)
>>> quantifier = ED()
>>> quantifier.fit(X, y)
>>> X_test, _ = make_classification(n_samples=100, n_classes=2, random_state=7)
>>> prevalences = quantifier.predict(X_test)
Source code in quack/quantifiers/_probabilities.py
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
class ED(BaseQuantifier):
  """Energy Distance Minimization (ED) Quantifier.

  A non-parametric, feature-space mixture model that estimates target class 
  prevalences by minimizing the Energy Distance divergence between the joint 
  training distributions and the unlabelled test batch. It uses an exact 
  analytical solution for binary settings and a quadratic programming solver 
  (via CVXPY) for multiclass problems.

  Parameters
  ----------
  n_jobs : int, default = None
    Number of jobs to run in parallel while computing the pairwise-distance
    sums that make up `class_distances_matrix_` (during `fit`) and
    `test_cross_distances` (during `predict`). Each `(class_i, class_j)`
    pair (fit) or `(class_i, test_bag)` pair (predict) is mutually
    independent, so they are dispatched as independent `joblib` jobs.
    `None` means sequential (matching the previous behavior); `-1` uses
    all available processors.
  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the distance-sum jobs.

  Attributes
  ----------
  classes_ : ndarray of shape (n_classes,)
    The distinct class labels found during the training phase.

  n_classes_ : int
    The total number of unique classes.

  train_class_samples_ : list of ndarray
    A list of length `n_classes_` where each element stores a subset of the 
    training feature matrix belonging strictly to that specific class index.

  class_distances_matrix_ : ndarray of shape (n_classes, n_classes)
    Matrix 'A' representing the expected cross-class average pairwise distances 
    calculated across the training data subsets.

  quadratic_matrix_ : ndarray of shape (n_classes - 1, n_classes - 1)
    Matrix 'B' storing the transformed quadratic form coefficients used to 
    solve multiclass optimization steps. Only populated if `n_classes_ > 2`.

  References
  ----------
  Hideko Kawakubo, Marthinus Christoffel du Plessis, and Masashi Sugiyama.
  Computationally efficient class-prior estimation under class balance change using
  energy distance. IEICE Transactions on Information and Systems, 99(1):176-186, 2016.

  Examples
  --------
  >>> from sklearn.datasets import make_classification
  >>> X, y = make_classification(n_samples=500, n_classes=2, random_state=0)
  >>> quantifier = ED()
  >>> quantifier.fit(X, y)
  >>> X_test, _ = make_classification(n_samples=100, n_classes=2, random_state=7)
  >>> prevalences = quantifier.predict(X_test)
  """

  def __init__(self, n_jobs: int = None, parallel_backend: str = "loky"):
    # energy distance operates directly on raw features, bypassing an underlying classifier
    super().__init__(classifier=None)
    self.n_jobs = n_jobs
    self.parallel_backend = parallel_backend
    self.class_distances_matrix_ = None
    self.quadratic_matrix_ = None
    self.train_class_samples_ = None

  def fit(self, X: np.ndarray, y: np.ndarray) -> 'ED':
    """Fits the ED quantifier by computing expected intra-class pairwise distances.

    Parameters
    ----------
    X : ndarray of shape (n_samples, n_features)
      The training feature matrix.
    y : ndarray of shape (n_samples,)
      The target class labels.

    Returns
    -------
    self : object
      Returns the instance itself.
    """
    X, y = check_X_y(X, y, accept_sparse=False)
    self.classes_ = np.unique(y)
    self.n_classes_ = len(self.classes_)

    if self.n_classes_ < 2:
      raise ValueError("Energy Distance requires at least 2 distinct classes.")

    # isolate training coordinates grouped by class
    self.train_class_samples_ = [X[y == class_label] for class_label in self.classes_]
    class_sizes = np.array([samples.shape[0] for samples in self.train_class_samples_])

    # dispatch every upper-triangular (i, j) pair as an independent job,
    # since each pairwise distance sum only depends on its own two class
    # blocks; matters most when n_classes_ or the class blocks are large
    pairs = [(i, j) for i in range(self.n_classes_) for j in range(i, self.n_classes_)]
    jobs = [
      delayed(_sum_pairwise_distances)(self.train_class_samples_[i], self.train_class_samples_[j])
      for i, j in pairs
    ]
    pair_sums = Parallel(n_jobs=self.n_jobs, backend=self.parallel_backend)(jobs)

    self.class_distances_matrix_ = np.zeros((self.n_classes_, self.n_classes_))
    for (i, j), total_distance in zip(pairs, pair_sums):
      value = total_distance / (class_sizes[i] * class_sizes[j])
      self.class_distances_matrix_[i, j] = value
      self.class_distances_matrix_[j, i] = value  # exploit symmetry

    # construct the optimization matrix (Matrix B) for multi-dimensional spaces
    if self.n_classes_ > 2:
      last_idx = self.n_classes_ - 1
      A = self.class_distances_matrix_
      # fully vectorized: quadratic_matrix_[i, j] = -A[i,j] + A[i,last] + A[last,j] - A[last,last].
      # Since A is symmetric this expression is automatically symmetric in
      # (i, j) too, matching the previous loop's explicit upper-triangle-then-mirror.
      self.quadratic_matrix_ = (
        -A[:last_idx, :last_idx]
        + A[:last_idx, last_idx][:, np.newaxis]
        + A[last_idx, :last_idx][np.newaxis, :]
        - A[last_idx, last_idx]
      )

    return self

  def predict(self, X: np.ndarray) -> np.ndarray:
    """Estimates class prevalences for the given unlabelled test data batch.

    Parameters
    ----------
    X : ndarray of shape (n_samples, n_features)
      The testing data matrix.

    Returns
    -------
    final_prevalences : ndarray of shape (n_classes,)
      A normalized probability vector indicating the estimated prevalence 
      proportions for each class.
    """
    check_is_fitted(self)
    X = check_array(X, accept_sparse=False)

    n_test_samples = X.shape[0]
    class_sizes = np.array([samples.shape[0] for samples in self.train_class_samples_])

    # compute the average distance profile from each training class subset
    # to the test bag; independent per class, dispatched as parallel jobs
    jobs = [
      delayed(_sum_pairwise_distances)(self.train_class_samples_[i], X)
      for i in range(self.n_classes_)
    ]
    class_sums = Parallel(n_jobs=self.n_jobs, backend=self.parallel_backend)(jobs)
    test_cross_distances = np.array(class_sums) / (class_sizes * n_test_samples)

    A = self.class_distances_matrix_
    s = test_cross_distances

    # Route A: analytical optimization for binary scenarios (n_classes < 3)
    if self.n_classes_ < 3:
      p = (s[1] - s[0] + A[0, 1] - A[1, 1]) / (-A[0, 0] + 2 * A[0, 1] - A[1, 1])

      if p < 0:
        return np.array([0.0, 1.0])
      if p > 1:
        return np.array([1.0, 0.0])
      return np.array([p, 1.0 - p])

    # Route B: constrained Quadratic Programming optimization for multiclass scenarios
    else:
      last_idx = self.n_classes_ - 1
      # fully vectorized: linear_vector[i] = -s[i] + A[i,last] + s[last] - A[last,last]
      linear_vector = -s[:last_idx] + A[:last_idx, last_idx] + s[last_idx] - A[last_idx, last_idx]

      # set up the constrained convex problem: minimize (P.T @ B @ P) - (2 * P.T @ t)
      estimated_proportions = cvx.Variable(last_idx)
      constraints = [estimated_proportions >= 0, cvx.sum(estimated_proportions) <= 1.0]

      objective_function = cvx.Minimize(
        cvx.quad_form(estimated_proportions, self.quadratic_matrix_) - 2 * estimated_proportions.T @ linear_vector
      )
      problem = cvx.Problem(objective_function, constraints)
      problem.solve()

      # post-process and append the pivot remaining probability profile element
      solved_proportions = np.clip(np.array(estimated_proportions.value).squeeze(), 0.0, 1.0)
      final_prevalences = np.append(solved_proportions, 1.0 - np.sum(solved_proportions))

      return final_prevalences

fit(X, y)

Fits the ED quantifier by computing expected intra-class pairwise distances.

Parameters:

Name Type Description Default
X ndarray of shape (n_samples, n_features)

The training feature matrix.

required
y ndarray of shape (n_samples,)

The target class labels.

required

Returns:

Name Type Description
self object

Returns the instance itself.

Source code in quack/quantifiers/_probabilities.py
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
def fit(self, X: np.ndarray, y: np.ndarray) -> 'ED':
  """Fits the ED quantifier by computing expected intra-class pairwise distances.

  Parameters
  ----------
  X : ndarray of shape (n_samples, n_features)
    The training feature matrix.
  y : ndarray of shape (n_samples,)
    The target class labels.

  Returns
  -------
  self : object
    Returns the instance itself.
  """
  X, y = check_X_y(X, y, accept_sparse=False)
  self.classes_ = np.unique(y)
  self.n_classes_ = len(self.classes_)

  if self.n_classes_ < 2:
    raise ValueError("Energy Distance requires at least 2 distinct classes.")

  # isolate training coordinates grouped by class
  self.train_class_samples_ = [X[y == class_label] for class_label in self.classes_]
  class_sizes = np.array([samples.shape[0] for samples in self.train_class_samples_])

  # dispatch every upper-triangular (i, j) pair as an independent job,
  # since each pairwise distance sum only depends on its own two class
  # blocks; matters most when n_classes_ or the class blocks are large
  pairs = [(i, j) for i in range(self.n_classes_) for j in range(i, self.n_classes_)]
  jobs = [
    delayed(_sum_pairwise_distances)(self.train_class_samples_[i], self.train_class_samples_[j])
    for i, j in pairs
  ]
  pair_sums = Parallel(n_jobs=self.n_jobs, backend=self.parallel_backend)(jobs)

  self.class_distances_matrix_ = np.zeros((self.n_classes_, self.n_classes_))
  for (i, j), total_distance in zip(pairs, pair_sums):
    value = total_distance / (class_sizes[i] * class_sizes[j])
    self.class_distances_matrix_[i, j] = value
    self.class_distances_matrix_[j, i] = value  # exploit symmetry

  # construct the optimization matrix (Matrix B) for multi-dimensional spaces
  if self.n_classes_ > 2:
    last_idx = self.n_classes_ - 1
    A = self.class_distances_matrix_
    # fully vectorized: quadratic_matrix_[i, j] = -A[i,j] + A[i,last] + A[last,j] - A[last,last].
    # Since A is symmetric this expression is automatically symmetric in
    # (i, j) too, matching the previous loop's explicit upper-triangle-then-mirror.
    self.quadratic_matrix_ = (
      -A[:last_idx, :last_idx]
      + A[:last_idx, last_idx][:, np.newaxis]
      + A[last_idx, :last_idx][np.newaxis, :]
      - A[last_idx, last_idx]
    )

  return self

predict(X)

Estimates class prevalences for the given unlabelled test data batch.

Parameters:

Name Type Description Default
X ndarray of shape (n_samples, n_features)

The testing data matrix.

required

Returns:

Name Type Description
final_prevalences ndarray of shape (n_classes,)

A normalized probability vector indicating the estimated prevalence proportions for each class.

Source code in quack/quantifiers/_probabilities.py
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
def predict(self, X: np.ndarray) -> np.ndarray:
  """Estimates class prevalences for the given unlabelled test data batch.

  Parameters
  ----------
  X : ndarray of shape (n_samples, n_features)
    The testing data matrix.

  Returns
  -------
  final_prevalences : ndarray of shape (n_classes,)
    A normalized probability vector indicating the estimated prevalence 
    proportions for each class.
  """
  check_is_fitted(self)
  X = check_array(X, accept_sparse=False)

  n_test_samples = X.shape[0]
  class_sizes = np.array([samples.shape[0] for samples in self.train_class_samples_])

  # compute the average distance profile from each training class subset
  # to the test bag; independent per class, dispatched as parallel jobs
  jobs = [
    delayed(_sum_pairwise_distances)(self.train_class_samples_[i], X)
    for i in range(self.n_classes_)
  ]
  class_sums = Parallel(n_jobs=self.n_jobs, backend=self.parallel_backend)(jobs)
  test_cross_distances = np.array(class_sums) / (class_sizes * n_test_samples)

  A = self.class_distances_matrix_
  s = test_cross_distances

  # Route A: analytical optimization for binary scenarios (n_classes < 3)
  if self.n_classes_ < 3:
    p = (s[1] - s[0] + A[0, 1] - A[1, 1]) / (-A[0, 0] + 2 * A[0, 1] - A[1, 1])

    if p < 0:
      return np.array([0.0, 1.0])
    if p > 1:
      return np.array([1.0, 0.0])
    return np.array([p, 1.0 - p])

  # Route B: constrained Quadratic Programming optimization for multiclass scenarios
  else:
    last_idx = self.n_classes_ - 1
    # fully vectorized: linear_vector[i] = -s[i] + A[i,last] + s[last] - A[last,last]
    linear_vector = -s[:last_idx] + A[:last_idx, last_idx] + s[last_idx] - A[last_idx, last_idx]

    # set up the constrained convex problem: minimize (P.T @ B @ P) - (2 * P.T @ t)
    estimated_proportions = cvx.Variable(last_idx)
    constraints = [estimated_proportions >= 0, cvx.sum(estimated_proportions) <= 1.0]

    objective_function = cvx.Minimize(
      cvx.quad_form(estimated_proportions, self.quadratic_matrix_) - 2 * estimated_proportions.T @ linear_vector
    )
    problem = cvx.Problem(objective_function, constraints)
    problem.solve()

    # post-process and append the pivot remaining probability profile element
    solved_proportions = np.clip(np.array(estimated_proportions.value).squeeze(), 0.0, 1.0)
    final_prevalences = np.append(solved_proportions, 1.0 - np.sum(solved_proportions))

    return final_prevalences

HDy

quack.quantifiers._dmm.HDy

Bases: DyS

Hellinger Distance y (HDy) Quantifier.

A specialized instance of the DyS framework that minimizes the Hellinger Distance over binned score histograms using a Logistic Regression classifier.

Parameters:

Name Type Description Default
classifier estimator object

The underlying base classifier. Defaults to LogisticRegression().

None
n_bins int

The total number of histogram bins.

10
cv int

The number of cross-validation folds.

10
use_convex_solver bool

If True, optimizes via CVXPY.

True
predict_proba bool

If True, forces the model to use probabilistic predict_proba outputs.

False
n_jobs int

Number of jobs to run in parallel while fitting the cv folds.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"
References

Víctor González-Castro, Rocío Alaiz-Rodríguez, and Enrique Alegre. Class distribution estimation based on the Hellinger distance. Information Sciences, 218(1):146-164, 2013.

Source code in quack/quantifiers/_dmm.py
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
class HDy(DyS):
  """Hellinger Distance y (HDy) Quantifier.

  A specialized instance of the DyS framework that minimizes the Hellinger 
  Distance over binned score histograms using a Logistic Regression classifier.

  Parameters
  ----------
  classifier : estimator object, default=None
      The underlying base classifier. Defaults to `LogisticRegression()`.

  n_bins : int, default=10
      The total number of histogram bins.

  cv : int, default=10
      The number of cross-validation folds.

  use_convex_solver : bool, default=True
      If True, optimizes via CVXPY.

  predict_proba : bool, default=False
      If True, forces the model to use probabilistic `predict_proba` outputs.

  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds.

  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  References
  ----------
  Víctor González-Castro, Rocío Alaiz-Rodríguez, and Enrique Alegre. Class distribution
  estimation based on the Hellinger distance. Information Sciences, 218(1):146-164, 2013.
  """

  def __init__(self,
               classifier: BaseEstimator = LogisticRegression(),
               n_bins: int = 10,
               cv: int = 10,
               use_convex_solver: bool = True,
               predict_proba: bool = False,
               n_jobs: int = None,
               parallel_backend: str = "loky"):
    super().__init__(classifier=classifier, distance_metric="HD", n_bins=n_bins, 
                     cv=cv, use_convex_solver=use_convex_solver, predict_proba=predict_proba,
                     n_jobs=n_jobs, parallel_backend=parallel_backend)

Distribution y Similarity (DyS)

quack.quantifiers._dmm.DyS

Bases: BaseScoreMixtureQuantifier

Distribution y-Similarity (DyS) Quantifier.

An adjusting prediction mixture model built strictly for binary quantification. It partitions out-of-fold continuous classification scores into a specified number of histograms bins to match training and testing distributions.

Parameters:

Name Type Description Default
classifier estimator object

The underlying base classifier. If None, defaults to SVC().

None
distance_metric str

The distance metric minimized ('L1', 'L2', 'HD', 'TS').

'TS'
n_bins int

The total number of histogram bins used to slice the distribution profiles.

10
cv int

The number of cross-validation folds for out-of-fold scoring.

10
use_convex_solver bool

If True, optimizes via CVXPY.

True
predict_proba bool

If True, forces the model to use probabilistic predict_proba outputs. If False, falls back to raw decision boundary scores.

False
n_jobs int

Number of jobs to run in parallel while fitting the cv folds. See BaseCalibratedQuantifier.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"

Attributes:

Name Type Description
classes_ ndarray of shape (n_classes,)

The distinct class labels found during the training phase.

n_classes_ int

The total number of unique classes.

train_prevalence_ ndarray of shape (n_classes,)

The baseline prevalence proportion of each class observed in the training data.

conditional_matrix_ ndarray of shape (n_bins, n_classes)

The binned score conditional matrix built using out-of-fold calibration data.

score_range_ tuple of float (min, max)

The minimum and maximum boundaries used to define histogram bins.

References

André Maletzke, Denis dos Reis, Everton Cherman, and Gustavo Batista. DyS: A framework for mixture models in quantification. In Proceedings of the AAAI Conference on Artificial Intelligence, pages 4552-4560, Honolulu, Hawaii, 2019.

Source code in quack/quantifiers/_dmm.py
 10
 11
 12
 13
 14
 15
 16
 17
 18
 19
 20
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
class DyS(BaseScoreMixtureQuantifier):
  """Distribution y-Similarity (DyS) Quantifier.

  An adjusting prediction mixture model built strictly for binary quantification. 
  It partitions out-of-fold continuous classification scores into a specified number 
  of histograms bins to match training and testing distributions.

  Parameters
  ----------
  classifier : estimator object, default=None
    The underlying base classifier. If None, defaults to `SVC()`.

  distance_metric : str, default='TS'
    The distance metric minimized ('L1', 'L2', 'HD', 'TS').

  n_bins : int, default=10
    The total number of histogram bins used to slice the distribution profiles.

  cv : int, default=10
    The number of cross-validation folds for out-of-fold scoring.

  use_convex_solver : bool, default=True
    If True, optimizes via CVXPY.

  predict_proba : bool, default=False
    If True, forces the model to use probabilistic `predict_proba` outputs. 
    If False, falls back to raw decision boundary scores.

  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds. See
    `BaseCalibratedQuantifier`.

  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  Attributes
  ----------
  classes_ : ndarray of shape (n_classes,)
    The distinct class labels found during the training phase.

  n_classes_ : int
    The total number of unique classes.

  train_prevalence_ : ndarray of shape (n_classes,)
    The baseline prevalence proportion of each class observed in the training data.

  conditional_matrix_ : ndarray of shape (n_bins, n_classes)
    The binned score conditional matrix built using out-of-fold calibration data.

  score_range_ : tuple of float (min, max)
    The minimum and maximum boundaries used to define histogram bins.

  References
  ----------
  André Maletzke, Denis dos Reis, Everton Cherman, and Gustavo Batista. DyS: A framework
  for mixture models in quantification. In Proceedings of the AAAI Conference on Artificial
  Intelligence, pages 4552-4560, Honolulu, Hawaii, 2019.
  """

  def __init__(self,
               classifier: BaseEstimator = SVC(),
               distance_metric: str = "TS",
               n_bins: int = 10, cv: int = 10, use_convex_solver: bool = True,
               predict_proba: bool = False, n_jobs: int = None, parallel_backend: str = "loky"):
    super().__init__(classifier=classifier, distance_metric=distance_metric, cv=cv,
                     use_convex_solver=use_convex_solver, predict_proba=predict_proba,
                     n_jobs=n_jobs, parallel_backend=parallel_backend)
    self.n_bins = n_bins
    self.score_range_ = None

  def _calibrate(self, y_true_oof: np.ndarray, y_pred_oof: np.ndarray):
    y_scores = self._extract_1d_scores(y_pred_oof)
    self.score_range_ = (0.0, 1.0) if self.predict_proba else (np.min(y_scores), np.max(y_scores))

    conditional_blocks = []
    for class_label in self.classes_:
      class_mask = (y_true_oof == class_label)
      counts, _ = np.histogram(y_scores[class_mask], bins=self.n_bins, range=self.score_range_)
      conditional_blocks.append(counts)

    _, class_counts = np.unique(y_true_oof, return_counts=True)
    self.conditional_matrix_ = np.vstack(conditional_blocks).T / class_counts

  def _compute_score(self, X: np.ndarray) -> np.ndarray:
    prediction_method = getattr(self.classifier_, self._get_oof_method())
    raw_predictions = prediction_method(X)
    y_scores = self._extract_1d_scores(raw_predictions)

    test_frequencies, _ = np.histogram(y_scores, bins=self.n_bins, range=self.score_range_)
    if not self.predict_proba:
      test_frequencies[0] += np.sum(y_scores < self.score_range_[0])
      test_frequencies[-1] += np.sum(y_scores > self.score_range_[1])

    return test_frequencies / X.shape[0]

Expectation Maximization Quantifier (EMQ)

quack.quantifiers._iterators.EM

Bases: BaseCalibratedQuantifier

Expectation Maximization (EM) Quantifier.

An iterative quantification algorithm that adapts a classifier's output probabilities to a target test set by maximizing the likelihood of the test data. It recursively updates sample posteriors and prior estimations until convergence.

Parameters:

Name Type Description Default
classifier estimator object

The underlying base classifier implementing predict_proba. If None, defaults to LogisticRegression().

= None
cv int, cross-validation generator or an iterable

Determines the cross-validation splitting strategy for the calibration phase.

= 10
epsilon float

The convergence tolerance threshold. The iterative loop terminates when the Euclidean norm between consecutive steps is smaller than this value.

= 1e-06
max_iter int

The maximum allowable optimization iterations.

= 1000
n_jobs int

Number of jobs to run in parallel while fitting the cv folds (plus the final full-data classifier refit). See BaseCalibratedQuantifier.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"

Attributes:

Name Type Description
classes_ ndarray of shape (n_classes,)

The distinct class labels found during the training phase.

n_classes_ int

The total number of unique classes.

train_prevalence_ ndarray of shape (n_classes,)

The baseline prevalence proportion of each class observed in the training data.

classifier_ estimator object

The final trained base classifier adjusted on the entire training dataset.

References

Marco Saerens, Patrice Latinne, and Christine Decaestecker. Adjusting the outputs of a classifier to new a priori probabilities: A simple procedure. Neural Computation, 14(1): 21-41, 2002.

Source code in quack/quantifiers/_iterators.py
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
class EM(BaseCalibratedQuantifier):
  """Expectation Maximization (EM) Quantifier.

  An iterative quantification algorithm that adapts a classifier's output probabilities 
  to a target test set by maximizing the likelihood of the test data. It recursively 
  updates sample posteriors and prior estimations until convergence.

  Parameters
  ----------
  classifier : estimator object, default = None
    The underlying base classifier implementing `predict_proba`. If None, 
    defaults to `LogisticRegression()`.

  cv : int, cross-validation generator or an iterable, default = 10
    Determines the cross-validation splitting strategy for the calibration phase.

  epsilon : float, default = 1e-06
    The convergence tolerance threshold. The iterative loop terminates when the 
    Euclidean norm between consecutive steps is smaller than this value.

  max_iter : int, default = 1000
    The maximum allowable optimization iterations.

  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds (plus
    the final full-data classifier refit). See `BaseCalibratedQuantifier`.

  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  Attributes
  ----------
  classes_ : ndarray of shape (n_classes,)
    The distinct class labels found during the training phase.

  n_classes_ : int
    The total number of unique classes.

  train_prevalence_ : ndarray of shape (n_classes,)
    The baseline prevalence proportion of each class observed in the training data.

  classifier_ : estimator object
    The final trained base classifier adjusted on the entire training dataset.

  References
  ----------
  Marco Saerens, Patrice Latinne, and Christine Decaestecker.
  Adjusting the outputs of a classifier to new a priori probabilities: A simple procedure.
  Neural Computation, 14(1): 21-41, 2002.
  """

  def __init__(self, classifier: BaseEstimator = None, cv: int = 10,
               epsilon: float = 1e-06, max_iter: int = 1000,
               n_jobs: int = None, parallel_backend: str = "loky"):
    super().__init__(classifier=classifier, cv=cv, n_jobs=n_jobs, parallel_backend=parallel_backend)
    self.epsilon = epsilon
    self.max_iter = max_iter

  def _get_oof_method(self) -> str:
    return "predict_proba"

  def _calibrate(self, y_true_oof: np.ndarray, y_pred_oof: np.ndarray):
    """EM handles calibration adjustments dynamically inside the prediction loop."""
    pass

  def _quantify(self, X: np.ndarray) -> np.ndarray:
    # extract probabilistic scores
    predicted_probabilities = self.classifier_.predict_proba(X)

    # initialize convergence tracking arrays matching the original calculus state
    prevalence_new = self.train_prevalence_
    prevalence_old = np.ones(self.train_prevalence_.shape)
    iteration_count = 0

    # convergence logic loop
    while (np.linalg.norm(prevalence_old - prevalence_new) > self.epsilon) and iteration_count < self.max_iter:
      prevalence_old = np.array(prevalence_new)

      # vectorized update: row-wise multiply every sample's posterior
      # vector by the (prior_new / prior_train) ratio via broadcasting,
      # then renormalize each row to sum to 1.0 — mathematically identical
      # to the previous per-sample Python loop, just without the loop
      posterior_matrix = predicted_probabilities * (prevalence_old / self.train_prevalence_)
      posterior_matrix /= posterior_matrix.sum(axis=1, keepdims=True)

      # update step: average the adjusted sample posterior profiles
      prevalence_new = posterior_matrix.mean(axis=0)
      iteration_count += 1

    return prevalence_new

Class Distribution Estimation (CDE)

quack.quantifiers._iterators.CDE

Bases: BaseCalibratedQuantifier

Class Conditional Density Estimation (CDE) Quantifier.

An iterative binary threshold-adjusting quantifier that modulates prediction cutoffs dynamically by evaluating relative target distribution shifts.

Parameters:

Name Type Description Default
classifier estimator object

The underlying base classifier implementing predict_proba. If None, defaults to LogisticRegression().

= None
cv int, cross-validation generator or an iterable

Determines the cross-validation splitting strategy for the calibration phase.

= 10
epsilon float

The convergence tolerance threshold.

= 1e-06
max_iter int

The maximum allowable optimization iterations.

= 1000
n_jobs int

Number of jobs to run in parallel while fitting the cv folds. See BaseCalibratedQuantifier.

= None
parallel_backend str

joblib.Parallel backend used for the CV/final-fit jobs.

= "loky"

Attributes:

Name Type Description
classes_ ndarray of shape (2,)

The binary class labels found during the training phase.

n_classes_ int

The total number of unique classes (strictly equals 2).

train_prevalence_ ndarray of shape (2,)

The baseline prevalence proportion of each class observed in the training data.

classifier_ estimator object

The final trained base classifier adjusted on the entire training dataset.

References

Dirk Tasche. Fisher consistency for prior probability shift. Journal of Machine Learning Research, 18(95):1-32, 2017.

Source code in quack/quantifiers/_iterators.py
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
class CDE(BaseCalibratedQuantifier):
  """Class Conditional Density Estimation (CDE) Quantifier.

  An iterative binary threshold-adjusting quantifier that modulates prediction 
  cutoffs dynamically by evaluating relative target distribution shifts.

  Parameters
  ----------
  classifier : estimator object, default = None
    The underlying base classifier implementing `predict_proba`. If None, 
    defaults to `LogisticRegression()`.

  cv : int, cross-validation generator or an iterable, default = 10
    Determines the cross-validation splitting strategy for the calibration phase.

  epsilon : float, default = 1e-06
    The convergence tolerance threshold.

  max_iter : int, default = 1000
    The maximum allowable optimization iterations.

  n_jobs : int, default = None
    Number of jobs to run in parallel while fitting the `cv` folds. See
    `BaseCalibratedQuantifier`.

  parallel_backend : str, default = "loky"
    `joblib.Parallel` backend used for the CV/final-fit jobs.

  Attributes
  ----------
  classes_ : ndarray of shape (2,)
    The binary class labels found during the training phase.

  n_classes_ : int
    The total number of unique classes (strictly equals 2).

  train_prevalence_ : ndarray of shape (2,)
    The baseline prevalence proportion of each class observed in the training data.

  classifier_ : estimator object
    The final trained base classifier adjusted on the entire training dataset.

  References
  ----------
  Dirk Tasche. Fisher consistency for prior probability shift.
  Journal of Machine Learning Research, 18(95):1-32, 2017.
  """

  def __init__(self, classifier: BaseEstimator = None, cv: int = 10,
               epsilon: float = 1e-06, max_iter: int = 1000,
               n_jobs: int = None, parallel_backend: str = "loky"):
    super().__init__(classifier=classifier, cv=cv, n_jobs=n_jobs, parallel_backend=parallel_backend)
    self.epsilon = epsilon
    self.max_iter = max_iter

  def _get_oof_method(self) -> str:
    return "predict_proba"

  def _calibrate(self, y_true_oof: np.ndarray, y_pred_oof: np.ndarray):
    """CDE thresholding adjustments are executed entirely during the testing phase."""
    pass

  def fit(self, X: np.ndarray, y: np.ndarray) -> 'CDE':
    # check target dimensionality before triggering the pipeline execution
    unique_classes = np.unique(y)
    if len(unique_classes) > 2:
      raise ValueError(
        "CDE only works for binary quantification. Multiclass is possible via the "
        "OVR strategy, but not recommended due to theoretical issues with that approach."
      )

    return super().fit(X, y)

  def _quantify(self, X: np.ndarray) -> np.ndarray:
    predicted_probabilities = self.classifier_.predict_proba(X)
    pos_probs = predicted_probabilities[:, 1]

    # initialize directional weight arrays matching the original state
    weights = np.ones(2)
    weights_old = np.zeros(2)

    positive_prevalence = 2.0
    iteration_count = 0

    # strict preservation of your original termination criteria (<= max_iter)
    while np.linalg.norm(weights - weights_old) > self.epsilon and iteration_count <= self.max_iter:
      # vectorized hard-label assignment: replaces the previous
      # np.apply_along_axis(lambda, ...) call, which is a thin Python-loop
      # wrapper around each row and not actually vectorized; np.where
      # performs the exact same row-wise comparison in a single C-level pass
      threshold = weights[0] / np.sum(weights)
      threshold_labels = np.where(pos_probs > threshold, self.classes_[1], self.classes_[0])
      weights_old = np.copy(weights)

      # calculate the empirical mean of positive labels
      positive_prevalence = np.mean(threshold_labels == self.classes_[1])

      # re-weight updates based on baseline training rates discrepancies
      weights[0] = (1.0 - positive_prevalence) / self.train_prevalence_[0]
      weights[1] = positive_prevalence / self.train_prevalence_[1]
      iteration_count += 1

    if iteration_count >= self.max_iter:
      warnings.warn("The CDE iteration has not converged.")

    return np.array([1.0 - positive_prevalence, positive_prevalence])