机器学习实战（Peter Harrington）阅读笔记（一）

from numpy import *

def createDataSet():
	group = array([[1.0,1.1],[1.0,1.0],[0,0],[0,0.1]])
	labels = ['A','A','B','B']
	return group, labels

def kNN(X, dataSet, labels, k):
	"""
	X: 要分类的样本
	dataSet： 数据集
	labels：标签
	k：k最邻近算法的参数
	"""
	m, n = dataSet.shape #样本数目和特征数
	# 计算样本与数据集之间的距离，采用欧拉距离
	d = np.sum( (np.tile(X,(m,1)) - dataSet)**2,1)**0.5
	# 对距离进行排序
	d = d.argsort()
	print(d)
	#对类别出现频次进行技术
	classCount={}
	for i in range(k):
		l = labels[d[i]]
		classCount[l] = classCount.get(l,0) + 1
	return sorted(classCount.items(),
		key = lambda asd:asd[1],
		reverse=True)[0][0]

def file2matrix(filename):
	"""将文件转化为训练样本矩阵和标签向量X、y"""
	fr = open(filename)
	lines = fr.readlines() #打开读取文件
	X = np.zeros((len(lines), 3)) #初始化样本矩阵和标签向量
	y = []
	for idx, line in enumerate(lines): #解析填充数据
		line = line.strip()
		xy = line.split('\t')
		X[idx,:] = xy[0:3]
		y.append(xy[-1])
	return X, np.array(y)

import matplotlib
import matplotlib.pyplot as plt
fig = plt.figure()
ax = fig.add_subplot(111)
# ax.scatter(X[:,1], X[:,2])
ax.scatter(X[:,1], X[:,2], 15.0*y, 15.0*y)
plt.xlabel('Percentage of Time Spent Playing Video Games')
plt.ylabel('Liters of Ice Cream Consumed Per Week')
plt.show()

def autoNorm(X):
	"""归一化特征值：newValue = (oldValue-min)/(max-min)"""
	minV = X.min(0)
	maxV = X.max(0)
	ranges = maxV - minV
	normX = np.zeros(X.shape)
	m = X.shape[0]
	normX = (X -minV) / ranges
	return normX, ranges, minV

def datingClassTest():
	X,y = file2matrix('datingTestSet2.txt')  #从文件中加载数据集
	normX, ranges, minV = autoNorm(X)
	m = normX.shape[0] # 样本数目
	tlen = int(m*0.1) # 测试集长度test length
	errCnt = 0
	for i in range(tlen):
		res = kNN(normX[i,:],normX[tlen:m,:],y[tlen:m],3)
		print ("分类器返回: %d, 正确答案是: %d" % (res, y[i]))
		if (res != y[i]): errCnt += 1
	print ("错误率为: %f" % (errCnt/tlen))
	print (errCnt)

def classifyPerson():
	resultList = ['一点也不', '有一点', '非常喜欢']
	percentTats = float(input(\
								  "玩视频游戏所耗时间百分比?"))
	ffMiles = float(input("每年获得的飞行常客里程数?"))
	iceCream = float(input("每周消费的冰淇淋公升数?"))
	datingDataMat, datingLabels = file2matrix('datingTestSet2.txt')
	normMat, ranges, minVals = autoNorm(datingDataMat)
	inArr = np.array([ffMiles, percentTats, iceCream, ])
	classifierResult = kNN((inArr - \
								  minVals)/ranges, normMat, datingLabels, 3)
	print ("你将可能喜欢这个人： %s" % resultList[classifierResult - 1])

def img2vector(filename):
	"""读取图片"""
	returnVect = np.zeros((1,1024))
	fr = open(filename)
	for i in range(32):
		lineStr = fr.readline()
		for j in range(32):
			returnVect[0,32*i+j] = int(lineStr[j])
	return returnVect


def handwritingClassTest():
	"""使用kNN算法对手写数字进行分类"""
	hwLabels = []
	trainingFileList = os.listdir('trainingDigits')  #加载训练集
	m = len(trainingFileList)
	trainingMat = np.zeros((m,1024))
	for i in range(m):
		fileNameStr = trainingFileList[i]
		fileStr = fileNameStr.split('.')[0]	 #take off .txt
		classNumStr = int(fileStr.split('_')[0])
		hwLabels.append(classNumStr)
		trainingMat[i,:] = img2vector('trainingDigits/%s' % fileNameStr)
	testFileList = os.listdir('testDigits')		#iterate through the test set
	errorCount = 0.0
	mTest = len(testFileList)
	for i in range(mTest):
		fileNameStr = testFileList[i]
		fileStr = fileNameStr.split('.')[0]	 #take off .txt
		classNumStr = int(fileStr.split('_')[0])
		vectorUnderTest = img2vector('testDigits/%s' % fileNameStr)
		classifierResult = kNN(vectorUnderTest, trainingMat, hwLabels, 3)
		print ("分类器返回: %d, 真正的答案: %d" % (classifierResult, classNumStr))
		if (classifierResult != classNumStr): errorCount += 1.0
	print ("\n出错总数: %d" % errorCount)
	print ("\n错误率: %f" % (errorCount/float(mTest)))

