自定义ColumnTransformer出现NotFittedError问题求助

问题描述 #

我构建了一个包含两个自定义列转换器的Scikit-learn Pipeline，其中SkipSimpleImputer可正常运行，但CustomZeroImputer抛出以下错误：

NotFittedError: This CustomZeroImputer instance is not fitted yet. Call 'fit' with appropriate arguments before using this estimator.

完整Pipeline代码如下：

class SkipSimpleImputer(SimpleImputer):
    def __init__(self, **kwargs):
        super().__init__(**kwargs)

    def transform(self, X, y=None):
        if 'MARITAL_STATUS' in X.columns:
            print('\t---- MARITAL STATUS found in skipsimpleimpute, all cols are: ', X.columns)
            transformed_X = super().transform(X['MARITAL_STATUS'])
            X['MARITAL_STATUS'] = transformed_X
        return X

    def fit(self, X, y=None):
        return self

drop_cols =  Pipeline(steps=[
        ("columnDropper", columnDropperTransformer(id_cols))
    ])

feature_remover = Pipeline(steps=[
        ("columnDropper", missingRemover())
    ])

class CustomZeroImputer(SimpleImputer):
    def __init__(self, columns, **kwargs):
        self.columns = columns
        print('---- Got zeroImp request: ', columns)
        super().__init__(**kwargs)
        
    def fit(self, X, y=None):
        return self

    def transform(self, X, y=None):
        cols_to_imp = [col for col in self.columns if col in X.columns]
        
        print('---- Imputing zero cols: ', cols_to_imp)
        X[cols_to_imp] = super().transform(X[cols_to_imp])
        return X

fill_na_zero_transformer = Pipeline(steps=[
    ('zero_imputer', CustomZeroImputer(fill_zero_cols, strategy='constant', fill_value=0))
])

numeric_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy = "constant", fill_value=-1, add_indicator=True)),
    ('scaler', StandardScaler())
])

categorical_transformer = Pipeline(steps=[
    ('categorical_imputer', SkipSimpleImputer(strategy="constant", fill_value='Unknown')),
    ('encoder', OneHotEncoder(handle_unknown='ignore'))
])

preprocess_ppl = ColumnTransformer(
    transformers=[
        ('encode', categorical_transformer, make_column_selector(dtype_include=object)),
        ('zero_impute', fill_na_zero_transformer, make_column_selector(dtype_include=np.number)),
        ('numeric', numeric_transformer, make_column_selector(dtype_include=np.number))
    ]
)
pipeline2 = Pipeline(
    steps=[
        ('dropper', drop_cols),
        ('remover',feature_remover),
        ("preprocessor", preprocess_ppl),
        ("estimator", customOLS(sm.OLS))
        ]
)

print('---- Fitting pipeline')
pipeline2.fit(X_train,y_train)

问题原因 #

CustomZeroImputer的fit方法直接返回self，没有调用父类SimpleImputer的fit方法。虽然SkipSimpleImputer也采用了同样的写法，但它仅处理单一列且使用常量填充策略，父类在这种场景下无需依赖拟合后的统计量即可运行；而CustomZeroImputer调用super().transform()时，父类内部会强制检查是否完成拟合流程，因此触发了NotFittedError。

解决方案 #

修改CustomZeroImputer的fit方法，针对目标列调用父类的fit方法，同时保留“只处理当前存在列”的逻辑：

class CustomZeroImputer(SimpleImputer):
    def __init__(self, columns, **kwargs):
        self.columns = columns
        print('---- Got zeroImp request: ', columns)
        super().__init__(**kwargs)
        
    def fit(self, X, y=None):
        # 仅对当前数据中存在的目标列执行拟合
        cols_to_fit = [col for col in self.columns if col in X.columns]
        if cols_to_fit:
            super().fit(X[cols_to_fit], y)
        return self

    def transform(self, X, y=None):
        cols_to_imp = [col for col in self.columns if col in X.columns]
        
        print('---- Imputing zero cols: ', cols_to_imp)
        if cols_to_imp:
            X[cols_to_imp] = super().transform(X[cols_to_imp])
        return X

补充说明 #

• 该修改既让父类SimpleImputer完成了必要的拟合流程，避免了错误，又保留了你原本的容错逻辑；
• 严格来说，SkipSimpleImputer也应该补全fit方法的父类调用，避免后续场景变更时出现潜在问题。

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