如何基于Scikit-learn构建两阶段回归Pipeline：X1拟合y后用X2拟合残差

要实现你描述的两步回归流程，核心是自定义符合Scikit-learn接口的组件，将残差计算环节整合到Pipeline中。以下是完整的实现方案：

1. 自定义残差堆叠回归器 #

我们需要一个复合回归器，继承Scikit-learn的基础类来适配Pipeline，它会完成"用X1训练回归器→计算残差→用X2训练回归器预测残差"的完整逻辑。

2. 完整代码实现 #

import numpy as np
from sklearn.base import BaseEstimator, RegressorMixin
from sklearn.pipeline import Pipeline
from sklearn.linear_model import LinearRegression
from sklearn.ensemble import RandomForestRegressor
from sklearn.model_selection import train_test_split
from sklearn.metrics import mean_squared_error

# 自定义堆叠回归器，实现残差预测逻辑
class ResidualStackedRegressor(BaseEstimator, RegressorMixin):
    def __init__(self, regressor1, regressor2, x1_cols, x2_cols):
        self.regressor1 = regressor1
        self.regressor2 = regressor2
        self.x1_cols = x1_cols  # X1在合并特征矩阵中的列索引
        self.x2_cols = x2_cols  # X2在合并特征矩阵中的列索引
        
    def fit(self, X, y):
        # 提取X1特征训练第一个回归器
        X1 = X[:, self.x1_cols]
        self.regressor1.fit(X1, y)
        # 计算残差
        y_pred1 = self.regressor1.predict(X1)
        residual = y - y_pred1
        # 提取X2特征训练第二个回归器预测残差
        X2 = X[:, self.x2_cols]
        self.regressor2.fit(X2, residual)
        return self
    
    def predict(self, X):
        # 用X1得到第一阶段预测值
        X1 = X[:, self.x1_cols]
        y_pred1 = self.regressor1.predict(X1)
        # 用X2得到残差预测值
        X2 = X[:, self.x2_cols]
        y_pred_residual = self.regressor2.predict(X2)
        # 总预测值为两者之和
        return y_pred1 + y_pred_residual

# 生成模拟数据
np.random.seed(42)
X1 = np.random.rand(1000, 10)  # 10个X1特征
X2 = np.random.rand(1000, 10)  # 10个X2特征
X = np.hstack([X1, X2])  # 合并为单特征矩阵
y = 2 * X1[:, 0] + 3 * X1[:, 1] + np.log(X2[:, 2] + 1) + np.random.randn(1000) * 0.1

# 拆分数据集
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 构建Pipeline
pipeline = Pipeline([
    ('stacked_reg', ResidualStackedRegressor(
        regressor1=LinearRegression(),
        regressor2=RandomForestRegressor(n_estimators=100, random_state=42),
        x1_cols=list(range(10)),  # X1对应前10列
        x2_cols=list(range(10, 20))  # X2对应后10列
    ))
])

# 训练与预测
pipeline.fit(X_train, y_train)
y_pred = pipeline.predict(X_test)

# 评估效果
print(f"测试集MSE: {mean_squared_error(y_test, y_pred):.4f}")

3. 扩展：添加特征预处理 #

如果需要对X1、X2分别做预处理（比如标准化），可以在Pipeline中加入ColumnTransformer实现分组处理：

from sklearn.preprocessing import StandardScaler
from sklearn.compose import ColumnTransformer

# 分组预处理：分别标准化X1和X2
preprocessor = ColumnTransformer(
    transformers=[
        ('scaler_x1', StandardScaler(), list(range(10))),
        ('scaler_x2', StandardScaler(), list(range(10, 20)))
    ])

# 带预处理的完整Pipeline
full_pipeline = Pipeline([
    ('preprocess', preprocessor),
    ('stacked_reg', ResidualStackedRegressor(
        regressor1=LinearRegression(),
        regressor2=RandomForestRegressor(n_estimators=100, random_state=42),
        x1_cols=list(range(10)),
        x2_cols=list(range(10, 20))
    ))
])

4. 注意事项 #

• 如果X1和X2是独立的特征矩阵，建议合并为单个矩阵后再传入Pipeline，这更符合Scikit-learn的设计逻辑。
• regressor1和regressor2可以替换为任意Scikit-learn兼容的回归器（如SVR、GradientBoostingRegressor等）。

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