KNN算法总结-1.png
KNN算法总结-2.png
KNN算法总结-3.png
KNN算法总结-4.png
KNN算法总结-5.png
KNN算法总结-6.png
KNN算法总结-7.png
KNN算法总结-8.png
KNN算法总结-9.png
程序
main.py
# %%
import numpy as np
from Fisher import Fisher
import process_data as pd
import Eval
import time
import os
from Knn import KNN
import plot
flag=0#0usps,1=sonar
# To->main
if flag:
file_name = './Sonar.csv'
else:
file_name = './usps_3&8.csv'
### t0=time.time()
#
cross_validation_number = 8
class_number = 2
k=9
# dataset = dict()
def fff():
color=['r','b']
data = dict(x=[[], []], y=[[], []])
val_data = dict(x=[[], []], y=[[], []])
Pred_True = [0] * class_number
pred_num=[0,0]
L = [0, 0]
last_j=float('-inf')
datas, L[0], L[1], Lall = pd.read_datas(file_name)
data_block = pd.split_datas(datas, cross_validation_number)
i = 0
for cla, val in data_block:
# print(i)
i += 1
for j in range(len(cla)):
(data['x'][j], data['y'][j], val_data['x'][j], val_data['y'][j]) = (
cla[j][:, :-1], cla[j][:, -1:], val[j][:, :-1], val[j][:, -1:])
myfisher = Fisher(L[0], L[1], Lall, data)
W = myfisher.W_Direction() # 投影参数
J=Eval.simple_Fisher_Friterion(W,myfisher.between_class_scatter_matrix())#评估一下投影效果，并更新投影向量
if J>last_j:
last_j=J
W_great=W
W0 = myfisher.OneKey_W0(W) # 阈值
# %%
for j in range(class_number):
predict_y = myfisher.Pred_result(W, val_data['x'][j], W0)
# print(predict_y)
# 画图
plot.draw(predict_y,color[j])
pred_num[j]+=len(val_data['x'][j])
Pred_True[j] += Eval.Comp_as1(predict_y)
plot.show(str(i))
# print(time.time()-t0)
print(' 预测->> 第1类 第2类 全部')
for j in range(class_number):
print('实际第{0}类 {1} {2} {3}'.format(j+1,Pred_True[j],pred_num[j]-Pred_True[j],pred_num[j]))
print('mixed_accuracy：%f' % ((Pred_True[0]+pred_num[1]-Pred_True[1])/(pred_num[0]+pred_num[1])))
# print('评价函数最大的W保存在',os.path.abspath(Eval.w_filename))
# Eval.save_par(W_great)
pass
def kkk():
data = dict(x=[[], []], y=[[], []])
val_data = dict(x=[[], []], y=[[], []])
Pred_True = [0] * class_number
pred_num = [0, 0]
L = [0, 0]
datas, L[0], L[1], Lall = pd.read_datas(file_name)
data_block = pd.split_datas(datas, cross_validation_number)
for cla, val in data_block:
for j in range(len(cla)):
(data['x'][j], data['y'][j], val_data['x'][j], val_data['y'][j]) = (
cla[j][:, :-1], cla[j][:, -1:], val[j][:, :-1], val[j][:, -1:])
myknn=KNN(data,k)
for i in range(class_number):
predict_y = list()
for x in val_data['x'][i]:
predict_y.append(myknn.predict(x))
pred_num[i] += len(predict_y) # 预测总数量
Pred_True[i] += Eval.Comp_as1(predict_y)
pass
print(' 预测第1类 预测第2类 全部')
for j in range(class_number):
print('实际第{0}类 {1} {2} {3}'.format(j + 1, Pred_True[j], pred_num[j] - Pred_True[j], pred_num[j]))
print('mixed_accuracy：%f' % ((Pred_True[0] + pred_num[1] - Pred_True[1]) / (pred_num[0] + pred_num[1])))
if __name__ == '__main__':
fff()
# kkk()
pass
'''
sonar
预测 第1类 第2类 全部
实际第1类 63 33 96
实际第2类 36 72 108
mixed_accuracy：0.661765
usps_3&8
预测 第1类 第2类 全部
实际第1类 789 31 820
实际第2类 20 680 700
mixed_accuracy：0.966447
'''
'''
预测第1类 预测第2类 全部
实际第1类 44 46 90
实际第2类 43 67 110
mixed_accuracy：0.555000
预测第1类 预测第2类 全部
实际第1类 803 17 820
实际第2类 10 690 700
mixed_accuracy：0.982237
'''
Fisher.py
import numpy as np
x = [] # n*d
w = [] # d*1
y = [] # 0*n
x = np.array(x)
w = np.array(w)
y = np.array(y)
def within_class_scatter_matrix(x, m):
'''
:param x: n*d->n*d*1
:param m: d*1->1*d*1
:return:类內离散度矩阵 d*d
'''
x = x[:, :, np.newaxis]
m_d = len(m)
m = m[np.newaxis, :, :] # 1*d*1
n, d = np.shape(x)[0], np.shape(x)[1]
assert d == m_d
temp = x - m
temp_tran = np.transpose(temp, (0, 2, 1)) # n*1*d
# print(list(map(np.shape,[x-m,x_transpose-m_transpose])))
# 三维点乘变循环
s = []
for a, b in zip(temp, temp_tran):
s.append(np.dot(a, b))
s = np.array(s)
s = np.sum(s, 0)
assert s.shape == (d, d)
return s
def pooled_within_class_scatter_matrix(wcsm1, wcsm2, L1, L2, Lall):
'''
:param wcsm1: d*d
:param wcsm2:
:return: 总类內离散度矩阵 d*d
'''
return (L1 * wcsm1 + L2 * wcsm2) / Lall
def projective_mean(w, m):
'''
:param w: 投影方向 d*1
:param m: 类均值向量 d*1
:return:投影后的均值 0*0 scalar
'''
def pred_y(x, w):
'''
:param x: n*d
:param w: d*1
:return: 预测值 0*n
'''
return x.dot(w).ravel()
assert np.shape(w) == np.shape(m)
# print('w:',np.transpose(w).shape,'m',m.shape)
# print(pred_y(np.transpose(w),m)[0])
return pred_y(np.transpose(w), m)[0]
def projective_within_class_scatter_matrix(y, M):
'''
:param y: 投影后的样本0*n
:param M: 均值向量投影后 0*0
:return:投影后类內离散度0*0 scalar
'''
M = np.array([M]) # 0*1
return pow(np.linalg.norm(y - M), 2)
def projective_between_class_scatter_matrix(m1, m2):
'''
:param m1: 投影后均值
:param m2:
:return: 投影后类间离散度0*0 scalar
'''
return pow((m1 - m2), 2)
def w_direction(S_W, m1, m2):
'''
:param S_W:总类內离散度矩阵 d*d
:param m1: 类均值向量 d*1
:param m2:
:return: d*1 投影参数（大小无所谓）
'''
# print(S_W**-1)
# print()
return np.dot(np.linalg.inv(S_W), (m1 - m2))
def W0(m1, m2):
'''
不考虑先验概率
:param m1: scalar 投影后均值
:param m2:
:return: 阈值
'''
return -(m1 + m2) / 2
class Fisher(object):
def __init__(self, L1, L2, Lall, data):
self.L1 = L1
self.L2 = L2
self.Lall = Lall
self.M = [[], []] # 均值向量
self.M[0], self.M[1] = map(self.__class_mean_vector, [data['x'][0], data['x'][1]])
self.data_x=data['x']
def __class_mean_vector(self, x):
'''
:param x:n*d
:return:类均值向量 d*1
'''
return np.transpose(np.mean(x, 0, keepdims=True))
def Pred_result(self, w, x, w0):
'''
:param w:投影参数d*1
:param x: n*d
:param w0: 0*0
:return: 决策结果 (n,)
'''
w0 = np.array([w0])
temp = np.dot(x, w)
return temp.ravel() + w0
def W_Direction(self):
Sw1, Sw2 = within_class_scatter_matrix(self.data_x[0], self.M[0]), within_class_scatter_matrix(self.data_x[1], self.M[1])
SW = pooled_within_class_scatter_matrix(Sw1, Sw2, self.L1, self.L2, self.Lall)
W = w_direction(SW, self.M[0], self.M[1])
return W
def OneKey_W0(self, W):
m1, m2 = projective_mean(W, self.M[0]), projective_mean(W, self.M[1])
return W0(m1, m2)
def between_class_scatter_matrix(self):
'''
:param m1: 第一类均值向量 d*1
:param m2: 第二类均值向量
:return: 类间离散度矩阵 d*d
'''
m_ = self.M[0] - self.M[1]
return np.dot(m_, np.transpose(m_))
Knn.py
import numpy as np
class KNN(object):
def __init__(self,data,K):
'''
:param data: (n,d)
'''
self.data_x=np.concatenate((data['x'][0],data['x'][1]),0)
self.data_y=np.concatenate((data['y'][0],data['y'][1]),0)
self.k=K
#参与投票者必须小于全体投票者
assert self.k<=len(self.data_x)
pass
def predict(self,x):
'''
:param x:
:return:
'''
def candidate( m):
j = m # 工作索引
c = ballot[m] # 候选/被比较的候选众数
count = 1 # 计数器
while j < n and count > 0:
j += 1
# 如果现在的j使得列表越界，可以执行else的操作并退出循环
try:
ballot[j]
except IndexError:
count -= 1
continue
# 有一个一样计数器+1
if ballot[j] == c:
count += 1
# 不一样的抵消，计数器-1
else:
count -= 1
# 工作索引和长度一样，走完序列，返回当前剩余的候选众数
if j == n:
return c
else:
return candidate(j + 1)
pass
'''
:param x: (1,d)
:return:(k,)
'''
distance=np.linalg.norm((self.data_x-x),axis=1)
position=np.argsort(distance)[:self.k]
ballot=list()
for p in position:
ballot.append(self.data_y[p][0])
# print(ballot)
n=self.k
result=candidate(0)
return result
Eval.py
import json
import numpy as np
#To->Eval
w_filename='w.json'
def save_par(w):
with open(w_filename,'w') as fp:
json.dump(list(w.ravel()),fp)
def load_par():
with open(w_filename,'r') as fp:
return np.array(json.load(fp))[:,np.newaxis]
def Comp_as1(y):
# y=(y>0.0 if label>0 else y<=0.0)
if isinstance(y,list):
y=np.array(y)
y= y>0.0
return np.sum(y)
def simple_Fisher_Friterion(w,S_b):
'''
Fisher简化版的判别函数
:param w: 投影向量d*1
:param S_b: 类间离散度矩阵
:return:
'''
return np.dot(np.dot(np.transpose(w),S_b),w)[0][0]
plot.py
import matplotlib.pyplot as plt
from threading import Thread
from time import sleep
def draw(x,color):
plt.hist(x, bins=50, color=color)
def show(name):
def close(time=2):
sleep(time)
# plt.savefig('./img/'+name)
plt.close()
pass
thread1 = Thread(target=close, args=())
thread1.start()
plt.show()
pass
process_data.py
import logging
import numpy as np
import random
from main import flag
# To->process_data
#%%
if flag:
## ---sonar
label=('R','M')
file_size=(208,61)
else:
## ---usps
label=('3','8')
file_size=(9298,257)
label_dict={label[0]:1,label[1]:-1}
def save_38(data):
import csv
with open('usps_3&8.csv','w',newline='') as csvfile:
writer=csv.writer(csvfile)
writer.writerows(data)
print('写完了')
exit(0)
def read_datas(file_name):
'''
读取并处理文件
:param file_name:
:return:
'''
datas1,datas2=[],[]
with open(file_name,'r') as f:
datas=f.readlines()
#%%
for i in range(len(datas)):
datas[i]=datas[i][:-1]
datas[i]=datas[i].split(',')
# print(i)
datas[i][:-1]=list(map(float,datas[i][:-1]))
# print(datas[i][-1])
# try:
datas[i][-1]=label_dict[datas[i][-1]]#替换成数字
# except KeyError:
# continue
# print('数据大小',np.array(datas).shape)
# assert np.array(datas).shape==file_size
# 可以用转成数组处理
for data in datas:
# print(data[-1])
#分别添加到两个数据
if data[-1]==label_dict[label[0]]:
datas1.append(data)
elif data[-1]==label_dict[label[1]]:
datas2.append(data)
else:
# continue
print('这个标签 %s 未知，请看一下，在第%d行' % (data[-1],datas.index(data)))
# 可以用转成数组处理
#随机一下顺序
# save_38(datas1+datas2)
map(random.shuffle,[datas1,datas2])
L1,L2,Lall=len(datas1),len(datas2),file_size[0]
# assert (L1+L2)==Lall
print('——————数据读取完成！——————')
return ((datas1,datas2),L1,L2,Lall)
#%%
def split_datas(datas,n):
'''
验证集和训练集分开
:param n: n折交叉验证
:param datas: 打乱后的数据，
:return: 若n=10，第一类，第二类，交叉验证集20*61
'''
data_block_len=[len(data)//n for data in datas]
for i in range(n):
val,cla=[],[]
for j,data in enumerate(datas):
val_left,val_right=data_block_len[j]*(i),data_block_len[j]*(i+1)
val.append(np.array(data[val_left:val_right]))
cla.append(np.array(data[:val_left]+data[val_right:]))
yield cla,val
