贝叶斯分类器

发表于 | 分类于

贝叶斯决策论是在概率框架下实施决策的一种基本方法。对分类问题而言，朴素贝叶斯理论在大数定律和属性条件独立性假设的基础上，根据先验概率和类条件概率计算后验概率，从而得到新样本属于某一类的概率。

基本原理

基于贝叶斯定理，后验概率\(P(c|\vec{x})\)可以通过先验概率和条件概率计算得到： \[ P(c|\vec{x}) = \frac{P(c)P(\vec{x})|c}{p(\vec{x})} \] 其中\(P(c)\)是类先验概率（prior probability），\(P(\vec{x})|c)\)是样本\(\vec{x}\)相对于类标记\(c\)的类条件概率（class-conditional probability），或者“似然”（likelihood），\(P(\vec{x})\)是用于归一化的证据（evidence）因子。证据因子和类标记无关，因此估计后验概率P(c|)的问题转化为如何基于训练数据\(D\)来估计先验\(P(c)\)和似然\(P(\vec{x})|c)\)。
1. 类先验概率\(P(c)\)表达了样本空间中各类样本所占的比例。根据大数定律，当训练集包含充足的独立同分布样本时，\(P(c)\)可通过各类样本出现的频率来进行估计： \[ P(c) = \frac{|D_c|}{|D|} \] 其中\(D_c\)表示训练集\(D\)中第\(c\)类样本组成的集合，运算\(|D|\)表示集合\(D\)的样本个数。 2. 类条件概率\(P(\vec{x})|c)\)指的是属性\(\vec{x}\)的联合概率，直接根据样本出现的频率来估计将会遇到严重的困难。比如，样本共有\(d\)个属性，每个属性都是二值的，那么样本空间将有\(2^d\)种可能的取值，在现实应用中，这个值往往远大于训练样本个数\(m\)。如果各个属性是相互独立的，那么需要的样本数就可以减少到\(2*d\)。 \[ P(x_i|c) = \frac{|D_{c,x_i}|}{|D_c|} \] \[ P(\vec{x}|c) = \prod_{i=1}^{d} P(x_i|c) \] 3. “未被观测到”不等于“出现概率为0”，为此，需要进行平滑处理，常用“拉普拉斯修正”， \[ P(c) = \frac{|D_c|+1}{|D|+N} \] 其中\(N\)表示训练集\(D\)中的类别数； \[ P(x_i|c) = \frac{|D_{c,x_i}|+1}{|D_c|+N_i} \] 其中\(N_i\)表示第\(i\)个属性的取值数。 4. 多次连乘小的数容易导致下溢，为了避免这个问题，取连乘结果的对数： \[ P(\vec{x}|c) = \sum_{i=1}^{d} log(P(x_i|c)) \]

最终的朴素贝叶斯(Naive Bayes)分类器的表达式为： \[ h_{nb}(\vec{x}) = argmax_{c \in K} \left[log(P(c)) + \sum_{i=1}^{d} log(P(x_i|c)) \right] \]

算法实现

数据准备

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
def loadDataSet():
    """
    斑点犬爱好者留言板
    """
    postingList = [['my','dog','has','flea',\
                   'problem','help','please'],
                   ['maybe','not','take','him',\
                   'to','dog','park','stupid'],
                   ['my','dalmation','is','so','cute',\
                   'I','love','him'],
                   ['stop','posting','stupid','worthless','garbage'],
                   ['mr','licks','ate','my','steak','how',\
                   'to','stop','him'],
                   ['quit','buying','worthless','dog','food','stupid']]                   
    classVec = [0, 1, 0, 1, 0, 1] # 1代表侮辱性文字，0代表正常言论
    return postingList, classVec

def createVocabList(dataSet):
    """
    创建词汇表（每个单词只出现一次）
    """
    vocabSet = set([]) # set构造函数返回一个不重复的词表
    for document in dataSet:
        vocabSet = vocabSet | set(document) #求并集，也是按位“或”操作符
    return list(vocabSet)

def setOfWord2Vec(vocabList, inputSet):
    """
    文档转化为向量
    词集模型
    vocabList: 词汇表
    inputSet: 输入文档
    returnVec: 返回和词汇表长度相同，输入文本中出现词汇表中单词的地方置1，没有置0
    """
    returnVec = [0]*len(vocabList) # 创建一个所有元素都为0的序列
    for word in inputSet:
        if word in vocabList:
            returnVec[vocabList.index(word)] = 1 # 每个词只出现1次
        else:
            print("the word: %s is not in my vocabulary!" %word)
    return returnVec

