關聯分析（Apriori）詳解和python實現

時間 2019-12-02

標籤關聯分析 apriori 詳解 python 實現欄目 Python 简体版

原文原文鏈接

關聯分析

關聯關係是一種很是有用的數據挖掘算法，它能夠分析出數據內在的關聯關係。其中比較著名的是啤酒和尿不溼的案例python

交易號	清單
0	豆奶，萵苣
1	萵苣，尿布，啤酒，甜菜
2	豆奶，尿布，啤酒，橙汁
3	萵苣，豆奶，尿布，啤酒
4	萵苣，豆奶，尿布，橙汁

當超市在分析顧客的購物清單時發現一個比較奇怪的問題，爲何大部顧客在購買啤酒的時候還會買啤酒呢？後來通過超市的調查發現，顧客的妻子提醒丈夫買尿不溼時丈父會把本身的啤酒也一塊兒帶上。這是超市調查發現的尿不溼和啤酒的關係，若是數據量小咱們仍是能夠處理的，可是涉及到大數據時，其複雜度就很是高，那咱們有沒有其它方式去尋找這種關係呢？其實咱們可使用關聯算法去挖掘咱們商品之間的關聯關係，這些關係能夠有兩種形式：頻繁項集或者關聯規則。頻繁項集（frequent item sets)是常常出如今一塊的物品的集合（即常常被一塊兒購買的商品），關聯規則(association rules)暗示兩種物品之間可能存在很強的關係。算法

支持度和置信度

爲了有效定義頻繁和關聯，咱們引入兩個概念，支持度和置信度。數組

支持度(support)，即事件發生的頻率，記做 $support(\{A, B\}) = num(A\cup B) = P(A \cap B)$ ，例如一共有5條記錄，啤酒和尿布出現的次數是3次，這樣啤酒的支持度就是 3 / 5 = 0.6，支持度越大表格商品出現的次數就越多。網絡

置信度(confidence),置信度揭事了若是事件A發生了，則事件B發生的的機率，記做 $confidence(A \to B) = \frac{support(\{A ,B\})}{support(\{A\})}$ ，ide

例如啤酒和尿布被購買時，橙汁也一塊兒被購買的機率記做 $confidence(\{啤酒, 尿布\} \to \{橙汁\}) = \frac{support(\{\{啤酒, 尿布\} ,\{橙汁\}\})}{(\{橙汁\})} = \frac{\frac{1}{5}}{\frac{2}{5}} = \frac{1}{2}$ ，置信度的值表示事件A和B同時發生後的機率佔了A事件出現的機率的百份比，值越大代表買了啤酒和尿刷的顧客購買橙汁的機率比較大。oop

Apriori原理

由上面的介紹能夠知道，咱們須要計算全部組合的支持度和置信度，從而計算出全部可能被購買的頻繁項集這樣咱們就能夠對商品進行合理的佈局。佈局

支持度

下面爲了說明問題咱們先假設咱們有n種商品，那咱們全部的商品排列組合爲學習

$totalSupport = \sum_{k=2}^{n}C_{n}^{k} ({k<=n})（1）$

其中k表示頻繁集合中元素的個數，也就是說咱們要運算出全部的頻繁集合的話，時間複雜度爲，以下圖(來自網絡)例示咱們有四種商品0, 1, 2, 3，那咱們就有種頻繁集合，若是咱們有1000種頻繁集合的話，可想而知這個數據量是很是的龐大的。測試

置信度

置信度的計算方式是咱們遍歷全部的頻繁項集，而後找出每個項集的全部的子集再使用上面的公式來計算出全部子集的置信度，這些子集的意思就是那些商品最可能被一塊兒購買的商品組合，舉個例子若是咱們有頻繁項集{0, 2, 3}那它可能出現的全部的子集是{0, 2}, {0, 3}, {2, 3}, {0, 2, 3}。使用排列組合的知識咱們同理能夠得出:大數據

$totalNumOfConfidence = \sum_{i=2}^{n}C_{k}^{i} ({i<=k}) （2）$

個排列組合，根據(1), (2)兩個公式咱們能夠獲得計算全部子集的計算複雜度爲 $O\lgroup\sum_{k=2}^{n}\sum_{i=2}^{k}C_{k}^{i} \rgroup ({k<=n, i<=k})$ , 從公式能夠看出咱們若是要計算的時間複雜度很是高，咱們須要把計算置信度方式進行降維。

下面咱們先介紹Apriori的兩個定理：

定理1： 若是一個項集是頻繁的，那麼其全部的子集（subsets）也必定是頻繁的。這個比較容易證實，由於某項集的子集的支持度必定不小於該項集。

定理2： 若是一個項集是非頻繁的，那麼其全部的超集（supersets）也必定是非頻繁的。定理2是上一條定理的逆反定理。根據定理2，能夠對項集樹進行以下剪枝(下圖來自網絡)：

由於數據量大的話，生成頻集的計算量也是很是大的，而Apriori給出的兩個生成頻繁集項的方法：

