Consensus API¶

SRF is a non-convex method, so different random initializations can land on slightly different embeddings. The consensus transformers run several random starts and aggregate them into a single, stable embedding. They follow the scikit-learn transformer API (fit/transform), so you can drop them into a Pipeline in place of a single SRF. See examples for how to use them in a workflow.

EnsembleFit ¶

EnsembleFit(
    base_estimator: BaseEstimator,
    n_runs: int = 50,
    random_state: int = 0,
    n_jobs: int = -1,
    threads_per_worker: int | str | None = None,
)

Bases: BaseEstimator, TransformerMixin

Fit the base estimator multiple times and stack the embeddings.

Each run uses a different random seed, producing embeddings that differ only in column permutation. The stacked output has shape (n_samples, rank * n_runs) and can be passed to a consensus method.

Parameters:

Name	Type	Description	Default
`base_estimator`	`BaseEstimator`	Estimator with fit/transform (e.g. SRF)	required
`n_runs`	`int`	Number of independent runs	`50`
`random_state`	`int`	Base random seed (run i uses random_state + i)	`0`
`n_jobs`	`int`	Number of parallel jobs	`-1`

Attributes:

Name	Type	Description
`embeddings_`	`ndarray of shape (n_samples, rank * n_runs)`	Horizontally stacked embeddings from all runs
`estimators_`	`list of BaseEstimator`	Fitted estimators from each run

Source code in pysrf/consensus.py

def __init__(
    self,
    base_estimator: BaseEstimator,
    n_runs: int = 50,
    random_state: int = 0,
    n_jobs: int = -1,
    threads_per_worker: int | str | None = None,
):
    self.base_estimator = base_estimator
    self.n_runs = n_runs
    self.random_state = random_state
    self.n_jobs = n_jobs
    self.threads_per_worker = threads_per_worker

fit ¶

fit(x: ndarray, y: ndarray | None = None) -> EnsembleFit

Fit n_runs independent estimators in parallel.

Parameters:

Name	Type	Description	Default
`x`	`ndarray of shape (n_samples, n_samples)`	Symmetric similarity matrix	required
`y`	`Ignored`		`None`

Returns:

Name	Type	Description
`self`	`EnsembleFit`

Source code in pysrf/consensus.py

def fit(self, x: np.ndarray, y: np.ndarray | None = None) -> EnsembleFit:
    """Fit n_runs independent estimators in parallel.

    Parameters
    ----------
    x : ndarray of shape (n_samples, n_samples)
        Symmetric similarity matrix
    y : Ignored

    Returns
    -------
    self : EnsembleFit
    """

    n_jobs = n_jobs_for_tasks(self.n_jobs, self.n_runs)
    limits = blas_limits_for_workers(self.threads_per_worker, n_jobs)

    def _fit_one(seed):
        # Limit BLAS threads per worker so parallel runs do not
        # oversubscribe the machine
        limit = threadpool_limits(limits=limits) if limits else nullcontext()
        with limit:
            est = clone(self.base_estimator)
            est.set_params(random_state=seed)
            return est, est.fit_transform(x)

    if n_jobs == 1:
        results = [_fit_one(self.random_state + i) for i in range(self.n_runs)]
    else:
        results = Parallel(n_jobs=n_jobs, prefer="processes")(
            delayed(_fit_one)(self.random_state + i) for i in range(self.n_runs)
        )

    estimators, embeddings = zip(*results)
    self.estimators_ = list(estimators)
    self.embeddings_ = np.hstack(embeddings)
    return self

transform ¶

transform(
    x: ndarray, y: ndarray | None = None
) -> np.ndarray

Return the stacked embeddings (x is ignored).

Parameters:

Name	Type	Description	Default
`x`	`Ignored`		required
`y`	`Ignored`		`None`

Returns:

Name	Type	Description
`embeddings`	`ndarray of shape (n_samples, rank * n_runs)`

Source code in pysrf/consensus.py

def transform(self, x: np.ndarray, y: np.ndarray | None = None) -> np.ndarray:
    """Return the stacked embeddings (x is ignored).

    Parameters
    ----------
    x : Ignored
    y : Ignored

    Returns
    -------
    embeddings : ndarray of shape (n_samples, rank * n_runs)
    """
    check_is_fitted(self, "embeddings_")
    return self.embeddings_