createBranch():
	If so return 类标签；
	Else
		寻找划分数据集的最好特征
		划分数据集
		创建分支节点
			for 每个划分的子集
				调用函数createBranch并增加返回结果到分支节点中
		return 分支节点

def entropy(y):
	"""计算信息熵"""
	m = y.shape[0]
	yCnt = {}
	for l in y:
		yCnt[l] = yCnt.get(l, 0) + 1
	ret = 0.0
	for k in yCnt:
		p = yCnt[k] / m
		ret -= p * math.log(p,2)
	return ret

def createDataSet():
	"""创建测试数据"""
	X = np.array([[1, 1],
				[1, 1],
				[1, 0],
				[0, 1],
				[0, 1]])
	y = np.array(['yes','yes','no','no','no'])
	featNames = ['no surfacing','flippers'] #各个特征的成员名字
	return X,y,featNames

def splitDataSet(X, y, axis, value):
	"""分割数据集，去除axis轴的特征，
	选取该特征为value的样本"""
	retDataSet = []
	retX, rety = [], []
	for i, x in enumerate(X):
		if x[axis] == value: # 选择符合要求的样本
			retX.append(np.append(x[:axis],x[axis+1:]))
			rety.append(y[i])
	return np.array(retX),np.array(rety)

def bestSplit(X,y):
	"""选择最好的划分，返回最好的划分的特征序号"""
	m, n = X.shape
	baseEntropy = entropy(y)
	bestFeature = -1 #特征序号初始化为-1
	bestInfoGain = 0.0 #最好的信息增益初始化为0.0
	for i in range(n):
		featSet = set(X[:,i])
		newEntropy = 0.0
		for val in featSet: #计算条件熵
			subX, suby = splitDataSet(X,y,i,val)
			p = subX.shape[0] / m
			newEntropy += p*entropy(suby)
		infoGain = baseEntropy - newEntropy #计算信息增益
		if(infoGain > bestInfoGain):
			bestInfoGain = infoGain
			bestFeature = i
	return bestFeature

def createDTree(X,y,featNames=None):
	"""递归构建决策树
	featName： python数组类型，长度为n，特征名列表"""
	if(featNames == None):
		featNames = ["feature%d"%i for i in range(X.shape[1])]
	if(np.sum(y == y[0]) == len(y)): # 类别完全相同则停止继续划分
		return y[0]
	if(X.shape[1] == 0): # 没有可以划分的特征
		return Counter(y).most_common()[0] #返回
	bestFeat = bestSplit(X,y) #最优划分的特征序号
	bestFeatName = featNames[bestFeat] #最优划分的特征名
	dtree = {bestFeatName:{}}
	#del(featNames[bestFeat]) #从参数列表中删除
	featSet = set(X[:,bestFeat]) #获取该特征的所有取值
	for val in featSet: #构造子节点
		subFeatNames = featNames[:bestFeat] + featNames[bestFeat+1:]
		dtree[bestFeatName][val] = createDTree(
			*splitDataSet(X,y,bestFeat,val),
			subFeatNames)
	return dtree

def classify(dtree,x, featNames = None):
	"""分类器"""
	if(featNames == None):
		featNames = ["feature%d"%i for i in range(X.shape[1])]
	featName = list(dtree.keys())[0] #获取特征名
	subTree = dtree[featName] #获取子树
	featIdx = featNames.index(featName) #获取当前特征名在x的索引位置
	for key in subTree.keys():
		if(x[featIdx] == key):
			if(type(subTree[key]).__name__=='dict'):
				y = classify(subTree[key],x,featNames)
			else: y = subTree[key]
	return y

def saveDTree(dtree,filename):
	"""持久化树"""
	import pickle
	fw = open(filename,'w')
	pickle.dump(inputTree,fw)
	fw.close()

def loadDTree(filename):
	"""加载树"""
	import pickle
	fr = open(filename)
	return pickle.load(fr)

