Skip to content

Source robust scaler

neureptrace.decoding.source_robust_scaler implements source-only robust feature scaling.

The protocol is Category 1 / strict source-only. Location and scale statistics are estimated from source rows only. Evaluation rows are transformed with the fitted statistics but are not used to fit them.

Typical usage:

from neureptrace.decoding.source_robust_scaler import fit_source_robust_scaler

result = fit_source_robust_scaler(
    source_features=X_source,
    test_features=X_target,
    config={"center": "median", "scale": "iqr"},
)

X_source_scaled = result.train_features
X_target_scaled = result.test_features

Supported center modes: median, mean, none.

Supported scale modes: iqr, mad, std, none.

neureptrace.decoding.source_robust_scaler

Source-only robust feature scaling for cross-subject decoding.

This module fits feature-wise location and scale statistics from source rows only and applies the fitted transform to source and evaluation feature matrices. The implementation is a strict Protocol-1 preprocessing helper: evaluation rows are transformed but never used to fit the scaler.

SourceRobustScalerConfig dataclass

Configuration for source-only robust feature scaling.

Source code in src/neureptrace/decoding/source_robust_scaler.py 
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
@dataclass(frozen=True, slots=True)
class SourceRobustScalerConfig:
    """Configuration for source-only robust feature scaling."""

    center: str = "median"
    scale: str = "iqr"
    lower_quantile: float = DEFAULT_LOWER_QUANTILE
    upper_quantile: float = DEFAULT_UPPER_QUANTILE
    epsilon: float = DEFAULT_EPSILON

    def __post_init__(self) -> None:
        """Validate direct dataclass construction as strictly as the public helper."""

        lower = _unit_interval_float(self.lower_quantile, name="lower_quantile")
        upper = _unit_interval_float(self.upper_quantile, name="upper_quantile")
        if lower >= upper:
            raise ValueError("lower_quantile must be smaller than upper_quantile.")
        object.__setattr__(self, "center", normalize_center_mode(self.center))
        object.__setattr__(self, "scale", normalize_scale_mode(self.scale))
        object.__setattr__(self, "lower_quantile", lower)
        object.__setattr__(self, "upper_quantile", upper)
        object.__setattr__(self, "epsilon", _positive_float(self.epsilon, name="epsilon"))

__post_init__()

Validate direct dataclass construction as strictly as the public helper.

Source code in src/neureptrace/decoding/source_robust_scaler.py 
36
37
38
39
40
41
42
43
44
45
46
47
def __post_init__(self) -> None:
    """Validate direct dataclass construction as strictly as the public helper."""

    lower = _unit_interval_float(self.lower_quantile, name="lower_quantile")
    upper = _unit_interval_float(self.upper_quantile, name="upper_quantile")
    if lower >= upper:
        raise ValueError("lower_quantile must be smaller than upper_quantile.")
    object.__setattr__(self, "center", normalize_center_mode(self.center))
    object.__setattr__(self, "scale", normalize_scale_mode(self.scale))
    object.__setattr__(self, "lower_quantile", lower)
    object.__setattr__(self, "upper_quantile", upper)
    object.__setattr__(self, "epsilon", _positive_float(self.epsilon, name="epsilon"))

SourceRobustScalerStats dataclass

Fitted source-only scaler statistics.

Source code in src/neureptrace/decoding/source_robust_scaler.py 
50
51
52
53
54
55
56
@dataclass(frozen=True, slots=True)
class SourceRobustScalerStats:
    """Fitted source-only scaler statistics."""

    location: np.ndarray
    scale: np.ndarray
    n_rows: int

SourceRobustScalerResult dataclass

Scaled feature matrices and fitted source-only statistics.

Source code in src/neureptrace/decoding/source_robust_scaler.py 
59
60
61
62
63
64
65
66
@dataclass(frozen=True, slots=True)
class SourceRobustScalerResult:
    """Scaled feature matrices and fitted source-only statistics."""

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

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

Fit robust scaler statistics on source rows and transform matrices.

Source code in src/neureptrace/decoding/source_robust_scaler.py 
 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
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
def fit_source_robust_scaler(
    *,
    source_features: Sequence[Sequence[float]] | np.ndarray,
    test_features: Sequence[Sequence[float]] | np.ndarray,
    config: SourceRobustScalerConfig | Mapping[str, Any] | None = None,
) -> SourceRobustScalerResult:
    """Fit robust scaler statistics on source rows and transform matrices."""

    cfg = source_robust_scaler_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]}.")
    stats = fit_source_robust_scaler_stats(source, config=cfg)
    train = apply_source_robust_scaler(source, stats=stats)
    test_scaled = apply_source_robust_scaler(test, stats=stats)
    metadata = {
        "source_robust_scaler": True,
        "source_robust_scaler_protocol": SOURCE_ROBUST_SCALER_PROTOCOL,
        "source_robust_scaler_protocol_category": SOURCE_ROBUST_SCALER_CATEGORY,
        "source_robust_scaler_uses_source_features": True,
        "source_robust_scaler_uses_source_labels": False,
        "source_robust_scaler_uses_test_features_for_fitting": False,
        "source_robust_scaler_uses_test_labels": False,
        "source_robust_scaler_valid_for_strict_source_only": True,
        "source_robust_scaler_valid_for_benchmark": True,
        "source_robust_scaler_n_source_rows": int(source.shape[0]),
        "source_robust_scaler_n_test_rows": int(test.shape[0]),
        "source_robust_scaler_feature_dim": int(source.shape[1]),
        "source_robust_scaler_center": cfg.center,
        "source_robust_scaler_scale": cfg.scale,
        "source_robust_scaler_lower_quantile": float(cfg.lower_quantile),
        "source_robust_scaler_upper_quantile": float(cfg.upper_quantile),
        "source_robust_scaler_epsilon": float(cfg.epsilon),
    }
    return SourceRobustScalerResult(
        train_features=train.astype(np.float32, copy=False),
        test_features=test_scaled.astype(np.float32, copy=False),
        stats=stats,
        metadata=metadata,
    )

