频繁模式挖掘与关联规则：数据仓库大作业的Python实现与实战

简介：面向数据仓库与数据挖掘课程期末大作业的完整项目资源，采用Python实现频繁模式挖掘。方案基于Apriori算法，从多角度、多篮子粒度进行关联规则挖掘，在Gutenberg与DBLP数据集上设计了多个应用任务，包括作者活跃度分析、团体挖掘和主题关联等，场景覆盖较广。代码模块化清晰，包含8个Python源文件，并附有详细注释，搭配PDF报告与Markdown说明文档，即使新手也能快速理解核心逻辑，简单部署即可运行。压缩包共41个文件，约5.84MB，另有txt数据文件、结果图片、pyc缓存等，目录结构合理，便于按需查阅。项目已有817人学习，适合作为课程设计、期末大作业的高分参考，可帮助读者掌握频繁模式挖掘从理论到落地的完整实现流程。

1. 频繁模式挖掘大作业，数据仓库课程最需要的是一整条交付链

数据仓库和数据挖掘课程里，频繁模式挖掘几乎是每届都有的经典选题。可真正让大作业拉开分差的，往往不是 Apriori 或 FP-Growth 的算法推导，而是源代码和报告之间有没有一条完整交付链。只准备一个能运行的挖掘脚本，演示时点两下输出几行频繁项集，老师追问数据怎么从仓库取出来、支持度为什么这么设，现场就容易冷场。我一般会按题目中的“源代码+文档说明+报告pdf”三件套反推工作顺序：先确定事务数据格式，再写模块化代码，最后用脚本生成带执行结果的报告。下面按这个顺序把数据仓库事实表到事务集、Apriori 源码结构、规则评估、PDF 报告制作串一遍，并给出可直接改用的 Python 代码。适合正在做数据仓库与数据挖掘课程设计的人，也适合想在下一次购物篮分析工程中复用这套代码的人。

2. 频繁模式挖掘的算法选型与 Python 源码结构

2.1 Apriori 和 FP-Growth，在什么数据量下选谁更稳妥

频繁模式挖掘的经典实现有两个：Apriori 和 FP-Growth。Apriori 的核心是一个递推性质：一个 k 项集如果不频繁，它的所有超集也一定不频繁，所以每一轮都只需在上轮频繁 k‑1 项集上生成候选，再扫描事务库统计支持度。FP-Growth 的思路则是把事务库压缩成一颗 FP 树，在树上递归处理条件模式基，省掉层层的全库扫描。两个算法在数据仓库大作业里都能成立，区别主要在数据量和你打算在报告里展开的技术深度。

教学用数据仓库导出的订单表，事务量通常从几千到几十万行。几千行时，两个算法的耗时差距是秒级甚至更小，Apriori 的实现更短、调参更直观；事务量真正到几十万行时，FP-Growth 对重复项的压缩效果更明显，但纯 Python 手写树的排错成本也更高。我一般先用一个小支持度试探跑一次 Apriori，如果耗时超过五秒，再考虑切 FP-Growth。两个算法的差异可以直接做进报告文档，下面是常用对比表。

从答辩角度说，Apriori 每一轮的候选生成、计数、剪枝都能对应到一张过程表，FP-Growth 的树结构则要画得足够准确才能讲明白。再考虑作业查重，我通常建议源代码以 Apriori 为主线，把 FP-Growth 的树构建思路作为扩展章节写进文档说明。等以后真遇到大规模订单数据，再按 FP-Growth 重构不迟。

2.2 把频繁模式挖掘源代码拆成加载器、挖掘器、规则器

拿到一个现成的 Apriori 例程后，第一件事不是调参，而是把源码拆成 loader、miner、rules 三个文件。原因有两个：老师检查源代码时能一眼看出数据的进出口在哪里；另一个原因是，很多免费 python 源码大全里下载的例程，几十行全堆在一个 main.py 里，查重时整个文件都会被判定为网络已有实现。拆开之后，每个文件都加入针对本次数据仓库结构的自定义逻辑，命中率反而下降。推荐的目录结构如下。

assignment_python_frequent/ ├── data/ │ ├── transactions.csv # 整理后的事务数据 │ └── warehouse_query.sql # 从数据仓库取数的 SQL ├── loader.py # 读取数据并得到 List[Set[str]] ├── miner.py # Apriori 算法主体 ├── rules.py # 关联规则生成与评估 ├── main.py # 命令行入口 ├── output/ # 挖掘结果和图表 ├── report.md # 报告源文件 └── README.md # 文档说明

