一文学会朴素贝叶斯并且从头开始用 Python 实现朴素贝叶斯算法

朴素贝叶斯算法是简单并且有效的算法，而且应该是你尝试解决分类问题的第一个方法。

在这个教程中，你将会学习朴素贝叶斯算法，包括它如何工作并且如何从头开始用 Python 实现。

关于朴素贝叶斯

朴素贝叶斯是一种使用每一个属性属于每一个类别的直觉方法来做出预测。它是一个有监督的学习过程，你需要建模一个概率预测模型问题的时候，你应该想到的方法。

给定一个属性的值了之后某一个类别的概率叫做条件概率。通过累乘某一个类别的所有属性的条件概率，我们得到了一个数据实例属于某一个类别的概率。

为了做出一个预测，我们可以计算数据属于每一个类别的概率，然后选择最高概率的类别作为结果。

朴素贝叶斯经常被描述为使用类别数据，因为他非常容易描述并且计算。更加实用的算法版本支持数值属性，并且假设每一个数值属性是正态分布。再一次，这是一个很强的假设，但是仍然给出了健壮的结果。

预测糖尿病的发病

我们在这个教程中将会使用皮马印第安人糖尿病问题作为测试问题。

这个问题有768 个包含皮马印第安人医学细节的观察对象。这些记录描述了从患者身上获取的即时观测值，比如说他们的年龄，怀孕和验血次数。所有的患者都是年龄21 或者更大的女性。所有的属性都是数值类型的，并且属性之间的单位不一样。

每一个记录有个类别值表明这个病人在这次测量的五年内是否患糖尿病。

下面是从皮马印第安人数据集中拿出的一个样例，来对我们即将处理的数据有一些感觉。

数据下载：

6,148,72,35,0,33.6,0.627,50,1

1,85,66,29,0,26.6,0.351,31,0

8,183,64,0,0,23.3,0.672,32,1

1,89,66,23,94,28.1,0.167,21,0

0,137,40,35,168,43.1,2.288,33,1

朴素贝叶斯算法教程

这个教程被分为一下几个步骤

处理数据 从 CSV 中载入数据，并且拆分为训练和测试数据集汇总数据 总结这些训练集中的属性，然后我们可以计算概率，并且做出预测。预测一个 根据汇总的数据来生成一个预测批量预测 针对测试数据做出批量预测评估准备率 评估预测测试数据集类别额准确率，计算预测类别正确的个数占总数的比例聚合代码 将所有的代码片段整理为一个完整且独立的朴素贝叶斯算法实现处理数据

第一件我们需要做的事情就是载入我们的数据。数据在 CSV 格式中没有标题行或者注释。我们可以在 csv 模块中，使用 open 函数来打开文件和 使用 reader 函数来读取数据行。

我们也需要转换属性，从载入时的字符串格式到数值，这样我们才能继续处理它们。下面就是 loadCsv() 函数用来载入皮马印第安数据集。

import csvdef loadCsv(filename):	lines = csv.reader(open(filename, "rb"))	dataset = list(lines)	for i in range(len(dataset)):		dataset[i] = [float(x) for x in dataset[i]]	return dataset

我们可以测试这个函数通过载入皮马印第安数据集并且打印数据实例的个数。

filename = 'pima-indians-diabetes.data.csv'dataset = loadCsv(filename)print('Loaded data file {0} with {1} rows').format(filename, len(dataset))

运行这个测试，你应该看到如下输出：

Loaded data file pima-indians-diabetes.data.csv rows

下一步我们将会拆分训练集，然后朴素贝叶斯可以用来做出预测，还有测试机这样我们可以评估模型的准确率。我们需要随机的拆分训练集和测试集，比例为 67% 作为训练，33% 作为测试（这是一个很常见的训练集和测试集的拆分比例）。

接下来是 splitDataset() 函数，它将会按比例拆分一个给定的数据集。

import randomdef splitDataset(dataset, splitRatio):	trainSize = int(len(dataset) * splitRatio)	trainSet = []	copy = list(dataset)	while len(trainSet) < trainSize:		index = random.randrange(len(copy))		trainSet.append(copy.pop(index))	return [trainSet, copy]

我们可以通过定义一个玩具数据来测试这个函数，来拆分数据集为训练集和测试集，并且打印出来。

