处理不平衡数据:当 99% 准确率是个谎言

处理不平衡数据:当 99% 准确率是个谎言

本节摘要:当你 99% 的数据是「正常」时,准确率是个谎言。你搭了个欺诈检测模型,拿到 99.9% 准确率,正庆祝,却发现它对每笔交易都预测「非欺诈」。这不是 bug——当只有 0.1% 交易欺诈时,永远猜多数类是最小化总体误差的理性选择,模型技术上正确、完全无用。这种事在真实分类要紧的地方处处发生:疾病诊断阳性率 1%,网络入侵攻击 0.01%,制造缺陷 0.5%,垃圾过滤 20%,流失预测流失者 5%——少数类越要紧,它往往越稀有。准确率失败因为它把所有正确预测同等对待:正确标记合法交易和正确抓欺诈都算一分,但抓欺诈是模型存在的全部理由。本节讲透为何准确率危险、改用 F1/AUPRC/MCC,从零实现 SMOTE(合成少数类过采样)、随机过/欠采样、类权重、阈值调优,并搭一条结合 SMOTE+类权重+阈值优化的完整不平衡数据流水线。

学习目标

阅读完本节,你应当能够:

1. 从零实现 SMOTE,解释合成过采样与随机复制的区别。
2. 用 F1、AUPRC、马修斯相关系数(MCC)代替准确率评估不平衡分类器。
3. 对比类权重、阈值调优、重采样策略,为给定不平衡比选对方法。
4. 搭一条结合 SMOTE、类权重、阈值优化的完整不平衡数据流水线。

一、问题与直觉

你搭了个欺诈检测模型。它拿 99.9% 准确率。你庆祝。然后你发现它对每笔交易都预测「非欺诈」。

这不是 bug。当只有 0.1% 交易欺诈时,永远猜多数类最小化总体误差,模型学到了这一点。它技术上正确、完全无用。

这种事在真实分类要紧的地方处处发生。疾病诊断:1% 阳性率。网络入侵:0.01% 攻击。制造缺陷:0.5% 缺陷。垃圾过滤:20% 垃圾。流失预测:5% 流失者。少数类越要紧,它往往越稀有。

准确率失败,因为它把所有正确预测同等对待。正确标记合法交易和正确抓欺诈都算一分准确率。但抓欺诈是模型存在的全部理由。我们需要指标、技术、训练策略,强制模型关注稀有但重要的类。

为何准确率失败

考虑一个 1000 样本的数据集:990 负、10 正。一个永远预测负的模型:

准确率 = (0 + 990) / 1000 = 99.0%

模型抓住零欺诈、零疾病、零缺陷。但准确率说 99%。这就是为何准确率对不平衡问题危险。

更好的指标

精确率(Precision) = TP / (TP + FP)。所有被标为正的里,实际多少是?高精确率意味假告警少。

召回率(Recall) = TP / (TP + FN)。所有实际为正的里,我们抓了多少?高召回率意味漏报少。

F1 分数 = 2 * 精确率 * 召回率 / (精确率 + 召回率)。调和平均。比算术平均更惩罚精确率与召回率的极端失衡。

F-beta 分数 = (1 + beta^2) * 精确率 * 召回率 / (beta^2 * 精确率 + 召回率)。beta > 1 时召回率更重要。beta < 1 时精确率更重要。F2 在欺诈检测里常见(漏欺诈比假告警更糟)。

AUPRC(精确率-召回率曲线下面积)。类似 AUC-ROC 但对不平衡数据信息量更大。随机分类器的 AUPRC 等于正类率(不像 ROC 的 0.5),这让改进更容易看出。

马修斯相关系数(MCC) = (TPTN - FPFN) / sqrt((TP+FP)(TP+FN)(TN+FP)(TN+FN))。范围 -1 到 +1。只有模型在两类上都好才给高分。即使两类大小差异巨大也平衡。

对上面那个「永远预测负」的模型:精确率 = 0/0(未定义,常设 0),召回率 = 0/10 = 0,F1 = 0,MCC = 0。这些指标正确地识别出模型一文不值。

