Skip to content

Source-domain selection

neureptrace.decoding.source_selection provides a generic source-domain selection and weighting utility for cross-subject transfer experiments.

The protocol is Category 2 / unlabeled target-adaptive. It uses source features, source domain identifiers, optional source labels for class-balanced weighting, and unlabeled target features. It does not accept target labels.

neureptrace.decoding.source_selection

Source-domain selection and weighting for cross-subject transfer.

This module implements a generic Category-2 source-domain selection helper that can be used outside the MEKT-specific transfer path. Source subjects are scored by similarity to an unlabeled held-out target feature distribution, then converted into a selected-domain mask and optional sample weights. Target labels are not accepted by the public API.

SourceDomainSelectionResult dataclass

Target-similarity source-domain selection result.

Source code in src/neureptrace/decoding/source_selection.py 
28
29
30
31
32
33
34
35
36
37
@dataclass(frozen=True, slots=True)
class SourceDomainSelectionResult:
    """Target-similarity source-domain selection result."""

    selected_domains: tuple[Hashable, ...]
    domain_distances: Mapping[Hashable, float]
    domain_scores: Mapping[Hashable, float]
    sample_weights: np.ndarray
    selected_mask: np.ndarray
    metadata: dict[str, Any] = field(default_factory=dict)

select_source_domains_by_target_similarity(source_features, source_domains, target_features, *, metric=DEFAULT_SOURCE_SELECTION_METRIC, top_k=None, max_distance=None, min_selected_domains=1, softmax_temperature=DEFAULT_SOURCE_SELECTION_TEMPERATURE, source_labels=None, class_balance=False)

Select or weight source domains by similarity to unlabeled target features.

Source code in src/neureptrace/decoding/source_selection.py 
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def select_source_domains_by_target_similarity(
    source_features: Sequence[Sequence[float]] | np.ndarray,
    source_domains: Sequence[Hashable] | np.ndarray,
    target_features: Sequence[Sequence[float]] | np.ndarray,
    *,
    metric: str | None = DEFAULT_SOURCE_SELECTION_METRIC,
    top_k: int | str | None = None,
    max_distance: float | str | None = None,
    min_selected_domains: int | str = 1,
    softmax_temperature: float | str = DEFAULT_SOURCE_SELECTION_TEMPERATURE,
    source_labels: Sequence[Any] | np.ndarray | None = None,
    class_balance: bool | int | str = False,
) -> SourceDomainSelectionResult:
    """Select or weight source domains by similarity to unlabeled target features."""

    source_matrix = _feature_matrix(source_features, name="source_features")
    target_matrix = _feature_matrix(target_features, name="target_features")
    if source_matrix.shape[1] != target_matrix.shape[1]:
        raise ValueError(
            "source_features and target_features must have the same feature width: "
            f"{source_matrix.shape[1]} != {target_matrix.shape[1]}."
        )

    domain_vector = _domain_vector(source_domains, expected_length=source_matrix.shape[0])
    domains = _unique_domains(domain_vector)
    selected_min = _normalize_positive_int(min_selected_domains, name="min_selected_domains")
    if selected_min > len(domains):
        raise ValueError(f"min_selected_domains={selected_min} exceeds the number of available source domains ({len(domains)}).")

    resolved_top_k = _normalize_optional_positive_int(top_k, name="top_k")
    if resolved_top_k is not None and resolved_top_k > len(domains):
        raise ValueError(f"top_k={resolved_top_k} exceeds the number of available source domains ({len(domains)}).")
    if resolved_top_k is not None and resolved_top_k < selected_min:
        raise ValueError("top_k must be greater than or equal to min_selected_domains.")

    resolved_max_distance = _normalize_optional_nonnegative_float(max_distance, name="max_distance")
    normalized_metric = normalize_source_selection_metric(metric)
    balance_classes = _normalize_bool(class_balance, name="class_balance")
    distances = {
        domain: _domain_distance(source_matrix[_object_equal_mask(domain_vector, domain)], target_matrix, metric=normalized_metric)
        for domain in domains
    }
    ordered_domains = tuple(sorted(domains, key=lambda domain: (distances[domain], repr(domain))))
    selected = _select_domains(
        ordered_domains,
        distances,
        top_k=resolved_top_k,
        max_distance=resolved_max_distance,
        min_selected_domains=selected_min,
    )
    scores = _distance_scores(distances, temperature=softmax_temperature)
    selected_set = set(selected)
    selected_mask = np.asarray([domain in selected_set for domain in domain_vector.tolist()], dtype=bool)
    sample_weights = np.zeros(source_matrix.shape[0], dtype=float)
    for domain in selected:
        sample_weights[_object_equal_mask(domain_vector, domain)] = scores[domain]
    if balance_classes:
        if source_labels is None:
            raise ValueError("source_labels are required when class_balance=True.")
        sample_weights = _class_balanced_weights(sample_weights, source_labels, selected_mask)
    sample_weights = _normalize_selected_weights(sample_weights, selected_mask)

    metadata = _metadata(
        metric=normalized_metric,
        n_source_rows=source_matrix.shape[0],
        n_target_rows=target_matrix.shape[0],
        feature_dim=source_matrix.shape[1],
        n_source_domains=len(domains),
        selected_domains=selected,
        top_k=resolved_top_k,
        max_distance=resolved_max_distance,
        min_selected_domains=selected_min,
        softmax_temperature=softmax_temperature,
        class_balance=balance_classes,
        distances=distances,
        scores=scores,
    )
    return SourceDomainSelectionResult(
        selected_domains=selected,
        domain_distances=distances,
        domain_scores=scores,
        sample_weights=sample_weights,
        selected_mask=selected_mask,
        metadata=metadata,
    )

