Scikit-learn Pipeline集成自定义筛选器在GridSearchCV中报错排查

问题场景 #

构建包含自定义特征筛选Transformer、OptBinning BinningProcess、Logistic Regression的Pipeline，并在GridSearchCV中执行，目标是基于IV值筛选高相关性特征以避免模型偏差。单独运行自定义Transformer正常，但集成到Pipeline后全部拟合失败，报错ValueError: 列数必须与variable_names的长度相等。

核心报错原因分析 #

1. 参数传递顺序错误：自定义CorrBivariateColumnsSelector初始化参数顺序与调用时传入的参数不匹配，导致内部逻辑使用错误的参数进行特征筛选，输出的特征列与后续BinningProcess预期的列数/列名不匹配。
2. BinningProcess静态初始化：BinningProcess初始化时指定了原始全部特征，但经过自定义Transformer筛选后特征数量减少，导致输入列数与BinningProcess预设的variable_names长度不一致。
3. Transformer不符合Sklearn API规范：自定义Transformer的fit方法未使用交叉验证时传入的当前fold的y，而是使用初始化时传入的全局y，既不符合Pipeline设计逻辑，也可能导致数据泄漏；同时fit阶段未完成特征筛选计算，延迟到transform阶段，导致每次transform都重复计算且无法保存筛选结果。
4. 函数内部参数硬编码：corr_bivariate_filter函数内部硬编码覆盖了传入的corr_threshold和monotonic参数，导致外部配置无效。

修复步骤与代码调整 #

1. 修复自定义Transformer的参数与逻辑 #

• 修正__init__参数顺序，确保调用时传入正确的参数
• 在fit阶段完成特征筛选计算，保存筛选后的列名到实例属性
• 使用fit时传入的当前fold的y进行IV计算，避免数据泄漏
• 移除corr_bivariate_filter内部硬编码的参数值

修复后的自定义Transformer代码：

import pandas as pd
import numpy as np
from sklearn.base import BaseEstimator, TransformerMixin
from optbinning import BinningProcess, OptimalBinning


def corr_bivariate_filter(
    X: pd.DataFrame,
    y: pd.DataFrame,
    filtered_features: list,
    corr_threshold=0.8,
    monotonic=True,
    binning_fit_params={},
):
    num_vars = list(X.select_dtypes(include=[int, float]).columns)
    cat_vars = list(X.select_dtypes(include=[object]).columns)

    if monotonic:
        binning_fit_params = {
            v: {"monotonic_trend": "auto_asc_desc"} for v in num_vars + cat_vars
        }

    X_filtered = X[filtered_features].copy()

    binning_process = BinningProcess(
        filtered_features,
        categorical_variables=cat_vars,
        binning_fit_params=binning_fit_params,
    )
    binning_process.fit(X_filtered, y)

    bivariate_df = binning_process.summary()
    iv_dict = bivariate_df.set_index('name')['iv'].to_dict()

    num_vars_filtered = list(X_filtered.select_dtypes(include=["int", "float"]).columns)
    corr_matrix = X_filtered[num_vars_filtered].corr().abs()
    upper = corr_matrix.where(np.triu(np.ones(corr_matrix.shape), k=1).astype(bool))
    
    correlation_groups = []
    for c in upper.columns:
        correlations = {}
        df = upper[upper[c].notna()]
        variables = list(df.index.values)
        if variables:
            for v in variables:
                correlations[v] = iv_dict[v]
            correlations[c] = iv_dict[c]
            correlation_groups.append(correlations)

    post_filter_columns = []
    for d in correlation_groups:
        post_filter_columns.append(max(d, key=d.get))
    post_filter_columns = list(dict.fromkeys(post_filter_columns))

    unpacked_vars = []
    for d in correlation_groups:
        unpacked_vars.extend(d.keys())
    to_drop = list(set(unpacked_vars) - set(post_filter_columns))

    columns_after_correlation_filter = [
        c for c in X_filtered.columns if c not in to_drop
    ]
    return columns_after_correlation_filter


