如何在Scikit-learn Pipeline中实现三个分位数回归预测？解决FeatureUnion报错

错误原因 #

FeatureUnion要求每个传入的估计器必须实现fit和transform方法，但TransformedTargetRegressor是回归器，仅提供predict方法，没有transform接口，因此触发报错。

替代方案 #

方案1：自定义包装器将回归器转为Transformer #

编写一个简单的包装类，把回归器的predict方法包装成transform方法，适配FeatureUnion的要求：

from sklearn.base import BaseEstimator, TransformerMixin
import numpy as np

class RegressorTransformer(BaseEstimator, TransformerMixin):
    def __init__(self, regressor):
        self.regressor = regressor
    
    def fit(self, X, y=None):
        self.regressor.fit(X, y)
        return self
    
    def transform(self, X):
        # 返回二维数组，匹配FeatureUnion的输出格式
        return self.regressor.predict(X).reshape(-1, 1)

修改Pipeline代码，用该包装器包裹每个TransformedTargetRegressor：

from sklearn.pipeline import Pipeline, FeatureUnion
from sklearn.preprocessing import StandardScaler
from sklearn.compose import TransformedTargetRegressor
from sklearn.linear_model import QuantileRegressor

# 假设已定义log和exp转换函数
log_func = np.log
exp_func = np.exp

pipe = Pipeline([
    ('scaler', StandardScaler()),
    ('estimator', FeatureUnion([
        ('median', RegressorTransformer(
            TransformedTargetRegressor(
                regressor=QuantileRegressor(quantile=0.50),
                func=log_func, inverse_func=exp_func, check_inverse=True
            )
        )),
        ('lower', RegressorTransformer(
            TransformedTargetRegressor(
                regressor=QuantileRegressor(quantile=0.10),
                func=log_func, inverse_func=exp_func, check_inverse=True
            )
        )),
        ('upper', RegressorTransformer(
            TransformedTargetRegressor(
                regressor=QuantileRegressor(quantile=0.90),
                func=log_func, inverse_func=exp_func, check_inverse=True
            )
        ))
    ]))
])

# 拟合与预测
pipe.fit(X_train, y_train)
predictions = pipe.predict(X_test)
# predictions列顺序：中位数、下限、上限

方案2：手动管理独立Pipeline #

无需自定义类，直接创建三个结构一致的Pipeline，拟合后合并预测结果：

# 先拟合Scaler，避免重复计算
scaler = StandardScaler()
X_train_scaled = scaler.fit_transform(X_train)
X_test_scaled = scaler.transform(X_test)

# 定义三个分位数回归器
median_reg = TransformedTargetRegressor(
    regressor=QuantileRegressor(quantile=0.50),
    func=log_func, inverse_func=exp_func, check_inverse=True
)
lower_reg = TransformedTargetRegressor(
    regressor=QuantileRegressor(quantile=0.10),
    func=log_func, inverse_func=exp_func, check_inverse=True
)
upper_reg = TransformedTargetRegressor(
    regressor=QuantileRegressor(quantile=0.90),
    func=log_func, inverse_func=exp_func, check_inverse=True
)

# 分别拟合
median_reg.fit(X_train_scaled, y_train)
lower_reg.fit(X_train_scaled, y_train)
upper_reg.fit(X_train_scaled, y_train)

# 合并预测结果
y_median = median_reg.predict(X_test_scaled)
y_lower = lower_reg.predict(X_test_scaled)
y_upper = upper_reg.predict(X_test_scaled)

predictions = np.column_stack([y_median, y_lower, y_upper])

方案3：自定义多分位数回归器 #

将三个分位数的回归逻辑封装成单个回归器，简化调用流程：

from sklearn.base import BaseEstimator, RegressorMixin

class MultiQuantileRegressor(BaseEstimator, RegressorMixin):
    def __init__(self, quantiles=[0.5, 0.1, 0.9], func=None, inverse_func=None):
        self.quantiles = quantiles
        self.func = func
        self.inverse_func = inverse_func
        self.regressors = [QuantileRegressor(q) for q in quantiles]
    
    def fit(self, X, y):
        # 应用目标转换
        y_transformed = self.func(y) if self.func else y
        for reg in self.regressors:
            reg.fit(X, y_transformed)
        return self
    
    def predict(self, X):
        preds_transformed = [reg.predict(X) for reg in self.regressors]
        # 反转转换
        if self.inverse_func:
            preds = [self.inverse_func(p) for p in preds_transformed]
        else:
            preds = preds_transformed
        return np.column_stack(preds)

# 构建Pipeline
pipe = Pipeline([
    ('scaler', StandardScaler()),
    ('regressor', MultiQuantileRegressor(
        quantiles=[0.5, 0.1, 0.9],
        func=log_func, inverse_func=exp_func
    ))
])

pipe.fit(X_train, y_train)
predictions = pipe.predict(X_test)

方案对比 #

• 方案1：保留FeatureUnion并行结构，代码整洁，但需要自定义Transformer。
• 方案2：逻辑直观，无额外自定义代码，适合快速实现，可通过复用Scaler优化性能。
• 方案3：封装性强，适合需要频繁复用多分位数预测的场景，调用更简洁。

内容的提问来源于stack exchange，提问作者Selva Kumaresan Ramakrishnan