Skip to content

Source variance filter

neureptrace.decoding.source_variance_filter implements source-only variance-based feature filtering.

The protocol is Category 1 / strict source-only. Feature variances and the selected feature mask are estimated from source rows only. Evaluation rows are transformed with the fitted mask but are not used to fit it.

Typical usage:

from neureptrace.decoding.source_variance_filter import fit_source_variance_filter

result = fit_source_variance_filter(
    source_features=X_source,
    test_features=X_target,
    config={"variance_threshold": 0.0, "top_k": 128},
)

X_source_filtered = result.train_features
X_target_filtered = result.test_features

neureptrace.decoding.source_variance_filter

Source-only variance feature filtering.

This module fits a feature-selection mask from source rows only and applies that mask to source and evaluation feature matrices. It is a strict Protocol-1 preprocessing helper for removing constant or low-variance features without using any evaluation-domain statistics.

SourceVarianceFilterConfig dataclass

Configuration for source-only variance feature filtering.

Source code in src/neureptrace/decoding/source_variance_filter.py 
22
23
24
25
26
27
28
29
30
31
32
33
34
35
@dataclass(frozen=True, slots=True)
class SourceVarianceFilterConfig:
    """Configuration for source-only variance feature filtering."""

    variance_threshold: float = DEFAULT_VARIANCE_THRESHOLD
    top_k: int | None = None
    ddof: int = 1

    def __post_init__(self) -> None:
        """Normalize and validate direct dataclass construction."""

        object.__setattr__(self, "variance_threshold", _nonnegative_float(self.variance_threshold, name="variance_threshold"))
        object.__setattr__(self, "top_k", _optional_positive_int(self.top_k, name="top_k"))
        object.__setattr__(self, "ddof", _nonnegative_int(self.ddof, name="ddof"))

__post_init__()

Normalize and validate direct dataclass construction.

Source code in src/neureptrace/decoding/source_variance_filter.py 
30
31
32
33
34
35
def __post_init__(self) -> None:
    """Normalize and validate direct dataclass construction."""

    object.__setattr__(self, "variance_threshold", _nonnegative_float(self.variance_threshold, name="variance_threshold"))
    object.__setattr__(self, "top_k", _optional_positive_int(self.top_k, name="top_k"))
    object.__setattr__(self, "ddof", _nonnegative_int(self.ddof, name="ddof"))

SourceVarianceFilterResult dataclass

Filtered feature matrices and fitted source-only mask.

Source code in src/neureptrace/decoding/source_variance_filter.py 
38
39
40
41
42
43
44
45
46
@dataclass(frozen=True, slots=True)
class SourceVarianceFilterResult:
    """Filtered feature matrices and fitted source-only mask."""

    train_features: np.ndarray
    test_features: np.ndarray
    selected_indices: np.ndarray
    variances: np.ndarray
    metadata: dict[str, Any] = field(default_factory=dict)

fit_source_variance_filter(*, source_features, test_features, config=None)

Fit a variance-based feature mask from source rows and transform matrices.