文件名里越能体现数据仓库上下文越好，warehouse_query.sql 的存在就是大作业与实际数据结合的直接证明。loader 只负责把 CSV、数据库结果转成内存中的事务列表；miner 只接收事务列表并返回频繁项集字典；rules 再把频繁项集推导成关联规则。main.py 只做三件事：解析参数、组装流程、写输出文件。后面排错时，只要 loader 输出的数据格式不变，miner 某个函数改坏了，马上能定位到出错文件。

2.3 事务数据类型统一为 List[Set[str]]，避免重复埋坑

频繁模式挖掘代码对输入数据的假设不多，但类型必须稳定。我习惯把一切来源统一成List[Set[str]]，列表的每一项代表一个订单，集合里的字符串代表该订单中的商品。用集合而不是列表，是因为 Apriori 里的子集判断issubset()在集合上走哈希运算，而且集合天然去重，同一订单重复购买的商品不会干扰计数。下面这个函数直接放到 loader.py 里。

# loader.py from typing import List, Set def to_transactions(raw_rows: List[List[str]]) -> List[Set[str]]: """把原始行数据转换为事务集。""" transactions = [] for row in raw_rows: items = {cell.strip() for cell in row if cell and cell.strip()} if items: transactions.append(items) return transactions

函数入参raw_rows来自 csv.reader 或 pandas 的values.tolist()，它是行数据的二维矩阵。集合推导式里的if cell过滤空字符串，cell.strip()去掉头尾空格，这是数据仓库导出数据最常出现的问题：同一个商品名在不同批次被写成牛奶和牛奶，strip 之后才能算作同一个项。函数返回时既去重也打乱了顺序，但顺序不影响挖掘结果，因为后续频繁项集判断走的是集合运算，不是按位置比较。

如果 loader 输出的事务集传给算法后，频繁项集数量与预期偏差很大，先检查是否漏了这步 strip。另外，事务之间必须独立，一个订单的商品绝不会跑到另一个订单里，分组逻辑应当在 loader 之前完成，不要等到 miner 里去强行按订单号拆分。

3. Python 实现频繁模式挖掘源代码：Apriori 的可运行脚本

3.1 候选集生成与剪枝：Apriori 循环的主体代码

这一部分直接给出 miner.py 的完整核心，代码可以原样复制到项目里跑通。两个函数加在一起就是 Apriori 的主体：candidate_gen负责由 k‑1 频繁项集生成 k 项候选并剪枝，apriori负责迭代计数。

# miner.py from itertools import combinations from typing import Dict, List, Set def candidate_gen(prev_itemsets: Set[frozenset], k: int) -> Set[frozenset]: """由 k-1 频繁项集生成 k 项候选，并执行 Apriori 剪枝。""" candidates = set() prev = list(prev_itemsets) for combo in combinations(prev, 2): union = combo[0] | combo[1] if len(union) == k: candidates.add(union) # 剪枝：候选的任意 k-1 维子集都必须出现在上一轮频繁项集中 pruned = set() for cand in candidates: if all(frozenset(sub) in prev_itemsets for sub in combinations(cand, k - 1)): pruned.add(cand) return pruned def apriori(transactions: List[Set[str]], min_support: float) -> Dict[int, Set[frozenset]]: """返回 {项集大小: 频繁项集合}，支持度统一用 float 表示。""" n = max(1, len(transactions)) item_count: Dict[str, int] = {} for t in transactions: for item in t: item_count[item] = item_count.get(item, 0) + 1 freq: Dict[int, Set[frozenset]] = {1: set()} for item, cnt in item_count.items(): if cnt / n >= min_support: freq[1].add(frozenset([item])) k = 2 while freq[k - 1]: candidates = candidate_gen(freq[k - 1], k) count = {cand: 0 for cand in candidates} for t in transactions: ts = set(t) for cand in candidates: if cand.issubset(ts): count[cand] += 1 freq[k] = {cand for cand, cnt in count.items() if cnt / n >= min_support} k += 1 return {size: itemsets for size, itemsets in freq.items() if itemsets}

