python实现KNN解析

KNN分类算法（K-Nearest-Neighbors Classification），又叫K近邻算法，是一个概念极其简单，而分类效果又很优秀的分类算法。

他的核心思想就是，要确定测试样本属于哪一类，就寻找所有训练样本中与该测试样本“距离”最近的前K个样本，然后看这K个样本大部分属于哪一类，那么就认为这个测试样本也属于哪一类。简单的说就是让最相似的K个样本来投票决定。

KNN算法简单有效，但没有优化的暴力法效率容易达到瓶颈。如样本个数为N，特征维度为D的时候，该算法时间复杂度呈O（DN)增长。

所以通常KNN的实现会把训练数据构建成K-D Tree（K-dimensional tree），构建过程很快，甚至不用计算D维欧氏距离，而搜索速度高达O（D*log（N））。

当然，KNN算法也存在一切问题。比如如果训练数据大部分都属于某一类，投票算法就有很大问题了。这时候就需要考虑设计每个投票者票的权重了。

KNN函数

#-*- coding:utf-8 -*-  #有中文时必须加上这一句
from math import pow
from collections import defaultdict  #关于集合的库
from multiprocessing import Process, cpu_count, Queue  #关于进程的库
i
4000
mport numpy as np

class Neighbor(object):
"""
一个结构体，用来描述一个邻居所属的类别和与该邻居的距离
"""

def __init__(self, class_label, distance):
"""
:param class_label: 类别(y).
:param distance: 距离.
初始化。
"""
self.class_label = class_label
self.distance = distance

class KNeighborClassifier(object):
"""
K-近邻算法分类器(k-Nearest Neighbor, KNN)，无KD树优化。
"""

def __init__(self, n_neighbors=5, metric='euclidean'):
"""
:param n_neighbors: 近邻数，默认为5.
:param metric: 测算距离采用的度量,默认为欧氏距离.
初始化。
"""
self.n_neighbors = n_neighbors

# p=2为欧氏距离，p=1为曼哈顿距离，其余的方式可自行添加。
if metric == 'euclidean':
self.p = 2
elif metric == 'manhattan':
self.p = 1

def fit(self, train_x, train_y):
"""
:param train_x: 训练集X.
:param trian_y: 训练集Y.
:return: None
接收训练参数
"""
self.train_x = train_x.astype(np.float32)
self.train_y = train_y

def predict_one(self, one_test):
'''
:param one_test: 测试集合的一个样本
:return: test_x的类别
预测单个样本
'''
# 用于储存所有样本点与测试点之间的距离
neighbors = []
for x, y in zip(self.train_x, self.train_y):
distance = self.get_distance(x, one_test)
neighbors.append(Neighbor(y, distance))

# 将邻居根据距离由小到大排序
'''sorted 和list.sort 都接受key, reverse定制。
但是区别是。list.sort()是列表中的方法，只能用于列表。而sorted可以用于任何可迭代的对象。
list.sort()是在原序列上进行修改，不会产生新的序列。所以如果你不需要旧的序列，可以选择
list.sort()。 sorted() 会返回一个新的序列。旧的对象依然存在。'''

neighbors.sort(key=lambda x: x.distance)

# 如果近邻值大于训练集的样本数，则用后者取代前者
if self.n_neighbors > len(self.train_x):
self.n_neighbors = len(self.train_x)

# 用于储存不同标签的近邻数
cls_count = defaultdict(int)

for i in range(self.n_neighbors):
cls_count[neighbors[i].class_label] += 1

# 返回结果
ans = max(cls_count, key=cls_count.get)
return ans

def predict(self, test_x):
'''
:param test_x: 测试集
:return: 测试集的预测值
预测一个测试集
'''
return np.array([self.predict_one(x) for x in test_x])

def get_distance(self, input, x):
"""
:param input: 训练集的一个样本.
:param x: 测试集合.
:return: 两点距离
工具方法，求两点之间的距离.
"""
if self.p == 2:
return np.linalg.norm(input - x)#计算矩阵范数
ans = 0
for i, t in zip(input, x):
ans += pow(abs(i - t), self.p)
return pow(ans, 1 / self.p)

class ParallelKNClassifier(KNeighborClassifier):
"""
并行K近邻算法分类器
"""

def __init__(self, n_neighbors=5, metric='euclidean'):
super(ParallelKNClassifier, self).__init__(n_neighbors, metric)
self.task_queue = Queue()
self.ans_queue = Queue()