fr = open('lenses.txt')
lenses = np.array([ l.strip().split('\t') for l in fr.readlines()])
X, y = lenses[:,0:-1], lenses[:,-1]
featNames = ['age', 'prescript', 'astigmatic', 'tearRate']
lensesTree = dtree.createDTree(X,y,featNames)
print(lensesTree)
treePlotter.createPlot(lensesTree)

import numpy as np
from collections import Counter

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']]
	y = np.array([0,1,0,1,0,1])	#1 侮辱性词汇, 0 没有
	return postingList, y


def createVocabList(dataSet):
	"""返回词典向量"""
	vocabSet = set()  #创建一个空集
	for document in dataSet:
		vocabSet = vocabSet | set(document) # 求并集
	return list(vocabSet)

def words2Vec(vocabList, words):
	"""将词汇列表转换为0,1向量"""
	returnVec = [0]*len(vocabList)
	for word in words:
		if word in vocabList:
			returnVec[vocabList.index(word)] = 1
		else: print ("单词: %s 不在我的词汇表中!" % word)
	return np.array(returnVec)

def words2Mat(vocabList, dataSet):
	return np.array([words2Vec(vocabList, words) for words in dataSet])

def trainNB(X,y):
	"""计算概率的一些项，返回：
	(p(x|y), p(y), 标签名)"""
	m, n = X.shape
	_y = Counter(y)
	for k in _y: _y[k] /= m
	p_y = np.array(list(_y.values())) # 计算p(y)
	labelNames = list(_y.keys()) # 对应多个的标签名
	p_x_y = np.zeros([len(labelNames), n])# 初始化p(x|y)
	for idx, l in enumerate(labelNames):
		subX = X[y==l]
		p_x_y[idx, :] += np.log((subX.sum(0) + 1)/ (subX.sum()+2.0))
	return p_x_y, p_y, labelNames

def classifyNB(x,p_x_y,p_y,labelNames):
	return labelNames[np.argmax(np.dot(p_x_y,x))]

def testingNB():
	dataSet, y = loadDataSet()
	vocabList = createVocabList(dataSet)
	X = words2Mat(vocabList, dataSet)
	p_x_y, p_y, labelNames = trainNB(X,y)
	testEntry = ['love', 'my', 'dalmation']
	thisDoc = np.array(words2Vec(vocabList, testEntry))
	print (testEntry,'被分为: ',classifyNB(thisDoc,p_x_y,p_y,labelNames))
	testEntry = ['stupid', 'garbage']
	thisDoc = np.array(words2Vec(vocabList, testEntry))
	print (testEntry,'被分为: ',classifyNB(thisDoc,p_x_y,p_y,labelNames))

def bagOfWords2Vec(vocabList, words):
	returnVec = [0]*len(vocabList)
	for word in words:
		if word in vocabList:
			returnVec[vocabList.index(word)] += 1
	return np.array(returnVec)

def bagOfWords2Mat(vocabList, dataSet):
	return np.array([bagOfWords2Vec(vocabList, words) for words in dataSet])

def textParse(bigString):	#input is big string, #output is word list
	import re
	listOfTokens = re.split(r'\W*', bigString)
	return [tok.lower() for tok in listOfTokens if len(tok) > 2]

def spamTest():
	docList=[]; classList = []; fullText =[]
	for i in range(1,26):
		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)#create vocabulary
	trainingSet = list(range(50)); testSet=[]		   #create test set
	for i in range(10):
		randIndex = int(np.random.uniform(0,len(trainingSet)))
		testSet.append(trainingSet[randIndex])
		del(trainingSet[randIndex])
	trainMat=[]; trainClasses = []
	for docIndex in trainingSet:#train the classifier (get probs) trainNB0
		trainMat.append(bagOfWords2Vec(vocabList, docList[docIndex]))
		trainClasses.append(classList[docIndex])
	p_x_y, p_y, labelNames = trainNB(np.array(trainMat),np.array(trainClasses))
	errorCount = 0
	for docIndex in testSet:		#classify the remaining items
		wordVector = bagOfWords2Vec(vocabList, docList[docIndex])
		if classifyNB(wordVector, p_x_y, p_y, labelNames) != classList[docIndex]:
			errorCount += 1
			print ("分类错误邮件为：",docList[docIndex])
	print ('错误率为: ',float(errorCount)/len(testSet))
	# return vocabList,fullText

