简介：这份资源面向计算机、数据科学与大数据技术、人工智能等专业的在校学生与教师，提供基于Python实现的数据挖掘经典频繁项集挖掘算法对比测试项目，可用于课程设计、期末大作业、毕业设计或入门进阶练习。压缩包共3个文件，包含2个py源码文件与1个md项目说明文档，整体约6KB，其中源码分别实现Apriori算法与FP-Growth算法，说明文档则交代项目背景、运行方式与对比思路，便于快速理解两种算法的实现差异与效率表现。目前已有413人学习下载，具备一定参考热度。读者可借此掌握频繁项集挖掘的核心流程，对比Apriori的候选集生成与FP-Growth的FP树构建在时间与空间开销上的不同，并在此基础上拓展关联规则分析、数据集替换或性能测试等实验内容，为后续数据挖掘课程实践与项目立项提供可直接复用的代码基础。
1. 从一份 Python 数据挖掘源码说起：Apriori 和 FP-Growth 到底该用哪个
电商订单表里躺着几十万条交易记录，老板让你找出“买了啤酒的人还会买什么”。你打开 Python，搜到两个关键词：Apriori 和 FP-Growth。一个是数据挖掘课的经典算法，一个是号称快一个数量级的改进方案。问题是，它们跑在同一份数据上，结果一样吗？速度差多少？内存吃多少？我见过太多人把 Apriori 代码复制过来跑一遍，看到频繁项集就交差，结果上线时数据量翻十倍直接卡死。这份“基于 Python 实现数据挖掘 Apriori 算法与 FP-Growth 算法对比测试”的源码和项目说明，核心价值不是教你写两个算法，而是让你在同一套数据、同一套评估口径下，亲眼看到两者的分水岭在哪里。适合已经会 Python 基础语法、想真正把关联规则用到业务里的人，也适合正在做数据挖掘课程设计、需要一份可复现对比实验的读者。
2. 关联规则挖掘的两个主角：Apriori 的候选生成与 FP-Growth 的树结构
2.1 Apriori 算法为什么需要反复扫描数据集
Apriori 的核心思想建立在先验性质上：如果一个项集是频繁的，那么它的所有子集也必须是频繁的。反过来，如果一个项集是非频繁的，那么它的所有超集也一定是非频繁的。这个性质让算法可以逐层剪枝，不用穷举所有组合。
但代价也很明显。假设有 1000 个商品，理论上可能产生的项集数量是 2 的 1000 次方减一，这是天文数字。Apriori 的做法是：第一轮扫描数据库，统计每个单个商品的出现次数，筛掉低于最小支持度的；第二轮用剩下的频繁一项集两两组合，生成候选二项集，再扫描数据库统计；第三轮生成候选三项集……每一轮都要完整遍历一次数据集。
我做过一个测试，用一份 5 万条交易记录、平均每条 8 个商品的数据集，最小支持度设为 0.01。Apriori 在生成候选四项集时，扫描次数已经到了第 4 轮，耗时接近 40 秒。如果最小支持度降到 0.005，候选集数量爆炸，内存直接飙到 2GB 以上。这就是 Apriori 的命门：候选集生成和重复扫描数据库，两者叠加导致它在稠密数据集上非常吃力。
2.2 FP-Growth 如何用一棵树把扫描次数压到两次
FP-Growth 换了一个思路。它不生成候选集，而是把整个数据集压缩成一棵频繁模式树（FP-Tree），然后在这棵树上递归挖掘。
具体流程分三步。第一步，扫描一次数据库，统计每个商品的出现频次，去掉不满足最小支持度的商品，然后把剩下的商品按频次降序排列。第二步，再扫描一次数据库，对每条交易记录，按频次降序处理商品，依次插入 FP-Tree。如果路径上已有相同节点就共享，否则创建新节点，并在节点头部维护一个链表，把相同商品串起来。第三步，从频次最低的商品开始，在 FP-Tree 上找到它的所有前缀路径，构成条件模式基，递归构建条件 FP-Tree，直到树中只剩一条路径，直接枚举所有组合。
整个过程只扫描数据库两次。后续所有操作都在内存中的树上完成。这就是 FP-Growth 在速度上碾压 Apriori 的根本原因。但树结构也有代价：构建 FP-Tree 需要额外内存，而且当数据极度稀疏时，树的压缩效果不明显，优势会缩小。
2.3 用 Python 跑通 Apriori 的最小代码骨架
下面这段代码不依赖任何第三方数据挖掘库，只用 Python 标准库实现 Apriori 的核心逻辑。你可以直接复制到本地运行，观察每一步的输出。
# apriori_basic.py # 最小化 Apriori 实现，用于理解候选生成与剪枝过程 def load_dataset(): """返回一个列表，每个元素是一条交易记录（商品集合）""" return [ ['牛奶', '面包', '尿布'], ['可乐', '面包', '尿布', '啤酒'], ['牛奶', '尿布', '啤酒', '鸡蛋'], ['面包', '牛奶', '尿布', '啤酒'], ['面包', '牛奶', '尿布', '可乐'], ] def create_c1(dataset): """生成候选一项集：遍历所有交易，统计每个商品出现次数""" c1 = {} for transaction in dataset: for item in transaction: c1[frozenset([item])] = c1.get(frozenset([item]), 0) + 1 return c1 def scan_dataset(dataset, candidates, min_support_count): """扫描数据集，统计候选项集出现次数，返回满足最小支持度的项集""" ss_cnt = {} for transaction in dataset: for candidate in candidates: if candidate.issubset(transaction): ss_cnt[candidate] = ss_cnt.get(candidate, 0) + 1 ret_list = [] for key in ss_cnt: if ss_cnt[key] >= min_support_count: ret_list.append(key) return ret_list, ss_cnt def apriori_gen(freq_sets, k): """由频繁 k-1 项集生成候选 k 项集，并做剪枝""" ret_list = [] len_lk = len(freq_sets) for i in range(len_lk): for j in range(i + 1, len_lk): l1 = list(freq_sets[i])[:k - 2] l2 = list(freq_sets[j])[:k - 2] l1.sort() l2.sort() if l1 == l2: ret_list.append(freq_sets[i] | freq_sets[j]) return ret_list def apriori(dataset, min_support=0.5): """主函数：逐层生成频繁项集""" c1 = create_c1(dataset) dataset_len = len(dataset) min_support_count = min_support * dataset_len l1, support_data = scan_dataset(dataset, c1.keys(), min_support_count) l = [l1] k = 2 while len(l[k - 2]) > 0: ck = apriori_gen(l[k - 2], k) lk, sup_k = scan_dataset(dataset, ck, min_support_count) support_data.update(sup_k) if len(lk) == 0: break l.append(lk) k += 1 return l, support_data if __name__ == '__main__': data = load_dataset() L, support = apriori(data, min_support=0.4) for i, freq_set in enumerate(L): print(f"频繁 {i+1} 项集: {freq_set}")这段代码的逻辑说明：create_c1完成第一次扫描，统计单个商品频次。scan_dataset是核心扫描函数，对每个候选项集判断是否是某条交易的子集，如果是就计数。apriori_gen负责由频繁 k-1 项集生成候选 k 项集，这里用了一个简化剪枝：只合并前 k-2 项相同的两个项集。主循环从 k=2 开始，逐层生成频繁项集，直到某一层为空。
参数说明：min_support是最小支持度，取值范围 0 到 1。设为 0.4 表示一个项集至少要在 40% 的交易中出现才算频繁。这个值越低，产生的频繁项集越多，计算时间越长。实际业务中通常从 0.01 到 0.1 开始试，根据数据稀疏程度调整。
2.4 FP-Growth 的 Python 实现：从建树到递归挖掘
FP-Growth 的代码量比 Apriori 大，但核心结构清晰。下面是一个完整的单机实现，同样只用标准库。
# fpgrowth_basic.py # 最小化 FP-Growth 实现，展示 FP-Tree 构建与条件模式基递归 class FPNode: def __init__(self, name, count, parent): self.name = name self.count = count self.parent = parent self.children = {} self.node_link = None def increment(self, count): self.count += count def create_tree(dataset, min_support=1): """构建 FP-Tree，返回根节点、头指针表、频繁项集支持度""" header_table = {} for transaction in dataset: for item in transaction: header_table[item] = header_table.get(item, 0) + 1 # 去掉不满足最小支持度的项 header_table = {k: v for k, v in header_table.items() if v >= min_support} freq_item_set = set(header_table.keys()) if len(freq_item_set) == 0: return None, None, None for k in header_table: header_table[k] = [header_table[k], None] ret_tree = FPNode('Null Set', 1, None) for transaction in dataset: local_d = {} for item in transaction: if item in freq_item_set: local_d[item] = header_table[item][0] if len(local_d) > 0: ordered_items = [v[0] for v in sorted(local_d.items(), key=lambda p: p[1], reverse=True)] update_tree(ordered_items, ret_tree, header_table, 1) return ret_tree, header_table, freq_item_set def update_tree(items, in_tree, header_table, count): """将一条排序后的交易记录插入 FP-Tree""" if items[0] in in_tree.children: in_tree.children[items[0]].increment(count) else: in_tree.children[items[0]] = FPNode(items[0], count, in_tree) if header_table[items[0]][1] is None: header_table[items[0]][1] = in_tree.children[items[0]] else: update_header(header_table[items[0]][1], in_tree.children[items[0]]) if len(items) > 1: update_tree(items[1:], in_tree.children[items[0]], header_table, count) def update_header(node_to_test, target_node): """更新头指针表的链表""" while node_to_test.node_link is not None: node_to_test = node_to_test.node_link node_to_test.node_link = target_node def ascend_tree(leaf_node, prefix_path): """从叶子节点向上回溯，收集前缀路径""" if leaf_node.parent is not None: prefix_path.append(leaf_node.name) ascend_tree(leaf_node.parent, prefix_path) def find_prefix_path(base_pat, tree_node): """找到以 base_pat 为后缀的所有前缀路径""" cond_pats = {} while tree_node is not None: prefix_path = [] ascend_tree(tree_node, prefix_path) if len(prefix_path) > 1: cond_pats[frozenset(prefix_path[1:])] = tree_node.count tree_node = tree_node.node_link return cond_pats def mine_tree(in_tree, header_table, min_support, prefix, freq_item_list): """递归挖掘 FP-Tree""" big_l = [v[0] for v in sorted(header_table.items(), key=lambda p: p[1][0])] for base_pat in big_l: new_freq_set = prefix.copy() new_freq_set.add(base_pat) freq_item_list.append(new_freq_set) cond_patt_bases = find_prefix_path(base_pat, header_table[base_pat][1]) my_cond_tree, my_head, _ = create_tree(cond_patt_bases, min_support) if my_head is not None: mine_tree(my_cond_tree, my_head, min_support, new_freq_set, freq_item_list) if __name__ == '__main__': data = [ ['牛奶', '面包', '尿布'], ['可乐', '面包', '尿布', '啤酒'], ['牛奶', '尿布', '啤酒', '鸡蛋'], ['面包', '牛奶', '尿布', '啤酒'], ['面包', '牛奶', '尿布', '可乐'], ] min_sup = 2 tree, header, freq = create_tree(data, min_sup) freq_list = [] if header is not None: mine_tree(tree, header, min_sup, set(), freq_list) for itemset in freq_list: print(itemset)逻辑说明：create_tree完成两次扫描。第一次统计频次并过滤，第二次按频次降序将每条交易插入树中。update_tree是递归插入函数，如果当前项已在子节点中则增加计数，否则新建节点并更新头指针链表。find_prefix_path从某个节点的链表头开始，沿 node_link 遍历所有相同名称的节点，对每个节点向上回溯到根，收集前缀路径。mine_tree是递归挖掘入口，从频次最低的项开始，构建条件模式基，递归生成条件 FP-Tree。
参数说明：min_support这里用的是绝对计数，不是比例。因为示例数据集只有 5 条记录，设为 2 表示至少出现 2 次。在实际项目中，你通常先算min_support_count = min_support_ratio * len(dataset)，再传入。
2.5 两个算法在同一份数据上的输出对比
把上面两段代码跑在同一份 5 条交易的数据上，Apriori 设min_support=0.4，FP-Growth 设min_support=2，两者等价。输出结果应该完全一致：频繁一项集 5 个，频繁二项集若干，频繁三项集若干。如果结果不一致，先检查最小支持度的换算是否正确，再检查 FP-Growth 建树时是否漏掉了频次排序这一步。
| 对比维度 | Apriori | FP-Growth |
|---|---|---|
| 数据库扫描次数 | 每层一次，共 k 次 | 固定两次 |
| 候选集生成 | 需要，可能爆炸 | 不需要 |
| 内存占用 | 候选集大时高 | 树结构，通常更低 |
| 实现复杂度 | 较低 | 较高 |
| 适合数据规模 | 小到中等 | 中到大型 |
| 稀疏数据表现 | 候选集少时还行 | 树压缩效果一般 |
3. 对比测试怎么做：数据集、参数与评估指标
3.1 数据集从哪来、怎么构造
做对比测试，数据集的选择直接决定结论有没有说服力。我一般会准备三份数据：一份是公开的零售交易数据集，比如经典的超市购物篮数据；一份是自己用 Python 随机生成的稠密数据集，控制商品数量和每条交易的平均长度；一份是真实业务脱敏后的订单数据。
随机生成数据的代码很简单，但要注意控制稀疏度。下面这个脚本生成一份指定交易数、商品数、平均长度的数据集。
# generate_dataset.py import random def generate_transactions(num_transactions, num_items, avg_len): """生成随机交易数据集 num_transactions: 交易条数 num_items: 商品总数 avg_len: 每条交易平均商品数 """ items = [f'item_{i}' for i in range(num_items)] dataset = [] for _ in range(num_transactions): length = max(1, int(random.gauss(avg_len, avg_len * 0.3))) length = min(length, num_items) transaction = random.sample(items, length) dataset.append(transaction) return dataset if __name__ == '__main__': data = generate_transactions(10000, 200, 8) with open('transactions.txt', 'w', encoding='utf-8') as f: for t in data: f.write(','.join(t) + '\n') print(f'生成 {len(data)} 条交易，商品数 200，平均长度约 8')参数说明：num_transactions控制数据规模，从 1000 到 100000 都可以试。num_items是商品种类数，越大数据越稀疏。avg_len是每条交易的平均商品数，越大数据越稠密，频繁项集越多。我通常会用(1000, 50, 5)、(10000, 200, 8)、(50000, 500, 10)三组配置来观察算法在不同稀疏度下的表现。
3.2 最小支持度怎么选：从业务含义反推
最小支持度不是拍脑袋定的。它的业务含义是：一个商品组合至少在多少比例的交易中出现，才值得关注。设得太高，规则太少，可能漏掉有价值的弱关联；设得太低，规则爆炸，计算时间不可控。
我的经验是分两步走。第一步，先用一个较高的值跑一遍，比如 0.05 或 0.1，看看能产出多少规则，人工判断是否合理。第二步，逐步降低支持度，观察规则数量和运行时间的变化曲线。通常存在一个拐点，支持度降到某个值以下，规则数量急剧上升，运行时间也非线性增长。这个拐点就是你的业务能承受的下限。
对于 FP-Growth，因为速度快，可以比 Apriori 设得更低一些，挖掘出更多长尾规则。但要注意，支持度太低时，很多规则只是统计噪声，需要配合提升度（lift）和置信度（confidence）一起过滤。
3.3 评估指标：运行时间、内存峰值、规则数量、规则一致性
对比测试不能只看“谁快”。我一般会记录四个指标：
运行时间：用time.perf_counter()包住算法主逻辑，跑三次取平均值。注意要把数据加载时间排除在外，只计算算法本身耗时。
内存峰值：用tracemalloc或memory_profiler监控。Apriori 在候选集生成阶段内存增长最明显，FP-Growth 在建树阶段内存增长最明显。
规则数量：在相同最小支持度和最小置信度下，两个算法产出的频繁项集应该完全一致。如果数量不同，说明实现有问题。规则数量本身也是评估指标，反映算法在不同参数下的产出能力。
规则一致性：把两个算法产出的频繁项集分别转成排序后的元组集合，做集合比较。一致则通过，不一致则排查。
# benchmark.py import time import tracemalloc from apriori_basic import apriori, load_dataset from fpgrowth_basic import create_tree, mine_tree def run_apriori(dataset, min_support): tracemalloc.start() start = time.perf_counter() L, support = apriori(dataset, min_support) elapsed = time.perf_counter() - start current, peak = tracemalloc.get_traced_memory() tracemalloc.stop() return elapsed, peak / 1024 / 1024, L def run_fpgrowth(dataset, min_support_count): tracemalloc.start() start = time.perf_counter() tree, header, freq = create_tree(dataset, min_support_count) freq_list = [] if header is not None: mine_tree(tree, header, min_support_count, set(), freq_list) elapsed = time.perf_counter() - start current, peak = tracemalloc.get_traced_memory() tracemalloc.stop() return elapsed, peak / 1024 / 1024, freq_list if __name__ == '__main__': data = load_dataset() min_sup_ratio = 0.4 min_sup_count = int(min_sup_ratio * len(data)) t1, m1, r1 = run_apriori(data, min_sup_ratio) t2, m2, r2 = run_fpgrowth(data, min_sup_count) print(f'Apriori: 耗时 {t1:.4f}s, 内存峰值 {m1:.2f}MB, 频繁项集数 {sum(len(x) for x in r1)}') print(f'FP-Growth: 耗时 {t2:.4f}s, 内存峰值 {m2:.2f}MB, 频繁项集数 {len(r2)}')这段代码把两个算法的运行时间和内存峰值放在同一口径下对比。注意tracemalloc只追踪 Python 对象的内存分配，不包括底层 C 扩展或系统缓存，但对于纯 Python 实现足够用。
3.4 用 matplotlib 把对比结果画出来
数字表格不够直观，我习惯把不同支持度下的运行时间画成折线图。下面这段代码生成对比图。
# plot_compare.py import matplotlib.pyplot as plt from apriori_basic import apriori, load_dataset from fpgrowth_basic import create_tree, mine_tree import time def measure(dataset, min_support_ratio): min_support_count = int(min_support_ratio * len(dataset)) start = time.perf_counter() apriori(dataset, min_support_ratio) t_apriori = time.perf_counter() - start start = time.perf_counter() tree, header, freq = create_tree(dataset, min_support_count) freq_list = [] if header is not None: mine_tree(tree, header, min_support_count, set(), freq_list) t_fpgrowth = time.perf_counter() - start return t_apriori, t_fpgrowth if __name__ == '__main__': data = load_dataset() ratios = [0.2, 0.3, 0.4, 0.5, 0.6] apriori_times = [] fpgrowth_times = [] for r in ratios: ta, tf = measure(data, r) apriori_times.append(ta) fpgrowth_times.append(tf) plt.plot(ratios, apriori_times, marker='o', label='Apriori') plt.plot(ratios, fpgrowth_times, marker='s', label='FP-Growth') plt.xlabel('min_support') plt.ylabel('time (s)') plt.legend() plt.title('Apriori vs FP-Growth runtime') plt.savefig('compare.png') plt.show()参数说明：ratios列表控制横轴的最小支持度取值。在真实数据集上，这个范围要根据数据稀疏度调整。稠密数据可以从 0.05 到 0.3，稀疏数据从 0.01 到 0.1。marker参数让两条线更容易区分。
4. 避坑与排查：对比测试中最容易翻车的五个地方
4.1 最小支持度口径不一致导致结果对不上
现象：Apriori 和 FP-Growth 跑同一份数据，频繁项集数量差很多，甚至一项集都对不上。
原因：Apriori 代码里用的是比例，FP-Growth 代码里用的是绝对计数，两者没有做换算。或者一个用了>=，另一个用了>，边界条件不一致。
解决：统一口径。在调用两个算法之前，先算好min_support_count = min_support_ratio * len(dataset)，然后确认两个算法内部都使用同一个比较符号。我一般会在测试脚本里打印出实际使用的支持度计数，肉眼核对一遍。
4.2 FP-Growth 建树时忘记按频次降序排列
现象：FP-Growth 产出的频繁项集比 Apriori 少，或者某些明显频繁的组合没有出现。
原因：FP-Tree 的压缩效果依赖于按频次降序插入。如果顺序乱了，相同前缀的路径无法共享，树会变得又高又胖，条件模式基的挖掘也会出错。
解决：在create_tree函数里，对每条交易记录，先过滤掉非频繁项，然后按header_table[item][0]降序排列，再调用update_tree。这个排序步骤不能省。
4.3 递归深度过大导致栈溢出
现象：在较大的数据集上跑 FP-Growth，报RecursionError: maximum recursion depth exceeded。
原因：mine_tree是递归函数，当频繁项集很长时，递归深度可能超过 Python 默认的 1000 层限制。
解决：在脚本开头加import sys; sys.setrecursionlimit(10000)。但更根本的办法是检查数据，如果频繁项集长度超过 20，说明最小支持度设得太低，或者数据本身有问题。我一般会先限制最大项集长度，比如只挖掘到 5 项集，避免无意义的深层递归。
4.4 内存监控方式不对导致数据失真
现象：用tracemalloc监控 Apriori 内存，发现峰值只有几 MB，和实际感受不符。
原因：tracemalloc只追踪 Python 对象的内存分配。如果算法内部用了大量临时列表和字典，这些对象在 GC 回收后不会被计入峰值。另外，如果数据加载在tracemalloc.start()之前完成，那部分内存也不会被统计。
解决：把tracemalloc.start()放在数据加载之后、算法调用之前。对于更精确的内存监控，可以用memory_profiler的@profile装饰器，或者用psutil读取进程的 RSS 内存。我通常两个都用，tracemalloc看 Python 对象分配，psutil看进程整体内存。
4.5 规则评估只看支持度和置信度，忽略提升度
现象：挖出一堆规则，比如“买尿布的人 80% 会买啤酒”，但啤酒本身在数据集里的出现率就有 75%，这个规则几乎没有价值。
原因：置信度高不代表关联强。如果后件本身就很频繁，前件对后件的提升作用很小。
解决：在规则生成阶段加入提升度过滤。提升度 = 置信度 / 后件支持度。提升度大于 1 才说明前件对后件有正向促进作用。我一般会设提升度阈值 1.2 以上，再结合业务判断。Apriori 和 FP-Growth 都只负责挖频繁项集，规则生成和过滤需要额外写代码，这部分不能省。
5. 从对比测试到落地：参数调优与增量挖掘的实用技巧
跑通对比测试只是第一步。真正要把关联规则用到业务里，还有两件事要做：参数调优和增量更新。
参数调优的核心是找到支持度、置信度、提升度三者的平衡点。我的习惯是先用 FP-Growth 快速扫一遍低支持度，看看规则数量的分布，然后画一张“支持度-规则数”曲线，找到拐点。拐点之前，规则数随支持度降低缓慢增长；拐点之后，规则数爆炸。这个拐点对应的支持度就是你的起点。然后在这个起点附近，用 Apriori 和 FP-Growth 各跑一遍，确认两者结果一致，再根据运行时间决定线上用哪个。如果数据量在十万条以内，Apriori 优化一下候选集生成也能用；超过十万条，直接上 FP-Growth，别犹豫。
增量挖掘是另一个实战中绕不开的问题。业务数据每天新增，你不可能每天全量重跑。FP-Growth 在这方面有天然优势：FP-Tree 可以增量更新。新来一批交易，按频次降序插入现有树中，更新头指针链表和节点计数，然后只对受影响的子树重新挖掘。Apriori 做增量就麻烦得多，因为候选集生成依赖全局频繁项集，新增数据可能改变所有层的频繁性判断。
下面是一个简化的 FP-Tree 增量更新示例，展示核心思路。
# incremental_fpgrowth.py from fpgrowth_basic import create_tree, mine_tree, update_tree def incremental_update(existing_tree, existing_header, new_transactions, min_support_count): """在已有 FP-Tree 上增量插入新交易，并重新挖掘""" # 先统计新数据中的项频次，合并到现有头指针表 for transaction in new_transactions: for item in transaction: if item in existing_header: existing_header[item][0] += 1 else: existing_header[item] = [1, None] # 过滤掉不满足最小支持度的项 freq_items = {k: v for k, v in existing_header.items() if v[0] >= min_support_count} for transaction in new_transactions: local_d = {} for item in transaction: if item in freq_items: local_d[item] = freq_items[item][0] if len(local_d) > 0: ordered = [v[0] for v in sorted(local_d.items(), key=lambda p: p[1], reverse=True)] update_tree(ordered, existing_tree, existing_header, 1) # 重新挖掘 freq_list = [] mine_tree(existing_tree, existing_header, min_support_count, set(), freq_list) return freq_list if __name__ == '__main__': data = [ ['牛奶', '面包', '尿布'], ['可乐', '面包', '尿布', '啤酒'], ['牛奶', '尿布', '啤酒', '鸡蛋'], ] min_sup = 2 tree, header, freq = create_tree(data, min_sup) new_data = [ ['面包', '牛奶', '尿布', '啤酒'], ['面包', '牛奶', '尿布', '可乐'], ] result = incremental_update(tree, header, new_data, min_sup) for itemset in result: print(itemset)这段代码的逻辑：先更新头指针表中的频次计数，然后对新交易按频次降序插入现有树，最后重新挖掘。注意，增量更新后，原来不频繁的项可能因为新数据而变得频繁，所以头指针表要保留所有项，只在插入时过滤。重新挖掘时，mine_tree会遍历整个头指针表，确保新频繁项不被遗漏。
参数说明：min_support_count在增量场景下通常保持不变，但如果你发现新数据导致规则数量激增，可以适当调高。new_transactions是新增的交易列表，格式和原始数据一致。
最后说一个我踩过的坑：增量更新后一定要做一次全量校验。我遇到过因为头指针表链表断裂导致部分节点丢失的情况，增量挖掘结果和全量重跑对不上。后来我养成了一个习惯，每次增量更新后，随机抽 100 条交易，用全量算法跑一遍，对比频繁项集是否一致。这个校验步骤花不了多少时间，但能避免线上规则悄悄出错。希望帮到你。
本文还有配套的精品资源，点击获取