def do_parallel_task(self):
'''
:return: None
单个进程的，并行任务。
进程不断从任务队列里取出测试样本，
计算完成后将参数放入答案队列
'''
while not self.task_queue.empty():
id, one = self.task_queue.get()
ans = self.predict_one(one)
self.ans_queue.put((id, ans))

def predict(self, test_x):
'''
:param test_x: 测试集
:return: 测试集的预测值
预测一个测试集
'''
for i, v in enumerate(test_x):
self.task_queue.put((i, v))

pool = []
for i in range(cpu_count()):
process = Process(target=self.do_parallel_task)
pool.append(process)
process.start()
for i in pool:
i.join()

ans = []
while not self.ans_queue.empty():
ans.append(self.ans_queue.get())

ans.sort(key=lambda x: x[0])
ans = np.array([i[1] for i in ans])
return ans

上面充分体现了python面向对象编程(Object Oriented Programming，简称OOP)思想。OOP把对象作为程序的基本单元，一个对象包含了数据和操作数据的函数。

面向过程的程序设计把计算机程序视为一系列的命令集合，即一组函数的顺序执行。为了简化程序设计，面向过程把函数继续切分为子函数，即把大块函数通过切割成小块函数来降低系统的复杂度。

而面向对象的程序设计把计算机程序视为一组对象的集合，而每个对象都可以接收其他对象发过来的消息，并处理这些消息，计算机程序的执行就是一系列消息在各个对象之间传递。

在Python中，所有数据类型都可以视为对象，当然也可以自定义对象。自定义的对象数据类型就是面向对象中的类（Class）的概念。

面向过程的例子(首先定义学生实例，然后调用函数打印学生成绩）

std1 = { 'name': 'Michael', 'score': 98 }
std2 = { 'name': 'Bob', 'score': 81 }
def print_score(std):
print '%s: %s' % (std['name'], std['score'])
print_score(std1)
print_score(std2)

面向对象的例子(首先必须创建出这个学生对应的对象，然后，给对象发一个print_score消息，让对象自己把自己的数据打印出来)

class Student(object):

def __init__(self, name, score):
self.name = name
self.score = score

def print_score(self):
print '%s: %s' % (self.name, self.score)
bart = Student('Bart Simpson', 59)
lisa = Student('Lisa Simpson', 87)
bart.print_score()
lisa.print_score()

画出边界utils函数

#-*- coding:utf-8 -*-
"""
工具包，包含了一些实用的函数。
"""

import numpy as np
import matplotlib.pyplot as plt

def plot_decision_boundary(pred_func, X, y):
'''
:param pred_func: predicet函数
:param X: 训练集X
:param y: 训练集Y
:return: None
分类器画图函数，可画出样本点和决策边界
'''

# Set min and max values and give it some padding
x_min, x_max = X[:, 0].min() - .5, X[:, 0].max() + .5
y_min, y_max = X[:, 1].min() - .5, X[:, 1].max() + .5
h = 0.8
# Generate a grid of points with distance h between them
xx, yy = np.meshgrid(np.arange(x_min, x_max, h), np.arange(y_min, y_max, h))
# Predict the function value for the whole gid
Z = pred_func(np.c_[xx.ravel(), yy.ravel()])
Z = Z.reshape(xx.shape)
# Plot the contour and training examples
plt.contourf(xx, yy, Z, cmap=plt.cm.Spectral)
plt.scatter(X[:, 0], X[:, 1], s=40, c=y, cmap=plt.cm.Spectral)
plt.show()

main函数，调用knn中的ParallelKNClassifier确定测试数据的类别标记，调用utils中的import plot_decision_boundary画图

import os
import sys

sys.path.insert(0, os.path.abspath('.'))

from mlearn.knn import ParallelKNClassifier
from mlearn.utils import plot_decision_boundary
import numpy as np

def main():
train_x = np.array([[1, 1], [0.1, 0.1], [0.5, 0.7], [10, 10], [10, 11]])
train_y = np.array(['A', 'A', 'A', 'B', 'B'])
test_x = np.array([[11, 12], [12, 13], [11, 13], [0.05, 0.1]])

k = ParallelKNClassifier(3)
k.fit(train_x, train_y)
print(k.predict(test_x))
import sklearn.datasets
np.random.seed(0)
X, y = sklearn.datasets.make_moons(200, noise=0.20)
clf = ParallelKNClassifier(3)
clf.fit(X, y)
testX,_ =sklearn.datasets.make_moons(10, noise=0.40)
a = clf.predict(testX)
print(a)
plot_decision_boundary(clf.predict, X, y)

if __name__ == '__main__':
main()