Skip to content

Source correlation filter

neureptrace.decoding.source_correlation_filter implements source-only correlation-based feature filtering.

The protocol is Category 1 / strict source-only. Feature correlations, feature importances, 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_correlation_filter import fit_source_correlation_filter

result = fit_source_correlation_filter(
    source_features=X_source,
    test_features=X_target,
    config={"max_abs_correlation": 0.98, "max_features": 256},
)

X_source_filtered = result.train_features
X_target_filtered = result.test_features

neureptrace.decoding.source_correlation_filter

Source-only correlation feature filtering.

This module fits a feature-redundancy mask from source rows only. Features are ranked by a source-only importance vector, then greedily kept when their absolute source correlation with previously kept features does not exceed the configured threshold. Evaluation rows are transformed with the fitted mask but never used to fit it.

SourceCorrelationFilterConfig dataclass

Configuration for source-only correlation feature filtering.

Source code in src/neureptrace/decoding/source_correlation_filter.py 
24
25
26
27
28
29
30
31
@dataclass(frozen=True, slots=True)
class SourceCorrelationFilterConfig:
    """Configuration for source-only correlation feature filtering."""

    max_abs_correlation: float = DEFAULT_MAX_ABS_CORRELATION
    max_features: int | None = None
    min_features: int = 1
    epsilon: float = DEFAULT_EPSILON

SourceCorrelationFilterResult dataclass

Filtered feature matrices and fitted source-only mask.

Source code in src/neureptrace/decoding/source_correlation_filter.py 
34
35
36
37
38
39
40
41
42
43
@dataclass(frozen=True, slots=True)
class SourceCorrelationFilterResult:
    """Filtered feature matrices and fitted source-only mask."""

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

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

Fit a source-only correlation mask and transform feature matrices.

Source code in src/neureptrace/decoding/source_correlation_filter.py 
48
49
50
51
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
93
94
95
96
def fit_source_correlation_filter(
    *,
    source_features: Sequence[Sequence[float]] | np.ndarray,
    test_features: Sequence[Sequence[float]] | np.ndarray,
    config: SourceCorrelationFilterConfig | Mapping[str, Any] | None = None,
) -> SourceCorrelationFilterResult:
    """Fit a source-only correlation mask and transform feature matrices."""

    cfg = source_correlation_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]}.")
    correlation = source_feature_correlation(source, epsilon=cfg.epsilon)
    importance = source_feature_importance(source, epsilon=cfg.epsilon)
    selected = select_uncorrelated_features(
        correlation,
        importance=importance,
        max_abs_correlation=cfg.max_abs_correlation,
        max_features=cfg.max_features,
        min_features=cfg.min_features,
    )
    metadata = {
        "source_correlation_filter": True,
        "source_correlation_filter_protocol": SOURCE_CORRELATION_FILTER_PROTOCOL,
        "source_correlation_filter_protocol_category": SOURCE_CORRELATION_FILTER_CATEGORY,
        "source_correlation_filter_uses_source_features": True,
        "source_correlation_filter_uses_source_labels": False,
        "source_correlation_filter_uses_test_features_for_fitting": False,
        "source_correlation_filter_uses_test_labels": False,
        "source_correlation_filter_valid_for_strict_source_only": True,
        "source_correlation_filter_valid_for_benchmark": True,
        "source_correlation_filter_n_source_rows": int(source.shape[0]),
        "source_correlation_filter_n_test_rows": int(test.shape[0]),
        "source_correlation_filter_input_dim": int(source.shape[1]),
        "source_correlation_filter_output_dim": int(selected.shape[0]),
        "source_correlation_filter_max_abs_correlation": float(cfg.max_abs_correlation),
        "source_correlation_filter_max_features": "" if cfg.max_features is None else int(cfg.max_features),
        "source_correlation_filter_min_features": int(cfg.min_features),
        "source_correlation_filter_selected_indices": "|".join(str(int(index)) for index in selected.tolist()),
    }
    return SourceCorrelationFilterResult(
        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),
        correlation=correlation.astype(np.float32, copy=False),
        importance=importance.astype(np.float32, copy=False),
        metadata=metadata,
    )

source_correlation_filter_config(*, max_abs_correlation=DEFAULT_MAX_ABS_CORRELATION, max_features=None, min_features=1, epsilon=DEFAULT_EPSILON)

Normalize public source-correlation-filter options.