Source code in src/neureptrace/decoding/source_variance_filter.py 
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
def fit_source_variance_filter(
    *,
    source_features: Sequence[Sequence[float]] | np.ndarray,
    test_features: Sequence[Sequence[float]] | np.ndarray,
    config: SourceVarianceFilterConfig | Mapping[str, Any] | None = None,
) -> SourceVarianceFilterResult:
    """Fit a variance-based feature mask from source rows and transform matrices."""

    cfg = source_variance_filter_config() if config is None else _coerce_config(config)
    source = _feature_matrix(source_features, name="source_features")
    test = _feature_matrix(test_features, name="test_features")
    if source.shape[1] != test.shape[1]:
        raise ValueError(f"source_features and test_features must have the same feature width: {source.shape[1]} != {test.shape[1]}.")
    variances = source_feature_variances(source, ddof=cfg.ddof)
    selected = select_variance_features(variances, variance_threshold=cfg.variance_threshold, top_k=cfg.top_k)
    metadata = {
        "source_variance_filter": True,
        "source_variance_filter_protocol": SOURCE_VARIANCE_FILTER_PROTOCOL,
        "source_variance_filter_protocol_category": SOURCE_VARIANCE_FILTER_CATEGORY,
        "source_variance_filter_uses_source_features": True,
        "source_variance_filter_uses_source_labels": False,
        "source_variance_filter_uses_test_features_for_fitting": False,
        "source_variance_filter_uses_test_labels": False,
        "source_variance_filter_valid_for_strict_source_only": True,
        "source_variance_filter_valid_for_benchmark": True,
        "source_variance_filter_n_source_rows": int(source.shape[0]),
        "source_variance_filter_n_test_rows": int(test.shape[0]),
        "source_variance_filter_input_dim": int(source.shape[1]),
        "source_variance_filter_output_dim": int(selected.shape[0]),
        "source_variance_filter_variance_threshold": float(cfg.variance_threshold),
        "source_variance_filter_top_k": "" if cfg.top_k is None else int(cfg.top_k),
        "source_variance_filter_ddof": int(cfg.ddof),
        "source_variance_filter_selected_indices": "|".join(str(int(index)) for index in selected.tolist()),
    }
    return SourceVarianceFilterResult(
        train_features=source[:, selected].astype(np.float32, copy=False),
        test_features=test[:, selected].astype(np.float32, copy=False),
        selected_indices=selected.astype(int, copy=False),
        variances=variances.astype(np.float32, copy=False),
        metadata=metadata,
    )

source_variance_filter_config(*, variance_threshold=DEFAULT_VARIANCE_THRESHOLD, top_k=None, ddof=1)

Normalize public variance-filter options.

Source code in src/neureptrace/decoding/source_variance_filter.py 
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
def source_variance_filter_config(
    *,
    variance_threshold: float | str = DEFAULT_VARIANCE_THRESHOLD,
    top_k: int | str | None = None,
    ddof: int | str = 1,
) -> SourceVarianceFilterConfig:
    """Normalize public variance-filter options."""

    return SourceVarianceFilterConfig(
        variance_threshold=_nonnegative_float(variance_threshold, name="variance_threshold"),
        top_k=_optional_positive_int(top_k, name="top_k"),
        ddof=_nonnegative_int(ddof, name="ddof"),
    )

source_feature_variances(source_features, *, ddof=1)

Return source-only feature variances.

Source code in src/neureptrace/decoding/source_variance_filter.py 
110
111
112
113
114
115
116
117
def source_feature_variances(source_features: Sequence[Sequence[float]] | np.ndarray, *, ddof: int = 1) -> np.ndarray:
    """Return source-only feature variances."""

    source = _feature_matrix(source_features, name="source_features")
    resolved_ddof = _nonnegative_int(ddof, name="ddof")
    if source.shape[0] <= resolved_ddof:
        resolved_ddof = 0
    return np.var(source, axis=0, ddof=resolved_ddof).astype(float, copy=False)

select_variance_features(variances, *, variance_threshold=DEFAULT_VARIANCE_THRESHOLD, top_k=None)

Return selected feature indices sorted in original feature order.

Source code in src/neureptrace/decoding/source_variance_filter.py 
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
def select_variance_features(
    variances: Sequence[float] | np.ndarray,
    *,
    variance_threshold: float = DEFAULT_VARIANCE_THRESHOLD,
    top_k: int | None = None,
) -> np.ndarray:
    """Return selected feature indices sorted in original feature order."""

    values = np.asarray(variances, dtype=float)
    if values.ndim != 1 or values.size < 1 or not np.all(np.isfinite(values)) or np.any(values < 0.0):
        raise ValueError("variances must be a non-empty one-dimensional finite non-negative vector.")
    threshold = _nonnegative_float(variance_threshold, name="variance_threshold")
    selected = np.flatnonzero(values > threshold)
    if top_k is not None:
        k = min(_positive_int(top_k, name="top_k"), values.size)
        ranked = np.argsort(values, kind="mergesort")[-k:]
        selected = np.intersect1d(selected, ranked, assume_unique=False)
    if selected.size == 0:
        selected = np.asarray([int(np.argmax(values))], dtype=int)
    return np.sort(selected).astype(int, copy=False)