selected_source_subset(source_features, source_labels, result)

Return selected source rows, labels, and normalized non-zero weights.

Source code in src/neureptrace/decoding/source_selection.py 
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
def selected_source_subset(
    source_features: Sequence[Sequence[float]] | np.ndarray,
    source_labels: Sequence[Any] | np.ndarray,
    result: SourceDomainSelectionResult,
) -> tuple[np.ndarray, np.ndarray, np.ndarray]:
    """Return selected source rows, labels, and normalized non-zero weights."""

    features = _feature_matrix(source_features, name="source_features")
    labels = _label_vector(source_labels, expected_length=features.shape[0])
    mask = np.asarray(result.selected_mask, dtype=bool)
    if mask.shape[0] != features.shape[0]:
        raise ValueError("result.selected_mask length does not match source_features rows.")
    weights = np.asarray(result.sample_weights, dtype=float).reshape(-1)
    if weights.shape[0] != features.shape[0]:
        raise ValueError("result.sample_weights length does not match source_features rows.")
    return features[mask], labels[mask], weights[mask]

normalize_source_selection_metric(metric)

Normalize public aliases for target-similarity metrics.

Source code in src/neureptrace/decoding/source_selection.py 
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
def normalize_source_selection_metric(metric: str | None) -> str:
    """Normalize public aliases for target-similarity metrics."""

    normalized = DEFAULT_SOURCE_SELECTION_METRIC if metric is None else str(metric).strip().lower().replace("-", "_")
    normalized = {
        "mean_distance": "mean",
        "centroid": "mean",
        "centroid_distance": "mean",
        "cov": "covariance",
        "covariance_distance": "covariance",
        "coral": "covariance",
        "mean_cov": "mean_covariance",
        "mean_plus_covariance": "mean_covariance",
        "mean_coral": "mean_covariance",
        "mmd_rbf": "mmd",
        "rbf_mmd": "mmd",
        "maximum_mean_discrepancy": "mmd",
    }.get(normalized, normalized)
    if normalized not in SOURCE_DOMAIN_SELECTION_METRICS:
        raise ValueError(
            f"Unknown source-selection metric {metric!r}. "
            f"Available metrics: {', '.join(SOURCE_DOMAIN_SELECTION_METRICS)}."
        )
    return normalized