src/features/selector.py

"""Feature selection, implemented so that fitting outside a CV fold is impossible.



THE CONSTRAINT. n is 124 cells. Selecting features on the full dataset is the

fastest way to manufacture a result that looks strong and does not replicate:

the selector sees the labels of the cells it will later be evaluated on, and

"which features correlate with the target" is exactly the information a test

fold is supposed to withhold.



HOW IT IS ENFORCED HERE. Every selector below is an sklearn transformer with a

``fit``/``transform`` split. Composed into a ``Pipeline`` and passed to

``cross_val_score`` or ``GridSearchCV``, ``fit`` is called on the training fold

ONLY -- scikit-learn guarantees that, so the guarantee does not depend on this

project remembering to do it. There is deliberately no module-level

"select_features(X, y)" convenience function, because that is the shape of API

that invites full-data use.



``configs/features.yaml`` sets ``selection.fit_inside_cv_only: true`` and the

schema in ``src/utils/config.py`` REJECTS the config if it is ever false.

"""

from __future__ import annotations

import numpy as np
import pandas as pd
from scipy import stats
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.feature_selection import mutual_info_classif, mutual_info_regression
from sklearn.impute import SimpleImputer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.utils.validation import check_is_fitted

from src.utils.config import ConfigBundle


def _abs_spearman(x: np.ndarray, y: np.ndarray) -> float:
    """|Spearman rho| between two aligned 1-D arrays, 0.0 if undefined.



    Takes arrays rather than Series on purpose: pandas correlates on the shared

    INDEX, which silently mis-pairs a bootstrap subset (original row labels)

    against a freshly built target (RangeIndex).

    """
    if x.size != y.size or x.size < 3:
        return 0.0
    finite = np.isfinite(x) & np.isfinite(y)
    if finite.sum() < 3:
        return 0.0
    x, y = x[finite], y[finite]
    if np.ptp(x) == 0 or np.ptp(y) == 0:
        return 0.0
    rho = stats.spearmanr(x, y).statistic
    return float(abs(rho)) if np.isfinite(rho) else 0.0


def _encode_target(y) -> tuple[np.ndarray, bool]:
    """Return a numeric target and whether it is categorical.



    The selectors serve both routes of the project: regression on log10 cycle

    life, and the secondary direct classification on grade labels. Grades arrive

    as strings, which neither mutual information nor a rank correlation can

    consume.



    Sorted order is used for the encoding, which for the grade labels A/B/C

    yields 0/1/2 -- their true ordinal order, most demanding first. That is

    meaningful rather than arbitrary here, so a rank statistic computed on the

    codes measures something real.

    """
    array = np.asarray(y).ravel()
    if array.dtype.kind in "biufc":
        return array.astype("float64"), False
    categories = np.unique(array)
    lookup = {value: index for index, value in enumerate(categories)}
    return np.array([lookup[v] for v in array], dtype="float64"), True


def _abs_spearman_matrix(X: np.ndarray, y: np.ndarray) -> np.ndarray:
    """|Spearman rho| between every column of ``X`` and ``y``, vectorised.



    Spearman is Pearson on ranks, so ranking each column and then taking a

    single centred, normalised matrix product is EXACTLY equivalent to calling

    a Spearman routine per column -- it is a speed change, not an approximation.

    Ties are handled by ``rankdata``'s average method, matching

    ``scipy.stats.spearmanr``.



    Columns that are constant, or all-NaN, score 0.0 rather than NaN: they carry

    no information, and a NaN would propagate into the ranking.

    """
    n_rows, n_cols = X.shape
    if n_rows < 3:
        return np.zeros(n_cols)

    # NaNs cannot be ranked. Column medians keep the column usable without
    # inventing signal, and a fully-NaN column is zeroed out below anyway.
    if np.isnan(X).any():
        medians = np.nanmedian(np.where(np.isfinite(X), X, np.nan), axis=0)
        medians = np.where(np.isfinite(medians), medians, 0.0)
        X = np.where(np.isfinite(X), X, medians)

    ranks_x = stats.rankdata(X, axis=0)
    ranks_y = stats.rankdata(y)

    xc = ranks_x - ranks_x.mean(axis=0)
    yc = ranks_y - ranks_y.mean()

    x_norm = np.sqrt((xc**2).sum(axis=0))
    y_norm = np.sqrt((yc**2).sum())

    with np.errstate(invalid="ignore", divide="ignore"):
        rho = (xc * yc[:, None]).sum(axis=0) / (x_norm * y_norm)
    return np.nan_to_num(np.abs(rho), nan=0.0, posinf=0.0, neginf=0.0)


