moreo
/
QuaPy
1
0
Fork 1
Code Issues Pull Requests Releases Wiki Activity

Add GMNet, a Gaussian-mixture neural quantifier 

Ports GMNet (from https://github.com/pglez84/gmnet) into quapy/method/_gmnet.py,
mirroring how HistNetQ was ported: dropping that repo's quantificationlib-backed
bag generators in favor of QuaPy's own sampling protocols, and adding geotorch
(now a 'neural' extra dependency) to keep the Gaussian layers' covariance matrices
positive-definite during training.

- GMNet represents each bag instance by its likelihood under one or more learned
  mixtures of Gaussians ("GM branches"), mean-pools these representations over the
  bag, and predicts prevalence from the result. Supports multiple stacked GM
  branches with an optional CKA-regularization term encouraging their latent
  representations to be dissimilar.
- Fixes two aspects of the original architecture that assumed a fixed, training-time
  bag_size baked into the network (a reshape step, and forward-hook-based activation
  capture for CKA): both are now computed from the actual input shape/plain
  attributes at forward time, so the model also works on predict()'s arbitrary-sized
  test samples, not just same-size bags.
- Factors the bag-based training loop shared by HistNetQ and GMNet (bag generation,
  fit/fit_from_samples, early stopping, LR scheduling, checkpointing, predict) out of
  _histnet.py into a new BagTrainedQuantifier base class in
  quapy/method/_neural_bags.py; HistNetQ's public API and behavior are unchanged.
- Aliased in meta.py (torch/geotorch-optional, mirroring HistNetQ/QuaNet) and
  registered in META_METHODS.
- Adds test_gmnet covering single-branch and multi-branch+CKA (via
  fit_from_samples/mix_bags) variants.

Co-Authored-By: Claude Sonnet 5 <noreply@anthropic.com>
This commit is contained in:
Alejandro Moreo Fernandez 2026-09-08 14:47:03 +02:00
parent d6fbd13ecd
commit 89548d3a8b
7 changed files with 864 additions and 310 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

3
quapy/method/__init__.py
View File

@ -81,7 +81,8 @@ NON_AGGREGATIVE_METHODS = {
META_METHODS = {
meta.Ensemble,
meta.QuaNet,
meta.HistNetQ
meta.HistNetQ,
meta.GMNet
}
QUANTIFICATION_METHODS = AGGREGATIVE_METHODS | NON_AGGREGATIVE_METHODS | META_METHODS

328
quapy/method/_gmnet.py Normal file
View File

@ -0,0 +1,328 @@
"""
GMNet implementation.
Ported from the reference implementation at https://github.com/pglez84/gmnet (the `GMNet`/
`DLQuantification` classes in that repo), adapted to QuaPy's own protocol-based sample generation
(replacing that repo's custom, `quantificationlib`-backed bag generators), and reusing the shared
bag-based training loop already factored out for :class:`quapy.method._histnet.HistNetQ` (see
:class:`quapy.method._neural_bags.BagTrainedQuantifier`).
The overall architecture is: one or more "GM branches" -- each a small per-branch feature extractor
followed by a layer of Gaussian likelihoods (a :class:`_GMLayer`) evaluated at every instance of a bag
-- concatenated and mean-pooled over the bag, followed by the shared quantification MLP head. Like
HistNetQ (and QuaNet), GMNet is trained end-to-end by minimizing a quantification loss over samples
("bags") of known prevalence, rather than over individually labeled instances.
Two deliberate deviations from the reference implementation, both required for the model to satisfy
QuaPy's `predict(X)` contract (i.e., to be usable on a real test collection of arbitrary size, as
opposed to only on bags resampled at the fixed `bag_size` used for training):
* the original `GMNet_Module` reshapes each branch's per-instance likelihoods around a *fixed*,
constructor-time `bag_size` (via `torch.nn.Unflatten(0, (-1, bag_size))`), which only works when
every forward pass is fed bags of exactly that size. Here, the reshape is instead computed from the
actual input shape at forward time (see :class:`_GMBranch`), which is equivalent when the bag size
matches but also supports bags (or, at prediction time, whole test samples) of any other size.
* the forward hooks used by the original code to capture each branch's pre-Gaussian latent activations
(for the CKA regularization term) are replaced by simply storing that activation as an attribute
during `forward` (see :attr:`_GMBranch.latent_activation`), since branches are now implemented with a
plain `forward` method rather than an opaque `torch.nn.Sequential`.
"""
import numpy as np
import scipy.spatial.distance
import torch
import torch.nn as nn
import geotorch
from quapy.method._neural_bags import BagTrainedQuantifier
from quapy.protocol import UPP
def _cka(latent_activations):
"""Feature-space linear CKA (Centered Kernel Alignment), averaged over every pair of latent
activations, following the `CKARegularization` class in the reference implementation. Used to
encourage the Gaussian components learned by different GM branches to capture complementary
(dissimilar) aspects of the instances.
:param latent_activations: a list of tensors, one per GM branch, all of shape (n_instances, dim_i)
(dim_i may differ across branches)
"""
cka_sum = 0.
n_pairs = 0
for i in range(len(latent_activations)):
for j in range(i + 1, len(latent_activations)):
x = latent_activations[i]
y = latent_activations[j]
x = x - torch.mean(x, dim=0, keepdim=True)
y = y - torch.mean(y, dim=0, keepdim=True)
dot_product_similarity = torch.norm(torch.matmul(x.t(), y)) ** 2
normalization_x = torch.norm(torch.matmul(x.t(), x))
normalization_y = torch.norm(torch.matmul(y.t(), y))
cka_sum = cka_sum + dot_product_similarity / (normalization_x * normalization_y)
n_pairs += 1
return cka_sum / n_pairs
class _GMLayer(nn.Module):
"""A layer of `num_gaussians` (unnormalized) Gaussian likelihoods, evaluated at every instance of
a bag. `centers` and `covariance` are learned; `covariance` is constrained to stay positive-definite
throughout training via `geotorch.positive_definite`.
"""
def __init__(self, n_features, num_gaussians):
super().__init__()
self.n_features = n_features
self.num_gaussians = num_gaussians
self.centers = nn.Parameter(torch.rand(num_gaussians, n_features))
self.covariance = nn.Parameter(torch.eye(n_features).repeat(num_gaussians, 1, 1))
geotorch.positive_definite(self, "covariance")
# initialize the centers' covariance from the (squared, halved) nearest-neighbor distance
# between the randomly initialized centers, so that gaussians start with a sensible spread
centers = self.centers.detach().cpu().numpy()
distances = scipy.spatial.distance.cdist(centers, centers)
np.fill_diagonal(distances, np.inf)
cov = (np.mean(np.min(distances, axis=1)) / 2) ** 2
self.covariance = torch.eye(n_features).repeat(num_gaussians, 1, 1) * cov
def forward(self, x):
# x: (batch_size, bag_size, n_features)
centers = self.centers.unsqueeze(0).unsqueeze(0) # (1, 1, num_gaussians, n_features)
diff = x.unsqueeze(2) - centers # (batch_size, bag_size, num_gaussians, n_features)
cov_inv = torch.inverse(self.covariance)
det_cov = torch.linalg.det(self.covariance)
mahalanobis = torch.einsum('...i,...ij,...j->...', diff, cov_inv.unsqueeze(0).unsqueeze(0), diff)
normalization_term = torch.log((2 * torch.pi) ** self.n_features * det_cov).unsqueeze(0).unsqueeze(0)
log_probs = -0.5 * (mahalanobis + normalization_term)
return torch.exp(log_probs) # (batch_size, bag_size, num_gaussians)
class _GMBranch(nn.Module):
"""One GM branch: an optional small MLP mapping the (already feature-extracted) instances into a
`gaussian_dimensions`-sized latent space, followed by a Sigmoid, a :class:`_GMLayer`, and a
BatchNorm applied instance-wise (i.e., over the merged batch*bag_size dimension, matching the
reference implementation).
"""
def __init__(self, input_size, num_gaussians, gaussian_dimensions, hidden_size_fe, dropout_fe):
super().__init__()
self.pre = nn.Sequential()
prev_size = input_size
latent_size = gaussian_dimensions if gaussian_dimensions is not None else input_size
if gaussian_dimensions is not None:
for j, layer_size in enumerate(hidden_size_fe or ()):
self.pre.add_module(f'hidden_{j}', nn.Linear(prev_size, layer_size))
self.pre.add_module(f'leakyrelu_{j}', nn.LeakyReLU())
self.pre.add_module(f'dropout_{j}', nn.Dropout(dropout_fe))
prev_size = layer_size
self.pre.add_module('latent_linear', nn.Linear(prev_size, gaussian_dimensions))
self.pre.add_module('sigmoid', nn.Sigmoid())
self.gm_layer = _GMLayer(n_features=latent_size, num_gaussians=num_gaussians)
self.batch_norm = nn.BatchNorm1d(num_features=num_gaussians)
self.output_size = num_gaussians
self.latent_activation = None # populated on every forward(), read by GMNet's CKA regularization
def forward(self, x):
# x: (batch_size, bag_size, input_size)
batch_size, bag_size = x.shape[0], x.shape[1]
latent = self.pre(x)
self.latent_activation = latent.reshape(-1, latent.shape[-1])
likelihoods = self.gm_layer(latent) # (batch_size, bag_size, num_gaussians)
flat = self.batch_norm(likelihoods.reshape(batch_size * bag_size, -1))
return flat.reshape(batch_size, bag_size, -1)
class _GMNetModule(nn.Module):
"""The quantification module for GMNet: one or more :class:`_GMBranch` instances, each producing a
per-instance representation that is concatenated across branches and mean-pooled over the bag, as
required by :class:`quapy.method._neural_bags.BagTrainedQuantifier`.
"""
def __init__(self, input_size, num_gaussians, n_gm_layers, gaussian_dimensions, hidden_size_fe=None,
dropout_fe=0., cka_regularization=0.):
super().__init__()
if len(num_gaussians) != n_gm_layers:
raise ValueError('num_gaussians should be a tuple of the same size as n_gm_layers')
if len(gaussian_dimensions) != n_gm_layers:
raise ValueError('gaussian_dimensions should be a tuple of the same size as n_gm_layers')
self.n_gm_layers = n_gm_layers
self.cka_regularization = cka_regularization
self.branches = nn.ModuleList([
_GMBranch(input_size, num_gaussians[i], gaussian_dimensions[i], hidden_size_fe, dropout_fe)
for i in range(n_gm_layers)
])
self.output_size = sum(num_gaussians)
def forward(self, x):
outputs = [branch(x) for branch in self.branches]
return torch.mean(torch.cat(outputs, dim=-1), dim=1)
def apply_regularization(self):
"""Whether the CKA regularization term should be added to the training loss: requires at least
two GM branches (CKA is a pairwise measure) and a nonzero `cka_regularization` weight."""
return self.n_gm_layers > 1 and self.cka_regularization != 0
def regularization_term(self):
latent_activations = [branch.latent_activation for branch in self.branches]
return self.cka_regularization * _cka(latent_activations)
class GMNet(BagTrainedQuantifier):
"""
Implementation of `GMNet <https://github.com/pglez84/gmnet>`_, a neural network for quantification
that represents each instance of a bag by its likelihood under one or more learned mixtures of
Gaussians, mean-pools these representations over the bag, and predicts the class prevalence from the
result, trained end-to-end by minimizing a quantification loss over many samples ("bags") of known
prevalence.
Like :class:`quapy.method._histnet.HistNetQ` and :class:`quapy.method.meta.QuaNet`, GMNet does not
follow the classify-then-aggregate pattern of :class:`quapy.method.aggregative.AggregativeQuantifier`;
it is instead trained and evaluated end-to-end on whole bags (see
:class:`quapy.method._neural_bags.BagTrainedQuantifier` for the shared training/prediction logic,
including the two entry points, :meth:`fit` and :meth:`fit_from_samples`).
:param feature_extraction_module: a `torch.nn.Module` exposing an `output_size` attribute, used to
embed each instance before it is passed to every GM branch. If None (default), an identity
module is used, i.e., the instances in `X` are assumed to already be in their final numeric
representation.
:param n_gm_layers: number of GM branches (default 1).
:param num_gaussians: number of gaussians per branch: either a single int (used for every branch) or
a tuple/list of `n_gm_layers` ints (default 4).
:param gaussian_dimensions: dimensionality of the latent space in which each branch's gaussians live:
either a single int/None (used for every branch) or a tuple/list of `n_gm_layers` int/None
values. If None for a given branch, that branch's gaussians operate directly on the
feature-extracted instances, with no extra per-branch projection (default None).
:param hidden_size_fe: sizes of the hidden layers of the small per-branch MLP that maps the
feature-extracted instances into the latent space (only used when `gaussian_dimensions` is not
None for the corresponding branch); default None (no hidden layers, i.e., a single linear
projection).
:param dropout_fe: dropout applied after each of the `hidden_size_fe` layers (default 0).
:param cka_regularization: weight of the CKA regularization term encouraging the different branches'
latent representations to be dissimilar; only applied when `n_gm_layers > 1` (default 0, i.e.,
disabled).
:param linear_sizes: tuple of ints with the sizes of the linear layers used in the shared
quantification head, after the GM branches (default empty, i.e., only the final classification
layer is used).
:param dropout: dropout applied after each of the `linear_sizes` layers (default 0).
:param output_function: either 'softmax' or 'normalize' (L1); both yield a valid prevalence vector
(default 'softmax').
:param bag_size: number of instances per training/validation bag (default 500).
:param n_bags_train: number of bags generated per training epoch (default 500).
:param n_bags_val: number of bags generated per validation epoch (default 500).
:param train_epochs: maximum number of training epochs (default 200).
:param patience: number of epochs without improvement in validation loss before early-stopping
(default 20).
:param start_lr: initial learning rate (default 1e-3).
:param end_lr: once the learning rate decays below this value, training stops (default 1e-6).
:param lr_factor: factor by which the learning rate is reduced after `patience` epochs without
improvement (default 0.1).
:param weight_decay: L2 regularization (default 0).
:param quant_loss: the quantification loss to minimize (default `torch.nn.L1Loss()`), called as
`quant_loss(true_prevalences, predicted_prevalences)`.
:param batch_size: number of bags per gradient update (default 16).
:param protocol: the :class:`quapy.protocol.AbstractStochasticSeededProtocol` subclass used by
:meth:`fit` to resample bags from the given labelled collection (default
:class:`quapy.protocol.UPP`, which draws bags with prevalence sampled uniformly at random from
the simplex).
:param protocol_params: dict of extra keyword arguments passed to `protocol` (besides `data`,
`sample_size`, `repeats`, and `random_state`, which are set internally); default None.
:param val_split: float in (0,1), the proportion of the collection given to :meth:`fit` that is held
out (via stratified sampling) for validation and early stopping (default 0.4).
:param device: `'cpu'` or `'cuda'` (default 'cpu').
:param random_state: seed used for the train/validation split and for the (fixed) validation
sampling sequence, as well as for the random initialization of the GM branches (default 0).
:param checkpointdir: directory where the best model found during training is stored (default
'../checkpoint').
:param checkpointname: name of the checkpoint file; if None (default), a random name is generated.
:param verbose: verbosity level; if >0, shows a progress bar with the current losses (default 0).
"""
def __init__(self,
feature_extraction_module=None,
n_gm_layers=1,
num_gaussians=4,
gaussian_dimensions=None,
hidden_size_fe=None,
dropout_fe=0.,
cka_regularization=0.,
linear_sizes=(),
dropout=0.,
output_function='softmax',
bag_size=500,
n_bags_train=500,
n_bags_val=500,
train_epochs=200,
patience=20,
start_lr=1e-3,
end_lr=1e-6,
lr_factor=0.1,
weight_decay=0.,
quant_loss=None,
batch_size=16,
protocol=UPP,
protocol_params=None,
val_split=0.4,
device='cpu',
random_state=0,
checkpointdir='../checkpoint',
checkpointname=None,
verbose=0):
super().__init__(
feature_extraction_module=feature_extraction_module,
linear_sizes=linear_sizes,
dropout=dropout,
output_function=output_function,
bag_size=bag_size,
n_bags_train=n_bags_train,
n_bags_val=n_bags_val,
train_epochs=train_epochs,
patience=patience,
start_lr=start_lr,
end_lr=end_lr,
lr_factor=lr_factor,
weight_decay=weight_decay,
quant_loss=quant_loss,
batch_size=batch_size,
protocol=protocol,
protocol_params=protocol_params,
val_split=val_split,
device=device,
random_state=random_state,
checkpointdir=checkpointdir,
checkpointname=checkpointname,
verbose=verbose,
)
self.n_gm_layers = n_gm_layers
self.num_gaussians = num_gaussians if isinstance(num_gaussians, (tuple, list)) \
else [num_gaussians] * n_gm_layers
self.gaussian_dimensions = gaussian_dimensions if isinstance(gaussian_dimensions, (tuple, list)) \
else [gaussian_dimensions] * n_gm_layers
self.hidden_size_fe = hidden_size_fe
self.dropout_fe = dropout_fe
self.cka_regularization = cka_regularization
@property
def _checkpoint_prefix(self):
return 'GMNet'
def _build_quantmodule(self, n_features):
torch.manual_seed(self.random_state)
return _GMNetModule(
input_size=n_features,
num_gaussians=self.num_gaussians,
n_gm_layers=self.n_gm_layers,
gaussian_dimensions=self.gaussian_dimensions,
hidden_size_fe=self.hidden_size_fe,
dropout_fe=self.dropout_fe,
cka_regularization=self.cka_regularization,
)
def _extra_loss(self):
quantmodule = self.model.quantmodule
if quantmodule.apply_regularization():
return quantmodule.regularization_term()
return 0.

355
quapy/method/_histnet.py
View File

@ -15,33 +15,15 @@ arXiv preprint arXiv:2012.06311 (2020).
The overall architecture is: feature_extraction -> Sigmoid -> histogram layer -> small MLP -> softmax,
trained by minimizing a quantification loss over samples ("bags") of known prevalence, rather than
over individually labeled instances (in the spirit of QuaNet, see method/_quanet.py).
over individually labeled instances (in the spirit of QuaNet, see method/_quanet.py). The bag-based
training loop itself (bag generation, early stopping, LR scheduling, checkpointing, prediction) is
shared with :class:`quapy.method._gmnet.GMNet` via :class:`quapy.method._neural_bags.BagTrainedQuantifier`.
"""
import copy
import os
import random
import numpy as np
import torch
import torch.nn as nn
from tqdm import tqdm
from quapy.data import LabelledCollection
from quapy.method.base import BaseQuantifier
from quapy.protocol import AbstractProtocol, UPP
from quapy.util import EarlyStop
class _IdentityFeatureExtractionModule(nn.Module):
"""Used when no feature extraction module is provided: instances are assumed to already be in
their final numeric representation."""
def __init__(self, input_size):
super().__init__()
self.output_size = input_size
def forward(self, x):
return x
from quapy.method._neural_bags import BagTrainedQuantifier
from quapy.protocol import UPP
class _HardHistogramLayer(nn.Module):
@ -99,84 +81,22 @@ class _HardHistogramLayer(nn.Module):
return result
class _HistNetModule(nn.Module):
"""The full HistNetQ network: feature extraction, histogram, and the quantification MLP."""
class _SigmoidHistogram(nn.Module):
"""The quantification module for HistNetQ: squashes the (already feature-extracted) instances
through a Sigmoid and builds a differentiable histogram of them, as required by
:class:`quapy.method._neural_bags.BagTrainedQuantifier`."""
def __init__(self, feature_extraction_module, n_classes, n_bins=8, quantiles=False, linear_sizes=(),
dropout=0., output_function='softmax'):
def __init__(self, n_features, n_bins=8, quantiles=False):
super().__init__()
self.feature_extraction_module = feature_extraction_module
self.sigmoid = nn.Sigmoid()
self.histogram = _HardHistogramLayer(
n_features=feature_extraction_module.output_size, n_bins=n_bins, quantiles=quantiles
)
self.histogram = _HardHistogramLayer(n_features=n_features, n_bins=n_bins, quantiles=quantiles)
self.output_size = self.histogram.output_size
self.output_function = output_function
self.output_module = nn.Sequential()
prev_size = self.histogram.output_size
for i, linear_size in enumerate(linear_sizes):
self.output_module.add_module(f'linear_{i}', nn.Linear(prev_size, linear_size))
self.output_module.add_module(f'leakyrelu_{i}', nn.LeakyReLU())
self.output_module.add_module(f'dropout_{i}', nn.Dropout(dropout))
prev_size = linear_size
self.output_module.add_module('last_linear', nn.Linear(prev_size, n_classes))
if output_function == 'softmax':
self.output_module.add_module('softmax', nn.Softmax(dim=1))
elif output_function == 'normalize':
self.output_module.add_module('relu', nn.ReLU())
else:
raise ValueError(f"unknown {output_function=}; valid ones are 'softmax', 'normalize'")
def forward(self, bag):
# bag: (batch_size, bag_size, n_features)
features = self.feature_extraction_module(bag)
features = self.sigmoid(features)
histogram = self.histogram(features)
out = self.output_module(histogram)
if self.output_function == 'normalize':
out = nn.functional.normalize(out, p=1, dim=1)
return out
def forward(self, input):
return self.histogram(self.sigmoid(input))
def _to_tensor(x, device):
if torch.is_tensor(x):
return x.to(device=device, dtype=torch.float32)
if hasattr(x, 'toarray'): # scipy sparse
x = x.toarray()
return torch.as_tensor(np.asarray(x), dtype=torch.float32, device=device)
def _stack_bags(bags, device):
"""
:param bags: an iterable of (X_bag, prevalence) pairs, all X_bag with the same number of instances
:return: a pair of tensors (X, P) of shape (n_bags, bag_size, n_features) and (n_bags, n_classes)
"""
Xs, ps = zip(*bags)
X = torch.stack([_to_tensor(x, device) for x in Xs])
P = torch.stack([_to_tensor(p, device) for p in ps])
return X, P
def _mix_two_bags(bag_a, bag_b, bag_size, rng):
"""Synthesizes a new bag of size `bag_size` by mixing two given bags with a random ratio, following
the "mixer" idea from the original HistNetQ repo (`UnlabeledMixerBagGenerator`): useful when the
only available training material is a modest number of pre-built samples (e.g., LeQua's dev
samples) and one wants extra intermediate-prevalence bags without access to instance-level labels.
"""
Xa, pa = bag_a
Xb, pb = bag_b
m = rng.random()
na = round(m * bag_size)
nb = bag_size - na
idx_a = rng.choices(range(len(Xa)), k=na) if na > 0 else []
idx_b = rng.choices(range(len(Xb)), k=nb) if nb > 0 else []
Xa, Xb = np.asarray(Xa), np.asarray(Xb)
X_mixed = np.concatenate([Xa[idx_a], Xb[idx_b]], axis=0)
p_mixed = m * np.asarray(pa, dtype=float) + (1 - m) * np.asarray(pb, dtype=float)
return X_mixed, p_mixed
class HistNetQ(BaseQuantifier):
class HistNetQ(BagTrainedQuantifier):
"""
Implementation of `HistNetQ <https://github.com/pglez84/histnetq>`_, a neural network for
quantification that learns a differentiable histogram-based representation of a sample, trained
@ -193,14 +113,9 @@ class HistNetQ(BaseQuantifier):
end-to-end on whole samples rather than on individually labeled instances, following a symmetric problem setting
(learning from bags, predicting on bags).
Training data can be provided in two ways:
* via :meth:`fit`, from a plain labelled collection (`X`, `y`): training/validation bags are then
generated by resampling from it using a QuaPy sampling protocol (:class:`quapy.protocol.UPP` by
default).
* via :meth:`fit_from_samples`, from a :class:`quapy.protocol.AbstractProtocol` that already yields
the training bags (e.g., :class:`quapy.data._lequa.SamplesFromDir` for LeQua-style pre-built
samples), optionally enriched with synthetic bags mixed from the given ones.
Training data can be provided in two ways: via :meth:`fit`, from a plain labelled collection; or via
:meth:`fit_from_samples`, from a protocol that already yields the training bags. See
:class:`quapy.method._neural_bags.BagTrainedQuantifier` for details.
:param feature_extraction_module: a `torch.nn.Module` exposing an `output_size` attribute, used to
embed each instance before computing the histogram (e.g., a small MLP for tabular data, a CNN
@ -270,213 +185,37 @@ class HistNetQ(BaseQuantifier):
checkpointdir='../checkpoint',
checkpointname=None,
verbose=0):
self.feature_extraction_module = feature_extraction_module
super().__init__(
feature_extraction_module=feature_extraction_module,
linear_sizes=linear_sizes,
dropout=dropout,
output_function=output_function,
bag_size=bag_size,
n_bags_train=n_bags_train,
n_bags_val=n_bags_val,
train_epochs=train_epochs,
patience=patience,
start_lr=start_lr,
end_lr=end_lr,
lr_factor=lr_factor,
weight_decay=weight_decay,
quant_loss=quant_loss,
batch_size=batch_size,
protocol=protocol,
protocol_params=protocol_params,
val_split=val_split,
device=device,
random_state=random_state,
checkpointdir=checkpointdir,
checkpointname=checkpointname,
verbose=verbose,
)
self.n_bins = n_bins
self.quantiles = quantiles
self.linear_sizes = linear_sizes
self.dropout = dropout
self.output_function = output_function
self.bag_size = bag_size
self.n_bags_train = n_bags_train
self.n_bags_val = n_bags_val
self.train_epochs = train_epochs
self.patience = patience
self.start_lr = start_lr
self.end_lr = end_lr
self.lr_factor = lr_factor
self.weight_decay = weight_decay
self.quant_loss = quant_loss if quant_loss is not None else torch.nn.L1Loss()
self.batch_size = batch_size
self.protocol = protocol
self.protocol_params = protocol_params
self.val_split = val_split
self.device = torch.device(device)
self.random_state = random_state
if checkpointname is None:
local_random = random.Random()
random_code = '-'.join(str(local_random.randint(0, 1000000)) for _ in range(5))
checkpointname = 'HistNetQ-' + random_code
self.checkpointdir = checkpointdir
self.checkpoint = os.path.join(checkpointdir, checkpointname)
self.verbose = verbose
self._classes_ = None
@property
def classes_(self):
return self._classes_
def _checkpoint_prefix(self):
return 'HistNetQ'
def fit(self, X, y):
"""
Trains HistNetQ from a plain labelled collection, generating training and validation bags by
resampling from it via `self.protocol` (a fresh random sequence of bags every epoch for
training, and a fixed, reproducible sequence for validation).
:param X: the training instances
:param y: the labels of X
:return: self
"""
data = LabelledCollection(X, y)
self._classes_ = data.classes_
train_data, val_data = data.split_stratified(train_prop=1 - self.val_split, random_state=self.random_state)
protocol_params = self.protocol_params or {}
def train_bags():
sampler = self.protocol(
train_data, sample_size=self.bag_size, repeats=self.n_bags_train, random_state=None,
**protocol_params
)
return sampler()
def val_bags():
sampler = self.protocol(
val_data, sample_size=self.bag_size, repeats=self.n_bags_val, random_state=self.random_state,
**protocol_params
)
return sampler()
n_features = train_data.instances.shape[1]
self._fit_loop(train_bags, val_bags, n_features, n_bags_train=self.n_bags_train, n_bags_val=self.n_bags_val)
return self
def fit_from_samples(self, protocol: AbstractProtocol, val_protocol: AbstractProtocol = None,
mix_bags=False, mix_bags_proportion=0.5):
"""
Trains HistNetQ from a protocol that already yields the training bags (e.g.,
:class:`quapy.data._lequa.SamplesFromDir`, for LeQua-style pre-built samples), instead of
resampling from a labelled collection. This is the entry point to use whenever only bags of
known prevalence are available (no instance-level labels).
:param protocol: an :class:`AbstractProtocol` yielding `(sample, prevalence)` pairs; consumed
once and kept in memory (expected to be of modest size, as is typical of pre-built sample
collections).
:param val_protocol: an optional, separate protocol providing the validation bags; if None, a
`val_split` fraction of the bags returned by `protocol` is held out instead.
:param mix_bags: if True, in addition to the bags returned by `protocol`, synthesize extra bags
each epoch by mixing random pairs of the given bags with a random ratio (a substitute for
the original repo's `UnlabeledMixerBagGenerator`, useful to broaden the coverage of
prevalence values beyond what the given bags exhibit).
:param mix_bags_proportion: proportion (relative to the number of base training bags) of extra
mixed bags to generate per epoch when `mix_bags=True` (default 0.5).
:return: self
"""
assert isinstance(protocol, AbstractProtocol), 'protocol must be an instance of AbstractProtocol'
base_bags = list(protocol())
n_classes = len(np.asarray(base_bags[0][1]))
self._classes_ = np.arange(n_classes)
if val_protocol is not None:
val_bags_list = list(val_protocol())
else:
n_val = max(1, int(len(base_bags) * self.val_split))
val_bags_list = base_bags[:n_val]
base_bags = base_bags[n_val:]
rng = random.Random(self.random_state)
n_mixed = round(len(base_bags) * mix_bags_proportion) if mix_bags else 0
def train_bags():
bags = list(base_bags)
if n_mixed > 0:
for _ in range(n_mixed):
a, b = rng.choice(base_bags), rng.choice(base_bags)
bags.append(_mix_two_bags(a, b, self.bag_size, rng))
rng.shuffle(bags)
return bags
def val_bags():
return val_bags_list
n_features = np.asarray(base_bags[0][0]).shape[1]
self._fit_loop(
train_bags, val_bags, n_features,
n_bags_train=len(base_bags) + n_mixed, n_bags_val=len(val_bags_list)
)
return self
def _fit_loop(self, train_bags_fn, val_bags_fn, n_features, n_bags_train, n_bags_val):
os.makedirs(self.checkpointdir, exist_ok=True)
n_classes = len(self._classes_)
fe = self.feature_extraction_module
if fe is None:
fe = _IdentityFeatureExtractionModule(n_features)
self.histnet = _HistNetModule(
fe, n_classes, n_bins=self.n_bins, quantiles=self.quantiles, linear_sizes=self.linear_sizes,
dropout=self.dropout, output_function=self.output_function
).to(self.device)
optim = torch.optim.Adam(self.histnet.parameters(), lr=self.start_lr, weight_decay=self.weight_decay)
scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optim, patience=self.patience, factor=self.lr_factor)
early_stop = EarlyStop(self.patience, lower_is_better=True)
best_state = copy.deepcopy(self.histnet.state_dict())
for epoch in range(self.train_epochs):
tr_loss = self._run_epoch(train_bags_fn(), n_bags_train, optim, train=True, epoch=epoch)
va_loss = self._run_epoch(val_bags_fn(), n_bags_val, optim=None, train=False, epoch=epoch)
early_stop(va_loss, epoch)
if early_stop.IMPROVED:
best_state = copy.deepcopy(self.histnet.state_dict())
torch.save(best_state, self.checkpoint)
elif early_stop.STOP:
if self.verbose > 0:
print(f'[HistNetQ] training ended by patience exhausted at epoch {epoch}; '
f'restoring best model from epoch {early_stop.best_epoch}')
break
scheduler.step(va_loss)
if optim.param_groups[0]['lr'] < self.end_lr:
if self.verbose > 0:
print(f'[HistNetQ] early stopping in epoch {epoch} (learning rate below end_lr)')
break
self.histnet.load_state_dict(best_state)
def _run_epoch(self, bags, n_bags, optim, train, epoch):
self.histnet.train(mode=train)
losses = []
pbar = tqdm(bags, total=n_bags, disable=self.verbose == 0)
batch = []
def process_batch(batch):
X, P = _stack_bags(batch, self.device)
if train:
optim.zero_grad()
P_hat = self.histnet.forward(X)
loss = self.quant_loss(P, P_hat)
loss.backward()
optim.step()
else:
with torch.no_grad():
P_hat = self.histnet.forward(X)
loss = self.quant_loss(P, P_hat)
return loss.item()
for bag in pbar:
batch.append(bag)
if len(batch) == self.batch_size:
losses.append(process_batch(batch))
batch = []
pbar.set_description(
f'[HistNetQ] epoch={epoch} {"train" if train else "val"}-loss={np.mean(losses):.5f}'
)
if batch:
losses.append(process_batch(batch))
return np.mean(losses) if losses else float('inf')
def predict(self, X):
"""
Generates a class prevalence estimate for the sample `X`, via a single forward pass of the
trained network (the histogram layer aggregates over however many instances are given, so `X`
need not match the `bag_size` used during training).
:param X: the test instances
:return: `np.ndarray` of shape `(n_classes,)` with the class prevalence estimates
"""
self.histnet.eval()
with torch.no_grad():
X_t = _to_tensor(X, self.device).unsqueeze(0)
prevalence = self.histnet.forward(X_t)
return prevalence.cpu().numpy().flatten()
def _build_quantmodule(self, n_features):
return _SigmoidHistogram(n_features=n_features, n_bins=self.n_bins, quantiles=self.quantiles)

443
quapy/method/_neural_bags.py Normal file
View File

@ -0,0 +1,443 @@
"""
Shared machinery for QuaPy's "bag-trained" neural quantifiers, i.e., methods that -- like
:class:`quapy.method.meta.QuaNet` -- do not follow the classify-then-aggregate pattern of
:class:`quapy.method.aggregative.AggregativeQuantifier`, but are instead trained and evaluated
end-to-end on whole samples ("bags") of known prevalence.
:class:`quapy.method._histnet.HistNetQ` and :class:`quapy.method._gmnet.GMNet` share the same overall
architecture (feature_extraction -> quantification module -> small MLP head -> softmax/normalize) and
the same bag-based training protocol (bag generation via a QuaPy sampling protocol, early stopping, LR
scheduling, checkpointing). This module factors that common part out into :class:`BagTrainedQuantifier`;
concrete subclasses only need to supply the quantification module placed between the feature extractor
and the shared head (see :meth:`BagTrainedQuantifier._build_quantmodule`).
"""
import copy
import os
import random
from abc import ABC, abstractmethod
import numpy as np
import torch
import torch.nn as nn
from tqdm import tqdm
from quapy.data import LabelledCollection
from quapy.method.base import BaseQuantifier
from quapy.protocol import AbstractProtocol, UPP
from quapy.util import EarlyStop
class IdentityFeatureExtractionModule(nn.Module):
"""Used when no feature extraction module is provided: instances are assumed to already be in
their final numeric representation."""
def __init__(self, input_size):
super().__init__()
self.output_size = input_size
def forward(self, x):
return x
def to_tensor(x, device):
if torch.is_tensor(x):
return x.to(device=device, dtype=torch.float32)
if hasattr(x, 'toarray'): # scipy sparse
x = x.toarray()
return torch.as_tensor(np.asarray(x), dtype=torch.float32, device=device)
def stack_bags(bags, device):
"""
:param bags: an iterable of (X_bag, prevalence) pairs, all X_bag with the same number of instances
:return: a pair of tensors (X, P) of shape (n_bags, bag_size, n_features) and (n_bags, n_classes)
"""
Xs, ps = zip(*bags)
X = torch.stack([to_tensor(x, device) for x in Xs])
P = torch.stack([to_tensor(p, device) for p in ps])
return X, P
def mix_two_bags(bag_a, bag_b, bag_size, rng):
"""Synthesizes a new bag of size `bag_size` by mixing two given bags with a random ratio, following
the "mixer" idea from the original HistNetQ repo (`UnlabeledMixerBagGenerator`): useful when the
only available training material is a modest number of pre-built samples (e.g., LeQua's dev
samples) and one wants extra intermediate-prevalence bags without access to instance-level labels.
"""
Xa, pa = bag_a
Xb, pb = bag_b
m = rng.random()
na = round(m * bag_size)
nb = bag_size - na
idx_a = rng.choices(range(len(Xa)), k=na) if na > 0 else []
idx_b = rng.choices(range(len(Xb)), k=nb) if nb > 0 else []
Xa, Xb = np.asarray(Xa), np.asarray(Xb)
X_mixed = np.concatenate([Xa[idx_a], Xb[idx_b]], axis=0)
p_mixed = m * np.asarray(pa, dtype=float) + (1 - m) * np.asarray(pb, dtype=float)
return X_mixed, p_mixed
def build_output_head(input_size, n_classes, linear_sizes, dropout, output_function):
"""Builds the small MLP + output activation shared by every bag-trained quantifier's head: a stack
of (Linear, LeakyReLU, Dropout) blocks sized by `linear_sizes`, followed by a final Linear to
`n_classes` and either a softmax or an L1-normalization (applied in :class:`BagNetworkModule`), both
yielding a valid prevalence vector."""
output_module = nn.Sequential()
prev_size = input_size
for i, linear_size in enumerate(linear_sizes):
output_module.add_module(f'linear_{i}', nn.Linear(prev_size, linear_size))
output_module.add_module(f'leakyrelu_{i}', nn.LeakyReLU())
output_module.add_module(f'dropout_{i}', nn.Dropout(dropout))
prev_size = linear_size
output_module.add_module('last_linear', nn.Linear(prev_size, n_classes))
if output_function == 'softmax':
output_module.add_module('softmax', nn.Softmax(dim=1))
elif output_function == 'normalize':
output_module.add_module('relu', nn.ReLU())
else:
raise ValueError(f"unknown {output_function=}; valid ones are 'softmax', 'normalize'")
return output_module
class BagNetworkModule(nn.Module):
"""The full network shared by every bag-trained quantifier: feature extraction, a pluggable
quantification module (mapping a bag of instance-level features to a single per-bag
representation), and the shared MLP head.
:param quantmodule: a `torch.nn.Module` exposing an `output_size` attribute, mapping a tensor of
shape (batch_size, bag_size, n_features) to one of shape (batch_size, quantmodule.output_size).
"""
def __init__(self, feature_extraction_module, quantmodule, n_classes, linear_sizes, dropout, output_function):
super().__init__()
self.feature_extraction_module = feature_extraction_module
self.quantmodule = quantmodule
self.output_function = output_function
self.output_module = build_output_head(
quantmodule.output_size, n_classes, linear_sizes, dropout, output_function
)
def forward(self, bag):
# bag: (batch_size, bag_size, n_features)
features = self.feature_extraction_module(bag)
representation = self.quantmodule(features)
out = self.output_module(representation)
if self.output_function == 'normalize':
out = nn.functional.normalize(out, p=1, dim=1)
return out
class BagTrainedQuantifier(BaseQuantifier, ABC):
"""
Base class for QuaPy's neural quantifiers trained end-to-end on whole samples ("bags") of known
prevalence, rather than following the classify-then-aggregate pattern of
:class:`quapy.method.aggregative.AggregativeQuantifier` (in the spirit of
:class:`quapy.method.meta.QuaNet`). Concrete subclasses only need to provide the quantification
module placed between the feature extractor and the shared MLP head (see
:meth:`_build_quantmodule`) and a checkpoint-name prefix (see :attr:`_checkpoint_prefix`); bag
generation, the training/validation loop, early stopping, LR scheduling, checkpointing, and
prediction are all shared.
Training data can be provided in two ways:
* via :meth:`fit`, from a plain labelled collection (`X`, `y`): training/validation bags are then
generated by resampling from it using a QuaPy sampling protocol (:class:`quapy.protocol.UPP` by
default).
* via :meth:`fit_from_samples`, from a :class:`quapy.protocol.AbstractProtocol` that already yields
the training bags (e.g., :class:`quapy.data._lequa.SamplesFromDir` for LeQua-style pre-built
samples), optionally enriched with synthetic bags mixed from the given ones.
:param feature_extraction_module: a `torch.nn.Module` exposing an `output_size` attribute, used to
embed each instance before the quantification module (e.g., a small MLP for tabular data, a CNN
for images). If None (default), an identity module is used, i.e., the instances in `X` are
assumed to already be in their final numeric representation.
:param linear_sizes: tuple of ints with the sizes of the linear layers used in the shared head
(default empty, i.e., only the final classification layer is used).
:param dropout: dropout applied after each of the `linear_sizes` layers (default 0).
:param output_function: either 'softmax' or 'normalize' (L1); both yield a valid prevalence vector
(default 'softmax').
:param bag_size: number of instances per training/validation bag (default 500).
:param n_bags_train: number of bags generated per training epoch (default 500).
:param n_bags_val: number of bags generated per validation epoch (default 500).
:param train_epochs: maximum number of training epochs (default 200).
:param patience: number of epochs without improvement in validation loss before early-stopping
(default 20).
:param start_lr: initial learning rate (default 1e-3).
:param end_lr: once the learning rate decays below this value, training stops (default 1e-6).
:param lr_factor: factor by which the learning rate is reduced after `patience` epochs without
improvement (default 0.1).
:param weight_decay: L2 regularization (default 0).
:param quant_loss: the quantification loss to minimize (default `torch.nn.L1Loss()`), called as
`quant_loss(true_prevalences, predicted_prevalences)`.
:param batch_size: number of bags per gradient update (default 16).
:param protocol: the :class:`quapy.protocol.AbstractStochasticSeededProtocol` subclass used by
:meth:`fit` to resample bags from the given labelled collection (default
:class:`quapy.protocol.UPP`, which draws bags with prevalence sampled uniformly at random from
the simplex).
:param protocol_params: dict of extra keyword arguments passed to `protocol` (besides `data`,
`sample_size`, `repeats`, and `random_state`, which are set internally); default None.
:param val_split: float in (0,1), the proportion of the collection given to :meth:`fit` that is held
out (via stratified sampling) for validation and early stopping (default 0.4).
:param device: `'cpu'` or `'cuda'` (default 'cpu').
:param random_state: seed used for the train/validation split and for the (fixed) validation
sampling sequence (default 0).
:param checkpointdir: directory where the best model found during training is stored (default
'../checkpoint').
:param checkpointname: name of the checkpoint file; if None (default), a random name prefixed by
:attr:`_checkpoint_prefix` is generated.
:param verbose: verbosity level; if >0, shows a progress bar with the current losses (default 0).
"""
def __init__(self,
feature_extraction_module=None,
linear_sizes=(),
dropout=0.,
output_function='softmax',
bag_size=500,
n_bags_train=500,
n_bags_val=500,
train_epochs=200,
patience=20,
start_lr=1e-3,
end_lr=1e-6,
lr_factor=0.1,
weight_decay=0.,
quant_loss=None,
batch_size=16,
protocol=UPP,
protocol_params=None,
val_split=0.4,
device='cpu',
random_state=0,
checkpointdir='../checkpoint',
checkpointname=None,
verbose=0):
self.feature_extraction_module = feature_extraction_module
self.linear_sizes = linear_sizes
self.dropout = dropout
self.output_function = output_function
self.bag_size = bag_size
self.n_bags_train = n_bags_train
self.n_bags_val = n_bags_val
self.train_epochs = train_epochs
self.patience = patience
self.start_lr = start_lr
self.end_lr = end_lr
self.lr_factor = lr_factor
self.weight_decay = weight_decay
self.quant_loss = quant_loss if quant_loss is not None else torch.nn.L1Loss()
self.batch_size = batch_size
self.protocol = protocol
self.protocol_params = protocol_params
self.val_split = val_split
self.device = torch.device(device)
self.random_state = random_state
if checkpointname is None:
local_random = random.Random()
random_code = '-'.join(str(local_random.randint(0, 1000000)) for _ in range(5))
checkpointname = f'{self._checkpoint_prefix}-{random_code}'
self.checkpointdir = checkpointdir
self.checkpoint = os.path.join(checkpointdir, checkpointname)
self.verbose = verbose
self._classes_ = None
self.model = None
@property
def classes_(self):
return self._classes_
@property
@abstractmethod
def _checkpoint_prefix(self):
"""Short name used as the default checkpoint filename prefix (e.g. 'HistNetQ', 'GMNet')."""
...
@abstractmethod
def _build_quantmodule(self, n_features):
"""Builds the module placed between the (already feature-extracted) instances and the shared
MLP head. Must expose an `output_size` attribute and accept input of shape
(batch_size, bag_size, n_features), returning one of shape (batch_size, output_size)."""
...
def _extra_loss(self):
"""Optional additional term added to the quantification loss during training (e.g., GMNet's CKA
regularization across GM layers). Returns 0 by default."""
return 0.
def fit(self, X, y):
"""
Trains the quantifier from a plain labelled collection, generating training and validation bags
by resampling from it via `self.protocol` (a fresh random sequence of bags every epoch for
training, and a fixed, reproducible sequence for validation).
:param X: the training instances
:param y: the labels of X
:return: self
"""
data = LabelledCollection(X, y)
self._classes_ = data.classes_
train_data, val_data = data.split_stratified(train_prop=1 - self.val_split, random_state=self.random_state)
protocol_params = self.protocol_params or {}
def train_bags():
sampler = self.protocol(
train_data, sample_size=self.bag_size, repeats=self.n_bags_train, random_state=None,
**protocol_params
)
return sampler()
def val_bags():
sampler = self.protocol(
val_data, sample_size=self.bag_size, repeats=self.n_bags_val, random_state=self.random_state,
**protocol_params
)
return sampler()
n_features = train_data.instances.shape[1]
self._fit_loop(train_bags, val_bags, n_features, n_bags_train=self.n_bags_train, n_bags_val=self.n_bags_val)
return self
def fit_from_samples(self, protocol: AbstractProtocol, val_protocol: AbstractProtocol = None,
mix_bags=False, mix_bags_proportion=0.5):
"""
Trains the quantifier from a protocol that already yields the training bags (e.g.,
:class:`quapy.data._lequa.SamplesFromDir`, for LeQua-style pre-built samples), instead of
resampling from a labelled collection. This is the entry point to use whenever only bags of
known prevalence are available (no instance-level labels).
:param protocol: an :class:`AbstractProtocol` yielding `(sample, prevalence)` pairs; consumed
once and kept in memory (expected to be of modest size, as is typical of pre-built sample
collections).
:param val_protocol: an optional, separate protocol providing the validation bags; if None, a
`val_split` fraction of the bags returned by `protocol` is held out instead.
:param mix_bags: if True, in addition to the bags returned by `protocol`, synthesize extra bags
each epoch by mixing random pairs of the given bags with a random ratio (a substitute for
the original HistNetQ repo's `UnlabeledMixerBagGenerator`, useful to broaden the coverage of
prevalence values beyond what the given bags exhibit).
:param mix_bags_proportion: proportion (relative to the number of base training bags) of extra
mixed bags to generate per epoch when `mix_bags=True` (default 0.5).
:return: self
"""
assert isinstance(protocol, AbstractProtocol), 'protocol must be an instance of AbstractProtocol'
base_bags = list(protocol())
n_classes = len(np.asarray(base_bags[0][1]))
self._classes_ = np.arange(n_classes)
if val_protocol is not None:
val_bags_list = list(val_protocol())
else:
n_val = max(1, int(len(base_bags) * self.val_split))
val_bags_list = base_bags[:n_val]
base_bags = base_bags[n_val:]
rng = random.Random(self.random_state)
n_mixed = round(len(base_bags) * mix_bags_proportion) if mix_bags else 0
def train_bags():
bags = list(base_bags)
if n_mixed > 0:
for _ in range(n_mixed):
a, b = rng.choice(base_bags), rng.choice(base_bags)
bags.append(mix_two_bags(a, b, self.bag_size, rng))
rng.shuffle(bags)
return bags
def val_bags():
return val_bags_list
n_features = np.asarray(base_bags[0][0]).shape[1]
self._fit_loop(
train_bags, val_bags, n_features,
n_bags_train=len(base_bags) + n_mixed, n_bags_val=len(val_bags_list)
)
return self
def _fit_loop(self, train_bags_fn, val_bags_fn, n_features, n_bags_train, n_bags_val):
os.makedirs(self.checkpointdir, exist_ok=True)
n_classes = len(self._classes_)
fe = self.feature_extraction_module
if fe is None:
fe = IdentityFeatureExtractionModule(n_features)
quantmodule = self._build_quantmodule(fe.output_size)
self.model = BagNetworkModule(
fe, quantmodule, n_classes, linear_sizes=self.linear_sizes, dropout=self.dropout,
output_function=self.output_function
).to(self.device)
optim = torch.optim.Adam(self.model.parameters(), lr=self.start_lr, weight_decay=self.weight_decay)
scheduler = torch.optim.lr_scheduler.ReduceLROnPlateau(optim, patience=self.patience, factor=self.lr_factor)
early_stop = EarlyStop(self.patience, lower_is_better=True)
best_state = copy.deepcopy(self.model.state_dict())
for epoch in range(self.train_epochs):
self._run_epoch(train_bags_fn(), n_bags_train, optim, train=True, epoch=epoch)
va_loss = self._run_epoch(val_bags_fn(), n_bags_val, optim=None, train=False, epoch=epoch)
early_stop(va_loss, epoch)
if early_stop.IMPROVED:
best_state = copy.deepcopy(self.model.state_dict())
torch.save(best_state, self.checkpoint)
elif early_stop.STOP:
if self.verbose > 0:
print(f'[{self._checkpoint_prefix}] training ended by patience exhausted at epoch {epoch}; '
f'restoring best model from epoch {early_stop.best_epoch}')
break
scheduler.step(va_loss)
if optim.param_groups[0]['lr'] < self.end_lr:
if self.verbose > 0:
print(f'[{self._checkpoint_prefix}] early stopping in epoch {epoch} (learning rate below end_lr)')
break
self.model.load_state_dict(best_state)
def _run_epoch(self, bags, n_bags, optim, train, epoch):
self.model.train(mode=train)
losses = []
pbar = tqdm(bags, total=n_bags, disable=self.verbose == 0)
batch = []
def process_batch(batch):
X, P = stack_bags(batch, self.device)
if train:
optim.zero_grad()
P_hat = self.model.forward(X)
loss = self.quant_loss(P, P_hat) + self._extra_loss()
loss.backward()
optim.step()
else:
with torch.no_grad():
P_hat = self.model.forward(X)
loss = self.quant_loss(P, P_hat)
return loss.item()
for bag in pbar:
batch.append(bag)
if len(batch) == self.batch_size:
losses.append(process_batch(batch))
batch = []
pbar.set_description(
f'[{self._checkpoint_prefix}] epoch={epoch} {"train" if train else "val"}-'
f'loss={np.mean(losses):.5f}'
)
if batch:
losses.append(process_batch(batch))
return np.mean(losses) if losses else float('inf')
def predict(self, X):
"""
Generates a class prevalence estimate for the sample `X`, via a single forward pass of the
trained network (the quantification module aggregates over however many instances are given, so
`X` need not match the `bag_size` used during training).
:param X: the test instances
:return: `np.ndarray` of shape `(n_classes,)` with the class prevalence estimates
"""
self.model.eval()
with torch.no_grad():
X_t = to_tensor(X, self.device).unsqueeze(0)
prevalence = self.model.forward(X_t)
return prevalence.cpu().numpy().flatten()

11
quapy/method/meta.py
View File

@ -37,6 +37,17 @@ if _histnet:
else:
HistNetQ = "HistNetQ is not available due to missing torch package"
try:
from . import _gmnet
except ModuleNotFoundError:
_gmnet = None
if _gmnet:
GMNet = _gmnet.GMNet
else:
GMNet = "GMNet is not available due to missing torch and/or geotorch packages"
class MedianEstimator(BinaryQuantifier):
"""

32
quapy/tests/test_methods.py
View File

@ -127,6 +127,38 @@ class TestMethods(unittest.TestCase):
estim_prevalences2 = model2.predict(dataset.test.X)
self.assertTrue(check_prevalence_vector(estim_prevalences2))
def test_gmnet(self):
try:
import torch
import geotorch
except ModuleNotFoundError:
print('the torch and/or geotorch packages are not installed; skipping unit test for GMNet')
return
from quapy.method.meta import GMNet
from quapy.protocol import UPP
for dataset in TestMethods.datasets:
# single GM layer, no CKA regularization
model = GMNet(
bag_size=20, n_bags_train=10, n_bags_val=5, train_epochs=2, patience=1, batch_size=2,
device='cpu', checkpointdir='./checkpoint_test_gmnet'
)
model.fit(*dataset.training.Xy)
estim_prevalences = model.predict(dataset.test.X)
self.assertTrue(check_prevalence_vector(estim_prevalences))
# multiple GM layers + CKA regularization, and fit_from_samples
given_samples = UPP(dataset.training, sample_size=20, repeats=8, random_state=1)
val_samples = UPP(dataset.training, sample_size=20, repeats=4, random_state=2)
model2 = GMNet(
n_gm_layers=2, num_gaussians=3, gaussian_dimensions=4, cka_regularization=0.1,
bag_size=20, train_epochs=2, patience=1, batch_size=2, device='cpu',
checkpointdir='./checkpoint_test_gmnet'
)
model2.fit_from_samples(given_samples, val_protocol=val_samples, mix_bags=True)
estim_prevalences2 = model2.predict(dataset.test.X)
self.assertTrue(check_prevalence_vector(estim_prevalences2))
def test_composable(self):
try:

2
setup.py
View File

@ -131,7 +131,7 @@ setup(
# projects.
extras_require={ # Optional
'bayes': ['jax', 'jaxlib', 'numpyro', 'pystan', 'setuptools<82'],
'neural': ['torch'],
'neural': ['torch', 'geotorch'],
'tests': ['certifi'],
'docs' : ['pydata-sphinx-theme', 'myst-parser', 'sphinx-design'],
},