机器学习
基础

from dataclasses import dataclass
from datetime import timedelta

@dataclass(frozen=True)
class TaskSpec:
    name: str
    entity: str
    prediction_time: str
    label_window: timedelta
    task_type: str
    metric: str

spec = TaskSpec(
    name="七天退款风险",
    entity="order",
    prediction_time="paid_at",
    label_window=timedelta(days=7),
    task_type="binary_classification",
    metric="recall_at_precision_0.6",
)

assert spec.entity and spec.prediction_time
assert spec.task_type in {"regression", "binary_classification", "multiclass"}
print(spec)

from datetime import datetime, timezone

PREDICTION_TIME = datetime(2026, 8, 1, tzinfo=timezone.utc)

def feature_row(user_id, events):
    before = [event for event in events
              if event["at"] <= PREDICTION_TIME]
    minutes = sum(event["minutes"] for event in before)
    count = len(before)
    return {
        "user_id": user_id,
        "reading_minutes_7d": minutes,
        "reading_events_7d": count,
        "has_history": int(bool(before)),
    }

events = [{"at": datetime(2026, 7, 29, tzinfo=timezone.utc), "minutes": 18}]
row = feature_row("u-17", events)
assert row["reading_minutes_7d"] == 18
assert set(row) == {"user_id", "reading_minutes_7d", "reading_events_7d", "has_history"}
print(row)

from math import sqrt

def mae(actual, predicted):
    return sum(abs(a - p) for a, p in zip(actual, predicted)) / len(actual)

def rmse(actual, predicted):
    mse = sum((a - p) ** 2 for a, p in zip(actual, predicted)) / len(actual)
    return sqrt(mse)

train_y = [10.0, 12.0, 13.0, 15.0]
test_y = [11.0, 16.0, 14.0]
constant = sum(train_y) / len(train_y)
baseline = [constant] * len(test_y)
print({"baseline": constant, "mae": mae(test_y, baseline), "rmse": rmse(test_y, baseline)})
assert len(baseline) == len(test_y)

from random import Random

def split_rows(rows, seed=7, test_ratio=0.2, valid_ratio=0.2):
    if not rows:
        raise ValueError("数据不能为空")
    if test_ratio <= 0 or valid_ratio <= 0 or test_ratio + valid_ratio >= 1:
        raise ValueError("比例必须为正且总和小于 1")
    shuffled = list(rows)
    Random(seed).shuffle(shuffled)
    test_size = max(1, round(len(shuffled) * test_ratio))
    valid_size = max(1, round(len(shuffled) * valid_ratio))
    test = shuffled[:test_size]
    valid = shuffled[test_size:test_size + valid_size]
    train = shuffled[test_size + valid_size:]
    assert set(map(id, train)).isdisjoint(map(id, valid))
    assert set(map(id, train)).isdisjoint(map(id, test))
    return train, valid, test

rows = [{"id": i, "label": i % 2} for i in range(20)]
train, valid, test = split_rows(rows)
print(len(train), len(valid), len(test))

from collections import defaultdict

def group_split(rows, group_key, train_groups, valid_groups):
    train, valid, overlap = [], [], set()
    train_groups = set(train_groups)
    valid_groups = set(valid_groups)
    overlap = train_groups & valid_groups
    if overlap:
        raise ValueError(f"群组重叠: {sorted(overlap)}")
    for row in rows:
        group = row[group_key]
        if group in train_groups:
            train.append(row)
        elif group in valid_groups:
            valid.append(row)
    return train, valid

rows = [
    {"user": "a", "at": 1, "label": 0},
    {"user": "a", "at": 2, "label": 1},
    {"user": "b", "at": 3, "label": 0},
    {"user": "c", "at": 4, "label": 1},
]
train, valid = group_split(rows, "user", {"a"}, {"b", "c"})
assert {row["user"] for row in train}.isdisjoint({row["user"] for row in valid})
print(train, valid)

from datetime import datetime, timezone

def assert_no_future_features(rows, feature_time_key, prediction_time_key):
    violations = []
    for row in rows:
        feature_time = row[feature_time_key]
        prediction_time = row[prediction_time_key]
        if feature_time > prediction_time:
            violations.append(row["id"])
    if violations:
        raise AssertionError(f"发现未来特征: {violations}")

rows = [
    {"id": "ok", "feature_at": datetime(2026, 7, 1, tzinfo=timezone.utc), "predict_at": datetime(2026, 7, 2, tzinfo=timezone.utc)},
    {"id": "bad", "feature_at": datetime(2026, 7, 3, tzinfo=timezone.utc), "predict_at": datetime(2026, 7, 2, tzinfo=timezone.utc)},
]
try:
    assert_no_future_features(rows, "feature_at", "predict_at")