class _FrameAwareSelector(BaseEstimator, TransformerMixin):
    """Base class that preserves feature names across transforms."""

    def _record_names(self, X) -> np.ndarray:
        if isinstance(X, pd.DataFrame):
            self.feature_names_in_ = np.asarray(X.columns)
        else:
            self.feature_names_in_ = np.asarray([f"x{i}" for i in range(np.shape(X)[1])])
        return self.feature_names_in_

    def _subset(self, X):
        if isinstance(X, pd.DataFrame):
            return X.loc[:, self.support_]
        return np.asarray(X)[:, self.support_]

    def get_feature_names_out(self, input_features=None) -> np.ndarray:
        check_is_fitted(self, "support_")
        return self.feature_names_in_[self.support_]


class VarianceFilter(_FrameAwareSelector):
    """Drop near-constant and all-missing columns.



    A column with no variance in the TRAINING fold cannot inform a model fitted

    on that fold. This also removes the degenerate columns the builder

    deliberately leaves in place (``policy_c3`` is all-NaN in this corpus and

    ``policy_n_steps`` is constant), which is where that decision belongs.

    """

    def __init__(self, threshold: float = 1e-8):
        self.threshold = threshold

    def fit(self, X, y=None):
        names = self._record_names(X)
        frame = pd.DataFrame(X, columns=names)
        variances = frame.var(axis=0, skipna=True)
        all_missing = frame.isna().all(axis=0)
        self.support_ = (variances > self.threshold).to_numpy() & ~all_missing.to_numpy()
        if not self.support_.any():  # never hand an empty matrix downstream
            self.support_ = np.ones(len(names), dtype=bool)
        return self

    def transform(self, X):
        check_is_fitted(self, "support_")
        return self._subset(X)


class CorrelationFilter(_FrameAwareSelector):
    """Drop one of every pair of features correlated above a threshold.



    WHY it matters here specifically: the curve features are collinear by

    construction -- ``dq_var`` and ``log_abs_dq_var`` are monotone transforms of

    one another, and several ΔQ(V) statistics move together. Severe collinearity

    destabilises linear-model coefficients, which in turn makes the SHAP

    attributions of Phase 9 unstable, and an unstable attribution cannot support

    an auditable scrap decision.



    Of each correlated pair the LATER column is dropped, so the ordering of the

    feature registry determines precedence deterministically rather than by

    whichever the correlation matrix happened to list first.

    """

    def __init__(self, threshold: float = 0.95):
        self.threshold = threshold

    def fit(self, X, y=None):
        names = self._record_names(X)
        frame = pd.DataFrame(X, columns=names)
        # .copy() because pandas returns a read-only view here, which
        # np.fill_diagonal cannot write into.
        corr = frame.corr(method="spearman").abs().to_numpy().copy()
        np.fill_diagonal(corr, 0.0)
        corr = np.nan_to_num(corr, nan=0.0)

        keep = np.ones(len(names), dtype=bool)
        for j in range(len(names)):
            if not keep[j]:
                continue
            for k in range(j + 1, len(names)):
                if keep[k] and corr[j, k] > self.threshold:
                    keep[k] = False
        self.support_ = keep
        return self

    def transform(self, X):
        check_is_fitted(self, "support_")
        return self._subset(X)


