基于朴素贝叶斯的垃圾邮件分类Python实现

背景

垃圾邮件的问题一直困扰着人们,传统的垃圾邮件分类的方法主要有"关键词法"和"校验码法"等,然而这两种方法效果并不理想。其中,如果使用的是“关键词”法,垃圾邮件中如果这个关键词被拆开则可能识别不了,比如,“中奖”如果被拆成“中 ~~~ 奖”可能会识别不了。后来,直到提出了使用“贝叶斯”的方法才使得垃圾邮件的分类达到一个较好的效果,而且随着邮件数目越来越多,贝叶斯分类的效果会更加好。

我们想采用的分类方法是通过多个词来判断是否为垃圾邮件,但这个概率难以估计,通过贝叶斯公式,可以转化为求垃圾邮件中这些词出现的概率。

贝叶斯公式的介绍

贝叶斯定理由英国数学家贝叶斯 ( Thomas Bayes 1702-1761 ) 发展,用来描述两个条件概率之间的关系,比如 P(A|B) 和 P(B|A)。

(1)条件概率公式

设A,B是两个事件,且P(B)>0,则在事件B发生的条件下,事件A发生的条件概率为:

P(A|B) = P(AB)/P(B)

(2)由条件概率公式得出乘法公式: P(AB)=P(A|B)P(B)=P(B|A)P(A)

P(A|B) = P(AB)/P(B)可变形为: 

可以这样来看贝叶斯公式: 

  • P(A) 称为”先验概率”,即在B事件发生之前,我们对A事件概率的一个判断。如:正常收到一封邮件,该邮件为垃圾邮件的概率就是“先验概率”
  • P(A|B)称为”后验概率”, 即在B事件发生之后,我们对A事件概率的重新评估。如:邮件中含有“中奖”这个词,该邮件为垃圾邮件的概率就是“后验概率”

(3)全概率公式:

全概率公式的意义在于,当直接计算P(A)较为困难,而P(Bi), P(A|Bi) (i=1,2,...)的计算较为简单时,可以利用全概率公式计算P(A)。

 下面给出《概率论与数理统计》浙大四版对于全概率公式和贝叶斯公式的详细讲述:

本次分类任务的主要思路:

分类标准:当 P(垃圾邮件|文字内容)> P(正常邮件|文字内容)时,我们认为该邮件为垃圾邮件,但是单凭单个词而做出判断误差肯定相当大,因此我们可以将所有的词一起进行联合判断。

这里假设:所有词语彼此之间是不相关的(严格说这个假设不成立;实际上各词语之间不可能完全没有相关性,但可以忽略)。

假如我们进行判断的词有“中奖”、“免费”、“无套路”,则需要判断P(垃圾邮件|中奖,免费,无套路)与P(正常|中奖,免费,无套路),使用贝叶斯公式,可以变为

 同理,P(正常|中奖,免费,无套路)可以变为:

因此,对P(垃圾邮件|中奖,免费,无套路)与P(正常|中奖,免费,无套路)的比较,只需要对分子进行对比。

但是如果对多个词的P(内容|正常/垃圾)进行相乘时,可能会因为某个词的概率很小从而导致最后的结果为0(超出计算机的精度),因此可以对P(内容|正常/垃圾)取自然对数,即ln P(内容|正常/垃圾)。

因此可以变为

 但是还有一个要注意的问题,如果某个词只出现在垃圾邮件中,而没有出现在正常的邮件中,这就会导致P(内容|正常)为0,从而导致整个分子都变为0,这会让上面的判别失效,为此,可以采取拉普拉斯平滑(laplace smoothing)。

拉普拉斯平滑(laplace smoothing)

主要的思想是对词的个数+1,对训练数据进行平滑处理。当训练样本很大时,每个词的个数+1造成的概率变化并不大,在误差允许的范围之内。

数据集的介绍:

本次实验中,所采用的数据集为Enron Email Dataset。该数据集已经对正常邮件和垃圾邮件进行了分类。由于该数据集是真实的电子邮件数据集,因此它包含真实的垃圾邮件,杀毒软件可能会对其中部分的邮件进行删除,因此进行本次实验时请暂时关闭杀毒软件。

代码实现:

1.需要导入的包:

import os
import re
import string
import math

2.数据的读入:

