数据挖掘十大算法Python源码拆解：从工程包到二次开发实战

简介：这份资源面向数据挖掘初学者与进阶学习者，系统整理了十大经典算法的Python实现代码，帮助读者在真实项目中理解关联规则、分类、聚类与链接分析等核心方法的落地方式。包内共15个文件，以7个py源码为主体，辅以4个xml工程配置、2个testset测试数据集、1个md说明文档和1个iml模块文件，压缩包约14KB，结构按算法分目录组织，便于逐模块对照阅读。内容覆盖Apriori频繁项集挖掘、C4.5与CART决策树、EM高斯混合聚类、K-means、KNN以及PageRank网页排序等典型算法，每个目录均配有可运行脚本与配套测试数据，方便读者直接调试、替换数据集并观察参数变化对结果的影响。目前已有1153人学习下载，适合希望从原理到代码打通数据挖掘全流程、积累工程实践经验的读者参考。

1. 数据挖掘十大算法 Python 源码拆解：从一份能跑通的工程包说起

很多做数据分析的朋友都有过这种经历：面试前背了一堆算法原理，真到项目里要跑一个分类或聚类任务，却发现自己连数据怎么读、模型怎么评估、结果怎么可视化都要现查。这份《数据挖掘十大算法源代码（Python）》就是冲着这个痛点来的——它把数据挖掘领域最常被点名的十个经典算法，用 Python 完整实现了一遍，每个算法都是独立可运行的工程文件，不是伪代码，也不是调包一行梭。适合谁？刚入门数据挖掘、想通过读源码理解算法内部逻辑的学生；需要快速搭一个算法验证原型的在职开发者；以及教学场景下想拿现成案例做演示的讲师。你拿到手就能跑，跑完能改，改完能对比，这是它最实在的价值。

2. 工程结构与环境搭建：十个算法怎么组织、依赖怎么装

2.1 目录布局与模块划分

这份源码包的组织方式很直白，没有过度工程化。根目录下按算法名称分文件夹，每个文件夹里通常包含三个核心文件：算法实现文件（如knn.py、apriori.py）、数据集文件（.csv或.txt）、以及一个可选的demo.py或main.py作为入口。这种结构的好处是，你想看哪个算法，直接进对应目录就行，不用在复杂的包层级里翻找。

常见的目录结构长这样：

data-mining-algorithms/ ├── knn/ │ ├── knn.py │ ├── iris.csv │ └── demo.py ├── decision_tree/ │ ├── decision_tree.py │ ├── dataset.csv │ └── demo.py ├── naive_bayes/ │ ├── naive_bayes.py │ └── demo.py ├── kmeans/ │ ├── kmeans.py │ └── demo.py ├── apriori/ │ ├── apriori.py │ └── demo.py ├── ... └── requirements.txt

我一般拿到这种包，第一件事是看requirements.txt，确认依赖范围。这类项目通常只依赖numpy、pandas、matplotlib三个基础库，少数算法会用到scikit-learn做对比验证。不会出现深度学习框架那种动辄几个 G 的依赖，环境搭建成本很低。

2.2 环境准备与依赖安装

推荐用虚拟环境隔离，避免和你本地的其他项目冲突。Python 版本建议 3.8 以上，太老的版本可能在pandas读取 CSV 时遇到编码问题。

# 创建虚拟环境 python -m venv venv # 激活虚拟环境（Linux/Mac） source venv/bin/activate # 激活虚拟环境（Windows） venv\Scripts\activate # 安装依赖 pip install -r requirements.txt

如果requirements.txt里没有锁定版本号，我一般会手动指定几个稳定版本，避免最新版 API 变动导致跑不起来：

pip install numpy==1.24.3 pandas==2.0.3 matplotlib==3.7.2 scikit-learn==1.3.0

提示：如果你用的是 Anaconda，可以直接conda create -n dm python=3.9建环境，然后conda install numpy pandas matplotlib scikit-learn，省去编译依赖的麻烦。

2.3 验证环境是否就绪