1. $F_{k} = F_{k-1} * F_1$ 即第k-1項的全部頻繁集項和第一項的頻繁集項進行組合生成第k項候選頻繁集量,可是並非全部的結果都是知足需求的，咱們要設定最小頻繁項集閥值，只有大於閥值纔會轉正成爲頻繁項集。

2. $F_{k} = F_{k - 1} * F_{k - 1}$ ,選擇前項均相同的 $f_{k-2}$ 進行合併，生成 $F_{k−1}$ 頻繁集，這裏有個要求是 $F_{k−1}$ 必須是有序的，不然生成出來的項集並並不會符合要求。

算法實現

# -*- coding:gbk -*-
''' Created on 2018年2月12日 @author: Belle '''
from sklearn.feature_extraction import DictVectorizer
from dask.array.chunk import arange
import time;  # 引入time模塊
from apriori2 import apriori

SUPPORT_DIVIDER = ","

CONFIDENCE_DIVIDER = "=>"

''' 構建模型 '''
class Apriori():
    def __init__(self, dataSet, minSupport, minConfidence):
        self.vec = DictVectorizer()
        '''最小支持度'''
        self.minSupport = minSupport
        '''最小置信度'''
        self.minConfidence = minConfidence
        '''整個列表，數組的行表示單個特證向量，裏面的特證不重複，並且每一行的長度有可能不同'''
        self.dataSet = dataSet
        self.numOfTypes = len(dataSet)
        '''構建全部種類出現的次數'''
        self.dataTypeMap = {}
        '''初始化一項式'''
        self.dataTypeMap[1] = createTrainSet(self.dataSet)

''' 構學無監督學習的數據 '''
def createTrainSet(dataTypeMap):
    dataTypeMapResult = {}
    for row in range(len(dataTypeMap)):
        rowValues = dataTypeMap[row]
        rowValues.sort()
        for column in range(len(rowValues)):
            value = str(rowValues[column])
            if value in dataTypeMapResult:
                '''更新當前鍵出現的次數'''
                dataTypeMapResult[value] = dataTypeMapResult[value] + 1
            else:
                '''第一次出現的數據值爲1'''
                dataTypeMapResult[value] = 1
    return dataTypeMapResult

''' analize_x 爲n*k列距陣 '''
def analize(dataSet, minSupport = 0.15, minConfidence = 0.7):
    row = 2
    apriori = Apriori(dataSet, minSupport, minConfidence)
    '''從C(2, n), C(3, n)....到C(n, n)'''
    while True:
        if innerLoop(apriori, row) == 0:
            break
        row = row + 1
    '''生成關規則'''
    generateRule(apriori)
    return apriori
        

''' 計算經過k-1項，計算k項的數據 '''
def innerLoop(apriori, kSet):
    '''候選 k項式修選集'''
    kSetItems = {}
    beforeLenght = len(kSetItems)
    
    '''選擇k項式的值'''
    print("選擇{0}項式的值開始...".format(kSet))
    startTime = time.time()
    scanKMinusItems(kSetItems, apriori, kSet)
    print("獲取候選{0}項式時間：".format(kSet) + str(time.time() - startTime))
    
    '''對候選集進行剪枝'''
    print("剪枝開始,剪枝數量{0}...".format(len(kSetItems)))
    startTime = time.time()
    sliceBranch(kSetItems, apriori)
    print("剪枝花費時間:" + str(time.time() - startTime))
    '''存在下一個key_set,則放在結果中'''
    afterLength = len(kSetItems)
    if afterLength != beforeLenght:
        apriori.dataTypeMap[kSet] = kSetItems
        return 1
    else:
        return 0

''' 經過1項式和k-1項式生成k項式 '''
def scanKMinusItems(kSetItems, apriori, kSet):
    '''頻集1項式和k-1項式，組成新的k項式，而後把不知足的項式去掉'''
    '''頻集1項式'''
    keys = list(apriori.dataTypeMap[1].keys())
    
    '''k-1項式,,1項式和k-1項式組成k項式'''
    kMinusOneKeys = list(apriori.dataTypeMap[kSet - 1].keys())
    '''生成候選集'''
    for row in range(len(keys)):
        for column in range(len(kMinusOneKeys)):
            '''2項式時，因爲1項式和1項式進行組合，須要去除相同的組合數'''
            if kSet == 2 and row == column:
                continue
            calc(keys[row], kMinusOneKeys[column], kSetItems)

''' 生成候選頻繁集 @param oneDataSetKey: 1項式的key值 @param dataSet: 訓練集1項式 @param kMinusOneItemKey: k - 1項式的key值 @param kDataSetItems: k項式map數據 '''
def calc(oneDataSetKey, kMinusOneItemKey, kDataSetItems):
    if kMinusOneItemKey.find(oneDataSetKey) != -1:
        return
    kDataSetItemKeys = kMinusOneItemKey + SUPPORT_DIVIDER + str(oneDataSetKey)
    '''分割成數組,再進行排序'''
    kItemKeys = kDataSetItemKeys.split(SUPPORT_DIVIDER)
    kItemKeys.sort()
    '''list合成字段串'''
    kDataSetItemKeys = SUPPORT_DIVIDER.join(kItemKeys)
    '''kDataSetItemKeys初始值爲0'''
    if kDataSetItemKeys in kDataSetItems.keys():
        kDataSetItems[kDataSetItemKeys] += 1
    else:
        kDataSetItems[kDataSetItemKeys] = 0


