Ensemble Base

base

CLASS	DESCRIPTION
Ensemble	Abstract base class for ensemble methods.

ATTRIBUTE	DESCRIPTION
ModelType	Type alias for models that can be used in ensembles.

Attributes

ModelType = Estimator | Parametric module-attribute

Type alias for models that can be used in ensembles.

A union type representing models that can be included in ensemble methods. Can be either an Estimator (non-parametric methods like PCA, diffusion maps) or a Parametric model (neural network-based methods like autoencoders, spectral maps).

Type

Union[Estimator, Parametric]

Classes

Ensemble(estimators: list[ModelType] | dict[str, ModelType], fit_manager: FitManager | None = None)

Bases: Estimator, ABC

Abstract base class for ensemble methods.

Provides common functionality for combining multiple estimators or parametric models, including model validation, Procrustes alignment, score aggregation, and weighting.

This class provides a unified interface for ensemble learning across different model types:

• Estimator methods (i.e., non-parametric): diffusion maps, PCA, kernel PCA
• Parametric models: autoencoders, spectral maps

PARAMETER	DESCRIPTION
estimators	Models to create an ensemble from. Can be list for unnamed models or dictionary for named models. Each model must be of type Parametric or Estimator (ModelType). TYPE: list[ModelType] | dict[str, ModelType]
fit_manager	Manager for unified training of both Estimator and Parametric models. If provided, enables ensemble methods to work with PyTorch Lightning-based Parametric models by handling Trainer creation and configuration. If None, uses direct estimator.fit() calls (backward compatible). See FitManager for configuration options. TYPE: FitManager | None DEFAULT: None

ATTRIBUTE	DESCRIPTION
estimators	Dictionary of fitted estimators with names as keys. TYPE: dict[str, ModelType]
estimator_names	List of estimator names in order. TYPE: list[str]
is_fitted	Whether the ensemble has been fitted. TYPE: bool

Notes

Subclasses must implement:

• Fit ensemble to data: fit( X: torch.Tensor, weights: torch.Tensor | None = None, target: torch.Tensor | None = None)
• Generate ensemble predictions: predict(X: torch.Tensor)
• Compute ensemble score: score( X: torch.Tensor, weights: torch.Tensor | None = None, target: torch.Tensor | None = None)

The class automatically handles:

• Procrustes alignment between embeddings
• Model cloning for independent fitting
• Score aggregation across estimators
• Weight normalization

Examples:

Create ensemble with named estimators:

>>> from spectre.decomposition import DiffusionMap, PCA
>>> from spectre.ensemble import EmbeddingEnsemble
>>>
>>> dm = DiffusionMap(...)
>>> pca = PCA(...)
>>>
>>> ensemble = EmbeddingEnsemble(
...     estimators={"diffusion": dm, "pca": pca},
...     reduce="mean",
... )
>>> ensemble.fit(X)
>>> X_embedded = ensemble.predict(X)

Create ensemble with unnamed estimators:

>>> estimators = [dm1, dm2, dm3]
>>> ensemble = EmbeddingEnsemble(estimators, reduce="median")
>>> ensemble.fit(X)

Access individual estimators:

>>> for name, estimator in ensemble.estimators.items():
...     print(f"{name}: {estimator.out_features} features")

Compute scores with different reduction methods:

>>> mean_score = ensemble.reduce_scores(X, reduce="mean")
>>> weighted_score = ensemble.reduce_scores(
...     X, reduce="weighted_mean", estimator_weights={"diffusion": 0.7, "pca": 0.3}
... )

METHOD	DESCRIPTION
clone_estimator	Clone an estimator for independent fitting.
align	Align embedding to target using Procrustes analysis.
reduce_scores	Compute scores from all estimators and reduce them to a scalar.
predict_ensemble	Get predictions from all estimators.
reduce_predictions	Reduce predictions from multiple estimators.
normalize_estimator_weights	Normalize estimator weights to sum to 1.
get_equal_estimator_weights	Get equal estimator weights for all estimators.
fit	Fit the ensemble to data.
predict	Generate ensemble predictions.
score	Compute ensemble score.

Source code in spectre/ensemble/base.py

def __init__(
    self,
    estimators: list[ModelType] | dict[str, ModelType],
    fit_manager: FitManager | None = None,
):
    super().__init__()

    # Convert estimators to dict with names
    if isinstance(estimators, dict):
        self._is_named = True
        self._estimators = estimators
        self._estimator_names = list(estimators.keys())
    elif isinstance(estimators, list):
        self._is_named = False
        self._estimators = {
            f"estimator_{i}": est for i, est in enumerate(estimators)
        }
        self._estimator_names = list(self._estimators.keys())
    else:
        raise TypeError(
            f"`estimators` must be list or dict, got {type(estimators).__name__}."
        )

    # Validate all are ModelType
    for name, estimator in self._estimators.items():
        if not isinstance(estimator, ModelType):
            raise TypeError(
                f"Estimator '{name}' must be Estimator or Parametric, "
                f"got {type(estimator).__name__}."
            )

    if not self._estimators:
        raise ValueError("At least one estimator must be provided.")

    # Store FitManager for unified training
    self._fit_manager = fit_manager