dataset = [[1], [2], [3], [4], [5]]splitRatio = 0.67train, test = splitDataset(dataset, splitRatio)print('Split {0} rows into train with {1} and test with {2}').format(len(dataset), train, test)

运行这个测试文件，你将会看到：

Split 5 rows into train with [[4], [3], [5]] and test with [[1], [2]]

朴素贝叶斯由训练集的数据摘要组成。这个数据摘要又用来预测。

训练集的摘要包括数据每一个属性的平均值和标准方差。例如，如果有两个类别，7 个数值属性，那么我们需要每一个属性的平均值和标准方差和类别结合在一起，那就是有 14 个属性的摘要。

我们可以将数据摘要的准备拆分为以下子任务：

基于类别分开数据计算平均值计算标准方差汇总整个数据基于类别汇总属性

基于类别分开数据

第一个任务是基于类别的值来将训练集分开，这样我们就可以计算每一个类别的统计值。我们可以通过为每一个类别创建一个字典来保存所有属于这个类别的列表，并且将整个数据集的实例排序为一个合适的列表。

def separateByClass(dataset):	separated = {}	for i in range(len(dataset)):		vector = dataset[i]		if (vector[-1] not in separated):			separated[vector[-1]] = []		separated[vector[-1]].append(vector)	return separated

你可以看到这个函数假设最后一个属性保存的是类别值。这个函数返回了一个词典，分别是类别和类别下的一系列数据。

我们可以使用一些样本数据来测试这个函数，如下所示：

dataset = [[1,20,1], [2,21,0], [3,22,1]]separated = separateByClass(dataset)print('Separated instances: {0}').format(separated)

运行这个测试，你将会看到如下所示：

Separated instances: {0: [[2, 21, 0]], 1: [[1, 20, 1], [3, 22, 1]]}

计算平均值

我们需要计算每一个类别下的每一个属性的平均值。平均值是数据的中心或者中心的趋势，然后我们在计算概率的时候，将会用它作为我们高斯分布的平均值。

我们也需要计算每一个类别下所有属性的标准方差。这个标准方差描述了数据变化的趋势，并且我们在计算概率时，将会用它作为高斯分布的标准差。

标准差是方差的平方根。方差计算为每一个属性的值与平均值的差的平方的平均值。我们使用的是N-1 方法，他在计算的时候会把属性值的数量减一。

def mean(numbers):	return sum(numbers)/float(len(numbers)) def stdev(numbers):	avg = mean(numbers)	variance = sum([pow(x-avg,2) for x in numbers])/float(len(numbers)-1)	return math.sqrt(variance)

我们可以通过计算 1到5 数字的平均值来测试这个方法：

numbers = [1,2,3,4,5]print('Summary of {0}: mean={1}, stdev={2}').format(numbers, mean(numbers), stdev(numbers))

运行这个测试，你可以看到如下结果。

Summary of [1, 2, 3, 4, 5]: mean=3.0, stdev=1.58113883008

汇总数据集

现在我们有工具来汇总数据集。对于给定的一列数据实例，我们可以计算每一个属性的平均值和标准差。

zip 函数将每一个属性的平均值和标准差组合起来了。

def summarize(dataset):	summaries = [(mean(attribute), stdev(attribute)) for attribute in zip(*dataset)]	del summaries[-1]	return summaries

我们可以通过一些测试数据来测试这个摘要函数，它表明第一个属性和第二个属性的平均值和标准差有非常显著的不同。

dataset = [[1,20,0], [2,21,1], [3,22,0]]summary = summarize(dataset)print('Attribute summaries: {0}').format(summary)

运行这个测试，你将会看到如下结果：

Attribute summaries: [(2.0, 1.0), (21.0, 1.0)]

基于类别来汇总数据

我们可以将这个流程梳理为，先将同一个类别的数据放在一起。然后计算每一个属性的摘要。

def summarizeByClass(dataset):	separated = separateByClass(dataset)	summaries = {}	for classValue, instances in separated.iteritems():		summaries[classValue] = summarize(instances)	return summaries

我们可以使用一组小测试集来测试 summarizeByClass() 函数

