Apriori算法手写实现：从频繁项集到关联规则挖掘实战

1. 别被名字吓到，Apriori算法其实就干了三件事

最近带团队做商品关联分析，好几个刚接触数据挖掘的同学看到“Apriori”这个名字就发怵，总觉得背后有什么高深莫测的推导公式。实际用代码实现一遍你会发现，这个算法比你想象中朴素得多。它解决的是一个非常具体的业务问题：在交易数据中，哪些商品经常同时被购买，以及能不能从“购买了A”推导出“大概率会买B”。

说白了，Apriori算法的全部核心就三条：找频繁项集、算置信度、评估提升度。所谓频繁项集，就是“一起出现次数足够多”的商品组合，比如“啤酒+尿布”同时出现的概率超过了你设定的阈值；置信度是“买了A的人里面有多少人买了B”；提升度则回答“A对B的促进效果是正向还是负向”这个问题。

我当年第一次啃论文时被各种符号绕得七荤八素，真正上手写代码才明白，这算法压根不需要高等数学，它依赖的只是计数和集合运算。这篇文章我不打算贴一堆公式，而是直接带着你从零手写一份Python实现，每一步都拆开讲清楚。

适合谁来读？两类人：一类是刚接触关联规则分析、想搞懂底层的初学者；另一类是工作中要跑频繁项集挖掘，但不想只调包、想理解包背后逻辑的工程师。读完之后你能自己写一个完整可用的Apriori实现，还能针对不同数据规模做优化。

2. 三个核心指标和一个关键性质：不搞清楚这些，代码就是抄了个寂寞

2.1 支持度、置信度、提升度到底在算什么

先说支持度（Support）。假设我们有五笔订单，其中“牛奶→面包”同时出现在三笔订单里，那它的支持度就是3/5=60%。这个指标回答的问题是“这个组合整体出现的概率有多大”，它的作用是把那些稀稀拉拉、偶然一次出现的组合过滤掉。你设一个最小支持度阈值，比如5%，那出现频率低于这个线的组合直接不考虑。

再看置信度（Confidence）。这个是用来衡量条件概率的：在买了牛奶的订单中，有多少比例同时也买了面包。假设买牛奶的订单有4笔，其中3笔也买了面包，那“牛奶→面包”的置信度就是3/4=75%。注意这里箭头是有方向的，“牛奶→面包”和“面包→牛奶”是两个不同的规则，置信度也可能完全不同。

最后是提升度（Lift）。这个指标很多人会忽视，但它恰恰是最能说明“规则是否有实际价值”的。计算方法是：置信度除以“面包在所有订单中的出现频率”。为什么要这么做？因为有些商品本身就是畅销品，就算没有牛奶的带动，它也经常被购买。如果“牛奶→面包”的置信度是75%，但面包在所有订单中本身的出现概率就有70%，那牛奶对面包的拉动效应其实很弱。提升度大于1说明A对B有正向促进作用，等于1说明两者独立，小于1说明A反而抑制B的出现。

2.2 Apriori性质：这个算法效率高的根基

理解了三个指标，还得明白一个性质——先验性质（Apriori property），也叫反单调性。一句话总结：如果一个项集是频繁的，那它的所有子集一定也是频繁的；反过来，如果一个项集是非频繁的，那它的所有超集一定是非频繁的。

举个例子，假设“牛奶、面包、鸡蛋”这个三项组合的支持度只有2%，小于你设定的5%阈值，它是非频繁项集。那你可以直接断定：“牛奶、面包、鸡蛋、黄油”这个四项组合的支持度一定也小于5%，连算都不用算。为什么？因为一项集的子集包含在更小的组合里，多一个条件只会让同时出现的概率更低。

这个性质是Apriori算法剪枝策略的理论基础。实际计算的时候，它帮我们砍掉了大量永远不可能达标的候选集。如果没有这个性质做剪枝，数据稍微大一点，所有组合的排列组合数量就会爆炸，算法根本跑不动。

