Apriori算法（原理步骤、Python实现、apyori库实现）

本文使用的源码以及数据集Data-Mining/Apriori at Stellaris.github.io · Stellaris123/Data-Mining

Apriori

​ Apriori 算法采用的策略是将数管来拿规则挖掘任务分为①产生频繁项集、②产生规则，两主要步骤。

详细步骤

一、产生频繁项集：从数据中发现满足最小支持度阈值的所有项集。

（1）确定数据集 $D$ 中所包含的项集，并具体到每一个独立数据点（生成候选一项集）

（2）将候选集中每个项目单独进行扫描，根据支持度公式进行计算支持度，并筛掉不满足设定的最小支持度阈值的项。

（3）从$k-1$项集生成候选 $k$ 项集：判断 $k-1$ 项集中每两个的项中是否满足两个项的 $k-2$ 前缀是相同的，要是相同就表示可以构造（这是通过一个项满足支持则其子项也一定满足支持得出）。

（4）将候选 $k$ 项集进行筛选，产生 $k$ 项集。

（5）重复 3~4 步骤直到项集小于等于 1。

二、从产生的频繁项集中找出满足最小置信度阈值的规则。

（1）对于每个频繁项集，产生其所有的非空子集。

（2）对于每个非空子集，根据置信度公式进计算置信度，并筛掉不满足设定的最小置信度阈值的项。

（3）循环1~2步骤，直到挖出所有关联规则。

支持度和置信度

公式定义

$$\begin{gathered} Support(A\cap B)=\frac{Support\_count(A\cap B)}{sum}=\frac{A,B同时发生的事物个数}{所有事物个数} \\ Confidence(A\Rightarrow B)=P(A|B)=\frac{Support(A\cap B)}{Support(A)}=\frac{Support\_count(A\cap B)}{Support\_count(A)}=\frac{A,B同时发生的事物个数}{发生A的事物个数} \end{gathered}$$

数据例子

存在以下 $9$ 条购物记录：1为选购，0为未选购。

（1）支持度（$A\cap B$为例）

$Support\_count(A\cap B)=5$，$sum=9$，所以 $Support(A\cap B)=\frac{Support\_count(A\cap B)}{sum}=\frac{5}{9}$

（2）置信度（$A\Rightarrow B$为例）

$Support\_count(A\cap B)=5$，$Support\_count(A)=6$，所以$Confidence(A\Rightarrow B)=\cdots =\frac{5}{6}$

基于 Python 实现

Apriori类

