Skip to content

Source balancing

neureptrace.decoding.source_balance implements strict source-only sample weighting and balanced resampling helpers.

The protocol is Category 1 / strict source-only. The helpers use source labels and optional source-domain identifiers only. Held-out target features and labels are not accepted.

Supported strategies:

  • none
  • class
  • domain
  • class_domain

Supported group targets:

  • max
  • min
  • mean

neureptrace.decoding.source_balance

Strict source-only class/domain balancing helpers.

SourceBalanceConfig dataclass

Configuration for source-only sample weighting and resampling.

Source code in src/neureptrace/decoding/source_balance.py 
19
20
21
22
23
24
25
26
@dataclass(frozen=True, slots=True)
class SourceBalanceConfig:
    """Configuration for source-only sample weighting and resampling."""

    strategy: str = "class_domain"
    target: str = "max"
    normalize_weights: bool = True
    random_state: int | None = 13

SourceBalanceResult dataclass

Per-row source weights and group metadata.

Source code in src/neureptrace/decoding/source_balance.py 
29
30
31
32
33
34
35
36
37
@dataclass(frozen=True, slots=True)
class SourceBalanceResult:
    """Per-row source weights and group metadata."""

    sample_weights: np.ndarray
    group_keys: tuple[Hashable, ...]
    group_counts: Mapping[Hashable, int]
    group_target_count: float
    metadata: dict[str, Any] = field(default_factory=dict)

SourceResampleResult dataclass

Balanced source-row resample.

Source code in src/neureptrace/decoding/source_balance.py 
40
41
42
43
44
45
46
47
48
@dataclass(frozen=True, slots=True)
class SourceResampleResult:
    """Balanced source-row resample."""

    features: np.ndarray
    labels: np.ndarray
    domains: np.ndarray | None
    source_indices: np.ndarray
    metadata: dict[str, Any] = field(default_factory=dict)

compute_source_balance_weights(source_labels, *, source_domains=None, config=None)

Compute Protocol-1 source sample weights by class/domain groups.

Source code in src/neureptrace/decoding/source_balance.py 
 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
def compute_source_balance_weights(
    source_labels: Iterable[Any] | np.ndarray,
    *,
    source_domains: Iterable[Hashable] | np.ndarray | None = None,
    config: SourceBalanceConfig | Mapping[str, Any] | None = None,
) -> SourceBalanceResult:
    """Compute Protocol-1 source sample weights by class/domain groups."""

    cfg = source_balance_config() if config is None else _coerce_config(config)
    labels = _vector(source_labels, name="source_labels")
    domains = _domain_vector(source_domains, expected_length=labels.shape[0])
    keys = _group_keys(labels, domains, strategy=cfg.strategy)
    counts = _count_groups(keys)
    target_count = _target_count(counts, target=cfg.target)
    if cfg.strategy == "none":
        weights = np.ones(labels.shape[0], dtype=float)
    else:
        weights = np.asarray([target_count / counts[key] for key in keys], dtype=float)
    if cfg.normalize_weights and weights.size:
        weights = weights / float(np.mean(weights))
    return SourceBalanceResult(
        sample_weights=weights.astype(np.float32, copy=False),
        group_keys=tuple(keys),
        group_counts=counts,
        group_target_count=float(target_count),
        metadata=_metadata(cfg, n_source_rows=labels.shape[0], n_groups=len(counts), group_counts=counts, group_target_count=target_count, n_output_rows=""),
    )

resample_source_rows_balanced(source_features, source_labels, *, source_domains=None, config=None)

Resample source rows to equalize class/domain groups.

Source code in src/neureptrace/decoding/source_balance.py 
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
def resample_source_rows_balanced(
    source_features: Iterable[Iterable[float]] | np.ndarray,
    source_labels: Iterable[Any] | np.ndarray,
    *,
    source_domains: Iterable[Hashable] | np.ndarray | None = None,
    config: SourceBalanceConfig | Mapping[str, Any] | None = None,
) -> SourceResampleResult:
    """Resample source rows to equalize class/domain groups."""

    cfg = source_balance_config() if config is None else _coerce_config(config)
    features = _feature_matrix(source_features, name="source_features")
    labels = _vector(source_labels, name="source_labels", expected_length=features.shape[0])
    if labels.shape[0] != features.shape[0]:
        raise ValueError("source_labels must contain one value per feature row.")
    domains = _domain_vector(source_domains, expected_length=features.shape[0])
    keys = _group_keys(labels, domains, strategy=cfg.strategy)
    counts = _count_groups(keys)
    target_count = int(round(_target_count(counts, target=cfg.target)))
    rng = np.random.default_rng(cfg.random_state)
    if cfg.strategy == "none":
        indices = np.arange(features.shape[0], dtype=int)
    else:
        picked: list[int] = []
        for key in ordered_unique(keys):
            group_indices = np.asarray([index for index, row_key in enumerate(keys) if values_equal(row_key, key)], dtype=int)
            picked.extend(rng.choice(group_indices, size=target_count, replace=group_indices.size < target_count).astype(int).tolist())
        indices = np.asarray(picked, dtype=int)
    out_domains = None if source_domains is None else domains[indices]
    metadata = _metadata(cfg, n_source_rows=features.shape[0], n_groups=len(counts), group_counts=counts, group_target_count=target_count, n_output_rows=indices.shape[0])
    metadata["source_balance_resampled"] = True
    return SourceResampleResult(features=features[indices].astype(np.float32, copy=False), labels=labels[indices], domains=out_domains, source_indices=indices, metadata=metadata)

source_balance_config(*, strategy='class_domain', target='max', normalize_weights=True, random_state=13)

Normalize source-balancing options.

Source code in src/neureptrace/decoding/source_balance.py 
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
def source_balance_config(
    *,
    strategy: str | None = "class_domain",
    target: str | None = "max",
    normalize_weights: bool | str | int | float = True,
    random_state: int | str | None = 13,
) -> SourceBalanceConfig:
    """Normalize source-balancing options."""

    return SourceBalanceConfig(
        strategy=normalize_balance_strategy(strategy),
        target=normalize_balance_target(target),
        normalize_weights=_bool_config(normalize_weights, name="normalize_weights"),
        random_state=_optional_nonnegative_int(random_state, name="random_state"),
    )

normalize_balance_strategy(value)

Normalize balance strategy aliases.

Source code in src/neureptrace/decoding/source_balance.py 
173
174
175
176
177
178
179
180
def normalize_balance_strategy(value: str | None) -> str:
    """Normalize balance strategy aliases."""

    normalized = "class_domain" if value is None else str(value).strip().lower().replace("-", "_")
    normalized = {"off": "none", "label": "class", "labels": "class", "subject": "domain", "source_domain": "domain", "class_subject": "class_domain", "domain_class": "class_domain"}.get(normalized, normalized)
    if normalized not in BALANCE_STRATEGIES:
        raise ValueError(f"Unknown balance strategy {value!r}.")
    return normalized

normalize_balance_target(value)

Normalize group target aliases.

Source code in src/neureptrace/decoding/source_balance.py 
183
184
185
186
187
188
189
190
def normalize_balance_target(value: str | None) -> str:
    """Normalize group target aliases."""

    normalized = "max" if value is None else str(value).strip().lower().replace("-", "_")
    normalized = {"largest": "max", "oversample": "max", "smallest": "min", "undersample": "min", "average": "mean"}.get(normalized, normalized)
    if normalized not in BALANCE_TARGETS:
        raise ValueError(f"Unknown balance target {value!r}.")
    return normalized