如何将自定义特征工程类集成到Sklearn Pipeline的XGBoost代码中？

1. 先确保自定义类符合Sklearn接口规范 #

Sklearn Pipeline要求所有组件必须实现fit()和transform()方法，最简便的方式是继承BaseEstimator和TransformerMixin基类。你的CustomFeatureEngineering类需要按如下规范实现：

import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin

class CustomFeatureEngineering(BaseEstimator, TransformerMixin):
    def __init__(self, target_num_cols=None):
        # 初始化：指定要生成组合特征的数值列
        self.target_num_cols = target_num_cols if target_num_cols else []
    
    def fit(self, X, y=None):
        # 特征组合无需拟合数据，直接返回self即可
        return self
    
    def transform(self, X):
        # 复制输入数据，避免修改原数据集
        X_processed = X.copy()
        
        # 示例：生成数值列的加减乘除组合特征
        if len(self.target_num_cols) >= 2:
            col_a, col_b = self.target_num_cols[0], self.target_num_cols[1]
            X_processed[f"{col_a}_add_{col_b}"] = X_processed[col_a] + X_processed[col_b]
            X_processed[f"{col_a}_mul_{col_b}"] = X_processed[col_a] * X_processed[col_b]
            X_processed[f"{col_a}_sub_{col_b}"] = X_processed[col_a] - X_processed[col_b]
        
        # 可添加更多自定义特征逻辑
        return X_processed

2. 将自定义类嵌入Pipeline流程 #

Pipeline的执行顺序是从上到下，所以要把自定义特征工程放在最前面，之后衔接常规预处理步骤，最后是XGBoost模型：

from sklearn.pipeline import Pipeline
from sklearn.impute import SimpleImputer
from sklearn.preprocessing import OneHotEncoder
from sklearn.compose import ColumnTransformer
import xgboost as xgb

# 定义数据集列类型
numeric_cols = ['age', 'monthly_income']
categorical_cols = ['gender', 'industry']

# 数值列预处理管道
numeric_pipeline = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='median')),
])

# 分类列预处理管道
categorical_pipeline = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('onehot', OneHotEncoder(handle_unknown='ignore'))
])

# 统一处理不同类型列的转换器
preprocessor = ColumnTransformer(
    transformers=[
        ('num', numeric_pipeline, numeric_cols),
        ('cat', categorical_pipeline, categorical_cols)
    ])

# 完整Pipeline：自定义特征工程 → 通用预处理 → XGBoost分类器
full_pipeline = Pipeline(steps=[
    ('custom_fe', CustomFeatureEngineering(target_num_cols=numeric_cols)),
    ('preprocessor', preprocessor),
    ('xgb_clf', xgb.XGBClassifier(objective='binary:logistic'))
])

3. 结合Optuna进行超参数调优 #

调参时需要用组件名__参数名的格式指定Pipeline内各组件的参数，比如自定义类的列选择、XGBoost的模型参数：

import optuna
from sklearn.model_selection import cross_val_score

def objective(trial):
    # 定义待调优参数
    params = {
        # XGBoost参数
        'xgb_clf__max_depth': trial.suggest_int('max_depth', 3, 12),
        'xgb_clf__learning_rate': trial.suggest_float('learning_rate', 0.01, 0.3),
        'xgb_clf__n_estimators': trial.suggest_int('n_estimators', 100, 1200),
        # 可选：调优自定义特征工程的目标列组合
        # 'custom_fe__target_num_cols': trial.suggest_categorical('fe_cols', [['age','monthly_income'], ['age','credit_score']])
    }
    
    # 给Pipeline设置参数
    full_pipeline.set_params(**params)
    
    # 交叉验证评估模型
    accuracy = cross_val_score(full_pipeline, X_train, y_train, cv=5, scoring='accuracy').mean()
    return accuracy

# 启动调优
study = optuna.create_study(direction='maximize')
study.optimize(objective, n_trials=60)

# 用最优参数训练完整模型
best_model = full_pipeline.set_params(**study.best_params)
best_model.fit(X_train, y_train)

# 保存模型
import joblib
joblib.dump(best_model, 'xgb_custom_fe_pipeline.pkl')

常见问题排查 #

• 若出现NotFittedError：检查CustomFeatureEngineering的fit()方法是否正确返回self，确保transform前已执行fit。
• 若新特征未生成：检查transform()方法是否复制了输入数据，以及特征生成逻辑是否存在索引或列名错误。
• 若参数调优不生效：确认参数名格式为组件名__参数名，比如custom_fe__target_num_cols对应自定义类的初始化参数。

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