不平衡数据流水线

SMOTE:合成少数类过采样技术

随机过采样复制已有少数类样本。这有效但有过拟合风险,因为模型反复看到相同点。

SMOTE 创造新的、合理但非复制的合成少数类样本。算法:

1. 对每个少数类样本 x,在其他少数类样本里找其 k 近邻。
2. 随机选一个邻居。
3. 在 x 和那个邻居的连线段上造一个新样本。

公式:new_sample = x + random(0, 1) * (neighbor - x)

这在真实少数类点之间插值,在与它们相同的特征空间区域创造样本,而非简单复制已有数据。

采样策略对比

随机过采样:复制少数类样本以匹配多数类计数。

• 优点:简单,无信息损失
• 缺点:精确复制导致过拟合,增加训练时间

随机欠采样:移除多数类样本以匹配少数类计数。

• 优点:训练快,简单
• 缺点:丢弃可能有用的多数类数据,方差更高

SMOTE:通过插值创造合成少数类样本。

• 优点:生成新数据点,相比随机过采样降过拟合
• 缺点:可能在决策边界附近造噪声样本,不考虑多数类分布

类权重

不改数据,改模型如何对待错误。给错分少数类赋更高权重。

对一个 950 负、50 正的二元问题:

• 负类权重 = n_samples / (2 * n_negative) = 1000 / (2 * 950) = 0.526
• 正类权重 = n_samples / (2 * n_positive) = 1000 / (2 * 50) = 10.0

正类拿到 19 倍权重。错分一个正样本的代价等于错分 19 个负样本。模型被迫关注少数类。

在逻辑回归里,这修改损失函数:

weighted_loss = -sum(w_i * [y_i * log(p_i) + (1-y_i) * log(1-p_i)])
​

其中 w_i 取决于样本 i 的类。

类权重在期望上数学等价于过采样,但不创造新数据点。这使其更快,且避免复制样本的过拟合风险。

阈值调优

多数分类器输出概率。默认阈值 0.5:若 P(正) >= 0.5,预测正。但 0.5 是任意的。当类别不平衡时,最优阈值通常低得多。

流程:

1. 训练模型。
2. 在验证集上拿预测概率。
3. 扫阈值 0.0 到 1.0。
4. 在每个阈值算 F1(或你选的指标)。
5. 选最大化你指标的阈值。

一个模型可能对一笔欺诈交易输出 P(欺诈)=0.15。阈值 0.5 时归为非欺诈。阈值 0.10 时被正确抓住。概率校准不如排序重要——只要欺诈拿到比非欺诈更高的概率,就存在一个能分开它们的阈值。

代价敏感学习

类权重的推广。不用统一代价,赋具体错分代价:

漏一笔欺诈交易(FN)的代价是假告警(FP)的 100 倍。模型优化总代价,而非总错误数。

当你能估计真实世界代价时,这是最有原则的方法。漏诊癌症的代价与导致额外活检的假告警代价天差地别。把这些代价显式化,强制正确的权衡。

决策流程图

二、从零实现

第 1 步:生成不平衡数据集

import numpy as np


def make_imbalanced_data(n_majority=950, n_minority=50, seed=42):
    rng = np.random.RandomState(seed)

    X_maj = rng.randn(n_majority, 2) * 1.0 + np.array([0.0, 0.0])
    X_min = rng.randn(n_minority, 2) * 0.8 + np.array([2.5, 2.5])

    X = np.vstack([X_maj, X_min])
    y = np.concatenate([np.zeros(n_majority), np.ones(n_minority)])

    shuffle_idx = rng.permutation(len(y))
    return X[shuffle_idx], y[shuffle_idx]
​

第 2 步:从零实现 SMOTE

def euclidean_distance(a, b):
    return np.sqrt(np.sum((a - b) ** 2))


def find_k_neighbors(X, idx, k):
    distances = []
    for i in range(len(X)):
        if i == idx:
            continue
        d = euclidean_distance(X[idx], X[i])
        distances.append((i, d))
    distances.sort(key=lambda x: x[1])
    return [d[0] for d in distances[:k]]


