机器学习基础实操：手写最近邻和逻辑回归，看懂过拟合与数据泄漏

预测谁需要提醒，先别急着看准确率

假设你想在学习站上给“可能跟不上”的学员发一条提醒，手里有每周学习分钟和完成练习数。本课用标准库手写两个最小模型：最近邻分类和逻辑回归，做一次规范的训练/验证划分，然后亲眼看到三件事：训练集上满分不代表学会了；准确率 0.75 可能一个需要帮助的人都没找出来；一个“好得不像真的”的特征往往是数据泄漏。

完成后你会得到四个脚本和两份数据文件：make_data.py（造数据并划分）、knn.py、logistic.py、leakage.py。准备 Python 3.12 和一个空目录 ml-lab，按顺序运行，后三个脚本读取第一个生成的 train.csv 和 val.csv。数据全部由固定随机种子生成，是虚构学员，不对应任何真人，也不来自本站的学习记录。

这些概念不只属于传统机器学习：大模型评测里的“保留测试集”“题目泄漏进训练数据”“裁判判断的召回率”，都是同一套想法。

第一步：造一份虚构数据，只划分一次

每个学员有两个可以提前知道的特征（每周学习分钟 minutes、完成练习数 exercises），一个考完才知道的小测分数 quiz_score，以及要预测的标签 need_help（小测低于 60 分记为 1，表示需要提醒）。quiz_score 故意留在文件里，第四步用它演示泄漏。保存为 make_data.py：

import csv
import random

# 全部是用固定随机种子生成的虚构学员，不对应任何真人；隐藏规则只用来造数据，模型看不到
rng = random.Random(2026)
rows = []
for i in range(1, 241):
    minutes = rng.randint(20, 420)                                        # 每周学习分钟
    exercises = min(15, max(0, round(minutes / 40 + rng.gauss(0, 2.5))))  # 完成练习数
    ability = 0.010 * minutes + 0.25 * exercises + rng.gauss(0, 1.0) - 1.8
    quiz_score = max(0, min(100, round(60 + 12 * ability)))               # 小测分数：考完才知道
    rows.append({"id": f"S{i:03d}", "minutes": minutes, "exercises": exercises,
                 "quiz_score": quiz_score, "need_help": int(quiz_score < 60)})

# 分层划分：两类各自打乱后按 7:3 切开，让验证集里“需要提醒”的比例和训练集接近
train, val = [], []
for label in (0, 1):
    group = [r for r in rows if r["need_help"] == label]
    rng.shuffle(group)
    cut = round(len(group) * 0.7)
    train += group[:cut]
    val += group[cut:]

# 划分只做一次并存成文件，之后所有模型都用同一份验证集比较
for name, part in (("train.csv", train), ("val.csv", val)):
    with open(name, "w", newline="", encoding="utf-8") as f:
        writer = csv.DictWriter(f, fieldnames=list(part[0]))
        writer.writeheader()
        writer.writerows(sorted(part, key=lambda r: r["id"]))
    print(f"{name}: {len(part)} 行，需要提醒 {sum(r['need_help'] for r in part)} 人")
print("示例：", rows[0])
assert not {r["id"] for r in train} & {r["id"] for r in val}

实际输出：

train.csv: 168 行，需要提醒 42 人
val.csv: 72 行，需要提醒 18 人
示例： {'id': 'S001', 'minutes': 80, 'exercises': 0, 'quiz_score': 75, 'need_help': 0}

三个做法值得照搬。训练集用来“学”，验证集只用来“考”，模型训练时不能看见验证集。需要提醒的人只占四分之一，随手一切可能让验证集里几乎没有这类人，所以按类别分别切。划分结果存成文件，后面每个模型都在同一份验证集上比，结果才可比。示例里 S001 学得少却考了 75 分，真实数据也有这种噪声，任何模型都不可能全对。

第二步：最近邻分类，看过拟合长什么样

最近邻的想法最直接：找训练集里最像的 k 个人，多数需要提醒就判为需要。分钟数在几十到几百之间，练习数只有 0 到 15，直接算距离几乎只看分钟，所以先标准化；均值和标准差只能用训练集算，否则验证集的信息就提前混进来了。保存为 knn.py：

import csv
import math

FEATURES = ["minutes", "exercises"]


def load(path):
    with open(path, encoding="utf-8") as f:
        return [{k: (v if k == "id" else int(v)) for k, v in row.items()} for row in csv.DictReader(f)]


