wehub-resource-sync
256 lines
8.3 KiB
Python
256 lines
8.3 KiB
Python
# Python Machine Learning by Sebastian Raschka, Packt Publishing Ltd. 2015
|
|
# Code Repository: https://github.com/rasbt/python-machine-learning-book
|
|
# Code License: MIT License
|
|
|
|
import numpy as np
|
|
import sys
|
|
|
|
|
|
class NeuralNetMLP(object):
|
|
""" Feedforward neural network / Multi-layer perceptron classifier.
|
|
|
|
Parameters
|
|
------------
|
|
n_hidden : int (default: 30)
|
|
Number of hidden units.
|
|
l2 : float (default: 0.)
|
|
Lambda value for L2-regularization.
|
|
No regularization if l2=0. (default)
|
|
epochs : int (default: 100)
|
|
Number of passes over the training set.
|
|
eta : float (default: 0.001)
|
|
Learning rate.
|
|
shuffle : bool (default: True)
|
|
Shuffles training data every epoch if True to prevent circles.
|
|
minibatche_size : int (default: 1)
|
|
Number of training samples per minibatch.
|
|
seed : int (default: None)
|
|
Random seed for initializing weights and shuffling.
|
|
|
|
Attributes
|
|
-----------
|
|
eval_ : dict
|
|
Dictionary collecting the cost, training accuracy,
|
|
and validation accuracy for each epoch during training.
|
|
|
|
"""
|
|
def __init__(self, n_hidden=30,
|
|
l2=0., epochs=100, eta=0.001,
|
|
shuffle=True, minibatch_size=1, seed=None):
|
|
|
|
self.random = np.random.RandomState(seed)
|
|
self.n_hidden = n_hidden
|
|
self.l2 = l2
|
|
self.epochs = epochs
|
|
self.eta = eta
|
|
self.shuffle = shuffle
|
|
self.minibatch_size = minibatch_size
|
|
|
|
def _onehot(self, y, n_classes):
|
|
"""Encode labels into one-hot representation
|
|
|
|
Parameters
|
|
------------
|
|
y : array, shape = [n_samples]
|
|
Target values.
|
|
|
|
Returns
|
|
-----------
|
|
onehot : array, shape = (n_samples, n_labels)
|
|
|
|
"""
|
|
onehot = np.zeros((n_classes, y.shape[0].astype(int)))
|
|
for idx, val in enumerate(y):
|
|
onehot[val, idx] = 1.
|
|
return onehot.T
|
|
|
|
def _sigmoid(self, z):
|
|
"""Compute logistic function (sigmoid)"""
|
|
return 1. / (1. + np.exp(-np.clip(z, -250, 250)))
|
|
|
|
def _forward(self, X):
|
|
"""Compute forward propagation step"""
|
|
|
|
# step 1: net input of hidden layer
|
|
# [n_samples, n_features] dot [n_features, n_hidden]
|
|
# -> [n_samples, n_hidden]
|
|
z_h = np.dot(X, self.w_h) + self.b_h
|
|
|
|
# step 2: activation of hidden layer
|
|
a_h = self._sigmoid(z_h)
|
|
|
|
# step 3: net input of output layer
|
|
# [n_samples, n_hidden] dot [n_hidden, n_classlabels]
|
|
# -> [n_samples, n_classlabels]
|
|
|
|
z_out = np.dot(a_h, self.w_out) + self.b_out
|
|
|
|
# step 4: activation output layer
|
|
a_out = self._sigmoid(z_out)
|
|
|
|
return z_h, a_h, z_out, a_out
|
|
|
|
def _compute_cost(self, y_enc, output):
|
|
"""Compute cost function.
|
|
|
|
Parameters
|
|
----------
|
|
y_enc : array, shape = (n_samples, n_labels)
|
|
one-hot encoded class labels.
|
|
output : array, shape = [n_samples, n_output_units]
|
|
Activation of the output layer (forward propagation)
|
|
|
|
Returns
|
|
---------
|
|
cost : float
|
|
Regularized cost
|
|
|
|
"""
|
|
L2_term = (self.l2 *
|
|
(np.sum(self.w_h ** 2.) +
|
|
np.sum(self.w_out ** 2.)))
|
|
|
|
term1 = -y_enc * (np.log(output))
|
|
term2 = (1. - y_enc) * np.log(1. - output)
|
|
cost = np.sum(term1 - term2) + L2_term
|
|
return cost
|
|
|
|
def predict(self, X):
|
|
"""Predict class labels
|
|
|
|
Parameters
|
|
-----------
|
|
X : array, shape = [n_samples, n_features]
|
|
Input layer with original features.
|
|
|
|
Returns:
|
|
----------
|
|
y_pred : array, shape = [n_samples]
|
|
Predicted class labels.
|
|
|
|
"""
|
|
z_h, a_h, z_out, a_out = self._forward(X)
|
|
y_pred = np.argmax(z_out, axis=1)
|
|
return y_pred
|
|
|
|
def fit(self, X_train, y_train, X_valid, y_valid):
|
|
""" Learn weights from training data.
|
|
|
|
Parameters
|
|
-----------
|
|
X_train : array, shape = [n_samples, n_features]
|
|
Input layer with original features.
|
|
y_train : array, shape = [n_samples]
|
|
Target class labels.
|
|
X_valid : array, shape = [n_samples, n_features]
|
|
Sample features for validation during training
|
|
y_valid : array, shape = [n_samples]
|
|
Sample labels for validation during training
|
|
|
|
Returns:
|
|
----------
|
|
self
|
|
|
|
"""
|
|
n_output = np.unique(y_train).shape[0] # number of class labels
|
|
n_features = X_train.shape[1]
|
|
|
|
########################
|
|
# Weight initialization
|
|
########################
|
|
|
|
# weights for input -> hidden
|
|
self.b_h = np.zeros(self.n_hidden)
|
|
self.w_h = self.random.normal(loc=0.0, scale=0.1,
|
|
size=(n_features, self.n_hidden))
|
|
|
|
# weights for hidden -> output
|
|
self.b_out = np.zeros(n_output)
|
|
self.w_out = self.random.normal(loc=0.0, scale=0.1,
|
|
size=(self.n_hidden, n_output))
|
|
|
|
epoch_strlen = len(str(self.epochs)) # for progress formatting
|
|
self.eval_ = {'cost': [], 'train_acc': [], 'valid_acc': []}
|
|
|
|
y_train_enc = self._onehot(y_train, n_output)
|
|
|
|
# iterate over training epochs
|
|
for i in range(self.epochs):
|
|
|
|
# iterate over minibatches
|
|
indices = np.arange(X_train.shape[0])
|
|
|
|
if self.shuffle:
|
|
self.random.shuffle(indices)
|
|
|
|
for start_idx in range(0, indices.shape[0] - self.minibatch_size +
|
|
1, self.minibatch_size):
|
|
batch_idx = indices[start_idx:start_idx + self.minibatch_size]
|
|
|
|
# forward propagation
|
|
z_h, a_h, z_out, a_out = self._forward(X_train[batch_idx])
|
|
|
|
##################
|
|
# Backpropagation
|
|
##################
|
|
|
|
# [n_samples, n_classlabels]
|
|
sigma_out = a_out - y_train_enc[batch_idx]
|
|
|
|
# [n_samples, n_hidden]
|
|
sigmoid_derivative_h = a_h * (1. - a_h)
|
|
|
|
# [n_samples, n_classlabels] dot [n_classlabels, n_hidden]
|
|
# -> [n_samples, n_hidden]
|
|
sigma_h = (np.dot(sigma_out, self.w_out.T) *
|
|
sigmoid_derivative_h)
|
|
|
|
# [n_features, n_samples] dot [n_samples, n_hidden]
|
|
# -> [n_features, n_hidden]
|
|
grad_w_h = np.dot(X_train[batch_idx].T, sigma_h)
|
|
grad_b_h = np.sum(sigma_h, axis=0)
|
|
|
|
# [n_hidden, n_samples] dot [n_samples, n_classlabels]
|
|
# -> [n_hidden, n_classlabels]
|
|
grad_w_out = np.dot(a_h.T, sigma_out)
|
|
grad_b_out = np.sum(sigma_out, axis=0)
|
|
|
|
# Regularization and weight updates
|
|
delta_w_h = (grad_w_h + self.l2*self.w_h)
|
|
delta_b_h = grad_b_h # bias is not regularized
|
|
self.w_h -= self.eta * delta_w_h
|
|
self.b_h -= self.eta * delta_b_h
|
|
|
|
delta_w_out = (grad_w_out + self.l2*self.w_out)
|
|
delta_b_out = grad_b_out # bias is not regularized
|
|
self.w_out -= self.eta * delta_w_out
|
|
self.b_out -= self.eta * delta_b_out
|
|
|
|
#############
|
|
# Evaluation
|
|
#############
|
|
|
|
# Evaluation after each epoch during training
|
|
z_h, a_h, z_out, a_out = self._forward(X_train)
|
|
cost = self._compute_cost(y_enc=y_train_enc,
|
|
output=a_out)
|
|
|
|
y_train_pred = self.predict(X_train)
|
|
y_valid_pred = self.predict(X_valid)
|
|
|
|
train_acc = ((np.sum(y_train == y_train_pred)).astype(np.float) /
|
|
X_train.shape[0])
|
|
valid_acc = ((np.sum(y_valid == y_valid_pred)).astype(np.float) /
|
|
X_valid.shape[0])
|
|
|
|
sys.stderr.write('\r%0*d/%d | Cost: %.2f '
|
|
'| Train/Valid Acc.: %.2f%%/%.2f%% ' %
|
|
(epoch_strlen, i+1, self.epochs, cost,
|
|
train_acc*100, valid_acc*100))
|
|
sys.stderr.flush()
|
|
|
|
self.eval_['cost'].append(cost)
|
|
self.eval_['train_acc'].append(train_acc)
|
|
self.eval_['valid_acc'].append(valid_acc)
|
|
|
|
return self |