Batch Normalization

May 15, 2024

1 Batch Normalization From Scratch in Python

By Cristian Leo

[1]: import numpy as np

import [Link] as plt

# Custom classes (built from scratch)

from [Link] import WeightInitializer
from [Link] import PlotManager, LSTMTrainer, TimeSeriesDataset

[2]: class BatchNorm:

def __init__(self, hidden_size):
self.hidden_size = hidden_size
self.x = None
[Link] = [Link]((hidden_size, 1))
[Link] = [Link]((hidden_size, 1))

1
 def forward(self, x):
self.x = x
[Link] = [Link](x, axis=0)
[Link] = [Link](x, axis=0)
self.x_norm = (x - [Link]) / [Link]([Link] + 1e-6)
out = [Link] * self.x_norm + [Link]
return out

def backward(self, dout):

N = [Link][0]
dgamma = [Link](dout * self.x_norm, axis=0)
dbeta = [Link](dout, axis=0)
dx_norm = dout * [Link]
dvar = [Link](dx_norm * (self.x - [Link]) * -0.5 * ([Link] +␣
↪1e-8)**-1.5, axis=0)

dmu = [Link](dx_norm * -1 / [Link]([Link] + 1e-8), axis=0) + dvar *␣

↪[Link](-2 * (self.x - [Link]), axis=0)

dx = dx_norm / [Link]([Link] + 1e-8) + dvar * 2 * (self.x - [Link])␣

↪/ N + dmu / N

return dx, dgamma, dbeta

def plot_batch_norm(self):
# Compute the histograms of the pre-normalized and post-normalized data
pre_norm_hist, pre_norm_bins = [Link](self.x, bins=30)
post_norm_hist, post_norm_bins = [Link](self.x_norm, bins=30)

# Plot the pre-normalized data

[Link](pre_norm_bins[:-1], pre_norm_bins, weights=pre_norm_hist,␣
↪alpha=0.5, label='Pre-Normalization')

# Plot the post-normalized data

[Link](post_norm_bins[:-1], post_norm_bins, weights=post_norm_hist,␣
↪alpha=0.5, label='Post-Normalization')

# Add labels, a title, and a legend

[Link]('Value')
[Link]('Frequency')
[Link]('Pre-Normalization vs. Post-Normalization')
[Link]()

# Display the plot

[Link]()

def plot_activation_distribution(self, activation_function):

# Compute the activation before batch normalization
pre_norm_activation = activation_function(self.x)

2
 # Compute the activation after batch normalization
post_norm_activation = activation_function(self.x_norm)

# Compute the histograms of the pre-normalized and post-normalized␣

↪activations
pre_norm_hist, pre_norm_bins = [Link](pre_norm_activation,␣
↪bins=30)

post_norm_hist, post_norm_bins = [Link](post_norm_activation,␣

↪bins=30)

# Plot the pre-normalized activation

[Link](pre_norm_bins[:-1], pre_norm_bins, weights=pre_norm_hist,␣
↪alpha=0.5, label='Pre-Normalization')

# Plot the post-normalized activation

[Link](post_norm_bins[:-1], post_norm_bins, weights=post_norm_hist,␣
↪alpha=0.5, label='Post-Normalization')

# Add labels, a title, and a legend

