链接网址：http://blog.csdn.net/heyongluoyao8/article/details/48213345
BPNN
人工神经网络
我们知道，人的脑袋具有很强的学习、记忆、联想等功能，虽然人类还没有完全搞明白人类的大脑，但是我们已经知道它的基本单位就是一个个神经元，即一个神经细胞，人的神级细胞个数大约为1011个，海兔大约为2000多个，数目越多就越复杂，处理信息的能力就越强。如下图所示，每个神经元由细胞体与突起构成，细胞体的结构与一般的细胞相似，包括细胞膜，细胞质与细胞核，突起是神经元胞体的延伸部分，按照功能的不同分为树突与轴突。树突的功能是接受冲动并将冲动传入细胞体，即接受其它神经元传来的信号，并将信号传给神经元它是传入神经末梢。而轴突将本神经元的冲动传递给其它的与之相连的神经元，因此它是输出神经末梢。
大脑的神经元便是通过这样的连接进行信号传递来处理信息的，每个神经元通过其接受的信号来使得其兴奋或抑制。神经元内部信号的产生、传导采用电信号的方式进行，而神经元之间、神经元与肌肉之间则通常采用化学递质方式进行传导，即上级神经元的轴突在有电信号传递时释放出化学递质，作用于下一级神经元的树突，树突受到递质作用后产生出点信号，从而实现了神经元间的信息传递。而神经元的兴奋与抑制是由接受到的递质决定，有些递质起兴奋作用，有些起抑制作用。当传入神经元的冲动经整合，使细胞膜电位升高，超过动作电位的阈值时，为兴奋状态，产生神经冲动，由轴突神经末梢传出。当传入神经元的冲动经整合，使得细胞膜电位降低，低于阈值时，为抑制状态，不产生神经冲动。
大脑可通过自组织、自学习，不断适应外界环境的变化。大脑的自组织、自学习性，来源于神经网络结构的可塑性，主要反映在神经元之间连接强度的可变性上。由于神经元结构的可塑性，突触的传递作用可增强与减弱，因此神经元具有学习与遗忘功能。
人工神经网络起源于上世纪40～50年代，它是在基于人脑的基本单元－神经元的建模与联结，模拟人脑神经系统，形成一种具有学习、联想、记忆和模式识别等智能信息处理的人工系统，称为人工神经网络（Artificial Neural Networks，简称ANN），它是一种连接模型（Connection Model）。1943年神经生物学家MeCulloch与青年数学家Pitts合作，提出了第一个人工神经网络模型，并在此基础上抽象出神经元的数理模型（即神经元的阈值模型，超过阈值则兴奋，否则抑制），被称为MP模型（名字命名），这是ANN的启蒙阶段，ANN正式进入研究阶段。1958年Rosenblatt在原有的MP模型的基础上增加了学习机制，他提出的感知器模型，首次把神经网络理论付诸于工程实现，ANN研究迎来第一次高潮期。1969年出版的轰动一时的《Perceptrons》一书指出简单的线性感知器的功能是有限的，它无非解决线性不可分的而分类问题，如简单的线性感知器不能实现“异或”的逻辑关系，加上神经网络就和黑夹子一样，很多东西不透明，模型的解释性不强，参数过多，容易出错，容易过拟合，无法保证全局最优等问题，同时70年代集成电路和微电子技术的迅猛发展，使得传统的Von Neumenn计算机进入全盛时期，基于逻辑符号处理方法的人工智能得到了迅速发展并取得了显著的成果。ANN进入了长达10年的低潮期。1982年，美国科学院发表了著名的Hopfield网络模型的理论，不仅对ANN信息存储和提取功能进行了非线性数学概括，提出了动力方程和学习方程，使得ANN的构造与学习有了理论指导。这一研究激发了ANN的研究热情。1986年，由Rumelhat和McChekkand等16位作者参加撰写的《Parallel Distributed Processing: Exploration in Microstructures of Cognition》一书，建立了并行分布式处理理论，并提出了BP算法，解决了长期以来ANN中的权值调整问题没有有效解决方法的难题，可以求解感知器所不能解决的问题，回答了《Perceptrons》一书中关于神经网络局限性的问题，从实践中证实了ANN的有效性。BP算法产生了深远而广泛的影响，从至ANN的研究进入了蓬勃发展期。后来学者与研究者们又不断提出了RNN（Recurrent Neural Networks，递归神经网络）以及Deep Learning（深度学习）等模型。
'''
该网络中隐含层所有的偏置项使用在输入层增加一个节点（节点输入输出值恒为1.0来代替），而输出层没有偏置项
@author hy
'''
import numpy as np
'''
激活函数类
'''
class Sigmoid:
#sigmoid函数
def f(self,x):
return 1/(1+np.exp(-x)) #sigmoid函数的导数
def df(self,x):
y = self.f(x)
return y-np.multiply(y,y) class Tanh:
#双正切函数
def f(self,x):
return (np.exp(x)-np.exp(-x))/(np.exp(x)+np.exp(-x)) #双正切函数的导数
def df(self,x):
y = self.f(x)
return 1-np.multiply(y,y)
'''
工具类
'''
class Utils:
def uniform_distribution(self,x1,x2):
return np.random.uniform(x1,x2)
def create_random_matrix(self,row,col,periphery):
m = np.zeros((row,col))
for i in range(row):
for j in range(col):
m[i][j] = self.uniform_distribution(-1*periphery, periphery)
return m
def create_matrix(self,row,col,init_value=None):
m = np.zeros((row,col))
if init_value is not None:
for i in range(row):
for j in range(col):
m[i][j] = init_value
return m
def matrix_fill_zeros(self,m):
return np.zeros(np.shape(m))
'''
BP神经网络类
输入层、隐含层、输出层三层网络结构
'''
class BPNN:
#网络初始化
def __init__(self,input_node_num,hidden_node_num,output_node_num):
#输入层节点个数,增加一个偏置项
self.i_n = input_node_num+1
#隐含层层节点个数
self.h_n = hidden_node_num
#输出层节点个数
self.o_n = output_node_num self.utils = Utils() #输入层的输入(第i个样本),增加的偏置项节点的输入输出恒为1
self.i_i = self.utils.create_matrix(self.i_n,1,1.0)
#隐含层对第i个样本的输入
self.h_i = self.utils.create_matrix(self.h_n,1,0.0)
#输出层对第i个样本的输入
self.o_i = self.utils.create_matrix(self.o_n,1,0.0) #输入层的第i个样本输出,增加的偏置项节点的输入输出恒为1
self.i_o = self.utils.create_matrix(self.i_n,1,1.0)
#隐含层对于第i个样本的输出
self.h_o = self.utils.create_matrix(self.h_n,1,0.0)
#o_o是预测值，第i个样本的预测值
self.o_o = self.utils.create_matrix(self.o_n,1,0.0) #初始化连接权值矩阵
self.w_i_h = self.utils.create_random_matrix(self.i_n,self.h_n,0.2)
self.w_h_o = self.utils.create_random_matrix(self.h_n,self.o_n,2) #delta
self.o_delta = self.utils.create_matrix(self.o_n, 1,0.0)
self.h_delta = self.utils.create_matrix(self.h_n, 1,0.0) #delta w和，batch_size个样本的训练的和，也就是批量更新
self.w_h_o_delta = self.utils.create_matrix(self.h_n,self.o_n,0.0)
self.w_i_h_delta = self.utils.create_matrix(self.i_n,self.h_n,0.0) #训练
def train(self,hidden_activation,output_activation,train_inputs,train_outputs,alpha,error_threshold,iteration_num,batch_percent):
#隐含层的激活函数
self.h_activation = hidden_activation
#输出层的激活函数
self.o_activation = output_activation
#学习步长
self.alpha = alpha
#训练样本的个数
self.train_sample_n = np.shape(train_inputs)[0]
#这次迭代的总误差
self.train_error = 0.0
#误差阈值
self.error_threshold = error_threshold
#最大迭代次数
self.iteration_num = iteration_num
#每一次批量更新需要使用的样本个数
if batch_percent>100:
batch_percent = 100
self.batch_size = (int)(self.train_sample_n*batch_percent/100) #训练样本输入,矩阵,每一行为一个样本特征
self.train_inputs = train_inputs
#训练样本真实值,矩阵,每一行为一个样本的值
self.train_outputs = train_outputs #开始训练
self.batch() #测试
def test(self,test_inputs,test_outputs=None):
#测试样本个数
self.test_sample_n = np.shape(test_inputs)[0]
#预测集合
predict_outputs = self.utils.create_matrix(self.test_sample_n, self.o_n,0.0)
for i in range(self.test_sample_n):
#预测第i个测试样本
self.predict(test_inputs[i:i+1:,::])
#预测结果在self.o_o中
predict_outputs[i:i+1:,::] = np.transpose(self.o_o)
print "actural: ",test_outputs[i]
'''
print "predict values:"
print predict_outputs
#如果测试样本有结果，则输出真实结果以及预测总误差
if test_outputs is not None:
diff = test_outputs-predict_outputs
self.test_error = 0.5*np.sum(np.sum(np.multiply(diff,diff),axis=1),axis=0)[0,0]
print "actural values:"
print test_outputs
print "test error:"
print self.test_error
''' #预测一个样本
def predict(self,test_input):
#输入层的输出，即为其输入,i_nx1矩阵,因为有个偏置项，所有两个矩阵行数不一样，需要进行这样的赋值操作
self.i_o[0:self.i_n-1:,0:1:] = np.transpose(test_input[0:1:,::])
#计算隐含层每个节点的输出h_o
self.h_o = self.h_activation.f(np.transpose(self.w_i_h)*self.i_o) #计算输出层每个节点的输出o_o
self.o_o = self.o_activation.f(np.transpose(self.w_h_o)*self.h_o) print "predict: ",self.o_o #批量更新
def batch(self):
#一次batch使用的样本的编号集
inputs_indexs = [i for i in range(self.batch_size)]
#下次batch的需要使用的第一个样本的编号
last_index = (inputs_indexs[-1]+1)%self.train_sample_n
#批量更新，直到误差小于阈值或者达到最大迭代次数
while True:
#print "inputs_indexs:",inputs_indexs
self.one_batch(inputs_indexs)
#print "error: ",self.train_error
#剩余的迭代次数减1
self.iteration_num -= 1
#判断误差是否不再改变或者达到最大迭代次数
if self.terminate():
break
else:#否则继续迭代
#得到下次batch所需要使用的样本集所对应的编号集
'''
举例：训练样本集所对应的编号集是[0,1,2,3,4,5],样本个数为6，即train_sample_n＝6
如果batch_size=4，即每一次batch使用四个样本，
那么第一次使用的batch样本集所对应的编号集inputs_indexs=[0,1,2,3]
last_index = 4
第二次使用的batch样本集所对应的编号集inputs_indexs=[4,5,0,1]
即以后每次batch的inputs_indexs为：
for i in range(self.batch_size):
inputs_indexs.append((i+last_index)%self.train_sample_n)
'''
inputs_indexs = []
for i in range(self.batch_size):
inputs_indexs.append((i+last_index)%self.train_sample_n)
last_index = (inputs_indexs[-1]+1)%self.train_sample_n #一次batch
def one_batch(self,inputs_indexs):
#对每一个样本，首先使用前向传递，然后使用误差反向传播
for i in inputs_indexs:
#前向传递
self.fp(i)
#break
#反向传播
self.bp(i)
#更新权值
self.update() #第i个样本前向传递
def fp(self,i):
#输入层的输出，即为其输入,第i个样本, i_nx1矩阵
self.i_o[0:self.i_n-1:,0:1:] = np.transpose(self.train_inputs[i:i+1:,::]) #计算隐含层每个节点的输入h_i,以及隐含层的输出h_o
self.h_i = np.transpose(self.w_i_h)*np.matrix(self.i_o)
self.h_o = self.h_activation.f(self.h_i) #计算输出层每个节点的输入o_i,以及隐含层的输出o_o
self.o_i = np.transpose(self.w_h_o)*self.h_o
self.o_o = self.o_activation.f(self.o_i) #计算平方误差和
actural = np.transpose(self.train_outputs[i:i+1:,::])
tmp = self.o_o-actural
self.train_error = self.train_error + (np.transpose(tmp)*tmp)[0,0] #第i个样本误差反向传播
def bp(self,i): #对输出层每一个节点，计算\Delta_{ij}^{T-1}=(y_{ij}-\hat{y}_{ij}) \cdot f'^{T}(v)|_{v=I_{ij}^{T}}
self.o_delta = np.multiply((self.o_o-np.transpose(self.train_outputs[i:i+1:,::])),
self.o_activation.df(self.o_i))
#print "self.o_delta:",self.o_delta
#使用公式\frac{1}{p}\sum_{i=1}^{p}\Delta_{ij}^{T-1} \cdot o^{T-1}_{ik}计算\Delta {w_{kj}^{T-1}}
#前面的系数frac{1}{p}还没乘
self.w_h_o_delta = self.w_h_o_delta + self.h_o*np.transpose(self.o_delta)
#print "self.w_h_o_delta:",self.w_h_o_delta #对隐含层每一个节点，计算\Delta_{ij}^{T-2}=\sum_{s=1}^{s_{T}}(y_{is}-\hat{y}_{is}) \cdot f'^{T}(v)|_{v=I_{is}^{T}}\cdot w_{js}^{T-1} \cdot f'^{T-1}(v)|_{v=I_{ij}^{T-1}}
self.h_delta = np.multiply(self.w_h_o*np.multiply((self.o_o-np.transpose(self.train_outputs[i:i+1:,::])),self.o_activation.df(self.o_i)),self.h_activation.df(self.h_i))
#print "self.h_delta:",self.h_delta
#使用公式\frac{1}{p}\sum_{i=1}^{p}\Delta_{ij}^{T-2}\cdot o^{T-2}_{ik}计算\Delta {w_{kj}^{T-2}}
#前面的系数frac{1}{p}还没乘
self.w_i_h_delta = self.w_i_h_delta + self.i_o*np.transpose(self.h_delta)
#print "self.w_i_h_delta:",self.w_i_h_delta #更新权值W
def update(self):
#按照公式更新输入层与隐含层之间的w: w^{T-2}_{kj}:=w^{T-2}_{kj}-\alpha\frac{1}{p}\sum_{i=1}^{p}\Delta_{ij}^{T-2}\cdot o^{T-2}_{ik}
self.w_i_h = self.w_i_h - self.alpha/self.batch_size*self.w_i_h_delta
#按照公式更新隐含层与输出层之间的w: w^{T-1}_{kj}:=w^{T-1}_{kj}-\alpha\frac{1}{p}\sum_{i=1}^{p}\Delta_{ij}^{T-1} \cdot o^{T-1}_{ik}
self.w_h_o = self.w_h_o - self.alpha/self.batch_size*self.w_h_o_delta
#delta清零
self.w_i_h_delta = Utils().matrix_fill_zeros(self.w_i_h_delta)
self.w_h_o_delta = Utils().matrix_fill_zeros(self.w_h_o_delta) #判断迭代是否终止
def terminate(self):
if(0.5*self.train_error/self.batch_size<self.error_threshold
or self.iteration_num<=0):
return True
else:
print "error: ",self.train_error
self.train_error = 0.0
return False if __name__=="__main__":
'''
inputs = np.matrix([[0,0],[0,1],[1,0],[1,1]])
outputs = np.matrix([[0],[1],[1],[0]])
bpnn = BPNN(2, 5, 1) bpnn.train(Tanh(), Tanh(), inputs,
outputs, 0.1, 0.001,
10000, 100)
bpnn.test(inputs)
'''
#读取数据
r_fp = open("data","r")
#输入层节点数
input_level_node_num = 0
#隐藏层层节点数
hidden_level_node_num = 0
#输出层节点数
output_level_node_num = 0
#输入数据
inputs = []
#数据的真实值
outputs = []
i = 0
for line in r_fp:
strs = line.strip().split(",")
#第一行是每层的节点数
if i==0:
input_level_node_num = int(strs[0])
hidden_level_node_num = int(strs[1])
output_level_node_num = int(strs[2])
else:
#数据，最后一列是真实值
#特征值向量
i_v = []
#真实值向量
o_v = []
for i in range(len(strs)-1):
i_v.append(float(strs[i]))
o_v.append(float(strs[-1]))
inputs.append(i_v)
outputs.append(o_v)
i+=1
inputs = np.matrix(inputs)
outputs = np.matrix(outputs)
#每个特征以及结果的的最大值最小值归一化
max_inputs = np.max(inputs,axis=0)
min_inputs = np.min(inputs,axis=0)
normalize_inputs = (inputs-min_inputs)/(max_inputs-min_inputs)
max_outputs = np.max(outputs,axis=0)
min_outputs = np.min(outputs,axis=0)
normalize_outputs = (outputs-min_outputs)/(max_outputs-min_outputs)
#获取总共样本的个数
smaple_n = np.shape(normalize_inputs)[0]
#将数据按照2:1拆分成训练集与测试集
train_sample_n = smaple_n*2.0/3
train_inputs = normalize_inputs[0:train_sample_n:1,::]
train_outputs = normalize_outputs[0:train_sample_n:1,::]
test_inputs = normalize_inputs[train_sample_n:smaple_n:1,::]
test_outputs = normalize_outputs[train_sample_n:smaple_n:1,::] #构建网络
bpnn = BPNN(input_level_node_num, hidden_level_node_num, output_level_node_num)
#训练
bpnn.train(Sigmoid(), Sigmoid(), train_inputs,
train_outputs, 0.2, 0.01,
1000, 100)
#测试,其实最后预测值需要进行反归一化，这里没有做此步骤
bpnn.test(test_inputs,test_outputs)