处理不平衡数据

当你 99% 的数据都是"正常"时，准确率就是个谎言。

类型： Build 语言： Python 前置要求： 阶段 2 第 01-09 课（尤其是评估指标） 预计时间： ~90 分钟

学习目标

从零实现 SMOTE，并解释合成过采样和随机复制有什么不同
用 F1、AUPRC 和马修斯相关系数代替准确率来评估不平衡分类器
对比类别加权、阈值调节和重采样策略，为给定的不平衡比例选对方法
构建一条完整的不平衡数据流水线，结合 SMOTE、类别权重和阈值优化

问题背景

你做了一个欺诈检测模型。它拿到 99.9% 准确率。你庆祝。然后你意识到它对每一笔交易都预测"非欺诈"。

这不是 bug。当只有 0.1% 的交易是欺诈时，这是理性的做法。模型学到永远猜多数类能让总体误差最小。它技术上正确、却完全没用。

这在每一个真实分类重要的地方都会发生。疾病诊断：1% 阳性率。网络入侵：0.01% 攻击。制造缺陷：0.5% 次品。垃圾邮件过滤：20% 垃圾。流失预测：5% 流失者。少数类越重要，它往往越稀少。

准确率失败是因为它平等对待所有正确预测。正确标注一笔合法交易和正确抓住一笔欺诈都算一分准确率。但抓住欺诈才是这个模型存在的全部理由。我们需要能强迫模型关注那个稀少但重要的类的指标、技术和训练策略。

核心概念

为什么准确率失败

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

	预测为正	预测为负
实际为正	0 (TP)	10 (FN)
实际为负	0 (FP)	990 (TN)

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

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

更好的指标

精确率 = TP / (TP + FP)。在所有被标为正的里，有多少真的是？高精确率意味着少误报。

召回率 = TP / (TP + FN)。在所有真正为正的里，我们抓到了多少？高召回率意味着少漏掉的正例。

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

F-beta 分数 = (1 + beta^2) * precision * recall / (beta^2 * precision + recall)。beta > 1 时召回率更重要。beta < 1 时精确率更重要。F2 在欺诈检测里常见（漏掉欺诈比误报更糟）。

AUPRC（精确率-召回率曲线下面积）。像 AUC-ROC 但对不平衡数据更有信息量。随机分类器的 AUPRC 等于正类率（不像 ROC 那样是 0.5）。这让改进更容易看出来。

马修斯相关系数 = (TP * TN - FP * FN) / sqrt((TP+FP)(TP+FN)(TN+FP)(TN+FN))。从 -1 到 +1。只有模型在两个类上都做得好时才给高分。即使类别大小差异很大也保持平衡。

对上面那个"永远预测负"的模型：精确率 = 0/0（无定义，常设为 0）、召回率 = 0/10 = 0、F1 = 0、MCC = 0。这些指标正确地把模型识别为一文不值。

不平衡数据流水线

mermaid

flowchart TD
    A[不平衡数据集] --> B{不平衡比例？}
    B -->|轻度：80/20| C[类别权重]
    B -->|中度：95/5| D[SMOTE + 阈值调节]
    B -->|重度：99/1| E[SMOTE + 类别权重 + 阈值]
    C --> F[训练模型]
    D --> F
    E --> F
    F --> G[用 F1 / AUPRC / MCC 评估]
    G --> H{够好了吗？}
    H -->|否| I[试不同策略]
    H -->|是| J[带监控部署]
    I --> B

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

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

SMOTE 创建新的合成少数类样本，它们合理但不是副本。算法：

对每个少数类样本 x，在其他少数类样本里找它的 k 个最近邻
随机挑一个邻居
在 x 和那个邻居之间的线段上创建一个新样本

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

这在真实少数类点之间插值，在特征空间的同一区域里创建样本，而不只是复制已有数据。

mermaid

