数据挖掘实战：Apriori算法原理详解与Python实现

1次阅读

没有评论

共计 3868 个字符，预计需要花费 10 分钟才能阅读完成。

关联规则挖掘是数据挖掘中的重要技术，用于发现大量数据项之间的有趣关联或相关关系。想象一下超市的购物篮数据，我们可能发现 ” 购买啤酒的顾客也经常购买尿布 ” 这样的关联规则，这就是著名的 ” 啤酒与尿布 ” 案例。

Apriori 算法正是解决这类问题的经典算法，它通过寻找频繁项集来发现数据中的关联规则。对于初学者来说，理解 Apriori 算法是进入关联规则挖掘领域的良好起点。

项集(Itemset): 一组物品的集合，如{牛奶, 面包}
支持度(Support): 项集在所有交易中出现的频率
置信度(Confidence): 在包含 X 的交易中，也包含 Y 的条件概率
频繁项集(Frequent Itemset): 支持度大于最小支持度阈值的项集
强关联规则: 同时满足最小支持度和最小置信度的规则

Apriori 算法的核心是 ” 先验原理 ”：

如果一个项集是频繁的，那么它的所有子集也一定是频繁的
反之，如果一个项集是非频繁的，那么它的所有超集也一定是非频繁的

这一性质大大减少了需要考察的项集数量，是算法高效的关键。

下面我们用一个完整的 Python 实现来演示 Apriori 算法的工作原理。这个实现只使用 Python 标准库，适合学习算法本质。

def load_dataset():
    """创建示例数据集"""
    return [['牛奶', '面包', '尿布'],
        ['可乐', '面包', '尿布', '啤酒'],
        ['牛奶', '尿布', '啤酒', '鸡蛋'],
        ['面包', '牛奶', '尿布', '啤酒'],
        ['面包', '牛奶', '尿布', '可乐']
    ]

def create_c1(dataset):
    """创建初始候选项集"""
    c1 = []
    for transaction in dataset:
        for item in transaction:
            if [item] not in c1:
                c1.append([item])
    c1.sort()
    return list(map(frozenset, c1))

def scan_d(dataset, candidates, min_support):
    """扫描数据集，计算候选项集的支持度"""
    item_count = {}
    for transaction in dataset:
        for candidate in candidates:
            if candidate.issubset(transaction):
                item_count[candidate] = item_count.get(candidate, 0) + 1

    num_items = float(len(dataset))
    freq_items = []
    support_data = {}

    for item in item_count:
        support = item_count[item] / num_items
        if support >= min_support:
            freq_items.append(item)
        support_data[item] = support

    return freq_items, support_data

def apriori_gen(freq_items, k):
    """生成候选项集"""
    candidates = []
    len_freq = len(freq_items)

    for i in range(len_freq):
        for j in range(i+1, len_freq):
            l1 = list(freq_items[i])[:k-2]
            l2 = list(freq_items[j])[:k-2]
            l1.sort()
            l2.sort()

            if l1 == l2:
                candidates.append(freq_items[i] | freq_items[j])

    return candidates

def apriori(dataset, min_support=0.5):
    """Apriori 算法主函数"""
    c1 = create_c1(dataset)
    d = list(map(set, dataset))
    l1, support_data = scan_d(d, c1, min_support)

    l = [l1]
    k = 2

    while len(l[k-2]) > 0:
        ck = apriori_gen(l[k-2], k)
        lk, sup_k = scan_d(d, ck, min_support)
        support_data.update(sup_k)
        l.append(lk)
        k += 1

    return l, support_data

def generate_rules(l, support_data, min_confidence=0.7):
    """从频繁项集中生成关联规则"""
    rules = []

    for i in range(1, len(l)):
        for freq_set in l[i]:
            subsets = [frozenset([item]) for item in freq_set]

            if i == 1:
                calculate_confidence(freq_set, subsets, support_data, rules, min_confidence)
            else:
                calculate_confidence(freq_set, subsets, support_data, rules, min_confidence)
                ap_gen_rules(freq_set, subsets, support_data, rules, min_confidence)

    return rules

def calculate_confidence(freq_set, subsets, support_data, rules, min_confidence):
    """计算规则的置信度"""
    pruned = []

    for subset in subsets:
        remain = freq_set - subset
        confidence = support_data[freq_set] / support_data[subset]

        if confidence >= min_confidence:
            rules.append((subset, remain, confidence))
            pruned.append(subset)

    return pruned

def ap_gen_rules(freq_set, subsets, support_data, rules, min_confidence):
    """递归生成候选规则"""
    m = len(subsets[0])

    if len(freq_set) > (m + 1):
        next_subsets = apriori_gen(subsets, m + 1)
        next_subsets = calculate_confidence(freq_set, next_subsets, support_data, rules, min_confidence)

        if len(next_subsets) > 1:
            ap_gen_rules(freq_set, next_subsets, support_data, rules, min_confidence)

# 使用示例
dataset = load_dataset()
L, support_data = apriori(dataset, min_support=0.6)
rules = generate_rules(L, support_data, min_confidence=0.7)

print("频繁项集:")
for itemset in L:
    print(itemset)

print("\n 关联规则:")
for rule in rules:
    print(f"{rule[0]} => {rule[1]} (置信度: {rule[2]:.2f})")

Apriori 算法的时间复杂度主要取决于：