Ensemble Consensus

consensus

CLASS	DESCRIPTION
ConsensusEmbedding	Consensus embedding by aligning and reducing multiple embeddings.

Attributes

Classes

ConsensusEmbedding(estimators: list[ModelType] | dict[str, ModelType], consensus_method: str = 'procrustes_mean', reference_idx: int | str | None = None, estimator_weights: dict[str, float] | None = None, n_iter: int = 5, fit_manager: FitManager | None = None)

Bases: Ensemble

Consensus embedding by aligning and reducing multiple embeddings.

Finds consensus low-dimensional representation by aligning embeddings from multiple methods using Procrustes analysis and reduction with various strategies (mean, median, weighted).

PARAMETER	DESCRIPTION
estimators	Estimators to combine. TYPE: list[ModelType] | dict[str, ModelType]
consensus_method	Method for finding consensus: "procrustes_mean", "procrustes_median", "weighted_procrustes", "reference_alignment". TYPE: str, by default "procrustes_mean" DEFAULT: 'procrustes_mean'
reference_idx	Index or name of reference estimator for "reference_alignment". TYPE: int | str, optional, by default None DEFAULT: None
estimator_weights	Weights for each estimator in weighted methods. Keys must match estimator names/indices. TYPE: dict[str, float], optional, by default None DEFAULT: None
n_iter	Number of iterations for iterative alignment methods. TYPE: int, by default 5 DEFAULT: 5
fit_manager	Manager for unified training of both Estimator and Parametric models. Required for using Parametric (PyTorch Lightning) models in consensus ensemble. See FitManager for configuration options. TYPE: FitManager | None DEFAULT: None

ATTRIBUTE	DESCRIPTION
estimators	Fitted estimators with names as keys. TYPE: dict[str, ModelType]
consensus_embedding	The consensus embedding from training data. TYPE: Tensor

METHOD	DESCRIPTION
fit	Fit all estimators and compute consensus embedding.
predict	Project new data to consensus embedding space.
score	Compute consensus quality score (average alignment score).

Source code in spectre/ensemble/consensus.py

def __init__(
    self,
    estimators: list[ModelType] | dict[str, ModelType],
    consensus_method: str = "procrustes_mean",
    reference_idx: int | str | None = None,
    estimator_weights: dict[str, float] | None = None,
    n_iter: int = 5,
    fit_manager: FitManager | None = None,
):
    super().__init__(estimators, fit_manager=fit_manager)

    allowed_consensus_methods = [
        "procrustes_mean",
        "procrustes_median",
        "weighted_procrustes",
        "reference_alignment",
    ]
    if consensus_method not in allowed_consensus_methods:
        raise ValueError(
            f"Unknown consensus_method: {consensus_method}. "
            f"Valid options: {allowed_consensus_methods}"
        )
    self.consensus_method = consensus_method

    # Validate reference_idx for reference_alignment
    if consensus_method == "reference_alignment":
        if reference_idx is None:
            raise ValueError(
                "`reference_idx` must be provided for reference_alignment method."
            )

        # Convert int to string key if using list estimators
        if isinstance(reference_idx, int):
            if reference_idx < 0 or reference_idx >= len(self._estimator_names):
                raise ValueError(
                    f"`reference_idx` {reference_idx} out of bounds. "
                    f"Valid range: [0, {len(self._estimator_names) - 1}]."
                )
            reference_idx = self._estimator_names[reference_idx]
        elif not isinstance(reference_idx, str):
            raise TypeError(
                f"`reference_idx` must be int or str, got {type(reference_idx).__name__}."
            )

        # Check if resolved key exists
        if reference_idx not in self._estimators:
            raise ValueError(
                f"`reference_idx` '{reference_idx}' not found in estimators."
            )
    self.reference_idx = reference_idx

    # Validate estimator weights for weighted methods
    if consensus_method == "weighted_procrustes":
        if estimator_weights is None:
            raise ValueError(
                "Estimator weights must be provided for `weighted_procrustes` method."
            )
        self._estimator_weights = self.normalize_estimator_weights(
            estimator_weights
        )
    else:
        self._estimator_weights = estimator_weights

    if not isinstance(n_iter, int) or n_iter < 1:
        raise ValueError(f"n_iter must be positive int, got {n_iter}.")
    self.n_iter = n_iter

    # Internal storage
    self._consensus_embedding = None

Attributes

consensus_embedding: torch.Tensor property

Return consensus embedding.

Functions

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

Fit all estimators and compute consensus embedding.

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

RETURNS	DESCRIPTION
self	Fitted estimator. TYPE: ConsensusEmbedding

Source code in spectre/ensemble/consensus.py

def fit(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
) -> "ConsensusEmbedding":
    """
    Fit all estimators and compute consensus embedding.

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

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

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

    Returns
    -------
    self : ConsensusEmbedding
        Fitted estimator.
    """
    self.in_features = X.shape[1]

    # Fit all estimators and get embeddings
    embeddings = {}
    for name, estimator in self._estimators.items():
        self._fit_estimator(
            estimator, X, weights=weights, target=target, estimator_name=name
        )
        embeddings[name] = estimator.predict(X)

    # Set output features from first embedding
    self.out_features = list(embeddings.values())[0].shape[1]

    # Compute consensus embedding
    if self.consensus_method == "reference_alignment":
        self._consensus_embedding = self._reference_alignment(embeddings)
    elif self.consensus_method == "procrustes_mean":
        self._consensus_embedding = self._iterative_procrustes(embeddings, "mean")
    elif self.consensus_method == "procrustes_median":
        self._consensus_embedding = self._iterative_procrustes(embeddings, "median")
    elif self.consensus_method == "weighted_procrustes":
        self._consensus_embedding = self._weighted_procrustes(embeddings)

    self.is_fitted = True

    return self

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

Project new data to consensus embedding space.

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

RETURNS	DESCRIPTION
X_consensus	Consensus embedding of shape (n_samples, out_features). TYPE: Tensor

Source code in spectre/ensemble/consensus.py

def predict(self, X: torch.Tensor) -> torch.Tensor:
    """
    Project new data to consensus embedding space.

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

    Returns
    -------
    X_consensus : torch.Tensor
        Consensus embedding of shape (n_samples, out_features).
    """
    self.check_fitted()

    # Get embeddings from all estimators
    embeddings = self.predict_ensemble(X)

    # Align to consensus using same method
    if self.consensus_method == "reference_alignment":
        return self._align_to_reference(embeddings)
    elif self.consensus_method == "procrustes_mean":
        return self._align_to_consensus(embeddings, "mean")
    elif self.consensus_method == "procrustes_median":
        return self._align_to_consensus(embeddings, "median")
    elif self.consensus_method == "weighted_procrustes":
        return self._align_to_consensus_weighted(embeddings)

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

Compute consensus quality score (average alignment score).

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

RETURNS	DESCRIPTION
score	Average reconstruction score across estimators. TYPE: float

Source code in spectre/ensemble/consensus.py

def score(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
) -> float:
    """
    Compute consensus quality score (average alignment score).

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

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

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

    Returns
    -------
    score : float
        Average reconstruction score across estimators.
    """
    return self.reduce_scores(
        X,
        weights=weights,
        target=target,
        reduce="mean",
    )