FIX keep CrossSubject results subject-specific for grouped folds

docs/source/whats_new.rst

@@ -63,6 +63,7 @@ Requirements
 
 Bugs
 ~~~~
+- Keep :class:`moabb.evaluations.CrossSubjectEvaluation` result provenance subject-specific when a grouped cross-validation fold holds out multiple subjects at once. The estimator is still fitted once per fold, while scores, cache identities, and saved-model paths are emitted per held-out subject and session instead of assigning the whole fold to its first subject (by `lindicaphxag-tech`_).
 - Fix :func:`moabb.datasets.Dataset.convert_to_bids` crashing on datasets whose
   MOABB run-label suffix is literally ``"calibration"`` or ``"crosstalk"``
   (Wang2025, Leeuwis2021, Brandl2020, Romani_BF2025_ERP). After the ``acq-``

moabb/evaluations/base.py

@@ -139,6 +139,7 @@ def _evaluate_fold(
     cv_ind,
     split_metadata=None,
     calib_idx=None,
+    score_subjects=None,
 ):
     """Evaluate a single CV fold. Pure function, no shared mutable state.
 
@@ -166,6 +167,10 @@ def _evaluate_fold(
         Cross-validation fold index.
     split_metadata : dict | None
         Extra metadata from the splitter.
+    score_subjects : sequence | None
+        Held-out subjects whose results are still missing for this pipeline.
+        When provided, the fold is fitted once and scored separately for these
+        subjects.
 
     Returns
     -------
@@ -177,6 +182,7 @@ def _evaluate_fold(
     param_grid = config["param_grid"]
     additional_columns = config["additional_columns"]
     score_per_session = config["score_per_session"]
+    score_per_subject = config["score_per_subject"]
     mne_labels = config["mne_labels"]
     codecarbon_config = config["codecarbon_config"]
     trialwise = config.get("trialwise", False)
@@ -257,37 +263,66 @@ def _evaluate_fold(
     if tracker is not None:
         tracker.stop()
 
-    # Optionally save model
+    # Optionally save model. A multi-subject cross-subject fold still trains
+    # exactly one model, but each held-out subject owns its result/cache path.
     hdf5_path = config["hdf5_path"]
     eval_type = config["eval_type"]
+    result_subjects = (
+        list(score_subjects)
+        if score_per_subject and score_subjects is not None
+        else [subject]
+    )
     if hdf5_path is not None and config["save_model"]:
-        model_save_path = _create_save_path(
-            hdf5_path=hdf5_path,
-            code=dataset.code,
-            subject=subject,
-            session="" if score_per_session else session,
-            name=pipeline_name,
-            grid=is_search,
-            eval_type=eval_type,
-            paradigm=config["paradigm"],
-            suffix=config["suffix"],
-        )
-        _save_model_cv(model=cvclf, save_path=model_save_path, cv_index=str(cv_ind))
+        for save_subject in result_subjects:
+            model_save_path = _create_save_path(
+                hdf5_path=hdf5_path,
+                code=dataset.code,
+                subject=save_subject,
+                session="" if score_per_session else session,
+                name=pipeline_name,
+                grid=is_search,
+                eval_type=eval_type,
+                paradigm=config["paradigm"],
+                suffix=config["suffix"],
+            )
+            _save_model_cv(model=cvclf, save_path=model_save_path, cv_index=str(cv_ind))
 
     scorer = None if trialwise else _create_scorer(cvclf, scoring)
 
-    # Build score groups: per-session or full test set
-    if score_per_session:
-        test_sessions = metadata.iloc[test_idx]["session"].values
-        score_groups = [
-            (test_idx[test_sessions == s], y_test[test_sessions == s], s)
-            for s in np.unique(test_sessions)
-        ]
+    # Build score groups. Cross-subject folds can contain multiple held-out
+    # subjects when using GroupKFold or another custom grouped splitter.
+    test_meta = metadata.iloc[test_idx]
+    test_sessions = test_meta["session"].to_numpy()
+    test_subjects = test_meta["subject"].to_numpy()
+    score_groups = []
+
+    if score_per_subject:
+        selected_subjects = (
+            np.asarray(score_subjects)
+            if score_subjects is not None
+            else np.unique(test_subjects)
+        )
+        for group_subject in selected_subjects:
+            subject_mask = test_subjects == group_subject
+            if score_per_session:
+                for group_session in np.unique(test_sessions[subject_mask]):
+                    mask = subject_mask & (test_sessions == group_session)
+                    score_groups.append(
+                        (test_idx[mask], y_test[mask], group_subject, group_session)
+                    )
+            else:
+                score_groups.append(
+                    (test_idx[subject_mask], y_test[subject_mask], group_subject, session)
+                )
+    elif score_per_session:
+        for group_session in np.unique(test_sessions):
+            mask = test_sessions == group_session
+            score_groups.append((test_idx[mask], y_test[mask], subject, group_session))
     else:
-        score_groups = [(test_idx, y_test, session)]
+        score_groups = [(test_idx, y_test, subject, session)]
 
     results = []
-    for group_idx, group_y, group_session in score_groups:
+    for group_idx, group_y, group_subject, group_session in score_groups:
         is_error = False
         try:
             if trialwise:
@@ -303,7 +338,7 @@ def _evaluate_fold(
         res = {
             "time": duration,
             "dataset": dataset,
-            "subject": subject,
+            "subject": group_subject,
             "session": group_session,
             "n_samples": len(train_idx),
             "n_samples_test": len(group_y),
@@ -416,6 +451,7 @@ class BaseEvaluation(ABC):
     search = False
     _eval_type = None
     _score_per_session = False
+    _score_per_subject = False
     _needs_all_subjects = False
     _aggregate_folds = False
 
@@ -759,6 +795,7 @@ def _build_eval_config(self, param_grid):
                 self.emissions.codecarbon_config if _carbonfootprint else None
             ),
             "score_per_session": self._score_per_session,
+            "score_per_subject": self._score_per_subject,
             "trialwise": getattr(self, "trialwise", False),
             "param_grid": None,  # overridden per-task below if needed
         }
@@ -814,14 +851,32 @@ def _build_task_list(
         for cv_ind, train_idx, calib_idx, test_idx, split_meta in fold_preview:
             test_meta = metadata.iloc[test_idx]
             self._validate_test_fold_metadata(test_meta)
-            subject = test_meta["subject"].iloc[0]
-
-            if subject not in work_plan:
-                continue
-            run_pipes = work_plan[subject]
             session = test_meta["session"].iloc[0]
 
-            for name, clf in run_pipes.items():
+            if self._score_per_subject:
+                test_subjects = np.unique(test_meta["subject"].to_numpy())
+                pipeline_subjects = {
+                    name: [
+                        subject
+                        for subject in test_subjects
+                        if name in work_plan.get(subject, {})
+                    ]
+                    for name in pipelines
+                }
+                task_specs = [
+                    (name, work_plan[subjects[0]][name], subjects[0], subjects)
+                    for name, subjects in pipeline_subjects.items()
+                    if subjects
+                ]
+            else:
+                subject = test_meta["subject"].iloc[0]
+                if subject not in work_plan:
+                    continue
+                task_specs = [
+                    (name, clf, subject, None) for name, clf in work_plan[subject].items()
+                ]
+
+            for name, clf, subject, score_subjects in task_specs:
                 task_config = dict(config)
                 if param_grid is not None and name in param_grid:
                     task_param_grid = {name: deepcopy(param_grid[name])}
@@ -841,6 +896,7 @@ def _build_task_list(
                         "cv_ind": cv_ind,
                         "split_metadata": split_meta,
                         "calib_idx": calib_idx,
+                        "score_subjects": score_subjects,
                     }
                 )
         return tasks

moabb/evaluations/evaluations.py

@@ -523,6 +523,7 @@ class CrossSubjectEvaluation(BaseEvaluation):
 
     _eval_type = "CrossSubject"
     _score_per_session = True
+    _score_per_subject = True
     _needs_all_subjects = True
 
     def __init__(

moabb/tests/test_evaluations.py

@@ -1785,6 +1785,151 @@ def test_cross_subject_process_structure(self, tmp_path):
         # 2 subjects × 2 sessions = 4 results
         assert len(results) == 4
 
+    def test_cross_subject_multisubject_fold_preserves_provenance(self, tmp_path):
+        """A fold holding out multiple subjects is scored per subject/session."""
+        ds = FakeDataset(["left_hand", "right_hand"], n_subjects=4, n_sessions=2, seed=23)
+        paradigm = FakeImageryParadigm()
+        evaluation = ev.CrossSubjectEvaluation(
+            paradigm=paradigm,
+            datasets=[ds],
+            n_splits=2,
+            overwrite=True,
+            hdf5_path=str(tmp_path / "cross_subject_multisubject"),
+            save_model=True,
+        )
+
+        results = evaluation.process(pipelines)
+
+        assert len(results) == 8
+        assert set(results["subject"]) == {str(subject) for subject in ds.subject_list}
+        assert (results.groupby(["subject", "session"], observed=True).size() == 1).all()
+
+        _, _, metadata = paradigm.get_data(ds)
+        metadata = metadata.copy()
+        metadata["subject"] = metadata["subject"].astype(str)
+        expected_sizes = metadata.groupby(["subject", "session"], observed=True).size()
+        actual_sizes = results.set_index(["subject", "session"])["samples_test"]
+        actual_sizes = actual_sizes.reindex(expected_sizes.index)
+        np.testing.assert_array_equal(actual_sizes.to_numpy(), expected_sizes.to_numpy())
+
+        for subject in ds.subject_list:
+            model_dir = (
+                tmp_path
+                / "cross_subject_multisubject"
+                / "Models_CrossSubject"
+                / type(paradigm).__name__
+                / ds.code
+                / str(subject)
+                / "C"
+            )
+            assert list(model_dir.glob("fitted_model_*.pkl"))
+
+    def test_cross_subject_multisubject_fold_respects_partial_work_plan(self, tmp_path):
+        """A cached first subject must not suppress work for another held-out subject."""
+        ds = FakeDataset(["left_hand", "right_hand"], n_subjects=4, n_sessions=2, seed=42)
+        evaluation = ev.CrossSubjectEvaluation(
+            paradigm=FakeImageryParadigm(),
+            datasets=[ds],
+            hdf5_path=str(tmp_path / "cross_subject_partial_cache"),
+            n_splits=2,
+        )
+        metadata = pd.DataFrame(
+            {"subject": np.repeat([1, 2, 3, 4], 2), "session": ["0", "1"] * 4}
+        )
+        y = np.array([0, 1] * 4)
+        splitter = evaluation._create_splitter()
+        folds = list(splitter.split(y, metadata))
+        held_out = list(pd.unique(metadata.iloc[folds[0][1]]["subject"]))
+        assert len(held_out) == 2
+
+        target_subject = held_out[1]
+        pipeline = Dummy(strategy="most_frequent")
+        work_plan = {target_subject: {"dummy": pipeline}}
+        tasks = evaluation._build_task_list(
+            ds,
+            None,
+            y,
+            metadata,
+            evaluation._create_splitter(),
+            work_plan,
+            {"dummy": pipeline},
+            None,
+        )
+
+        assert len(tasks) == 1
+        assert tasks[0]["subject"] == target_subject
+        assert tasks[0]["score_subjects"] == [target_subject]
+        assert target_subject in set(metadata.iloc[tasks[0]["test_idx"]]["subject"])
+
+    def test_cross_subject_partial_cache_fits_remaining_fold_once(self, tmp_path):
+        """A partially cached grouped fold is fitted once for the missing subject."""
+
+        class CountingDummy(Dummy):
+            fit_calls = 0
+
+            def fit(self, X, y, sample_weight=None):
+                type(self).fit_calls += 1
+                return super().fit(X, y, sample_weight=sample_weight)
+
+        ds = FakeDataset(["left_hand", "right_hand"], n_subjects=4, n_sessions=2, seed=42)
+        paradigm = FakeImageryParadigm()
+        evaluation = ev.CrossSubjectEvaluation(
+            paradigm=paradigm,
+            datasets=[ds],
+            hdf5_path=str(tmp_path / "cross_subject_cached_fold"),
+            n_splits=2,
+        )
+        pipeline = CountingDummy(strategy="most_frequent")
+        pipeline_dict = {"counting": pipeline}
+
+        X, y, metadata = paradigm.get_data(ds)
+        folds = list(evaluation._create_splitter().split(y, metadata))
+        held_out_by_fold = [
+            list(pd.unique(metadata.iloc[test_idx]["subject"])) for _, test_idx in folds
+        ]
+        assert all(len(subjects) == 2 for subjects in held_out_by_fold)
+
+        cached_target = held_out_by_fold[0][0]
+        missing_target = held_out_by_fold[0][1]
+        cached_subjects = [cached_target, *held_out_by_fold[1]]
+
+        process_pipeline = paradigm.make_process_pipelines(ds)[0]
+        for cached_subject in cached_subjects:
+            cached_result = {
+                "score": 0.0,
+                "time": 0.0,
+                "dataset": ds,
+                "subject": cached_subject,
+                "session": "cached",
+                "n_samples": 1,
+                "n_channels": X.shape[1],
+                "carbon_emission": 0.0,
+            }
+            evaluation.results.add(
+                {"counting": cached_result},
+                pipelines=pipeline_dict,
+                process_pipeline=process_pipeline,
+            )
+
+        before = evaluation.results.to_dataframe(
+            pipelines=pipeline_dict, process_pipeline=process_pipeline
+        )
+        before_counts = before.groupby("subject", observed=True).size().to_dict()
+
+        CountingDummy.fit_calls = 0
+        after = evaluation.process(pipeline_dict)
+
+        assert CountingDummy.fit_calls == 1
+        after_counts = after.groupby("subject", observed=True).size().to_dict()
+        increased_subjects = {
+            subject
+            for subject, count in after_counts.items()
+            if count > before_counts.get(subject, 0)
+        }
+        assert increased_subjects == {str(missing_target)}
+        assert after_counts[str(cached_target)] == before_counts[str(cached_target)]
+        assert after_counts[str(missing_target)] == ds.n_sessions
+
     def test_learning_curve_parallel(self, tmp_path):
         """LearningCurve evaluation via parallel process()."""
         evaluation = ev.WithinSessionEvaluation(