关键参数min_support是 0 到 1 之间的小数，0.3 表示一个项集至少出现在 30% 的订单中才算频繁。函数第一步统计所有单物品出现次数，过滤出频繁 1 项集；随后 while 循环从频繁 k‑1 项集生成 k 项候选，每个候选都去事务库做子集命中计数，只有计数达到cnt / n >= min_support才进入下一轮。剪枝条件写得更严格一些：候选项的每一个 k‑1 子集都必须是上轮频繁项，这能提前砍掉大量无效条目。

用combinations(prev, 2)做两两连接是常见做法，它强调两个 k‑1 项集共享 k‑2 个元素时才可能得到长度为 k 的并集。课程报告里可以把每轮候选数量和剪枝后的数量做成表格，这比堆代码更能说明掌握程度。

3.2 关联规则生成：置信度、提升度怎么算

频繁项集本身只是中间产物，大作业展示的主体通常是关联规则。rules.py 里生成规则时，我习惯把置信度和提升度一起算出来，后者能过滤一批“看起来强但实际无意义”的规则。

# rules.py from typing import Dict, List, Set def generate_rules( freq: Dict[int, Set[frozenset]], transactions: List[Set[str]], min_confidence: float = 0.6, ): """返回规则列表，每条规则包含前件、后件、支持度、置信度、提升度。""" total = len(transactions) # 统计单个商品的支持度，用于计算提升度 item_support: Dict[str, int] = {} for t in transactions: for item in t: item_support[item] = item_support.get(item, 0) + 1 rules = [] for k, itemsets in freq.items(): if k < 2: continue for itemset in itemsets: for conseq in itemset: ante = itemset - {conseq} support_ante = ( sum(1 for t in transactions if ante.issubset(t)) / total ) support_rules = ( sum(1 for t in transactions if itemset.issubset(t)) / total ) confidence = support_rules / support_ante if support_ante else 0 lift = confidence / (item_support[conseq] / total) if confidence >= min_confidence: rules.append((ante, conseq, support_rules, confidence, lift)) return rules

min_confidence是指定规则最低置信度的阈值，0.6 表示前件出现时后件至少有 60% 的概率跟着出现。lift的计算是置信度除以后件无条件支持度，值大于 1 时才说明前件对后件有正向促进作用，等于 1 表示两者独立，小于 1 甚至可能是一种抑制关系。这里为了控制代码长度，规则只枚举了单后项的情况，作业要求完整规则集时，把for conseq in itemset换成对非空真子集的组合遍历即可。

3.3 main.py 命令行封装与 CSV 输出

命令行入口做成 argparse 形式，评分老师拿到 README 后不需要打开源码找参数。

python main.py --data data/transactions.csv --min-support 0.2 --min-confidence 0.5

main.py 只负责组装 loader、miner、rules 三段流程，不写任何算法细节。

# main.py import argparse import csv from loader import to_transactions, load_csv from miner import apriori from rules import generate_rules def main(): parser = argparse.ArgumentParser(description="频繁模式挖掘-数据仓库大作业") parser.add_argument("--data", required=True) parser.add_argument("--min-support", type=float, default=0.2) parser.add_argument("--min-confidence", type=float, default=0.5) args = parser.parse_args() txs = to_transactions(load_csv(args.data)) freq = apriori(txs, args.min_support) rules = generate_rules(freq, txs, args.min_confidence) with open("output/rules.csv", "w", newline="", encoding="utf-8-sig") as f: writer = csv.writer(f) writer.writerow(["前项", "后项", "支持度", "置信度", "提升度"]) for ante, conseq, sup, conf, lift in rules: writer.writerow([",".join(ante), conseq, f"{sup:.3f}", f"{conf:.3f}", f"{lift:.3f}"]) if __name__ == "__main__": main()

encoding="utf-8-sig"是常用做法，Excel 直接打开 CSV 时中文不会乱码。load_csv在 loader.py 里封装 csv.reader 即可，注意只读表头后按行组织原始数据。输出规则表保留前项、后项、支持度、置信度、提升度五列，后两列在报告分析和筛选时最有用。

提示：Windows 下调试时，如果控制台输出中文报错，先确认终端代码页支持 UTF-8，代码文件本身也要以 UTF-8 保存。

3.4 用微型数据集验证频繁项集和规则合理性

写完代码后先用四五个订单的小样本验证逻辑，不要直接上全量仓库数据。下面这个数据集故意构造了一条“知名反例”：买牛奶的人经常买面包，但提升度可能接近 1。

