|
import numpy as np |
|
import gzip |
|
import pickle |
|
import pandas as pd |
|
|
|
class Network(object): |
|
|
|
def __init__(self, sizes): |
|
self.num_layers = len(sizes) |
|
self.sizes = sizes |
|
self.biases = [np.random.randn(y, 1) for y in sizes[1:]] |
|
self.weights = [np.random.randn(y, x) for x,y in zip(sizes[:-1], sizes[1:])] |
|
|
|
|
|
|
|
|
|
|
|
|
|
|
|
def feedforward(self, a): |
|
|
|
for b, w in zip(self.biases, self.weights): |
|
a = sigmoid(np.dot(w, a) + b) |
|
|
|
return a |
|
|
|
|
|
def SGD(self, training_data, epochs, mini_batch_size, eta, k, test_data=None): |
|
if test_data: |
|
n_test = len(test_data) |
|
|
|
for j in range(epochs): |
|
np.random.shuffle(training_data) |
|
k_fold = self.k_fold_split(training_data, k) |
|
|
|
for fold in range(k): |
|
validation_data = k_fold[fold] |
|
train_data = [item for sublist in k_fold[:fold] + k_fold[fold + 1:] for item in sublist] |
|
|
|
for mini_batch in [train_data[k:k+mini_batch_size] for k in range(0, len(train_data), mini_batch_size)]: |
|
self.update_mini_batch(mini_batch, eta) |
|
|
|
print(f"Epoch {j}, Fold {fold}: {self.evaluate(validation_data)}/{len(validation_data)}") |
|
|
|
if test_data: |
|
print(f"Epoch {j}: {self.evaluate(test_data)}/{n_test}") |
|
|
|
def k_fold_split(self, data, k): |
|
fold_size = len(data) // k |
|
return [data[i*fold_size:(i+1)*fold_size] for i in range(k)] |
|
|
|
def update_mini_batch(self, mini_batch, eta): |
|
nabla_b = [np.zeros(b.shape) for b in self.biases] |
|
nabla_w = [np.zeros(w.shape) for w in self.weights] |
|
|
|
for x, y in mini_batch: |
|
delta_nabla_b, delta_nabla_w = self.backdrop(x, y) |
|
|
|
nabla_b = [nb+dnb for nb, dnb in zip(nabla_b, delta_nabla_b)] |
|
nabla_w = [nw+dnw for nw, dnw in zip(nabla_w, delta_nabla_w)] |
|
|
|
self.weights = [w-(eta/len(mini_batch))*nw for w, nw in zip(self.weights, nabla_w)] |
|
self.biases = [b-(eta/len(mini_batch))*nb for b, nb in zip(self.biases, nabla_b)] |
|
|
|
def evaluate(self, test_data): |
|
test_results = [(np.argmax(self.feedforward(x)), np.argmax(y)) for (x, y) in test_data] |
|
return sum(int(x == y) for (x, y) in test_results) |
|
|
|
def cost_derivative(self, output_activations, y): |
|
return output_activations - y |
|
|
|
def backdrop(self, x, y): |
|
nabla_b = [np.zeros(b.shape) for b in self.biases] |
|
nabla_w = [np.zeros(w.shape) for w in self.weights] |
|
activation = x |
|
activations = [x] |
|
zs = [] |
|
|
|
for b, w in zip(self.biases, self.weights): |
|
z = np.dot(w, activation)+b |
|
zs.append(z) |
|
activation = sigmoid(z) |
|
activations.append(activation) |
|
delta = self.cost_derivative(activations[-1], y) * sigmoid_prime(zs[-1]) |
|
nabla_b[-1] = delta |
|
nabla_w[-1] = np.dot(delta, activations[-2].transpose()) |
|
|
|
for l in range(2, self.num_layers): |
|
z = zs[-l] |
|
sp = sigmoid_prime(z) |
|
delta = np.dot(self.weights[-l+1].transpose(), delta) * sp |
|
nabla_b[-l] = delta |
|
nabla_w[-l] = np.dot(delta, activations[-l-1].transpose()) |
|
return (nabla_b, nabla_w) |
|
|
|
|
|
def sigmoid(z): |
|
sig = np.zeros_like(z) |
|
sig[z >= 0] = 1 / (1 + np.exp(-z[z >= 0])) |
|
sig[z < 0] = np.exp(z[z < 0]) / (1 + np.exp(z[z < 0])) |
|
return sig |
|
|
|
def sigmoid_prime(z): |
|
return sigmoid(z)*(1-sigmoid(z)) |
|
|
|
|
|
|
|
|
|
|
|
def load_data(): |
|
with gzip.open('C:\\Users\\tt235\\Desktop\\Code\\code\\代码复现\\BP神经网络\\mnist.pkl.gz', 'rb') as f: |
|
training_data, validation_data, test_data = pickle.load(f, encoding='latin1') |
|
|
|
return (training_data, validation_data, test_data) |
|
|
|
def load_data_wrapper(): |
|
tr_d, va_d, te_d = load_data() |
|
training_inputs = [np.reshape(x, (784, 1)) for x in tr_d[0]] |
|
training_results = [vectorized_result(y) for y in tr_d[1]] |
|
training_data = list(zip(training_inputs, training_results)) |
|
validation_inputs = [np.reshape(x, (784, 1)) for x in va_d[0]] |
|
validation_data = list(zip(validation_inputs, va_d[1])) |
|
test_inputs = [np.reshape(x, (784, 1)) for x in te_d[0]] |
|
test_results = [vectorized_result(y) for y in te_d[1]] |
|
test_data = list(zip(test_inputs, test_results)) |
|
return (training_data, validation_data, test_data) |
|
|
|
def vectorized_result(j): |
|
e = np.zeros((10, 1)) |
|
e[j] = 1.0 |
|
return e |
|
|
|
|
|
training_data, validation_data, test_data = load_data_wrapper() |
|
net = Network([784, 41, 10]) |
|
|
|
net.SGD(training_data, 3, 8, 3.0, 5, test_data=test_data) |
|
|