scikit-learn 嵌套 Pipeline 添加特征后为何报 NotFittedError？

2026-11-04数据小许

结论

根据给出的代码和检查结果，直接原因是 AddComboFeatures.fit() 只执行了 return self，没有留下可被 scikit-learn 识别的拟合状态。它是内层 add_features Pipeline 的最后一步，会影响内层 Pipeline 的拟合状态判断。

让自定义转换器在 fit() 中记录输入特征信息，再重新创建并拟合整个 Pipeline，就可以针对这一原因修复报错。列名硬编码也需要维护，但当前堆栈首先暴露的是拟合状态问题。

为什么调用了 fit，仍被判定为未拟合？

check_is_fitted() 通常通过名称以 _ 结尾的实例属性识别拟合结果，例如 age_cap_、n_features_in_。它也支持通过 __sklearn_is_fitted__() 等机制判断状态，因此不能简单理解为“检查 fit 是否调用过”。详见 check_is_fitted 文档。

两段代码的区别在于：

# AddCensoredFeatures.fit() 会创建拟合属性
self.age_cap_ = X["housing_median_age"].max()

# AddComboFeatures.fit() 没有创建任何拟合属性
return self

因此，检查结果中的 NOT fitted 不能证明 AddComboFeatures.fit() 没有执行。

Pipeline 的相关实现会检查最后一个有效步骤的拟合状态。你的结构是：

外层 Pipeline
├── add_features：内层 Pipeline
│   ├── add_censored_cols
│   └── add_combo_cols
└── preprocessing：ColumnTransformer

如果内层最后一步 add_combo_cols 的检查失败，内层 Pipeline 就会被判定为未拟合。外层 Pipeline 的状态检查可能因为末尾的 ColumnTransformer 已拟合而通过，但执行 transform() 时仍会进入内层 Pipeline，触发它自己的检查。这个判断方式可参照 Pipeline 源码中的 sklearn_is_fitted。这里引用源码是为了解释机制，并不代表已经确定你的安装版本。

fit_transform() 可以在拟合过程中直接调用各步骤的拟合和转换方法。因此，训练阶段得到输出，后续独立调用 transform() 时仍不一定能通过状态检查。不同版本的检查时机和警告行为可能有所差异。

修改自定义转换器的 fit

将 AddComboFeatures.fit() 替换为下面的方法，现有的特征计算逻辑可以保留：

def fit(self, X, y=None):
    required_columns = [
        "total_rooms",
        "total_bedrooms",
        "population",
        "median_income",
        "households",
    ]
    missing_columns = [
        column for column in required_columns
        if column not in X.columns
    ]
    if missing_columns:
        raise ValueError(
            f"The following columns are missing: {missing_columns}"
        )

    self.feature_names_in_ = np.asarray(X.columns, dtype=object)
    self.n_features_in_ = X.shape[1]
    return self

这些属性记录了真实的输入信息，也符合 scikit-learn 对表格输入估计器的约定，参见自定义估计器开发文档。

AddCensoredFeatures 也应记录相同的输入信息。当 add_censored_age=False 时，它不会创建 age_cap_，因此单独检查该转换器时也可能被判定为未拟合：

def fit(self, X, y=None):
    if self.add_censored_age:
        if "housing_median_age" not in X.columns:
            raise ValueError(
                "housing_median_age is not a column in this data."
            )
        self.age_cap_ = X["housing_median_age"].max()

    self.feature_names_in_ = np.asarray(X.columns, dtype=object)
    self.n_features_in_ = X.shape[1]
    return self

如果希望自定义转换器阻止未拟合调用，可以在两个 transform() 方法开头加入：

from sklearn.utils.validation import check_is_fitted

# 放在 transform() 方法内部
check_is_fitted(self, ["feature_names_in_", "n_features_in_"])

保存这些属性不会自动校验后续输入的列名和顺序。如果需要严格约束，还应在 transform() 中增加对应检查。

使用 pandas 输出时，补齐特征名称处理

原来的 get_feature_names_out() 直接执行 list(input_features)。当调用方没有显式传入名称时，input_features 默认为 None，会产生另一个错误。

