Cross-subject transfer learning via CrossSubjectEvaluation (calibration_size)

moabb/evaluations/base.py

@@ -12,9 +12,11 @@
 import numpy as np
 import pandas as pd
 from joblib import Parallel, delayed
+from sklearn import config_context
 from sklearn.base import BaseEstimator, clone
 from sklearn.model_selection import StratifiedKFold
 from sklearn.preprocessing import LabelEncoder
+from sklearn.utils.metadata_routing import get_routing_for_object
 
 from moabb.analysis import Results
 from moabb.datasets.base import (  # noqa: F401 - CacheConfig used in type hints
@@ -44,6 +46,25 @@
 
 log = logging.getLogger(__name__)
 
+
+def _route_transfer_metadata(estimator, subjects, calib=None):
+    """Keep only the transfer metadata an estimator requests at ``fit``.
+
+    ``subjects`` is the per-trial source-subject array; ``calib`` is an optional
+    dict carrying the (raw) calibration slice, e.g.
+    ``{"X_target_unlabeled": X[calib_idx]}``. Steps opt in via
+    ``set_fit_request(...)``; plain pipelines request nothing, so this returns
+    ``{}`` and the fit is unchanged. The framework passes the slice raw -- the
+    estimator owns the target representation.
+    """
+    candidate = {"subjects": subjects}
+    if calib:
+        candidate.update(calib)
+    with config_context(enable_metadata_routing=True):
+        kept = get_routing_for_object(estimator).consumes("fit", set(candidate))
+    return {k: v for k, v in candidate.items() if k in kept}
+
+
 # Making the optuna soft dependency
 
 
@@ -100,6 +121,7 @@ def _evaluate_fold(
     session,
     cv_ind,
     split_metadata=None,
+    calib_idx=None,
 ):
     """Evaluate a single CV fold. Pure function, no shared mutable state.
 
@@ -174,14 +196,30 @@ def _evaluate_fold(
         tracker = emissions_obj.create_tracker()
         tracker.start()
 
+    # Optional transfer-learning calibration slice (raw). Offer it under every
+    # name; _route_transfer_metadata keeps only what the estimator requests, so
+    # the estimator's set_fit_request decides labeled vs unlabeled. Empty for
+    # ordinary evaluations -> plain fit.
+    calib = None
+    if calib_idx is not None and len(calib_idx):
+        y_calib = y[calib_idx] if mne_labels else le.transform(y[calib_idx])
+        calib = {
+            "X_target_unlabeled": X[calib_idx],
+            "X_target_labeled": X[calib_idx],
+            "y_target_labeled": y_calib,
+        }
+    subjects_train = metadata["subject"].to_numpy()[train_idx]
+    fit_params = _route_transfer_metadata(cvclf, subjects_train, calib)
+
     # Fit model
     task_name = None
     emissions = math.nan
     if tracker is not None:
         task_name = str(uuid4())
         tracker.start_task(task_name)
     t_start = perf_counter()
-    cvclf.fit(X[train_idx], y_train)
+    with config_context(enable_metadata_routing=True):
+        cvclf.fit(X[train_idx], y_train, **fit_params)
     duration = perf_counter() - t_start
     if tracker is not None:
         emissions_data = tracker.stop_task()
@@ -463,13 +501,14 @@ def __init__(
         )
 
     def _resolve_cv(self, default_class, default_kwargs=None):
-        """Resolve the cross-validation class and kwargs for a splitter."""
-        if self.cv_class is None:
-            cv_class = default_class
-            cv_kwargs = {} if default_kwargs is None else dict(default_kwargs)
-        else:
-            cv_class = self.cv_class
-            cv_kwargs = dict(self.cv_kwargs)
+        """Resolve the cross-validation class and kwargs for a splitter.
+
+        ``self.cv_kwargs`` always overrides the defaults, whether or not a custom
+        ``cv_class`` is set -- so splitter options (e.g. ``calibration_size``)
+        passed via ``cv_kwargs`` are honored with the default ``cv_class`` too.
+        """
+        cv_class = default_class if self.cv_class is None else self.cv_class
+        cv_kwargs = {**(default_kwargs or {}), **self.cv_kwargs}
         return cv_class, cv_kwargs
 
     def _load_data(
@@ -565,15 +604,16 @@ def _build_scored_result(
             res["score"] = self.error_score
             return res
 
-    def _fit_cv(self, model, X_train, y_train, tracker=None):
+    def _fit_cv(self, model, X_train, y_train, tracker=None, fit_params=None):
         """Fit a model for a CV fold with optional CodeCarbon tracking."""
         task_name = None
         emissions = math.nan
         if tracker is not None:
             task_name = str(uuid4())
             tracker.start_task(task_name)
         t_start = perf_counter()
-        model.fit(X_train, y_train)
+        with config_context(enable_metadata_routing=True):
+            model.fit(X_train, y_train, **(fit_params or {}))
         duration = perf_counter() - t_start
         if tracker is not None:
             emissions_data = tracker.stop_task()
@@ -674,13 +714,16 @@ def _build_eval_config(self, param_grid):
     def _preview_splits(splitter, y, metadata):
         """Materialize folds up front with optional splitter metadata."""
         preview = []
-        for cv_ind, (train_idx, test_idx) in enumerate(splitter.split(y, metadata)):
+        # ``*cal`` absorbs the optional calibration slice from a transfer
+        # splitter; a plain 2-tuple splitter gives cal == [] (no calibration).
+        for cv_ind, (train_idx, *cal, test_idx) in enumerate(splitter.split(y, metadata)):
+            calib_idx = cal[0] if cal else train_idx[:0]
             split_metadata = None
             if hasattr(splitter, "get_metadata"):
                 split_metadata = splitter.get_metadata()
                 if split_metadata is not None:
                     split_metadata = dict(split_metadata)
-            preview.append((cv_ind, train_idx, test_idx, split_metadata))
+            preview.append((cv_ind, train_idx, calib_idx, test_idx, split_metadata))
         return preview
 
     def _build_task_list(
@@ -691,7 +734,7 @@ def _build_task_list(
         config = self._build_eval_config(param_grid)
         fold_preview = self._preview_splits(splitter, y, metadata)
 
-        for cv_ind, train_idx, test_idx, split_meta in fold_preview:
+        for cv_ind, train_idx, calib_idx, test_idx, split_meta in fold_preview:
             test_meta = metadata.iloc[test_idx]
             subject = test_meta["subject"].iloc[0]
 
@@ -719,6 +762,7 @@ def _build_task_list(
                         "session": session,
                         "cv_ind": cv_ind,
                         "split_metadata": split_meta,
+                        "calib_idx": calib_idx,
                     }
                 )
         return tasks

moabb/evaluations/evaluations.py

@@ -7,7 +7,7 @@
 from sklearn.preprocessing import LabelEncoder
 from tqdm import tqdm
 
-from moabb.evaluations.base import BaseEvaluation
+from moabb.evaluations.base import BaseEvaluation, _route_transfer_metadata
 from moabb.evaluations.splitters import (
     CrossSessionSplitter,
     CrossSubjectSplitter,
@@ -455,6 +455,18 @@ class CrossSubjectEvaluation(BaseEvaluation):
     n_splits : int or None
         Number of splits for cross-validation. If None, the number of splits
         is equal to the number of subjects. Defaults to ``None``.
+    cv_class : type or None
+        Cross-validation strategy used to hold out subjects (e.g.
+        ``LeaveOneGroupOut``, ``GroupShuffleSplit``, ``GroupKFold``). Defaults to
+        ``None`` (``LeaveOneGroupOut``, or ``GroupKFold`` when ``n_splits`` is set).
+    cv_kwargs : dict
+        Keyword arguments for ``cv_class``. ``calibration_size`` (float in
+        ``[0, 1]``, default ``0.0``) enables transfer learning: when ``> 0`` each
+        fold becomes ``(train, calibration, test)`` and the held-out calibration
+        slice is routed (raw) to the pipeline steps that request it via
+        ``set_fit_request`` -- ``subjects`` and ``X_target_unlabeled`` /
+        ``X_target_labeled`` / ``y_target_labeled`` (the estimator's requests
+        decide labeled vs unlabeled). Steps that request nothing are unaffected.
 
     Notes
     -----
@@ -467,7 +479,12 @@ class CrossSubjectEvaluation(BaseEvaluation):
     _needs_all_subjects = True
 
     def _create_splitter(self):
-        """Create the CrossSubjectSplitter for parallel evaluation."""
+        """Create the CrossSubjectSplitter for parallel evaluation.
+
+        ``calibration_size`` passed via ``cv_kwargs`` turns each fold into a
+        ``(train, calibration, test)`` transfer split (see
+        :class:`~moabb.evaluations.splitters.CrossSubjectSplitter`).
+        """
         if self.n_splits is None:
             default_class = LeaveOneGroupOut
             default_kwargs = {}
@@ -533,13 +550,15 @@ def evaluate(
             tracker.start()
 
         # Progressbar at subject level
-        for cv_ind, (train, test) in enumerate(
+        # ``*cal`` absorbs the optional calibration slice from a transfer split.
+        for cv_ind, (train, *cal, test) in enumerate(
             tqdm(
                 self.cv.split(y, metadata),
                 total=n_subjects,
                 desc=f"{dataset.code}-CrossSubject",
             )
         ):
+            calib = cal[0] if cal else train[:0]
             subject = groups[test[0]]
             # now we can check if this subject has results
             run_pipes = self.results.not_yet_computed(
@@ -552,8 +571,21 @@ def evaluate(
                 )
                 cvclf = clone(clf)
 
+                calib_md = None
+                if len(calib):
+                    calib_md = {
+                        "X_target_unlabeled": X[calib],
+                        "X_target_labeled": X[calib],
+                        "y_target_labeled": y[calib],
+                    }
+                fit_params = _route_transfer_metadata(cvclf, groups[train], calib_md)
+
                 duration, emissions, task_name = self._fit_cv(
-                    cvclf, X[train], y[train], tracker if _carbonfootprint else None
+                    cvclf,
+                    X[train],
+                    y[train],
+                    tracker if _carbonfootprint else None,
+                    fit_params=fit_params,
                 )
                 self._maybe_save_model_cv(
                     cvclf, dataset, subject, "", name, cv_ind, eval_type="CrossSubject"

moabb/evaluations/splitters.py

@@ -507,6 +507,13 @@ class CrossSubjectSplitter(BaseCrossValidator):
         Cross-validation strategy for splitting the subjects between train and test sets.
         By default, use LeaveOneGroupOut, which keeps one subject as a test.
         Defaults to ``LeaveOneGroupOut``.
+    calibration_size : float
+        Transfer-learning calibration, in ``[0, 1]``. ``0.0`` (default) is the
+        ordinary cross-subject split, yielding ``(train, test)``. When ``> 0``,
+        the first ``calibration_size`` fraction of each held-out subject's trials
+        is set aside for adaptation and the splitter yields
+        ``(train, calibration, test)`` instead; ``1.0`` is transductive
+        (calibration equals test).
     random_state : int or None
         Controls the randomness of the cross-validation.
         Pass an int for reproducible output across multiple calls.
@@ -519,17 +526,26 @@ class CrossSubjectSplitter(BaseCrossValidator):
     train : ndarray
         The training set indices for that split.
 
+    calibration : ndarray
+        The held-out calibration indices. Only yielded when ``calibration_size > 0``.
+
     test : ndarray
         The testing set indices for that split.
     """
 
     def __init__(
         self,
         cv_class: type[BaseCrossValidator] = LeaveOneGroupOut,
+        calibration_size: float = 0.0,
         random_state: int = None,
         **cv_kwargs,
     ):
+        if not 0.0 <= calibration_size <= 1.0:
+            raise ValueError(
+                f"calibration_size must be in [0, 1]. Got {calibration_size!r}."
+            )
         self.cv_class = cv_class
+        self.calibration_size = calibration_size
         self.cv_kwargs = cv_kwargs
         self._cv_kwargs = dict(**cv_kwargs)
 
@@ -584,13 +600,50 @@ def split(self, y, metadata):
 
         for train_session_idx, test_session_idx in splitter.split(**split_kwargs):
             self._last_split_metadata = _splitter_metadata(splitter)
-            yield all_index[train_session_idx], all_index[test_session_idx]
+            train_idx = all_index[train_session_idx]
+            group_idx = all_index[test_session_idx]
+            if self.calibration_size > 0:
+                # Transfer learning: carve a calibration slice off the held-out
+                # subject -> (train, calibration, test).
+                calib_idx, test_idx = _split_target_fraction(
+                    group_idx, self.calibration_size
+                )
+                yield train_idx, calib_idx, test_idx
+            else:
+                yield train_idx, group_idx
 
     def get_metadata(self):
         """Return metadata for the most recent split."""
         return self._last_split_metadata
 
 
+def _split_target_fraction(
+    target_idx: np.ndarray, calibration_size: float
+) -> tuple[np.ndarray, np.ndarray]:
+    """Slice a held-out target subject's trials into (calibration, test).
+
+    The first ``calibration_size`` fraction of ``target_idx`` (in trial order) is
+    the calibration/adaptation slice; the rest is evaluated. ``0.0`` yields no
+    calibration; ``1.0`` is transductive (calibration and test are identical).
+    For any fraction strictly between 0 and 1, at least one trial is kept on each
+    side.
+    """
+    target_idx = np.asarray(target_idx, dtype=int)
+    if len(target_idx) == 0:
+        raise ValueError("Empty held-out target index.")
+    if not 0.0 <= calibration_size <= 1.0:
+        raise ValueError(f"calibration_size must be in [0, 1]. Got {calibration_size!r}.")
+
+    if calibration_size == 0.0:
+        return np.array([], dtype=int), target_idx
+    if calibration_size == 1.0:
+        return target_idx, target_idx
+
+    n_calib = int(round(calibration_size * len(target_idx)))
+    n_calib = min(max(n_calib, 1), len(target_idx) - 1)
+    return target_idx[:n_calib], target_idx[n_calib:]
+
+
 class CrossDatasetSplitter(BaseCrossValidator):
     """Data splitter for leave-dataset-out style evaluation.