def smote(X_minority, k=5, n_synthetic=100, seed=42):
    rng = np.random.RandomState(seed)
    n_samples = len(X_minority)
    k = min(k, n_samples - 1)
    synthetic = []

    for _ in range(n_synthetic):
        idx = rng.randint(0, n_samples)
        neighbors = find_k_neighbors(X_minority, idx, k)
        neighbor_idx = neighbors[rng.randint(0, len(neighbors))]
        t = rng.random()
        new_point = X_minority[idx] + t * (X_minority[neighbor_idx] - X_minority[idx])
        synthetic.append(new_point)

    return np.array(synthetic)
​

第 3 步:随机过采样和欠采样

def random_oversample(X, y, seed=42):
    rng = np.random.RandomState(seed)
    classes, counts = np.unique(y, return_counts=True)
    max_count = counts.max()

    X_resampled = list(X)
    y_resampled = list(y)

    for cls, count in zip(classes, counts):
        if count < max_count:
            cls_indices = np.where(y == cls)[0]
            n_needed = max_count - count
            chosen = rng.choice(cls_indices, size=n_needed, replace=True)
            X_resampled.extend(X[chosen])
            y_resampled.extend(y[chosen])

    X_out = np.array(X_resampled)
    y_out = np.array(y_resampled)
    shuffle = rng.permutation(len(y_out))
    return X_out[shuffle], y_out[shuffle]


def random_undersample(X, y, seed=42):
    rng = np.random.RandomState(seed)
    classes, counts = np.unique(y, return_counts=True)
    min_count = counts.min()

    X_resampled = []
    y_resampled = []

    for cls in classes:
        cls_indices = np.where(y == cls)[0]
        chosen = rng.choice(cls_indices, size=min_count, replace=False)
        X_resampled.extend(X[chosen])
        y_resampled.extend(y[chosen])

    X_out = np.array(X_resampled)
    y_out = np.array(y_resampled)
    shuffle = rng.permutation(len(y_out))
    return X_out[shuffle], y_out[shuffle]
​

第 4 步:带类权重的逻辑回归

def sigmoid(z):
    return 1.0 / (1.0 + np.exp(-np.clip(z, -500, 500)))


def logistic_regression_weighted(X, y, weights, lr=0.01, epochs=200):
    n_samples, n_features = X.shape
    w = np.zeros(n_features)
    b = 0.0

    for _ in range(epochs):
        z = X @ w + b
        pred = sigmoid(z)
        error = pred - y
        weighted_error = error * weights

        gradient_w = (X.T @ weighted_error) / n_samples
        gradient_b = np.mean(weighted_error)

        w -= lr * gradient_w
        b -= lr * gradient_b

    return w, b


def compute_class_weights(y):
    classes, counts = np.unique(y, return_counts=True)
    n_samples = len(y)
    n_classes = len(classes)
    weight_map = {}
    for cls, count in zip(classes, counts):
        weight_map[cls] = n_samples / (n_classes * count)
    return np.array([weight_map[yi] for yi in y])
​

第 5 步:阈值调优

def find_optimal_threshold(y_true, y_probs, metric="f1"):
    best_threshold = 0.5
    best_score = -1.0

    for threshold in np.arange(0.05, 0.96, 0.01):
        y_pred = (y_probs >= threshold).astype(int)
        tp = np.sum((y_pred == 1) & (y_true == 1))
        fp = np.sum((y_pred == 1) & (y_true == 0))
        fn = np.sum((y_pred == 0) & (y_true == 1))

        if metric == "f1":
            precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0
            recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0
            score = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0
        elif metric == "recall":
            score = tp / (tp + fn) if (tp + fn) > 0 else 0.0
        elif metric == "precision":
            score = tp / (tp + fp) if (tp + fp) > 0 else 0.0

        if score > best_score:
            best_score = score
            best_threshold = threshold

    return best_threshold, best_score
​

第 6 步:评估函数