class MutualInfoSelector(_FrameAwareSelector):
    """Keep the top-k features by mutual information with the target.



    Mutual information rather than a linear correlation because several

    relationships here are monotone but not linear -- cycle life against

    resistance is a clear case. Requires ``y``, so it can only ever be fitted

    where labels are legitimately available, i.e. the training fold.

    """

    def __init__(self, k: int | None = None, n_neighbors: int = 3, random_state: int = 42):
        self.k = k
        self.n_neighbors = n_neighbors
        self.random_state = random_state

    def fit(self, X, y=None):
        names = self._record_names(X)
        if y is None:
            raise ValueError("MutualInfoSelector requires y; it cannot be fitted unsupervised.")
        frame = pd.DataFrame(X, columns=names)
        # MI cannot consume NaN. Impute with training-fold medians for the
        # ranking only; the returned mask is applied to the untouched data.
        filled = frame.fillna(frame.median(numeric_only=True)).fillna(0.0)

        # The selectors serve both project routes, so the estimator must match
        # the target type: mutual_info_regression cannot consume the string
        # grade labels the secondary classification route supplies.
        target, is_categorical = _encode_target(y)
        estimator = mutual_info_classif if is_categorical else mutual_info_regression
        scores = estimator(
            filled, target.astype("int64") if is_categorical else target,
            n_neighbors=self.n_neighbors, random_state=self.random_state,
        )
        self.scores_ = scores
        k = len(names) if self.k is None else min(self.k, len(names))
        order = np.argsort(scores)[::-1][:k]
        self.support_ = np.zeros(len(names), dtype=bool)
        self.support_[order] = True
        return self

    def transform(self, X):
        check_is_fitted(self, "support_")
        return self._subset(X)