DATA_DIR = 'enron'
target_names = ['ham', 'spam']
def get_data(DATA_DIR):
    subfolders = ['enron%d' % i for i in range(1,7)]
    data = []
    target = []
    for subfolder in subfolders:
        # spam
        spam_files = os.listdir(os.path.join(DATA_DIR, subfolder, 'spam'))
        for spam_file in spam_files:
            with open(os.path.join(DATA_DIR, subfolder, 'spam', spam_file), encoding="latin-1") as f:
                data.append(f.read())
                target.append(1)
        # ham
        ham_files = os.listdir(os.path.join(DATA_DIR, subfolder, 'ham'))
        for ham_file in ham_files:
            with open(os.path.join(DATA_DIR, subfolder, 'ham', ham_file), encoding="latin-1") as f:
                data.append(f.read())
                target.append(0)
    return data, target

X, y = get_data(DATA_DIR)#读取数据

在这个代码段中,我们读入了所有邮件内容和标签,其中邮件内容存储在data中,标签存储在target当中,“1”表示为垃圾邮件,“0”表示为正常邮件。

3.数据的预处理

下面我们定义一个类对数据进行预处理

class SpamDetector_1(object):
    #清除标点符号
    def clean(self, s):
        translator = str.maketrans("", "", string.punctuation)
        return s.translate(translator)
    #将字符串标记为单词
    def tokenize(self, text):
        text = self.clean(text).lower()
        return re.split("W+", text)
    #计算某个单词出现的次数
    def get_word_counts(self, words):
        word_counts = {}
        for word in words:
            word_counts[word] = word_counts.get(word, 0.0) + 1.0
        return word_counts

4.数据处理阶段

在我们开始实际算法之前,我们需要做三件事:计算(对数)类先验,即计算P(垃圾邮件)和P(正常邮件);词汇表(即正常邮件和垃圾邮件中出现的所有单词,方便进行拉普拉斯平滑);垃圾邮件和非垃圾邮件的词频,即给定词在垃圾邮件和非垃圾邮件中出现的次数。

class SpamDetector_2(SpamDetector_1):
    # X:data,Y:target标签(垃圾邮件或正常邮件)
    def fit(self, X, Y):
        self.num_messages = {}
        self.log_class_priors = {}
        self.word_counts = {}
        # 建立一个集合存储所有出现的单词
        self.vocab = set()
        # 统计spam和ham邮件的个数
        self.num_messages['spam'] = sum(1 for label in Y if label == 1)
        self.num_messages['ham'] = sum(1 for label in Y if label == 0)

        # 计算先验概率,即所有的邮件中,垃圾邮件和正常邮件所占的比例
        self.log_class_priors['spam'] = math.log(
            self.num_messages['spam'] / (self.num_messages['spam'] + self.num_messages['ham']))
        self.log_class_priors['ham'] = math.log(
            self.num_messages['ham'] / (self.num_messages['spam'] + self.num_messages['ham']))

        self.word_counts['spam'] = {}
        self.word_counts['ham'] = {}

        for x, y in zip(X, Y):
            c = 'spam' if y == 1 else 'ham'
            # 构建一个字典存储单封邮件中的单词以及其个数
            counts = self.get_word_counts(self.tokenize(x))
            for word, count in counts.items():
                if word not in self.vocab:
                    self.vocab.add(word)#确保self.vocab中含有所有邮件中的单词
                # 下面语句是为了计算垃圾邮件和非垃圾邮件的词频,即给定词在垃圾邮件和非垃圾邮件中出现的次数。
                # c是0或1,垃圾邮件的标签
                if word not in self.word_counts[c]:
                    self.word_counts[c][word] = 0.0
                self.word_counts[c][word] += count

 可以利用下面的语句进行debug,判断是否运行正确,若正确,log_class_priors of spam应该为-0.6776,log_class_priors of ham应该为-0.7089。

我们选取了第100封之后的邮件作为训练集,前面一百封邮件作为测试集。

MNB = SpamDetector_2()
# 选取了第100封之后的邮件作为训练集,前面一百封邮件作为测试集
MNB.fit(X[100:], y[100:])

# print("log_class_priors of spam", MNB.log_class_priors['spam']) #-0.6776
# print("log_class_priors of ham", MNB.log_class_priors['ham']) #-0.7089

5.测试阶段

下面定义一个类 SpamDetector对测试集进行测试。主要的思路是对

 进行比较,从而判断是垃圾邮件还是正常邮件

