Feature Selection Sequential

sequential

CLASS	DESCRIPTION
SequentialFeatureSelector	Sequential feature selection with optional floating variant.

Classes

SequentialFeatureSelector(estimator: Any, score_fn: Callable[[Any], torch.Tensor], out_features: int = 0, forward: bool = True, floating: bool = False, labels: list[str] | None = None, n_cv: int | Any = 5, n_jobs: int = -1, maximize: bool = True, random_state: int | None = None)

Sequential feature selection with optional floating variant.

Implements forward and backward sequential feature selection algorithms with optional floating search to escape local optima. Uses cross-validation to evaluate feature subsets.

PARAMETER	DESCRIPTION
estimator	Estimator with fit method. TYPE: Any
score_fn	User-defined score function with signature: score_fn( estimator: Any, X_train: Tensor, X_test: Tensor, weights_train: Tensor, weights_test: Tensor, target_train: Tensor, target_test: Tensor ) -> torch.Tensor TYPE: Callable
out_features	Target number of features to select. Must be positive and not exceed total number of features in data. TYPE: int DEFAULT: 0
forward	If True, performs forward selection (starts empty, adds features). If False, performs backward elimination (starts full, removes features). TYPE: bool, by default True DEFAULT: True
floating	If True, enables floating variant that conditionally removes (forward) or adds (backward) features after each step to escape local optima. Note: Progress bar may not accurately reflect progress with floating enabled due to variable iteration counts. TYPE: bool, by default False DEFAULT: False
labels	Optional feature names. If provided, must have one name per feature. TYPE: list[str] | None, optional, by default None DEFAULT: None
n_cv	Cross-validation strategy. If int, number of folds for KFold with shuffle. Otherwise, any sklearn splitter object with a split() method (e.g., StratifiedKFold, GroupKFold, TimeSeriesSplit). TYPE: int | sklearn splitter, by default 5 DEFAULT: 5
n_jobs	Number of parallel jobs for cross-validation. -1 uses all processors. TYPE: int, by default -1 DEFAULT: -1
maximize	If True, higher scores are better. If False, lower scores are better. TYPE: bool, by default True DEFAULT: True
random_state	Random seed for cross-validation shuffle. TYPE: int | None, optional, by default None DEFAULT: None

Properties

features : tuple[int] | None Indices of selected features after fitting.

scores : torch.Tensor | None Cross-validation scores for selected feature subset of shape (n_cv, ).

labels : list[str] | None Names of selected features if labels was provided.

metrics : dict[str, float] | None Dictionary of selected feature subset metrics.

METHOD	DESCRIPTION
check_fitted	Check if the estimator is fitted and raise an error if not.
score	Compute cross-validation score for feature subset.
select	Select best feature to add (forward) or remove (backward).
fit	Fit selector to find optimal feature subset.
transform	Transform data by selecting best features.
fit_transform	Fit selector and transform data in one step.

ATTRIBUTE	DESCRIPTION
features	Indices of selected features after fitting. TYPE: tuple[int, ...]
scores	Cross-validation scores for selected feature subset. TYPE: Tensor
labels	Names of selected features if labels was provided. TYPE: list[str] | None
metrics	Dictionary of selected feature metrics. TYPE: dict[str, float]

Source code in spectre/feature_selection/sequential.py

