Domain importance weighting

neureptrace.decoding.domain_importance estimates source sample weights from a binary source-versus-target domain classifier.

This is a Category 2 / unlabeled target-adaptive protocol. It uses source features and unlabeled target features to estimate source-row weights. It does not use target labels.

Typical usage:

from neureptrace.decoding.domain_importance import fit_domain_classifier_importance_weights

result = fit_domain_classifier_importance_weights(
    source_features=X_source,
    target_features=X_target,
)

sample_weight = result.sample_weights

The returned weights can be passed to any downstream classifier that supports sample_weight.

neureptrace.decoding.domain_importance

Domain-classifier importance weighting for unlabeled target adaptation.

This module implements a compact covariate-shift weighting helper for cross-subject M/EEG transfer. A binary domain classifier is trained to separate labeled source feature rows from unlabeled held-out target feature rows. Its source-row domain posteriors are converted into non-negative source sample weights.

The public API intentionally has no target-label argument. This is a Category-2 protocol because unlabeled target features affect source training weights.

DomainImportanceConfig dataclass

Configuration for domain-classifier source weighting.

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

@dataclass(frozen=True, slots=True)
class DomainImportanceConfig:
    """Configuration for domain-classifier source weighting."""

    clip: tuple[float, float] | None = DEFAULT_WEIGHT_CLIP
    normalize: bool = True
    account_for_sample_priors: bool = True
    epsilon: float = DEFAULT_EPSILON

DomainImportanceResult dataclass

Source weights and domain probabilities from unlabeled target adaptation.

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

@dataclass(frozen=True, slots=True)
class DomainImportanceResult:
    """Source weights and domain probabilities from unlabeled target adaptation."""

    sample_weights: np.ndarray
    source_target_probabilities: np.ndarray
    target_target_probabilities: np.ndarray
    domain_classifier: BaseEstimator
    metadata: dict[str, Any] = field(default_factory=dict)

fit_domain_classifier_importance_weights(source_features, target_features, *, estimator=None, config=None)

Estimate source sample weights from an unlabeled target domain classifier.

Parameters:

Name	Type	Description	Default
source_features	Sequence[Sequence[float]] | ndarray	Source feature rows that will later be used with source labels by a downstream classifier.	required
target_features	Sequence[Sequence[float]] | ndarray	Unlabeled held-out target rows used only to estimate source/target feature density ratios.	required
estimator	BaseEstimator | None	Optional sklearn-compatible probabilistic binary classifier. If omitted, a standardized logistic regression classifier is used.	None
config	DomainImportanceConfig | Mapping[str, Any] | None	Weight clipping and normalization settings. A mapping is normalized through :func:domain_importance_config.	None

Returns:

Type	Description
DomainImportanceResult	One non-negative sample weight per source row, plus protocol metadata.

Notes

The API intentionally has no target_labels parameter. The returned source weights are Category-2 because they depend on unlabeled X_t.

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

