用于改变X和y的sklearn管道的自定义变换器

Question

我想创建自己的变换器以与sklearn Pipeline一起使用.因此,我正在创建一个实现fit和transform方法的类.变压器的目的将是从具有比的NaN指定数目的多个矩阵中删除行.所以我面临的问题是如何更改传递给变压器的X和y矩阵？我相信这必须在fit方法中完成,因为它可以同时访问X和y.因为一旦我将X重新分配给具有较少行的新矩阵,python就会通过赋值传递参数,因此对原始X的引用将丢失(当然对于y也是如此).是否可以保留此参考？

我正在使用pandas DataFrame来轻松删除具有太多NaN的行,这可能不适合我的用例.当前代码如下所示:

class Dropna():

    # thresh is max number of NaNs allowed in a row
    def __init__(self, thresh=0):
        self.thresh = thresh

    def fit(self, X, y):
        total = X.shape[1]
        # +1 to account for 'y' being added to the dframe                                                                                                                            
        new_thresh = total + 1 - self.thresh
        df = pd.DataFrame(X)
        df['y'] = y
        df.dropna(thresh=new_thresh, inplace=True)
        X = df.drop('y', axis=1).values
        y = df['y'].values
        return self

    def transform(self, X):
        return X

Answer 1

修改样本轴,例如删除样本,(但是？)不符合scikit-learn变换器API.因此,如果您需要这样做,您应该在任何调用scikit学习之外进行预处理.

就像现在一样,变换器API用于将给定样本的特征转换为新的特征.这可以隐含地包含来自其他样本的信息,但是样本永远不会被删除.

另一种选择是尝试估算缺失值.但同样,如果您需要删除样本,请在使用scikit learn之前将其视为预处理.

Answer 2

您必须修改 sklearn 的内部代码Pipeline。

我们定义一个转换器，用于删除在拟合期间至少特征或目标的值为 NaN 的样本 ( fit_transform)。而它在推理过程中删除了至少特征值为 NaN 的样本 ( transform)。需要注意的是，我们的转换器返回 X 和 y，fit_transform因此我们需要在 sklearn 中处理此行为Pipeline。

class Dropna():

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

    def fit_transform(self, X, y):
        
        mask = (np.isnan(X).any(-1) | np.isnan(y))
        if hasattr(X, 'loc'):
            X = X.loc[~mask]
        else:
            X = X[~mask]
        if hasattr(y, 'loc'):
            y = y.loc[~mask]
        else:
            y = y[~mask]
        
        return X, y   ###### make fit_transform return X and y
    
    def transform(self, X):
        
        mask = np.isnan(X).any(-1)
        if hasattr(X, 'loc'):
            X = X.loc[~mask]
        else:
            X = X[~mask]
        
        return X

我们只需修改原来的sklearnPipeline中的两个特定点fit即可_fit。其余保持不变。

from sklearn import pipeline
from sklearn.base import clone
from sklearn.utils import _print_elapsed_time
from sklearn.utils.validation import check_memory

class Pipeline(pipeline.Pipeline):

    def _fit(self, X, y=None, **fit_params_steps):
        self.steps = list(self.steps)
        self._validate_steps()
        memory = check_memory(self.memory)

        fit_transform_one_cached = memory.cache(pipeline._fit_transform_one)

        for (step_idx, name, transformer) in self._iter(
            with_final=False, filter_passthrough=False
        ):
                        
            if transformer is None or transformer == "passthrough":
                with _print_elapsed_time("Pipeline", self._log_message(step_idx)):
                    continue

            try:
                # joblib >= 0.12
                mem = memory.location
            except AttributeError:
                mem = memory.cachedir
            finally:
                cloned_transformer = clone(transformer) if mem else transformer

            X, fitted_transformer = fit_transform_one_cached(
                cloned_transformer,
                X,
                y,
                None,
                message_clsname="Pipeline",
                message=self._log_message(step_idx),
                **fit_params_steps[name],
            )
            
            if isinstance(X, tuple):    ###### unpack X if is tuple: X = (X,y)
                X, y = X
            
            self.steps[step_idx] = (name, fitted_transformer)
        
        return X, y
    
    def fit(self, X, y=None, **fit_params):
        fit_params_steps = self._check_fit_params(**fit_params)
        Xt = self._fit(X, y, **fit_params_steps)
        
        if isinstance(Xt, tuple):    ###### unpack X if is tuple: X = (X,y)
            Xt, y = Xt 
        
        with _print_elapsed_time("Pipeline", self._log_message(len(self.steps) - 1)):
            if self._final_estimator != "passthrough":
                fit_params_last_step = fit_params_steps[self.steps[-1][0]]
                self._final_estimator.fit(Xt, y, **fit_params_last_step)

        return self

这是为了解压Dropna().fit_transform(X, y)newX和中生成的值所必需的y。

这是完整的工作流程：

from sklearn.linear_model import Ridge

X = np.random.uniform(0,1, (100,3))
y = np.random.uniform(0,1, (100,))
X[np.random.uniform(0,1, (100)) < 0.1] = np.nan
y[np.random.uniform(0,1, (100)) < 0.1] = np.nan

pipe = Pipeline([('dropna', Dropna()), ('model', Ridge())])
pipe.fit(X, y)

pipe.predict(X).shape

另一项带有进一步中间预处理步骤的试验：

from sklearn.preprocessing import StandardScaler

pipe = Pipeline([('dropna', Dropna()), ('scaler', StandardScaler()), ('model', Ridge())])
pipe.fit(X, y)

pipe.predict(X).shape

根据需要可以通过其他简单修改来实现更复杂的行为。如果您也有兴趣Pipeline().fit_transform或者Pipeline().fit_predict您需要进行相同的更改。

Answer 3

imblearn构建在 之上的包sklearn包含一个估计器FunctionSampler ，它允许在管道步骤中操作特征数组X和目标数组。y

请注意，在管道步骤中使用它需要使用继承自 in 中Pipeline的类。此外，默认情况下，在 的上下文中，如果没有立即调用该方法（如），则该方法不会执行任何操作。因此，请提前阅读文档。imblearnsklearnPipelineresamplefitfit_resample