如何在sklearn pipeline分类步骤前对目标列y进行预处理分箱

报错的根因是make_column_transformer默认仅对输入特征X做转换，你调用pipeline.fit()时传入的X仅包含特征列，不存在target字段，因此触发键值错误。另外标准Sklearn Pipeline的原生设计中，预处理步骤默认只能修改特征X，无法修改目标变量y，TransformedTargetRegressor仅适配回归场景，确实无法直接用于分类任务。

可行实现方案 #

我们可以自定义一个兼容Sklearn API的分类目标转换包装类，将y的分箱逻辑和分类模型封装到一起，既符合Pipeline的接口规范，也能避免数据泄露，所有逻辑可完全内置到Pipeline中。

自定义包装类代码 #

from sklearn.base import BaseEstimator, ClassifierMixin
from sklearn.preprocessing import KBinsDiscretizer

class BinnedTargetClassifier(BaseEstimator, ClassifierMixin):
    def __init__(self, classifier, n_bins=3, encode='ordinal', strategy='quantile'):
        self.classifier = classifier
        self.n_bins = n_bins
        self.encode = encode
        self.strategy = strategy
        # 初始化目标分箱器
        self.target_binner = KBinsDiscretizer(n_bins=self.n_bins, encode=self.encode, strategy=self.strategy)
    
    def fit(self, X, y):
        # 仅用训练集的y拟合分箱器，避免数据泄露
        y_binned = self.target_binner.fit_transform(y.reshape(-1, 1)).ravel()
        # 用分箱后的y拟合分类器
        self.classifier.fit(X, y_binned)
        self.classes_ = self.target_binner.bin_edges_[0] # 可选，保存分箱边界
        return self
    
    def predict(self, X):
        return self.classifier.predict(X)
    
    def predict_proba(self, X):
        return self.classifier.predict_proba(X)

完整可运行Pipeline示例 #

import sklearn
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LogisticRegression
from sklearn.preprocessing import StandardScaler
import pandas as pd
from sklearn.datasets import load_boston
from sklearn.model_selection import train_test_split
from sklearn.metrics import accuracy_score

# 加载数据
data = load_boston()
X = data['data']
y = data['target']
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 构建Pipeline：所有预处理+模型逻辑全部内置
pipeline = Pipeline(steps = [
    ('feature_scaler', StandardScaler()), # 这里可以加任意对X的预处理步骤
    ('classifier', BinnedTargetClassifier(classifier=LogisticRegression(), n_bins=3))
])

# 训练、评估
pipeline.fit(X_train, y_train)
y_pred = pipeline.predict(X_test)
# 测试集y先做相同分箱再计算准确率
y_test_binned = pipeline.named_steps['classifier'].target_binner.transform(y_test.reshape(-1,1)).ravel()
print(f"分类准确率：{accuracy_score(y_test_binned, y_pred):.2f}")

方案优势 #

• 所有转换逻辑完全内置在Pipeline中，没有额外样板代码，不存在数据泄露风险（分箱器仅在fit阶段用训练集y拟合，不会接触测试集数据）
• 整个Pipeline符合Sklearn的标准接口，可以直接对接Databricks模型注册中心等部署服务，不需要额外处理目标变量的逻辑。
• 包装类的分类器参数可灵活替换，支持所有兼容Sklearn接口的分类模型，比如随机森林、XGBoost等。

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