朴素贝叶斯算法--垃圾邮件过滤
文章目录
- 一、概述
- 第一部分 贝叶斯决策理论
-
第二部分 条件概率 -
第三部分 朴素贝叶斯 -
第四部分 朴素贝叶斯的一般流程
第二章 概述
-
- 数据预处理与特征提取: 基于文本构建词向量
-
- 模型训练与概率计算: 基于词向量的概率估算
-
- 模型评估与优化: 基于实际情况调整分类策略
-
- 垃圾邮件分类
一、朴素贝叶斯概述
1、贝叶斯决策理论
考虑到我们拥有的数据集分为两个类别构成,并令P₁(X)代表样本X属于类1的概率以及令P₂(X)代表样本X属于类2的概率那么对于一个新的待判别样本X我们可以基于以下准则进行分类:当样本X落入类别的概率高于另一类时则将其划分为类1;反之则划分为类2这正是贝叶斯决策理论的基本原则:总是选择概率最高的类别
2、条件概率
这个概念可以用公式来表示:P(A|B) = \frac{P(AB)}{P(B)}
3、朴素贝叶斯
贝叶斯定理为交换条件概率中的条件与结果提供了指导原则,并建议采用以下计算公式:P(A|B) = \frac{P(B|A)P(A)}{P(B)}

