实现自定义转换器

尽管 Scikit-learn 为数据转换提供了丰富的工具集，但您会经常遇到特定特征工程逻辑或预处理步骤未被标准库涵盖的情况。此时，创建自己的自定义转换器就变得非常必要。自定义转换器允许您将独特的数据处理逻辑封装成可复用组件，这些组件能与 Scikit-learn 的 Pipeline 对象和 GridSearchCV 等模型选择工具良好配合。

要在 Scikit-learn 生态系统中构建一个行为正确的组件，您需要遵循其既定的 API 规范。对于转换器而言，这通常涉及继承 Scikit-learn 提供的两个基类：BaseEstimator 和 TransformerMixin。

典型的自定义 Scikit-learn 转换器的继承结构。

• BaseEstimator：这是 Scikit-learn 中所有估计器的根本基类。继承它会为您的转换器提供两个重要方法：get_params() 和 set_params()。这些方法由 GridSearchCV 和 Pipeline 等工具在内部使用，用于检查和修改转换器的参数（在 __init__ 方法中定义的超参数）。为了使这些方法正常工作，您必须在 __init__ 方法中将超参数作为显式关键字参数接受，并将其作为公共属性不变地存储（例如，self.parameter = parameter）。
• TransformerMixin：这个混入类提供 fit_transform() 方法。默认情况下，它通过简单地调用 fit() 后接 transform() 来实现 fit_transform。虽然方便，但如果存在计算上更高效的方法可以同时执行拟合和转换以满足您的特定逻辑，您偶尔会覆盖此方法。

需要实现的核心方法

自定义转换器通常需要实现以下方法：

1. __init__(self, param1, param2, ...)：构造函数。您在此处定义转换器的超参数。请记住将这些参数作为与参数同名的公共属性存储，并避免在此处进行任何输入验证或超出简单赋值的逻辑。

   # __init__ 结构示例
   from sklearn.base import BaseEstimator, TransformerMixin
   
   class MyCustomTransformer(BaseEstimator, TransformerMixin):
       def __init__(self, strategy='mean', fill_value=None):
           # 直接存储超参数
           self.strategy = strategy
           self.fill_value = fill_value
           # 此处无复杂逻辑或验证
   

2. fit(self, X, y=None)：此方法负责从数据 X 中学习。它可选择使用目标信息 y，尽管对于转换器而言，这比对于监督估计器更不常见。fit 方法应根据训练数据 X 估计任何必要参数，并将其存储为带下划线后缀的属性（例如，self.mean_、self.scale_）。此命名约定将学习到的参数与初始化时设置的超参数区分开。非常重要的一点是，fit 方法必须返回 self。

   # fit 结构示例（用于填充器）
   import numpy as np
   
   class MeanImputer(BaseEstimator, TransformerMixin):
       def __init__(self):
           # 这个简单示例不需要超参数
           pass
   
       def fit(self, X, y=None):
           # 从 X 中学习每个特征的均值
           # 存储学习到的均值，带下划线后缀
           self.means_ = np.nanmean(X, axis=0)
           # 始终返回 self
           return self
   

   在 fit 方法中：

   • X：输入数据（通常是 NumPy 数组或 Pandas DataFrame）。
   • y：目标标签（可选，通常在转换器中被忽略）。
   • 该方法从 X 中计算必要的统计量（如均值、方差、映射）。
   • 这些计算出的统计量存储在以单下划线结尾的实例变量中（例如，self.means_）。
   • 它返回实例 self。

