Source feature scaling

neureptrace.decoding.source_scaling implements source-only gain augmentation for feature matrices.

Synthetic rows are scaled copies of source rows. The method uses source rows and labels only.

neureptrace.decoding.source_scaling

Strict source-only feature scaling augmentation.

This module creates scaled copies of labeled source feature rows. It is intended as a dependency-light domain-generalization baseline for M/EEG feature matrices. Synthetic rows keep the source label of the sampled content row while global-row or per-feature gain factors are sampled from source-only settings.

SourceFeatureScalingConfig dataclass

Configuration for strict source-only feature scaling.

Source code in src/neureptrace/decoding/source_scaling.py

@dataclass(frozen=True, slots=True)
class SourceFeatureScalingConfig:
    """Configuration for strict source-only feature scaling."""

    synthetic_per_class: int = 0
    scale_std: float = DEFAULT_SCALE_STD
    scaling_mode: str = "row"
    distribution: str = "lognormal"
    preserve_original: bool = True
    random_state: int | None = 13
    epsilon: float = DEFAULT_EPSILON

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

        object.__setattr__(self, "synthetic_per_class", _nonnegative_int(self.synthetic_per_class, name="synthetic_per_class"))
        object.__setattr__(self, "scale_std", _nonnegative_float(self.scale_std, name="scale_std"))
        object.__setattr__(self, "scaling_mode", normalize_scaling_mode(self.scaling_mode))
        object.__setattr__(self, "distribution", normalize_scaling_distribution(self.distribution))
        object.__setattr__(self, "preserve_original", _bool_value(self.preserve_original, name="preserve_original"))
        object.__setattr__(self, "random_state", _optional_nonnegative_int(self.random_state, name="random_state"))
        object.__setattr__(self, "epsilon", _positive_float(self.epsilon, name="epsilon"))

    @property
    def enabled(self) -> bool:
        """Whether synthetic rows should be generated."""

        return self.synthetic_per_class > 0 and self.scale_std > 0.0

enabled property

Whether synthetic rows should be generated.

__post_init__()

Normalize and validate direct dataclass construction.

Source code in src/neureptrace/decoding/source_scaling.py

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

    object.__setattr__(self, "synthetic_per_class", _nonnegative_int(self.synthetic_per_class, name="synthetic_per_class"))
    object.__setattr__(self, "scale_std", _nonnegative_float(self.scale_std, name="scale_std"))
    object.__setattr__(self, "scaling_mode", normalize_scaling_mode(self.scaling_mode))
    object.__setattr__(self, "distribution", normalize_scaling_distribution(self.distribution))
    object.__setattr__(self, "preserve_original", _bool_value(self.preserve_original, name="preserve_original"))
    object.__setattr__(self, "random_state", _optional_nonnegative_int(self.random_state, name="random_state"))
    object.__setattr__(self, "epsilon", _positive_float(self.epsilon, name="epsilon"))

SourceFeatureScalingResult dataclass

Augmented source rows and provenance.

Source code in src/neureptrace/decoding/source_scaling.py

@dataclass(frozen=True, slots=True)
class SourceFeatureScalingResult:
    """Augmented source rows and provenance."""

    features: np.ndarray
    labels: np.ndarray
    synthetic_mask: np.ndarray
    content_indices: np.ndarray
    scale_factors: np.ndarray
    metadata: dict[str, Any] = field(default_factory=dict)

    @property
    def n_synthetic(self) -> int:
        """Number of generated rows in the returned matrix."""

        return int(np.sum(self.synthetic_mask))

n_synthetic property

Number of generated rows in the returned matrix.

augment_source_with_feature_scaling(source_features, source_labels, *, source_domains=None, config=None)

Append gain-scaled source-row copies.

Parameters:

Name	Type	Description	Default
source_features	Sequence[Sequence[float]] | ndarray	Source feature matrix with rows as trials/windows and columns as features.	required
source_labels	Sequence[Any] | ndarray	One source label per row. Synthetic rows inherit the sampled row label.	required
source_domains	Sequence[Hashable] | ndarray | None	Optional source-domain ids recorded only for provenance.	None
config	SourceFeatureScalingConfig | Mapping[str, Any] | None	Scaling options. Mappings are normalized through :func:source_feature_scaling_config.	None

Source code in src/neureptrace/decoding/source_scaling.py