''' 對後選頻煩集進行剪枝 @param kDataSetItems '''
def sliceBranch(kDataSetItems, apriori):
    kItemKeyArrays = list(kDataSetItems.keys())
    '''計算kItemKeys數組裏面的全部元素同時在dataSet中出現的次數'''
    dataSets = {}
    for kItemKeys in kItemKeyArrays:
        kItemKeyArray = kItemKeys.split(SUPPORT_DIVIDER)
        kDataSetItemCount = 0
        setData = set(kItemKeyArray)
        for rowIndex in range(len(apriori.dataSet)):
            if rowIndex in dataSets:
                rowArray = dataSets[rowIndex]
            else:
                rowArray = set(apriori.dataSet[rowIndex])
                dataSets[rowIndex] = rowArray
            '''長度大於數據長度'''
            if len(rowArray) < len(kItemKeyArray):
                continue
            '''判斷全部元素是否都在列表中同時存在'''
            if setData.issubset(set(rowArray)):
                kDataSetItemCount += 1
        '''小於最小支持度，則不添加到列表中'''
        if apriori.minSupport > kDataSetItemCount / apriori.numOfTypes:
            del kDataSetItems[kItemKeys]
        else:
            kDataSetItems[kItemKeys] = kDataSetItemCount

''' 計算置信度 @param kDataSetItems: 頻集數據集{1:{'1, 2, 3':次數}} @param apriori: 關聯數據類 '''
def generateRule(apriori):
    cacheKeySet = {}
    resultConfidence = {}
    '''key是頻集集合,value表明是K項式的k值'''
    for key in apriori.dataTypeMap:
        if key == 1:
            continue
        innerMap = apriori.dataTypeMap[key]
        for innerKey in innerMap:
            keyArray = innerKey.split(SUPPORT_DIVIDER)
            generateRule2(apriori, keyArray, innerMap[innerKey], resultConfidence, len(keyArray) - 1)

''' 目標繁集項和源繁集項兩兩結合在一塊兒 @param kDataSetItems: 二項式繁集項 @param targetItems: 某個目標繁集 @param sourceItems: 源繁集項 '''
def generateRule2(apriori, targetItems, supportTargetItems, resultConfidence, childIndex):
    if childIndex <= 0:
        return
    dataMap = apriori.dataTypeMap
    '''數據長度'''
    dataLength = len(targetItems)
    totalSets = set(targetItems)
    '''構造targetItems非空真子集,並計算至信度'''
    for index in range(dataLength):
        '''超過了數組長度'''
        if index + childIndex > dataLength:
            break
        '''從index開始取childIndex個數據表示是leftDataSet'''
        leftDataArray = targetItems[index:childIndex + index]
        leftDataArray.sort()
        '''取總列表與左邊的列表相減，就是右列key'''
        rightDataArray = list(totalSets ^ set(leftDataArray))
        rightDataArray.sort()
        
        leftDataKeyString = SUPPORT_DIVIDER.join(leftDataArray)
        rightDataKeyString = SUPPORT_DIVIDER.join(rightDataArray)
        
        '''不存在數量爲數組長度的頻集'''
        if (len(leftDataArray) not in dataMap) or (len(rightDataArray) not in dataMap):
            continue
        
        '''非頻集'''
        if (leftDataKeyString not in dataMap[len(leftDataArray)]) or \
            (rightDataKeyString not in dataMap[len(rightDataArray)]):
            continue
        
        '''leftDataKey出現的時,rightDataKeyString出現的機率,即頻集列表中兩個出現的數量'''
        confidence = supportTargetItems / \
            dataMap[len(leftDataArray)][leftDataKeyString]
        if confidence > apriori.minConfidence:
            '''置信度大於閥值'''
            print("{0}===>>{1}: confidence = {2}".format(leftDataKeyString, rightDataKeyString, str(confidence)))
            resultConfidence[leftDataKeyString + CONFIDENCE_DIVIDER + rightDataKeyString] = confidence
        else:
            '''置信度小於閥值,放在ignore例表中，用於下次判'''
    '''遞規的方式雲偏歷'''
    generateRule2(apriori, targetItems, supportTargetItems, resultConfidence, childIndex - 1)
        

複製代碼

測試代碼

# -*- coding:gbk -*-
''' Created on 2018年2月12日 @author: Belle '''
import Apriori

analize_x = [["豆奶", "萵苣"],\
             ["萵苣", "尿布", "啤酒", "甜菜"],\
             ["豆奶", "尿布", "啤酒", "橙汁"],\
             ["萵苣", "豆奶", "尿布", "啤酒"],\
             ["萵苣", "豆奶", "尿布", "橙汁"]]

apriori = Apriori.analize(analize_x)
print(apriori.dataTypeMap)

複製代碼

相關標籤/搜索

每日一句

每一个你不满意的现在，都有一个你没有努力的曾经。