Ensemble Adaptive

adaptive

CLASS	DESCRIPTION
AdaptiveWeightedEnsemble	Ensemble with adaptive weight learning based on performance metrics.

Attributes

Classes

AdaptiveWeightedEnsemble(estimators: list[ModelType] | dict[str, ModelType], weight_method: Literal['cv_score', 'reconstruction', 'stability', 'embedding_quality', 'spectral_quality', 'uniform'] = 'cv_score', n_folds: int = 5, regularization: float = 0.0, eps: float = torch.finfo(torch.float32).eps, fit_manager: FitManager | None = None)

Bases: Ensemble

Ensemble with adaptive weight learning based on performance metrics.

Learns optimal estimator weights for combining estimators using various strategies: cross-validation scores, reconstruction quality, stability measures, embedding quality, or spectral quality.

PARAMETER	DESCRIPTION
estimators	Estimators to combine adaptively. TYPE: list[ModelType] | dict[str, ModelType]
weight_method	Weight learning method. "cv_score": Cross-validation based weights "reconstruction": Based on reconstruction error "stability": Based on pairwise Procrustes stability "embedding_quality": Based on trustworthiness + continuity "spectral_quality": Based on eigenvalue/eigenvector alignment "uniform": Equal weights (no adaptation) TYPE: str, by default "cv_score" DEFAULT: 'cv_score'
reduce	How to reduce predictions (must be "weighted_mean" for adaptive). TYPE: str, by default "weighted_mean"
n_folds	Number of CV folds for weight_method="cv_score". TYPE: int, by default 5 DEFAULT: 5
regularization	L2 regularization on estimator weights (encourages equal weights). TYPE: float, by default 0.0 DEFAULT: 0.0
eps	Small constant to avoid division by zero when normalizing weights. TYPE: float, by default torch.finfo(torch.float32).eps DEFAULT: eps
fit_manager	Manager for unified training of both Estimator and Parametric models. Required for using Parametric (PyTorch Lightning) models in adaptive ensemble. See FitManager for configuration options. TYPE: FitManager | None DEFAULT: None

ATTRIBUTE	DESCRIPTION
estimators	Dictionary of fitted estimators. TYPE: dict[str, ModelType]
estimator_weights	Learned optimal weights for each estimator. TYPE: dict[str, float]

Examples:

Adaptive ensemble based on cross-validation:

>>> from spectre.decomposition import DiffusionMap, PCA
>>> from spectre.ensemble import AdaptiveWeightedEnsemble
>>>
>>> dm = DiffusionMap(...)
>>> pca = PCA(...)
>>>
>>> adaptive = AdaptiveWeightedEnsemble(
...     estimators={"dm": dm, "pca": pca},
...     weight_method="cv_score",
...     n_folds=5,
... )
>>> adaptive.fit(X)
>>> print(adaptive.estimator_weights)  # {'dm': 0.7, 'pca': 0.3}

Stability-based weighting:

>>> adaptive_stable = AdaptiveWeightedEnsemble(
...     estimators=[dm1, dm2, dm3],
...     weight_method="stability",
... )
>>> adaptive_stable.fit(X)
>>> X_embedded = adaptive_stable.predict(X)

Embedding quality weighting:

>>> adaptive_quality = AdaptiveWeightedEnsemble(
...     estimators={"method1": est1, "method2": est2},
...     weight_method="embedding_quality",
... )
>>> adaptive_quality.fit(X)

METHOD	DESCRIPTION
fit	Fit all estimators and learn optimal estimator_weights.
predict	Predict using weighted combination of estimators.
score	Compute weighted score across estimators.

Source code in spectre/ensemble/adaptive.py

def __init__(
    self,
    estimators: list[ModelType] | dict[str, ModelType],
    weight_method: Literal[
        "cv_score",
        "reconstruction",
        "stability",
        "embedding_quality",
        "spectral_quality",
        "uniform",
    ] = "cv_score",
    n_folds: int = 5,
    regularization: float = 0.0,
    eps: float = torch.finfo(torch.float32).eps,
    fit_manager: FitManager | None = None,
):
    super().__init__(estimators, fit_manager=fit_manager)

    allowed_weight_method = [
        "cv_score",
        "reconstruction",
        "stability",
        "embedding_quality",
        "spectral_quality",
        "uniform",
    ]
    if weight_method not in allowed_weight_method:
        raise ValueError(
            f"Unknown weight_method '{weight_method}'. Options: {allowed_weight_method}."
        )
    self.weight_method = weight_method
    self.reduce = "weighted_mean"

    if not isinstance(n_folds, int):
        raise TypeError(f"n_folds must be an integer, got {type(n_folds)}.")
    check_in_interval(n_folds, "[2, inf)")
    self.n_folds = n_folds

    if not isinstance(regularization, (int, float)):
        raise TypeError(
            f"`regularization` must be a number, got {type(regularization).__name__}."
        )
    if regularization < 0:
        raise ValueError(f"`regularization` must be >= 0, got {regularization}.")
    self.regularization = float(regularization)

    self.eps = eps

    # Storage for learned estimator weights
    self._estimator_weights = None

