Subspace adaptation

neureptrace.decoding.subspace_adaptation implements Category-2 TCA-style source-target latent projection.

The fit uses source features and unlabeled target features. Optional source labels can be used to balance the source side of the marginal domain discrepancy. Held-out target labels are not part of the public API.

neureptrace.decoding.subspace_adaptation

Category-2 source-target subspace adaptation utilities.

The helpers in this module implement a small dependency-light Transfer Component Analysis style projection for cross-subject M/EEG features. The projection is fit from source features and unlabeled target features; optional source labels can be used only to balance the source side of the marginal domain discrepancy. Held-out target labels are intentionally absent from the public API.

SubspaceAdaptationConfig dataclass

Configuration for Category-2 TCA-style feature projection.

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

@dataclass(frozen=True, slots=True)
class SubspaceAdaptationConfig:
    """Configuration for Category-2 TCA-style feature projection."""

    method: str = DEFAULT_SUBSPACE_METHOD
    n_components: int | str = DEFAULT_SUBSPACE_COMPONENTS
    regularization: float = DEFAULT_SUBSPACE_REGULARIZATION
    eigen_ridge: float = DEFAULT_SUBSPACE_EIGEN_RIDGE
    standardize: bool = True
    class_balance_source: bool = False
    normalize_latent: bool = False

SubspaceAdaptationResult dataclass

Source/target features projected into a target-adaptive latent space.

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

@dataclass(frozen=True, slots=True)
class SubspaceAdaptationResult:
    """Source/target features projected into a target-adaptive latent space."""

    source_features: np.ndarray
    target_features: np.ndarray
    projection: np.ndarray
    feature_mean: np.ndarray
    feature_scale: np.ndarray
    eigenvalues: np.ndarray
    source_weights: np.ndarray
    metadata: dict[str, Any] = field(default_factory=dict)

fit_subspace_adaptation(source_features, target_features, *, source_labels=None, config=None, method=None, n_components=None, regularization=None, eigen_ridge=None, standardize=None, class_balance_source=None, normalize_latent=None)

Fit a TCA-style subspace using source and unlabeled target features.

Parameters:

Name	Type	Description	Default
source_features	Sequence[Sequence[float]] | ndarray	Source and held-out target feature matrices with the same feature width. Target rows are used without labels to fit the shared projection.	required
target_features	Sequence[Sequence[float]] | ndarray	Source and held-out target feature matrices with the same feature width. Target rows are used without labels to fit the shared projection.	required
source_labels	Sequence[Any] | ndarray | None	Optional source labels. They are used only when source class balancing is requested, so source classes contribute equal mass to the marginal source-target discrepancy.	None
config	SubspaceAdaptationConfig | dict[str, Any] | None		None
method	SubspaceAdaptationConfig | dict[str, Any] | None		None
n_components	SubspaceAdaptationConfig | dict[str, Any] | None		None
regularization	SubspaceAdaptationConfig | dict[str, Any] | None		None
eigen_ridge	SubspaceAdaptationConfig | dict[str, Any] | None		None
standardize	SubspaceAdaptationConfig | dict[str, Any] | None		None
class_balance_source	bool | None	Configuration values. Explicit keyword values override config.	None
normalize_latent	bool | None	Configuration values. Explicit keyword values override config.	None

Returns:

Type	Description
SubspaceAdaptationResult	Projected source and target features plus projection/provenance fields.

Notes

This is a Category-2 protocol. It uses X_s and unlabeled X_t; it may use y_s for source-side class balancing; it never accepts target labels.

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