def confusion_matrix_values(y_true, y_pred):
    tp = np.sum((y_pred == 1) & (y_true == 1))
    tn = np.sum((y_pred == 0) & (y_true == 0))
    fp = np.sum((y_pred == 1) & (y_true == 0))
    fn = np.sum((y_pred == 0) & (y_true == 1))
    return tp, tn, fp, fn


def compute_metrics(y_true, y_pred):
    tp, tn, fp, fn = confusion_matrix_values(y_true, y_pred)
    accuracy = (tp + tn) / (tp + tn + fp + fn)
    precision = tp / (tp + fp) if (tp + fp) > 0 else 0.0
    recall = tp / (tp + fn) if (tp + fn) > 0 else 0.0
    f1 = 2 * precision * recall / (precision + recall) if (precision + recall) > 0 else 0.0

    denom = np.sqrt(float((tp + fp) * (tp + fn) * (tn + fp) * (tn + fn)))
    mcc = (tp * tn - fp * fn) / denom if denom > 0 else 0.0

    return {
        "accuracy": accuracy,
        "precision": precision,
        "recall": recall,
        "f1": f1,
        "mcc": mcc,
    }
​

第 7 步:对比所有方法

X, y = make_imbalanced_data(950, 50, seed=42)
split = int(0.8 * len(y))
X_train, X_test = X[:split], X[split:]
y_train, y_test = y[:split], y[split:]

# 基线:不处理
w_base, b_base = logistic_regression_weighted(
    X_train, y_train, np.ones(len(y_train)), lr=0.1, epochs=300
)
probs_base = sigmoid(X_test @ w_base + b_base)
preds_base = (probs_base >= 0.5).astype(int)

# 过采样
X_over, y_over = random_oversample(X_train, y_train)
w_over, b_over = logistic_regression_weighted(
    X_over, y_over, np.ones(len(y_over)), lr=0.1, epochs=300
)
preds_over = (sigmoid(X_test @ w_over + b_over) >= 0.5).astype(int)

# SMOTE
minority_mask = y_train == 1
X_minority = X_train[minority_mask]
synthetic = smote(X_minority, k=5, n_synthetic=len(y_train) - 2 * int(minority_mask.sum()))
X_smote = np.vstack([X_train, synthetic])
y_smote = np.concatenate([y_train, np.ones(len(synthetic))])
w_sm, b_sm = logistic_regression_weighted(
    X_smote, y_smote, np.ones(len(y_smote)), lr=0.1, epochs=300
)
preds_smote = (sigmoid(X_test @ w_sm + b_sm) >= 0.5).astype(int)

# 类权重
sample_weights = compute_class_weights(y_train)
w_cw, b_cw = logistic_regression_weighted(
    X_train, y_train, sample_weights, lr=0.1, epochs=300
)
probs_cw = sigmoid(X_test @ w_cw + b_cw)
preds_cw = (probs_cw >= 0.5).astype(int)

# 阈值调优(在留出验证集上调,不在测试集)
probs_val = sigmoid(X_val @ w_cw + b_cw)
best_thresh, best_f1 = find_optimal_threshold(y_val, probs_val, metric="f1")
preds_thresh = (probs_cw >= best_thresh).astype(int)
​

代码文件把这一切跑在一个脚本里并打印结果。

三、框架对比

用 scikit-learn 和 imbalanced-learn,这些技术都是一行:

from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report, f1_score
from sklearn.model_selection import train_test_split
from imblearn.over_sampling import SMOTE
from imblearn.under_sampling import RandomUnderSampler
from imblearn.pipeline import Pipeline

X_train, X_test, y_train, y_test = train_test_split(X, y, stratify=y)

model_weighted = LogisticRegression(class_weight="balanced")
model_weighted.fit(X_train, y_train)
print(classification_report(y_test, model_weighted.predict(X_test)))

smote = SMOTE(random_state=42)
X_resampled, y_resampled = smote.fit_resample(X_train, y_train)
model_smote = LogisticRegression()
model_smote.fit(X_resampled, y_resampled)
print(classification_report(y_test, model_smote.predict(X_test)))