dataset = [[1,20,1], [2,21,0], [3,22,1], [4,22,0]]summary = summarizeByClass(dataset)print('Summary by class value: {0}').format(summary)

运行这个测试，你将会看到以下内容：

Summary by class value: {0: [(3.0, 1.4142135623730951), (21.5, 0.7071067811865476)], 1: [(2.0, 1.4142135623730951), (21.0, 1.4142135623730951)]}

我们现在已经准备好使用训练数据的摘要来做出预测了。做出预测需要计算给定数据在每一个类别下的概率，然后选择最大概率的类别作为预测值。

我们可以将它拆分为以下步骤：

计算高斯概率密度函数计算每一个类别的概率做出预测评估准确率

计算机高斯概率密度函数

我们可以用高斯函数来估计一个给定属性值的概率，提供这个属性从训练集中计算出来的平均值和方差即可。

给定属性从属性和类别中的汇总数据，结果就是给定某一个属性的值之后某一个类别的条件概率。

可以查看引用来了解更多高斯概率密度函数等式的细节。在数据摘要中我们已经知道了高斯分布的平均值和方差，而且知道了某一个属性值属于某一个类别的似然。

在 calculaeProbability() 函数中，我们可以先计算指数部分，然后计算主要的除法。这让我们将等式很好的放在两行中。

import mathdef calculateProbability(x, mean, stdev):	exponent = math.exp(-(math.pow(x-mean,2)/(2*math.pow(stdev,2))))	return (1 / (math.sqrt(2*math.pi) * stdev)) * exponent

我们可以用样本数据来测试这个公式。

x = 71.5mean = 73stdev = 6.2probability = calculateProbability(x, mean, stdev)print('Probability of belonging to this class: {0}').format(probability)

运行这个测试，你将会看到以下内容：

Probability of belonging to this class: 0.0624896575937

计算类别的概率

现在我们可以计算某一个属性的值属于每一个类别的概率了，我们可以结合所有的属性值的概率，最后得到整个数据属于每一个类别的概率。

我们通过连乘把所有的概率结合在一起，在下面的 calculateClassProbabilities() 中，给定数据的概率通过连乘所有的属性概率得到，结果是一个词典包含了属性值和对应的概率。

def calculateClassProbabilities(summaries, inputVector):	probabilities = {}	for classValue, classSummaries in summaries.iteritems():		probabilities[classValue] = 1		for i in range(len(classSummaries)):			mean, stdev = classSummaries[i]			x = inputVector[i]			probabilities[classValue] *= calculateProbability(x, mean, stdev)	return probabilities

我们可以测试一下 calculateClassProbabilities() 函数：

summaries = {0:[(1, 0.5)], 1:[(20, 5.0)]}inputVector = [1.1, '?']probabilities = calculateClassProbabilities(summaries, inputVector)print('Probabilities for each class: {0}').format(probabilities)

运行这个测试，你将会看到以下内容：

Probabilities for each class: {0: 0.7820853879509118, 1: 6.298736258150442e-05}

做出预测

现在可以计算一个数据实例属于每一个类别的概率了，我们可以寻找最大的概率，然后返回对应的类别。

predict() 函数

def predict(summaries, inputVector):	probabilities = calculateClassProbabilities(summaries, inputVector)	bestLabel, bestProb = None, -1	for classValue, probability in probabilities.iteritems():		if bestLabel is None or probability > bestProb:			bestProb = probability			bestLabel = classValue	return bestLabel

我们可以测试 predict() 函数：

summaries = {'A':[(1, 0.5)], 'B':[(20, 5.0)]}inputVector = [1.1, '?']result = predict(summaries, inputVector)print('Prediction: {0}').format(result)

运行这个测试，你将会看到以下内容：

Prediction: A

批量预测

最后，我们可以预估模型的准确率，通过预测每一个测试数据集。这 getPredictions() 将会返回每一个实例的类别。

def getPredictions(summaries, testSet):	predictions = []	for i in range(len(testSet)):		result = predict(summaries, testSet[i])		predictions.append(result)	return predictions

我们可以测试一下 getPredictions() 函数：

summaries = {'A':[(1, 0.5)], 'B':[(20, 5.0)]}testSet = [[1.1, '?'], [19.1, '?']]predictions = getPredictions(summaries, testSet)print('Predictions: {0}').format(predictions)