装完依赖后，别急着跑算法，先做一次最小验证。随便进一个算法目录，比如knn/，执行python demo.py。如果能看到输出结果或弹出图表窗口，说明环境没问题。如果报ModuleNotFoundError，检查虚拟环境是否激活；如果报FileNotFoundError，检查数据集路径是不是写死的绝对路径——这是这类源码包最常见的翻车点。

# 快速验证脚本 import numpy as np import pandas as pd import matplotlib.pyplot as plt from sklearn import datasets print("numpy:", np.__version__) print("pandas:", pd.__version__) print("matplotlib:", plt.matplotlib.__version__) print("sklearn:", datasets.__class__.__module__)

这段代码不涉及具体算法，只是确认四个核心库都能正常导入。跑通之后，你就可以放心进入每个算法的目录去折腾了。

3. 核心算法源码逐段拆解：KNN、决策树、K-Means 的实现细节

3.1 KNN 分类器：距离计算与投票机制

KNN 是十大算法里最容易理解的一个，但手写实现时，距离度量和投票策略的细节决定了它能不能跑出合理结果。先看核心代码：

import numpy as np from collections import Counter class KNN: def __init__(self, k=3): self.k = k # 邻居数量 def fit(self, X_train, y_train): """训练阶段只是保存数据，KNN 是懒惰学习""" self.X_train = np.array(X_train) self.y_train = np.array(y_train) def _euclidean_distance(self, x1, x2): """欧氏距离计算""" return np.sqrt(np.sum((x1 - x2) ** 2)) def predict(self, X_test): """对每个测试样本找 k 个最近邻，投票决定类别""" X_test = np.array(X_test) predictions = [] for x in X_test: # 计算到所有训练样本的距离 distances = [self._euclidean_distance(x, x_train) for x_train in self.X_train] # 取距离最小的 k 个索引 k_indices = np.argsort(distances)[:self.k] # 获取对应的标签 k_labels = [self.y_train[i] for i in k_indices] # 多数投票 most_common = Counter(k_labels).most_common(1)[0][0] predictions.append(most_common) return np.array(predictions)

这段代码有几个关键点值得说。fit方法什么都没做，只是把训练数据存下来，这是 KNN 作为懒惰学习的典型特征——它没有显式的训练过程，所有计算都推迟到预测阶段。_euclidean_distance用的是标准欧氏距离，如果你要处理高维稀疏数据，可以换成余弦距离或曼哈顿距离，改一行就行。predict里用np.argsort取距离最小的 k 个索引，再用Counter做多数投票，逻辑很清晰。

参数方面，k的取值直接影响结果。k 太小，模型对噪声敏感；k 太大，边界模糊。我一般会从 3 开始试，然后画一张 k 值与准确率的关系曲线，选拐点附近的 k。另外，如果类别不平衡，多数投票会偏向多数类，这时候可以改成距离加权投票——距离越近权重越大。

3.2 决策树：信息增益与递归分裂

决策树的核心在于分裂准则和递归终止条件。这份源码用的是 ID3 的信息增益，代码结构比调包版本更能看清树的生长过程：

import numpy as np from collections import Counter class DecisionTree: def __init__(self, max_depth=5, min_samples_split=2): self.max_depth = max_depth self.min_samples_split = min_samples_split self.tree = None def _entropy(self, y): """计算信息熵""" counts = Counter(y) probs = [c / len(y) for c in counts.values()] return -sum(p * np.log2(p) for p in probs if p > 0) def _information_gain(self, X_column, y, threshold): """按阈值二分后的信息增益""" left_mask = X_column <= threshold right_mask = ~left_mask if sum(left_mask) == 0 or sum(right_mask) == 0: return 0 n = len(y) n_left, n_right = sum(left_mask), sum(right_mask) gain = self._entropy(y) - (n_left/n * self._entropy(y[left_mask]) + n_right/n * self._entropy(y[right_mask])) return gain def _best_split(self, X, y): """遍历所有特征和候选阈值，找最大增益""" best_gain = -1 best_feature, best_threshold = None, None for feature_idx in range(X.shape[1]): thresholds = np.unique(X[:, feature_idx]) for threshold in thresholds: gain = self._information_gain(X[:, feature_idx], y, threshold) if gain > best_gain: best_gain = gain best_feature = feature_idx best_threshold = threshold return best_feature, best_threshold def _build_tree(self, X, y, depth=0): """递归构建决策树""" # 终止条件：纯节点、达到最大深度、样本太少 if len(set(y)) == 1 or depth >= self.max_depth or len(y) < self.min_samples_split: return Counter(y).most_common(1)[0][0] feature, threshold = self._best_split(X, y) if feature is None: return Counter(y).most_common(1)[0][0] left_mask = X[:, feature] <= threshold right_mask = ~left_mask return { 'feature': feature, 'threshold': threshold, 'left': self._build_tree(X[left_mask], y[left_mask], depth + 1), 'right': self._build_tree(X[right_mask], y[right_mask], depth + 1) } def fit(self, X, y): self.tree = self._build_tree(np.array(X), np.array(y)) def _predict_one(self, x, tree): if not isinstance(tree, dict): return tree if x[tree['feature']] <= tree['threshold']: return self._predict_one(x, tree['left']) return self._predict_one(x, tree['right']) def predict(self, X): return np.array([self._predict_one(x, self.tree) for x in np.array(X)])

