机器学习(4):BP神经网络原理及其python实现

# # encoding=utf8
'''
Created on 2017-7-3 @author: Administrator
'''
import random
import pandas as pd
import numpy as np
from matplotlib import pyplot as plt
from sklearn.model_selection import train_test_split as ttsplit class LossFun:
def __init__(self, lf_type="least_square"):
self.name = "loss function"
self.type = lf_type def cal(self, t, z):
loss = 0
if self.type == "least_square":
loss = self.least_square(t, z)
return loss def cal_deriv(self, t, z):
delta = 0
if self.type == "least_square":
delta = self.least_square_deriv(t, z)
return delta def least_square(self, t, z):
zsize = z.shape
sample_num = zsize[1]
return np.sum(0.5 * (t - z) * (t - z) * t) / sample_num def least_square_deriv(self, t, z):
return z - t class ActivationFun:
'''
激活函数
'''
def __init__(self, atype="sigmoid"):
self.name = "activation function library"
self.type = atype; def cal(self, a):
z = 0
if self.type == "sigmoid":
z = self.sigmoid(a)
elif self.type == "relu":
z = self.relu(a)
return z def cal_deriv(self, a):
z = 0
if self.type == "sigmoid":
z = self.sigmoid_deriv(a)
elif self.type == "relu":
z = self.relu_deriv(a)
return z def sigmoid(self, a):
return 1 / (1 + np.exp(-a)) def sigmoid_deriv(self, a):
fa = self.sigmoid(a)
return fa * (1 - fa) def relu(self, a):
idx = a <= 0
a[idx] = 0.1 * a[idx]
return a # np.maximum(a, 0.0) def relu_deriv(self, a):
# print a
a[a > 0] = 1.0
a[a <= 0] = 0.1
# print a
return a class Layer:
'''
神经网络层
'''
def __init__(self, num_neural, af_type="sigmoid", learn_rate=0.5):
self.af_type = af_type # active function type
self.learn_rate = learn_rate
self.num_neural = num_neural
self.dim = None
self.W = None self.a = None
self.X = None
self.z = None
self.delta = None
self.theta = None
self.act_fun = ActivationFun(self.af_type) def fp(self, X):
'''
Foward Propagation
'''
self.X = X
xsize = X.shape
self.dim = xsize[0]
self.num = xsize[1] if self.W == None:
# self.W = np.random.random((self.dim, self.num_neural))-0.5
# self.W = np.random.uniform(-1,1,size=(self.dim,self.num_neural))
if(self.af_type == "sigmoid"):
self.W = np.random.normal(0, 1, size=(self.dim, self.num_neural)) / np.sqrt(self.num)
elif(self.af_type == "relu"):
self.W = np.random.normal(0, 1, size=(self.dim, self.num_neural)) * np.sqrt(2.0 / self.num)
if self.theta == None:
# self.theta = np.random.random((self.num_neural, 1))-0.5
# self.theta = np.random.uniform(-1,1,size=(self.num_neural,1)) if(self.af_type == "sigmoid"):
self.theta = np.random.normal(0, 1, size=(self.num_neural, 1)) / np.sqrt(self.num)
elif(self.af_type == "relu"):
self.theta = np.random.normal(0, 1, size=(self.num_neural, 1)) * np.sqrt(2.0 / self.num)
# calculate the foreward a
self.a = (self.W.T).dot(self.X)
###calculate the foreward z####
self.z = self.act_fun.cal(self.a)
return self.z def bp(self, delta):
'''
Back Propagation
'''
self.delta = delta * self.act_fun.cal_deriv(self.a)
self.theta = np.array([np.mean(self.theta - self.learn_rate * self.delta, 1)]).T # 求所有样本的theta均值
dW = self.X.dot(self.delta.T) / self.num
self.W = self.W - self.learn_rate * dW
delta_out = self.W.dot(self.delta);
return delta_out class BpNet:
'''
BP神经网络
'''
def __init__(self, net_struct, stop_crit, max_iter, batch_size=10):
self.name = "net work"
self.net_struct = net_struct
if len(self.net_struct) == 0:
print "no layer is specified!"
return self.stop_crit = stop_crit
self.max_iter = max_iter
self.batch_size = batch_size
self.layers = []
self.num_layers = 0;
# 创建网络
self.create_net(net_struct)
self.loss_fun = LossFun("least_square"); def create_net(self, net_struct):
'''
创建网络
'''
self.num_layers = len(net_struct)
for i in range(self.num_layers):
self.layers.append(Layer(net_struct[i][0], net_struct[i][1], net_struct[i][2])) def train(self, X, t, Xtest=None, ttest=None):
'''
训练网络
'''
eva_acc_list = []
eva_loss_list = [] xshape = X.shape;
num = xshape[0]
dim = xshape[1] for k in range(self.max_iter):
# i = random.randint(0,num-1)
idxs = random.sample(range(num), self.batch_size)
xi = np.array([X[idxs, :]]).T[:, :, 0]
ti = np.array([t[idxs, :]]).T[:, :, 0]
# 前向计算
zi = self.fp(xi) # 偏差计算
delta_i = self.loss_fun.cal_deriv(ti, zi) # 反馈计算
self.bp(delta_i) # 评估精度
if Xtest != None:
if k % 100 == 0:
[eva_acc, eva_loss] = self.test(Xtest, ttest)
eva_acc_list.append(eva_acc)
eva_loss_list.append(eva_loss)
print "%4d,%4f,%4f" % (k, eva_acc, eva_loss)
else:
print "%4d" % (k)
return [eva_acc_list, eva_loss_list] def test(self, X, t):
'''
测试模型精度
'''
xshape = X.shape;
num = xshape[0]
z = self.fp_eval(X.T)
t = t.T
est_pos = np.argmax(z, 0)
real_pos = np.argmax(t, 0)
corrct_count = np.sum(est_pos == real_pos)
acc = 1.0 * corrct_count / num
loss = self.loss_fun.cal(t, z)
# print "%4f,loss:%4f"%(loss)
return [acc, loss] def fp(self, X):
'''
前向计算
'''
z = X
for i in range(self.num_layers):
z = self.layers[i].fp(z)
return z def bp(self, delta):
'''
反馈计算
'''
z = delta
for i in range(self.num_layers - 1, -1, -1):
z = self.layers[i].bp(z)
return z def fp_eval(self, X):
'''
前向计算
'''
layers = self.layers
z = X
for i in range(self.num_layers):
z = layers[i].fp(z)
return z def z_score_normalization(x):
mu = np.mean(x)
sigma = np.std(x)
x = (x - mu) / sigma;
return x; def sigmoid(X, useStatus):
if useStatus:
return 1.0 / (1 + np.exp(-float(X)));
else:
return float(X); def plot_curve(data, title, lege, xlabel, ylabel):
num = len(data)
idx = range(num)
plt.plot(idx, data, color="r", linewidth=1) plt.xlabel(xlabel, fontsize="xx-large")
plt.ylabel(ylabel, fontsize="xx-large")
plt.title(title, fontsize="xx-large")
plt.legend([lege], fontsize="xx-large", loc='upper left');
plt.show() if __name__ == "__main__":
print ('This is main of module "bp_nn.py"') print("Import data")
raw_data = pd.read_csv('./train.csv', header=0)
data = raw_data.values
imgs = data[0::, 1::]
labels = data[::, 0]
train_features, test_features, train_labels, test_labels = ttsplit(
imgs, labels, test_size=0.33, random_state=23323) train_features = z_score_normalization(train_features)
test_features = z_score_normalization(test_features)
sample_num = train_labels.shape[0]
tr_labels = np.zeros([sample_num, 10])
for i in range(sample_num):
tr_labels[i][train_labels[i]] = 1 sample_num = test_labels.shape[0]
te_labels = np.zeros([sample_num, 10])
for i in range(sample_num):
te_labels[i][test_labels[i]] = 1 print train_features.shape
print tr_labels.shape
print test_features.shape
print te_labels.shape stop_crit = 100 # 停止
max_iter = 10000 # 最大迭代次数
batch_size = 100 # 每次训练的样本个数
net_struct = [[100, "relu", 0.01], [10, "sigmoid", 0.1]] # 网络结构[[batch_size,active function, learning rate]]
# net_struct = [[200,"sigmoid",0.5],[100,"sigmoid",0.5],[10,"sigmoid",0.5]] 网络结构[[batch_size,active function, learning rate]] bpNNCls = BpNet(net_struct, stop_crit, max_iter, batch_size);
# train model [acc, loss] = bpNNCls.train(train_features, tr_labels, test_features, te_labels)
# [acc, loss] = bpNNCls.train(train_features, tr_labels)
print("training model finished")
# create test data
plot_curve(acc, "Bp Network Accuracy", "accuracy", "iter", "Accuracy")
plot_curve(loss, "Bp Network Loss", "loss", "iter", "Loss") # test model
[acc, loss] = bpNNCls.test(test_features, te_labels);
print "test accuracy:%f" % (acc)