Sklearn Pipeline中IsolationForest离群值移除器报错问题

错误原因 #

你的IsolationForestOutlierRemover类违反了Sklearn Transformer的接口规范：Transformer的transform方法只能返回特征矩阵X，不能同时返回X和y。

当你把这个类放进Pipeline后，Pipeline会将transform返回的元组(X_filtered, y_filtered)直接作为下一个步骤（随机森林）的输入X，导致随机森林收到的是一个包含两个数组的元组，而非正常的特征矩阵，从而触发形状不匹配的报错。

另外，即使你修改transform只返回X，Pipeline默认仍会将原始的y传递给下一个步骤，导致过滤后的X（95个样本）和原始y（100个样本）样本数不匹配，同样会报错。

修复方案 #

方案1：调整自定义Transformer，配合手动过滤y（适合简单场景） #

先修改IsolationForestOutlierRemover的transform方法，只返回过滤后的X：

class IsolationForestOutlierRemover(BaseEstimator, TransformerMixin):
    def __init__(self, contamination=0.05):
        self.contamination = contamination
        self.isolation_forest = IsolationForest(contamination=self.contamination)

    def fit(self, X, y=None):
        self.isolation_forest.fit(X)
        mask = self.isolation_forest.predict(X) == 1
        self.mask = mask
        return self

    def transform(self, X, y=None):
        # 只返回过滤后的特征矩阵X
        return X[self.mask]

    def fit_transform(self, X, y=None):
        self.fit(X, y)
        if y is not None:
            return X[self.mask], y[self.mask]
        else:
            return X[self.mask]

然后在训练时，先手动过滤X和y，再传入Pipeline：

# 先过滤离群值
X_filtered, y_filtered = IsolationForestOutlierRemover().fit_transform(X1, y1)

# 用过滤后的数据集训练Pipeline
pipelinet = Pipeline(
    [
        ("random_forest", RandomForestClassifier()),
    ]
)
pipelinet.fit(X_filtered, y_filtered)

方案2：自定义MetaEstimator，实现端到端的离群值移除+模型训练（适合Pipeline集成） #

如果你想把整个流程封装成一个可复用的组件，可以自定义一个MetaEstimator，内部先处理离群值过滤，再训练模型：

from sklearn.base import BaseEstimator
from sklearn.ensemble import IsolationForest, RandomForestClassifier
import numpy as np

class IsolationForestOutlierRemover(BaseEstimator, TransformerMixin):
    def __init__(self, contamination=0.05):
        self.contamination = contamination
        self.isolation_forest = IsolationForest(contamination=self.contamination)

    def fit(self, X, y=None):
        self.isolation_forest.fit(X)
        mask = self.isolation_forest.predict(X) == 1
        self.mask = mask
        return self

    def transform(self, X, y=None):
        if y is not None:
            return X[self.mask], y[self.mask]
        else:
            return X[self.mask]

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

# 自定义MetaEstimator，整合离群值移除和分类器
class OutlierResistantClassifier(BaseEstimator):
    def __init__(self, outlier_remover, classifier):
        self.outlier_remover = outlier_remover
        self.classifier = classifier
    
    def fit(self, X, y):
        # 先过滤离群值
        X_filtered, y_filtered = self.outlier_remover.fit_transform(X, y)
        # 训练分类器
        self.classifier.fit(X_filtered, y_filtered)
        return self
    
    def predict(self, X):
        # 预测时不需要过滤（离群值移除只在训练时做）
        return self.classifier.predict(X)
    
    def predict_proba(self, X):
        return self.classifier.predict_proba(X)

使用方式：

# 初始化整合后的模型
clf = OutlierResistantClassifier(
    outlier_remover=IsolationForestOutlierRemover(contamination=0.05),
    classifier=RandomForestClassifier()
)

# 直接用原始数据训练
clf.fit(X1, y1)
# 预测
predictions = clf.predict(X1)

完整可运行代码 #

from sklearn.datasets import make_classification
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.ensemble import IsolationForest, RandomForestClassifier
import numpy as np

# 生成测试数据
X1, y1 = make_classification(n_samples=100, n_features=10, n_informative=5, n_classes=3)

# 自定义离群值移除器
class IsolationForestOutlierRemover(BaseEstimator, TransformerMixin):
    def __init__(self, contamination=0.05):
        self.contamination = contamination
        self.isolation_forest = IsolationForest(contamination=self.contamination)

    def fit(self, X, y=None):
        self.isolation_forest.fit(X)
        mask = self.isolation_forest.predict(X) == 1
        self.mask = mask
        return self

    def transform(self, X, y=None):
        if y is not None:
            return X[self.mask], y[self.mask]
        else:
            return X[self.mask]

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

# 自定义整合模型
class OutlierResistantClassifier(BaseEstimator):
    def __init__(self, outlier_remover, classifier):
        self.outlier_remover = outlier_remover
        self.classifier = classifier
    
    def fit(self, X, y):
        X_filtered, y_filtered = self.outlier_remover.fit_transform(X, y)
        self.classifier.fit(X_filtered, y_filtered)
        return self
    
    def predict(self, X):
        return self.classifier.predict(X)

# 训练并测试
clf = OutlierResistantClassifier(
    outlier_remover=IsolationForestOutlierRemover(contamination=0.05),
    classifier=RandomForestClassifier()
)
clf.fit(X1, y1)
print("预测结果：", clf.predict(X1[:5]))

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