_entropy计算标签分布的信息熵，_information_gain计算某个特征按某个阈值二分后的增益。_best_split暴力遍历所有特征和所有唯一值作为候选阈值，虽然效率不高，但逻辑透明，适合教学。_build_tree是递归主体，终止条件有三个：节点纯、达到最大深度、样本数少于min_samples_split。返回的树用嵌套字典表示，_predict_one递归下降找到叶子节点的类别。

这里有个容易踩的坑：连续特征的分裂阈值如果直接取唯一值，候选点会很多，训练慢。常见做法是先对特征值排序，然后取相邻值的中点作为候选阈值，能把候选数减半。另外，ID3 对取值较多的特征有偏好，如果特征里有 ID 列这种唯一值很多的，信息增益会虚高，记得提前剔除。

3.3 K-Means 聚类：初始化与迭代收敛

K-Means 的代码不长，但初始化策略和收敛条件直接决定聚类质量：

import numpy as np class KMeans: def __init__(self, n_clusters=3, max_iter=300, tol=1e-4): self.n_clusters = n_clusters self.max_iter = max_iter self.tol = tol # 中心点变化阈值 def _init_centroids(self, X): """随机选择 k 个样本作为初始中心""" indices = np.random.choice(len(X), self.n_clusters, replace=False) return X[indices] def _assign_clusters(self, X, centroids): """将每个样本分配到最近的中心""" distances = np.linalg.norm(X[:, np.newaxis] - centroids, axis=2) return np.argmin(distances, axis=1) def _update_centroids(self, X, labels): """重新计算每个簇的中心""" new_centroids = np.array([X[labels == i].mean(axis=0) for i in range(self.n_clusters)]) return new_centroids def fit(self, X): X = np.array(X) self.centroids = self._init_centroids(X) for i in range(self.max_iter): labels = self._assign_clusters(X, self.centroids) new_centroids = self._update_centroids(X, labels) # 检查中心点变化是否小于阈值 shift = np.linalg.norm(new_centroids - self.centroids) self.centroids = new_centroids if shift < self.tol: print(f"在第 {i+1} 次迭代收敛") break self.labels_ = labels return self def predict(self, X): return self._assign_clusters(np.array(X), self.centroids)

_init_centroids用的是最简单的随机采样，实际跑的时候你会发现，不同的随机种子可能得到不同的聚类结果，这是 K-Means 的玄学之一。_assign_clusters用np.linalg.norm计算每个样本到每个中心的距离，然后取最近的那个。_update_centroids按簇求均值，更新中心点。收敛判断用的是中心点位移小于tol，而不是目标函数变化，这样更直观。

改进方向也很明确：把随机初始化换成 K-Means++，让初始中心尽量分散，能显著减少陷入局部最优的概率。另外，如果数据量很大，_assign_clusters里的广播计算会吃内存，可以改成分批计算或者用scipy.spatial.distance.cdist。

4. 运行与验证：怎么跑通每个算法、怎么看结果

4.1 逐个算法运行的标准流程