def fit_subspace_adaptation(
    source_features: Sequence[Sequence[float]] | np.ndarray,
    target_features: Sequence[Sequence[float]] | np.ndarray,
    *,
    source_labels: Sequence[Any] | np.ndarray | None = None,
    config: SubspaceAdaptationConfig | dict[str, Any] | None = None,
    method: str | None = None,
    n_components: int | str | None = None,
    regularization: float | str | None = None,
    eigen_ridge: float | str | None = None,
    standardize: bool | None = None,
    class_balance_source: bool | None = None,
    normalize_latent: bool | None = None,
) -> SubspaceAdaptationResult:
    """Fit a TCA-style subspace using source and unlabeled target features.

    Parameters
    ----------
    source_features, target_features:
        Source and held-out target feature matrices with the same feature width.
        Target rows are used without labels to fit the shared projection.
    source_labels:
        Optional source labels.  They are used only when source class balancing is
        requested, so source classes contribute equal mass to the marginal
        source-target discrepancy.
    config, method, n_components, regularization, eigen_ridge, standardize,
    class_balance_source, normalize_latent:
        Configuration values.  Explicit keyword values override ``config``.

    Returns
    -------
    SubspaceAdaptationResult
        Projected source and target features plus projection/provenance fields.

    Notes
    -----
    This is a Category-2 protocol.  It uses ``X_s`` and unlabeled ``X_t``; it may
    use ``y_s`` for source-side class balancing; it never accepts target labels.
    """

    cfg = _resolve_config(
        config,
        method=method,
        n_components=n_components,
        regularization=regularization,
        eigen_ridge=eigen_ridge,
        standardize=standardize,
        class_balance_source=class_balance_source,
        normalize_latent=normalize_latent,
    )
    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]}.")
    if cfg.class_balance_source and source_labels is None:
        raise ValueError("class_balance_source=True requires source_labels.")
    labels = None if source_labels is None else _object_vector(source_labels, expected_length=source.shape[0], name="source_labels")

    joint = np.vstack([source, target]).astype(float, copy=False)
    mean = np.mean(joint, axis=0) if cfg.standardize else np.zeros(joint.shape[1], dtype=float)
    centered = joint - mean
    scale = np.std(centered, axis=0, ddof=1 if joint.shape[0] > 1 else 0) if cfg.standardize else np.ones(joint.shape[1], dtype=float)
    scale = np.maximum(scale, MIN_SCALE)
    z = centered / scale

    source_weights = _source_weights(source.shape[0], labels=labels, class_balance=cfg.class_balance_source)
    target_weights = np.full(target.shape[0], 1.0 / float(target.shape[0]), dtype=float)
    domain_vector = np.concatenate([source_weights, -target_weights])
    mmd_matrix = np.outer(domain_vector, domain_vector)
    centering = np.eye(joint.shape[0], dtype=float) - np.full((joint.shape[0], joint.shape[0]), 1.0 / float(joint.shape[0]))

    feature_dim = z.shape[1]
    n_components_resolved = _effective_components(cfg.n_components, n_samples=z.shape[0], n_features=feature_dim)
    a_matrix = z.T @ mmd_matrix @ z + cfg.regularization * np.eye(feature_dim, dtype=float)
    b_matrix = z.T @ centering @ z + cfg.eigen_ridge * np.eye(feature_dim, dtype=float)
    values, vectors = eigh(a_matrix, b_matrix, check_finite=True)
    order = np.argsort(values)
    selected = order[:n_components_resolved]
    projection = _canonicalize_projection(vectors[:, selected])
    latent = z @ projection
    if cfg.normalize_latent:
        latent_scale = np.maximum(np.std(latent, axis=0, ddof=1 if latent.shape[0] > 1 else 0), MIN_SCALE)
        latent = latent / latent_scale
        projection = projection / latent_scale.reshape(1, -1)
    source_latent = latent[: source.shape[0]]
    target_latent = latent[source.shape[0] :]

    raw_gap = _weighted_mean_gap((source - mean) / scale, (target - mean) / scale, source_weights=source_weights, target_weights=target_weights)
    latent_gap = _weighted_mean_gap(source_latent, target_latent, source_weights=source_weights, target_weights=target_weights)
    metadata = _metadata(
        cfg=cfg,
        n_source_rows=source.shape[0],
        n_target_rows=target.shape[0],
        feature_dim=feature_dim,
        n_components=n_components_resolved,
        source_labels_used=labels is not None and cfg.class_balance_source,
        raw_gap=raw_gap,
        latent_gap=latent_gap,
        eigenvalues=values[selected],
    )
    return SubspaceAdaptationResult(
        source_features=source_latent.astype(np.float32, copy=False),
        target_features=target_latent.astype(np.float32, copy=False),
        projection=projection.astype(np.float32, copy=False),
        feature_mean=mean.astype(np.float32, copy=False),
        feature_scale=scale.astype(np.float32, copy=False),
        eigenvalues=np.asarray(values[selected], dtype=float),
        source_weights=source_weights.astype(float, copy=False),
        metadata=metadata,
    )

transform_subspace_features(features, result)

Transform new rows with an already fitted subspace adaptation result.

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

def transform_subspace_features(features: Sequence[Sequence[float]] | np.ndarray, result: SubspaceAdaptationResult) -> np.ndarray:
    """Transform new rows with an already fitted subspace adaptation result."""

    matrix = _feature_matrix(features, name="features")
    if matrix.shape[1] != result.projection.shape[0]:
        raise ValueError(f"features width {matrix.shape[1]} does not match projection width {result.projection.shape[0]}.")
    return (((matrix - result.feature_mean) / result.feature_scale) @ result.projection).astype(np.float32, copy=False)

subspace_adaptation_config(*, method=DEFAULT_SUBSPACE_METHOD, n_components=DEFAULT_SUBSPACE_COMPONENTS, regularization=DEFAULT_SUBSPACE_REGULARIZATION, eigen_ridge=DEFAULT_SUBSPACE_EIGEN_RIDGE, standardize=True, class_balance_source=False, normalize_latent=False)

Normalize public configuration values.

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

def subspace_adaptation_config(
    *,
    method: str | None = DEFAULT_SUBSPACE_METHOD,
    n_components: int | str | None = DEFAULT_SUBSPACE_COMPONENTS,
    regularization: float | str = DEFAULT_SUBSPACE_REGULARIZATION,
    eigen_ridge: float | str = DEFAULT_SUBSPACE_EIGEN_RIDGE,
    standardize: bool = True,
    class_balance_source: bool = False,
    normalize_latent: bool = False,
) -> SubspaceAdaptationConfig:
    """Normalize public configuration values."""

    normalized_method = normalize_subspace_method(method)
    balance = bool(class_balance_source or normalized_method == "balanced_tca")
    return SubspaceAdaptationConfig(
        method=normalized_method,
        n_components=_normalize_components_request(n_components),
        regularization=_nonnegative_float(regularization, name="regularization"),
        eigen_ridge=_positive_float(eigen_ridge, name="eigen_ridge"),
        standardize=bool(standardize),
        class_balance_source=balance,
        normalize_latent=bool(normalize_latent),
    )

normalize_subspace_method(method)

Normalize aliases for subspace-adaptation methods.

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

def normalize_subspace_method(method: str | None) -> str:
    """Normalize aliases for subspace-adaptation methods."""

    normalized = DEFAULT_SUBSPACE_METHOD if method is None else str(method).strip().lower().replace("-", "_")
    normalized = {
        "transfer_component_analysis": "tca",
        "marginal_tca": "tca",
        "source_balanced_tca": "balanced_tca",
        "class_balanced_tca": "balanced_tca",
        "balanced_transfer_component_analysis": "balanced_tca",
    }.get(normalized, normalized)
    if normalized not in SUBSPACE_METHODS:
        raise ValueError(f"Unknown subspace method {method!r}. Available methods: {', '.join(SUBSPACE_METHODS)}.")
    return normalized