class SpamDetector(SpamDetector_2):
    def predict(self, X):
        result = []
        flag_1 = 0
        # 遍历所有的测试集
        for x in X:
            counts = self.get_word_counts(self.tokenize(x))  # 生成可以记录单词以及该单词出现的次数的字典
            spam_score = 0
            ham_score = 0
            flag_2 = 0
            for word, _ in counts.items():
                if word not in self.vocab: continue

                #下面计算P(内容|垃圾邮件)和P(内容|正常邮件),所有的单词都要进行拉普拉斯平滑
                else:
                    # 该单词存在于正常邮件的训练集和垃圾邮件的训练集当中
                    if word in self.word_counts['spam'].keys() and word in self.word_counts['ham'].keys():
                        log_w_given_spam = math.log(
                            (self.word_counts['spam'][word] + 1) / (sum(self.word_counts['spam'].values()) + len(self.vocab)))
                        log_w_given_ham = math.log(
                            (self.word_counts['ham'][word] + 1) / (sum(self.word_counts['ham'].values()) + len(
                                self.vocab)))
                    # 该单词存在于垃圾邮件的训练集当中,但不存在于正常邮件的训练集当中
                    if word in self.word_counts['spam'].keys() and word not in self.word_counts['ham'].keys():
                        log_w_given_spam = math.log(
                            (self.word_counts['spam'][word] + 1) / (sum(self.word_counts['spam'].values()) + len(self.vocab)))
                        log_w_given_ham = math.log( 1 / (sum(self.word_counts['ham'].values()) + len(
                                self.vocab)))
                    # 该单词存在于正常邮件的训练集当中,但不存在于垃圾邮件的训练集当中
                    if word not in self.word_counts['spam'].keys() and word in self.word_counts['ham'].keys():
                        log_w_given_spam = math.log( 1 / (sum(self.word_counts['spam'].values()) + len(self.vocab)))
                        log_w_given_ham = math.log(
                            (self.word_counts['ham'][word] + 1) / (sum(self.word_counts['ham'].values()) + len(
                                self.vocab)))

                # 把计算到的P(内容|垃圾邮件)和P(内容|正常邮件)加起来
                spam_score += log_w_given_spam
                ham_score += log_w_given_ham

                flag_2 += 1

                # 最后,还要把先验加上去,即P(垃圾邮件)和P(正常邮件)
                spam_score += self.log_class_priors['spam']
                ham_score += self.log_class_priors['ham']

            # 最后进行预测,如果spam_score > ham_score则标志为1,即垃圾邮件
            if spam_score > ham_score:
                result.append(1)
            else:
                result.append(0)
                
            flag_1 += 1

        return result
MNB = SpamDetector()
MNB.fit(X[100:], y[100:])
pred = MNB.predict(X[:100])
true = y[:100]

accuracy = 0
for i in range(100):
    if pred[i] == true[i]:
        accuracy += 1
print(accuracy) # 0.98

测试集的分类的正确率达到98%,效果还是挺好的。

所有的代码:

import os
import re
import string
import math
DATA_DIR = 'enron'
target_names = ['ham', 'spam']
def get_data(DATA_DIR):
    subfolders = ['enron%d' % i for i in range(1,7)]
    data = []
    target = []
    for subfolder in subfolders:
        # spam
        spam_files = os.listdir(os.path.join(DATA_DIR, subfolder, 'spam'))
        for spam_file in spam_files:
            with open(os.path.join(DATA_DIR, subfolder, 'spam', spam_file), encoding="latin-1") as f:
                data.append(f.read())
                target.append(1)
        # ham
        ham_files = os.listdir(os.path.join(DATA_DIR, subfolder, 'ham'))
        for ham_file in ham_files:
            with open(os.path.join(DATA_DIR, subfolder, 'ham', ham_file), encoding="latin-1") as f:
                data.append(f.read())
                target.append(0)
    return data, target

X, y = get_data(DATA_DIR)

class SpamDetector_1(object):
    """Implementation of Naive Bayes for binary classification"""
    #清除空格
    def clean(self, s):
        translator = str.maketrans("", "", string.punctuation)
        return s.translate(translator)
    #分开每个单词
    def tokenize(self, text):
        text = self.clean(text).lower()
        return re.split("W+", text)
    #计算某个单词出现的次数
    def get_word_counts(self, words):
        word_counts = {}
        for word in words:
            word_counts[word] = word_counts.get(word, 0.0) + 1.0
        return word_counts

