如何用Python3从Kaggle训练集生成与测试集分布一致的验证集

当然有简便的方案！核心是让验证集的数据分布（特征、标签/目标变量的分布）尽可能和测试集对齐，这样模型在验证集上的评估结果才会更贴近最终在Kaggle测试集上的表现。下面结合Python 3给你分享几种实用方法：

方法1：基于标签的分层划分（适用于测试集有标签的场景） #

如果你的Kaggle测试集是带标签的（比如部分竞赛会提供公开的带标签测试集），或者原训练集和测试集本身来自同一分布，那直接按标签比例划分是最直接的方式。这种方法能保证验证集的标签分布和测试集（或原训练集）完全一致。

代码示例： #

import pandas as pd
from sklearn.model_selection import train_test_split

# 加载原始数据集
train_df = pd.read_csv("train.csv")
test_df = pd.read_csv("test.csv")

# 情况1：原训练集和测试集分布一致，直接用sklearn的分层划分
train_set, dev_set = train_test_split(
    train_df,
    test_size=0.2,  # 验证集占原训练集的比例，可根据数据量调整（10%-20%常用）
    stratify=train_df["target"],  # 按标签分层，保证分布一致
    random_state=42  # 固定随机种子，结果可复现
)

# 情况2：原训练集和测试集标签分布差异较大（比如测试集正例占30%，训练集只占20%）
# 手动按测试集的标签比例采样验证集
test_label_ratio = test_df["target"].value_counts(normalize=True)
dev_pos_sample = train_df[train_df["target"] == 1].sample(frac=0.2, random_state=42)
# 按测试集比例计算需要抽取的负例数量
dev_neg_count = int(len(dev_pos_sample) / test_label_ratio[1] - len(dev_pos_sample))
dev_neg_sample = train_df[train_df["target"] == 0].sample(n=dev_neg_count, random_state=42)

# 合并并打乱验证集
dev_set = pd.concat([dev_pos_sample, dev_neg_sample]).sample(frac=1, random_state=42)
train_set = train_df.drop(dev_set.index)

方法2：基于特征分布的匹配划分（测试集无标签的通用场景） #

大多数Kaggle测试集是不带标签的，这时候我们需要让验证集的特征分布和测试集对齐。这里用聚类+分层采样的思路，先找到数据的分布簇，再在每个簇里按测试集的比例抽取验证集。

代码示例： #

import pandas as pd
import numpy as np
from sklearn.cluster import KMeans
from sklearn.preprocessing import StandardScaler

# 加载数据
train_df = pd.read_csv("train.csv")
test_df = pd.read_csv("test.csv")

# 提取特征列（排除ID、标签等非特征列）
feature_cols = [col for col in train_df.columns if col not in ["id", "target"]]
X_train = train_df[feature_cols]
X_test = test_df[feature_cols]

# 特征预处理：归一化（KMeans对特征尺度敏感）
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)  # 仅用训练集拟合，避免数据泄露
X_test_scaled = scaler.transform(X_test)

# 合并训练和测试特征，训练聚类模型（找到共同分布簇）
X_combined = np.vstack([X_train_scaled, X_test_scaled])
# 簇数可通过肘部法调整，这里先设为10
kmeans = KMeans(n_clusters=10, random_state=42, n_init="auto")
cluster_labels = kmeans.fit_predict(X_combined)

# 拆分训练集和测试集的簇标签
train_cluster = cluster_labels[:len(X_train)]
test_cluster = cluster_labels[len(X_train):]

# 计算测试集在每个簇的占比
test_cluster_ratio = pd.Series(test_cluster).value_counts(normalize=True)

# 在每个簇中按测试集比例抽取验证集
dev_indices = []
total_dev_size = int(len(train_df) * 0.2)  # 验证集总大小
for cluster_id in test_cluster_ratio.index:
    # 获取当前簇在训练集中的所有索引
    cluster_train_indices = train_df[train_cluster == cluster_id].index
    # 计算当前簇需要抽取的验证集样本数
    cluster_dev_size = int(total_dev_size * test_cluster_ratio[cluster_id])
    # 随机抽取样本
    dev_indices.extend(np.random.choice(cluster_train_indices, size=cluster_dev_size, replace=False))

# 生成最终的训练集和验证集
dev_set = train_df.loc[dev_indices]
train_set = train_df.drop(dev_indices)

额外注意事项 #

• 避免数据泄露：预处理（如归一化、编码）时，只能用训练集的统计量拟合，绝对不能用测试集或合并后的数据集，否则会导致验证集评估结果失真。
• 验证分布一致性：划分后可以用统计方法验证，比如用KS检验对比连续特征的分布，用卡方检验对比离散特征的分布。
• 回归任务适配：如果是回归任务，可以把目标变量分桶（比如分成10个区间），然后按桶做分层划分，和分类任务的逻辑一致。

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