处理不平衡数据

问题

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

这不是 bug。当只有 0.1% 的交易是欺诈时，模型做出这样的预测是理性的。它学到一直猜多数类可以最小化整体误差。技术上正确，但完全没用。

在所有真正重要的分类场景中都会发生这种情况。疾病诊断：阳性率 1%。网络入侵：攻击占 0.01%。制造缺陷：次品率 0.5%。垃圾邮件过滤：垃圾占 20%。用户流失预测：流失者占 5%。少数类越重要，往往就越罕见。

准确率失效，因为它把所有正确预测同等看待。正确标记一笔正常交易和正确拦截欺诈都算 1 分准确率。但拦截欺诈才是这个模型存在的全部意义。我们需要指标、技术和训练策略，迫使模型关注罕见但重要的类别。

概念

为什么准确率会失效

考虑一个 1000 个样本的数据集：990 个负样本，10 个正样本。一个总是预测负样本的模型：

这个模型没抓到任何欺诈、任何疾病、任何缺陷。但准确率显示 99%。这就是准确率在不平衡问题中危险的原因。

更好的指标

精确率（Precision） = \text{TP} / (\text{TP} + \text{FP})。在所有被标记为正的样本中，有多少确实是正样本？高精确率意味着误报很少。

召回率（Recall） = \text{TP} / (\text{TP} + \text{FN})。在所有实际为正的样本中，我们抓到了多少？高召回率意味着漏掉的正样本很少。

F1 分数 = 2 \cdot \text{precision} \cdot \text{recall} / (\text{precision} + \text{recall})。调和平均。与算术平均相比，它更惩罚精确率和召回率之间严重失衡的情况。

F-beta 分数 = (1 + \beta^2) \cdot \text{precision} \cdot \text{recall} / (\beta^2 \cdot \text{precision} + \text{recall})。当 \beta > 1 时，召回率更重要；当 \beta < 1 时，精确率更重要。F2 在欺诈检测中很常见（漏掉欺诈比误报更糟）。

AUPRC（精确率-召回率曲线下面积）。与 AUC-ROC 类似，但对不平衡数据更具信息量。随机分类器的 AUPRC 等于正样本比例（不像 ROC 那样是 0.5）。这让改进更容易被察觉。

Matthews 相关系数 = (\text{TP} \cdot \text{TN} - \text{FP} \cdot \text{FN}) / \sqrt{(\text{TP}+\text{FP})(\text{TP}+\text{FN})(\text{TN}+\text{FP})(\text{TN}+\text{FN})}。取值范围 -1 到 +1。只有当模型在两个类别上都表现良好时才会给出高分。即使类别规模差异很大也能保持平衡。

对于上面「总是预测负样本」的模型：\text{precision} = 0/0（未定义，通常设为 0），\text{recall} = 0/10 = 0，\text{F1} = 0，\text{MCC} = 0。这些指标能正确识别该模型毫无价值。

不平衡数据处理 pipeline

flowchart TD
    A[Imbalanced Dataset] --> B{Imbalance Ratio?}
    B -->|Mild: 80/20| C[Class Weights]
    B -->|Moderate: 95/5| D[SMOTE + Threshold Tuning]
    B -->|Severe: 99/1| E[SMOTE + Class Weights + Threshold]
    C --> F[Train Model]
    D --> F
    E --> F
    F --> G[Evaluate with F1 / AUPRC / MCC]
    G --> H{Good Enough?}
    H -->|No| I[Try Different Strategy]
    H -->|Yes| J[Deploy with Monitoring]
    I --> B

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

随机过采样会复制现有的少数类样本。这样做有效，但存在过拟合风险，因为模型会反复看到完全相同的点。

SMOTE 会生成合理但不是复制品的新少数类样本。算法如下：

1. 对每个少数类样本 x，在其他少数类样本中找到它的 k 个最近邻
2. 随机挑选一个邻居
3. 在 x 与该邻居之间的线段上创建一个新样本

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

这在真实少数类点之间进行插值，在特征空间的同一区域内生成样本，而不仅仅是复制现有数据。

flowchart LR
    subgraph Original["Original Minority Points"]
        P1["x1 (1.0, 2.0)"]
        P2["x2 (1.5, 2.5)"]
        P3["x3 (2.0, 1.5)"]
    end
    subgraph SMOTE["SMOTE Generation"]
        direction TB
        S1["Pick x1, neighbor x2"]
        S2["random t = 0.4"]
        S3["new = x1 + 0.4*(x2-x1)"]
        S4["new = (1.2, 2.2)"]
        S1 --> S2 --> S3 --> S4
    end
    Original --> SMOTE
    subgraph Result["Augmented Set"]
        R1["x1 (1.0, 2.0)"]
        R2["x2 (1.5, 2.5)"]
        R3["x3 (2.0, 1.5)"]
        R4["synthetic (1.2, 2.2)"]
    end
    SMOTE --> Result

采样策略对比

随机过采样：复制少数类样本直到与多数类数量相等。

• 优点：简单，无信息损失
• 缺点：完全重复会导致过拟合，增加训练时间

随机欠采样：删除多数类样本直到与少数类数量相等。

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

SMOTE：通过插值生成合成少数类样本。

• 优点：生成新数据点，与随机过采样相比减少过拟合
• 缺点：可能在决策边界附近产生噪声样本，未考虑多数类分布

类别权重

与其改变数据，不如改变模型处理错误的方式。给少数类误分类分配更高的权重。

对于一个 950 个负样本、50 个正样本的二分类问题：