def augment_source_with_feature_scaling(
    source_features: Sequence[Sequence[float]] | np.ndarray,
    source_labels: Sequence[Any] | np.ndarray,
    *,
    source_domains: Sequence[Hashable] | np.ndarray | None = None,
    config: SourceFeatureScalingConfig | Mapping[str, Any] | None = None,
) -> SourceFeatureScalingResult:
    """Append gain-scaled source-row copies.

    Parameters
    ----------
    source_features:
        Source feature matrix with rows as trials/windows and columns as features.
    source_labels:
        One source label per row.  Synthetic rows inherit the sampled row label.
    source_domains:
        Optional source-domain ids recorded only for provenance.
    config:
        Scaling options.  Mappings are normalized through
        :func:`source_feature_scaling_config`.
    """

    cfg = source_feature_scaling_config() if config is None else _coerce_config(config)
    features = _feature_matrix(source_features, name="source_features")
    labels = _label_vector(source_labels, expected_length=features.shape[0], name="source_labels")
    domains = _domain_vector(source_domains, expected_length=features.shape[0])
    classes, _class_counts = label_counts(labels)
    domain_ids, _domain_counts = label_counts(domains)

    if not cfg.enabled:
        return SourceFeatureScalingResult(
            features=features.astype(np.float32, copy=False),
            labels=labels.copy(),
            synthetic_mask=np.zeros(features.shape[0], dtype=bool),
            content_indices=np.empty(0, dtype=int),
            scale_factors=np.empty((0, features.shape[1]), dtype=np.float32),
            metadata=_metadata(cfg, features.shape[0], 0, classes.shape[0], domain_ids.shape[0], features.shape[1]),
        )

    rng = np.random.default_rng(cfg.random_state)
    synthetic_rows: list[np.ndarray] = []
    synthetic_labels: list[Any] = []
    content_indices: list[int] = []
    scale_rows: list[np.ndarray] = []
    for class_label in classes.tolist():
        class_indices = np.flatnonzero(label_equal_mask(labels, class_label))
        if class_indices.size == 0:
            continue
        for _ in range(cfg.synthetic_per_class):
            content_index = int(rng.choice(class_indices))
            scale = sample_scaling_factors(
                features.shape[1],
                scale_std=cfg.scale_std,
                scaling_mode=cfg.scaling_mode,
                distribution=cfg.distribution,
                epsilon=cfg.epsilon,
                rng=rng,
            )
            synthetic_rows.append(features[content_index] * scale)
            synthetic_labels.append(class_label)
            content_indices.append(content_index)
            scale_rows.append(scale)

    synthetic_features = np.vstack(synthetic_rows).astype(np.float32, copy=False) if synthetic_rows else np.empty((0, features.shape[1]), dtype=np.float32)
    synthetic_label_array = _label_output_vector(synthetic_labels, dtype=labels.dtype)
    scale_array = np.vstack(scale_rows).astype(np.float32, copy=False) if scale_rows else np.empty((0, features.shape[1]), dtype=np.float32)
    if cfg.preserve_original:
        output_features = np.vstack([features, synthetic_features]).astype(np.float32, copy=False)
        output_labels = np.concatenate([labels, synthetic_label_array])
        synthetic_mask = np.concatenate([np.zeros(features.shape[0], dtype=bool), np.ones(synthetic_features.shape[0], dtype=bool)])
    else:
        output_features = synthetic_features
        output_labels = synthetic_label_array
        synthetic_mask = np.ones(synthetic_features.shape[0], dtype=bool)

    return SourceFeatureScalingResult(
        features=output_features,
        labels=output_labels,
        synthetic_mask=synthetic_mask,
        content_indices=np.asarray(content_indices, dtype=int),
        scale_factors=scale_array,
        metadata=_metadata(cfg, features.shape[0], synthetic_features.shape[0], classes.shape[0], domain_ids.shape[0], features.shape[1]),
    )

sample_scaling_factors(n_features, *, scale_std=DEFAULT_SCALE_STD, scaling_mode='row', distribution='lognormal', epsilon=DEFAULT_EPSILON, rng=None)

Sample positive scaling factors for one synthetic row.

Source code in src/neureptrace/decoding/source_scaling.py

def sample_scaling_factors(
    n_features: int | str,
    *,
    scale_std: float | str = DEFAULT_SCALE_STD,
    scaling_mode: str = "row",
    distribution: str = "lognormal",
    epsilon: float | str = DEFAULT_EPSILON,
    rng: np.random.Generator | None = None,
) -> np.ndarray:
    """Sample positive scaling factors for one synthetic row."""

    width = _positive_int(n_features, name="n_features")
    std = _nonnegative_float(scale_std, name="scale_std")
    eps = _positive_float(epsilon, name="epsilon")
    mode = normalize_scaling_mode(scaling_mode)
    dist = normalize_scaling_distribution(distribution)
    generator = np.random.default_rng() if rng is None else rng
    size = 1 if mode == "row" else width
    if dist == "lognormal":
        values = np.exp(generator.normal(0.0, std, size=size))
    elif dist == "uniform":
        values = generator.uniform(max(eps, 1.0 - std), 1.0 + std, size=size)
    elif dist == "normal":
        values = np.maximum(generator.normal(1.0, std, size=size), eps)
    else:  # pragma: no cover - guarded by normalization
        raise ValueError(f"Unhandled scaling distribution {dist!r}.")
    if mode == "row":
        values = np.full(width, float(values[0]), dtype=float)
    return np.asarray(values, dtype=np.float32)

source_feature_scaling_config(*, synthetic_per_class=0, scale_std=DEFAULT_SCALE_STD, scaling_mode='row', distribution='lognormal', preserve_original=True, random_state=13, epsilon=DEFAULT_EPSILON)

Normalize public feature-scaling options.

Source code in src/neureptrace/decoding/source_scaling.py

def source_feature_scaling_config(
    *,
    synthetic_per_class: int | str = 0,
    scale_std: float | str = DEFAULT_SCALE_STD,
    scaling_mode: str | None = "row",
    distribution: str | None = "lognormal",
    preserve_original: bool | int | str = True,
    random_state: int | str | None = 13,
    epsilon: float | str = DEFAULT_EPSILON,
) -> SourceFeatureScalingConfig:
    """Normalize public feature-scaling options."""

    return SourceFeatureScalingConfig(
        synthetic_per_class=_nonnegative_int(synthetic_per_class, name="synthetic_per_class"),
        scale_std=_nonnegative_float(scale_std, name="scale_std"),
        scaling_mode=normalize_scaling_mode(scaling_mode),
        distribution=normalize_scaling_distribution(distribution),
        preserve_original=_bool_value(preserve_original, name="preserve_original"),
        random_state=_optional_nonnegative_int(random_state, name="random_state"),
        epsilon=_positive_float(epsilon, name="epsilon"),
    )