class StabilitySelector(_FrameAwareSelector):
    """Keep features selected in a sufficient fraction of bootstrap resamples.



    WHY, at n = 124: a single fit will happily rank a noise feature highly by

    chance. Requiring a feature to survive repeated resampling of the TRAINING

    fold is a far stronger criterion and is the standard remedy for selection

    instability in the small-n regime. All resampling happens strictly within

    whatever data ``fit`` receives.

    """

    def __init__(

        self,

        n_bootstrap: int = 100,

        sample_fraction: float = 0.75,

        selection_frequency_threshold: float = 0.6,

        k_per_bootstrap: int | None = None,

        random_state: int = 42,

    ):
        self.n_bootstrap = n_bootstrap
        self.sample_fraction = sample_fraction
        self.selection_frequency_threshold = selection_frequency_threshold
        self.k_per_bootstrap = k_per_bootstrap
        self.random_state = random_state

    def fit(self, X, y=None):
        names = self._record_names(X)
        if y is None:
            raise ValueError("StabilitySelector requires y.")
        frame = pd.DataFrame(X, columns=names)
        filled = frame.fillna(frame.median(numeric_only=True)).fillna(0.0)
        # Grade labels are encoded to their ordinal codes so the rank statistic
        # below is meaningful; see _encode_target.
        target, _ = _encode_target(y)

        rng = np.random.default_rng(self.random_state)
        n_samples = len(filled)
        draw = max(5, int(round(self.sample_fraction * n_samples)))
        k = self.k_per_bootstrap or max(1, len(names) // 2)

        counts = np.zeros(len(names), dtype=float)
        for _ in range(self.n_bootstrap):
            idx = rng.choice(n_samples, size=draw, replace=False)
            sub, sub_y = filled.iloc[idx], target[idx]
            if len(np.unique(sub_y)) < 3:
                continue
            # Spearman |rho| as the per-bootstrap ranking statistic: robust to
            # the monotone-but-nonlinear relationships this corpus contains.
            #
            # Computed VECTORISED over all features at once, and on NUMPY
            # arrays. Both details matter:
            #
            # * Vectorised because the loop version issued one scipy call per
            #   feature per bootstrap -- 5,700 calls per fit -- and dominated
            #   the Phase 6 sweep runtime. Spearman is Pearson on ranks, so
            #   ranking each column once per bootstrap and taking a single
            #   matrix product is exactly equivalent, not an approximation.
            # * On arrays because pandas correlates on the INDEX: the bootstrap
            #   subset keeps the original row labels while the target gets a
            #   fresh RangeIndex, so ``Series.corr`` silently mis-pairs them.
            #   That bug scored dq_var at |rho| = 0.02 against a true 0.91,
            #   demoting the strongest predictor in the project to noise.
            scores = _abs_spearman_matrix(sub.to_numpy(dtype="float64"), sub_y)
            counts[np.argsort(scores)[::-1][:k]] += 1.0

        self.selection_frequency_ = counts / max(self.n_bootstrap, 1)
        self.support_ = self.selection_frequency_ >= self.selection_frequency_threshold
        if not self.support_.any():
            # Never return zero features: fall back to the most stable ones.
            top = np.argsort(self.selection_frequency_)[::-1][:max(1, len(names) // 4)]
            self.support_ = np.zeros(len(names), dtype=bool)
            self.support_[top] = True
        return self

    def transform(self, X):
        check_is_fitted(self, "support_")
        return self._subset(X)


def build_selection_pipeline(bundle: ConfigBundle, *, include_scaler: bool = True) -> Pipeline:
    """Compose the configured preprocessing and selection steps into a Pipeline.



    The returned object is intended to be the front of a model pipeline, so that

    imputation, scaling and selection are all fitted on the training fold only.



    Ordering rationale: variance and correlation filters are unsupervised and

    run first to shrink the problem cheaply; the supervised steps (mutual

    information, stability selection) run afterwards on the reduced set.

    Imputation precedes them because the supervised selectors cannot consume

    NaN, and it uses the median, which is robust to the residual outliers the

    plausibility mask does not catch.

    """
    selection = bundle.features.selection
    steps: list[tuple[str, object]] = [
        # keep_empty_features=True is load-bearing, not a default tweak. Without
        # it SimpleImputer silently DROPS all-NaN columns, so the matrix leaving
        # this step has fewer columns than the feature registry declares and
        # every downstream selector's feature names shift out of alignment.
        # Keeping the column (filled with zeros) lets the variance filter remove
        # it explicitly, per fold, which is where that decision belongs.
        ("impute", SimpleImputer(strategy=bundle.features.preprocessing.impute_strategy,
                                 keep_empty_features=True)),
        ("variance", VarianceFilter(threshold=selection.variance_threshold)),
        ("correlation", CorrelationFilter(threshold=selection.correlation_threshold)),
    ]

    if selection.stability_selection.enabled:
        stability = selection.stability_selection
        steps.append((
            "stability",
            StabilitySelector(
                n_bootstrap=stability.n_bootstrap,
                sample_fraction=stability.sample_fraction,
                selection_frequency_threshold=stability.selection_frequency_threshold,
                random_state=bundle.models.seeds.global_,
            ),
        ))

    if selection.mutual_information.enabled:
        steps.append((
            "mutual_info",
            MutualInfoSelector(
                n_neighbors=selection.mutual_information.n_neighbors,
                random_state=bundle.models.seeds.global_,
            ),
        ))

    if include_scaler and bundle.features.preprocessing.scaler == "standard":
        steps.append(("scale", StandardScaler()))

    pipeline = Pipeline(steps)
    # Emit DataFrames between steps so feature names survive the whole chain.
    # Without this, selectors after the first receive a bare ndarray and fall
    # back to positional names, which would make the Phase 9 SHAP attributions
    # impossible to map back to a physical quantity.
    return pipeline.set_output(transform="pandas")