2.3 我自己理解的逐层递进逻辑

算法的总体流程就是一层一层往上走的“广度优先搜索”。先找出所有出现频率达标的单个商品（1项频繁集），然后基于它们拼出所有可能的2项组合，过滤掉不达标的；再用剩下的2项频繁集拼出3项组合，继续过滤。每次往上一层，组合里的元素数量加一，直到发现频繁集为空或者没有更多组合可以拼为止。

这个过程带着明显的递进感。第k层的候选集，全部由第k-1层的频繁集拼接而成，而不是最初就穷举所有可能。以3项集为例，假设2项频繁集是{牛奶面包}、{牛奶鸡蛋}、{面包鸡蛋}、{面包黄油}这四个，拼接规则是找出前k-1个元素相同的两个集合作连接，比如{牛奶面包}和{牛奶鸡蛋}前1个元素都是“牛奶”，拼起来就得到{牛奶面包鸡蛋}。这样既避免了生成冗余组合，又保证了不漏掉任何潜在频繁集。

3. 代码实现前的准备：工具选型和数据组织

3.1 为什么我选pandas而不是纯Python列表

写Apriori实现，常见的做法有纯Python实现、用pandas处理数据、以及直接用现成的库比如mlxtend。我的建议是：学习阶段不要用mlxtend，那相当于看答案做题。但也不必刻意追求纯Python，因为你处理真实明细数据的时候，读Excel、CSV、数据库导出的表格，pandas能省掉大量数据清洗的功夫。

我在实际项目中通常是这样分工的：pandas负责数据加载、格式转换、统计支持度计数；核心的候选集生成和剪枝逻辑用原生Python写。这样既保留了算法的透明性，代码又不至于在集合操作上写得啰嗦。pandas处理行级统计的时候确实高效，接口也简单。

数据组织上要特别强调一点：事务数据必须整理成“事务ID+商品”的明细长表，或者直接转换成就地列表。比如标准格式就是这样一组数据：

这三列示例，左边是事务编号，右边是商品列表。真实场景下，一次订单可能涉及几十个商品，但每个事务对应的商品组合长度可以不一样。我们实现的时候，会把所有商品列表拼在一起做统一去重，生成商品全集；同时保留每个事务的商品列表用于逐单统计。

3.2 数据清洗的常见坑

很多人在预处理阶段就翻车，最常见的是这三类问题：商品字段里有空格或大小写不一致，导致同一商品被当成两个项；极少数订单商品数量特别多，导致这些订单里几乎没有非频繁项，扭曲了统计结果；以及商品编号是数值型，但部分商品编码带着前导零或特殊字符，需要统一转成字符串。

我自己的处理习惯是：所有商品名先做strip去空格，再用小写化统一，最后转成字符串类型。对于极端长订单，通常会设置一个长度上限，超过某个阈值就把订单截断或者移除。这不一定对所有业务都合理，但至少能防止少数异常订单对结果的干扰。如果你业务上必须保留长订单，那就不做截断，但要意识到它对结果的影响是存在的。

4. 完整代码逐块拆解：手写一个Apriori全流程

4.1 主框架设计：我希望一份代码能直接落地用

直接先贴总控代码，然后逐块讲里面的关键逻辑。整个实现分四个方法：load_data负责载入和预处理；generate_candidates负责从上一层频繁集生成候选集；calculate_support负责统计候选集的支持度计数并筛选；generate_rules负责从频繁项集生成关联规则并计算置信度和提升度。

import pandas as pd from itertools import combinations class AprioriMiner: def __init__(self, min_support=0.05, min_confidence=0.5, min_lift=1.0): self.min_support = min_support self.min_confidence = min_confidence self.min_lift = min_lift self.frequent_itemsets = {} # 键为项集的frozenset，值为支持度 self.rules = [] def load_data(self, orders): self.transactions = [sorted(set(order)) for order in orders] self.total_transactions = len(self.transactions) self.all_items = set() for t in self.transactions: self.all_items.update(t) return self.transactions