我们对A事件发生的初始概率进行评估,并将其定义为"先验概率"。
基于观察到的B事件结果后的更新后的概率则被称为"后验概率"(Posterior probability)。
通过计算P(B|A)/P(B),我们得到一个被称为"可能性函数"(Likelyhood)的关键指标;这个指标作为一个调整因子,在一定程度上反映了给定条件下观察到数据的可能性对比度。
基于这一假设模型的统计方法通常被称为朴素贝叶斯方法;该方法的核心在于假设各个特征之间相互独立。
4、朴素贝叶斯一般过程
⑴收集数据:可采用多种方式获取。
(2)准备数据:所需求解的问题必须基于数值型或布尔型数据。
(3)分析数据:当样本空间较为复杂时(即变量较多),单独评估每个变量的作用效果并不显著;而更适合采用直方图来进行多维度分布的可视化分析。
(4)训练算法:算法将被用来计算每个独立变量在给定条件下出现的概率。
(5)测试算法:通过计算模型预测结果与真实结果之间的差异来评估模型性能。
(6)使用算法:朴素贝叶斯方法的一个典型应用场景是文档分类任务;其优势在于能够有效地处理各类别问题,并不仅限于文本数据分析场景。
二、朴素贝叶斯算法–垃圾邮件
1、准备数据:从文本中构建词向量
将文本表示为单词或词条向量,并将每篇文档转换为相应的向量表示。具体步骤如下:首先通过分析所有文档中的关键词汇信息构建一个统一的词汇表集合;接着针对每一篇独立的文档进行词素化处理并将其映射到该词汇表中生成对应的二进制特征向量。整个过程涉及以下几个关键步骤:
通过调用loadDataSet()函数生成一系列实验样本数据集,并根据实际需求设定类别标签信息集合。该函数的主要作用是创建用于后续分析的基础数据结构框架。
调用createVocabList()函数完成对所有文档中出现过的不重复词语信息的统计汇总工作,并将其结果存储在一个有序排列的列表中作为统一的标准参考库使用
通过调用setOfWords2Vec()函数完成对特定文档特征信息的提取工作并生成相应的二进制特征向量表示结果;该过程的具体实现包括初始化一个全零初始状态下的固定长度特征空间模型,并根据输入文档中的实际出现词语情况动态更新模型中的相关位置标记值
# 函数loadDataSet() 创建了一些实验样本。该函数返回的第一个变量是进行词条切分后的文档集合,第二个变量为自定义的类别
def loadDataSet():
postingList=[['my', 'dog', 'has', 'flea', 'problems', '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]
return postingList, classVec
# 创建不重复词库列表
def createVocabList(dataSet):
vocabSet = set([]) #创建一个空集
for document in dataSet:
vocabSet = vocabSet | set(document) #创建两个集合的并集
return list(vocabSet)
# 输出文档向量
def setOfWords2Vec(vocabList, inputSet):
returnVec = [0]*len(vocabList) #创建一个元素都为0的向量
#遍历数据集单词
for word in inputSet:
#存在单词在词袋中则
if word in vocabList:
#index用于找到第一个与之匹配的下标
returnVec[vocabList.index(word)] = 1
else:
print ("the word: %s is not in my Vocabulary!" % word)
return returnVec
dataSet,classVec = loadDataSet()
vocabset = createVocabList(dataSet)
print(vocabset)
returnVec = setOfWords2Vec(vocabset,dataSet[0])
print(returnVec)
测试结果:可以看到把句子转化为了数字向量。

2、训练算法:从词向量计算概率
数据准备过程说明了如何将一组单词转换为一组数字,并探讨了如何利用这些数字来计算概率。假设当已知一个词是否出现在一篇文档中,并已知该文档所属的类别时,随后应用前述介绍的朴素贝叶斯公式来计算概率:P(x|C)表示给定类别C下某特征x出现的概率。

对每个类别都需要评估其特定值,并对这两个概率值进行分析比较。
具体而言,在统计学习过程中,
我们可以先通过类别1中出现的中文档数量除以总的文档数量来评估概率P(A)。
在此基础上,
我们需要运用朴素贝叶斯算法来计算条件概率P(B|A)。
通过引入条件独立性假设,
我们可以将上述联合概率表示为P(b1,b2,…bn|A)。
这一假设认为所有词之间都是相互独立的,
这使得条件独立性成为关键前提。
因此,在这种情况下就可以将联合概率分解为各个条件概率的乘积,
从而简化了整个概率计算过程。
from numpy import *
# 函数loadDataSet() 创建了一些实验样本。该函数返回的第一个变量是进行词条切分后的文档集合,第二个变量为自定义的类别
def loadDataSet():
postingList=[['my', 'dog', 'has', 'flea', 'problems', '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]
return postingList, classVec
# 创建不重复词库列表
def createVocabList(dataSet):
vocabSet = set([]) #创建一个空集
for document in dataSet:
vocabSet = vocabSet | set(document) #创建两个集合的并集
return list(vocabSet)
# 输出文档向量
def setOfWords2Vec(vocabList, inputSet):
returnVec = [0]*len(vocabList) #创建一个元素都为0的向量
#遍历数据集单词
for word in inputSet:
#存在单词在词袋中则
if word in vocabList:
#index用于找到第一个与之匹配的下标
returnVec[vocabList.index(word)] = 1
else:
print ("the word: %s is not in my Vocabulary!" % word)
return returnVec
# 训练算法,求概率
"""
该函数的伪代码如下:
计算每个类别中的文档数目
对每篇训练文档:
对每个类别:
如果词条出现文档中―增加该词条的计数值
增加所有词条的计数值
对每个类别:
对每个词条:
将该词条的数目除以总词条数目得到条件概率
返回每个类别的条件概率
"""
def trainNB0(trainMatrix,trainCategory):
numTrainDocs = len(trainMatrix)
numWords = len(trainMatrix[0])
pAbusive = sum(trainCategory)/float(numTrainDocs)
# 初始化概率
p0Num = zeros(numWords); p1Num = zeros(numWords)
p0Denom = 0.0; p1Denom = 0.0
# 遍历文档,向量相加
for i in range(numTrainDocs):
if trainCategory[i] == 1: # 侮辱类文档,向量相加
p1Num += trainMatrix[i]
p1Denom += sum(trainMatrix[i])
else: # 非侮辱类文档向量相加
p0Num += trainMatrix[i]
p0Denom += sum(trainMatrix[i])
p1Vect = p1Num/p1Denom # 各个单词在侮辱类中出现的概率
p0Vect = p0Num/p0Denom # 各个单词在非侮辱类中出现的概率
return p0Vect,p1Vect,pAbusive
dataSet, classVec = loadDataSet()
vocabset = createVocabList(dataSet)
trainMat = []
for postinDoc in dataSet:
trainMat.append(setOfWords2Vec(vocabset,postinDoc))
p0Vect, p1Vect, pAbusive = trainNB0(trainMat,classVec)
print(p0Vect) # 非侮辱类词向量概率
print(p1Vect) # 侮辱类词向量概率
print(pAbusive) #侮辱性文档的概率
两个类别的概率向量以及属于侮辱性文档的概率:

3、测试算法:根据现实情况修改分类器
问题一 :在应用贝叶斯分类器对文档进行分类时,在计算过程中需要求取多个条件概率的连乘积来确定文档属于某一类别的可能性大小, 即计算P(b₁|A) × P(b₂|A) × P(b₃|A) × … × P(bₙ|A) 。如果其中任何一个条件概率值为零,则该连乘积的结果也会是零。
解决 :为了减少这种情况下对结果的影响, 可以设定所有词汇的出现次数初始值均为1,并将分母初始值设为2。
p0Num = ones(numWords); p1Num = ones(numWords)
p0Denom = 2.0; p1Denom = 2.0
问题二 :当多个非常小的数值相乘时可能会导致结果溢出。具体原因在于多个非常小的数值相乘导致多个非常小的数值相乘可能会导致结果溢出或出现错误的结果。在实际应用中通常会对各条件概率取对数形式进行处理以避免计算过程中的数值下溢问题。
p1Vect = log(p1Num/p1Denom)
p0Vect = log(p0Num/p0Denom)
构建好完整的分类器后,进行测试分类。
from numpy import *
# 函数loadDataSet() 创建了一些实验样本。该函数返回的第一个变量是进行词条切分后的文档集合,第二个变量为自定义的类别
def loadDataSet():
postingList=[['my', 'dog', 'has', 'flea', 'problems', '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]
return postingList, classVec
# 创建不重复词库列表
def createVocabList(dataSet):
vocabSet = set([]) #创建一个空集
for document in dataSet:
vocabSet = vocabSet | set(document) #创建两个集合的并集
return list(vocabSet)
# 输出文档向量
def setOfWords2Vec(vocabList, inputSet):
returnVec = [0]*len(vocabList) #创建一个元素都为0的向量
#遍历数据集单词
for word in inputSet:
#存在单词在词袋中则
if word in vocabList:
#index用于找到第一个与之匹配的下标
returnVec[vocabList.index(word)] = 1
else:
print ("the word: %s is not in my Vocabulary!" % word)
return returnVec
def trainNB0(trainMatrix,trainCategory):
numTrainDocs = len(trainMatrix)
numWords = len(trainMatrix[0])
pAbusive = sum(trainCategory)/float(numTrainDocs)
p0Num = zeros(numWords); p1Num = zeros(numWords)
p0Denom = 0.0; p1Denom = 0.0
# 遍历文档,向量相加
for i in range(numTrainDocs):
if trainCategory[i] == 1: # 侮辱类文档,向量相加
p1Num += trainMatrix[i]
p1Denom += sum(trainMatrix[i])
else: # 非侮辱类文档向量相加
p0Num += trainMatrix[i]
p0Denom += sum(trainMatrix[i])
p1Vect = p1Num/p1Denom # 各个单词在侮辱类中出现的概率
p0Vect = p0Num/p0Denom # 各个单词在非侮辱类中出现的概率
return p0Vect,p1Vect,pAbusive
# 测试算法
def classifyNB(vec2Classify, p0Vec, p1Vec, pClass1):
#计算abusive的概率
p1 = sum(vec2Classify * p1Vec) + log(pClass1)
#计算not abusive概率
p0 = sum(vec2Classify * p0Vec) + log(1.0 - pClass1)
#看哪个概率大
if p1 > p0:
return 1
else:
return 0
def testingNB():
# test_list=[]; test_class=[0,1]
#加载数据集
listOPosts,listClasses = loadDataSet()
#创建词汇袋
myVocabList = createVocabList(listOPosts)
trainMat=[]
for postinDoc in listOPosts:
trainMat.append(setOfWords2Vec(myVocabList, postinDoc))
p0V,p1V,pAb = trainNB0(array(trainMat),array(listClasses))
testEntry = ['love', 'my', 'dalmation']
thisDoc = array(setOfWords2Vec(myVocabList, testEntry))
print (testEntry,'classified as: ',classifyNB(thisDoc,p0V,p1V,pAb))
testEntry = ['stupid', 'garbage']
thisDoc = array(setOfWords2Vec(myVocabList, testEntry))
print (testEntry,'classified as: ',classifyNB(thisDoc,p0V,p1V,pAb))
testingNB()
显示如下:

5、垃圾邮件分类
我们采用二进制特征来表示每个词汇的存在状态,并将其定义为"word set model"(WOSM)。这种表示方法被称为"Bag-of-Words Model"(BOWM)。在这种表示方法下,在word set model(WOSM)中允许同一个单词多次出现在向量表示里;而word set model(WOSM)中的每一个关键词只能以单一形式存在
from numpy import *
#统计列表中所有不重复的单词,词汇袋
def createVocabList(dataSet):
#定义词汇集
vocabSet = set([]) #set集合为不会有重复词的集合
#遍历数据集
for document in dataSet:
#把每个文档合并到词汇袋中
vocabSet = vocabSet | set(document) #union of the two sets
return list(vocabSet)
# 训练算法
def trainNB0(trainMatrix,trainCategory):
#计算文档的数目
numTrainDocs = len(trainMatrix)
#计算词袋单词的数目
numWords = len(trainMatrix[0])
#计算类别的概率 abusive的文档站文档总数的百分比
pAbusive = sum(trainCategory)/float(numTrainDocs)
#初始化计数器,1行*numWords列
#p0是not abusive p1是abusive
#为什么是不是0矩阵?
#因为如果是0的话,按照贝叶斯公式,p(W0|1)*p(W1|1)
#如果任意概率为0则全为0所以初始值为1
#同时下面分母初始化2
p0Num = ones(numWords); p1Num = ones(numWords)
#初始化分母
p0Denom = 2.0; p1Denom = 2.0
#遍历数据集中每个文档
for i in range(numTrainDocs):
#如果这文档是abusive类的计算它的abusive比例
if trainCategory[i] == 1:
#储存每个词在abusive下出现次数
p1Num += trainMatrix[i]
#储存abusive词的总数目
p1Denom += sum(trainMatrix[i])
else:
#储存每个词在not abusive的出现次数
p0Num += trainMatrix[i]
#储存not abusive词的总数目
p0Denom += sum(trainMatrix[i])
#使用log是为了防止数据向下溢出
# 计算abusive下每个词出现的概率
p1Vect = log(p1Num/p1Denom) #change to log()各个单词在侮辱类中出现的概率
# 计算not abusive下每个词出现的概率
p0Vect = log(p0Num/p0Denom) #change to log()各个单词在非侮辱类中出现的概率
#返回词出现的概率和文档为abusice的概率
return p0Vect,p1Vect,pAbusive
#贝叶斯分类
#第一个参数是要被训练的向量
#后面三个参数是根据上面一个函数处理训练样本后得到的参数
def classifyNB(vec2Classify, p0Vec, p1Vec, pClass1):
#计算abusive的概率
p1 = sum(vec2Classify * p1Vec) + log(pClass1)
#计算not abusive概率
p0 = sum(vec2Classify * p0Vec) + log(1.0 - pClass1)
#看哪个概率大
if p1 > p0:
return 1
else:
return 0
#计算文档单词在某个词袋中出现的次数
# 把单词转换成向量,用对比已经有得词袋,计算词出现的次数
def bagOfWords2VecMN(vocabList, inputSet):
returnVec = [0]*len(vocabList)
for word in inputSet:
if word in vocabList:
returnVec[vocabList.index(word)] += 1
return returnVec
# 输入字符串,输出单词列表
def textParse(bigString):
import re
listOfTokens = re.split(r'\W+', bigString)
return [tok.lower() for tok in listOfTokens if len(tok) > 2]
def spamTest():
# 定义docList文档列表,classList类别列表,fullText所有文档词汇
docList=[]; classList = []; fullText =[]
#遍历文件夹内文件
for i in range(1,26):
# 定义并读取垃圾邮件文件的词汇分割列表
wordList = textParse(open('email/spam/%d.txt' % i).read())
# 将词汇列表加到文档列表中
docList.append(wordList)
# 将所有词汇列表汇总到fullText中
fullText.extend(wordList)
# 文档类别为1,spam
classList.append(1)
# 读取非垃圾邮件的文档
wordList = textParse(open('email/ham/%d.txt' % i).read())
# 添加到文档列表中,注意append和extend的区别,append直接加列表,extend加元素
docList.append(wordList)
# 添加到所有词汇列表中
fullText.extend(wordList)
# 类别为0,非垃圾邮件
classList.append(0)
# 创建词汇列表
vocabList = createVocabList(docList)
# 定义训练集的索引和测试集
trainingSet = list(range(50)); testSet=[]
# 随机的选择10个作为测试集
for i in range(10):
#随机索引
randIndex = int(random.uniform(0,len(trainingSet)))
#将随机选择的文档加入测试集
testSet.append(trainingSet[randIndex])
#从训练集中删除随机选择的文档
del(trainingSet[randIndex])
#定义训练集的矩阵和类别
trainMat=[]; trainClasses = [];
# test_list=[];test_class=[]
#遍历训练集,求先验概率和条件概率
for docIndex in trainingSet:
#将词汇列表变成向量放到trainList中
trainMat.append(bagOfWords2VecMN(vocabList, docList[docIndex]))
#添加训练集的类标签
trainClasses.append(classList[docIndex])
#计算先验概率,条件概率
p0V,p1V,pSpam = trainNB0(array(trainMat),array(trainClasses))
errorCount = 0
#对测试集分类
for docIndex in testSet: #classify the remaining items
#将测试集向量化
wordVector = bagOfWords2VecMN(vocabList, docList[docIndex]) #bagOfWords2VecMN函数将词汇向量化和计算次数 setOfWords2Vec将词汇向量化
# 对测试数据进行分类
if classifyNB(array(wordVector),p0V,p1V,pSpam) != classList[docIndex]:
#错误的话错误计数加一
errorCount += 1
print( "classification error",docList[docIndex])
print ('the error rate is: ',float(errorCount)/len(testSet))
return vocabList,fullText
spamTest()