class LogisticRegression:
	def __init__(self):
		pass

	def fit(self, X, y, alpha=0.001, epochs=500):
		X = np.column_stack((np.ones(X.shape[0]),X))
		y = y.reshape((y.shape[0],1))
		m, n = np.shape(X)
		self.W = np.ones((n,1))
		for k in range(epochs):			  #heavy on matrix operations
			h = sigmoid(np.matmul(X, self.W))	 #matrix mult
			error = (y - h)			  #vector subtraction
			self.W = self.W + alpha * np.matmul(error.transpose(), X).transpose() #matrix mult

	def fit1(self, X, y, epochs=150):
		X = np.column_stack((np.ones(X.shape[0]),X))
		m, n = np.shape(X)
		self.W = np.ones(n)   #initialize to all ones
		for j in range(epochs):
			dataIndex = list(range(m))
			for i in range(m):
				alpha = 4/(1.0+j+i)+0.0001	#apha decreases with iteration, does not
				randIndex = int(np.random.uniform(0,len(dataIndex)))#go to 0 because of the constant
				h = sigmoid(np.sum(X[randIndex]*self.W))
				error = y[randIndex] - h
				self.W = self.W + alpha * error * X[randIndex]
				del(dataIndex[randIndex])
		self.W = self.W[:,np.newaxis]


	def predict(self, x):
		x = np.array(x)
		if(x.ndim == 1): x = x[np.newaxis,:]
		x = np.column_stack((np.ones(x.shape[0]),x))
		p = sigmoid(np.matmul(x,self.W))
		return p > 0.5

class AdaBoost:
	def __init__(self):
		pass

	def __stumpClassify(self, X,featIdx,threshVal,threshIneq):
		"""单层决策树对一组数据进行分类
		参数说明：(X, 特征下标，边界值，正类的选择)"""
		retArray = np.ones((np.shape(X)[0],1)) #shape(m,1)
		if threshIneq == 'lt':
			retArray[X[:,featIdx] <= threshVal] = -1.0
		else:
			retArray[X[:,featIdx] > threshVal] = -1.0
		return retArray

	def __buildStump(self, X, y, D):
		"""构建最佳的单层决策树"""
		m, n = np.shape(X)

		numSteps = 10.0 #划分数据边界的分割步数
		bestStump = {} #记录单层决策树的结构
		bestClasEst = np.mat(np.zeros((m,1))) #纪录决策树的预测值
		minError = np.inf #初始化误差和为正无穷
		for i in range(n): # 遍历所有特征
			rangeMin = X[:,i].min() #获取该特征的上下界
			rangeMax = X[:,i].max()
			stepSize = (rangeMax-rangeMin)/numSteps
			for j in range(-1,int(numSteps)+1):# 遍历所有换分
				for inequal in ['lt', 'gt']: #决定哪个为正类
					threshVal = (rangeMin + float(j) * stepSize) #边界值
					predictedVals = self.__stumpClassify(X,i,threshVal,inequal)#call stump classify with i, j, lessThan
					errArr = np.ones((m,1)) #shape(m,1)
					errArr[predictedVals == y] = 0
					weightedError = np.matmul(D.T, errArr)  #对不同的样本给予不同的权重
					if weightedError < minError:
						minError = weightedError
						bestClasEst = predictedVals.copy()
						bestStump['dim'] = i #划痕的特征下表达式
						bestStump['thresh'] = threshVal #边界值
						bestStump['ineq'] = inequal #大于还是下雨边界值的为正类
		return bestStump,minError,bestClasEst

	def fit(self, X, y, numIt=40):
		"""X：np.array类型，数据可以为连续的
		y：python list类型，二元取值，-1.0或者1.0"""
		y = np.array(y)
		y = y.reshape((y.shape[0], 1))

		weakClassArr = [] #弱分类器列表
		m = np.shape(X)[0]
		D = np.ones((m,1))/m   #初始化D为等概率
		aggClassEst = np.zeros((m,1))
		for i in range(3):
			# 构建一个决策树
			bestStump,error,classEst = self.__buildStump(X,y,D)
			#print ("D:",D.T)
			# 计算alpha
			alpha = float(0.5*np.log((1.0-error)/max(error,1e-16)))
			bestStump['alpha'] = alpha  # 每个分类器的权重
			#储存决策树
			weakClassArr.append(bestStump)
			#print ("classEst: ",classEst.T)
			expon = np.multiply(-1*alpha*y,classEst) #exponent for D calc, getting messy
			# 为下一次迭代更新D
			D = np.multiply(D,np.exp(expon))
			D = D/D.sum()
			#calc training error of all classifiers, if this is 0 quit for loop early (use break)
			aggClassEst += alpha*classEst #对分类器的结果进行加权求和
			#print ("aggClassEst: ",aggClassEst.T)
			#获得最终结果
			aggErrors = np.multiply(np.sign(aggClassEst) != y, np.ones((m,1)))
			errorRate = aggErrors.sum()/m
			print ("total error: ",errorRate)
			if errorRate == 0.0: break
		self.weakClassArr = weakClassArr

	def predict(self,x):
		classifierArr = self.weakClassArr

		dataMatrix = np.mat(x).A #do stuff similar to last aggClassEst in adaBoostTrainDS
		m = np.shape(dataMatrix)[0]
		aggClassEst = np.zeros((m,1))
		for i in range(len(classifierArr)):
			classEst = self.__stumpClassify(dataMatrix, classifierArr[i]['dim'],\
									 classifierArr[i]['thresh'],\
									 classifierArr[i]['ineq'])#call stump classify
			aggClassEst += classifierArr[i]['alpha']*classEst
			print (aggClassEst)
		return np.sign(aggClassEst)