train, val = load("train.csv"), load("val.csv")

# 标准化：只用训练集算均值和标准差，再套到验证集；分钟和练习数的量级差几十倍，不缩放的话距离几乎只看分钟
stats = {}
for name in FEATURES:
    values = [r[name] for r in train]
    mean = sum(values) / len(values)
    stats[name] = (mean, math.sqrt(sum((v - mean) ** 2 for v in values) / len(values)))


def vec(row):
    return [(row[n] - stats[n][0]) / stats[n][1] for n in FEATURES]


def knn_predict(row, k):
    x = vec(row)
    nearest = sorted(train, key=lambda t: (math.dist(x, vec(t)), t["id"]))[:k]
    return int(sum(t["need_help"] for t in nearest) * 2 > k)  # 过半邻居需要提醒才判为需要


def accuracy(rows, k):
    return sum(knn_predict(r, k) == r["need_help"] for r in rows) / len(rows)


print(" k   训练集准确率  验证集准确率")
for k in (1, 3, 7, 15, 31):
    print(f"{k:2}   {accuracy(train, k):10.2f}  {accuracy(val, k):12.2f}")

实际输出：

 k   训练集准确率  验证集准确率
 1         1.00          0.88
 3         0.92          0.83
 7         0.89          0.85
15         0.90          0.88
31         0.92          0.90

k=1 时训练集 1.00，因为每个训练样本最近的邻居就是它自己，这是背答案，不是学会了规律；验证集只有 0.88，差距就是过拟合的信号。k 变大后两者接近，模型更“平滑”。也要看到验证集只有 72 人，0.83 和 0.88 只差三四个人，这种差别可能只是运气。如果你反复在验证集上挑 k、挑特征，验证集就慢慢变成了训练的一部分；正式报告前应再留一份从未用过的测试集，或换一批新数据复核。

第三步：逻辑回归，读懂精确率和召回率

逻辑回归给每个特征一个权重，加权求和后用 S 形函数压到 0 到 1 之间，作为“需要提醒”的概率；训练就是用梯度下降不断调整权重，让预测概率贴近真实标签。保存为 logistic.py：

import csv
import math


def load(path):
    with open(path, encoding="utf-8") as f:
        return [{k: (v if k == "id" else int(v)) for k, v in row.items()} for row in csv.DictReader(f)]


def train_logistic(rows, features, epochs=2000, lr=0.1):
    stats = {}
    for n in features:
        values = [r[n] for r in rows]
        mean = sum(values) / len(values)
        stats[n] = (mean, math.sqrt(sum((v - mean) ** 2 for v in values) / len(values)) or 1.0)
    xs = [[(r[n] - stats[n][0]) / stats[n][1] for n in features] for r in rows]
    ys = [r["need_help"] for r in rows]
    w, b = [0.0] * len(features), 0.0
    for _ in range(epochs):  # 批量梯度下降，最小化对数损失
        grad_w, grad_b = [0.0] * len(features), 0.0
        for x, y in zip(xs, ys, strict=True):
            err = 1 / (1 + math.exp(-(sum(wi * xi for wi, xi in zip(w, x, strict=True)) + b))) - y
            grad_w = [g + err * xi for g, xi in zip(grad_w, x, strict=True)]
            grad_b += err
        w = [wi - lr * g / len(xs) for wi, g in zip(w, grad_w, strict=True)]
        b -= lr * grad_b / len(xs)

    def predict_proba(row):
        x = [(row[n] - stats[n][0]) / stats[n][1] for n in features]
        return 1 / (1 + math.exp(-(sum(wi * xi for wi, xi in zip(w, x, strict=True)) + b)))
    return predict_proba, dict(zip(features, (round(v, 2) for v in w), strict=True))


def report(name, rows, predict):
    tp = sum(predict(r) == 1 and r["need_help"] == 1 for r in rows)
    fp = sum(predict(r) == 1 and r["need_help"] == 0 for r in rows)
    fn = sum(predict(r) == 0 and r["need_help"] == 1 for r in rows)
    tn = len(rows) - tp - fp - fn
    precision = f"{tp / (tp + fp):.2f}" if tp + fp else "无法计算"  # 一个都没判为需要提醒
    recall = tp / (tp + fn)
    print(f"{name}: 准确率 {(tp + tn) / len(rows):.2f}  精确率 {precision}  召回率 {recall:.2f}"
          f"  [抓到 {tp}，误报 {fp}，漏掉 {fn}，正确放过 {tn}]")