Attributes

weights: dict[str, float] property

Return learned estimator weights.

Functions

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

Fit all estimators and learn optimal estimator_weights.

PARAMETER	DESCRIPTION
X	Training data of shape (n_samples, in_features). TYPE: Tensor
weights	Sample weights of shape (n_samples, ). TYPE: Tensor | None DEFAULT: None
target	Target values for supervised methods. TYPE: Tensor | None DEFAULT: None

RETURNS	DESCRIPTION
AdaptiveWeightedEnsemble	Returns self for method chaining.

Source code in spectre/ensemble/adaptive.py

def fit(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
) -> "AdaptiveWeightedEnsemble":
    """
    Fit all estimators and learn optimal estimator_weights.

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

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

    target : torch.Tensor | None, optional
        Target values for supervised methods.

    Returns
    -------
    AdaptiveWeightedEnsemble
        Returns self for method chaining.
    """
    check_2d(X)

    self.in_features = X.shape[1]

    for name, estimator in self._estimators.items():
        self._fit_estimator(
            estimator, X, weights=weights, target=target, estimator_name=name
        )

    # Set output features from first estimator
    dummy_pred = list(self._estimators.values())[0].predict(X[:1])
    self.out_features = dummy_pred.shape[1]

    # Learn estimator weights based on method
    if self.weight_method == "uniform":
        self._estimator_weights = self.get_equal_estimator_weights()
    elif self.weight_method == "cv_score":
        self._estimator_weights = self._learn_weights_cv(X, weights, target)
    elif self.weight_method == "reconstruction":
        self._estimator_weights = self._learn_weights_reconstruction(X)
    elif self.weight_method == "stability":
        self._estimator_weights = self._learn_weights_stability(X)
    elif self.weight_method == "embedding_quality":
        self._estimator_weights = self._learn_weights_embedding_quality(X)
    elif self.weight_method == "spectral_quality":
        self._estimator_weights = self._learn_weights_spectral_quality()
    else:
        raise ValueError(f"Unknown weight method: {self.weight_method}")

    self.is_fitted = True

    return self

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

Predict using weighted combination of estimators.

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

RETURNS	DESCRIPTION
Tensor	Weighted prediction of shape (n_samples, out_features).

Source code in spectre/ensemble/adaptive.py

def predict(self, X: torch.Tensor) -> torch.Tensor:
    """
    Predict using weighted combination of estimators.

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

    Returns
    -------
    torch.Tensor
        Weighted prediction of shape (n_samples, out_features).
    """
    self.check_fitted()
    check_2d(X)

    if X.shape[1] != self.in_features:
        raise ValueError(f"Expected {self.in_features} features, got {X.shape[1]}.")

    predictions = self.predict_ensemble(X)

    return self.reduce_predictions(
        predictions,
        reduce="weighted_mean",
        estimator_weights=self._estimator_weights,
    )

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

Compute weighted score across estimators.

PARAMETER	DESCRIPTION
X	Input data. TYPE: Tensor
weights	Sample weights. TYPE: Tensor | None DEFAULT: None
target	Target values. TYPE: Tensor | None DEFAULT: None

RETURNS	DESCRIPTION
float	Weighted mean score across estimators.

Source code in spectre/ensemble/adaptive.py

def score(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
) -> float:
    """
    Compute weighted score across estimators.

    Parameters
    ----------
    X : torch.Tensor
        Input data.

    weights : torch.Tensor | None, optional
        Sample weights.

    target : torch.Tensor | None, optional
        Target values.

    Returns
    -------
    float
        Weighted mean score across estimators.
    """
    return self.reduce_scores(
        X,
        weights=weights,
        target=target,
        reduce="weighted_mean",
        estimator_weights=self._estimator_weights,
    )

Functions