transactions = [ {"牛奶", "面包", "黄油"}, {"牛奶", "尿布", "啤酒"}, {"面包", "黄油"}, {"牛奶", "面包", "啤酒"}, ] freq = apriori(transactions, min_support=0.5) for k, items in freq.items(): for itemset in items: print(k, sorted(itemset)) rules = generate_rules(freq, transactions, min_confidence=0.5) for ante, conseq, sup, conf, lift in rules: print(set(ante), "->", conseq, f"support={sup:.2f}", f"confidence={conf:.2f}", f"lift={lift:.2f}")

min_support=0.5表示项集至少出现在两个订单中。这个例子会挖出频繁 1 项集“牛奶”“面包”“黄油”，频繁 2 项集{牛奶, 面包}和{面包, 黄油}。规则“牛奶 → 面包”的置信度是 2/3，提升度却只有 0.89，因为面包本身出现率就高；而“面包 → 黄油”的提升度大于 1，才真正说明两者存在正向关联。把这条解释写进报告，比放十张挖掘结果截图更能体现对频繁模式挖掘的理解。

4. 数据仓库与数据挖掘的结合：星型模型到事务集的预处理

4.1 用 SQL 在数据仓库中做订单聚合，而不是读明细文件

频繁模式挖掘的输入是一个个事务，而数据仓库常见输出是订单明细表。把明细按订单 ID 聚合成一行商品列表，这一步我一般留在 SQL 里完成，而不是先导出几万行明细再到 Python 里 groupby。数据库在聚合时还能顺带做状态过滤和空值处理，数据落盘后再校验一遍，整个流程更可控。下面这段 SQL 适合直接放到 data/warehouse_query.sql 里。

SELECT f.order_id, GROUP_CONCAT(DISTINCT p.product_name ORDER BY p.product_name) AS items FROM sales_fact f JOIN product_dim p ON f.product_key = p.product_key JOIN order_dim o ON f.order_key = o.order_key WHERE o.order_status = 'completed' AND p.product_name IS NOT NULL GROUP BY f.order_id;

这里的事实表 sales_fact 通过 product_key 与商品维度关联，通过 order_key 与订单维度关联，是数据仓库课程标准的星型模型写法。GROUP_CONCAT(DISTINCT ...)的作用是把一个订单下的多条商品记录拼成逗号分隔文本，DISTINCT 保证同一商品在事务中只出现一次。o.order_status = 'completed'是业务过滤，剔除了取消、退款订单，这条过滤规则的业务含义要写进文档说明。

需要注意 GROUP_CONCAT 是 MySQL 的方言，SQL Server 对应 STRING_AGG，PostgreSQL 对应 string_agg。如果数据量特别大，GROUP_CONCAT 默认最大长度可能截断商品列表，这时可以调整数据库会话参数或用两步聚合处理。SQL 执行结果导出为 CSV 后，loader 里按逗号拆分即可，不必再关注数据库方言差异。

4.2 数值字段离散化：用 pd.cut 把价格变成可挖掘的业务项

事务集中的项必须是类别值，而事实表里最常见的数值字段是单价和数量。如果不做离散化，两个不同价格的同款商品会被切割成“牛奶:3.99”和“牛奶:4.05”，支持度被严重压碎，规则失去业务解释。我一般用 pandas 的pd.cut把连续价格分箱，再把商品名和价格区间拼成复合项。

import pandas as pd df = pd.read_csv("sales_detail.csv") df["price_bin"] = pd.cut( df["unit_price"], bins=[0, 50, 100, 500, float("inf")], labels=["低价", "中低价", "中高价", "高价"], ) df["item_with_price"] = ( df["product_name"].astype(str) + "@" + df["price_bin"].astype(str) ) grouped = df.groupby("order_id")["item_with_price"] transactions = [set(items.dropna()) for order_id, items in grouped]

bins参数定义价格区间的边界，labels给每段命名，两者长度必须匹配。如果商品价格集中在 20 到 90 元，把边界改成[0, 30, 60, 120, float("inf")]会更贴近业务。复合项牛奶@低价让后续挖掘结果天然带业务上下文，报告中可以直接说“低价牛奶与中低价面包存在关联”，而不是对着一堆数字猜含义。分箱过细时支持度会骤降，如果最小支持度保持 0.3 而规则为空，优先减少区间数量，而不是把 min_support 压到 0.05。

