Source-domain probability ensemble

neureptrace.decoding.source_ensemble trains one classifier per source domain and combines held-out target probabilities.

Uniform weighting is Category 1 because target rows are only scored. Non-uniform weighting modes are Category 2 because unlabeled target predictions or target feature distributions affect the source-domain weights.

Supported weighting modes are uniform, target_confidence, target_entropy, and target_feature_similarity.

The public API intentionally has no target_labels argument.

neureptrace.decoding.source_ensemble

Source-domain probability ensembles for cross-subject decoding.

SourceDomainModel dataclass

One fitted source-domain classifier and its provenance.

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

@dataclass(frozen=True, slots=True)
class SourceDomainModel:
    """One fitted source-domain classifier and its provenance."""

    domain_id: Hashable
    model: BaseEstimator
    n_rows: int
    classes: np.ndarray

SourceDomainEnsembleResult dataclass

Target probabilities from a source-domain classifier ensemble.

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

@dataclass(frozen=True, slots=True)
class SourceDomainEnsembleResult:
    """Target probabilities from a source-domain classifier ensemble."""

    probabilities: np.ndarray
    predictions: np.ndarray
    classes: np.ndarray
    domain_weights: Mapping[Hashable, float]
    domain_probabilities: Mapping[Hashable, np.ndarray]
    models: Mapping[Hashable, SourceDomainModel]
    metadata: dict[str, Any] = field(default_factory=dict)

fit_source_domain_probability_ensemble(*, source_features, source_labels, source_domains, target_features, estimator=None, weighting='uniform', temperature=DEFAULT_TEMPERATURE, min_classes_per_domain=2, epsilon=DEFAULT_EPSILON)

Fit per-source-domain classifiers and ensemble their target probabilities.

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

def fit_source_domain_probability_ensemble(
    *,
    source_features: Sequence[Sequence[float]] | np.ndarray,
    source_labels: Sequence[Any] | np.ndarray,
    source_domains: Sequence[Hashable] | np.ndarray,
    target_features: Sequence[Sequence[float]] | np.ndarray,
    estimator: BaseEstimator | None = None,
    weighting: str = "uniform",
    temperature: float | str = DEFAULT_TEMPERATURE,
    min_classes_per_domain: int | str = 2,
    epsilon: float | str = DEFAULT_EPSILON,
) -> SourceDomainEnsembleResult:
    """Fit per-source-domain classifiers and ensemble their target probabilities."""

    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, name="source_labels")
    if labels.shape[0] != source.shape[0]:
        raise ValueError(f"source_labels must contain one value per source row: {labels.shape[0]} != {source.shape[0]}.")
    domains = _domain_vector(source_domains, expected_length=source.shape[0])
    classes = _unique_label_vector(labels)
    if classes.shape[0] < 2:
        raise ValueError("At least two source classes are required.")

    class_indices = {_label_key(label): index for index, label in enumerate(classes.tolist())}
    encoded_labels = np.asarray([class_indices[_label_key(label)] for label in labels.tolist()], dtype=int)
    mode = normalize_ensemble_weighting(weighting)
    temp = _positive_float(temperature, name="temperature")
    eps = _positive_float(epsilon, name="epsilon")
    min_classes = _positive_int(min_classes_per_domain, name="min_classes_per_domain")
    model_template = _default_estimator() if estimator is None else estimator

    models: dict[Hashable, SourceDomainModel] = {}
    domain_probabilities: dict[Hashable, np.ndarray] = {}
    for domain in tuple(dict.fromkeys(domains.tolist())):
        mask = domains == domain
        domain_classes = _unique_label_vector(labels[mask])
        if domain_classes.shape[0] < min_classes:
            continue
        model = clone(model_template)
        model.fit(source[mask], encoded_labels[mask])
        probabilities = _aligned_probabilities(model, target, classes=np.arange(classes.shape[0]), epsilon=eps)
        models[domain] = SourceDomainModel(domain_id=domain, model=model, n_rows=int(np.sum(mask)), classes=domain_classes)
        domain_probabilities[domain] = probabilities
    if not models:
        raise ValueError("No source domain had enough classes to train a domain classifier.")

    weights = _domain_weights(mode, domain_probabilities, source, domains, target, temperature=temp, epsilon=eps)
    probabilities = np.zeros((target.shape[0], classes.shape[0]), dtype=float)
    for domain, domain_probability in domain_probabilities.items():
        probabilities += weights[domain] * domain_probability
    probabilities = _normalize_probability_rows(probabilities, epsilon=eps)
    predictions = classes[np.argmax(probabilities, axis=1)]
    return SourceDomainEnsembleResult(
        probabilities=probabilities.astype(np.float32, copy=False),
        predictions=predictions,
        classes=classes,
        domain_weights=weights,
        domain_probabilities={domain: values.astype(np.float32, copy=False) for domain, values in domain_probabilities.items()},
        models=models,
        metadata=_metadata(
            mode=mode,
            n_source_rows=source.shape[0],
            n_target_rows=target.shape[0],
            feature_dim=source.shape[1],
            n_classes=classes.shape[0],
            n_source_domains=len(tuple(dict.fromkeys(domains.tolist()))),
            n_trained_domains=len(models),
            weights=weights,
            temperature=temp,
            min_classes=min_classes,
        ),
    )

normalize_ensemble_weighting(value)

Normalize public source-domain ensemble weighting aliases.

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

def normalize_ensemble_weighting(value: str | None) -> str:
    """Normalize public source-domain ensemble weighting aliases."""

    normalized = "uniform" if value is None else str(value).strip().lower().replace("-", "_")
    normalized = {
        "equal": "uniform",
        "mean": "uniform",
        "confidence": "target_confidence",
        "target_conf": "target_confidence",
        "low_entropy": "target_entropy",
        "entropy": "target_entropy",
        "feature_similarity": "target_feature_similarity",
        "target_similarity": "target_feature_similarity",
        "mean_covariance": "target_feature_similarity",
    }.get(normalized, normalized)
    if normalized not in ENSEMBLE_WEIGHTING_MODES:
        raise ValueError(f"Unknown source-domain ensemble weighting {value!r}. Available modes: {', '.join(ENSEMBLE_WEIGHTING_MODES)}.")
    return normalized