flowchart LR
    subgraph Original["原始少数类点"]
        P1["x1 (1.0, 2.0)"]
        P2["x2 (1.5, 2.5)"]
        P3["x3 (2.0, 1.5)"]
    end
    subgraph SMOTE["SMOTE 生成"]
        direction TB
        S1["挑 x1，邻居 x2"]
        S2["随机 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["增广后的集合"]
        R1["x1 (1.0, 2.0)"]
        R2["x2 (1.5, 2.5)"]
        R3["x3 (2.0, 1.5)"]
        R4["合成 (1.2, 2.2)"]
    end
    SMOTE --> Result

采样策略对比

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

优点：简单，无信息损失
缺点：完全相同的副本导致过拟合，增加训练时间

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

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

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

优点：生成新数据点，比随机过采样更少过拟合
缺点：可能在决策边界附近造出有噪声的样本，不考虑多数类分布

策略	改动的数据	风险	何时用
过采样	复制少数类	过拟合	小数据集、中度不平衡
欠采样	移除多数类	信息损失	大数据集、想训练快
SMOTE	加入合成少数类	边界噪声	中度不平衡、少数类样本够做 k-NN

类别权重

不改数据，而是改模型对待错误的方式。给错分少数类更高的权重。

对一个 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(positive) >= 0.5，预测正。但 0.5 是任意的。类别不平衡时，最优阈值通常低得多。

过程：

训练一个模型
在验证集上拿预测概率
把阈值从 0.0 扫到 1.0
在每个阈值上计算 F1（或你选的指标）
挑让你的指标最大化的阈值

mermaid

flowchart LR
    A[模型] --> B[预测概率]
    B --> C[扫阈值 0.0 到 1.0]
    C --> D[在每个上算 F1]
    D --> E[挑最佳阈值]
    E --> F[在生产里用]

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

代价敏感学习

类别权重的推广。不用统一代价，而是赋予具体的错分代价：

	预测正	预测负
实际为正	0（正确）	C_FN = 100
实际为负	C_FP = 1	0（正确）

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

当你能估计真实世界代价时，这是最有原则的方法。漏诊癌症和导致一次额外活检的误报，代价天差地别。把这些代价显式化，强迫做出正确的权衡。

决策流程图

mermaid

flowchart TD
    A[开始：不平衡数据集] --> B{有多不平衡？}
    B -->|"< 70/30"| C["轻度：先试类别权重"]
    B -->|"70/30 到 95/5"| D["中度：SMOTE + 类别权重"]
    B -->|"> 95/5"| E["重度：组合多种策略"]
    C --> F{数据够吗？}
    D --> F
    E --> F
    F -->|"< 1000 个样本"| G["过采样或 SMOTE，避免欠采样"]
    F -->|"1000-10000"| H["SMOTE + 阈值调节"]
    F -->|"> 10000"| I["欠采样可以，或类别权重"]
    G --> J[训练 + 用 F1/AUPRC 评估]
    H --> J
    I --> J
    J --> K{召回率够高吗？}
    K -->|否| L[降低阈值]
    K -->|是| M{精确率可接受吗？}
    M -->|否| N[提高阈值或加特征]
    M -->|是| O[交付]

figure

class-imbalance

动手构建

第 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 步：带类别权重的逻辑回归

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:]

# 基线：不做任何处理
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，这些技术都是一行：

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 插值。类别权重乘损失。阈值调节是对截断点做 for 循环。没魔法。

拿去用

本节课产出：

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

练习

Borderline-SMOTE：修改 SMOTE 实现，只为靠近决策边界的少数类点（那些 k 近邻里包含多数类样本的点）生成合成样本。在类别重叠的数据集上和标准 SMOTE 对比结果。
代价矩阵优化：实现代价敏感学习，把代价矩阵当参数。写一个函数，接收一个代价矩阵，返回最小化期望代价的最优预测。用不同代价比（1:10、1:100、1:1000）测试，画出精确率-召回率权衡如何变化。
阈值校准：实现 Platt 缩放（在模型原始输出上拟合一个逻辑回归来产生校准概率）。对比校准前后的精确率-召回率曲线。说明校准不改变排序（AUC 不变）但让概率更有意义。
平衡 bagging 集成：训练多个模型，每个在一个平衡的自助样本上（全部少数类 + 随机一部分多数类）。平均它们的预测。把这个方法和单个 SMOTE 模型对比。测量性能和跨运行的方差。
不平衡比例实验：拿一个平衡数据集，逐步增加不平衡比例（50/50、70/30、90/10、95/5、99/1）。对每个比例，分别在有和没有 SMOTE 下训练。把两种方法的 F1 对不平衡比例画出来。SMOTE 在哪个比例开始带来有意义的差别？