这里的核心点是frozenset。为什么不用普通set作为项集？因为set是可变对象，不能作为字典的键。而频繁项集在后续操作中有大量需要“查表”的场景，用字典按项集直接取支持度是最高效的。frozenset在保持内部元素不可变的同时，可以作为字典键使用，配合它天生的无序特性，刚好匹配“项集本身没有顺序”的数学定义。

4.2 候选集生成：连接步和剪枝步一个不能少

候选集生成是Apriori里最容易写错的地方。很多人图省事直接用combinations枚举所有可能组合，数据量一大就彻底跑不动。真正地按Apriori思路做，需要用到“自连接”加“剪枝”两部曲。

def _generate_candidates(self, freq_items, k): candidates = set() freq_list = list(freq_items) n = len(freq_list) for i in range(n): for j in range(i+1, n): items_i = list(freq_list[i]) items_j = list(freq_list[j]) items_i.sort() items_j.sort() if items_i[:k-2] == items_j[:k-2]: new_candidate = frozenset(items_i + [items_j[-1]]) if self._is_valid_candidate(new_candidate, freq_items): candidates.add(new_candidate) return candidates

连接步的规则很关键：两个k-1项集能拼接成一个k项候选集的前提是，它们的前k-2个元素完全相同，只有最后一个元素不同。这里要注意“前k-2个元素”指的是按固定顺序排列后的前k-2个元素，所以每次拼接前都要排序。这是代码里最容易被忽略的细节——两个集合本身无序，但为了做前缀比较，必须约定一个统一的排序规则。

剪枝步对应_is_valid_candidate这个辅助方法：

def _is_valid_candidate(self, candidate, freq_items): for subset in combinations(candidate, len(candidate)-1): if frozenset(subset) not in freq_items: return False return True

这段代码的作用是检查：对于一个候选k项集，它的所有k-1子集是否都已经在频繁集里。根据前面提到的先验性质，只要有一个k-1子集是不频繁的，这个候选集就可以直接丢弃。这个剪枝可以把那些“虽然有连接可能，但必然不达标”的组合提前拦截掉，避免后续做代价更高的全量计数。

4.3 支持度计算：别用蠢办法遍历所有事务

候选集生成之后，下一步是统计每个候选集在全部事务中出现的次数，这一步是整个算法的心跳。我见过很多人写成三层for循环：遍历所有候选集，对每个候选集再遍历所有事务，对每笔事务再遍历所有商品。数据量几千条时还能跑，但真实业务往往几十万条交易记录，这种写法会慢到怀疑人生。

def _calculate_support(self, candidates): support_count = {c: 0 for c in candidates} for transaction in self.transactions: trans_set = set(transaction) for candidate in candidates: if candidate.issubset(trans_set): support_count[candidate] += 1 frequent = {} for candidate, count in support_count.items(): support = count / self.total_transactions if support >= self.min_support: frequent[candidate] = support return frequent

注意这里用了一个非常巧妙的小优化：候选集如果连某个事务都排不上，就根本不去遍历它的内部元素。用issubset直接判断候选集是否是当前事务的子集，这个判断是内建方法，运算效率远高于手动遍历。你还可以用倒排索引做进一步优化：先统计每个商品出现在哪些事务里，然后候选集的支持度计数等于它的所有商品对应事务列表的交集大小。这种TID列表（事务ID列表）方法在数据量大的时候提升非常明显，后面我会专门讲。

4.4 主循环：逐层往上挖，直到挖不动为止

现在把前面两个块拼到一起，形成完整的迭代主流程：

def fit(self, orders): self.load_data(orders) single_freq = {} for item in self.all_items: count = sum(1 for t in self.transactions if item in t) support = count / self.total_transactions if support >= self.min_support: single_freq[frozenset([item])] = support self.frequent_itemsets[1] = single_freq k = 2 while True: candidates = self._generate_candidates(self.frequent_itemsets[k-1], k) if not candidates: break freq_k = self._calculate_support(candidates) self.frequent_itemsets[k] = freq_k if not freq_k: break k += 1 return self