3. transform(self, X)：此方法对数据 X 进行实际转换，使用在 fit 阶段学习到的参数。它不应更新转换器的状态（即，它不应改变任何以单下划线结尾的属性）。它接收数据 X（可以是训练数据或新的、未见过的数据），并且必须返回转换后的数据，通常是 NumPy 数组或 DataFrame。

   # transform 结构示例（MeanImputer 的延续）
   import numpy as np
   from sklearn.utils.validation import check_is_fitted
   
   class MeanImputer(BaseEstimator, TransformerMixin):
       # ... (之前的 __init__ 和 fit 方法) ...
   
       def fit(self, X, y=None):
           self.means_ = np.nanmean(X, axis=0)
           # 存储在 fit 期间看到的特征数量
           self.n_features_in_ = X.shape[1]
           return self
   
       def transform(self, X):
           # 检查是否已调用 fit
           check_is_fitted(self, 'means_')
   
           # 输入验证（可选但建议）
           # X = self._validate_data(X, accept_sparse=False, reset=False) # 需要较新的 scikit-learn 版本
   
           # 检查输入 X 是否具有与用于 fit 的数据相同数量的特征
           if X.shape[1] != self.n_features_in_:
               raise ValueError(f"Input X has {X.shape[1]} features, but MeanImputer expects {self.n_features_in_} features as input.")
   
           # 创建副本以避免修改原始数据
           X_transformed = X.copy()
   
           # 使用学习到的均值 means_ 应用转换
           for i in range(X.shape[1]):
               # 在当前列中查找 NaN 索引
               nan_indices = np.isnan(X_transformed[:, i])
               # 将 NaN 替换为该列的学习均值
               X_transformed[nan_indices, i] = self.means_[i]
   
           return X_transformed
   

   在 transform 方法中：

   • X：要转换的数据。
   • 它通常以 check_is_fitted(self) 开始，以确保 fit 已被调用。
   • 它使用学习到的参数（例如，self.means_）来修改 X。
   • 它应处理 X 可能具有与预期不同维度或类型的情况（输入验证）。添加对 fit 与 transform 期间看到的特征数量的检查非常重要。
   • 它返回 X 的转换版本。

示例：列选择器转换器

我们来创建一个简单的转换器，从 Pandas DataFrame 中选择特定列。当您希望在流水线中对不同列子集应用不同转换时，这很有用。

import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin

class ColumnSelector(BaseEstimator, TransformerMixin):
    """从 Pandas DataFrame 中选择指定列。"""
    def __init__(self, columns):
        # 检查 columns 是列表还是单个字符串
        if not isinstance(columns, list):
            self.columns = [columns] # 确保它是一个列表
        else:
            self.columns = columns

    def fit(self, X, y=None):
        # 没有要学习的参数，只需返回 self
        # 可选：如果 X 始终是 DataFrame，可以在此处添加验证以检查列是否存在于 X 中
        return self

    def transform(self, X):
        # 确保 X 是 DataFrame
        if not isinstance(X, pd.DataFrame):
            raise TypeError("Input X must be a Pandas DataFrame for ColumnSelector.")

        # 检查列是否存在（转换过程中的重要检查）
        missing_cols = set(self.columns) - set(X.columns)
        if missing_cols:
            raise ValueError(f"The following columns are missing from the DataFrame: {list(missing_cols)}")

        # 选择并返回指定的列
        return X[self.columns]

# --- 用法示例 ---
data = {'feature1': [1, 2, np.nan, 4],
        'feature2': [5, 6, 7, 8],
        'feature3': ['A', 'B', 'A', 'C']}
df = pd.DataFrame(data)

# 选择 'feature1' 和 'feature3'
selector = ColumnSelector(columns=['feature1', 'feature3'])

# 在此情况下 fit 不做任何事
selector.fit(df)

# transform 应用选择
transformed_df = selector.transform(df)
print("已选择的列:")
print(transformed_df)

# --- Mean Imputer 用法示例 ---
data_numeric = {'col_a': [1, 2, np.nan, 4, 5],
                'col_b': [np.nan, 7, 8, 9, 10]}
df_numeric = pd.DataFrame(data_numeric)

imputer = MeanImputer()

# fit 学习均值
imputer.fit(df_numeric.values) # 使用 .values 传递 NumPy 数组
print(f"\n学习到的均值: {imputer.means_}")

