Python脚本数据关联规则Apriori如何实现

wen 实用脚本 2026-07-18 181

本文目录导读：

方法一：使用mlxtend库（推荐）
方法二：手动实现Apriori算法
实际应用示例：购物篮分析
关键参数调优建议

我来详细介绍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库，它经过优化且功能完备，手动实现版本则有助于理解算法原理。

本文地址： https://www.bmjup.com/post/24539.html