这几个条件判断为什么都得有？if not candidates判断的是候选集生成阶段就没有任何可能组合了，说明上一层频繁集数量太少，拼不出更长的项集了。if not freq_k判断的是虽然生成了候选集，但全部不达标，说明即便有组合也都被阈值过滤掉了，算法到此自然收敛。这两个条件缺一不可，我调试的时候踩过这个坑——漏掉后面那个条件，循环会进入死循环，因为频繁集为空后，下一轮生成的候选集会变成空集，然后又开始乘循环。

4.5 关联规则生成：置信度和提升度的最终计算

频繁项集挖出来之后，工作只完成了一半。真正有业务价值的关联规则还需要筛选。生成规则的固定套路是：对每个频繁项集，枚举它的所有非空子集作为“条件部分”，剩余部分作为“结果部分”。

比如频繁项集是{牛奶、面包、鸡蛋}，可以生成{牛奶}→{面包、鸡蛋}、{面包}→{牛奶、鸡蛋}、{牛奶、面包}→{鸡蛋}等规则。候选规则生成之后计算置信度：规则“X→Y”的置信度等于项集X∪Y的支持度除以X的支持度。如果置信度不达标就舍弃；达标的再算提升度，判断规则的实际方向。

def generate_rules(self): self.rules = [] for k in range(2, len(self.frequent_itemsets)+1): for itemset in self.frequent_itemsets[k]: items = list(itemset) for m in range(1, k): for antecedent in combinations(items, m): antecedent_set = frozenset(antecedent) consequent_set = itemset - antecedent_set support_union = self.frequent_itemsets[k][itemset] support_ante = self.frequent_itemsets[len(antecedent_set)][antecedent_set] confidence = support_union / support_ante if confidence < self.min_confidence: continue support_consequent = self.frequent_itemsets[len(consequent_set)][consequent_set] lift = confidence / support_consequent if lift >= self.min_lift: self.rules.append({ 'antecedent': antecedent_set, 'consequent': consequent_set, 'support': support_union, 'confidence': confidence, 'lift': lift }) return self.rules

这段代码里有个隐藏的索引陷阱：第4行循环时，len(antecedent_set)对应的频繁集必须在self.frequent_itemsets里真实存在。换言之，如果某个组合压根不频繁，它不可能出现在频繁项集字典里。但因为我们是基于频繁项集本身来拆规则的，条件部分必然是某个频繁项集的子集，按性质它本身也可能是频繁的。不过有一种边界情况：当频繁项集中没有单独的2项集，只有3项集时（比如min_support设得很低），3项集的条件部分是1项集所以没问题，但2项集并不一定在频繁集中。所以当从3项集拆出2项条件时，self.frequent_itemsets[2]可能不存在，这就会抛KeyError。处理办法是在查表时做个存在性判断：

def _safe_support(self, itemset): k = len(itemset) if k not in self.frequent_itemsets: return 0.0 return self.frequent_itemsets[k].get(itemset, 0.0)

这个小函数我在所有需要查支持度的地方都用了，省了很多次调试崩溃。

4.6 演示数据实测

为了让整个流程可视，我构造了一份小型购物篮数据来验证代码正确性：

设定最小支持度0.3、最小置信度0.6、最小提升度1.0，运行之后的一部分输出如下：

频繁1项集：{牛奶}支持度0.83，{面包}支持度0.83，{鸡蛋}支持度0.67，{黄油}支持度0.50 频繁2项集：{牛奶,面包}支持度0.67，{牛奶,鸡蛋}支持度0.50，{面包,鸡蛋}支持度0.50，{面包,黄油}支持度0.50 频繁3项集：{牛奶,面包,鸡蛋}支持度0.33，{面包,鸡蛋,黄油}支持度0.33 规则示例： {黄油}→{面包} 置信度1.00 提升度1.20 {鸡蛋}→{牛奶} 置信度0.75 提升度0.90 <- 提升度小于1，存在抑制作用