def fit_domain_classifier_importance_weights(
    source_features: Sequence[Sequence[float]] | np.ndarray,
    target_features: Sequence[Sequence[float]] | np.ndarray,
    *,
    estimator: BaseEstimator | None = None,
    config: DomainImportanceConfig | Mapping[str, Any] | None = None,
) -> DomainImportanceResult:
    """Estimate source sample weights from an unlabeled target domain classifier.

    Parameters
    ----------
    source_features:
        Source feature rows that will later be used with source labels by a
        downstream classifier.
    target_features:
        Unlabeled held-out target rows used only to estimate source/target feature
        density ratios.
    estimator:
        Optional sklearn-compatible probabilistic binary classifier.  If omitted,
        a standardized logistic regression classifier is used.
    config:
        Weight clipping and normalization settings.  A mapping is normalized
        through :func:`domain_importance_config`.

    Returns
    -------
    DomainImportanceResult
        One non-negative sample weight per source row, plus protocol metadata.

    Notes
    -----
    The API intentionally has no ``target_labels`` parameter.  The returned source
    weights are Category-2 because they depend on unlabeled ``X_t``.
    """

    cfg = domain_importance_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]}.")
    classifier = clone(_default_estimator() if estimator is None else estimator)
    domain_features = np.vstack([source, target])
    domain_labels = np.concatenate([np.zeros(source.shape[0], dtype=int), np.ones(target.shape[0], dtype=int)])
    classifier.fit(domain_features, domain_labels)
    source_probabilities = _target_domain_probabilities(classifier, source, epsilon=cfg.epsilon)
    target_probabilities = _target_domain_probabilities(classifier, target, epsilon=cfg.epsilon)
    weights = _posterior_to_importance_weights(
        source_probabilities,
        n_source=source.shape[0],
        n_target=target.shape[0],
        account_for_sample_priors=cfg.account_for_sample_priors,
        epsilon=cfg.epsilon,
    )
    if cfg.clip is not None:
        weights = np.clip(weights, cfg.clip[0], cfg.clip[1])
    if cfg.normalize:
        mean_weight = float(np.mean(weights))
        if mean_weight <= 0.0 or not np.isfinite(mean_weight):
            raise ValueError("Domain-importance weights have non-positive or non-finite mean.")
        weights = weights / mean_weight
    metadata = _metadata(
        cfg,
        n_source_rows=source.shape[0],
        n_target_rows=target.shape[0],
        feature_dim=source.shape[1],
        weights=weights,
        source_probabilities=source_probabilities,
        target_probabilities=target_probabilities,
        estimator_name=type(classifier).__name__,
    )
    return DomainImportanceResult(
        sample_weights=weights.astype(np.float32, copy=False),
        source_target_probabilities=source_probabilities.astype(np.float32, copy=False),
        target_target_probabilities=target_probabilities.astype(np.float32, copy=False),
        domain_classifier=classifier,
        metadata=metadata,
    )

domain_importance_config(*, clip=DEFAULT_WEIGHT_CLIP, normalize=True, account_for_sample_priors=True, epsilon=DEFAULT_EPSILON)

Normalize public domain-importance options.

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

def domain_importance_config(
    *,
    clip: Sequence[float] | str | bool | None = DEFAULT_WEIGHT_CLIP,
    normalize: bool | str = True,
    account_for_sample_priors: bool | str = True,
    epsilon: float | str = DEFAULT_EPSILON,
) -> DomainImportanceConfig:
    """Normalize public domain-importance options."""

    return DomainImportanceConfig(
        clip=_normalize_clip(clip),
        normalize=_boolean(normalize, name="normalize"),
        account_for_sample_priors=_boolean(account_for_sample_priors, name="account_for_sample_priors"),
        epsilon=_positive_float(epsilon, name="epsilon"),
    )

apply_domain_importance_weights(source_features, source_labels, result)

Return source features, labels, and checked domain-importance weights.

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

def apply_domain_importance_weights(
    source_features: Sequence[Sequence[float]] | np.ndarray,
    source_labels: Sequence[Any] | np.ndarray,
    result: DomainImportanceResult,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Return source features, labels, and checked domain-importance weights."""

    features = _feature_matrix(source_features, name="source_features")
    labels = _label_vector(source_labels)
    if labels.shape[0] != features.shape[0]:
        raise ValueError(f"source_labels must contain one value per source row: {labels.shape[0]} != {features.shape[0]}.")
    weights = np.asarray(result.sample_weights, dtype=float).reshape(-1)
    if weights.shape[0] != features.shape[0]:
        raise ValueError(f"result.sample_weights length does not match source_features rows: {weights.shape[0]} != {features.shape[0]}.")
    if not np.all(np.isfinite(weights)) or np.any(weights < 0.0):
        raise ValueError("Domain-importance weights must be finite and non-negative.")
    return features, labels, weights