Conditional CORAL

neureptrace.decoding.conditional_coral implements pseudo-label class-conditional CORAL for cross-subject transfer.

The protocol is Category 2 / unlabeled target-adaptive. It uses source features and source labels, plus unlabeled target features with pseudo-labels or probability predictions. It does not accept held-out target labels.

Supported pseudo-label sources:

• caller-supplied target_pseudo_labels,
• caller-supplied target_probabilities in source-class order,
• a source-trained classifier fitted internally when neither is supplied.

If a pseudo-class has too few confident target rows, the implementation can fall back to global target statistics or raise an error.

neureptrace.decoding.conditional_coral

Pseudo-label conditional CORAL alignment for Category-2 transfer.

The helpers in this module implement a class-conditional CORAL transform for cross-subject M/EEG feature matrices. Source class distributions are aligned toward target pseudo-class distributions estimated from classifier predictions or caller-supplied pseudo-labels/probabilities.

The public API intentionally has no target-label argument. Target rows may be used for pseudo-label adaptation, but held-out target labels must remain reserved for scoring.

ConditionalCoralConfig dataclass

Configuration for pseudo-label conditional CORAL.

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

@dataclass(frozen=True, slots=True)
class ConditionalCoralConfig:
    """Configuration for pseudo-label conditional CORAL."""

    regularization: float = DEFAULT_CONDITIONAL_CORAL_REGULARIZATION
    min_target_rows_per_class: int = DEFAULT_CONDITIONAL_CORAL_MIN_TARGET_ROWS
    confidence_threshold: float = 0.0
    fallback: str = "global"
    center: bool = True
    random_state: int | None = 13

CoralClassStats dataclass

Class/domain feature statistics used by CORAL.

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

@dataclass(frozen=True, slots=True)
class CoralClassStats:
    """Class/domain feature statistics used by CORAL."""

    mean: np.ndarray
    covariance: np.ndarray
    n_rows: int

ConditionalCoralResult dataclass

Aligned train/test features and pseudo-label provenance.

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

@dataclass(frozen=True, slots=True)
class ConditionalCoralResult:
    """Aligned train/test features and pseudo-label provenance."""

    train_features: np.ndarray
    test_features: np.ndarray
    classes: np.ndarray
    pseudo_labels: np.ndarray
    pseudo_confidence: np.ndarray
    source_stats: Mapping[Any, CoralClassStats]
    target_stats: Mapping[Any, CoralClassStats]
    global_source_stats: CoralClassStats
    global_target_stats: CoralClassStats
    used_fallback_classes: tuple[Any, ...]
    metadata: dict[str, Any] = field(default_factory=dict)

fit_pseudo_label_conditional_coral(*, source_features, source_labels, target_features, config=None, estimator=None, target_pseudo_labels=None, target_probabilities=None)

Fit class-conditional CORAL using target pseudo-labels.

Parameters:

Name	Type	Description	Default
source_features	Sequence[Sequence[float]] | ndarray	Labeled source rows used to estimate source class distributions.	required
source_labels	Sequence[Sequence[float]] | ndarray	Labeled source rows used to estimate source class distributions.	required
target_features	Sequence[Sequence[float]] | ndarray	Unlabeled target rows. They are used to estimate pseudo-class target distributions, but not target labels.	required
config	ConditionalCoralConfig | Mapping[str, Any] | None	Conditional CORAL settings. A mapping is normalized through :func:conditional_coral_config.	None
estimator	BaseEstimator | None	Optional sklearn-style source classifier used when neither target_pseudo_labels nor target_probabilities are supplied.	None
target_pseudo_labels	Sequence[Any] | ndarray | None	Optional classifier-generated target pseudo-labels. These must be in the source class set and are not treated as true target labels.	None
target_probabilities	Sequence[Sequence[float]] | ndarray | None	Optional target class probabilities in source-class order. Argmax labels become pseudo-labels and max probability becomes pseudo-confidence.	None

Returns:

Type	Description
ConditionalCoralResult	Source rows aligned class-wise toward pseudo-target class distributions; target rows are returned in their native feature space.

Notes

This is a Category-2 protocol. The public API intentionally has no target_labels parameter.

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