注意最后那条规则的提升度小于1，按我们设定的min_lift=1.0阈值，它被过滤掉了。这个例子恰好说明提升度的重要意义：鸡蛋虽然经常和牛奶一起出现，但牛奶本身出现频率太高，鸡蛋对它没有净正向拉动效应。如果不加提升度检查，这条规则会被当成有效规则输出，业务上可能产生误导。

5. 实操调参记录与输出格式规范化

5.1 三个阈值怎么配合着调

很多初学者把min_support、min_confidence、min_lift当三个独立参数随手设，这是最常见的错误。实际上这三个参数是“漏斗”关系，前一个参数决定进入下一个环节的数据量。min_support越高，候选集越少，规则数量急剧下降；min_confidence则过滤掉低置信度的规则；min_lift过滤掉无关或负相关的规则。

我自己的经验是：先设定一个较大的min_support跑一遍，看生产了多少规则，再逐步降低。如果规则太多，优先调高min_confidence而不是min_support，因为业务上更关心的是“条件发生后结果发生的概率”，而不是“组合有多普遍”。反过来，如果规则太少，先降低min_support，而不是降低置信度，否则容易得到一堆高置信度但毫无代表性的规则。

还有一个实操点：阈值设置要考虑业务场景。比如我们分析的是零售购物篮，每天交易量可能有几万笔，那0.01的支持度可能就代表了几十上百笔订单，是有业务意义的。但如果分析的是一款SaaS产品的功能使用关联，总用户只有几百个，0.01的支持度就意味着不到几个人，统计分析毫无意义。一定要结合数据总量思考，不要盲目照搬别人的参数。

5.2 输出的结果要能直接给业务看

算法输出的原始规则长这样：{牛奶,面包}→{鸡蛋}，lift=1.32。这种格式给开发同事看没问题，给业务同学看就未必直观。我通常会在输出时做一层封装，把frozenset转成可读的字符串，再附加每个商品在整体订单中的出现频次，帮助业务方做综合判断。

def format_rules(self): results = [] for rule in self.rules: antecedent = ','.join(sorted(rule['antecedent'])) consequent = ','.join(sorted(rule['consequent'])) results.append(f"{antecedent} => {consequent} | " f"支持度={rule['support']:.2%} | " f"置信度={rule['confidence']:.2%} | " f"提升度={rule['lift']:.2f}") return results

输出格式看起来就像：

牛奶,面包 => 鸡蛋 | 支持度=33.33% | 置信度=50.00% | 提升度=1.20

这种格式放到报表里，业务方一看就懂。而且排序很重要，通常按提升度从高到低排，这样最有业务价值的规则最早被看到。我一般还会在结果里附带一列“规则涉及的商品组合基数”，用于粗筛掉那些组合过于庞大、难以落地的规则。

6. 性能优化：从理论能跑到能扛大数据的三种手段

6.1 瓶颈在哪，其实早就决定了

Apriori的性能瓶颈主要有两个：一是逐层生成候选集，每一层都要反复扫描整个事务数据库；二是当频繁项集数量很多时，候选集数量膨胀得异常恐怖。最小支持度越低，这个问题越严重。所以如果你只有几百MB或者几个GB的数据，直接跑原版Apriori问题不大；但如果事务数达到千万级别、商品种类上万，原版算法会慢到无法接受。

针对这个问题，有几个常见优化手段。事务压缩：如果一个事务不包含任何频繁k项集，那它也不可能包含频繁(k+1)项集，可以直接从后续扫描中移除。这个优化利用的还是Apriori性质，实现方式很简单，每次算完支持度后，标记哪些事务参与过频繁集匹配，下一轮扫描时只遍历这些事务。

哈希剪枝：对候选k项集做哈希，如果某个桶中所有项集出现次数加起来都低于最小支持度，那这个桶里的候选集直接删除。这种手段适合候选集特别庞大的场景，但实现复杂一点，需要设计合适的哈希函数。我通常是在数据量特别大的情况下才考虑。

