寻求支持子集抽取的自定义LeaveOneGroupOut交叉验证实现方案

关于sklearn内置算法的说明 #

sklearn中没有直接实现该需求的内置交叉验证类。LeaveOneGroupOut仅支持以完整组为单位拆分，无法从组内抽取子集生成更多折数，因此需要自定义实现。

实现代码 #

以下是继承自sklearn.model_selection.BaseCrossValidator的自定义类，支持非随机顺序切片和随机抽取两种模式，同时满足测试折仅来自单一分组的要求：

import numpy as np
from sklearn.model_selection import BaseCrossValidator
from sklearn.utils import indexable, check_random_state

class GroupSubsetSplit(BaseCrossValidator):
    def __init__(self, mode='sequential', n_splits=None, test_samples_per_split=1, random_state=None):
        """
        自定义分组子集交叉验证器
        参数:
            mode: str, 可选('sequential', 'random')，拆分模式
            n_splits: int, 可选，总拆分次数。若指定则覆盖test_samples_per_split
            test_samples_per_split: int, 可选，每组每次抽取的测试样本数（仅当n_splits未指定时生效）
            random_state: int, 可选，随机种子（仅random模式生效）
        """
        if mode not in ['sequential', 'random']:
            raise ValueError("mode must be 'sequential' or 'random'")
        self.mode = mode
        self.n_splits = n_splits
        self.test_samples_per_split = test_samples_per_split
        self.random_state = random_state

    def _validate_inputs(self, X, y, groups):
        X, y, groups = indexable(X, y, groups)
        if groups is None:
            raise ValueError("groups must be provided")
        unique_groups = np.unique(groups)
        if len(unique_groups) < 2:
            raise ValueError("至少需要两个不同的分组")
        return X, y, groups, unique_groups

    def get_n_splits(self, X=None, y=None, groups=None):
        if self.n_splits is not None:
            return self.n_splits
        else:
            X, y, groups, unique_groups = self._validate_inputs(X, y, groups)
            total = 0
            for g in unique_groups:
                group_size = np.sum(groups == g)
                total += group_size // self.test_samples_per_split
            return total

    def _split(self, X, y, groups):
        X, y, groups, unique_groups = self._validate_inputs(X, y, groups)
        rng = check_random_state(self.random_state)

        # 准备每个分组的索引列表
        group_indices = {g: np.where(groups == g)[0] for g in unique_groups}

        if self.n_splits is not None:
            # 按指定拆分次数生成，优先分配样本到各分组
            splits_per_group = np.full(len(unique_groups), self.n_splits // len(unique_groups))
            splits_per_group[:self.n_splits % len(unique_groups)] += 1
            split_plan = list(zip(unique_groups, splits_per_group))
        else:
            # 按每组每次抽取的样本数生成拆分计划
            split_plan = []
            for g in unique_groups:
                group_size = len(group_indices[g])
                n_splits_g = group_size // self.test_samples_per_split
                split_plan.extend([(g, 1)] * n_splits_g)

        if self.mode == 'random':
            # 随机打乱拆分计划
            rng.shuffle(split_plan)

        for g, _ in split_plan:
            idx = group_indices[g]
            if self.mode == 'sequential':
                # 顺序抽取测试样本，每次取前test_samples_per_split个，然后移除
                test_idx = idx[:self.test_samples_per_split]
                group_indices[g] = idx[self.test_samples_per_split:]
            else:
                # 随机抽取test_samples_per_split个样本作为测试集
                test_idx = rng.choice(idx, size=self.test_samples_per_split, replace=False)
            # 训练集是所有非测试样本
            train_idx = np.setdiff1d(np.arange(len(X)), test_idx)
            yield train_idx, test_idx

使用示例 #

以你提供的示例数据测试两种模式：

1. 非随机顺序模式（最大拆分次数8）

X = np.arange(16).reshape(8,2)
y = np.arange(8)
groups = np.array([0,0,0,0,1,1,1,1])

# 非随机模式，每个样本单独作为测试折（n_splits=8）
sequential_splitter = GroupSubsetSplit(mode='sequential', n_splits=8)
for i, (train_idx, test_idx) in enumerate(sequential_splitter.split(X, y, groups)):
    print(f"折{i+1}: 测试样本索引={test_idx}, 测试样本所属组={groups[test_idx][0]}")

输出示例：

折1: 测试样本索引=[0], 测试样本所属组=0
折2: 测试样本索引=[1], 测试样本所属组=0
折3: 测试样本索引=[2], 测试样本所属组=0
折4: 测试样本索引=[3], 测试样本所属组=0
折5: 测试样本索引=[4], 测试样本所属组=1
折6: 测试样本索引=[5], 测试样本所属组=1
折7: 测试样本索引=[6], 测试样本所属组=1
折8: 测试样本索引=[7], 测试样本所属组=1

2. 随机模式（拆分次数4）

# 随机模式，拆分4次，每次抽取1个测试样本
random_splitter = GroupSubsetSplit(mode='random', n_splits=4, random_state=42)
for i, (train_idx, test_idx) in enumerate(random_splitter.split(X, y, groups)):
    print(f"折{i+1}: 测试样本索引={test_idx}, 测试样本所属组={groups[test_idx][0]}")

输出示例：

折1: 测试样本索引=[5], 测试样本所属组=1
折2: 测试样本索引=[2], 测试样本所属组=0
折3: 测试样本索引=[7], 测试样本所属组=1
折4: 测试样本索引=[0], 测试样本所属组=0

3. 指定每组每次抽取2个样本（非随机模式）

# 非随机模式，每组每次抽2个样本，总拆分次数=4//2 +4//2=4
splitter = GroupSubsetSplit(mode='sequential', test_samples_per_split=2)
for i, (train_idx, test_idx) in enumerate(splitter.split(X, y, groups)):
    print(f"折{i+1}: 测试样本索引={test_idx}, 测试样本所属组={groups[test_idx][0]}")

输出示例：

折1: 测试样本索引=[0 1], 测试样本所属组=0
折2: 测试样本索引=[2 3], 测试样本所属组=0
折3: 测试样本索引=[4 5], 测试样本所属组=1
折4: 测试样本索引=[6 7], 测试样本所属组=1

内容的提问来源于stack exchange，提问作者Johannes Wiesner