except AssertionError as error:
    print(error)

def fit_line(xs, ys):
    if len(xs) != len(ys) or len(xs) < 2:
        raise ValueError("需要至少两个配对样本")
    x_mean = sum(xs) / len(xs)
    y_mean = sum(ys) / len(ys)
    numerator = sum((x - x_mean) * (y - y_mean) for x, y in zip(xs, ys))
    denominator = sum((x - x_mean) ** 2 for x in xs)
    if denominator == 0:
        raise ValueError("特征没有变化，无法确定斜率")
    weight = numerator / denominator
    bias = y_mean - weight * x_mean
    return weight, bias

def predict(xs, weight, bias):
    return [weight * x + bias for x in xs]

xs = [1.0, 2.0, 3.0, 4.0]
ys = [2.1, 3.9, 6.2, 8.0]
w, b = fit_line(xs, ys)
print({"weight": round(w, 3), "bias": round(b, 3), "pred": predict([5], w, b)})

from math import sqrt

def residual_report(actual, predicted):
    if len(actual) != len(predicted) or not actual:
        raise ValueError("长度必须一致且不能为空")
    residuals = [a - p for a, p in zip(actual, predicted)]
    mae = sum(abs(error) for error in residuals) / len(residuals)
    mse = sum(error * error for error in residuals) / len(residuals)
    return {
        "residuals": residuals,
        "mean_error": sum(residuals) / len(residuals),
        "mae": mae,
        "rmse": sqrt(mse),
        "worst_index": max(range(len(residuals)), key=lambda i: abs(residuals[i])),
    }

report = residual_report([10, 12, 20, 11], [11, 11, 15, 12])
print(report)
assert report["worst_index"] == 2

def dot(weights, features, bias=0.0):
    if len(weights) != len(features):
        raise ValueError("权重和特征维度不一致")
    return bias + sum(weight * value for weight, value in zip(weights, features))

features = [
    [30.0, 0.8, 2.0],  # 阅读分钟、完成比例、练习次数
    [10.0, 0.4, 0.0],
    [50.0, 0.9, 4.0],
]
labels = [88.0, 42.0, 96.0]
weights = [0.8, 25.0, 3.0]
predicted = [dot(weights, row, bias=5.0) for row in features]
for index, (actual, estimate) in enumerate(zip(labels, predicted)):
    print(index, {"actual": actual, "predicted": round(estimate, 2), "error": round(actual-estimate, 2)})
assert len(predicted) == len(labels)

from math import exp

def sigmoid(score):
    if score >= 0:
        value = exp(-score)
        return 1 / (1 + value)
    value = exp(score)
    return value / (1 + value)

def predict_probability(features, weights, bias):
    if len(features) != len(weights):
        raise ValueError("特征与权重维度不一致")
    score = bias + sum(x * w for x, w in zip(features, weights))
    return sigmoid(score)

weights = [0.08, 1.2, -0.7]
for row in ([20, 0.8, 1], [5, 0.2, 0]):
    probability = predict_probability(row, weights, bias=-2.0)
    print(round(probability, 4))
    assert 0 < probability < 1

def confusion(actual, predicted):
    if len(actual) != len(predicted):
        raise ValueError("标签长度不一致")
    matrix = {"tp": 0, "fp": 0, "fn": 0, "tn": 0}
    for truth, estimate in zip(actual, predicted):
        key = {(1, 1): "tp", (0, 1): "fp", (1, 0): "fn", (0, 0): "tn"}.get((truth, estimate))
        if key is None:
            raise ValueError("标签必须是 0 或 1")
        matrix[key] += 1
    return matrix

def metrics(matrix):
    tp, fp, fn, tn = (matrix[key] for key in ("tp", "fp", "fn", "tn"))
    precision = tp / (tp + fp) if tp + fp else 0.0
    recall = tp / (tp + fn) if tp + fn else 0.0
    f1 = 2 * precision * recall / (precision + recall) if precision + recall else 0.0
    accuracy = (tp + tn) / (tp + fp + fn + tn)
    return {"precision": precision, "recall": recall, "f1": f1, "accuracy": accuracy}

matrix = confusion([1, 1, 0, 0, 1], [1, 0, 1, 0, 1])
print(matrix, metrics(matrix))