6.2 用TID列表替代全量扫描

前面提过的TID列表是目前最实用的优化。与其反复扫描事务数据库，不如先做一次预处理，求每个商品对应的“事务ID列表”。然后一个候选集的支持度计数，就是它包含的每个商品的TID列表求交集后的大小。

比如商品“牛奶”出现在T1、T2、T4、T5、T6，“面包”出现在T1、T2、T3、T5、T6，那{牛奶,面包}的支持度就是这两个集合的交集，趟过一遍发现同时出现的交易是T1、T2、T5、T6，支持度计数为4。这样一次扫描就能得到候选集支持度，而且交集运算可以直接用Python的set。实际测试下来，当事务数超过十万、候选集超过千级别时，TID列表方法通常比遍历所有事务的方法快一个数量级。

def _build_tid_lists(self): self.tid_lists = {} for tid, transaction in enumerate(self.transactions): for item in transaction: self.tid_lists.setdefault(item, set()).add(tid) def _support_by_tid(self, itemset): if not itemset: return 0 result = None for item in itemset: tids = self.tid_lists.get(item, set()) result = tids if result is None else result & tids if not result: return 0 return len(result)

配合这个优化，_calculate_support可以改成对每个候选集用_support_by_tid，不再需要遍历事务。对我来说，TID列表实现简单、效果明显，是性价比最高的优化手段。

6.3 稀疏数据场景的选择

还有一个很容易被忽略的点：Apriori天生适合稀疏数据。如果一份数据平均每个事务包含的商品数量很多，那频繁项集的数量会爆炸式增长，Apriori家族算法会非常吃力。比如有几万种商品、平均每单包含上百项，用FP-Growth这类基于前缀树的方法可能更合适。反过来，如果平均每个事务只有3到10项，那Apriori配合TID列表的优化完全够用。

判断数据是否稀疏，有个经验指标：每个事务的平均项数除以商品种类总数，如果这个比值很低（比如小于千分之一），就是典型的稀疏数据。购物篮数据基本都是这种，而基因表达数据或文本关键词矩阵就不一定了。

7. 实战中经常翻车的四个问题

7.1 数据集空值导致候选集出现None

原始数据里商品列可能包含空值，清洗不彻底时，None会被当成一个正常的商品项参与计算。更隐蔽的是，一个事务可能因为空值被过滤掉，导致支持度分母不同。我遇到过几次调试了半天发现频繁项集里居然有一个None的情况，最后排查出来是数据源里某一行格式不对。

处理方式：数据加载后务必检查每个事务列表的有效性，过滤掉空值和非字符串类型，同时保证每笔事务至少保留一个有效商品。如果一笔事务全部都是空值，直接丢弃，但要在日志里记录丢弃数量，防止数据源出了大问题自己却不知道。

7.2 支持度计算用错了分母

支持度的分母是“总事务数”，但很多人写代码时不小心用了“包含条件商品的事务数”。比如算{牛奶,面包}的支持度时，有人会用“包含牛奶和面包的订单数”除以“包含牛奶的订单数”，这实际上是置信度而不是支持度。这种情况新手经常出现，因为思考时很容易从“买了牛奶的人里同时买了面包的比例”这个角度理解，写着写着就把分母写错了。

每次算支持度时，建议在代码里显式写出分母的数值来源：count / self.total_transactions，不要图省事复用变量。

7.3 频繁项集字典层级缺失导致KeyError

最典型的场景就是我在4.5节提到的：频繁项集中只有1项集和3项集，没有2项集。为什么会出现这种情况？因为最小支持度设得比较高，2项组合全被过滤掉了，但3项组合因为某种巧合支持度足够高（这在业务数据里不常见但在测试数据里容易发生）。生成规则时从3项集拆2项条件，查self.frequent_itemsets[2]就会报错。