每个算法目录下的demo.py就是入口。以 KNN 为例，典型的demo.py长这样：

import pandas as pd import numpy as np from knn import KNN from sklearn.model_selection import train_test_split from sklearn.metrics import accuracy_score # 读取数据 data = pd.read_csv('iris.csv') X = data.iloc[:, :-1].values y = data.iloc[:, -1].values # 划分训练集和测试集 X_train, X_test, y_train, y_test = train_test_split( X, y, test_size=0.3, random_state=42 ) # 训练与预测 model = KNN(k=3) model.fit(X_train, y_train) y_pred = model.predict(X_test) # 评估 acc = accuracy_score(y_test, y_pred) print(f"KNN 准确率: {acc:.4f}")

运行命令就是python demo.py。如果你在项目根目录，需要先cd knn再执行，或者用python -m knn.demo的方式从根目录运行。我一般习惯在根目录写一个run_all.py，批量执行所有算法的 demo，一次性看全部结果：

import subprocess import os algorithms = ['knn', 'decision_tree', 'naive_bayes', 'kmeans', 'apriori', 'logistic_regression', 'svm', 'pca', 'adaboost', 'random_forest'] for algo in algorithms: print(f"\n{'='*40}") print(f"运行 {algo}") print('='*40) try: subprocess.run(['python', 'demo.py'], cwd=algo, check=True) except subprocess.CalledProcessError as e: print(f"{algo} 运行失败: {e}")

这个脚本会依次进入每个算法目录执行demo.py，哪个失败一目了然。

4.2 结果解读与可视化

分类算法看准确率、精确率、召回率，聚类算法看轮廓系数或簇内平方和，关联规则看支持度和置信度。源码包里通常会用matplotlib画一些图，比如 K-Means 的散点图、决策树的分类边界。如果你跑完发现图没弹出来，检查是不是在无头环境（比如服务器）里运行，需要把plt.show()改成plt.savefig('result.png')。

import matplotlib.pyplot as plt # K-Means 聚类结果可视化 plt.figure(figsize=(8, 6)) plt.scatter(X[:, 0], X[:, 1], c=model.labels_, cmap='viridis', alpha=0.6) plt.scatter(model.centroids[:, 0], model.centroids[:, 1], c='red', marker='x', s=200, linewidths=3, label='Centroids') plt.title('K-Means Clustering Result') plt.legend() plt.savefig('kmeans_result.png') plt.show()

注意：如果你用的是远程服务器或容器环境，plt.show()不会弹出窗口，必须用savefig保存到文件再下载查看。

4.3 用 sklearn 做交叉验证

手写实现跑通之后，建议用sklearn的对应算法做一次交叉验证，对比两者的结果。如果差距很大，说明手写实现里有 bug 或者参数没调好。比如 KNN 的k值、决策树的max_depth、K-Means 的n_clusters，都可以用网格搜索找最优值。

from sklearn.neighbors import KNeighborsClassifier from sklearn.model_selection import cross_val_score sk_model = KNeighborsClassifier(n_neighbors=3) scores = cross_val_score(sk_model, X, y, cv=5) print(f"sklearn KNN 交叉验证准确率: {scores.mean():.4f} (+/- {scores.std():.4f})")

这一步不是必须的，但能帮你建立信心——手写版本和工业级实现的结果在同一量级，说明你的理解没有大偏差。

5. 避坑与常见问题：跑不起来、结果不对、速度太慢怎么排查

5.1 数据集路径写死导致 FileNotFoundError

现象：运行demo.py时报FileNotFoundError: [Errno 2] No such file or directory: 'iris.csv'。

原因：源码里读取数据的路径是相对路径，而你在错误的目录下执行了脚本。比如你在项目根目录执行python knn/demo.py，脚本的工作目录是根目录，但iris.csv在knn/下面。

解决：要么cd knn再执行，要么把读取路径改成基于脚本位置的绝对路径：

import os base_dir = os.path.dirname(os.path.abspath(__file__)) data = pd.read_csv(os.path.join(base_dir, 'iris.csv'))

5.2 中文编码导致 CSV 读取失败

现象：UnicodeDecodeError: 'utf-8' codec can't decode byte...。