def __init__(
    self,
    estimator: Any,
    score_fn: Callable[[Any], torch.Tensor],
    out_features: int = 0,
    forward: bool = True,
    floating: bool = False,
    labels: list[str] | None = None,
    n_cv: int | Any = 5,
    n_jobs: int = -1,
    maximize: bool = True,
    random_state: int | None = None,
) -> None:
    if not hasattr(estimator, "fit"):
        raise AttributeError(
            "Estimator must have a `fit()` method for feature selection."
        )
    self.estimator = estimator

    if not callable(score_fn):
        raise TypeError("`score_fn` must be callable.")
    self.score_fn = score_fn

    if not isinstance(out_features, int):
        raise TypeError(
            f"Expected `out_features` to be `int`, got {type(out_features)}."
        )
    check_in_interval(out_features, "[1, inf)")
    self.out_features = out_features

    if not isinstance(forward, bool):
        raise TypeError(f"Expected `forward` to be `bool`, got {type(forward)}.")
    self.forward = forward

    if not isinstance(floating, bool):
        raise TypeError(f"Expected `floating` to be `bool`, got {type(floating)}.")
    self.floating = floating

    if labels is not None:
        if not isinstance(labels, list):
            raise TypeError(f"Expected `labels` to be a list, got {type(labels)}.")
        if not all(isinstance(label, str) for label in labels):
            raise TypeError("Expected all elements in `labels` to be `str`.")
    self._labels = labels

    # Validate and store n_cv
    if isinstance(n_cv, int):
        check_in_interval(n_cv, "[2, inf)")
    elif not hasattr(n_cv, "split"):
        raise TypeError(
            f"Expected `n_cv` to be `int` or sklearn splitter with `split()` method, "
            f"got {type(n_cv)}."
        )
    self.n_cv = n_cv

    if not isinstance(n_jobs, int):
        raise TypeError(f"Expected `n_jobs` to be `int`, got {type(n_jobs)}.")
    self.n_jobs = n_jobs

    if not isinstance(maximize, bool):
        raise TypeError(f"Expected `maximize` to be `bool`, got {type(maximize)}.")
    self.maximize = maximize

    if random_state is not None and not isinstance(random_state, int):
        raise TypeError(
            f"Expected `random_state` to be `int` or None, got {type(random_state)}."
        )
    self.random_state = random_state

    # Attributes.
    self.is_fitted = False

    # Internal state.
    self._metrics: dict = {}
    self._features: tuple[int, ...] | None = None
    self._scores: torch.Tensor | None = None

Attributes

features: tuple[int, ...] property

Indices of selected features after fitting.

RETURNS	DESCRIPTION
tuple[int, ...]	Indices of selected features (out_features, ).

scores: torch.Tensor property

Cross-validation scores for selected feature subset.

RETURNS	DESCRIPTION
Tensor	CV fold scores of shape (n_splits, ) where n_splits is determined by the CV splitter (n_cv parameter).

labels: list[str] | None property

Names of selected features if labels was provided.

RETURNS	DESCRIPTION
list[str] | None	List of selected feature names (out_features, ), or None if labels not provided.

metrics: dict[str, float] property

Dictionary of selected feature metrics.

RETURNS	DESCRIPTION
dict[str, float]	Dictionary of selected feature metrics.

Functions

check_fitted() -> None

Check if the estimator is fitted and raise an error if not.

Source code in spectre/feature_selection/sequential.py

def check_fitted(self) -> None:
    """Check if the estimator is fitted and raise an error if not."""
    if not self.is_fitted:
        raise ValueError("Estimator is not fitted. Call `fit()` first.")

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

Compute cross-validation score for feature subset.

Fits estimator on each train fold and evaluates using score_fn. Uses K-fold cross-validation with internal parallelization disabled (n_jobs=1) to avoid nested parallelism conflicts with feature-level parallelization in select().

PARAMETER	DESCRIPTION
X	Input data with selected features of shape (n_samples, out_features). TYPE: Tensor
weights	Sample weights of shape (n_samples, ). TYPE: torch.Tensor | None, optional, by default None DEFAULT: None
target	Target values for supervised methods. Currently unused but kept for API extensibility. TYPE: torch.Tensor | None, optional, by default None DEFAULT: None

RETURNS	DESCRIPTION
tuple[Tensor, Tensor]	Mean CV score (scalar) and individual fold scores (n_cv, ).

Source code in spectre/feature_selection/sequential.py

def score(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
) -> tuple[torch.Tensor, torch.Tensor]:
    """
    Compute cross-validation score for feature subset.

    Fits estimator on each train fold and evaluates using `score_fn`.
    Uses K-fold cross-validation with internal parallelization disabled
    (`n_jobs=1`) to avoid nested parallelism conflicts with feature-level
    parallelization in `select()`.

    Parameters
    ----------
    X : torch.Tensor
        Input data with selected features of shape (n_samples, out_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 for supervised methods.
        Currently unused but kept for API extensibility.

    Returns
    -------
    tuple[torch.Tensor, torch.Tensor]
        Mean CV score (scalar) and individual fold scores (n_cv, ).
    """
    if isinstance(self.n_cv, int):
        cv = KFold(
            n_splits=self.n_cv,
            shuffle=True,
            random_state=self.random_state,
        )
    else:
        cv = self.n_cv

    scores = []
    for train_idx, test_idx in cv.split(X):
        # Fit estimator on train fold
        X_train = X[train_idx]
        weights_train = weights[train_idx] if weights is not None else None

        estimator = copy.deepcopy(self.estimator)
        estimator.fit(
            X_train,
            weights=weights_train,
            target=None,
        )

        X_test = X[test_idx]
        weights_test = weights[test_idx] if weights is not None else None

        score = self.score_fn(
            estimator,
            X_train,
            X_test,
            weights_train,
            weights_test,
            target_train=None,
            target_test=None,
        )
        scores.append(score)

    scores_tensor = torch.stack(scores)

    return scores_tensor.mean(), scores_tensor