这个问题的根源是“频繁项集的所有子集都是频繁的”这个性质。按道理3项集频繁，它的2项子集也应该频繁，所以正常情况下这种情况不该发生。但如果你自定义了频繁项集判断逻辑（比如用特殊加权支持度），或者数据本身有异常，就可能破坏这条性质。处理方式就是加_safe_support做保护。

7.4 事务太长导致候选集爆炸

有些订单会一次性购买几十上百个商品，比如仓库补货单。这种数据如果不做处理，频繁项集会非常密集，而且3项以上组合爆炸式增长。我在实际业务中遇到过一笔订单包含40多个商品的情况，即便最小支持度设到2%，3项候选集的数量还是突破了十几万。

处理方法分情况：如果这种长订单占比很小（比如不到1%），直接截断——把每个事务的商品列表限制在频繁程度最高的前20个商品以内；如果长订单是主流，那就说明这个业务场景本身不适合用Apriori，建议换FP-Growth或者先做一层商品类目降维，比如把几百个SKU映射到几十个类目再分析。

8. 一个提高效率的小技巧：提前算好单品支持度并剪枝

不管用原版Apriori还是优化版，每一步生成候选集都耗时，尤其是数据量大、频繁项集多的时候。有个经常被忽略但极其有效的小优化：在进入主循环之前，先把所有单品的支持度算好，并且直接过滤掉低于最小支持度的单品，后续所有操作只针对这些“高潜力单品”。

为什么这个优化有效？想想看，如果某种商品本身出现频率就不到5%，那包含它的任何组合，其支持度都不可能超过5%。所以它在任何一层都不可能是频繁项集的一部分。提前把它剔除，至少可以把候选集数量降低一个数量级。

def fit_with_pruning(self, orders): self.load_data(orders) item_count = {} for t in self.transactions: for item in t: item_count[item] = item_count.get(item, 0) + 1 self.frequent_1_items = set() self.frequent_1_map = {} for item, count in item_count.items(): sup = count / self.total_transactions if sup >= self.min_support: self.frequent_1_items.add(frozenset([item])) self.frequent_1_map[frozenset([item])] = sup self.frequent_itemsets[1] = self.frequent_1_map # 后续循环中生成候选集时只使用self.frequent_1_items中的商品

这样一来，_generate_candidates里所有拼接、组合操作都限制在已经验证过的高频单品集合内，不会出现一个低频单品混进候选集然后被支持度计数否定又被剪枝剪掉的重复劳动。

9. 我个人使用过程中的总结体会

Apriori这算法名声很大，但真正动手写一遍之后你会发现，它本质上就是一个带有剪枝策略的穷举法。它最大的优点是逻辑简单、结果可解释，能清楚地说出“为什么这条规则成立”，因为中间每一步都有明确的计数支撑。业务方问起来，你可以直接导出中间数据给他看：支持度是多少，置信度是多少，提升度是多少。这是很多黑盒机器学习模型做不到的事情。

不过它也确实是硬伤明显的算法。最小支持度设高了会漏掉低频高值的关联规则，设低了候选集数量爆炸，计算时间成倍增加。所以我现在的习惯是：数据量小、业务要求高解释性的场景，优先Apriori；数据集上了几十G，或者事务平均长度明显偏长，直接上并行化的FP-Growth或者用Spark MLlib里的FPGrowth实现。工具没有绝对的优劣，看场景选择罢了。

最后分享一个小技巧：如果你用这份代码分析真实业务数据，建完模后千万别直接拿全部规则上线。先做一轮人工抽查，至少看20条规则，确认里面没有明显的误导项。有时候支持度、置信度、提升度三项指标全部达标，规则在业务上仍然毫无意义，比如“A→A”这类恒真规则，或者两个高人气商品之间的同时出现只是因为节假日促销。指标只能告诉你统计关联，不能告诉你因果，做人肉过滤是最后一道关口。

这几次项目做完，我的复盘结论就一句话：Apriori教会我的不是算法本身，而是一个思路——“先剪枝，再穷举，最后用业务直觉兜底”。这套思路放到任何数据挖掘任务里，都同样适用。