class Apriori:
    """
    ---类名---
     Apriori
    ---数据成员---
     MIN_Support:    最小支持度
     MIN_Confidence: 最小置信度
     K_ItemSet:      k项集
     K_ItemSupprot:  k项集支持度
     rules:          规则
    ---方法---
     __init__(): 设置最小支持度和最小置信度（默认为0.5,0.5）
     init:       初始化，方便对象生成和调用不在一起
     get_candidate_one_set(): 获得候选第一项集
     get_k_itemset():     获得 k 项集
     sieve_itmeset():     筛选项集
     get_rules():         获取规则
     one_confidence():    计算并筛选规则
     more_confidence():   递归规则
     display_k_itemset(): 显示 k 项集
     display_rules():     显示规则
     run(): 运行入口
    """
    # 对象生成初始化
    def __init__(self, MIN_Support = 0.5, MIN_Confidence = 0.5):
        self.MIN_Support = MIN_Support
        self.MIN_Confidence = MIN_Confidence
        self.K_ItemSet = []
        self.K_ItemSupprot = []
        self.rules = []
    # 初始化
    def init(self):
        self.K_ItemSet = []
        self.K_ItemSupprot = []
        self.rules = []
    # 获得候选一项集
    def get_candidate_one_set(self, dataSet):
        one_set = []
        # 遍历添加
        for lst in dataSet:
            for item in lst:
                if not [item] in one_set:
                    one_set.append([item])
        # 排序
        one_set.sort()
        # 将列表冻结
        return list(map(frozenset, one_set))
    # 获得 K 项集
    def get_k_itemset(self, k_next, K):
        k_itemset = [] #K-项集
        # 两两组合
        for i in range(0,len(k_next)):
            for j in range(i+1,len(k_next)):
                # 判断前 k-2 项是否相同
                L1 = list(k_next[i])[:K-2]
                L2 = list(k_next[j])[:K-2]
                if L1==L2:
                    k_itemset.append(k_next[i]|k_next[j])
        return k_itemset
    # 筛选项集
    def sieve_itmeset(self, Data, itmeset):
        item_frequency = {}
        for item in Data:
            for element in itmeset:
                if element.issubset(item):
                    if not element in item_frequency:
                        item_frequency[element] = 1
                    else:
                        item_frequency[element] += 1
        
        numItems = float(len(Data)) #数据集中总的项集个数
        
        sieve_k_itemset = [] #频繁项集
        sieve_k_itemSupprot={} #项集中各项支持度
        for key in item_frequency:
            support = item_frequency[key]/numItems #支持度计算：元素出现次数 / 总项集数
            if support>=self.MIN_Support:  #满足最小支持度添加到频繁项集列表中
                sieve_k_itemset.insert(0, key)
                sieve_k_itemSupprot[key] = support
        return sieve_k_itemset, sieve_k_itemSupprot
    # 运行入口
    def run(self, dataSet):
        self.init()
        # 获取候选一项集
        candidate_one_set = self.get_candidate_one_set(dataSet)
        # 转换为字典
        Data = list(map(set, dataSet))
        # 过滤候选一项集
        sieve_one_set, self.K_ItemSupprot = self.sieve_itmeset(Data, candidate_one_set)
        # k=1的列表
        self.K_ItemSet = [[0], sieve_one_set]
        # 一直筛选直到小于等于 1 为止。
        K = 2
        while(len(self.K_ItemSet[K-1]) > 1):
            # 从 k-1 中构造候选 k 项集
            candidate_k_itemset = self.get_k_itemset(self.K_ItemSet[K-1], K)
            # 过滤候选 k 项集，并获取 k 项集支持度
            sieve_k_itemset, sieve_k_itemSupprot = self.sieve_itmeset(Data, candidate_k_itemset)
            
            # 更新项集
            self.K_ItemSupprot.update(sieve_k_itemSupprot)
            self.K_ItemSet.append(sieve_k_itemset)
            K += 1
        self.get_rules()
    # 获取规则
    def get_rules(self):
        self.rules = []
        # 将每一个分项集构造出关系
        for i in range(2, len(self.K_ItemSet)):
            for k_itemset in self.K_ItemSet[i]:
                fz_k_itemset = [frozenset([x]) for x in k_itemset]
                #print(fz_k_itemset)
                if(i > 2):
                    self.more_confidence(k_itemset,fz_k_itemset)
                else:
                    self.one_confidence(k_itemset,fz_k_itemset)
    # 计算并筛选规则
    def one_confidence(self, itemset, fz_k_itemset):
        #存储 fz_k_itemset 项中强关联规则
        item_realted = []
        for item in fz_k_itemset:
            # 计算置信度
            conf = self.K_ItemSupprot[itemset]/self.K_ItemSupprot[itemset - item]
            # 判断是否满足最小置信度
            if conf >= self.MIN_Confidence:
                self.rules.append((itemset - item, item, conf))
                item_realted.append(item)
        return item_realted                                                                                                                                      
    # 递归规则
    def more_confidence(self, itemset, fz_k_itemset):        
        if (len(itemset) > len(fz_k_itemset[0])+1):
            itemset_reunion = self.get_k_itemset(fz_k_itemset, len(fz_k_itemset[0]) + 1)
            # 计算置信度并过滤
            itemset_reunion_filtered = self.one_confidence(itemset, itemset_reunion)
            # 若存在强关联规则，递归组合计算置信度
            if(len(itemset_reunion_filtered) > 1):
                self.more_confidence(itemset, itemset_reunion_filtered)
    # 显示 k 项集
    def display_k_itemset(self):
        for k_itemset in self.K_ItemSet:
            print(k_itemset)
    # 显示规则
    def display_rules(self):
        for rule in self.rules:
            print(str(rule[0]) + " ---> " + str(rule[1]) + " Confidence:" + str(rule[2]))
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145

调用

apriori = Apriori(MIN_Support=0.4, MIN_Confidence=0.5)
apriori.run(dataSet)
apriori.display_rules()
apriori.display_k_itemset()
1
2
3
4

基于 apyori 实现

from apyori import apriori
pd.DataFrame(apriori(transactions=dataSet, min_support=0.5, min_confidence=0.6))
1
2

附录（关于数据预处理）

​ 存在一张表，如下：（第一行是列号，忽略）

可以使用下面这个方法将这个 $excel$ 转化为需要的列表：

import numpy as np
import pandas as pd
data = pd.read_excel("./menu_orders.xls")

df = pd.DataFrame(list(map(lambda x:pd.Series(1, index = x[pd.notnull(x)]), data.values))).fillna(0)
#df = df[sorted(df.columns)]

dataSet = []
for i in range(0,df.shape[0]):
    dataSet.append([x for x in df.columns if df[x][i]])
print(dataSet)
1
2
3
4
5
6
7
8
9
10
11

执行结果：

[['b', 'd'], ['b', 'c'], ['a', 'b', 'c', 'd'], ['a', 'b'], ['b', 'c'], ['a', 'b'], ['a', 'b', 'c', 'e'], ['a', 'b', 'c'], ['a', 'c', 'e']]
1