def standRegres(X,y):
	"""标准线性回归（普通最小二乘法)
	公式：\hat{w} = (X^TX)^(-1)X^Ty
	"""
	X = np.mat(X); y = np.mat(y).T
	xTx = X.T*X
	if np.linalg.det(xTx) == 0.0:
		print("This matrix is singular, cannot do inverse")
		return
	ws = xTx.I * (X.T*y)
	return ws

数据标准化，使其分布满足0均值和单位方差
在每轮迭代过程中：
	设置当前最小误差lowestError为正无穷
	对每个特征：
		增大或缩小：
			改变一个系数得到一个新的W
			计算新W下的误差
			如果误差Error小于当前最小误差lowestError：设置Wbest等于当前的W
		将W设置为新的Wbest

def regularize(xMat):#数据标准化
	inMat = xMat.copy()
	inMeans = np.mean(inMat,0)   #calc mean then subtract it off
	inVar = np.var(inMat,0)	  #calc variance of Xi then divide by it
	inMat = (inMat - inMeans)/inVar
	return inMat

def rssError(yArr,yHatArr): #yArr and yHatArr both need to be arrays
	return ((yArr-yHatArr)**2).sum()

def stageWise(xArr,yArr,eps=0.01,numIt=100):
	xMat = np.mat(xArr); yMat=np.mat(yArr).T
	yMean = np.mean(yMat,0)
	yMat = yMat - yMean	 #can also regularize ys but will get smaller coef
	xMat = regularize(xMat)
	m,n=np.shape(xMat)
	returnMat = np.zeros((numIt,n)) #testing code remove
	ws = np.zeros((n,1)); wsTest = ws.copy(); wsMax = ws.copy()
	for i in range(numIt):
		lowestError = np.inf;
		for j in range(n):
			for sign in [-1,1]:
				wsTest = ws.copy()
				wsTest[j] += eps*sign
				yTest = xMat*wsTest
				rssE = rssError(yMat.A,yTest.A)
				if rssE < lowestError:
					lowestError = rssE
					wsMax = wsTest
		ws = wsMax.copy()
		returnMat[i,:]=ws.T
	return returnMat

def isTree(obj):
	"""判断是否为一棵树"""
	return (type(obj)==dict)

def getMean(tree):
	"""该函数对树进行塌陷处理（即返回树平均值）"""
	if isTree(tree['right']): tree['right'] = getMean(tree['right'])
	if isTree(tree['left']): tree['left'] = getMean(tree['left'])
	return (tree['left']+tree['right'])/2.0

def prune(tree, testData):
	"""剪枝函数"""
	if np.shape(testData)[0] == 0:
		#若果没有测试数据，将整棵树（子树）进行塌陷处理器，
		#因为以下的划分都过拟合了，对于提高没有任何意义
		return getMean(tree)
	if (isTree(tree['right']) or isTree(tree['left'])):
		# 如果有一个孩子是一棵树，尝试分割测试集
		lSet, rSet = binSplitDataSet(testData, tree['spInd'], tree['spVal'])
	#递归调用
	if isTree(tree['left']):
		tree['left'] = prune(tree['left'], lSet)
	if isTree(tree['right']):
		tree['right'] =  prune(tree['right'], rSet)

	# 如果两个孩子是叶子节点，尝试是否可以合并
	if not isTree(tree['left']) and not isTree(tree['right']):
		lSet, rSet = binSplitDataSet(testData, tree['spInd'], tree['spVal'])
		# 没有合并的误差
		errorNoMerge = np.sum(np.power(lSet[:,-1] - tree['left'],2)) +\
			np.sum(np.power(rSet[:,-1] - tree['right'],2))
		# 整棵树的均值
		treeMean = (tree['left']+tree['right'])/2.0
		# 合并后的均值
		errorMerge = np.sum(np.power(testData[:,-1] - treeMean,2))
		if errorMerge < errorNoMerge:
			print ("merging")
			return treeMean
		else: return tree
	else: return tree