pipeline = Pipeline([
    ("smote", SMOTE()),
    ("model", LogisticRegression(class_weight="balanced")),
])
pipeline.fit(X_train, y_train)
print(classification_report(y_test, pipeline.predict(X_test)))
​

从零实现精确展示每个技术做什么。SMOTE 就是少数类上的 k 近邻插值。类权重乘损失。阈值调优是截断点上的循环。无魔法。

⚠️ 关键:重采样必须只发生在训练折内部,不能在交叉验证前对全集做。imbalanced-learn 的 Pipeline(注意是 imblearn.pipeline.Pipeline,不是 sklearn 的)会在每折训练时单独重采样,正确防泄漏。若用 sklearn 的 Pipeline 配 SMOTE,会泄漏。

四、可复用产物

本节产出 outputs/skill-imbalanced-data.md——一个处理不平衡分类问题的决策清单,根据不平衡比、数据规模、业务代价给出方法选择和阈值设置建议。

五、练习

1. 边界 SMOTE(Borderline-SMOTE):修改 SMOTE 实现,只为靠近决策边界的少数类点(其 k 近邻里含多数类样本的点)生成合成样本。在类别重叠的数据集上对比标准 SMOTE。

2. 代价矩阵优化:实现代价敏感学习,代价矩阵作为参数。写一个函数,接受代价矩阵返回最小化期望代价的最优预测。用不同代价比(1:10、1:100、1:1000)测试,画精确率-召回率权衡如何变化。

3. 阈值校准:实现 Platt 缩放(在模型原始输出上拟合逻辑回归产出校准概率)。对比校准前后的精确率-召回率曲线。展示校准不改排序(AUC 不变)但让概率更有意义。

4. 平衡 bagging 集成:训练多个模型,每个在一个平衡的自助样本上(全部少数类 + 随机多数类子集),平均预测。对比单模型配 SMOTE。测性能和跨运行的方差。

5. 不平衡比实验:取一个平衡数据集,逐步增不平衡比(50/50、70/30、90/10、95/5、99/1)。每个比例,有 SMOTE 和无 SMOTE 各训一次。画 F1 对不平衡比。SMOTE 在哪个比例开始有意义?

本节要点回顾

1. 99% 准确率可能是谎言:0.1% 异常下永远猜多数类拿 99.9% 准确率却抓不到任何少数类——准确率把所有正确预测同等对待,不平衡问题上一文不值。
2. 改用 F1/AUPRC/MCC:精确率(TP/(TP+FP),假告警少)、召回率(TP/(TP+FN),漏报少)、F1(调和平均)、AUPRC(随机基线=正类率,改进更易看出)、MCC(只有两类都好才给高分)。
3. SMOTE 是少数类 k 近邻插值:new = x + random(0,1) * (neighbor - x),在真实点之间造合理样本而非复制,降过拟合。
4. 三种采样策略各有取舍:过采样(复制,过拟合风险)、欠采样(移除,信息损失)、SMOTE(合成,边界噪声)。
5. 类权重在期望上等价于过采样:权重 = n_samples / (n_classes * count),但更快且不创造新点,避免复制样本过拟合。
6. 阈值调优:0.5 是任意的:不平衡下最优阈值通常低得多;扫阈值最大化 F1(或你选的指标),概率校准不如排序重要。
7. 代价敏感学习最有原则:赋具体错分代价(C_FN、C_FP),模型优化总代价而非错误数;当能估计真实代价时用。
8. 决策按不平衡比分档:轻度(<70/30)类权重、中度(70/30~95/5)SMOTE+阈值、重度(>95/5)组合多策略;数据规模也影响选择。
9. 重采样只在训练折内做:用 imblearn 的 Pipeline(不是 sklearn 的)按折重采样防泄漏;在全集重采样再交叉验证会泄漏。
10. 业务代价驱动阈值:欺诈检测漏报代价高设低阈值,设备维护假告警代价高设高阈值;基于假阳性/假阴性成本比选最低成本点。

下一节(也是本章最后一节),我们讲特征选择——过滤器、包装器、嵌入法如何去掉噪声特征、留下真正承载目标信息的特征,让模型训练更快、泛化更好、可解释性更强。