AlignedConsensus ¶

AlignedConsensus(rank: int)

Bases: BaseEstimator, TransformerMixin

Consensus via Hungarian alignment for symmetric NMF.

Symmetric NMF produces the same dimensions across runs but in different order (permutation ambiguity, no rotation). This class aligns dimensions via the Hungarian algorithm and returns the most central run, the one closest to the element-wise median across all aligned runs.

Parameters:

Name	Type	Description	Default
`rank`	`int`	Number of dimensions in each embedding	required

Attributes:

Name	Type	Description
`aligned_embeddings_`	`ndarray of shape (n_runs, n_samples, rank)`	Embeddings after Hungarian alignment
`consensus_median_`	`ndarray of shape (n_samples, rank)`	Element-wise median across aligned runs
`centrality_scores_`	`ndarray of shape (n_runs,)`	Frobenius distance of each run to the consensus median
`agreement_scores_`	`ndarray of shape (n_runs,)`	Mean pairwise cosine similarity per run (higher = more stable)
`selected_run_idx_`	`int`	Index of the most central run

Examples:

>>> from sklearn.pipeline import Pipeline
>>> from pysrf import SRF
>>> from pysrf.consensus import EnsembleFit, AlignedConsensus
>>>
>>> pipeline = Pipeline([
...     ('ensemble', EnsembleFit(SRF(rank=10), n_runs=50)),
...     ('consensus', AlignedConsensus(rank=10))
... ])
>>> embedding = pipeline.fit_transform(similarity_matrix)

Source code in pysrf/consensus.py

def __init__(self, rank: int):
    self.rank = rank

fit ¶

fit(
    x: ndarray, y: ndarray | None = None
) -> AlignedConsensus

Align embeddings and select the most central run.

Parameters:

Name	Type	Description	Default
`x`	`ndarray of shape (n_samples, n_runs * rank)`	Stacked embeddings from EnsembleFit	required
`y`	`Ignored`		`None`

Returns:

Name	Type	Description
`self`	`AlignedConsensus`

Source code in pysrf/consensus.py

def fit(self, x: np.ndarray, y: np.ndarray | None = None) -> AlignedConsensus:
    """Align embeddings and select the most central run.

    Parameters
    ----------
    x : ndarray of shape (n_samples, n_runs * rank)
        Stacked embeddings from EnsembleFit
    y : Ignored

    Returns
    -------
    self : AlignedConsensus
    """
    x = np.asarray(x)
    n_samples, n_total = x.shape
    n_runs = n_total // self.rank

    if n_total % self.rank != 0:
        raise ValueError(
            f"Number of columns ({n_total}) not divisible by rank ({self.rank})"
        )

    # (n_samples, n_runs, rank) -> (n_runs, n_samples, rank)
    embeddings = x.reshape(n_samples, n_runs, self.rank).transpose(1, 0, 2)

    self.aligned_embeddings_ = align_embeddings(embeddings, reference_idx=0)
    self.consensus_median_ = np.median(self.aligned_embeddings_, axis=0)

    diffs = self.aligned_embeddings_ - self.consensus_median_[np.newaxis]
    self.centrality_scores_ = np.sqrt(np.einsum("ijk,ijk->i", diffs, diffs))

    self.agreement_scores_ = pairwise_agreement(self.aligned_embeddings_)
    self.selected_run_idx_ = int(np.argmin(self.centrality_scores_))

    return self

transform ¶

transform(
    x: ndarray, y: ndarray | None = None
) -> np.ndarray

Return the most central run.

Parameters:

Name	Type	Description	Default
`x`	`Ignored`		required
`y`	`Ignored`		`None`

Returns:

Name	Type	Description
`w`	`ndarray of shape (n_samples, rank)`

Source code in pysrf/consensus.py

def transform(self, x: np.ndarray, y: np.ndarray | None = None) -> np.ndarray:
    """Return the most central run.

    Parameters
    ----------
    x : Ignored
    y : Ignored

    Returns
    -------
    w : ndarray of shape (n_samples, rank)
    """
    check_is_fitted(self, "aligned_embeddings_")
    return self.aligned_embeddings_[self.selected_run_idx_]