Attributes

estimators: dict[str, ModelType] property

Return fitted estimators.

estimator_names: list[str] property

Return list of estimator names.

is_named: bool property

Return whether estimators were provided with names.

Functions

clone_estimator(estimator: ModelType) -> ModelType

Clone an estimator for independent fitting.

PARAMETER	DESCRIPTION
estimator	Estimator or Parametric model to clone. TYPE: ModelType

RETURNS	DESCRIPTION
ModelType	Deep copy of the estimator.

Source code in spectre/ensemble/base.py

def clone_estimator(self, estimator: ModelType) -> ModelType:
    """
    Clone an estimator for independent fitting.

    Parameters
    ----------
    estimator : ModelType
        Estimator or Parametric model to clone.

    Returns
    -------
    ModelType
        Deep copy of the estimator.
    """
    return copy.deepcopy(estimator)

align(embedding: torch.Tensor, target: torch.Tensor, scaling: bool = True) -> torch.Tensor

Align embedding to target using Procrustes analysis.

See [procrustes][spectre.utils.compute.procrustes] function for detailed documentation.

PARAMETER	DESCRIPTION
embedding	Source embedding to align, shape (n_samples, n_components). TYPE: Tensor
target	Target embedding, shape (n_samples, n_components). TYPE: Tensor
scaling	Whether to apply optimal scaling after rotation. TYPE: bool, by default True DEFAULT: True

RETURNS	DESCRIPTION
Tensor	Aligned embedding, shape (n_samples, n_components).

Source code in spectre/ensemble/base.py

def align(
    self,
    embedding: torch.Tensor,
    target: torch.Tensor,
    scaling: bool = True,
) -> torch.Tensor:
    """
    Align embedding to target using Procrustes analysis.

    See [procrustes][spectre.utils.compute.procrustes] function
    for detailed documentation.

    Parameters
    ----------
    embedding : torch.Tensor
        Source embedding to align, shape (n_samples, n_components).

    target : torch.Tensor
        Target embedding, shape (n_samples, n_components).

    scaling : bool, by default True
        Whether to apply optimal scaling after rotation.

    Returns
    -------
    torch.Tensor
        Aligned embedding, shape (n_samples, n_components).
    """
    return procrustes(embedding, target, scaling=scaling)

reduce_scores(X: torch.Tensor, weights: torch.Tensor | None = None, target: torch.Tensor | None = None, reduce: str = 'mean', estimator_weights: dict[str, float] | None = None) -> float

Compute scores from all estimators and reduce them to a scalar.

Uses each estimator's score or score_step method to compute individual scores.

PARAMETER	DESCRIPTION
X	Input data of shape (n_samples, in_features). TYPE: Tensor
weights	Sample weights of shape (n_samples, ). TYPE: torch.Tensor | None, optional, by default None DEFAULT: None
target	Target values of shape (n_samples, ). TYPE: torch.Tensor | None, optional, by default None DEFAULT: None
reduce	Reduction method: "mean", "median", "weighted_mean", "min", "max". TYPE: str, by default "mean" DEFAULT: 'mean'
estimator_weights	Weights for weighted_mean reduction. Must match estimator names. TYPE: dict[str, float] | None, optional, by default None DEFAULT: None

RETURNS	DESCRIPTION
float	Reduced score across all estimators.

RAISES	DESCRIPTION
ValueError	If no estimators have a score method or reduction is unknown.

Source code in spectre/ensemble/base.py

