如何在OneVsOneClassifier流水线中逐对执行自定义欠采样

API 探路者

2026-8-28

问题说明

你正在实现基于SVM的自定义欠采样器NearSVUmdersampler，核心逻辑为：接收二分类数据集，训练RBF核SVC模型后，筛选多数类中靠近支持向量的样本，将多数类样本量下采样至与少数类一致，最终返回类别平衡的二分类数据集，类的初始实现代码如下：

import numpy as np
from collections import Counter
from sklearn.svm import SVC

class NearSVUmdersampler():
  def __init__(self, random_state=None):
    self.random_state = random_state
  
  def fit_resample(self, X, y):
    random_state = check_random_state(self.random_state)
    # 统计类别分布
    counter = Counter(y)
    maj_class = counter.most_common()[0][0]
    min_class = counter.most_common()[-1][0]
    num_minority = len(X[ y == min_class])
    svc = SVC(kernel='rbf', random_state=32)
    svc.fit(X,y)
    # 提取多数类对应的支持向量
    maj_sup_vector = svc.support_vectors_[maj_class]
    # 计算多数类样本到支持向量的距离
    distances = []
    for i, x in enumerate(X[y == maj_class]):
      d = np.linalg.norm(maj_sup_vector - x) 
      distances.append((i, d))
    # 按距离升序排序，选取与少数类等量的最近邻多数类样本
    distances.sort(key=lambda tup: tup[1])
    index = [i for i, d in distances][:num_minority]
    X_ds = np.concatenate((X[y == maj_class][index], X[y == min_class]))
    y_ds = np.concatenate((y[y == maj_class][index], y[y == min_class]))
    return X_ds, y_ds

你需要将该欠采样器接入多分类任务的imblearn.pipeline.Pipeline，采用one-vs-one（ovo）训练策略时，每训练一对类别的二分类子模型，都自动调用欠采样器仅对当前两个类别的样本做平衡处理。以4分类测试数据集为例，训练集类别分布为{0: 126, 1: 192, 2: 330, 3: 952}，共需训练6个ovo二分类子模型，每个子模型都将当前类别对中的多数类下采样至对应少数类的样本量。
当前采用的流水线结构为：第一步放置自定义欠采样器，第二步放置基估计器为SVC(kernel='rbf')的OneVsOneClassifier，调用fit方法训练时，欠采样器仅在全量多分类训练集上执行一次，仅保留全局最少数类（类别0）和全局最多数类（类别3）的平衡样本，其余类别样本全部丢失，仅能完成1个二分类子模型的训练，其余ovo子模型无法正常工作。需要调整现有流水线结构，实现每对ovo子模型训练前单独调用自定义欠采样器完成对应类别对样本平衡的需求。

解决方案

问题根因

当前结构失效的核心原因是重采样的作用域错误：imblearn.Pipeline的重采样步骤只会在全量输入数据集上全局执行1次，完全不会感知后续OneVsOneClassifier内部拆分二分类子集的逻辑，因此会在全局重采样阶段丢弃其他类别样本，导致后续OVO子模型无法获取对应类别的训练数据。

具体实现步骤

第一步：修复自定义欠采样器的原有逻辑bug
初始代码存在两处可导致运行错误的问题：

缺失check_random_state的导入
多数类支持向量提取逻辑错误：SVC.support_vectors_是按索引排列的所有支持向量数组，并非以类别为键的字典，无法直接用类别值做索引取值，需要结合SVC.support_（支持向量对应原训练集的索引位置）匹配标签筛选多数类支持向量。

修正后的欠采样器代码如下：

import numpy as np
from collections import Counter
from sklearn.svm import SVC
from sklearn.utils import check_random_state

class NearSVUmdersampler():
  def __init__(self, random_state=None):
    self.random_state = random_state
  
  def fit_resample(self, X, y):
    random_state = check_random_state(self.random_state)
    # 统计类别分布
    counter = Counter(y)
    maj_class = counter.most_common()[0][0]
    min_class = counter.most_common()[-1][0]
    num_minority = len(X[y == min_class])
    svc = SVC(kernel='rbf', random_state=32)
    svc.fit(X,y)
    # 筛选属于多数类的支持向量
    sv_maj_mask = y[svc.support_] == maj_class
    maj_sup_vectors = svc.support_vectors_[sv_maj_mask]
    # 计算多数类样本到多数类支持向量的最小距离
    distances = []
    maj_X = X[y == maj_class]
    for i, x in enumerate(maj_X):
      # 取样本到所有多数类支持向量的最小距离作为排序依据
      d = np.min(np.linalg.norm(maj_sup_vectors - x, axis=1))
      distances.append((i, d))
    # 按距离升序排序，选取与少数类等量的靠近支持向量的多数类样本
    distances.sort(key=lambda tup: tup[1])
    index = [i for i, d in distances][:num_minority]
    X_ds = np.concatenate((maj_X[index], X[y == min_class]))
    y_ds = np.concatenate((y[y == maj_class][index], y[y == min_class]))
    return X_ds, y_ds

注：原距离计算逻辑直接对支持向量矩阵和单样本做范数计算，返回的是矩阵差值的F范数，并非样本到每个支持向量的距离，修正后取样本到所有多数类支持向量的最小距离作为筛选依据，更符合"靠近支持向量"的筛选逻辑

第二步：调整流水线嵌套结构
不要把欠采样器放在OneVsOneClassifier的外层，而是将欠采样器和RBF核SVC打包成一个二分类专用的imblearn流水线，再把这个二分类流水线作为基估计器传入OneVsOneClassifier。
这种结构下，OVO每拆分出一对类别的二分类子集，就会自动触发二分类流水线的执行：先对当前两个类别的子集做欠采样平衡，再训练子SVC模型，完全满足逐对子集平衡的需求。
正确的流水线构建代码如下：

from imblearn.pipeline import Pipeline as ImbPipeline
from sklearn.multiclass import OneVsOneClassifier
from sklearn.svm import SVC

# 构建二分类基估计器流水线：先欠采样，再训练SVC
binary_clf = ImbPipeline(steps=[
    ('nearsv_under', NearSVUmdersampler(random_state=42)),
    ('svc', SVC(kernel='rbf', random_state=42))
])

# 将二分类流水线作为OVO的基估计器
ovo_clf = OneVsOneClassifier(estimator=binary_clf)

# 外层如果需要加其他预处理（比如标准化），可以放在imblearn流水线里
# 注意不要把欠采样器放在这一层
full_pipeline = ImbPipeline(steps=[
    # 其他预处理步骤比如scaler放这里
    ('ovo_clf', ovo_clf)
])

# 正常调用fit即可，内部会自动对每对OVO子集做欠采样
# full_pipeline.fit(X_train, y_train)

该结构在4分类数据集上运行时，6个OVO二分类子模型都会独立拿到对应两个类别的样本，自动完成类对内部的样本平衡，不会出现全局重采样丢失样本的问题。

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

如何在OneVsOneClassifier流水线中逐对执行自定义欠采样

问题根因 #

具体实现步骤 #

问题根因

具体实现步骤