运行这个测试，你将会看到以下内容：

Predictions: ['A', 'B']

获得准确率

预测值可以和测试集中的类别值对比，然后准确率可以被算出来。getAccuracy() 函数将会计算准确率。

def getAccuracy(testSet, predictions):	correct = 0	for x in range(len(testSet)):		if testSet[x][-1] == predictions[x]:			correct += 1	return (correct/float(len(testSet))) * 100.0

我们可以用样本数据来测试 getAccuracy() 函数：

testSet = [[1,1,1,'a'], [2,2,2,'a'], [3,3,3,'b']]predictions = ['a', 'a', 'a']accuracy = getAccuracy(testSet, predictions)print('Accuracy: {0}').format(accuracy)

运行这个测试，你将会看到以下内容：

Accuracy: 66.6666666667

最后我们将以上代码整理起来。

接下来我们将会提供从头开始用 Python 实现的朴素贝叶斯代码。

# Example of Naive Bayes implemented from Scratch in Pythonimport csvimport randomimport mathdef loadCsv(filename):	lines = csv.reader(open(filename, "rb"))	dataset = list(lines)	for i in range(len(dataset)):		dataset[i] = [float(x) for x in dataset[i]]	return datasetdef splitDataset(dataset, splitRatio):	trainSize = int(len(dataset) * splitRatio)	trainSet = []	copy = list(dataset)	while len(trainSet) < trainSize:		index = random.randrange(len(copy))		trainSet.append(copy.pop(index))	return [trainSet, copy]def separateByClass(dataset):	separated = {}	for i in range(len(dataset)):		vector = dataset[i]		if (vector[-1] not in separated):			separated[vector[-1]] = []		separated[vector[-1]].append(vector)	return separateddef mean(numbers):	return sum(numbers)/float(len(numbers))def stdev(numbers):	avg = mean(numbers)	variance = sum([pow(x-avg,2) for x in numbers])/float(len(numbers)-1)	return math.sqrt(variance)def summarize(dataset):	summaries = [(mean(attribute), stdev(attribute)) for attribute in zip(*dataset)]	del summaries[-1]	return summariesdef summarizeByClass(dataset):	separated = separateByClass(dataset)	summaries = {}	for classValue, instances in separated.iteritems():		summaries[classValue] = summarize(instances)	return summariesdef calculateProbability(x, mean, stdev):	exponent = math.exp(-(math.pow(x-mean,2)/(2*math.pow(stdev,2))))	return (1 / (math.sqrt(2*math.pi) * stdev)) * exponentdef calculateClassProbabilities(summaries, inputVector):	probabilities = {}	for classValue, classSummaries in summaries.iteritems():		probabilities[classValue] = 1		for i in range(len(classSummaries)):			mean, stdev = classSummaries[i]			x = inputVector[i]			probabilities[classValue] *= calculateProbability(x, mean, stdev)	return probabilities			def predict(summaries, inputVector):	probabilities = calculateClassProbabilities(summaries, inputVector)	bestLabel, bestProb = None, -1	for classValue, probability in probabilities.iteritems():		if bestLabel is None or probability > bestProb:			bestProb = probability			bestLabel = classValue	return bestLabeldef getPredictions(summaries, testSet):	predictions = []	for i in range(len(testSet)):		result = predict(summaries, testSet[i])		predictions.append(result)	return predictionsdef getAccuracy(testSet, predictions):	correct = 0	for i in range(len(testSet)):		if testSet[i][-1] == predictions[i]:			correct += 1	return (correct/float(len(testSet))) * 100.0def main():	filename = 'pima-indians-diabetes.data.csv'	splitRatio = 0.67	dataset = loadCsv(filename)	trainingSet, testSet = splitDataset(dataset, splitRatio)	print('Split {0} rows into train={1} and test={2} rows').format(len(dataset), len(trainingSet), len(testSet))	# prepare model	summaries = summarizeByClass(trainingSet)	# test model	predictions = getPredictions(summaries, testSet)	accuracy = getAccuracy(testSet, predictions)	print('Accuracy: {0}%').format(accuracy)main()

运行这个测试，你将会看到以下内容：

Split 768 rows into train=514 and test=254 rowsAccuracy: 76.3779527559%