Source MixUp

neureptrace.decoding.source_mixup implements dependency-light feature-space MixUp for source-only domain generalization.

The protocol is Category 1 / strict source-only. It uses source features, source labels, and optional source-domain identifiers. It does not accept target features or target labels.

Synthetic rows are convex combinations of source rows. The result contains both hard labels for existing scikit-learn style pipelines and soft label distributions for consumers that support probabilistic targets.

Typical usage:

from neureptrace.decoding.source_mixup import augment_source_with_mixup

result = augment_source_with_mixup(
    X_source,
    y_source,
    source_domains=subject_ids,
    config={
        "synthetic_per_class": 8,
        "same_class_partner": True,
        "cross_domain_partner": True,
        "random_state": 13,
    },
)

X_aug = result.features
y_aug = result.labels

neureptrace.decoding.source_mixup

Source-only MixUp feature augmentation for domain generalization.

The helpers in this module implement dependency-light feature-space MixUp for cross-subject M/EEG decoding. Synthetic rows are convex combinations of source rows. Labels are represented both as hard labels for existing scikit-learn style pipelines and as class-probability targets for consumers that support soft-label training.

This is a strict source-only / Protocol-1 utility: the public APIs use source features, source labels, and optional source-domain ids only. Target features and target labels are intentionally not accepted.

SourceMixUpConfig dataclass

Configuration for source-only feature-space MixUp.

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

@dataclass(frozen=True, slots=True)
class SourceMixUpConfig:
    """Configuration for source-only feature-space MixUp."""

    synthetic_per_class: int = 0
    alpha: float = DEFAULT_MIXUP_ALPHA
    random_state: int | None = 13
    same_class_partner: bool = True
    cross_domain_partner: bool = True
    hard_label_policy: str = "content"
    preserve_original: bool = True

    @property
    def enabled(self) -> bool:
        """Whether this config requests synthetic rows."""

        return self.synthetic_per_class > 0

enabled property

Whether this config requests synthetic rows.

SourceMixUpResult dataclass

Augmented features, hard labels, soft labels, and provenance metadata.

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

@dataclass(frozen=True, slots=True)
class SourceMixUpResult:
    """Augmented features, hard labels, soft labels, and provenance metadata."""

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

    @property
    def n_synthetic(self) -> int:
        """Number of synthetic rows in the output."""

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

n_synthetic property

Number of synthetic rows in the output.

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

Append source-only MixUp synthetic rows.

Parameters:

Name	Type	Description	Default
source_features	Sequence[Sequence[float]] | ndarray	Source feature matrix. Rows are trials/windows; columns are flattened M/EEG features or latent features.	required
source_labels	Sequence[Any] | ndarray	One source class label per row.	required
source_domains	Sequence[Hashable] | ndarray | None	Optional source-domain identifiers, usually source-subject ids. When cross_domain_partner=True, partner rows are drawn from another source domain whenever possible.	None
config	SourceMixUpConfig | Mapping[str, Any] | None	MixUp settings. A mapping is normalized through :func:source_mixup_config.	None

Returns:

Type	Description
SourceMixUpResult	Feature rows, hard labels, soft label distributions, synthetic mask, and protocol metadata.

Notes

The API intentionally has no target-feature or target-label arguments. This keeps the method valid for strict source-only Protocol 1 evaluation.

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

