【星海隨筆】關聯規則(二) python實現

def loadDataSet():
return [[1,2,5],[2,4],[2,3],[1,2,4],[1,3],[2,3],[1,3],[1,2,3,5],[1,2,3]]

#1.構建候選1項集C1
def createC1(dataSet):
C1 = []
for transaction in dataSet:
for item in transaction:
if not [item] in C1:
C1.append([item])
C1.sort()
return list(map(frozenset, C1))
#
# frozenset是凍結的集合，它是不可變的，存在哈希值。
# 好處是它可以作為字典的key，也可以作為其它集合的元素。
# 缺點是一旦創建便不能更改，沒有add，remove方法。
#

def apriori(dataSet, minSupport = 0.5):
C1 = createC1(dataSet)
#獲得一個集合，用來再scanD中，方便統計每個數字出現的概率。
L1, supportData = scanD(dataSet, C1, minSupport)
#使用 scanD 函數進行計算支持度，支持度的概念在 關聯規則 （一） 中已經講解。
L = [L1]
#
# 多加一層列表
#[[frozenset({1}), frozenset({3}), frozenset({2}), frozenset({5})]]
# 注意{4} 已經沒有了，因為他的出現的概率為 25% 。低於最低支持度50%
#
k = 2
while (len(L[k-2]) > 0):
#
# 第一次進入 L 如果存在則繼續
#
Lk_1 = L[k-2]
# 
Ck = aprioriGen(Lk_1, k)
#
# 循環獲得不同的 集合集，應為是集合所以默認去重
# 例如： [frozenset({1, 3}), frozenset({1, 2}), frozenset({1, 5}), frozenset({2, 3}), frozenset({3, 5}), frozenset({2, 5})]
# 第二次 為 [frozenset({1, 2}), frozenset({1, 5}), frozenset({2, 5}), frozenset({2, 4}), frozenset({2, 3}), frozenset({1, 3})]
# 第三次 為 [frozenset({1, 2, 5}), frozenset({1, 2, 3})]
#
print("ck:",Ck)
Lk, supK = scanD(dataSet, Ck, minSupport)
#
# 每個數字出現在列表中的概率，重復出現在一個列表中算一次
# 返回的是每個字符，和每個字符出現的概率。
# 循環計算支持度
#
supportData.update(supK)
print("lk:", Lk)
L.append(Lk)
k += 1
return L, supportData

# 過濾掉不符合支持度的集合
# 返回 頻繁項集列表retList 所有元素的支持度字典
# support = ssCnt[key] / numItems
# numItems 是len(D) 的結果，D 是data元數據，既開始load的數據的數量。
# Ck 是不同的集合。
def scanD(D, Ck, minSupport):
#候選集計數
ssCnt = {
}
for tid in D:
#
# D 傳進來的是列表。
# 最初的dataset
#
for can in Ck:
#Ck 傳進來的是 frozenset 類型 。
#frozenset是凍結的集合，它是不可變的，存在哈希值，好處是它可以作為字典的key，也可以作為其它集合的元素。缺點是一旦創建便不能更改，沒有add，remove方法。
if can.issubset(tid):
if can not in ssCnt.keys(): ssCnt[can] = 1
else: ssCnt[can] += 1
#
# 上半部分獲得了每個 數字 出現的數量，使用字典的方式保存。
# ssCnt 字典 key 為 frozenset，volume 為再列表中出現的次數，如果在同一個列表中出現兩次，也算一次，計算的是每個列表擁有這個數字的概率。
#
numItems = float(len(D))
Lk= []
# 候選集項Cn生成的頻繁項集Lk
supportData = {
}
#
# 候選集項Cn的支持度字典
# 計算候選項集的支持度, supportData key:候選項， value:支持度
#
for key in ssCnt:
support = ssCnt[key] / numItems
if support >= minSupport:
Lk.append(key)
supportData[key] = support
#
# 判斷 列表中 出現指定 key 的概率。
# 如果大於 minSupport 則保留。
#
return Lk, supportData
#
#return 的格式如下：
#{frozenset({1}): 0.5,
#frozenset({3}): 0.75,
#frozenset({4}): 0.25,
#frozenset({2}): 0.75,
#frozenset({5}): 0.75})
#