source_robust_scaler_config(*, center='median', scale='iqr', lower_quantile=DEFAULT_LOWER_QUANTILE, upper_quantile=DEFAULT_UPPER_QUANTILE, epsilon=DEFAULT_EPSILON)

Normalize public robust-scaler options.

Source code in src/neureptrace/decoding/source_robust_scaler.py 
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
def source_robust_scaler_config(
    *,
    center: str | None = "median",
    scale: str | None = "iqr",
    lower_quantile: float | str = DEFAULT_LOWER_QUANTILE,
    upper_quantile: float | str = DEFAULT_UPPER_QUANTILE,
    epsilon: float | str = DEFAULT_EPSILON,
) -> SourceRobustScalerConfig:
    """Normalize public robust-scaler options."""

    lower = _unit_interval_float(lower_quantile, name="lower_quantile")
    upper = _unit_interval_float(upper_quantile, name="upper_quantile")
    if lower >= upper:
        raise ValueError("lower_quantile must be smaller than upper_quantile.")
    return SourceRobustScalerConfig(
        center=normalize_center_mode(center),
        scale=normalize_scale_mode(scale),
        lower_quantile=lower,
        upper_quantile=upper,
        epsilon=_positive_float(epsilon, name="epsilon"),
    )

fit_source_robust_scaler_stats(source_features, *, config=None)

Fit source-only location and scale vectors.

Source code in src/neureptrace/decoding/source_robust_scaler.py 
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
def fit_source_robust_scaler_stats(
    source_features: Sequence[Sequence[float]] | np.ndarray,
    *,
    config: SourceRobustScalerConfig | Mapping[str, Any] | None = None,
) -> SourceRobustScalerStats:
    """Fit source-only location and scale vectors."""

    cfg = source_robust_scaler_config() if config is None else _coerce_config(config)
    source = _feature_matrix(source_features, name="source_features")
    if cfg.center == "median":
        location = np.median(source, axis=0)
    elif cfg.center == "mean":
        location = np.mean(source, axis=0)
    elif cfg.center == "none":
        location = np.zeros(source.shape[1], dtype=float)
    else:  # pragma: no cover - guarded by normalization
        raise ValueError(f"Unhandled center mode {cfg.center!r}.")

    if cfg.scale == "iqr":
        upper = np.quantile(source, cfg.upper_quantile, axis=0)
        lower = np.quantile(source, cfg.lower_quantile, axis=0)
        scale = upper - lower
    elif cfg.scale == "mad":
        med = np.median(source, axis=0)
        scale = 1.4826 * np.median(np.abs(source - med), axis=0)
    elif cfg.scale == "std":
        scale = np.std(source - np.mean(source, axis=0, keepdims=True), axis=0, ddof=1 if source.shape[0] > 1 else 0)
    elif cfg.scale == "none":
        scale = np.ones(source.shape[1], dtype=float)
    else:  # pragma: no cover - guarded by normalization
        raise ValueError(f"Unhandled scale mode {cfg.scale!r}.")
    scale = np.maximum(np.asarray(scale, dtype=float), cfg.epsilon)
    return SourceRobustScalerStats(location=np.asarray(location, dtype=float), scale=scale, n_rows=int(source.shape[0]))

apply_source_robust_scaler(features, *, stats)

Apply fitted source-only robust scaler statistics.

Source code in src/neureptrace/decoding/source_robust_scaler.py 
170
171
172
173
174
175
176
def apply_source_robust_scaler(features: Sequence[Sequence[float]] | np.ndarray, *, stats: SourceRobustScalerStats) -> np.ndarray:
    """Apply fitted source-only robust scaler statistics."""

    matrix = _feature_matrix(features, name="features")
    if matrix.shape[1] != stats.location.shape[0] or matrix.shape[1] != stats.scale.shape[0]:
        raise ValueError("features width must match fitted scaler statistics.")
    return (matrix - stats.location) / stats.scale

normalize_center_mode(value)

Normalize center-mode aliases.

Source code in src/neureptrace/decoding/source_robust_scaler.py 
179
180
181
182
183
184
185
186
def normalize_center_mode(value: str | None) -> str:
    """Normalize center-mode aliases."""

    normalized = "median" if value is None else str(value).strip().lower().replace("-", "_")
    normalized = {"med": "median", "average": "mean", "avg": "mean", "off": "none", "false": "none", "no": "none"}.get(normalized, normalized)
    if normalized not in CENTER_MODES:
        raise ValueError(f"Unknown center mode {value!r}. Available modes: {', '.join(CENTER_MODES)}.")
    return normalized

normalize_scale_mode(value)

Normalize scale-mode aliases.

Source code in src/neureptrace/decoding/source_robust_scaler.py 
189
190
191
192
193
194
195
196
def normalize_scale_mode(value: str | None) -> str:
    """Normalize scale-mode aliases."""

    normalized = "iqr" if value is None else str(value).strip().lower().replace("-", "_")
    normalized = {"interquartile": "iqr", "inter_quartile": "iqr", "median_absolute_deviation": "mad", "sd": "std", "off": "none", "false": "none", "no": "none"}.get(normalized, normalized)
    if normalized not in SCALE_MODES:
        raise ValueError(f"Unknown scale mode {value!r}. Available modes: {', '.join(SCALE_MODES)}.")
    return normalized