class SpamDetector_2(SpamDetector_1):
    # X:data,Y:target标签(垃圾邮件或正常邮件)
    def fit(self, X, Y):
        self.num_messages = {}
        self.log_class_priors = {}
        self.word_counts = {}
        # 建立一个集合存储所有出现的单词
        self.vocab = set()
        # 统计spam和ham邮件的个数
        self.num_messages['spam'] = sum(1 for label in Y if label == 1)
        self.num_messages['ham'] = sum(1 for label in Y if label == 0)

        # 计算先验概率,即所有的邮件中,垃圾邮件和正常邮件所占的比例
        self.log_class_priors['spam'] = math.log(
            self.num_messages['spam'] / (self.num_messages['spam'] + self.num_messages['ham']))
        self.log_class_priors['ham'] = math.log(
            self.num_messages['ham'] / (self.num_messages['spam'] + self.num_messages['ham']))

        self.word_counts['spam'] = {}
        self.word_counts['ham'] = {}

        for x, y in zip(X, Y):
            c = 'spam' if y == 1 else 'ham'
            # 构建一个字典存储单封邮件中的单词以及其个数
            counts = self.get_word_counts(self.tokenize(x))
            for word, count in counts.items():
                if word not in self.vocab:
                    self.vocab.add(word)#确保self.vocab中含有所有邮件中的单词
                # 下面语句是为了计算垃圾邮件和非垃圾邮件的词频,即给定词在垃圾邮件和非垃圾邮件中出现的次数。
                # c是0或1,垃圾邮件的标签
                if word not in self.word_counts[c]:
                    self.word_counts[c][word] = 0.0
                self.word_counts[c][word] += count

MNB = SpamDetector_2()
MNB.fit(X[100:], y[100:])

class SpamDetector(SpamDetector_2):
    def predict(self, X):
        result = []
        flag_1 = 0
        # 遍历所有的测试集
        for x in X:
            counts = self.get_word_counts(self.tokenize(x))  # 生成可以记录单词以及该单词出现的次数的字典
            spam_score = 0
            ham_score = 0
            flag_2 = 0
            for word, _ in counts.items():
                if word not in self.vocab: continue

                #下面计算P(内容|垃圾邮件)和P(内容|正常邮件),所有的单词都要进行拉普拉斯平滑
                else:
                    # 该单词存在于正常邮件的训练集和垃圾邮件的训练集当中
                    if word in self.word_counts['spam'].keys() and word in self.word_counts['ham'].keys():
                        log_w_given_spam = math.log(
                            (self.word_counts['spam'][word] + 1) / (sum(self.word_counts['spam'].values()) + len(self.vocab)))
                        log_w_given_ham = math.log(
                            (self.word_counts['ham'][word] + 1) / (sum(self.word_counts['ham'].values()) + len(
                                self.vocab)))
                    # 该单词存在于垃圾邮件的训练集当中,但不存在于正常邮件的训练集当中
                    if word in self.word_counts['spam'].keys() and word not in self.word_counts['ham'].keys():
                        log_w_given_spam = math.log(
                            (self.word_counts['spam'][word] + 1) / (sum(self.word_counts['spam'].values()) + len(self.vocab)))
                        log_w_given_ham = math.log( 1 / (sum(self.word_counts['ham'].values()) + len(
                                self.vocab)))
                    # 该单词存在于正常邮件的训练集当中,但不存在于垃圾邮件的训练集当中
                    if word not in self.word_counts['spam'].keys() and word in self.word_counts['ham'].keys():
                        log_w_given_spam = math.log( 1 / (sum(self.word_counts['spam'].values()) + len(self.vocab)))
                        log_w_given_ham = math.log(
                            (self.word_counts['ham'][word] + 1) / (sum(self.word_counts['ham'].values()) + len(
                                self.vocab)))

                # 把计算到的P(内容|垃圾邮件)和P(内容|正常邮件)加起来
                spam_score += log_w_given_spam
                ham_score += log_w_given_ham

                flag_2 += 1

                # 最后,还要把先验加上去,即P(垃圾邮件)和P(正常邮件)
                spam_score += self.log_class_priors['spam']
                ham_score += self.log_class_priors['ham']

            # 最后进行预测,如果spam_score > ham_score则标志为1,即垃圾邮件
            if spam_score > ham_score:
                result.append(1)
            else:
                result.append(0)

            flag_1 += 1

        return result

MNB = SpamDetector()
MNB.fit(X[100:], y[100:])
pred = MNB.predict(X[:100])
true = y[:100]

accuracy = 0
for i in range(100):
    if pred[i] == true[i]:
        accuracy += 1
print(accuracy) # 0.98

其他说明:

Enron Email Dataset数据集可以点击下面链接下载

链接:https://pan.baidu.com/s/1qYrIXxP4gaja19uHjrm1xA 
提取码:1234

下载后解压到.py文件的目录下即可使用

本图文内容来源于网友网络收集整理提供,作为学习参考使用,版权属于原作者。
THE END
分享
二维码
< <上一篇
下一篇>>