select(*, X: torch.Tensor, weights: torch.Tensor | None = None, target: torch.Tensor | None = None, search_set: set[int], fixed_set: set[int], forward: bool = True) -> tuple[tuple[int, ...], torch.Tensor, torch.Tensor]

Select best feature to add (forward) or remove (backward).

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 for supervised methods of shape (n_samples, ). TYPE: torch.Tensor | None, optional, by default None DEFAULT: None
search_set	Set of feature indices to search from. TYPE: set[int]
fixed_set	Set of feature indices that must be included. TYPE: set[int]
forward	If True, adds one feature from remaining. If False, removes one. TYPE: bool, by default True DEFAULT: True

RETURNS	DESCRIPTION
tuple[tuple[int, ...], Tensor, Tensor]	Best feature subset (indices), its mean score, and CV fold scores.

Source code in spectre/feature_selection/sequential.py

def select(
    self,
    *,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
    search_set: set[int],
    fixed_set: set[int],
    forward: bool = True,
) -> tuple[tuple[int, ...], torch.Tensor, torch.Tensor]:
    """
    Select best feature to add (forward) or remove (backward).

    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 for supervised methods of shape (n_samples, ).

    search_set : set[int]
        Set of feature indices to search from.

    fixed_set : set[int]
        Set of feature indices that must be included.

    forward : bool, by default True
        If True, adds one feature from remaining. If False, removes one.

    Returns
    -------
    tuple[tuple[int, ...], torch.Tensor, torch.Tensor]
        Best feature subset (indices), its mean score, and CV fold scores.
    """
    remaining = list(search_set - fixed_set)
    n_remaining = len(remaining)

    if n_remaining == 0:
        raise ValueError(
            "No features available to add or remove. "
            "This indicates an invalid state in the selection algorithm."
        )

    if forward:
        # Try adding each remaining feature one at a time
        feature_ndx = combinations(remaining, r=1)
    else:
        # Try removing each feature by keeping n_remaining - 1 features
        feature_ndx = combinations(remaining, r=n_remaining - 1)

    fixed_tuple = tuple(sorted(fixed_set))
    subsets = [tuple(sorted(ndx + fixed_tuple)) for ndx in feature_ndx]

    parallel = joblib.Parallel(
        n_jobs=min(self.n_jobs, n_remaining) if self.n_jobs > 0 else self.n_jobs,
        verbose=False,
        pre_dispatch="2*n_jobs",
    )

    # score() returns (mean_score, fold_scores) tuple
    results = parallel(
        joblib.delayed(self.score)(
            X=X[:, ndx],
            weights=weights,
            target=target,
        )
        for ndx in subsets
    )

    mean_scores, fold_scores_list = zip(*results)
    mean_scores_tensor = torch.stack(mean_scores, dim=0)

    if torch.any(torch.isnan(mean_scores_tensor)):
        raise ValueError(
            "NaN scores detected during feature selection. "
            "This may indicate numerical instability in the scoring function."
        )

    if self.maximize:
        best_idx = mean_scores_tensor.argmax().item()
    else:
        best_idx = mean_scores_tensor.argmin().item()

    return (
        subsets[best_idx],
        mean_scores_tensor[best_idx],
        fold_scores_list[best_idx],
    )

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

Fit selector to find optimal feature subset.

Performs sequential feature selection using forward/backward search with optional floating to escape local optima.

PARAMETER	DESCRIPTION
X	Training 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 for supervised methods of shape (n_samples, ). TYPE: torch.Tensor | None, optional, by default None DEFAULT: None

RETURNS	DESCRIPTION
SequentialFeatureSelector	Returns self for method chaining.

Source code in spectre/feature_selection/sequential.py