train, val = load("train.csv"), load("val.csv")
proba, weights = train_logistic(train, ["minutes", "exercises"])
print("标准化后的权重：", weights)
report("全判不需要提醒", val, lambda r: 0)
report("逻辑回归 阈值0.5", val, lambda r: int(proba(r) >= 0.5))
report("逻辑回归 阈值0.3", val, lambda r: int(proba(r) >= 0.3))
report("训练集 阈值0.5", train, lambda r: int(proba(r) >= 0.5))

实际输出：

标准化后的权重： {'minutes': -1.82, 'exercises': -1.88}
全判不需要提醒: 准确率 0.75  精确率 无法计算  召回率 0.00  [抓到 0，误报 0，漏掉 18，正确放过 54]
逻辑回归 阈值0.5: 准确率 0.90  精确率 0.79  召回率 0.83  [抓到 15，误报 4，漏掉 3，正确放过 50]
逻辑回归 阈值0.3: 准确率 0.83  精确率 0.61  召回率 0.94  [抓到 17，误报 11，漏掉 1，正确放过 43]
训练集 阈值0.5: 准确率 0.92  精确率 0.85  召回率 0.83  [抓到 35，误报 6，漏掉 7，正确放过 120]

先看第一行：什么都不做、全判“不需要”，准确率也有 0.75，因为四分之三的人本来就不需要提醒，可它一个需要帮助的人都没找到。所以类别不均衡时，准确率必须和这个“全猜多数”的基线放在一起看。

精确率回答“发出的提醒里有多少是该发的”（15/19≈0.79），召回率回答“该提醒的人里找到了多少”（15/18≈0.83）。把阈值从 0.5 降到 0.3，多抓到 2 人、只漏 1 人，代价是误报从 4 条涨到 11 条。阈值没有标准答案：漏掉一个跟不上的学员和多打扰一个学得好的学员，哪个代价更大，由产品决定。两个权重都是负数，意思是学得越多、练得越多，越不需要提醒，和常识一致；如果权重方向反常，先查数据而不是急着解释。训练集 0.92 与验证集 0.90 很接近，说明这个两参数模型没有明显过拟合。

第四步：亲手制造两种数据泄漏

泄漏是指训练时用到了预测那一刻本不该知道的信息，验证成绩因此虚高，上线后就露馅。下面先用“单特征阈值规则”分别试三个特征，再模拟一次“重复导出后按行划分”的事故。保存为 leakage.py：

import csv
import random


def load(path):
    with open(path, encoding="utf-8") as f:
        return [{k: (v if k == "id" else int(v)) for k, v in row.items()} for row in csv.DictReader(f)]


def fit_stump(rows, feature):
    """单特征规则：数值低于阈值就判为需要提醒；在训练集上挑准确率最高的阈值。"""
    best = (-1.0, None)
    for t in sorted({r[feature] for r in rows}):
        acc = sum(int(r[feature] < t) == r["need_help"] for r in rows) / len(rows)
        best = max(best, (acc, t))
    return best[1]


train, val = load("train.csv"), load("val.csv")
for feature in ("minutes", "exercises", "quiz_score"):
    t = fit_stump(train, feature)
    acc = sum(int(r[feature] < t) == r["need_help"] for r in val) / len(val)
    print(f"只用 {feature:10} 阈值 {t:4}：验证集准确率 {acc:.2f}")

overlap = {r["id"] for r in train} & {r["id"] for r in val}
print(f"训练集与验证集重复的学员编号：{len(overlap)} 个")

# 模拟常见事故：同一批学员被导出两次后合并，再按“行”随机划分，而不是按学员编号划分
doubled = (train + val) * 2
random.Random(7).shuffle(doubled)
cut = int(len(doubled) * 0.7)
leak_train, leak_val = doubled[:cut], doubled[cut:]


def memorize_then_guess(fit_rows, eval_rows):
    """见过一模一样的记录就照抄答案，没见过就用分钟数规则。"""
    table = {(r["minutes"], r["exercises"]): r["need_help"] for r in fit_rows}
    t = fit_stump(fit_rows, "minutes")
    hits = [table.get((r["minutes"], r["exercises"]), int(r["minutes"] < t)) == r["need_help"]
            for r in eval_rows]
    seen = sum((r["minutes"], r["exercises"]) in table for r in eval_rows)
    return sum(hits) / len(hits), seen / len(eval_rows)