4.3 挖掘结果分析：提升度排序和图表的报告用法

代码跑完之后，不要只把规则 CSV 往报告里一贴。先把提升度大于 1 的规则筛出来，再按提升度降序取 Top N，这样报告的核心结论才会集中在有实际价值的关联上。

import pandas as pd rules_df = pd.DataFrame( [ (",".join(ante), conseq, sup, conf, lift) for ante, conseq, sup, conf, lift in rules ], columns=["antecedent", "consequent", "support", "confidence", "lift"], ) top_rules = rules_df[rules_df["lift"] > 1].sort_values("lift", ascending=False) print(top_rules.head(10))

这段代码把 rules 列表转成 DataFrame，然后用布尔筛选把提升度不大于 1 的规则全部丢掉。提升度等于 1 表示前件与后件独立，这类规则写进报告会被认为是凑数。筛选后的 top_rules 还可以继续按支持度和置信度做二次过滤，具体阈值根据挖掘目标调整。

可视化层面，我常用置信度和提升度的散点图来展示规则分布，保存为 PNG 后插入报告 PDF。

import matplotlib.pyplot as plt plt.scatter(top_rules["confidence"], top_rules["lift"], alpha=0.6) plt.xlabel("confidence") plt.ylabel("lift") plt.savefig("output/rules_lift.png", dpi=200)

dpi=200保证图片插到 Word 或 PDF 里不至于发虚；散点图里右上角的规则是置信度又高、提升度又高的重点规则。图表标题和坐标轴标签要写中文时注意字体设置，否则会出现方块字。到这里，数据仓库取数、预处理、挖掘、分析四段流程已经完整，可以开始组装文档说明和报告 PDF。

5. 文档说明与报告 PDF：用脚本闭合大作业的最后一环

5.1 在 README 里给出可执行的验证命令

源代码和报告不一致是频繁模式挖掘大作业最常见的扣分项。我在 README 开头只放两段内容：运行环境，和一条完整的启动命令。命令必须是在干净环境里也跑得通的。

pip install pandas matplotlib python main.py --data data/transactions.csv --min-support 0.2 --min-confidence 0.5

运行后检查 output 目录下是否生成了 rules.csv 和 rules_lift.png。如果只有控制台输出、没有文件生成，说明 main.py 里输出目录不存在而没有创建，需要在代码里加os.makedirs("output", exist_ok=True)。README 里把这两步写清，老师不用找代码就能复现结果。

5.2 用 Pandoc 和 XeLaTeX 生成中文报告 PDF

报告 PDF 不建议用 Word 手工排，两者字体公式不一致时会非常难看。我一般用 Markdown 写报告，再用 Pandoc 转 PDF，公式、图表、代码块都能自动排版。转换命令如下。

pandoc report.md -o report.pdf --pdf-engine=xelatex \ -V CJKmainfont="Noto Sans CJK SC" \ -V geometry:margin=2.5cm

--pdf-engine=xelatex强制使用 XeLaTeX 引擎，这样才能处理中文 Unicode 字体；CJKmainfont指定中文字体名称，不同系统差异很大：Windows 可以写SimSun或Microsoft YaHei，macOS 可以用PingFang SC，Linux 用Noto Sans CJK SC或WenQuanYi Micro Hei。如果命令执行报错找不到字体，先查看系统已安装中文字体名再替换这一项。

报告中一旦涉及支持度公式和置信度公式，可以直接在 Markdown 里写 LaTeX 数学表达式，Pandoc 会自动渲染。没有安装完整 TeX 发行版的情况下，这条路比较麻烦，遇到环境问题时应优先补装 texlive-lang-chinese 组件，而不是改用其他 PDF 工具。

5.3 交付前做三次人眼核验

不管代码多完整，PDF 最后的检查要做三次。第一次，随便抽 5 个包含规则前件的订单，人工核对后件是否同时出现，这一步防的是事务拆分引入的脏数据。第二次，在文档说明里写清楚 min_support 分别取 0.1、0.2、0.3 时频繁项集数量的变化，这比只报告一个参数更有说服力。第三次，把 rules.csv 里的前 10 条规则翻译成一句业务话术，例如“低价牛奶与中低价面包同单率高”，然后放进报告结论段。把上面命令行预演一遍并保留输出截图，再生成 PDF，这份频繁模式挖掘源代码加文档说明加报告 pdf 就具备了拿到高分的完整条件。

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