def bagOfWord2Vec(vocabList, inputSet):
    """
    文档转化为向量
    词袋模型
    vocabList: 词汇表
    inputSet: 输入文档
    returnVec: 返回和词汇表长度相同，输入文本中出现词汇表中单词的地方置1，没有置0
    """
    returnVec = [0]*len(vocabList) # 创建一个所有元素都为0的序列
    for word in inputSet:
        if word in vocabList:
            returnVec[vocabList.index(word)] += 1 # 记录每个词出现的次数
    return returnVec
1
2
3
4
5
listOPosts, listClasses = loadDataSet()
myVocabList = createVocabList(listOPosts)
print(myVocabList)
print(setOfWord2Vec(myVocabList, listOPosts[0]))
print(setOfWord2Vec(myVocabList, listOPosts[3]))
['is', 'cute', 'food', 'him', 'so', 'posting', 'has', 'love', 'mr', 'dog', 'not', 'licks', 'take', 'I', 'stupid', 'to', 'my', 'help', 'flea', 'park', 'garbage', 'problem', 'maybe', 'how', 'please', 'steak', 'worthless', 'dalmation', 'quit', 'ate', 'stop', 'buying']
[0, 0, 0, 0, 0, 0, 1, 0, 0, 1, 0, 0, 0, 0, 0, 0, 1, 1, 1, 0, 0, 1, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0]
[0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 0, 0, 1, 0, 0, 0, 1, 0]

训练算法

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
from numpy import *
def trainNB0(trainMatrix, trainCategory):
    numTrainDocs = len(trainMatrix) # 训练样本数
    numWords = len(trainMatrix[0]) # 字典的词汇数量
    pAbusive = sum(trainCategory)/float(numTrainDocs) # 计算P(class=1)
    p0Num = ones(numWords); p1Num = ones(numWords) # 初始化条件概率p(x|c=0)和p(x|c=1)
    p0Denom = 2.0; p1Denom = 2.0 #N_i=2
    
    # 对训练样本循环
    for i in range(numTrainDocs):
        # 统计c=1
        if trainCategory[i] == 1:
            p1Num += trainMatrix[i] # 统计所有样本中(c=1)出现某个单词的次数，输出list
            p1Denom += sum(trainMatrix[i]) # 统计所有样本（c=1)总的单词数量
        # 统计c=0
        else:
            p0Num += trainMatrix[i]
            p0Denom += sum(trainMatrix[i])
            
    # 计算p(x|c=1)和p(x|c=0)
    p1Vect = log(p1Num/p1Denom)
    p0Vect = log(p0Num/p0Denom)
    
    return p0Vect, p1Vect, pAbusive
1
2
3
4
5
6
7
8
9
10
11
listOPosts, listClasses = loadDataSet()
myVocabList = createVocabList(listOPosts)
trainMat = []
# 每一行，将文本转成向量
for postinDoc in listOPosts:
    trainMat.append(setOfWord2Vec(myVocabList, postinDoc))
trainNB0(trainMat, listClasses)
p0V, p1V, pAb = trainNB0(trainMat, listClasses)
print(p0V)
print(p1V)
print(pAb)
[ 0.04166667  0.04166667  0.          0.08333333  0.04166667  0.
  0.04166667  0.04166667  0.04166667  0.04166667  0.          0.04166667
  0.          0.04166667  0.          0.04166667  0.125       0.04166667
  0.04166667  0.          0.          0.04166667  0.          0.04166667
  0.04166667  0.04166667  0.          0.04166667  0.          0.04166667
  0.04166667  0.        ]
[ 0.          0.          0.05263158  0.05263158  0.          0.05263158
  0.          0.          0.          0.10526316  0.05263158  0.
  0.05263158  0.          0.15789474  0.05263158  0.          0.          0.
  0.05263158  0.05263158  0.          0.05263158  0.          0.          0.
  0.10526316  0.          0.05263158  0.          0.05263158  0.05263158]
0.5

测试算法

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
def classifyNB(vec2Classify, p0Vec, p1Vec, pClass1):
    p1 = sum(vec2Classify * p1Vec) + log(pClass1)
    p0 = sum(vec2Classify * p0Vec) + log(1.0 - pClass1)
    if p1 > p0:
        return 1
    else:
        return 0
    
def testingNB():
    listOPosts, listClasses = loadDataSet()
    myVocabList = createVocabList(listOPosts)
    trainMat = []
    for postinDoc in listOPosts:
        trainMat.append(setOfWord2Vec(myVocabList, postinDoc))
    p0V, p1V, pAb = trainNB0(array(trainMat), array(listClasses))
    
    # The 1st example
    testEntry = ['love', 'my', 'dalmation']
    thisDoc = array(setOfWord2Vec(myVocabList, testEntry))
    print(testEntry, 'classified as: ', classifyNB(thisDoc, p0V, p1V, pAb))
    
    # The 2rd example
    testEntry = ['stupid', 'garbage']
    thisDoc = array(setOfWord2Vec(myVocabList, testEntry))
    print(testEntry, 'classified as: ', classifyNB(thisDoc, p0V, p1V, pAb))