def reduce_scores(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
    reduce: str = "mean",
    estimator_weights: dict[str, float] | None = None,
) -> float:
    """
    Compute scores from all estimators and reduce them to a scalar.

    Uses each estimator's `score` or `score_step` method to compute individual
    scores.

    Parameters
    ----------
    X : torch.Tensor
        Input data of shape (n_samples, in_features).

    weights : torch.Tensor | None, optional, by default None
        Sample weights of shape (n_samples, ).

    target : torch.Tensor | None, optional, by default None
        Target values of shape (n_samples, ).

    reduce : str, by default "mean"
        Reduction method: "mean", "median", "weighted_mean", "min", "max".

    estimator_weights : dict[str, float] | None, optional, by default None
        Weights for `weighted_mean` reduction. Must match estimator names.

    Returns
    -------
    float
        Reduced score across all estimators.

    Raises
    ------
    ValueError
        If no estimators have a score method or reduction is unknown.
    """
    self.check_fitted()

    scores = []
    score_names = []

    for name, estimator in self._estimators.items():
        # Try score method first (Estimator)
        if hasattr(estimator, "score") and callable(estimator.score):
            score = estimator.score(X, weights=weights, target=target)
            scores.append(float(score))
            score_names.append(name)
        # Fall back to score_step for Parametric models
        elif hasattr(estimator, "score_step") and callable(estimator.score_step):
            with torch.no_grad():
                score = estimator.score_step(X, weights=weights, target=target)
            scores.append(float(score))
            score_names.append(name)

    if not scores:
        raise ValueError("No estimators have a `score` or `score_step` method.")

    if not isinstance(reduce, str):
        raise TypeError(f"`reduce` must be a string, got {type(reduce).__name__}.")

    scores_tensor = torch.tensor(scores)

    if reduce == "mean":
        return float(scores_tensor.mean().item())
    elif reduce == "median":
        return float(scores_tensor.median().item())
    elif reduce == "weighted_mean":
        if estimator_weights is None:
            raise ValueError("`estimator_weights` required for weighted_mean.")

        # Filter weights to only include scored estimators
        weights_list = [estimator_weights[name] for name in score_names]
        weights_tensor = torch.tensor(weights_list)
        weights_tensor = weights_tensor / weights_tensor.sum()

        return float((scores_tensor * weights_tensor).sum().item())
    elif reduce == "min":
        return float(scores_tensor.min().item())
    elif reduce == "max":
        return float(scores_tensor.max().item())
    else:
        raise ValueError(
            f"Unknown reduction '{reduce}'. "
            f"Options: ['mean', 'median', 'weighted_mean', 'min', 'max']."
        )

predict_ensemble(X: torch.Tensor) -> dict[str, torch.Tensor]

Get predictions from all estimators.

PARAMETER	DESCRIPTION
X	Input data of shape (n_samples, in_features). TYPE: Tensor

RETURNS	DESCRIPTION
dict[str, Tensor]	Dictionary mapping estimator names to their predictions.

Source code in spectre/ensemble/base.py

def predict_ensemble(
    self,
    X: torch.Tensor,
) -> dict[str, torch.Tensor]:
    """
    Get predictions from all estimators.

    Parameters
    ----------
    X : torch.Tensor
        Input data of shape (n_samples, in_features).

    Returns
    -------
    dict[str, torch.Tensor]
        Dictionary mapping estimator names to their predictions.
    """
    self.check_fitted()

    prediction_values = {}
    for name, estimator in self._estimators.items():
        if isinstance(estimator, Parametric):
            # Use predict for Parametric (handles no_grad internally)
            # TODO: check if it works correctly with gradients.
            pred = estimator.predict(X)
        else:
            # Use predict for Estimator
            pred = estimator.predict(X)
        prediction_values[name] = pred

    return prediction_values

reduce_predictions(predictions: list[torch.Tensor] | dict[str, torch.Tensor], reduce: str = 'mean', estimator_weights: dict[str, float] | None = None) -> torch.Tensor

Reduce predictions from multiple estimators.

PARAMETER	DESCRIPTION
predictions	Predictions to aggregate. If dict, keys are estimator names. TYPE: list[Tensor] | dict[str, Tensor]
reduce	Reduction method. TYPE: str, by default "mean" DEFAULT: 'mean'
estimator_weights	Weights for 'weighted_mean' reduce. Keys must match prediction dict keys. TYPE: dict[str, float] | None DEFAULT: None

RETURNS	DESCRIPTION
Tensor	Reduced predictions.

Source code in spectre/ensemble/base.py