def confusion_counts(actual, probabilities, threshold):
    predicted = [int(value >= threshold) for value in probabilities]
    tp = sum(truth == 1 and guess == 1 for truth, guess in zip(actual, predicted))
    fp = sum(truth == 0 and guess == 1 for truth, guess in zip(actual, predicted))
    fn = sum(truth == 1 and guess == 0 for truth, guess in zip(actual, predicted))
    tn = sum(truth == 0 and guess == 0 for truth, guess in zip(actual, predicted))
    precision = tp / (tp + fp) if tp + fp else 0.0
    recall = tp / (tp + fn) if tp + fn else 0.0
    return {"threshold": threshold, "tp": tp, "fp": fp, "fn": fn, "tn": tn, "precision": precision, "recall": recall}

truth = [0, 1, 1, 0, 1, 0, 1, 0]
probabilities = [0.10, 0.35, 0.62, 0.55, 0.81, 0.20, 0.92, 0.44]
for threshold in (0.3, 0.5, 0.7):
    print(confusion_counts(truth, probabilities, threshold))
# 练习：加入“最多告警 3 条”的约束并选阈值

def mean(values):
    return sum(values) / len(values)

history = [
    {"epoch": 1, "train": 0.82, "valid": 0.86},
    {"epoch": 2, "train": 0.61, "valid": 0.68},
    {"epoch": 3, "train": 0.42, "valid": 0.51},
    {"epoch": 4, "train": 0.28, "valid": 0.55},
    {"epoch": 5, "train": 0.19, "valid": 0.63},
]
best = min(history, key=lambda item: item["valid"])
gap = best["valid"] - best["train"]
print({"best_epoch": best["epoch"], "valid_loss": best["valid"], "generalization_gap": round(gap, 3)})
assert best["epoch"] == 3
# 练习：把第 4、5 轮的 valid 改低，观察“最佳轮次”如何变化

def ridge_objective(errors, weights, strength):
    data_loss = sum(error * error for error in errors) / len(errors)
    penalty = strength * sum(weight * weight for weight in weights)
    return data_loss + penalty

errors = [0.2, -0.4, 0.1, 0.8]
weights = [2.0, -1.5, 0.2]
for strength in (0.0, 0.01, 0.1, 1.0):
    print(strength, round(ridge_objective(errors, weights, strength), 4))

assert ridge_objective(errors, weights, 0.0) < ridge_objective(errors, weights, 1.0)
# 正式训练还要在验证集比较预测误差，而不是只看目标函数

from statistics import mean, pstdev

def fold_score(rows, validation_ids, predictor):
    validation = [row for row in rows if row["id"] in validation_ids]
    train = [row for row in rows if row["id"] not in validation_ids]
    model = predictor(train)
    errors = [abs(row["y"] - model(row["x"])) for row in validation]
    return mean(errors)

rows = [{"id": i, "x": float(i), "y": 2 * i + 1} for i in range(12)]
folds = [set(range(start, start + 3)) for start in (0, 3, 6, 9)]

def fit_mean(train):
    average = mean(row["y"] for row in train)
    return lambda x: average

scores = [fold_score(rows, ids, fit_mean) for ids in folds]
print({"scores": scores, "mean": mean(scores), "std": pstdev(scores)})
assert len(scores) == 4

def gini(labels):
    if not labels:
        return 0.0
    total = len(labels)
    counts = {label: labels.count(label) for label in set(labels)}
    return 1 - sum((count / total) ** 2 for count in counts.values())

def weighted_gini(left, right):
    total = len(left) + len(right)
    return (len(left) * gini(left) + len(right) * gini(right)) / total

def gain(parent, left, right):
    return gini(parent) - weighted_gini(left, right)

parent = [0, 0, 0, 1, 1, 1]
candidate_a = ([0, 0, 0], [1, 1, 1])
candidate_b = ([0, 0, 1], [0, 1, 1])
print(gain(parent, *candidate_a), gain(parent, *candidate_b))
assert gain(parent, *candidate_a) > gain(parent, *candidate_b)

from random import Random

def bootstrap(rows, rng):
    return [rows[rng.randrange(len(rows))] for _ in rows]

def majority(labels):
    counts = {label: labels.count(label) for label in set(labels)}
    return max(counts, key=counts.get)

rows = [{"x": x, "y": int(x >= 5)} for x in range(10)]
rng = Random(11)
votes = {row["x"]: [] for row in rows}
for tree_id in range(7):
    sample = bootstrap(rows, rng)
    seen = {row["x"] for row in sample}
    for row in rows:
        if row["x"] not in seen:  # 简化版 out-of-bag 评估
            guess = int(row["x"] + (tree_id % 2) >= 5)
            votes[row["x"]].append(guess)
for x, guesses in votes.items():
    if guesses:
        print(x, majority(guesses), guesses)