def fit_pseudo_label_conditional_coral(
    *,
    source_features: Sequence[Sequence[float]] | np.ndarray,
    source_labels: Sequence[Any] | np.ndarray,
    target_features: Sequence[Sequence[float]] | np.ndarray,
    config: ConditionalCoralConfig | Mapping[str, Any] | None = None,
    estimator: BaseEstimator | None = None,
    target_pseudo_labels: Sequence[Any] | np.ndarray | None = None,
    target_probabilities: Sequence[Sequence[float]] | np.ndarray | None = None,
) -> ConditionalCoralResult:
    """Fit class-conditional CORAL using target pseudo-labels.

    Parameters
    ----------
    source_features, source_labels:
        Labeled source rows used to estimate source class distributions.
    target_features:
        Unlabeled target rows.  They are used to estimate pseudo-class target
        distributions, but not target labels.
    config:
        Conditional CORAL settings.  A mapping is normalized through
        :func:`conditional_coral_config`.
    estimator:
        Optional sklearn-style source classifier used when neither
        ``target_pseudo_labels`` nor ``target_probabilities`` are supplied.
    target_pseudo_labels:
        Optional classifier-generated target pseudo-labels.  These must be in the
        source class set and are not treated as true target labels.
    target_probabilities:
        Optional target class probabilities in source-class order.  Argmax labels
        become pseudo-labels and max probability becomes pseudo-confidence.

    Returns
    -------
    ConditionalCoralResult
        Source rows aligned class-wise toward pseudo-target class distributions;
        target rows are returned in their native feature space.

    Notes
    -----
    This is a Category-2 protocol.  The public API intentionally has no
    ``target_labels`` parameter.
    """

    cfg = conditional_coral_config() if config is None else _coerce_config(config)
    source = _feature_matrix(source_features, name="source_features")
    target = _feature_matrix(target_features, name="target_features")
    if source.shape[1] != target.shape[1]:
        raise ValueError(f"source_features and target_features must have the same feature width: {source.shape[1]} != {target.shape[1]}.")
    labels = _label_vector(source_labels, expected_length=source.shape[0], name="source_labels")
    classes, _class_counts = label_counts(labels)
    if classes.shape[0] < 2:
        raise ValueError("Conditional CORAL requires at least two source classes.")

    pseudo_labels, pseudo_confidence, pseudo_source = _resolve_pseudo_labels(
        source,
        labels,
        target,
        classes=classes,
        estimator=estimator,
        config=cfg,
        target_pseudo_labels=target_pseudo_labels,
        target_probabilities=target_probabilities,
    )
    confident_mask = pseudo_confidence >= cfg.confidence_threshold
    source_stats = {class_label: feature_stats(source[label_equal_mask(labels, class_label)], regularization=cfg.regularization) for class_label in classes.tolist()}
    global_source = feature_stats(source, regularization=cfg.regularization)
    global_target = feature_stats(target[confident_mask] if np.any(confident_mask) else target, regularization=cfg.regularization)

    target_stats: dict[Any, CoralClassStats] = {}
    fallback_classes: list[Any] = []
    for class_label in classes.tolist():
        class_mask = label_equal_mask(pseudo_labels, class_label) & confident_mask
        if np.count_nonzero(class_mask) >= cfg.min_target_rows_per_class:
            target_stats[class_label] = feature_stats(target[class_mask], regularization=cfg.regularization)
        elif cfg.fallback == "global":
            target_stats[class_label] = global_target
            fallback_classes.append(class_label)
        else:
            raise ValueError(
                "Target pseudo-class "
                f"{class_label!r} has {int(np.count_nonzero(class_mask))} rows, below min_target_rows_per_class={cfg.min_target_rows_per_class}."
            )

    aligned_source = np.empty_like(source, dtype=float)
    for class_label in classes.tolist():
        class_mask = label_equal_mask(labels, class_label)
        aligned_source[class_mask] = coral_align_features(
            source[class_mask],
            source_stats=source_stats[class_label],
            target_stats=target_stats[class_label],
            center=cfg.center,
        )
    metadata = _metadata(
        cfg,
        n_source_rows=source.shape[0],
        n_target_rows=target.shape[0],
        feature_dim=source.shape[1],
        n_classes=classes.shape[0],
        pseudo_source=pseudo_source,
        pseudo_labels=pseudo_labels,
        confident_mask=confident_mask,
        fallback_classes=tuple(fallback_classes),
    )
    return ConditionalCoralResult(
        train_features=aligned_source.astype(np.float32, copy=False),
        test_features=target.astype(np.float32, copy=False),
        classes=classes,
        pseudo_labels=pseudo_labels,
        pseudo_confidence=pseudo_confidence.astype(float, copy=False),
        source_stats=source_stats,
        target_stats=target_stats,
        global_source_stats=global_source,
        global_target_stats=global_target,
        used_fallback_classes=tuple(fallback_classes),
        metadata=metadata,
    )