• 负类权重 = n_{\text{samples}} / (2 \cdot n_{\text{negative}}) = 1000 / (2 \cdot 950) = 0.526
• 正类权重 = n_{\text{samples}} / (2 \cdot n_{\text{positive}}) = 1000 / (2 \cdot 50) = 10.0

正类权重是负类的 19 倍。误分类一个正样本的代价相当于误分类 19 个负样本。模型被迫关注少数类。

在 logistic regression 中，这会修改损失函数：

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

类别权重在期望上等价于过采样，但不会创建新数据点。这使它更快，并避免了重复样本带来的过拟合风险。

阈值调优

大多数分类器输出概率。默认阈值是 0.5：如果 P(\text{positive}) \ge 0.5，就预测为正。但 0.5 是任意取的。当类别不平衡时，最优阈值通常要低得多。

流程如下：

1. 训练模型
2. 在验证集上获得预测概率
3. 从 0.0 到 1.0 扫描阈值
4. 在每个阈值下计算 F1（或你选定的指标）
5. 选择最大化该指标的阈值

flowchart LR
    A[Model] --> B[Predict Probabilities]
    B --> C[Sweep Thresholds 0.0 to 1.0]
    C --> D[Compute F1 at Each]
    D --> E[Pick Best Threshold]
    E --> F[Use in Production]

一个模型对一笔欺诈交易可能输出 P(\text{fraud}) = 0.15。在阈值 0.5 下，它会被分类为非欺诈。在阈值 0.10 下，它能被正确拦截。概率校准的重要性低于排序——只要欺诈交易获得比非欺诈更高的概率，就存在一个阈值可以把它们分开。

代价敏感学习

类别权重的泛化。不再使用统一的代价，而是为每种误分类分配特定代价：

漏掉一笔欺诈交易（FN）的代价是误报（FP）的 100 倍。模型优化的是总代价，而不是总错误数。

当你能估计真实代价时，这是最原则性的方法。漏诊癌症的代价与导致额外活检的误报代价截然不同。明确这些代价能迫使模型做出正确的权衡。

决策流程图

flowchart TD
    A[Start: Imbalanced Dataset] --> B{How imbalanced?}
    B -->|"< 70/30"| C["Mild: try class weights first"]
    B -->|"70/30 to 95/5"| D["Moderate: SMOTE + class weights"]
    B -->|"> 95/5"| E["Severe: combine multiple strategies"]
    C --> F{Enough data?}
    D --> F
    E --> F
    F -->|"< 1000 samples"| G["Oversample or SMOTE, avoid undersampling"]
    F -->|"1000-10000"| H["SMOTE + threshold tuning"]
    F -->|"> 10000"| I["Undersampling OK, or class weights"]
    G --> J[Train + Evaluate with F1/AUPRC]
    H --> J
    I --> J
    J --> K{Recall high enough?}
    K -->|No| L[Lower threshold]
    K -->|Yes| M{Precision acceptable?}
    M -->|No| N[Raise threshold or add features]
    M -->|Yes| O[Ship it]

构建它

步骤 1：生成一个不平衡数据集

python复制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

python复制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：随机过采样和欠采样

python复制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：带类别权重的 logistic regression

python复制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：阈值调优

python复制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：评估函数

python复制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：比较所有方法

python复制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:]

# Baseline: no treatment
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)

# Oversampled
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)

# Class weights
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)

# Threshold tuning (tune on held-out validation set, not test set)
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 时，这些技术只需一行代码：

python复制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-NN 插值。类别权重会乘到损失上。阈值调优是在多个截断点上的循环。没有魔法。

交付它

本节课产出：

• outputs/skill-imbalanced-data.md —— 处理不平衡分类问题的决策清单

练习

1. Borderline-SMOTE：修改 SMOTE 实现，只为靠近决策边界的少数类点生成合成样本（即 k 近邻中包含多数类样本的点）。在类别重叠的数据集上与标准 SMOTE 比较结果。

2. 代价矩阵优化：实现代价敏感学习，将代价矩阵作为参数。编写一个函数，接收代价矩阵并返回最小化期望代价的最优预测。用不同代价比（1:10、1:100、1:1000）测试，并绘制精确率-召回率权衡的变化。

3. 阈值校准：实现 Platt scaling（在模型原始输出上拟合一个 logistic regression 以产生校准后的概率）。比较校准前后的精确率-召回率曲线。说明校准不会改变排序（AUC 保持不变），但会让概率更有意义。

4. 平衡 bagging 集成：训练多个模型，每个都在一个平衡的 bootstrap 样本上训练（全部少数类 + 随机多数类子集），然后平均它们的预测。与单个 SMOTE 模型比较。衡量性能与多次运行的方差。

5. 不平衡比例实验：取一个平衡数据集，逐步增加不平衡比例（50/50、70/30、90/10、95/5、99/1）。对每个比例，分别用和不用 SMOTE 训练。绘制 F1 随不平衡比例变化的曲线。在哪个比例下 SMOTE 开始产生明显差异？

关键术语

延伸阅读

• SMOTE: Synthetic Minority Over-sampling Technique (Chawla et al., 2002) —— 原始 SMOTE 论文，仍是不平衡学习领域被引用最多的工作
• Learning from Imbalanced Data (He & Garcia, 2009) —— 涵盖采样、代价敏感和算法方法的综合性综述
• imbalanced-learn documentation —— Python 库，包含 SMOTE 变体、欠采样策略和 pipeline 集成
• The Precision-Recall Plot Is More Informative than the ROC Plot (Saito & Rehmsmeier, 2015) —— 说明在不平衡问题中为何应优先使用 PR 曲线而非 ROC 曲线