Sklearn Pipeline处理加州住房数据集时NotFittedError问题排查

核心原因 #

你的add_combo_cols步骤大概率是自定义的特征处理逻辑，但没有遵循Scikit-learn的Transformer规范，导致Pipeline在fit阶段没有标记该步骤为已拟合状态，最终测试集transform时触发NotFittedError。

常见问题场景及修复方案 #

1. 自定义Transformer未继承Scikit-learn基类

如果你的add_combo_cols是普通类/函数，没有继承BaseEstimator和TransformerMixin，Scikit-learn无法识别它是可拟合的Transformer，也就不会在fit阶段执行必要的状态标记。

错误示例：

# 错误：无基类继承，无fit方法
class AddComboCols:
    def transform(self, X):
        X['room_per_household'] = X['total_rooms'] / X['households']
        return X

修复方案：
必须继承BaseEstimator和TransformerMixin，并实现fit方法（即使不需要拟合参数，也要返回self）：

from sklearn.base import BaseEstimator, TransformerMixin

class AddComboCols(BaseEstimator, TransformerMixin):
    def fit(self, X, y=None):
        # 可选：记录训练集特征列，确保测试集特征一致
        self.train_features_ = X.columns.tolist()
        return self  # 必须返回self，标记为已拟合
    
    def transform(self, X):
        # 避免修改原数据，返回副本
        X_copy = X.copy()
        # 添加加州住房数据集的组合特征
        X_copy['room_per_household'] = X_copy['total_rooms'] / X_copy['households']
        X_copy['bedroom_per_room'] = X_copy['total_bedrooms'] / X_copy['total_rooms']
        X_copy['population_per_household'] = X_copy['population'] / X_copy['households']
        return X_copy

2. Pipeline中使用原始函数而非Transformer实例

如果直接把自定义函数传入Pipeline（而非包装成Transformer），Scikit-learn无法跟踪其拟合状态。

错误示例：

from sklearn.pipeline import Pipeline

def add_combo_cols_func(X):
    X_copy = X.copy()
    X_copy['room_per_household'] = X_copy['total_rooms'] / X_copy['households']
    return X_copy

# 错误：直接传入函数，未包装成Transformer
pipe = Pipeline([
    ('add_combo', add_combo_cols_func),
    ('log_transform', FunctionTransformer(np.log1p)),
    ('scaler', StandardScaler())
])

修复方案：
用FunctionTransformer包装函数，并设置validate=True：

from sklearn.preprocessing import FunctionTransformer

pipe = Pipeline([
    ('add_combo', FunctionTransformer(add_combo_cols_func, validate=True)),
    ('log_transform', FunctionTransformer(np.log1p, validate=True)),
    ('scaler', StandardScaler())
])

3. Fit后重新初始化Pipeline/步骤

如果在fit训练集后，重新创建了Pipeline或add_combo_cols实例，新实例是未拟合状态，调用transform会报错。

错误示例：

pipe.fit(X_train, y_train)
# 错误：重新创建未拟合的Pipeline
pipe = Pipeline([
    ('add_combo', AddComboCols()),
    ('log_transform', FunctionTransformer(np.log1p)),
    ('scaler', StandardScaler())
])
pipe.transform(X_test)  # 触发NotFittedError

修复方案：
确保fit后的Pipeline实例是同一个，不要重新初始化。

完整可运行示例（结合加州住房数据集） #

import numpy as np
import pandas as pd
from sklearn.datasets import fetch_california_housing
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, FunctionTransformer
from sklearn.model_selection import train_test_split

# 加载数据集
housing = fetch_california_housing()
X = pd.DataFrame(housing.data, columns=housing.feature_names)
y = housing.target
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42)

# 符合规范的自定义Transformer
class AddComboCols(BaseEstimator, TransformerMixin):
    def fit(self, X, y=None):
        self.train_features_ = X.columns.tolist()
        return self
    
    def transform(self, X):
        if not set(self.train_features_).issubset(X.columns):
            raise ValueError("测试集缺失训练集中的必要特征")
        X_copy = X.copy()
        X_copy['room_per_household'] = X_copy['total_rooms'] / X_copy['households']
        X_copy['bedroom_per_room'] = X_copy['total_bedrooms'] / X_copy['total_rooms']
        X_copy['population_per_household'] = X_copy['population'] / X_copy['households']
        return X_copy

# 构建并运行Pipeline
pipe = Pipeline([
    ('add_combo', AddComboCols()),
    ('log_transform', FunctionTransformer(np.log1p, validate=True)),
    ('scaler', StandardScaler())
])

# 训练集处理
X_train_processed = pipe.fit_transform(X_train, y_train)
# 测试集处理
X_test_processed = pipe.transform(X_test)

print("训练集处理后形状:", X_train_processed.shape)
print("测试集处理后形状:", X_test_processed.shape)

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