def augment_source_with_mixup(
    source_features: Sequence[Sequence[float]] | np.ndarray,
    source_labels: Sequence[Any] | np.ndarray,
    *,
    source_domains: Sequence[Hashable] | np.ndarray | None = None,
    config: SourceMixUpConfig | Mapping[str, Any] | None = None,
) -> SourceMixUpResult:
    """Append source-only MixUp synthetic rows.

    Parameters
    ----------
    source_features:
        Source feature matrix.  Rows are trials/windows; columns are flattened
        M/EEG features or latent features.
    source_labels:
        One source class label per row.
    source_domains:
        Optional source-domain identifiers, usually source-subject ids.  When
        ``cross_domain_partner=True``, partner rows are drawn from another source
        domain whenever possible.
    config:
        MixUp settings.  A mapping is normalized through
        :func:`source_mixup_config`.

    Returns
    -------
    SourceMixUpResult
        Feature rows, hard labels, soft label distributions, synthetic mask, and
        protocol metadata.

    Notes
    -----
    The API intentionally has no target-feature or target-label arguments.  This
    keeps the method valid for strict source-only Protocol 1 evaluation.
    """

    cfg = source_mixup_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 = _object_array(_unique_values(labels))
    original_distributions = _one_hot(labels, classes)

    if not cfg.enabled:
        metadata = _metadata(
            cfg,
            n_source_rows=features.shape[0],
            n_synthetic_rows=0,
            n_classes=classes.shape[0],
            n_source_domains=len(_unique_values(domains)),
        )
        return SourceMixUpResult(
            features=features.astype(np.float32, copy=False),
            labels=labels.copy(),
            classes=classes,
            label_distributions=original_distributions,
            synthetic_mask=np.zeros(features.shape[0], dtype=bool),
            content_indices=np.empty(0, dtype=int),
            partner_indices=np.empty(0, dtype=int),
            lambdas=np.empty(0, dtype=float),
            metadata=metadata,
        )

    rng = np.random.default_rng(cfg.random_state)
    synthetic_rows: list[np.ndarray] = []
    synthetic_labels: list[Any] = []
    synthetic_distributions: list[np.ndarray] = []
    content_indices: list[int] = []
    partner_indices: list[int] = []
    lambdas: list[float] = []

    for class_label in classes.tolist():
        class_indices = np.flatnonzero(_object_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))
            partner_pool = _partner_pool(
                labels,
                domains,
                content_index=content_index,
                class_indices=class_indices,
                same_class_partner=cfg.same_class_partner,
                cross_domain_partner=cfg.cross_domain_partner,
            )
            partner_index = int(rng.choice(partner_pool))
            lam = float(rng.beta(cfg.alpha, cfg.alpha))
            row = mixup_rows(
                features[content_index : content_index + 1],
                features[partner_index : partner_index + 1],
                lambdas=np.asarray([lam], dtype=float),
            )[0]
            distribution = np.zeros(classes.shape[0], dtype=float)
            distribution[_find_value_index(classes, labels[content_index])] += lam
            distribution[_find_value_index(classes, labels[partner_index])] += 1.0 - lam
            synthetic_rows.append(row)
            synthetic_distributions.append(distribution)
            synthetic_labels.append(_hard_label(labels[content_index], labels[partner_index], lam, policy=cfg.hard_label_policy))
            content_indices.append(content_index)
            partner_indices.append(partner_index)
            lambdas.append(lam)

    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 = _object_array(synthetic_labels)
    synthetic_distribution_array = np.vstack(synthetic_distributions).astype(np.float32, copy=False) if synthetic_distributions else np.empty((0, classes.shape[0]), 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])
        output_distributions = np.vstack([original_distributions, synthetic_distribution_array]).astype(np.float32, copy=False)
        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
        output_distributions = synthetic_distribution_array
        synthetic_mask = np.ones(synthetic_features.shape[0], dtype=bool)

    metadata = _metadata(
        cfg,
        n_source_rows=features.shape[0],
        n_synthetic_rows=synthetic_features.shape[0],
        n_classes=classes.shape[0],
        n_source_domains=len(_unique_values(domains)),
    )
    return SourceMixUpResult(
        features=output_features,
        labels=output_labels,
        classes=classes,
        label_distributions=output_distributions,
        synthetic_mask=synthetic_mask,
        content_indices=np.asarray(content_indices, dtype=int),
        partner_indices=np.asarray(partner_indices, dtype=int),
        lambdas=np.asarray(lambdas, dtype=float),
        metadata=metadata,
    )

mixup_rows(content_features, partner_features, *, lambdas)

Return convex combinations of content and partner feature rows.

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

def mixup_rows(
    content_features: Sequence[Sequence[float]] | np.ndarray,
    partner_features: Sequence[Sequence[float]] | np.ndarray,
    *,
    lambdas: Sequence[float] | np.ndarray | float,
) -> np.ndarray:
    """Return convex combinations of content and partner feature rows."""

    content = _feature_matrix(content_features, name="content_features")
    partner = _feature_matrix(partner_features, name="partner_features")
    if content.shape != partner.shape:
        raise ValueError(f"content_features and partner_features must have the same shape: {content.shape} != {partner.shape}.")
    lam = np.asarray(lambdas, dtype=float)
    if lam.ndim == 0:
        lam = np.full(content.shape[0], float(lam), dtype=float)
    lam = lam.reshape(-1, 1)
    if lam.shape[0] != content.shape[0]:
        raise ValueError(f"lambdas must be scalar or contain one value per row: {lam.shape[0]} != {content.shape[0]}.")
    if not np.all(np.isfinite(lam)) or np.any(lam < 0.0) or np.any(lam > 1.0):
        raise ValueError("lambdas must be finite values in [0, 1].")
    return (lam * content + (1.0 - lam) * partner).astype(np.float32, copy=False)

source_mixup_config(*, synthetic_per_class=0, alpha=DEFAULT_MIXUP_ALPHA, random_state=13, same_class_partner=True, cross_domain_partner=True, hard_label_policy='content', preserve_original=True)

Normalize user-facing source-MixUp options.

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

def source_mixup_config(
    *,
    synthetic_per_class: int | str = 0,
    alpha: float | str = DEFAULT_MIXUP_ALPHA,
    random_state: Any = 13,
    same_class_partner: Any = True,
    cross_domain_partner: Any = True,
    hard_label_policy: str = "content",
    preserve_original: Any = True,
) -> SourceMixUpConfig:
    """Normalize user-facing source-MixUp options."""

    return _build_config(
        synthetic_per_class=synthetic_per_class,
        alpha=alpha,
        random_state=random_state,
        same_class_partner=same_class_partner,
        cross_domain_partner=cross_domain_partner,
        hard_label_policy=hard_label_policy,
        preserve_original=preserve_original,
    )

normalize_hard_label_policy(value)

Normalize hard-label policies for synthetic MixUp rows.

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

def normalize_hard_label_policy(value: str | None) -> str:
    """Normalize hard-label policies for synthetic MixUp rows."""

    normalized = "content" if value is None else str(value).strip().lower().replace("-", "_")
    normalized = {
        "content_label": "content",
        "source": "content",
        "partner_label": "partner",
        "style": "partner",
        "lambda_dominant": "dominant",
        "majority": "dominant",
        "argmax": "dominant",
    }.get(normalized, normalized)
    if normalized not in HARD_LABEL_POLICIES:
        raise ValueError(f"Unknown hard_label_policy {value!r}. Available policies: {', '.join(HARD_LABEL_POLICIES)}.")
    return normalized