def reduce_predictions(
    self,
    predictions: list[torch.Tensor] | dict[str, torch.Tensor],
    reduce: str = "mean",
    estimator_weights: dict[str, float] | None = None,
) -> torch.Tensor:
    """
    Reduce predictions from multiple estimators.

    Parameters
    ----------
    predictions : list[torch.Tensor] | dict[str, torch.Tensor]
        Predictions to aggregate. If dict, keys are estimator names.

    reduce : str, by default "mean"
        Reduction method.

    estimator_weights : dict[str, float] | None, optional
        Weights for 'weighted_mean' reduce. Keys must match prediction dict keys.

    Returns
    -------
    torch.Tensor
        Reduced predictions.
    """
    # Convert dict to list if needed
    if isinstance(predictions, dict):
        pred_list = list(predictions.values())
        pred_names = list(predictions.keys())
    else:
        pred_list = predictions
        pred_names = [f"pred_{i}" for i in range(len(pred_list))]

    # Stack predictions
    stacked = torch.stack(pred_list, dim=0)

    # Convert reduce to enum
    if not isinstance(reduce, str):
        raise ValueError(f"Unknown reduction '{reduce}'.")

    # Aggregate
    if reduce == "mean":
        return stacked.mean(dim=0)
    elif reduce == "median":
        return stacked.median(dim=0).values
    elif reduce == "weighted_mean":
        if estimator_weights is None:
            raise ValueError(
                "`estimator_weights` required for weighted_mean aggregation."
            )
        # Get weights as tensor
        if isinstance(predictions, dict):
            weights_list = [estimator_weights[name] for name in pred_names]
        else:
            weights_list = list(estimator_weights.values())
        weights_tensor = torch.tensor(
            weights_list,
            device=stacked.device,
            dtype=stacked.dtype,
        )
        # Normalize weights
        weights_tensor = weights_tensor / weights_tensor.sum()
        # Reshape for broadcasting: (n_estimators, 1, 1, ...)
        weights_tensor = weights_tensor.view(-1, *([1] * (stacked.ndim - 1)))

        return (stacked * weights_tensor).sum(dim=0)
    elif reduce == "min":
        return stacked.min(dim=0).values
    elif reduce == "max":
        return stacked.max(dim=0).values
    else:
        raise ValueError(f"Unknown reduce: {reduce}")

normalize_estimator_weights(estimator_weights: dict[str, float] | list[float]) -> dict[str, float]

Normalize estimator weights to sum to 1.

PARAMETER	DESCRIPTION
estimator_weights	Weights to normalize. If dict, keys must match estimator names. If list, must match number of estimators. TYPE: dict[str, float] | list[float]

RETURNS	DESCRIPTION
dict[str, float]	Normalized weights as dictionary.

RAISES	DESCRIPTION
ValueError	If weights structure doesn't match estimators.

Source code in spectre/ensemble/base.py

def normalize_estimator_weights(
    self,
    estimator_weights: dict[str, float] | list[float],
) -> dict[str, float]:
    """
    Normalize estimator weights to sum to 1.

    Parameters
    ----------
    estimator_weights : dict[str, float] | list[float]
        Weights to normalize. If dict, keys must match estimator names.
        If list, must match number of estimators.

    Returns
    -------
    dict[str, float]
        Normalized weights as dictionary.

    Raises
    ------
    ValueError
        If weights structure doesn't match estimators.
    """
    if isinstance(estimator_weights, dict):
        if set(estimator_weights.keys()) != set(self._estimator_names):
            raise ValueError(
                f"Weight keys {set(estimator_weights.keys())} must match "
                f"estimator keys {set(self._estimator_names)}."
            )
        total = sum(estimator_weights.values())
        return {k: v / total for k, v in estimator_weights.items()}
    elif isinstance(estimator_weights, list):
        if len(estimator_weights) != len(self._estimators):
            raise ValueError(
                f"Number of weights ({len(estimator_weights)}) must match "
                f"number of estimators ({len(self._estimators)})."
            )
        total = sum(estimator_weights)
        return {
            name: w / total
            for name, w in zip(self._estimator_names, estimator_weights)
        }
    else:
        raise TypeError(
            f"weights must be dict or list, got {type(estimator_weights).__name__}."
        )

get_equal_estimator_weights() -> dict[str, float]

Get equal estimator weights for all estimators.

RETURNS	DESCRIPTION
dict[str, float]	Dictionary mapping estimator names to equal weights (1/n_estimators).

Source code in spectre/ensemble/base.py

def get_equal_estimator_weights(self) -> dict[str, float]:
    """
    Get equal estimator weights for all estimators.

    Returns
    -------
    dict[str, float]
        Dictionary mapping estimator names to equal weights (1/n_estimators).
    """
    n_estimators = len(self._estimators)
    return {name: 1.0 / n_estimators for name in self._estimator_names}

fit(X: torch.Tensor, weights: torch.Tensor | None = None, target: torch.Tensor | None = None) -> Ensemble abstractmethod

Fit the ensemble to data.

Source code in spectre/ensemble/base.py

@abstractmethod
def fit(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
) -> "Ensemble":
    """Fit the ensemble to data."""
    raise NotImplementedError()

predict(X: torch.Tensor) -> torch.Tensor abstractmethod

Generate ensemble predictions.

Source code in spectre/ensemble/base.py

@abstractmethod
def predict(self, X: torch.Tensor) -> torch.Tensor:
    """Generate ensemble predictions."""
    raise NotImplementedError()

score(X: torch.Tensor, weights: torch.Tensor | None = None, target: torch.Tensor | None = None) -> float abstractmethod

Compute ensemble score.

Source code in spectre/ensemble/base.py

@abstractmethod
def score(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
) -> float:
    """Compute ensemble score."""
    raise NotImplementedError()