关键术语

术语	大家怎么说	它实际是什么
类别不平衡	"一个类的样本多得多"	数据集里类别分布显著偏斜，导致模型偏向多数类
SMOTE	"合成过采样"	通过在已有少数类样本和它们的 k 近邻少数类之间插值，创建新的少数类样本
类别权重	"让稀有类上的错误更贵"	把损失函数乘以类别特定的权重，让模型更狠地惩罚少数类错分
阈值调节	"移动决策边界"	把分类的概率截断点从默认 0.5 改成优化目标指标的值
精确率-召回率权衡	"你没法两者兼得"	降低阈值抓更多正例（召回率更高）但也标出更多假正例（精确率更低），反之亦然
AUPRC	"PR 曲线下面积"	把精确率-召回率曲线汇总成一个数字；类别严重不平衡时比 AUC-ROC 更有信息量
马修斯相关系数	"平衡的指标"	预测和真实标签之间的相关性，只有模型在两个类上都表现好时才给高分
代价敏感学习	"不同错误代价不同"	把真实世界的错分代价纳入训练目标，让模型为总代价而非错误数优化
随机过采样	"复制少数类"	重复少数类样本来平衡类别数量；简单但有过拟合到复制点的风险

Claude 认证课程

lessons

00-certification-strategy

01-claude-product-and-model-landscape

02-model-selection-and-token-economics

03-prompting-and-task-decomposition

04-context-knowledge-memory-and-caching

05-output-evaluation-and-validation

06-governance-safety-and-responsible-use

07-workflow-design-and-human-handoffs

08-messages-api-and-application-lifecycle

09-structured-output-and-defensive-parsing

10-tool-use-and-agentic-loops

11-mcp-server-design-and-integration

12-claude-agent-sdk-and-hooks

13-application-security-and-secrets

14-evals-testing-debugging-and-observability

15-claude-code-for-development-teams

16-multi-agent-orchestration-and-delegation

17-agent-sdk-sessions-subagents-and-context

18-tool-contracts-errors-and-progressive-discovery

19-claude-code-memory-rules-skills-and-ci

20-reliable-extraction-batch-and-reviewers

21-long-context-reliability-provenance-and-escalation

22-business-discovery-requirements-and-slas

23-end-to-end-architecture-and-value-tradeoffs

24-rag-retrieval-and-data-pipelines

25-integration-protocols-identity-and-least-privilege

26-production-observability-latency-and-cost

27-enterprise-governance-compliance-and-hitl

28-stakeholder-communication-adrs-and-lifecycle

29-associate-workflow-capstone

30-developer-application-capstone

31-architect-foundations-scenario-capstone

32-architect-professional-system-capstone

references

research

阶段 0：环境与工具

01-dev-environment

outputs

04-apis-and-keys

outputs

05-jupyter-notebooks

outputs

09-data-management

outputs

12-debugging-and-profiling

outputs

阶段 1：数学基础

01-linear-algebra-intuition

outputs

02-vectors-matrices-operations

outputs

03-matrix-transformations

outputs

04-calculus-for-ml

outputs

05-chain-rule-and-autodiff

outputs

06-probability-and-distributions

outputs

07-bayes-theorem

outputs

08-optimization

outputs

09-information-theory

outputs

10-dimensionality-reduction

outputs

11-singular-value-decomposition

outputs

12-tensor-operations

outputs

13-numerical-stability

outputs

14-norms-and-distances

outputs

15-statistics-for-ml

outputs

16-sampling-methods