如何实现自定义留一法交叉验证？适配scikit-learn Pipeline与RandomizedSearchCV

要实现你需要的这种定制化交叉验证策略，我们可以通过继承scikit-learn的BaseCrossValidator类来创建自己的交叉验证迭代器，这样就能无缝搭配Pipeline和RandomizedSearchCV使用。

核心思路 #

你的需求本质是按用户内的样本顺序进行留一验证：每个测试集是单个数据点，训练集的构成取决于该数据点在所属用户样本序列中的位置：

• 若测试点是用户的第一个数据点：训练集排除该用户的所有数据
• 若测试点是用户的第n个数据点（n>1）：训练集包含其他所有用户数据 + 该用户的前n-1个数据点

自定义交叉验证类实现 #

import numpy as np
from sklearn.model_selection import BaseCrossValidator

class OrderedUserLeaveOneOut(BaseCrossValidator):
    def __init__(self):
        pass
    
    def get_n_splits(self, X=None, y=None, groups=None):
        # 返回总拆分次数，等于样本总数
        return len(groups)
    
    def split(self, X, y=None, groups=None):
        # 按用户分组，获取每个用户对应的样本索引列表
        # 确保每个用户的索引保持原始数据中的顺序
        unique_users = np.unique(groups)
        user_indices = {user: np.where(groups == user)[0] for user in unique_users}
        
        # 遍历每个用户的每个样本
        for user in unique_users:
            indices = user_indices[user]
            for idx_in_user, test_idx in enumerate(indices):
                # 构建训练集索引
                # 1. 先收集所有其他用户的索引
                train_indices = []
                for other_user in unique_users:
                    if other_user != user:
                        train_indices.extend(user_indices[other_user])
                
                # 2. 如果不是当前用户的第一个样本，添加该用户的前idx_in_user个样本
                if idx_in_user > 0:
                    train_indices.extend(indices[:idx_in_user])
                
                # 转换为numpy数组并返回
                train_indices = np.array(train_indices)
                test_indices = np.array([test_idx])
                
                yield train_indices, test_indices

代码解释 #

1. get_n_splits方法：返回交叉验证的总拆分次数，这里等于样本总数（因为每个样本都要单独作为一次测试集）。
2. split方法：
   • 首先按用户分组，把每个用户对应的所有样本索引整理出来，严格保留原始数据中的顺序（这样才能准确判断样本是用户的第几个数据点）。
   • 逐个遍历每个用户的每个样本：
     • 先收集所有其他用户的样本索引作为训练集基础。
     • 如果当前测试样本不是该用户的第一个样本，就把该用户之前的所有样本加入训练集。
     • 生成训练集和测试集的索引对，测试集始终是单个样本。

与Pipeline和RandomizedSearchCV结合使用示例 #

from sklearn.pipeline import Pipeline
from sklearn.svm import SVC
from sklearn.model_selection import RandomizedSearchCV
import numpy as np

# 模拟你的数据：1000个样本，80个用户，每个用户约12个样本
X = np.random.rand(1000, 10)  # 特征矩阵
y = np.random.randint(0, 2, 1000)  # 标签
groups = np.repeat(np.arange(80), 12)[:1000]  # 用户标识，前79个用户12个样本，最后一个用户52个样本

# 创建Pipeline
pipe = Pipeline([
    ('svm', SVC())
])

# 定义超参数搜索空间
param_dist = {
    'svm__C': np.logspace(-3, 3, 7),
    'svm__gamma': np.logspace(-3, 3, 7),
    'svm__kernel': ['linear', 'rbf']
}

# 使用自定义交叉验证器
cv = OrderedUserLeaveOneOut()

# 初始化RandomizedSearchCV
random_search = RandomizedSearchCV(
    estimator=pipe,
    param_distributions=param_dist,
    n_iter=20,
    cv=cv,
    n_jobs=-1,
    verbose=2
)

# 执行搜索
random_search.fit(X, y, groups=groups)

# 查看最佳参数
print("Best parameters found: ", random_search.best_params_)

注意事项 #

• 确保你的groups参数正确对应每个样本的用户标识，且每个用户的样本在原始数据中的顺序就是你期望的"第1个到第n个"的顺序（我们是按原始索引顺序来判断样本在用户内的位置）。
• 由于这种交叉验证会生成1000个拆分，计算量会比LeaveOneGroupOut大很多，建议使用n_jobs=-1利用多核加速，或者根据实际情况调整RandomizedSearchCV的n_iter参数减少搜索次数。

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