简介这是一套基于Python实现的机器学习算法源码集合涵盖逻辑回归、SVM、KMeans、DBSCAN、岭回归、BP、MeanShift、矩阵分解、随机森林、协同过滤等十余种经典算法并配有对应训练/测试数据与说明文档适合具备一定Python基础、希望系统学习算法原理与实现细节的开发者也可作为高校机器学习课程的辅助教学材料。整个压缩包共54个文件以31个Python源文件为核心配合19个文本数据文件及少量专用测试数据总大小仅323KB目录按算法分章组织定位清晰方便按需查阅。目前已有361人学习使用。通过这份源码读者可以对照代码理解每种算法的内部计算流程与适用场景借助自带的数据集快速运行验证省去自行准备数据的麻烦还能从项目结构中学到如何组织一个多算法共存的学习型代码库是理论与实践结合的实用参考资料。1. 基于Python的机器学习算法设计源码为什么“手写一遍”比调包更值得很多人入门机器学习的第一件事是打开 sklearn 敲一句 import。模型是跑通了但一被问到“梯度下降每一步到底在更新什么”或者特征高度相关时闭式解为什么直接崩掉就答不上来。而标题里说的“基于 Python 的机器学习算法设计源码”本质上不是把别人的仓库拖下来改两行而是把常见算法从数学公式翻译成可运行、可测试、可扩展的 Python 代码。这篇笔记适合这么一批人已经会用 sklearn、看得懂 Python 语法但觉得模型是个黑匣子的调包工程师以及需要把算法改造成自定义损失函数或小规模嵌入式场景的开发者。下面会从统一接口设计讲起依次手写线性回归、KNN、决策树最后给出交叉验证和性能剖析的方法并重点说明手写源码时最容易翻车的几个细节。2. 先把算法统一成 fit/predict 接口源码骨架与评价函数设计2.1 为什么接口先行我见过不少人自己写算法时每个文件各写各的线性回归定义一个 train 函数KNN 又定义 distance 和 vote 两个函数。看着很灵活一旦要做多模型对比测试代码就得写好几份想写个自动调参脚本函数签名又对不上。sklearn 把 fit(X, y) 和 predict(X) 定为通用接口我们在手写源码时也应该照做只是不依赖 sklearn纯 numpy 实现。下面这个基类完全可以作为整个算法库的公共父类后续每个模型都继承它# base.py import numpy as np class BaseModel: 统一接口所有手写模型都继承这个基类。 def fit(self, X, y): raise NotImplementedError def predict(self, X): raise NotImplementedError def score(self, X, y): pred self.predict(X) return np.mean(pred y) # 分类默认用 accuracy回归模型可覆盖fit 负责学习参数predict 负责把学到的规律应用到新样本。score 先默认写成分类准确率回归模型后面可以覆盖成 R2 或 RMSE。继承带来的好处非常直接后面写交叉验证、网格搜索时只需要传入“模型对象”这一个抽象不需要为每个算法单独写测试入口。接口风格优点缺点sklearn 风格 fit/predict统一评估、方便交叉验证、便于扩展需要稍微约束一下函数签名各自为政写起来随意对比和调参时到处打补丁2.2 数据生成与评价函数源码设计不需要一上来就上真实数据集。自己造一个数据生成器能快速验证算法有没有 bug。下面是常用的回归数据生成函数和几个评价函数# utils.py import numpy as np def make_regression(n_samples200, n_features3, noise0.5, random_state42): rng np.random.default_rng(random_state) X rng.standard_normal((n_samples, n_features)) w rng.standard_normal(n_features) * 2.0 y X w rng.normal(0, noise, n_samples) return X, y def mse(y_true, y_pred): return np.mean((y_true - y_pred) ** 2) def rmse(y_true, y_pred): return np.sqrt(mse(y_true, y_pred)) def accuracy(y_true, y_pred): return np.mean(np.asarray(y_true) np.asarray(y_pred))n_samples 默认 200对源码调试来说足够快noise 控制信噪比想验证模型抗噪能力就把 noise 调大。random_state 传随机种子让每次生成的数据可复现。这里用np.random.default_rng(random_state)而不是np.random.seed是为了避免污染全局随机状态。mse 对离群点更敏感rmse 会把误差拉回原始量纲方便脑内评估accuracy 专门给分类算法用。2.3 数据划分先写一个不带泄漏的 train_test_split评估算法前必须把训练集和测试集分开。很多设计得比较随意的源码直接把整个数据集丢给 fit再拿同一份数据评估训练误差当然好看部署到真实数据上立刻现原形。手写一个划分函数# utils.py def train_test_split(X, y, test_size0.2, random_stateNone): rng np.random.default_rng(random_state) idx rng.permutation(len(X)) cut int(len(X) * (1 - test_size)) train_idx, test_idx idx[:cut], idx[cut:] return X[train_idx], X[test_idx], y[train_idx], y[test_idx]rng.permutation把索引整体打乱切出互不重叠的两份。test_size 默认 0.2也就是常见的 8:2 划分。注意这里没有做特征缩放因为缩放必须在划分之后单独拟合训练集统计量否则测试集信息会通过均值、标准差泄漏进模型这一点后面避坑章节会专门讲。3. 用纯 NumPy 写线性回归源码最小二乘、梯度下降与收敛判断3.1 最小二乘闭式解的实现线性回归是十大机器学习算法里最容易从零写起的一个。目标是找到权重 w让残差平方和最小。当特征数不多、矩阵可逆时可以直接解正规方程。为了避免 X^T X 奇异导致权重爆炸通常加一个很小的 l2 正则项。下面是一个同时支持闭式解和梯度下降的完整实现# linear_regression.py import numpy as np class LinearRegression: def __init__(self, solvernormal, l21e-4, lr0.01, max_iter1000, tol1e-6): self.solver solver self.l2 l2 self.lr lr self.max_iter max_iter self.tol tol self.w None def fit(self, X, y): X np.asarray(X, dtypefloat) y np.asarray(y, dtypefloat) X np.c_[np.ones(len(X)), X] # 加一列 1用来学偏置 if self.solver normal: A X.T X self.l2 * np.eye(X.shape[1]) self.w np.linalg.solve(A, X.T y) elif self.solver gd: self.w np.zeros(X.shape[1]) last_loss None for i in range(self.max_iter): pred X self.w loss np.mean((pred - y) ** 2) grad X.T (pred - y) / len(X) self.w - self.lr * grad if last_loss is not None and abs(last_loss - loss) self.tol: break last_loss loss else: raise ValueError(solver 只支持 normal 或 gd) return self def predict(self, X): X np.asarray(X, dtypefloat) X np.c_[np.ones(len(X)), X] return X self.w闭式解用np.linalg.solve而不是np.linalg.inv数值稳定性高一个档次而且速度更快。梯度下降部分grad 是均方误差对 w 的导数除以 len(X) 是为了让梯度不受样本量影响学习率 lr 默认 0.01实际使用时常需要降到 0.001 或更低。迭代终止条件是连续两轮 loss 变化小于 tol这里 tol 是绝对变化量如果数据本身量纲很大可以适当放宽到 1e-4。参数默认值作用什么情况调solvernormal选闭式解还是梯度下降特征数过万时考虑 gdl21e-4正则化强度特征共线时调到 1e-2lr0.01梯度下降步长loss 震荡时下调到 0.001max_iter1000最大迭代轮数loss 仍在下降时加大tol1e-6收敛阈值数据量纲大时放宽到 1e-43.2 特征缩放闭式解无所谓梯度下降容易被量纲带崩当一个特征范围是 0 到 1另一个是 0 到 100000梯度下降的更新方向基本被大数值特征主导loss 曲线会像锯齿一样震荡甚至直接发散。所以在梯度下降前做标准化是源码里的常规操作def zscore(X, meanNone, stdNone): X np.asarray(X, dtypefloat) if mean is None or std is None: mean X.mean(axis0) std X.std(axis0) std[std 0] 1.0 return (X - mean) / std, mean, stdzscore 把每个特征变成均值 0、方差 1。关键点在于fit 训练集时得到一个 mean 和 std测试集转换时必须复用同一组统计量绝不能重新计算。实际写源码时很多人在这里偷偷引入泄漏后面避坑章节会展开。闭式解不需要特征缩放因为正规方程对量纲不敏感梯度下降则强烈建议先做这一步。3.3 收敛判断与 loss 轨迹判断训练是否正常光看最终 loss 不够最好把每轮 loss 记录下来。可以在 fit 里加一行if i % 100 0: print(fiter {i:4d}, loss {loss:.4f})配合 3.2 的特征缩放lr 从 0.01 起步通常能稳定下降。如果 loss 一开始就在变大说明步长太大把 lr 除以 10 再试。还有一种常见情况loss 在某个值附近反复横跳但不再下降这时不一定是代码 bug而是 lr 偏大导致一直在最优点附近振荡把 lr 调小即可。我一般的做法是先用 solvernormal 跑通一遍拿到参考权重再切到梯度下降对照 loss 曲线这样能快速确认梯度计算没错。4. 从零实现 KNN 和决策树源码分类算法的两套不同思路4.1 KNNfit 只是记住数据predict 才需要算距离KNN 是最直观的有监督算法新样本和训练集里每个样本算距离取前 k 个近邻投票。它没有显式的训练过程fit 阶段只是把数据存下来。手写源码的核心在 predict 的距离矩阵计算# knn.py import numpy as np class KNN: def __init__(self, k5): self.k k def fit(self, X, y): self.X np.asarray(X) self.y np.asarray(y) return self def predict(self, X): X np.asarray(X) # 广播计算欧氏距离X: (m,d)训练集 self.X: (n,d) dists np.sqrt(((X[:, None, :] - self.X[None, :, :]) ** 2).sum(axis-1)) idx np.argsort(dists, axis1)[:, :self.k] knn_y self.y[idx] preds [] for row in knn_y: preds.append(np.bincount(row).argmax()) return np.array(preds)X[:, None, :]把形状变成 (m, 1, d)self.X[None, :, :]变成 (1, n, d)广播之后得到 (m, n, d) 的差矩阵最后求和开根号。np.bincount要求标签是非负整数用于统计每个近邻类别出现次数argmax 取出票数最多的类。k 太小预测容易翻车k 太大又把远距离样本拉进投票。实际参数选择建议从 k5 开始用交叉验证看曲线再调整。如果特征量纲差异大欧氏距离会被大数值特征主导所以 KNN 前同样要做 zscore。4.2 决策树按基尼指数做最优切分决策树分类的每个节点做一件事找一个特征和阈值把样本分成左右两组让两组内部纯度提升最大。基尼指数越小越纯公式是 1 - Σ p^2。实现时先写分裂评估函数def gini(y): _, counts np.unique(y, return_countsTrue) p counts / counts.sum() return 1 - (p ** 2).sum() def best_split(X, y): n len(y) parent_gini gini(y) best_gain, best_feat, best_thr -1, None, None for feat in range(X.shape[1]): vals np.unique(X[:, feat]) # 候选阈值取相邻取值的中位点 for thr in (vals[:-1] vals[1:]) / 2: mask X[:, feat] thr if mask.sum() 0 or (~mask).sum() 0: continue weighted (mask.sum() / n * gini(y[mask]) (~mask).sum() / n * gini(y[~mask])) gain parent_gini - weighted if gain best_gain: best_gain, best_feat, best_thr gain, feat, thr return best_feat, best_thr候选阈值取相邻特征值的均值np.unique自带排序保证阈值从小到大。跳过所有样本都落到同一侧的分裂方式因为这种分裂没有意义。基尼增益越大说明分裂后纯度提升越多。有了 best_split再套一个递归建树的类class DecisionTree: def __init__(self, max_depth5, min_samples_split2): self.max_depth max_depth self.min_samples_split min_samples_split def fit(self, X, y): X np.asarray(X) y np.asarray(y) self.tree_ self._build(X, y, depth0) return self def _build(self, X, y, depth): if (len(np.unique(y)) 1 or depth self.max_depth or len(y) self.min_samples_split): return {leaf: True, class: int(np.bincount(y).argmax())} feat, thr best_split(X, y) if feat is None: return {leaf: True, class: int(np.bincount(y).argmax())} mask X[:, feat] thr return { leaf: False, feature: feat, threshold: thr, left: self._build(X[mask], y[mask], depth 1), right: self._build(X[~mask], y[~mask], depth 1), } def predict(self, X): X np.asarray(X) return np.array([self._predict_one(x, self.tree_) for x in X]) def _predict_one(self, x, node): if node[leaf]: return node[class] if x[node[feature]] node[threshold]: return self._predict_one(x, node[left]) return self._predict_one(x, node[right])max_depth 默认 5能有效防止无限生长min_samples_split 表示节点样本数少于该值时不再分。四个停止条件分别是纯节点、超深、样本太少、找不到合法分裂。树节点用 dict 实现直观但内存占用偏高真到大规模场景可以改成数组索引或链表。对决策树来说不是深度越大越好源码里把这两个默认参数透出就是为了让你在项目里通过网格搜索找到甜点区。4.3 把两个分类模型放进同一个框架里对比接口统一之后评估代码缩到很短。先生成一个二分类数据def make_classification(n_samples200, n_features2, random_state42): rng np.random.default_rng(random_state) X rng.standard_normal((n_samples, n_features)) y ((X[:, 0] ** 2 X[:, 1] ** 2) 4).astype(int) return X, y然后用同一个训练流程测两个模型from utils import make_classification, train_test_split, accuracy from knn import KNN from decision_tree import DecisionTree X, y make_classification() X_train, X_test, y_train, y_test train_test_split(X, y, random_state0) for model in [KNN(k5), DecisionTree(max_depth5)]: model.fit(X_train, y_train) acc accuracy(y_test, model.predict(X_test)) print(type(model).__name__, facc{acc:.3f})KNN 对这种圆形分界通常适应良好决策树用轴平行切分则需要更深才能逼近。这个对比不是为了证明谁强而是验证刚写的源码没有低级 bug。如果某个模型准确率只有 0.5 左右先检查数据是否泄露、标签是否错位再怀疑算法实现。5. 手写算法源码最容易踩的五个坑数据泄漏、梯度爆炸与复现问题5.1 数据泄漏归一化在划分前做测试集信息被提前看到现象模型在“测试集”上准确率很高甚至比调出来的 sklearn 结果还高但一到真实场景就明显下滑。原因手写源码时图省事对全量 X 先做 zscore 或 min-max 缩放再划分训练测试。这样一来测试集的均值和标准差已经参与过特征变换相当于模型提前看到了测试集的分布信息。这种泄漏不报错结果又好看最容易骗到自己。解决先划分再单独算训练集统计量。X_train, X_test, y_train, y_test train_test_split(X, y) mean, std X_train.mean(axis0), X_train.std(axis0) X_train (X_train - mean) / std X_test (X_test - mean) / std测试集只能用训练集的 mean/std不能用自己重新算的。这个习惯要刻进源码模板里。5.2 梯度下降 loss 变成 NaN 或一直涨现象前几轮 loss 正常突然变成 nan或者每轮 loss 一轮比一轮大。原因学习率太大权重更新一步越过了最优点梯度越来越大最后溢出。特征没做归一化时即使 lr0.01 也可能在量纲大的特征上爆炸。解决先做 zscore把 lr 从 0.01 开始往下试。写个判断经验loss 波动但整体下降问题不大loss 一路向上立刻把 lr 除以 10。训练循环里加一行if np.isnan(loss): raise ValueError(loss is nan, try smaller lr or zscore)5.3 闭式解在特征共线时出现极大权重现象训练集里存在高度相关的两列比如 x2 2 * x1闭式解算出的权重数值极大预测值异常。原因X^T X 接近奇异np.linalg.solve虽然比求逆稳但接近奇异时结果仍然不可靠会把噪声放大。解决源码里给正规方程加 l2 正则项l2 从 1e-4 调到 1e-2 通常能压住权重。更彻底的做法是检查特征相关性并删掉冗余列。在线性回归源码实现里默认带 l2这是我很坚持的一个习惯。5.4 决策树过拟合max_depth 设太大叶子全是纯节点现象训练集准确率接近 100%测试集准确率明显更低而且树的结构很深。原因树不停分裂到每个叶子只剩一两个样本把噪声当成规律记了下来。手写源码时停止条件如果只写“纯节点才停”几乎必然过拟合。解决设 max_depth3 到 7min_samples_leaf5 起。观察测试集准确率随深度先升后降取峰值附近的深度。这个“深了反而差”的反直觉现象在决策树源码里几乎必现。5.5 复现不了别人的结果不一定是算法写错现象公式一样、数据一样跑出来的精度差几个点反复检查也找不到明显问题。原因随机种子不一致、数据切分顺序不同、初始化权重不同、归一化方式不同甚至是评价指标不一样有人用 accuracy 有人用 F1。解决把整个实验流程固定成可复现设置。数据生成统一走 utils 里的函数并传 random_state模型初始化不读全局随机状态记录训练集 mean/std。我现在的做法是每个项目根目录放一个 experiments.py所有核心实验只通过它进入参数不散落在各个脚本里这样下次回头看还能原样跑通。6. 给手写源码做基准验证交叉验证、超参数搜索与性能剖析6.1 手写 k 折交叉验证单次训练测试划分太看运气。交叉验证把样本切成 k 份轮流拿其中一份做验证集其余训练最后取均值。代码量不大import copy def kfold_cv(model, X, y, k5, random_state42): rng np.random.default_rng(random_state) idx rng.permutation(len(X)) scores [] for i in range(k): val_idx idx[i::k] tr_idx np.setdiff1d(idx, val_idx) m copy.deepcopy(model) # 避免上一个 fold 的状态残留 m.fit(X[tr_idx], y[tr_idx]) scores.append(accuracy(y[val_idx], m.predict(X[val_idx]))) return np.mean(scores), np.std(scores)返回值是均值加减标准差。标准差大说明模型对数据划分敏感不稳定。deepcopy 很重要因为线性回归 fit 后 w 会被覆盖不复制对象第二次训练就带着上一个 fold 的初始化状态。6.2 超参数搜索手写源码时不一定要引入网格搜索库循环就够。比如 KNN 选 kfor k in [1, 3, 5, 9, 15]: acc, std kfold_cv(KNN(kk), X, y) print(k, round(acc, 3), round(std, 3))看趋势而不是单个点k1 在验证集上方差很大k 太大偏差上升合适的区间通常在两头的某个稳定段。决策树同理把 max_depth 从 2 扫到 10找测试分数开始下降的位置。6.3 用 timeit 找出 KNN 的瓶颈KNN 预测要算每个样本到全部训练集的距离复杂度 O(mnd)样本量上来后明显变慢。用性能计数器测量最直接import time start time.perf_counter() pred knn.predict(X_test) elapsed time.perf_counter() - start print(fpredict elapsed: {elapsed:.3f}s)实测中几千样本、几十维特征numpy 向量化还能扛上万样本就开始卡。真要优化下个方向是 KD-Tree 或 BallTree而不是继续优化当前的距离矩阵循环。我现在写源码有个固定习惯每个模型文件夹里放一份干净版和一份带 debug 日志的版本改了算法先跑基准数据再跑小规模业务数据确认行为一致再上全量。手写机器学习算法常见的坑最后都指向“输入数据范围没留意”这一件事。希望这篇笔记对你手写 Python 机器学习算法源码有帮助。本文还有配套的精品资源点击获取