class CorrBivariateColumnsSelector(BaseEstimator, TransformerMixin):
    def __init__(
        self,
        filtered_features,
        corr_threshold=0.8,
        monotonic=True,
        binning_fit_params={},
    ):
        self.filtered_features = filtered_features
        self.corr_threshold = corr_threshold
        self.monotonic = monotonic
        self.binning_fit_params = binning_fit_params
        self.selected_columns = None

    def fit(self, X, y):
        # 在fit阶段完成特征筛选，使用当前fold的y
        self.selected_columns = corr_bivariate_filter(
            X,
            y,
            self.filtered_features,
            self.corr_threshold,
            self.monotonic,
            self.binning_fit_params,
        )
        return self

    def transform(self, X):
        # 直接返回预筛选后的列
        return X[self.selected_columns].copy()

2. 修复Pipeline与GridSearchCV调用逻辑 #

• 修正CorrBivariateColumnsSelector实例化的参数顺序
• 使用动态BinningProcess：在Pipeline中，BinningProcess会自动接收Transformer输出的特征列，无需提前指定variable_names（OptBinning的BinningProcess如果不指定variable_names，会自动使用输入数据的列名）

修复后的Pipeline调用代码：

import pandas as pd
import numpy as np
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import make_scorer
from utils import ks_score
from sklearn.pipeline import make_pipeline
from sklearn.model_selection import cross_validate, GridSearchCV, KFold
from optbinning import BinningProcess
from utils import CorrBivariateColumnsSelector

ks_scorer = make_scorer(ks_score, needs_proba=True, greater_is_better=True)

features = catalog.load("features_after_univariate_filter")
feature_list = features['features'].to_list()
df = catalog.load("sample_with_features")

df = df.set_index('request_id')[feature_list + ['30ever3']].copy()
X = df[feature_list]
y = df['30ever3'].values.ravel()

num_vars = list(X.select_dtypes(include=[int,float]).columns)
cat_vars = list(X.select_dtypes(include=[object]).columns)

# 初始化Transformer：参数顺序修正为(filtered_features, corr_threshold, monotonic, ...)
correlation_bivariate_filter = CorrBivariateColumnsSelector(
    filtered_features=feature_list,
    corr_threshold=0.8,
    monotonic=True,
    binning_fit_params={}
)

# 动态初始化BinningProcess，不指定variable_names，自动使用输入列名
binning_process = BinningProcess(
    categorical_variables=cat_vars,
    binning_fit_params={v: {"monotonic_trend": "auto_asc_desc"} for v in num_vars+cat_vars}
)

lr = LogisticRegression()

pipe = make_pipeline(
    correlation_bivariate_filter, 
    binning_process, 
    lr
)

inner_cv = KFold(n_splits=3, shuffle=True, random_state=1)
outer_cv = KFold(n_splits=3, shuffle=True, random_state=3)

p_grid = {
    "logisticregression__C": [0.5, 1, 2, 10],
    "logisticregression__penalty": ["l1"],
    "logisticregression__solver": ["liblinear"],
    "logisticregression__max_iter": [50, 80, 100, 150, 200],
}

scoring = {
    'auc': 'roc_auc', 
    'log_loss': 'neg_log_loss',
    'brier_score': 'neg_brier_score', 
    'ks': ks_scorer,
}

clf = GridSearchCV(estimator=pipe, param_grid=p_grid, cv=inner_cv)
cv_results = cross_validate(
    clf,
    X, 
    y,
    cv=outer_cv, 
    scoring=scoring,
    return_estimator=True, 
)

关键修复点说明 #

1. 参数顺序修正：确保CorrBivariateColumnsSelector实例化时参数顺序与__init__定义一致，避免传入错误的参数导致特征筛选逻辑混乱。
2. 符合Sklearn API规范：在fit阶段完成特征筛选计算并保存结果，transform阶段直接复用结果，同时使用当前fold的y进行IV计算，避免数据泄漏。
3. 动态BinningProcess：不提前指定variable_names，让BinningProcess自动适配Transformer输出的特征列，解决列数不匹配的问题。
4. 移除硬编码参数：让corr_bivariate_filter函数使用外部传入的参数，保证配置的灵活性。

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