def fit(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
) -> "SequentialFeatureSelector":
    """
    Fit selector to find optimal feature subset.

    Performs sequential feature selection using forward/backward search
    with optional floating to escape local optima.

    Parameters
    ----------
    X : torch.Tensor
        Training 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 for supervised methods of shape (n_samples, ).

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

    if weights is not None:
        check_1d(weights)
        check_same_len(X, weights)
        check_same_device(X, weights)

    self.in_features = X.shape[1]

    if self.out_features > self.in_features:
        raise ValueError(
            f"`out_features` ({self.out_features}) cannot exceed "
            f"number of features in X ({self.in_features})."
        )

    if self._labels is not None:
        if len(self._labels) != self.in_features:
            raise ValueError(
                f"Length of labels ({len(self._labels)}) must match "
                f"number of features in X ({self.in_features})."
            )

    current_features = self._initialize_features(forward=self.forward)
    n_current = len(current_features)
    if n_current > 0:
        mean_score, cv_scores = self.score(
            X=X[:, current_features],
            weights=weights,
            target=None,
        )
        self._record_best_subset(n_current, current_features, cv_scores, mean_score)

    all_features = set(range(self.in_features))

    # Progress bar shows estimated steps. With floating=True, actual iterations
    # may exceed this estimate due to additional optimization sweeps.
    iteration = 0
    total_steps = abs(n_current - self.out_features)
    pbar = tqdm(
        total=total_steps,
        desc="Feature selection",
        disable=False,
    )
    while n_current != self.out_features:
        prev_features = set(current_features)
        if self.forward:
            search_set = all_features
            fixed_set = prev_features
        else:
            search_set = prev_features
            fixed_set = set()

        current_features, mean_score, cv_scores = self.select(
            X=X,
            weights=weights,
            target=None,
            search_set=search_set,
            fixed_set=fixed_set,
            forward=self.forward,
        )

        n_current = len(current_features)

        # Check if this is the best score for this subset size
        if self._is_best_for_size(mean_score, n_current):
            self._record_best_subset(
                n_current, current_features, cv_scores, mean_score
            )
            pbar.set_postfix({"Score": mean_score.item(), "Dim": n_current})

        if self.floating:
            current_features, mean_score, cv_scores = self._perform_floating_search(
                X=X,
                weights=weights,
                target=None,
                current_features=current_features,
                mean_score=mean_score,
                cv_scores=cv_scores,
                prev_features=prev_features,
                all_features=all_features,
            )
            n_current = len(current_features)

        iteration += 1
        pbar.update(iteration - pbar.n)
    pbar.update(total_steps - pbar.n)
    pbar.close()

    self.is_fitted = True

    best_subset_size = self._find_best_subset()
    selected_set = self._metrics[best_subset_size]["features"]
    self._features = tuple(selected_set)
    self._scores = self._metrics[best_subset_size]["scores"]

    return self

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

Transform data by selecting best features.

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

RETURNS	DESCRIPTION
Tensor	Transformed data with selected features of shape (n_samples, out_features).

RAISES	DESCRIPTION
ValueError	If selector has not been fitted yet.

Source code in spectre/feature_selection/sequential.py

def transform(self, X: torch.Tensor) -> torch.Tensor:
    """
    Transform data by selecting best features.

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

    Returns
    -------
    torch.Tensor
        Transformed data with selected features of shape (n_samples, out_features).

    Raises
    ------
    ValueError
        If selector has not been fitted yet.
    """
    self.check_fitted()
    return X[:, self._features]

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

Fit selector and transform data in one step.

PARAMETER	DESCRIPTION
X	Training 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 for supervised methods of shape (n_samples, ). Not used. TYPE: torch.Tensor | None, optional, by default None DEFAULT: None

RETURNS	DESCRIPTION
Tensor	Transformed data with selected features of shape (n_samples, out_features).

Source code in spectre/feature_selection/sequential.py

def fit_transform(
    self,
    X: torch.Tensor,
    weights: torch.Tensor | None = None,
    target: torch.Tensor | None = None,
) -> torch.Tensor:
    """
    Fit selector and transform data in one step.

    Parameters
    ----------
    X : torch.Tensor
        Training 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 for supervised methods of shape (n_samples, ). Not used.

    Returns
    -------
    torch.Tensor
        Transformed data with selected features of shape (n_samples, out_features).
    """
    self.fit(X, weights=weights, target=None)
    return self.transform(X)

Functions