本文目录导读：
我来详细介绍Python中实现Apriori算法的两种方式：
使用mlxtend库（推荐）
这是最简单的实现方式：
# 安装：pip install mlxtend
import pandas as pd
from mlxtend.preprocessing import TransactionEncoder
from mlxtend.frequent_patterns import apriori, association_rules
# 示例数据：交易数据
dataset = [
['牛奶', '面包', '鸡蛋'],
['牛奶', '面包'],
['牛奶', '鸡蛋', '果酱'],
['面包', '果酱', '果冻'],
['牛奶', '面包', '果酱', '鸡蛋'],
['面包', '鸡蛋', '果冻'],
]
# 数据转换：将交易数据转换为One-Hot编码格式
te = TransactionEncoder()
te_ary = te.fit(dataset).transform(dataset)
df = pd.DataFrame(te_ary, columns=te.columns_)
print("转换后的数据格式：")
print(df.head())
# 1. 挖掘频繁项集
frequent_itemsets = apriori(df, min_support=0.4, use_colnames=True)
print("\n频繁项集：")
print(frequent_itemsets)
# 2. 生成关联规则
rules = association_rules(frequent_itemsets, metric="lift", min_threshold=1.0)
# 关注提升度高的规则
rules = rules.sort_values('lift', ascending=False)
print("\n关联规则：")
print(rules[['antecedents', 'consequents', 'support', 'confidence', 'lift']].head())
手动实现Apriori算法
如果你想深入了解原理,可以自己实现：
from itertools import combinations
from collections import defaultdict
class Apriori:
def __init__(self, min_support=0.3, min_confidence=0.6):
self.min_support = min_support
self.min_confidence = min_confidence
def fit(self, transactions):
"""执行Apriori算法"""
# 数据预处理
self.transactions = [set(t) for t in transactions]
self.num_transactions = len(self.transactions)
# 第1步：生成频繁1项集
item_counts = defaultdict(int)
for transaction in self.transactions:
for item in transaction:
item_counts[frozenset([item])] += 1
# 过滤支持度
self.freq_itemsets = {}
for itemset, count in item_counts.items():
support = count / self.num_transactions
if support >= self.min_support:
self.freq_itemsets[itemset] = support
# 第2步：迭代生成更大项集
k = 2
current_freq = self.freq_itemsets
while current_freq:
# 生成候选集
candidates = self._generate_candidates(list(current_freq.keys()), k)
# 计算候选集支持度
next_freq = {}
for candidate in candidates:
count = sum(1 for t in self.transactions if candidate.issubset(t))
support = count / self.num_transactions
if support >= self.min_support:
next_freq[candidate] = support
# 更新频繁项集
self.freq_itemsets.update(next_freq)
current_freq = next_freq
k += 1
return self
def _generate_candidates(self, freq_itemsets, k):
"""生成候选k项集"""
candidates = []
n = len(freq_itemsets)
for i in range(n):
for j in range(i+1, n):
# 合并两个k-1项集
union = freq_itemsets[i] | freq_itemsets[j]
if len(union) == k and union not in candidates:
# 剪枝：检查所有子集是否频繁
if self._has_infrequent_subset(union, freq_itemsets):
continue
candidates.append(union)
return candidates
def _has_infrequent_subset(self, candidate, freq_itemsets):
"""检查候选集的所有子集是否为频繁项集"""
# 生成所有k-1大小的子集
subsets = list(combinations(candidate, len(candidate)-1))
for subset in subsets:
if frozenset(subset) not in freq_itemsets:
return True
return False
def generate_rules(self):
"""从频繁项集生成关联规则"""
rules = []
for itemset in self.freq_itemsets:
if len(itemset) < 2:
continue
# 生成所有非空真子集作为前提
for i in range(1, len(itemset)):
for antecedent in combinations(itemset, i):
antecedent = frozenset(antecedent)
consequent = itemset - antecedent
if consequent:
# 计算置信度
support_antecedent = self.freq_itemsets.get(antecedent, 0)
support_itemset = self.freq_itemsets[itemset]
confidence = support_itemset / support_antecedent if support_antecedent > 0 else 0
if confidence >= self.min_confidence:
# 计算提升度
support_consequent = self.freq_itemsets.get(consequent, 0)
lift = confidence / support_consequent if support_consequent > 0 else 0
rules.append({
'antecedent': set(antecedent),
'consequent': set(consequent),
'support': support_itemset,
'confidence': confidence,
'lift': lift
})
return sorted(rules, key=lambda x: x['lift'], reverse=True)
# 使用手动实现的Apriori
if __name__ == "__main__":
# 示例数据
dataset = [
['牛奶', '面包', '鸡蛋'],
['牛奶', '面包'],
['牛奶', '鸡蛋', '果酱'],
['面包', '果酱', '果冻'],
['牛奶', '面包', '果酱', '鸡蛋'],
['面包', '鸡蛋', '果冻'],
]
# 创建并训练模型
ap = Apriori(min_support=0.4, min_confidence=0.5)
ap.fit(dataset)
print("手动实现的结果：")
rules = ap.generate_rules()
for rule in rules[:5]:
print(f"{rule['antecedent']} -> {rule['consequent']}: "
f"support={rule['support']:.2f}, "
f"confidence={rule['confidence']:.2f}, "
f"lift={rule['lift']:.2f}")
# 显示所有频繁项集
print(f"\n所有频繁项集 (support >= {ap.min_support}):")
for itemset, support in ap.freq_itemsets.items():
print(f"{set(itemset)}: support={support:.2f}")
实际应用示例：购物篮分析
import pandas as pd
from mlxtend.frequent_patterns import apriori, association_rules
from mlxtend.preprocessing import TransactionEncoder
def market_basket_analysis(transactions, min_support=0.05, min_confidence=0.5):
"""
购物篮分析完整流程
Parameters:
-----------
transactions : list of lists
交易数据
min_support : float
最小支持度
min_confidence : float
最小置信度
"""
# 1. 数据预处理
te = TransactionEncoder()
te_ary = te.fit(transactions).transform(transactions)
df = pd.DataFrame(te_ary, columns=te.columns_)
print("数据概览：")
print(f"交易数量: {len(transactions)}")
print(f"商品种类: {len(te.columns_)}")
# 2. 挖掘频繁项集
frequent_itemsets = apriori(df, min_support=min_support, use_colnames=True)
print(f"\n发现 {len(frequent_itemsets)} 个频繁项集")
# 3. 生成关联规则
rules = association_rules(frequent_itemsets, metric="lift", min_threshold=1.0)
# 4. 筛选有价值的规则
# 高置信度规则
high_conf_rules = rules[rules['confidence'] >= min_confidence].copy()
# 添加规则长度信息
high_conf_rules['antecedent_len'] = high_conf_rules['antecedents'].apply(len)
high_conf_rules['consequent_len'] = high_conf_rules['consequents'].apply(len)
# 推荐有价值的规则
print(f"\n高价值关联规则 (前10条):")
# 按提升度排序
top_rules = high_conf_rules.nlargest(10, 'lift')
for idx, rule in top_rules.iterrows():
ante = ', '.join(list(rule['antecedents']))
cons = ', '.join(list(rule['consequents']))
print(f"如果购买 [{ante}]，则可能购买 [{cons}]")
print(f" 支持度: {rule['support']:.2%}")
print(f" 置信度: {rule['confidence']:.2%}")
print(f" 提升度: {rule['lift']:.2f}")
print("-" * 50)
return frequent_itemsets, high_conf_rules
# 使用示例
if __name__ == "__main__":
# 模拟电商交易数据
transactions = [
['手机', '充电器', '耳机'],
['手机', '手机壳', '充电器'],
['手机', '充电器'],
['笔记本电脑', '鼠标', '包'],
['手机', '充电器', '耳机', '手机壳'],
['笔记本电脑', '包'],
['手机', '耳机'],
['手机', '充电器', '手机壳'],
['手机', '充电器', '耳机', '充电宝'],
['笔记本电脑', '鼠标'],
]
# 执行购物篮分析
frequent_itemsets, rules = market_basket_analysis(
transactions,
min_support=0.2, # 至少20%的交易中出现
min_confidence=0.6 # 至少60%的置信度
)
关键参数调优建议
def apriori_parameter_tuning(transactions, item_names=None):
"""
Apriori参数调优指南
"""
from mlxtend.preprocessing import TransactionEncoder
from mlxtend.frequent_patterns import apriori, association_rules
import pandas as pd
# 数据转换
te = TransactionEncoder()
te_ary = te.fit(transactions).transform(transactions)
df = pd.DataFrame(te_ary, columns=te.columns_)
# 不同min_support的对比
supports = [0.1, 0.2, 0.3, 0.4, 0.5]
print("不同最小支持度的影响:")
for sup in supports:
# 频繁项集数量
itemsets = apriori(df, min_support=sup, use_colnames=True)
# 生成的规则数
rules = association_rules(itemsets, metric="lift", min_threshold=1.0)
print(f"min_support={sup}: 频繁项集 {len(itemsets)}个, 规则 {len(rules)}条")
# 建议的参数设置
print("\n参数建议:")
print("-" * 40)
print("min_support: 0.01-0.1 (大数据集), 0.1-0.5 (小数据集)")
print("min_confidence: 0.5-0.8")
print("metric: 'lift' (推荐), 'confidence', 'support'")
print("min_threshold: 1.0-3.0 (对于lift)")
这些代码提供了从基础实现到实际应用的完整方案,对于生产环境，推荐使用mlxtend库，它经过优化且功能完备，手动实现版本则有助于理解算法原理。