conditional_coral_config(*, regularization=DEFAULT_CONDITIONAL_CORAL_REGULARIZATION, min_target_rows_per_class=DEFAULT_CONDITIONAL_CORAL_MIN_TARGET_ROWS, confidence_threshold=0.0, fallback='global', center=True, random_state=13)

Normalize public conditional-CORAL options.

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

def conditional_coral_config(
    *,
    regularization: float | str = DEFAULT_CONDITIONAL_CORAL_REGULARIZATION,
    min_target_rows_per_class: int | str = DEFAULT_CONDITIONAL_CORAL_MIN_TARGET_ROWS,
    confidence_threshold: float | str = 0.0,
    fallback: str = "global",
    center: bool = True,
    random_state: int | str | None = 13,
) -> ConditionalCoralConfig:
    """Normalize public conditional-CORAL options."""

    return ConditionalCoralConfig(
        regularization=_nonnegative_float(regularization, name="regularization"),
        min_target_rows_per_class=_positive_int(min_target_rows_per_class, name="min_target_rows_per_class"),
        confidence_threshold=_unit_interval_float(confidence_threshold, name="confidence_threshold"),
        fallback=normalize_conditional_coral_fallback(fallback),
        center=bool(center),
        random_state=None if random_state in {None, "", "none", "None"} else _nonnegative_int(random_state, name="random_state"),
    )

coral_align_features(features, *, source_stats, target_stats, center=True)

Apply CORAL whitening/recoloring from source stats to target stats.

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

def coral_align_features(
    features: Sequence[Sequence[float]] | np.ndarray,
    *,
    source_stats: CoralClassStats,
    target_stats: CoralClassStats,
    center: bool = True,
) -> np.ndarray:
    """Apply CORAL whitening/recoloring from source stats to target stats."""

    matrix = _feature_matrix(features, name="features")
    if matrix.shape[1] != source_stats.mean.shape[0] or matrix.shape[1] != target_stats.mean.shape[0]:
        raise ValueError("feature width must match source and target statistics.")
    source_inv_sqrt = _matrix_inv_sqrt_spd(source_stats.covariance)
    target_sqrt = _matrix_sqrt_spd(target_stats.covariance)
    centered = matrix - source_stats.mean
    recolored = centered @ source_inv_sqrt @ target_sqrt
    return recolored + target_stats.mean if center else recolored + source_stats.mean

feature_stats(features, *, regularization=DEFAULT_CONDITIONAL_CORAL_REGULARIZATION)

Return mean and regularized covariance for a feature matrix.

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

def feature_stats(features: Sequence[Sequence[float]] | np.ndarray, *, regularization: float = DEFAULT_CONDITIONAL_CORAL_REGULARIZATION) -> CoralClassStats:
    """Return mean and regularized covariance for a feature matrix."""

    matrix = _feature_matrix(features, name="features")
    reg = _nonnegative_float(regularization, name="regularization")
    mean = np.mean(matrix, axis=0)
    centered = matrix - mean
    if matrix.shape[0] <= 1:
        covariance = np.zeros((matrix.shape[1], matrix.shape[1]), dtype=float)
    else:
        covariance = centered.T @ centered / float(matrix.shape[0] - 1)
    covariance = _nearest_spd(covariance + reg * np.eye(matrix.shape[1], dtype=float))
    return CoralClassStats(mean=mean.astype(float, copy=False), covariance=covariance, n_rows=int(matrix.shape[0]))

normalize_conditional_coral_fallback(value)

Normalize fallback policy aliases.

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

def normalize_conditional_coral_fallback(value: str | None) -> str:
    """Normalize fallback policy aliases."""

    normalized = "global" if value is None else str(value).strip().lower().replace("-", "_")
    normalized = {"raise": "error", "strict": "error", "fail": "error"}.get(normalized, normalized)
    if normalized not in {"global", "error"}:
        raise ValueError("fallback must be 'global' or 'error'.")
    return normalized