Source code in src/neureptrace/decoding/source_correlation_filter.py 
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
def source_correlation_filter_config(
    *,
    max_abs_correlation: float | str = DEFAULT_MAX_ABS_CORRELATION,
    max_features: int | str | None = None,
    min_features: int | str = 1,
    epsilon: float | str = DEFAULT_EPSILON,
) -> SourceCorrelationFilterConfig:
    """Normalize public source-correlation-filter options."""

    return SourceCorrelationFilterConfig(
        max_abs_correlation=_unit_interval_float(max_abs_correlation, name="max_abs_correlation"),
        max_features=_optional_positive_int(max_features, name="max_features"),
        min_features=_positive_int(min_features, name="min_features"),
        epsilon=_positive_float(epsilon, name="epsilon"),
    )

source_feature_correlation(source_features, *, epsilon=DEFAULT_EPSILON)

Return the source-only feature correlation matrix.

Source code in src/neureptrace/decoding/source_correlation_filter.py 
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
def source_feature_correlation(source_features: Sequence[Sequence[float]] | np.ndarray, *, epsilon: float = DEFAULT_EPSILON) -> np.ndarray:
    """Return the source-only feature correlation matrix."""

    source = _feature_matrix(source_features, name="source_features")
    eps = _positive_float(epsilon, name="epsilon")
    centered = source - np.mean(source, axis=0, keepdims=True)
    norms = np.linalg.norm(centered, axis=0)
    safe_norms = np.maximum(norms, eps)
    normalized = centered / safe_norms[None, :]
    correlation = normalized.T @ normalized
    zero_var = norms <= eps
    if np.any(zero_var):
        correlation[zero_var, :] = 0.0
        correlation[:, zero_var] = 0.0
        correlation[zero_var, zero_var] = 1.0
    correlation = np.clip(correlation, -1.0, 1.0)
    np.fill_diagonal(correlation, 1.0)
    return correlation.astype(float, copy=False)

source_feature_importance(source_features, *, epsilon=DEFAULT_EPSILON)

Return source-only feature importance based on variance.

Source code in src/neureptrace/decoding/source_correlation_filter.py 
136
137
138
139
140
141
def source_feature_importance(source_features: Sequence[Sequence[float]] | np.ndarray, *, epsilon: float = DEFAULT_EPSILON) -> np.ndarray:
    """Return source-only feature importance based on variance."""

    source = _feature_matrix(source_features, name="source_features")
    variances = np.var(source, axis=0, ddof=1 if source.shape[0] > 1 else 0)
    return np.maximum(variances, _positive_float(epsilon, name="epsilon")).astype(float, copy=False)

select_uncorrelated_features(correlation, *, importance=None, max_abs_correlation=DEFAULT_MAX_ABS_CORRELATION, max_features=None, min_features=1)

Greedily select low-redundancy feature indices.

Source code in src/neureptrace/decoding/source_correlation_filter.py 
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
def select_uncorrelated_features(
    correlation: Sequence[Sequence[float]] | np.ndarray,
    *,
    importance: Sequence[float] | np.ndarray | None = None,
    max_abs_correlation: float = DEFAULT_MAX_ABS_CORRELATION,
    max_features: int | None = None,
    min_features: int = 1,
) -> np.ndarray:
    """Greedily select low-redundancy feature indices."""

    corr = np.asarray(correlation, dtype=float)
    if corr.ndim != 2 or corr.shape[0] != corr.shape[1] or corr.shape[0] < 1:
        raise ValueError("correlation must be a non-empty square matrix.")
    if not np.all(np.isfinite(corr)):
        raise ValueError("correlation must contain only finite values.")
    threshold = _unit_interval_float(max_abs_correlation, name="max_abs_correlation")
    min_keep = _positive_int(min_features, name="min_features")
    if min_keep > corr.shape[0]:
        raise ValueError("min_features cannot exceed the number of features.")
    max_keep = corr.shape[0] if max_features is None else min(_positive_int(max_features, name="max_features"), corr.shape[0])
    if max_keep < min_keep:
        raise ValueError("max_features must be greater than or equal to min_features.")
    scores = np.ones(corr.shape[0], dtype=float) if importance is None else np.asarray(importance, dtype=float).reshape(-1)
    if scores.shape[0] != corr.shape[0] or not np.all(np.isfinite(scores)):
        raise ValueError("importance must contain one finite score per feature.")
    order = np.lexsort((np.arange(scores.shape[0]), -scores))
    selected: list[int] = []
    for candidate in order.tolist():
        if len(selected) >= max_keep:
            break
        if not selected or np.all(np.abs(corr[candidate, selected]) <= threshold):
            selected.append(int(candidate))
    if len(selected) < min_keep:
        for candidate in order.tolist():
            if candidate not in selected:
                selected.append(int(candidate))
                if len(selected) >= min_keep:
                    break
    return np.sort(np.asarray(selected, dtype=int))