Source bagging decoder

neureptrace.decoding.source_bagging implements a strict source-only bootstrap ensemble for feature decoding.

Each base estimator is trained on a source-only row sample and optional feature subset. Held-out rows are only scored, and probabilities are averaged across estimators.

This is Category 1 / strict source-only.

neureptrace.decoding.source_bagging

Strict source-only bootstrap bagging decoder.

This module provides a dependency-light Protocol-1 ensemble baseline for cross-subject feature decoding. Each base estimator is trained on a bootstrap or subsample of labeled source rows, optionally with a source-only feature subset. Held-out rows are scored by averaging estimator probabilities.

SourceBaggingConfig dataclass

Configuration for the source-only bagging decoder.

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

@dataclass(frozen=True, slots=True)
class SourceBaggingConfig:
    """Configuration for the source-only bagging decoder."""

    n_estimators: int = DEFAULT_N_ESTIMATORS
    sample_fraction: float = DEFAULT_SAMPLE_FRACTION
    feature_fraction: float = DEFAULT_FEATURE_FRACTION
    bootstrap_rows: bool = True
    bootstrap_features: bool = False
    class_balanced: bool = True
    random_state: int | None = 13
    epsilon: float = DEFAULT_EPSILON

SourceBaggingResult dataclass

Bagged predictions and provenance metadata.

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

@dataclass(frozen=True, slots=True)
class SourceBaggingResult:
    """Bagged predictions and provenance metadata."""

    probabilities: np.ndarray
    predictions: np.ndarray
    classes: np.ndarray
    estimators: tuple[BaseEstimator, ...]
    row_indices: tuple[np.ndarray, ...]
    feature_indices: tuple[np.ndarray, ...]
    metadata: dict[str, Any] = field(default_factory=dict)

    @property
    def n_estimators(self) -> int:
        """Number of fitted base estimators."""

        return len(self.estimators)

n_estimators property

Number of fitted base estimators.

fit_source_bagging_decoder(*, source_features, source_labels, test_features, estimator=None, config=None)

Fit a strict source-only bagged classifier and score held-out rows.

Parameters:

Name	Type	Description	Default
source_features	Sequence[Sequence[float]] | ndarray	Labeled source rows used to fit all bootstrap estimators.	required
source_labels	Sequence[Sequence[float]] | ndarray	Labeled source rows used to fit all bootstrap estimators.	required
test_features	Sequence[Sequence[float]] | ndarray	Rows to score. These rows are never used for fitting.	required
estimator	BaseEstimator | None	Optional sklearn-compatible estimator. If omitted, a standardized balanced logistic-regression classifier is used.	None
config	SourceBaggingConfig | Mapping[str, Any] | None	Bagging options. Mappings are normalized through :func:source_bagging_config.	None

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

def fit_source_bagging_decoder(
    *,
    source_features: Sequence[Sequence[float]] | np.ndarray,
    source_labels: Sequence[Any] | np.ndarray,
    test_features: Sequence[Sequence[float]] | np.ndarray,
    estimator: BaseEstimator | None = None,
    config: SourceBaggingConfig | Mapping[str, Any] | None = None,
) -> SourceBaggingResult:
    """Fit a strict source-only bagged classifier and score held-out rows.

    Parameters
    ----------
    source_features, source_labels:
        Labeled source rows used to fit all bootstrap estimators.
    test_features:
        Rows to score.  These rows are never used for fitting.
    estimator:
        Optional sklearn-compatible estimator.  If omitted, a standardized
        balanced logistic-regression classifier is used.
    config:
        Bagging options.  Mappings are normalized through
        :func:`source_bagging_config`.
    """

    cfg = source_bagging_config() if config is None else _coerce_config(config)
    source = _feature_matrix(source_features, name="source_features")
    test = _feature_matrix(test_features, name="test_features")
    if source.shape[1] != test.shape[1]:
        raise ValueError(f"source_features and test_features must have the same feature width: {source.shape[1]} != {test.shape[1]}.")
    labels = _label_vector(source_labels, expected_length=source.shape[0], name="source_labels")
    classes = _unique_labels(labels)
    if classes.shape[0] < 2:
        raise ValueError("At least two source classes are required.")
    label_codes = np.asarray([_label_index(class_label, classes, name="source_labels") for class_label in labels.tolist()], dtype=int)
    encoded_classes = np.arange(classes.shape[0], dtype=int)
    template = _default_estimator(cfg) if estimator is None else estimator
    rng = np.random.default_rng(cfg.random_state)

    probabilities: list[np.ndarray] = []
    estimators: list[BaseEstimator] = []
    row_indices: list[np.ndarray] = []
    feature_indices: list[np.ndarray] = []
    for _ in range(cfg.n_estimators):
        rows = _sample_rows(labels, classes=classes, cfg=cfg, rng=rng)
        feats = _sample_features(source.shape[1], cfg=cfg, rng=rng)
        model = clone(template)
        model.fit(source[rows][:, feats], label_codes[rows])
        probabilities.append(_aligned_probabilities(model, test[:, feats], classes=encoded_classes, epsilon=cfg.epsilon))
        estimators.append(model)
        row_indices.append(rows.astype(int, copy=False))
        feature_indices.append(feats.astype(int, copy=False))

    mean_probabilities = _normalize_probability_rows(np.mean(probabilities, axis=0), epsilon=cfg.epsilon)
    predictions = classes[np.argmax(mean_probabilities, axis=1)]
    metadata = _metadata(cfg, n_source_rows=source.shape[0], n_test_rows=test.shape[0], feature_dim=source.shape[1], classes=classes)
    return SourceBaggingResult(
        probabilities=mean_probabilities.astype(np.float32, copy=False),
        predictions=predictions,
        classes=classes,
        estimators=tuple(estimators),
        row_indices=tuple(row_indices),
        feature_indices=tuple(feature_indices),
        metadata=metadata,
    )

source_bagging_config(*, n_estimators=DEFAULT_N_ESTIMATORS, sample_fraction=DEFAULT_SAMPLE_FRACTION, feature_fraction=DEFAULT_FEATURE_FRACTION, bootstrap_rows=True, bootstrap_features=False, class_balanced=True, random_state=13, epsilon=DEFAULT_EPSILON)

Normalize public source-bagging options.

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

def source_bagging_config(
    *,
    n_estimators: int | str = DEFAULT_N_ESTIMATORS,
    sample_fraction: float | str = DEFAULT_SAMPLE_FRACTION,
    feature_fraction: float | str = DEFAULT_FEATURE_FRACTION,
    bootstrap_rows: bool | str = True,
    bootstrap_features: bool | str = False,
    class_balanced: bool | str = True,
    random_state: int | str | None = 13,
    epsilon: float | str = DEFAULT_EPSILON,
) -> SourceBaggingConfig:
    """Normalize public source-bagging options."""

    return SourceBaggingConfig(
        n_estimators=_positive_int(n_estimators, name="n_estimators"),
        sample_fraction=_positive_fraction(sample_fraction, name="sample_fraction"),
        feature_fraction=_positive_fraction(feature_fraction, name="feature_fraction"),
        bootstrap_rows=_boolean(bootstrap_rows, name="bootstrap_rows"),
        bootstrap_features=_boolean(bootstrap_features, name="bootstrap_features"),
        class_balanced=_boolean(class_balanced, name="class_balanced"),
        random_state=_optional_nonnegative_int(random_state, name="random_state"),
        epsilon=_positive_float(epsilon, name="epsilon"),
    )