实例：过滤垃圾邮件

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
def textParse(bigString):
    import re
    listOfTokens = re.split(r'\W*', bigString) # 正则表达式，其中分隔符是除单词、数字以外的任意字符串
    return [tok.lower() for tok in listOfTokens if len(tok) >2] # 去掉空字符串和少于2个字母的字符串

def spamTest():
    docList = []; classList = []; fullText = []
    
    # 15次循环，分别使用15封垃圾邮件和正常邮件，共50封
    for i in range(1, 16):
        # 使用25封垃圾邮件
        wordList = textParse(open('email/spam/%d.txt'%i).read())
        docList.append(wordList) # 生成
        fullText.extend(wordList) # 
        classList.append(1)
        
        wordList = textParse(open('email/ham/%d.txt' %i).read())
        docList.append(wordList)
        fullText.extend(wordList)
        classList.append(0)
        
    vocabList = createVocabList(docList) # 生成词汇表
    trainingSet = list(range(30)); testSet = []
    
    # 留存交叉验证！ 
    # 从训练数据集中随机提取10个样本作为测试集，并从训练数据集中删除该样本
    for i in range(10):
        randIndex = int(random.uniform(0, len(trainingSet))) # 从0～50中产生一个随机数
        testSet.append(trainingSet[randIndex])
        del(trainingSet[randIndex]) # 删除被测试集所选中的样本
    
    # 准备训练所需要的数据，剩下40个样本作为训练集
    trainMat = []; trainClasses = []
    for docIndex in trainingSet:
        trainMat.append(setOfWord2Vec(vocabList, docList[docIndex]))
        trainClasses.append(classList[docIndex])
        
    # 训练算法
    p0V, p1V, pSpam = trainNB0(array(trainMat), array(trainClasses))
    
    # 测试算法，并计算错误率
    errorCount = 0    
    for docIndex in testSet:
        wordVector = setOfWord2Vec(vocabList, docList[docIndex])
        if classifyNB(array(wordVector), p0V, p1V, pSpam) != classList[docIndex]:
            errorCount += 1
    print ('the error rate is: ', float(errorCount)/len(testSet))
1
spamTest()
the error rate is:  0.1


/home/seisinv/anaconda3/envs/tensorflow/lib/python3.5/re.py:203: FutureWarning: split() requires a non-empty pattern match.
  return _compile(pattern, flags).split(string, maxsplit)

结论与讨论

结论

优点：在数据较少的情况下仍然有效，可以处理多分类问题。
缺点：对输入数据的准备方式较为敏感

讨论

  1. 对连续属性可考虑概率密度函数。比如第\(i\)个属性\(c_i\)连续变化，假定\(x_i\)满足正态分布，即\(p(x_i|c) \sim N(\mu_{c,i}, \sigma^2_{c,i})\)，则可以通过最大似然法估计均值和方差，最终得到该属性的概率密度函数： \[ p(x_i|c) = \frac{1}{\sqrt{2\pi}\sigma} exp \left( -\frac{(x_i-\mu_{c,i})^2}{2\sigma^2_{c,i}} \right) \]
  2. 为了提高预测效率，对给定训练数据集，可将朴素贝叶斯分类器所涉及到的所有概率估值事先存储起来，在预测的时候只需要“查表”即可进行判断；若任务数据更替频繁，则可采用“懒惰学习”（lazy learning)方式，先不进行任何训练，待收到预测请求的时候，再根据当前数据集进行概率估计；若数据不断增加，则可在现有估值的基础上，仅对新增样本属性所涉及的概率估值进行计数修正即可，实现增量学习。
  3. 朴素贝叶斯假定属性各个属性是独立的，实际情况往往并不是这样，为此，学者们提出半朴素贝叶斯分类器，基本思想是考虑一部分属性间的相互依赖关系。有“独依赖估计”(One-Dependent Estimator, ODE), SPODE(Super-Parent ODE), TAN(Tree Augmented naive Bayes), AODE(Averaged One-Dependent Esimator),贝叶斯网（Bayesian Network)或信念网（Belief network）。
  4. 朴素贝叶斯分类器假设所有属性变量值都可以被观测到，即训练数据是完整的，但是现实应用中往往会遇到“不完整”的训练样本，此时需要采用EM(Expectation Maximization)算法分两步估计隐变量（latent variable）和模型参数。

参考资料

  1. Peter Harrington, Machine learning in action, Manning Publications, 2012
  2. 周志华，机器学习，清华大学出版社， 2016