# transform 填充 NaN
transformed_numeric = imputer.transform(df_numeric.values)
print("\n填充后的数据:")
print(transformed_numeric)

集成到流水线中

自定义转换器的真正效用来自它们与 Scikit-learn Pipeline 对象的集成。它们可与内置转换器和估计器混合搭配使用。

from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler, OneHotEncoder
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer # 用于比较的内置方法

# 需要不同处理的示例数据
data_complex = {'numeric1': [1, 2, np.nan, 4],
                'numeric2': [10, 20, 30, 40],
                'category1': ['A', 'B', 'A', 'C'],
                'category2': ['X', 'X', 'Y', 'Y']}
df_complex = pd.DataFrame(data_complex)

# 定义列类型
numeric_features = ['numeric1', 'numeric2']
categorical_features = ['category1', 'category2']

# 为数值和类别特征创建预处理流水线
# 使用我们的自定义 MeanImputer（假设它如上定义）
# 注意：Scikit-learn 的 SimpleImputer 通常更适合生产环境
# 但我们在此处使用 MeanImputer 进行演示。
numeric_transformer_custom = Pipeline(steps=[
    ('selector', ColumnSelector(columns=numeric_features)), # 使用我们的自定义选择器
    ('imputer', MeanImputer()), # 使用我们的自定义填充器
    ('scaler', StandardScaler())
])

# 或使用内置填充器
numeric_transformer_builtin = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='mean')),
    ('scaler', StandardScaler())
])

categorical_transformer = Pipeline(steps=[
    ('imputer', SimpleImputer(strategy='most_frequent')),
    ('onehot', OneHotEncoder(handle_unknown='ignore'))
])

# 使用 ColumnTransformer 将不同转换器应用于不同列
# 选项 1：使用内置选择器 + 我们的填充器 + 缩放器
preprocessor_v1 = ColumnTransformer(
    transformers=[
        ('num', numeric_transformer_builtin, numeric_features), # 使用内置
        ('cat', categorical_transformer, categorical_features)
    ],
    remainder='passthrough' # 保留其他列（如果有）
)

# 选项 2：在数值流水线中使用我们的 ColumnSelector
# 注意：ColumnTransformer 直接将转换器应用于指定列，
# 因此在此处将 ColumnSelector 嵌入到数值流水线中是多余的，
# 但它演示了在需要更复杂路由时如何使用它。
# 使用 ColumnTransformer 的更简洁方法已在 preprocessor_v1 中展示。

# 拟合并转换数据
processed_data = preprocessor_v1.fit_transform(df_complex)

print("\n处理后数据的形状:", processed_data.shape)
# 注意：ColumnTransformer 之后输出将是 NumPy 数组

最佳实践和注意事项

• 无状态 transform：确保 transform 仅依赖于在 fit 中学习到的参数（以 _ 结尾的属性）以及在 __init__ 中设置的超参数。它不应修改转换器的状态。
• 输入验证：使用 Scikit-learn 工具，如 check_is_fitted 以及可能的 _validate_data（在较新版本中），或在 transform 内部进行手动检查，以确保输入数据具有预期格式（例如，特征数量）。
• 在 fit 中返回 self：这对于流水线兼容性而言是必须的。
• 不变性：避免在 transform 中原地修改输入 X。返回一个新的数组或 DataFrame。
• 测试：彻底测试您的自定义转换器，最好使用 Scikit-learn 的 check_estimator 工具（from sklearn.utils.estimator_checks import check_estimator）来验证其是否符合 API。
• 特征名称：对于添加、删除或重新排序特征的转换器，请考虑实现 get_feature_names_out() 方法（在较新的 Scikit-learn 版本中引入），以实现更好的集成和自省能力，尤其是在整个流水线中使用 Pandas DataFrames 时。

遵循这些模式，您可以创建可复用的数据转换组件，针对您特定的机器学习问题定制，从而扩展 Scikit-learn 框架的功能和灵活性。