将 AddComboFeatures.get_feature_names_out() 改为：

def get_feature_names_out(self, input_features=None):
    check_is_fitted(self, "feature_names_in_")
    if input_features is None:
        input_features = self.feature_names_in_

    return np.asarray(
        list(input_features) + [
            "rooms_per_bedroom",
            "population_per_room",
            "income_per_house",
            "income_per_population",
        ],
        dtype=object,
    )

AddCensoredFeatures 对应的方法改为：

def get_feature_names_out(self, input_features=None):
    check_is_fitted(self, "feature_names_in_")
    if input_features is None:
        input_features = self.feature_names_in_

    names = list(input_features)
    if self.add_censored_age:
        names.append("censored_age")
    return np.asarray(names, dtype=object)

这些方法假设新增列原先不存在，且输出列顺序与代码中声明的一致。

对数变换保持列数和列名不变，可以在 get_pipeline() 中明确声明：

log_transformer = FunctionTransformer(
    log_transform,
    inverse_func=inv_log_transform,
    validate=False,
    feature_names_out="one-to-one",
)

feature_names_out="one-to-one" 表示输入、输出特征名称一一对应，见 FunctionTransformer 文档。它用于处理特征名称接口，前面的修改则用于修复拟合状态。

修改后重新创建并拟合实例：

pipeline = features.get_pipeline().set_output(transform="pandas")

train_X = pipeline.fit_transform(train_X_raw)

feature_pipeline = pipeline.named_steps["add_features"]
check_is_fitted(feature_pipeline)
check_is_fitted(feature_pipeline.named_steps["add_combo_cols"])

test_X = pipeline.transform(test_X_raw)

这段代码用于检查修复后的行为。它能否完整执行，还取决于输入列、缺失值、数值范围和分类值是否满足后续步骤的要求。测试集应复用训练得到的同一个实例，不要对测试集再次调用 fit_transform()。

ColumnTransformer 应如何接收新增列？

当前顺序可以保留：先由 add_features 返回包含新增列的 DataFrame，再交给 ColumnTransformer 按名称选择列。因此，population_per_room 等列即使不在原始输入中，只要在前一步生成，就可以被后面的列选择器引用。

需要留意两个细节：

censored_age 没有出现在数值列列表中，目前通过 remainder="passthrough" 原样输出。如果希望这个二元标记也参与标准化，需要明确加入对应分支。
手工按 __ 拆分输出列名可能造成重名。如果希望去掉转换器前缀，可以考虑 ColumnTransformer(..., verbose_feature_names_out=False)，前提是最终特征名称唯一。

还需检查哪些数据问题？

除零后取对数

np.where(b == 0, 0, a / b) 会先计算 a / b，再选择结果，因此仍可能产生除零警告。可以使用带掩码的除法：

def safe_divide(a, b):
    a = np.asarray(a, dtype=float)
    b = np.asarray(b, dtype=float)
    return np.divide(
        a,
        b,
        out=np.zeros_like(a, dtype=float),
        where=b != 0,
    )

但分母为零时返回的 0，进入 np.log() 后仍会变成 -inf。如果保留自然对数，应确保输入为有限的正数；如果业务上允许对非负特征使用 np.log1p()，逆变换需要同步改为 np.expm1()。这会改变特征定义，应根据数据含义选择。

缺失值填充的数据泄漏

示例在划分训练集之前调用 clean_data()，其中位数会使用测试集信息。应先划分数据，再仅根据训练集计算填充值，并将同一填充值用于测试集。做交叉验证时，适合把填充放进 Pipeline，并安排在需要使用 total_bedrooms 的特征计算之前。

测试集的新分类值

如果 ocean_proximity 在测试集中出现训练阶段未见的值，默认的 OneHotEncoder 可能报错。如果业务允许将未知分类编码为该特征对应的一组零，可以设置：

OneHotEncoder(sparse_output=False, handle_unknown="ignore")

此外，代码没有给出 train_test_split 的来源。如果使用的是 sklearn 的同名函数，需先分离特征与标签，再按其返回顺序接收 X_train, X_test, y_train, y_test；如果是自定义函数，则以自定义实现为准。