# 循環獲得不同的 集合集，應為是集合所以默認去重
# 例如： [frozenset({1, 3}), frozenset({1, 2}), frozenset({1, 5}), frozenset({2, 3}), frozenset({3, 5}), frozenset({2, 5})]
# 第二次 為 [frozenset({1, 2}), frozenset({1, 5}), frozenset({2, 5}), frozenset({2, 4}), frozenset({2, 3}), frozenset({1, 3})]
# 第三次 為 [frozenset({1, 2, 5}), frozenset({1, 2, 3})]
def aprioriGen(Lk_1, k):
#
# apriori while進入
# 對上層的列表遞歸傳入。
# 第一次對整個 [frozenset({1}), frozenset({3}), frozenset({2}), frozenset({5})] 完整傳入。
# k 第一次傳入為 2 , 之後遞增
#
Ck = []
lenLk = len(Lk_1)
for i in range(lenLk):
L1 = list(Lk_1[i])[:k - 2]
#
# 第一次傳進來的時候在這個位置 列表的【：k- 2】 為空。
#
L1.sort()
for j in range(i + 1, lenLk):
#前k-2個項相同時，將兩個集合合並
L2 = list(Lk_1[j])[:k - 2]
L2.sort()
if L1 == L2:
Ck.append(Lk_1[i] | Lk_1[j])
#
#循環形成並集
# [frozenset({1, 3}), frozenset({1, 2}), frozenset({1, 5}), frozenset({2, 3}), frozenset({3, 5}), frozenset({2, 5})]
# 獲取每種組合，為了後面計算每種組合的概率。也就是支持度。
# 循環獲取自身的第一個出現的 字符，循環匹配在自己後面的組合，如果該組合的第一個字符與自身的第一個字符 相等 則並集。
#
return Ck

dataset = loadDataSet()
L, supportData = apriori(dataset, minSupport=0.2)

ck: [frozenset({1, 2}), frozenset({1, 5}), frozenset({1, 4}),
frozenset({1, 3}), frozenset({2, 5}), frozenset({2, 4}), frozenset({2,
3}), frozenset({4, 5}), frozenset({3, 5}), frozenset({3, 4})] lk:
[frozenset({1, 2}), frozenset({1, 5}), frozenset({2, 5}),
frozenset({2, 4}), frozenset({2, 3}), frozenset({1, 3})] ck:
[frozenset({1, 2, 5}), frozenset({1, 2, 3}), frozenset({1, 3, 5}),
frozenset({2, 4, 5}), frozenset({2, 3, 5}), frozenset({2, 3, 4})] lk:
[frozenset({1, 2, 5}), frozenset({1, 2, 3})] ck: [frozenset({1, 2, 3,
5})] lk: []

擴展一：

L

[[frozenset({
1}),
frozenset({
2}),
frozenset({
5}),
frozenset({
4}),
frozenset({
3})],
[frozenset({
1, 2}),
frozenset({
1, 5}),
frozenset({
2, 5}),
frozenset({
2, 4}),
frozenset({
2, 3}),
frozenset({
1, 3})],
[frozenset({
1, 2, 5}), frozenset({
1, 2, 3})],
[]]

supportData

{
frozenset({
1}): 0.6666666666666666,
frozenset({
2}): 0.7777777777777778,
frozenset({
5}): 0.2222222222222222,
frozenset({
4}): 0.2222222222222222,
frozenset({
3}): 0.6666666666666666,
frozenset({
1, 2}): 0.4444444444444444,
frozenset({
1, 5}): 0.2222222222222222,
frozenset({
2, 5}): 0.2222222222222222,
frozenset({
2, 4}): 0.2222222222222222,
frozenset({
2, 3}): 0.4444444444444444,
frozenset({
1, 4}): 0.1111111111111111,
frozenset({
1, 3}): 0.4444444444444444,
frozenset({
3, 5}): 0.1111111111111111,
frozenset({
1, 2, 5}): 0.2222222222222222,
frozenset({
1, 2, 3}): 0.2222222222222222,
frozenset({
1, 3, 5}): 0.1111111111111111,
frozenset({
2, 3, 5}): 0.1111111111111111,
frozenset({
1, 2, 3, 5}): 0.1111111111111111}