[Link]('Activation Value')
[Link]('Frequency')
[Link]('Activation Distribution: Pre-Normalization vs.␣
↪Post-Normalization')

[Link]()

# Display the plot

[Link]()

[3]: # Create an instance of BatchNorm with a hypothetical hidden size, for example,␣
↪50.

bn_layer = BatchNorm(50)

# Pass some data from a mixture of Gaussians through the BatchNorm layer
# Generate data from two different normal distributions and combine them
data1 = [Link](-2, 1, size=(500, 50))
data2 = [Link](2, 1, size=(500, 50))
data = [Link]([data1, data2])

bn_layer.forward(data.T)

# Now, call the visualization methods

bn_layer.plot_batch_norm()

3
[4]: # Create an instance of BatchNorm with a hypothetical hidden size, for example,␣
↪50.

bn_layer = BatchNorm(50)

bn_layer.forward(data.T)

# Now, call the visualization methods

bn_layer.plot_activation_distribution([Link]) # Use the tanh activation␣
↪function

4
[5]: class LSTM:
"""
Long Short-Term Memory (LSTM) network.

Parameters:
- input_size: int, dimensionality of input space
- hidden_size: int, number of LSTM units
- output_size: int, dimensionality of output space
- init_method: str, weight initialization method (default: 'xavier')
"""
def __init__(self, input_size, hidden_size, output_size,␣
↪init_method='xavier'):

self.input_size = input_size
self.hidden_size = hidden_size
self.output_size = output_size
self.weight_initializer = WeightInitializer(method=init_method)

# Initialize weights
[Link] = self.weight_initializer.initialize((hidden_size, hidden_size␣
↪+ input_size))

5
 [Link] = self.weight_initializer.initialize((hidden_size, hidden_size␣
↪+ input_size))
[Link] = self.weight_initializer.initialize((hidden_size, hidden_size␣
↪+ input_size))

[Link] = self.weight_initializer.initialize((hidden_size, hidden_size␣

↪+ input_size))

# Initialize biases
[Link] = [Link]((hidden_size, 1))
[Link] = [Link]((hidden_size, 1))
[Link] = [Link]((hidden_size, 1))
[Link] = [Link]((hidden_size, 1))

# Initialize output layer weights and biases

[Link] = self.weight_initializer.initialize((output_size,␣
↪hidden_size))

[Link] = [Link]((output_size, 1))

# Initialize batch normalization layers

self.bn_f = BatchNorm(hidden_size)
self.bn_i = BatchNorm(hidden_size)
self.bn_o = BatchNorm(hidden_size)
self.bn_c = BatchNorm(hidden_size)

@staticmethod
def sigmoid(z):
"""
Sigmoid activation function.

Parameters:
- z: [Link], input to the activation function

Returns:
- [Link], output of the activation function
"""
return 1 / (1 + [Link](-z))

@staticmethod
def dsigmoid(y):
"""
Derivative of the sigmoid activation function.

Parameters:
- y: [Link], output of the sigmoid activation function

Returns:

6
 - [Link], derivative of the sigmoid function
"""
return y * (1 - y)

@staticmethod
def dtanh(y):
"""
Derivative of the hyperbolic tangent activation function.

Parameters:
- y: [Link], output of the hyperbolic tangent activation function

Returns:
- [Link], derivative of the hyperbolic tangent function
"""
return 1 - y * y

def forward(self, x):

"""
Forward pass through the LSTM network.

Parameters:
- x: [Link], input to the network

Returns:
- [Link], output of the network
- list, caches containing intermediate values for backpropagation
"""
caches = []
h_prev = [Link]((self.hidden_size, 1))
c_prev = [Link]((self.hidden_size, 1))
h = h_prev
c = c_prev

for t in range([Link][0]):
x_t = x[t].reshape(-1, 1)
combined = [Link]((h_prev, x_t))

f = [Link](self.bn_f.forward([Link]([Link], combined) + self.

bf))
↪

i = [Link](self.bn_i.forward([Link]([Link], combined) + self.

bi))
↪

o = [Link](self.bn_o.forward([Link]([Link], combined) + self.

bo))
↪

c_ = [Link](self.bn_c.forward([Link]([Link], combined) + [Link]))

7
 c = f * c_prev + i * c_
h = o * [Link](c)

cache = (h_prev, c_prev, f, i, o, c_, x_t, combined, c, h)

[Link](cache)

h_prev, c_prev = h, c

y = [Link]([Link], h) + [Link]
return y, caches

def backward(self, dy, caches, clip_value=1.0):

"""
Backward pass through the LSTM network.

Parameters:
- dy: [Link], gradient of the loss with respect to the output
- caches: list, caches from the forward pass
- clip_value: float, value to clip gradients to (default: 1.0)

Returns:
- tuple, gradients of the loss with respect to the parameters
"""
dWf, dWi, dWo, dWc = [np.zeros_like(w) for w in ([Link], [Link], self.
↪wo, [Link])]

dbf, dbi, dbo, dbc = [np.zeros_like(b) for b in ([Link], [Link], self.

↪bo, [Link])]

dWhy = np.zeros_like([Link])
dby = np.zeros_like([Link])

dgamma_f, dbeta_f = np.zeros_like(self.bn_f.gamma), np.zeros_like(self.

↪bn_f.beta)
dgamma_i, dbeta_i = np.zeros_like(self.bn_i.gamma), np.zeros_like(self.
↪bn_i.beta)

dgamma_o, dbeta_o = np.zeros_like(self.bn_o.gamma), np.zeros_like(self.

↪bn_o.beta)

dgamma_c, dbeta_c = np.zeros_like(self.bn_c.gamma), np.zeros_like(self.

↪bn_c.beta)

dy = [Link](self.output_size, -1)
dh_next = [Link]((self.hidden_size, 1))
dc_next = np.zeros_like(dh_next)

for cache in reversed(caches):

h_prev, c_prev, f, i, o, c_, x_t, combined, c, h = cache

8
 dh = [Link]([Link].T, dy) + dh_next
dc = dc_next + (dh * o * [Link]([Link](c)))

df = dc * c_prev * [Link](f)
di = dc * c_ * [Link](i)
do = dh * [Link]([Link](c))
dc_ = dc * i * [Link](c_)

df, dgamma_f_, dbeta_f_ = self.bn_f.backward(df)

di, dgamma_i_, dbeta_i_ = self.bn_i.backward(di)
do, dgamma_o_, dbeta_o_ = self.bn_o.backward(do)
dc_, dgamma_c_, dbeta_c_ = self.bn_c.backward(dc_)

dgamma_f += dgamma_f_
dbeta_f += dbeta_f_
dgamma_i += dgamma_i_
dbeta_i += dbeta_i_
dgamma_o += dgamma_o_
dbeta_o += dbeta_o_
dgamma_c += dgamma_c_
dbeta_c += dbeta_c_

dcombined_f = [Link]([Link].T, df)

dcombined_i = [Link]([Link].T, di)
dcombined_o = [Link]([Link].T, do)
dcombined_c = [Link]([Link].T, dc_)

dcombined = dcombined_f + dcombined_i + dcombined_o + dcombined_c

dh_next = dcombined[:self.hidden_size]
dc_next = f * dc

dWf += [Link](df, combined.T)

dWi += [Link](di, combined.T)
dWo += [Link](do, combined.T)
dWc += [Link](dc_, combined.T)

dbf += [Link](axis=1, keepdims=True)

dbi += [Link](axis=1, keepdims=True)
dbo += [Link](axis=1, keepdims=True)
dbc += dc_.sum(axis=1, keepdims=True)

dWhy += [Link](dy, h.T)

dby += dy

gradients = (dWf, dWi, dWo, dWc, dbf, dbi, dbo, dbc, dWhy, dby,␣
dgamma_f, dbeta_f, dgamma_i, dbeta_i, dgamma_o, dbeta_o, dgamma_c, dbeta_c)
↪

9
 for i in range(len(gradients)):
[Link](gradients[i], -clip_value, clip_value, out=gradients[i])

return gradients

def update_params(self, grads, learning_rate):

"""
Update the parameters of the network using the gradients.
"""
dWf, dWi, dWo, dWc, dbf, dbi, dbo, dbc, dWhy, dby, dgamma_f, dbeta_f,␣
↪dgamma_i, dbeta_i, dgamma_o, dbeta_o, dgamma_c, dbeta_c = grads

[Link] -= learning_rate * dWf

[Link] -= learning_rate * dWi
[Link] -= learning_rate * dWo
[Link] -= learning_rate * dWc
[Link] -= learning_rate * dbf

[Link] -= learning_rate * dbi

[Link] -= learning_rate * dbo
[Link] -= learning_rate * dbc

[Link] -= learning_rate * dWhy

[Link] -= learning_rate * dby

self.bn_f.gamma -= learning_rate * dgamma_f

self.bn_f.beta -= learning_rate * dbeta_f
self.bn_i.gamma -= learning_rate * dgamma_i
self.bn_i.beta -= learning_rate * dbeta_i
self.bn_o.gamma -= learning_rate * dgamma_o
self.bn_o.beta -= learning_rate * dbeta_o
self.bn_c.gamma -= learning_rate * dgamma_c
self.bn_c.beta -= learning_rate * dbeta_c

[6]: # Instantiate the dataset

dataset = TimeSeriesDataset('2005-01-01', '2020-12-31', train_size=0.7)
trainX, trainY, testX, testY = dataset.get_train_test()

# Plot the data

# Combine train and test data
combined = [Link]((trainY, testY))

# Plot the data

[Link](figsize=(14, 5))
[Link](combined, label='Google Stock Price', linewidth=2, color='dodgerblue')
[Link]('Google Stock Price', fontsize=20)
[Link]('Time', fontsize=16)

10
 [Link]('Normalized Stock Price', fontsize=16)
[Link](True)
[Link](fontsize=14)
[Link](fontsize=12)
[Link](fontsize=12)
[Link]()

[7]: # Reshape input to be [samples, time steps, features]

trainX = [Link](trainX, ([Link][0], [Link][1], 1))
testX = [Link](testX, ([Link][0], [Link][1], 1))

look_back = 1 # Number of previous time steps to include in each sample

hidden_size = 256 # Number of LSTM units
output_size = 1 # Dimensionality of the output space

lstm = LSTM(input_size=1, hidden_size=hidden_size, output_size=output_size,␣

↪init_method='xavier')

# Create and train the LSTM using LSTMTrainer

trainer = LSTMTrainer(lstm, learning_rate=1e-3, patience=10, verbose=True,␣
↪delta=0.001)

[Link](trainX, trainY, testX, testY, epochs=100, batch_size=32)

Epoch 1/100 - Loss: 0.57602, Val Loss: 1.29345

Epoch 11/100 - Loss: 0.00009, Val Loss: 0.41595
Epoch 21/100 - Loss: 0.00006, Val Loss: 0.38544
Epoch 31/100 - Loss: 0.00005, Val Loss: 0.38508
Early stopping

[9]: plot_manager = PlotManager()

11
# Inside your training loop
plot_manager.plot_losses(trainer.train_losses, trainer.val_losses)

# After your training loop

plot_manager.show_plots()

12