for name, (fit_rows, eval_rows) in {"按学员划分": (train, val), "重复后按行划分": (leak_train, leak_val)}.items():
    acc, seen = memorize_then_guess(fit_rows, eval_rows)
    print(f"{name}：验证集里 {seen:.0%} 的行在训练集见过，准确率 {acc:.2f}")

实际输出：

只用 minutes    阈值   99：验证集准确率 0.89
只用 exercises  阈值    4：验证集准确率 0.82
只用 quiz_score 阈值   60：验证集准确率 1.00
训练集与验证集重复的学员编号：0 个
按学员划分：验证集里 8% 的行在训练集见过，准确率 0.88
重复后按行划分：验证集里 81% 的行在训练集见过，准确率 0.97

quiz_score 一个特征就拿到 1.00，因为标签本来就是由它定义的，而且它要考完才知道，提醒却要在考前发。这种“完美特征”上线时根本拿不到。第二种泄漏更隐蔽：同一批人导出两次再按行划分，验证集八成的行在训练集出现过，一个只会照抄的规则也能得 0.97。按学员划分时仍有 8% 的行“见过”，那是不同学员恰好分钟数和练习数相同，属于正常巧合，不是泄漏。

还要注意，只用分钟数的一条规则就有 0.89，和逻辑回归的 0.90 几乎一样。复杂模型必须先赢过这种简单基线，才值得维护。

加练：把噪声加倍，再跑一遍

真实数据通常比这份更“吵”。新建目录，把 make_data.py 中 rng.gauss(0, 1.0) 改成 rng.gauss(0, 2.0)，其余不动，按顺序重跑四个脚本。下表是本站按这个改动实际运行得到的几项数字：

噪声加倍后，k=1 的训练与验证差距从 0.12 拉大到 0.28，过拟合更明显；阈值 0.5 时一半以上需要帮助的人被漏掉，降到 0.3 才找回大部分。k=1 训练集不再是 1.00，是因为训练集里出现了两对分钟数和练习数完全相同、标签却不同的学员，每对至少会错一个，再怎么背也无法全对。按行划分的泄漏照样把成绩抬高了一大截。换种子或换噪声后先看趋势是否还在，不要只记某一个数字。

常见失败与修改

训练集很高，验证集明显更低。 典型过拟合。最近邻就增大 k，其他模型就减少特征或加正则；更根本的是增加数据。调参只在验证集上做，最后用一份没碰过的测试集确认一次，不要反复回头改。

验证集好得不像真的（接近 1.00）。 先逐个检查特征：预测那一刻能否拿到？是不是由标签算出来的？再检查同一个人、同一份作业是否同时出现在训练集和验证集，按人或按对象划分；有时间顺序的数据按时间切，用过去预测未来。

准确率不错，却几乎找不到需要帮助的人。 先算“全猜多数”的基线和召回率；再降低判定阈值，或在训练时给少数类更高权重，并同时报告误报增加了多少。

每次运行结果都不一样。 固定随机种子，把划分结果存成文件，记录 Python 版本和数据生成日期；比较两个模型时必须用同一份验证集。

标准化或填补缺失值用了全部数据。 均值、标准差、缺失值填补规则都只从训练集计算，再原样套用到验证集和新数据。

保存和使用结果

在 ml-lab 目录保留四个脚本、train.csv、val.csv 和一张结果表：模型、参数、训练集准确率、验证集准确率、精确率、召回率、阈值、日期。以后读到任何“准确率 95%”的说法，先问四个问题：基线是多少？召回率呢？验证数据和训练数据怎样划分？有没有泄漏检查？同样的思路可以直接用到评测实操和大模型裁判的一致率统计中。

自测与答案

1. 最近邻 k=1 训练集准确率 1.00，为什么不能说明模型好？每个训练样本最近的就是它自己，等于背答案，要看验证集。
2. 验证集里 75% 的人不需要提醒，一个准确率 0.78 的模型好吗？只比全猜多数高 3 个百分点，还要看召回率和精确率才能判断。
3. 想多抓到需要帮助的人，阈值该升还是降？降，召回率上升，但误报通常也会增加。
4. 为什么 quiz_score 不能当特征？标签由它定义，而且预测时还没有这个分数，属于数据泄漏。

交付物是四个脚本的实际输出和一张结果表。全部数字来自固定种子的虚构数据，只用于说明概念，不代表任何真实学员或真实平台的预测效果。