from random import Random

def accuracy(actual, predicted):
    return sum(a == p for a, p in zip(actual, predicted)) / len(actual)

def predictor(row):
    return int(row["minutes"] > 20 or row["exercises"] >= 3)

rows = [
    {"minutes": 10, "exercises": 0}, {"minutes": 30, "exercises": 0},
    {"minutes": 12, "exercises": 4}, {"minutes": 40, "exercises": 5},
]
truth = [0, 1, 1, 1]
base = accuracy(truth, [predictor(row) for row in rows])
rng = Random(3)
for name in ("minutes", "exercises"):
    shuffled = list(rows)
    values = [row[name] for row in shuffled]
    rng.shuffle(values)
    for row, value in zip(shuffled, values):
        row[name] = value
    score = accuracy(truth, [predictor(row) for row in shuffled])
    print(name, {"drop": round(base - score, 3), "permuted_score": score})

from math import dist
from random import Random

def mean_point(points):
    return tuple(sum(point[i] for point in points) / len(points) for i in range(len(points[0])))

def kmeans(points, k=2, steps=10, seed=4):
    if k < 1 or k > len(points):
        raise ValueError("k 必须在样本数量范围内")
    rng = Random(seed)
    centers = [tuple(point) for point in rng.sample(points, k)]
    assignments = []
    for _ in range(steps):
        assignments = [min(range(k), key=lambda index: dist(point, centers[index])) for point in points]
        new_centers = []
        for index, center in enumerate(centers):
            cluster = [point for point, label in zip(points, assignments) if label == index]
            new_centers.append(mean_point(cluster) if cluster else center)
        if new_centers == centers:
            break
        centers = new_centers
    return assignments, centers

points = [(1, 1), (1, 2), (2, 1), (9, 9), (10, 9), (9, 10)]
labels, centers = kmeans(points)
print(labels, centers)
assert len(labels) == len(points)

from math import dist

def within_sum(points, labels, centers):
    return sum(dist(point, centers[label]) ** 2 for point, label in zip(points, labels))

def cluster_sizes(labels):
    return {label: labels.count(label) for label in sorted(set(labels))}

candidate_results = [
    {"k": 2, "inertia": 12.4, "silhouette": 0.58, "stable": 0.91},
    {"k": 3, "inertia": 7.1, "silhouette": 0.49, "stable": 0.64},
    {"k": 4, "inertia": 4.9, "silhouette": 0.43, "stable": 0.51},
]
best = max(candidate_results, key=lambda item: (item["silhouette"], item["stable"]))
print({"candidate": best, "all": candidate_results})
assert best["k"] == 2
# 练习：加入业务约束，排除样本数小于 10 的簇

def percentile(values, fraction):
    ordered = sorted(values)
    index = min(len(ordered) - 1, round((len(ordered) - 1) * fraction))
    return ordered[index]

def distance_to_center(point, center):
    return sum((a - b) ** 2 for a, b in zip(point, center)) ** 0.5

normal = [(1, 1), (1, 2), (2, 1), (2, 2), (3, 2)]
center = (sum(row[0] for row in normal) / len(normal), sum(row[1] for row in normal) / len(normal))
scores = [distance_to_center(point, center) for point in normal]
threshold = percentile(scores, 0.95)
new_points = [(2, 1), (9, 9)]
for point in new_points:
    score = distance_to_center(point, center)
    print(point, {"score": round(score, 2), "anomaly": score > threshold})
# 练习：改成按群组分别计算阈值，比较误报

import math

def center(rows):
    means = [sum(row[i] for row in rows) / len(rows) for i in range(len(rows[0]))]
    return [[value - means[i] for i, value in enumerate(row)] for row in rows], means

def dot(left, right):
    return sum(a * b for a, b in zip(left, right))

def power_iteration(rows, steps=30):
    vector = [1.0] * len(rows[0])
    for _ in range(steps):
        covariance_times = [
            sum(dot(column(row, axis), row) for row in rows)
            for axis in range(len(vector))
        ]
        norm = math.sqrt(dot(covariance_times, covariance_times)) or 1.0
        vector = [value / norm for value in covariance_times]
    return vector

def column(row, axis):
    return row[axis]

rows, means = center([[1.0, 1.2], [2.0, 2.1], [3.0, 2.9], [4.0, 4.2]])
direction = power_iteration(rows)
scores = [dot(row, direction) for row in rows]
print({"means": means, "direction": direction, "scores": scores})
# 教学提示：正式 PCA 应使用成熟实现并核对中心化和 solver 配置