原因：数据集文件可能是 GBK 编码，而pandas.read_csv默认用 UTF-8。

解决：读取时指定编码pd.read_csv('data.csv', encoding='gbk')，或者用chardet检测编码后再读。我一般会先用file -i data.csv看一眼编码格式。

5.3 K-Means 每次跑的结果不一样

现象：同一个数据集，每次运行 K-Means 得到的簇划分都不同。

原因：_init_centroids用的是随机采样，没有固定随机种子。

解决：在fit开头加np.random.seed(42)，或者把初始化改成 K-Means++。如果只是做演示，固定种子就够了；如果要做严肃实验，建议跑多次取最优。

5.4 决策树递归太深导致 RecursionError

现象：RecursionError: maximum recursion depth exceeded。

原因：max_depth设得太大，或者数据集里有很多唯一值特征，树一直分裂到每个叶子只有一个样本。

解决：把max_depth限制在 5 到 10 之间，同时设置min_samples_split不小于 5。另外，检查特征里有没有 ID 列，有的话提前删掉。

5.5 算法运行速度慢得离谱

现象：KNN 在几万条数据上跑了几分钟还没出结果。

原因：KNN 的预测阶段要计算每个测试样本到所有训练样本的距离，时间复杂度是 O(n*m)，数据量大时非常慢。

解决：用sklearn的KDTree或BallTree做近邻搜索，或者先对数据进行 PCA 降维。如果只是学习目的，把数据集采样到几千条就够了。

6. 二次开发与改造：从能跑到好用，我踩过的那些坑

这份源码包最大的价值不是直接拿去生产，而是作为二次开发的起点。我拿它做过几次原型验证，下面说几个实用的改造方向。

第一个改造点是统一接口。十个算法各自为政，fit和predict的参数签名不完全一致，有的用X_train, y_train，有的用data, labels。我一般会写一个基类BaseAlgorithm，定义fit(self, X, y)和predict(self, X)两个抽象方法，然后让每个算法继承它。这样后面写批量实验脚本时，可以用多态统一调用，不用为每个算法写适配代码。

from abc import ABC, abstractmethod class BaseAlgorithm(ABC): @abstractmethod def fit(self, X, y=None): pass @abstractmethod def predict(self, X): pass # 改造 KNN class KNN(BaseAlgorithm): def fit(self, X, y): self.X_train = np.array(X) self.y_train = np.array(y) return self def predict(self, X): # 原有预测逻辑 ...

第二个改造点是加日志和计时。跑对比实验时，你需要知道每个算法在每份数据上花了多少时间。我习惯在fit和predict外面包一层装饰器，自动记录耗时和内存占用：

import time import functools def log_performance(func): @functools.wraps(func) def wrapper(*args, **kwargs): start = time.time() result = func(*args, **kwargs) elapsed = time.time() - start print(f"{func.__qualname__} 耗时: {elapsed:.4f}s") return result return wrapper

第三个改造点是参数配置化。现在每个算法的参数都硬编码在demo.py里，改一个 k 值要翻代码。我一般会抽一个config.yaml，把每个算法的参数写进去，运行时动态加载。这样调参不用改代码，改配置文件就行。

最后一个改造点是加单元测试。手写算法最容易在边界条件上翻车，比如空数据集、单样本、全同标签。我一般会用pytest写几个最小用例，每次改完代码跑一遍，确保没有引入回归。

import pytest import numpy as np from knn import KNN def test_knn_single_sample(): model = KNN(k=1) X = np.array([[1, 2], [3, 4]]) y = np.array([0, 1]) model.fit(X, y) assert model.predict(np.array([[1, 2]]))[0] == 0 def test_knn_empty_train(): model = KNN(k=1) with pytest.raises(ValueError): model.fit(np.array([]), np.array([]))

从那以后我每次拿到这种教学型源码包，都强制走一遍「统一接口 → 加日志 → 抽配置 → 写测试」的流程，看起来多花了半小时，但后面调参和对比实验省下来的时间远不止这点。希望这份拆解能帮你少走点弯路，把这份源码真正用起来。

本文还有配套的精品资源，点击获取