from numpy import *
# 構造數據
def loadDataSet():
return [[1, 3, 4], [2, 3, 5], [1, 2, 3, 5], [2, 5]]
# 將所有元素轉換為frozenset型字典，存放到列表中
def createC1(dataSet):
C1 = []
for transaction in dataSet:
for item in transaction:
if not [item] in C1:
C1.append([item])
C1.sort()
# 使用frozenset是為了後面可以將這些值作為字典的鍵
return list(map(frozenset, C1)) # frozenset一種不可變的集合，set可變集合
# 過濾掉不符合支持度的集合
# 返回 頻繁項集列表retList 所有元素的支持度字典
# support = ssCnt[key] / numItems
# numItems 是len(D) 的結果，D 是data元數據，既開始load的數據的數量。
# Ck 是不同的集合。
def scanD(D, Ck, minSupport):
ssCnt = {
}
for tid in D:
for can in Ck:
if can.issubset(tid): # 判斷can是否是tid的《子集》 （這裡使用子集的方式來判斷兩者的關系）
if can not in ssCnt: # 統計該值在整個記錄中滿足子集的次數（以字典的形式記錄，frozenset為鍵）
ssCnt[can] = 1
else:
ssCnt[can] += 1
numItems = float(len(D))
retList = [] # 重新記錄滿足條件的數據值（即支持度大於阈值的數據）
supportData = {
} # 每個數據值的支持度
for key in ssCnt:
support = ssCnt[key] / numItems
if support >= minSupport:
retList.insert(0, key)
supportData[key] = support
return retList, supportData # 排除不符合支持度元素後的元素 每個元素支持度
# 生成所有可以組合的集合
# 頻繁項集列表Lk 項集元素個數k [frozenset({2, 3}), frozenset({3, 5})] -> [frozenset({2, 3, 5})]
def aprioriGen(Lk, k):
retList = []
lenLk = len(Lk)
for i in range(lenLk): # 兩層循環比較Lk中的每個元素與其它元素
for j in range(i+1, lenLk):
L1 = list(Lk[i])[:k-2] # 將集合轉為list後取值
L2 = list(Lk[j])[:k-2]
L1.sort(); L2.sort() # 這裡說明一下：該函數每次比較兩個list的前k-2個元素，如果相同則求並集得到k個元素的集合
if L1==L2:
retList.append(Lk[i] | Lk[j]) # 求並集
return retList # 返回頻繁項集列表Ck
# 封裝所有步驟的函數
# 返回 所有滿足大於阈值的組合 集合支持度列表
def apriori(dataSet, minSupport = 0.5):
D = list(map(set, dataSet)) # 轉換列表記錄為字典 [{1, 3, 4}, {2, 3, 5}, {1, 2, 3, 5}, {2, 5}]
C1 = createC1(dataSet) # 將每個元素轉會為frozenset字典 [frozenset({1}), frozenset({2}), frozenset({3}), frozenset({4}), frozenset({5})]
L1, supportData = scanD(D, C1, minSupport) # 過濾數據
L = [L1]
k = 2
while (len(L[k-2]) > 0): # 若仍有滿足支持度的集合則繼續做關聯分析
Ck = aprioriGen(L[k-2], k) # Ck候選頻繁項集
Lk, supK = scanD(D, Ck, minSupport) # Lk頻繁項集
supportData.update(supK) # 更新字典（把新出現的集合:支持度加入到supportData中）
L.append(Lk)
k += 1 # 每次新組合的元素都只增加了一個，所以k也+1（k表示元素個數）
return L, supportData
# 獲取關聯規則的封裝函數
def generateRules(L, supportData, minConf=0.7): # supportData 是一個字典
bigRuleList = []
for i in range(1, len(L)): # 從為2個元素的集合開始
for freqSet in L[i]:
# 只包含單個元素的集合列表
H1 = [frozenset([item]) for item in freqSet] # frozenset({2, 3}) 轉換為 [frozenset({2}), frozenset({3})]
# 如果集合元素大於2個，則需要處理才能獲得規則
if (i > 1):
rulesFromConseq(freqSet, H1, supportData, bigRuleList, minConf) # 集合元素 集合拆分後的列表 。。。
else:
calcConf(freqSet, H1, supportData, bigRuleList, minConf)
return bigRuleList
# 對規則進行評估 獲得滿足最小可信度的關聯規則
def calcConf(freqSet, H, supportData, brl, minConf=0.7):
prunedH = [] # 創建一個新的列表去返回
for conseq in H:
conf = supportData[freqSet]/supportData[freqSet-conseq] # 計算置信度
if conf >= minConf:
print(freqSet-conseq,'-->',conseq,'conf:',conf)
brl.append((freqSet-conseq, conseq, conf))
prunedH.append(conseq)
return prunedH
# 生成候選規則集合
def rulesFromConseq(freqSet, H, supportData, brl, minConf=0.7):
m = len(H[0])
if (len(freqSet) > (m + 1)): # 嘗試進一步合並
Hmp1 = aprioriGen(H, m+1) # 將單個集合元素兩兩合並
Hmp1 = calcConf(freqSet, Hmp1, supportData, brl, minConf)
if (len(Hmp1) > 1): #need at least two sets to merge
rulesFromConseq(freqSet, Hmp1, supportData, brl, minConf)
dataSet = loadDataSet()
L,suppData = apriori(dataSet,minSupport=0.5)
rules = generateRules(L,suppData,minConf=0.7)
# rules = generateRules(L,suppData,minConf=0.5)
print("具有強關聯規則的項集為：",rules)

擴展二：

https://github.com/apachecn/AiLearning/blob/master/src/py2.x/ml/11.Apriori/apriori.py