def mse(actual, reconstructed):
    values = [
        (left - right) ** 2
        for row_a, row_b in zip(actual, reconstructed)
        for left, right in zip(row_a, row_b)
    ]
    return sum(values) / len(values)

original = [[2.0, 2.1], [3.0, 3.2], [5.0, 4.8]]
reconstructed = [[2.1, 2.0], [3.1, 3.1], [4.9, 4.7]]
error = mse(original, reconstructed)
print({"mse": round(error, 4)})
assert error >= 0
# 练习：分别在原始尺度和标准化尺度计算误差并解释差异

def make_features(row, mode):
    base = [row["minutes"], row["exercises"]]
    if mode == "base":
        return base
    if mode == "interaction":
        return base + [row["minutes"] * row["exercises"]]
    if mode == "ratio":
        return base + [row["minutes"] / max(row["sessions"], 1)]
    raise ValueError("未知特征方案")

rows = [
    {"minutes": 20, "exercises": 2, "sessions": 4},
    {"minutes": 40, "exercises": 3, "sessions": 5},
]
for mode in ("base", "interaction", "ratio"):
    values = [make_features(row, mode) for row in rows]
    print(mode, {"dimension": len(values[0]), "first": values[0]})
# 练习：把每个方案接到同一评估器，比较验证指标而非训练误差

class Standardizer:
    def fit(self, values):
        self.mean = sum(values) / len(values)
        variance = sum((value - self.mean) ** 2 for value in values) / len(values)
        self.scale = variance ** 0.5 or 1.0
        return self

    def transform(self, values):
        if not hasattr(self, "mean"):
            raise RuntimeError("必须先 fit 训练数据")
        return [(value - self.mean) / self.scale for value in values]

class MeanRegressor:
    def fit(self, values, labels):
        self.value = sum(labels) / len(labels)
        return self

    def predict(self, values):
        return [self.value] * len(values)

train_x = [10.0, 12.0, 14.0]
train_y = [1.0, 2.0, 3.0]
valid_x = [13.0, 16.0]
scaler = Standardizer().fit(train_x)
model = MeanRegressor().fit(scaler.transform(train_x), train_y)
print(model.predict(scaler.transform(valid_x)))
assert scaler.mean == 12.0

from collections import defaultdict
from math import exp

def read_rows():
    return [
        {"id": "a1", "user": "u1", "title_len": 18, "minutes_7d": 80, "label": 1},
        {"id": "a2", "user": "u1", "title_len": 8, "minutes_7d": 12, "label": 0},
        {"id": "b1", "user": "u2", "title_len": 25, "minutes_7d": 66, "label": 1},
        {"id": "b2", "user": "u2", "title_len": 6, "minutes_7d": 5, "label": 0},
        {"id": "c1", "user": "u3", "title_len": 14, "minutes_7d": 40, "label": 0},
        {"id": "c2", "user": "u3", "title_len": 22, "minutes_7d": 95, "label": 1},
    ]

def make_features(row):
    return [row["title_len"] / 25, row["minutes_7d"] / 100]

def score(features, weights=(2.4, 1.7), bias=-2.0):
    value = bias + sum(x * w for x, w in zip(features, weights))
    return 1 / (1 + exp(-value))

def report(rows, threshold):
    counts = defaultdict(int)
    for row in rows:
        probability = score(make_features(row))
        prediction = int(probability >= threshold)
        counts[(row["label"], prediction)] += 1
    tp, fp = counts[(1, 1)], counts[(0, 1)]
    fn, tn = counts[(1, 0)], counts[(0, 0)]
    precision = tp / (tp + fp) if tp + fp else 0.0
    recall = tp / (tp + fn) if tp + fn else 0.0
    return {"tp": tp, "fp": fp, "fn": fn, "tn": tn, "precision": precision, "recall": recall}

rows = read_rows()
assert len({row["id"] for row in rows}) == len(rows)
print(report(rows, threshold=0.5))
# 练习：按用户群组切分，再只在验证集选择 threshold

review = {
    "task": "预测未来七天是否收藏",
    "sample_unit": "一次展示",
    "prediction_time": "展示时刻",
    "split": "按用户分组并保持时间顺序",
    "baseline": "多数类与固定规则",
    "main_metric": "precision_at_recall_0.7",
    "leakage_checks": ["未来字段", "重复用户", "全量预处理"],
    "next_topic": "神经元、前向传播、损失与反向传播",
}

required = {"task", "sample_unit", "prediction_time", "split", "baseline", "main_metric"}
missing = required - review.keys()
if missing:
    raise ValueError(f"复盘缺少: {sorted(missing)}")
print("chapter review passed")