# A Neural Network

I programmed a Neural Network in python. Feedback every kind is appreciated. I tried to use some vectorization but it turned out to become quite a mess. Because you can't append to numpy arrays I sometimes needed to use numpy arrays in list and sometime I could use just numpy arrays.

Is there a way to make It look cleaner?

"""
Author: Lupos
Purpose: Practising coding NN
Date: 17.11.2019
Description: test NN
"""

from typing import List, Dict  # used for typehints

import numpy as np  # used for forward pass, weight init, ect.

# logging
import logging  # used to log errors and info's in a file
from datetime import date  # used to get a name for the log file
import os  # used for creating a folder

import pandas as pd
import seaborn as sns
import matplotlib.pyplot as plt

class NeuralNetwork:
def __init__(self, x: List[float], y: List[float], nn_architecture: List[Dict], alpha: float, seed: int, custom_weights_data: List = [], custom_weights: bool = False) -> None:
"""
Constructor of the class Neural Network.

Parameters
----------
x : List[float]
Input data on which the Neural Network should get trained on
y : List[float]
Target/corresponding values for the input data.
nn_architecture : List[Dict]
Describes the architecture of the Neural Network.
alpha : float
The learning rate.
seed: int
Seed for numpy.For creating random values. For creating reproducible results.

Returns
-------
None
"""
self.level_of_debugging = logging.INFO
self.logger: object = self.init_logging(self.level_of_debugging)  # initializing of logging
# Dimension checks
self.check_input_output_dimension(x, y, nn_architecture)

np.random.seed(seed)  # set seed for reproducibility

self.y: List = y

self.output_model: float = np.zeros(y.shape)
self.alpha: float = alpha
self.layer_cache = {}  # later used for derivatives
self.error_term_cache = []

self.nn_architecture: List[Dict] = nn_architecture

self.weights: List = []  # np.array([])
self.init_weights(custom_weights, custom_weights_data)  # initializing of weights
self.w_d: List = []  # gardient in perspective to the weight

self.curr_layer: List = []
self.weight_change_cache: List = []

self.logger.info("__init__ executed")

# for visuliozing
self.x_train_loss_history = []
self.y_train_loss_history = []

self.bias_weight_tmp = []

x = np.array([np.insert(x, 0, 1)])
return x

def check_input_output_dimension(self, x, y, nn_architecture):
"""
Gets executed from the constructor "__init__". Is used
to check if the dimensions of input and output values correspond to the neuron size
in the input and output layer.

Parameters
----------
x
Input values
y
Output Values
nn_architecture : List[Dict]
Architecture of the neural network.

Returns
-------
None
"""
assert len(x[0]) == nn_architecture[0][
"layer_size"], 'Check the number of input Neurons and "X".'  # check if the first element in "x" has the right shape
assert len(y[0]) == nn_architecture[-1][
"layer_size"], 'Check the number of output Neurons and "Y".'  # check if the first element in "y" has the right shape
assert len(x) == len(y), "Check that X and Y have the corresponding values."

# mean square root
def loss(self, y: List[float], y_hat: List[float]) -> List[float]:
return np.sum(1 / 2 * (y - y_hat) ** 2)

def loss_derivative(self, y: List[float], y_hat: List[float]) -> List[float]:
y = np.array([y]).T
return np.array(-(y - y_hat))

def sigmoid(self, x: List[float]) -> List[float]:
return 1 / (1 + np.exp(-x))

def sigmoid_derivative(self, x: List[float]) -> List[float]:
return self.sigmoid(x) * (1 - self.sigmoid(x))

def relu(self, x: List[float]) -> List[float]:
return np.maximum(0, x)

def relu_derivative(self, x: List[float]) -> List[float]:
x[x <= 0] = 0
x[x > 0] = 1
return x

def linear(self, x: List[float]) -> List[float]:
return x

def activation_derivative(self, layer: Dict, curr_layer: List[float]) -> List[float]:
if layer["activation_function"] == "linear":
return np.array(self.linear(curr_layer))

elif layer["activation_function"] == "relu":
return np.array(self.relu_derivative(curr_layer))

elif layer["activation_function"] == "sigmoid":
return np.array(self.sigmoid_derivative(curr_layer))

else:
raise Exception("Activation function not supported!")

def communication(self, curr_epoch: int, curr_trainingsdata: int, data: List[float], target: List[float], how_often: int = 10) -> None:
"""
Gets executed from the method "train". Communicates information
about the current status of training progress.

Parameters
----------
i : int
A paramter that gets hand over. Current Iteration in a foor loop.
how_often: int
Is used to determine the frequently of updates from the training progress.

Returns
-------
None
"""
if curr_epoch % how_often == 0:
print("For iteration/trainings-example: #" + str(curr_epoch) + "/#"+ str(curr_trainingsdata))
print("Input: " + str(data))
print("Actual Output: " + str(target))
print("Predicted Output: " + str(self.output_model))
print("Loss: " + str(self.loss(y=target, y_hat=self.output_model)))
print("Value of last weight change: " + str(self.weight_change_cache[-1]))
print("\n")

def init_logging(self, level_of_debugging: str) -> object:
"""
Gets executed from the constructor "__init__". Initializes the logger.

Parameters
----------
level_of_debugging: {"logging.DEBUG", "logging.INFO", "logging.CRITICAL", "logging.WARNING", "logging.ERROR"}
Which error get logged.

Returns
-------
Object
return a logger object which is used to log errors.
"""
# creating a directory for "logs" if the directory doesnt exist
path = os.getcwd()
name = "logs"
full_path = path + "\\" + name
try:
if not os.path.isdir(full_path):
os.mkdir(full_path)
except OSError:
print("ERROR: Couldn't create a log folder.")

# create and configure logger
today = date.today()  # get current date
today_eu = today.strftime("%d-%m-%Y")  # european date format

LOG_FORMAT: str = "%(levelname)s  - %(asctime)s - %(message)s"  # logging format

logging.basicConfig(filename=full_path + "\\" + today_eu + ".log", level=level_of_debugging, format=LOG_FORMAT)
logger = logging.getLogger()

# Test logger
logger.info("------------------------------------------------")
logger.info("Start of the program")
logger.info("------------------------------------------------")

return logger

# TODO: "init_weights" is work in progress.
# TODO: "init_weights" init bias.
def init_weights(self, custom_weights: bool, custom_weights_data: List) -> List[float]:
"""
Gets executed from the constructor "__init__".
Initializes the weight in the whole Neural Network.

Returns
-------
List
Weights of the Neural Network.
"""
self.logger.info("init_weights executed")
for idx in range(0, len(self.nn_architecture) - 1):  # "len() - 1" because the output layer doesn't has weights

if not custom_weights:
# "self.nn_architecture[idx]["layer_size"] + 1" "+ 1" because we also have a bias term
weights_temp = 2 * np.random.rand(self.nn_architecture[idx + 1]["layer_size"], self.nn_architecture[idx]["layer_size"] + 1) - 1

self.weights.append(weights_temp)

if custom_weights:
self.weights = custom_weights_data

return self.weights

def activate_neuron(self, x: List[float], layer: Dict) -> List[float]:
"""
Gets executed from the method "forward" and "full_forward".
Activates the neurons in the current layer with the specified activation function.

Parameters
----------
x: List[float]
This are the values which get activated.
layer: Dict
A Dictionary with different attributes about the current layer.
Returns
-------
List
Outputs a List with activated values/neurons.
"""

if layer["activation_function"] == "relu":
temp_acti = self.relu(x)

# add bias to cache when not output layer
if not layer["layer_type"] == "output_layer":
else:
tmp_temp_acti_for_chache = temp_acti.T

# the name of the key of the dict is the index of current layer
idx_name = self.nn_architecture.index(layer)
self.layer_cache.update({"a" + str(idx_name): tmp_temp_acti_for_chache})

return temp_acti

elif layer["activation_function"] == "sigmoid":
temp_acti = self.sigmoid(x)

# add bias to cache when not output layer
if not layer["layer_type"] == "output_layer":
else:
tmp_temp_acti_for_chache = temp_acti.T

# the name of the key of the dict is the index of current layer
idx_name = self.nn_architecture.index(layer)
self.layer_cache.update({"a" + str(idx_name): tmp_temp_acti_for_chache})

return temp_acti

else:
raise Exception("Activation function not supported!")

def forward(self, weight: List[float], x: List[float], layer: Dict, idx: int) -> List[float]:
"""
Gets executed from the method "full_forward". This method make´s one
forward propagation step.

Parameters
----------
weight : List[float]
The weights of each associated Neurons in a List.
x : List[float]
The Input from the current layer which gets multiplicated with the weights and summed up.
layer : Dict
A Dictionary with different attributes about the current layer.

Returns
-------
List
List with values from the output of the one step forward propagation.
"""

curr_layer = np.dot(weight, x.T)

# add bias to cache when not output layer
if not layer["layer_type"] == "output_layer":
else:
tmp_curr_layer_for_chache = curr_layer.T

# the name of the key of the dict is the index of current layer
idx_name = self.nn_architecture.index(layer)
tmp_dict = {"z" + str(idx_name):  tmp_curr_layer_for_chache}
self.layer_cache.update(tmp_dict)   # append the "z" value | not activated value

curr_layer = self.activate_neuron(curr_layer, layer)

return curr_layer

# TODO: "full_forward" is work in progress
def full_forward(self, data):
"""
Gets executed from the method "forward_backprop". Makes the full forward propagation
through the whole Architecture of the Neural Network.

Returns
-------
List
List with the values of the output Layer.
"""
self.logger.info("full_forward executed")
self.layer_cache = {}  # delete cache used from previous iteration
for idx in range(0, len(self.nn_architecture) - 1):
self.logger.debug("Current-index (full_forward methode): " + str(idx))

if self.nn_architecture[idx]["layer_type"] == "input_layer":
self.layer_cache.update({"z0": data})
self.layer_cache.update({"a0": data})
self.curr_layer = self.forward(self.weights[idx], data, self.nn_architecture[idx + 1], idx=idx)  # "idx + 1" to fix issue regarding activation function
else:
self.curr_layer = self.forward(self.weights[idx], self.curr_layer, self.nn_architecture[idx + 1], idx=idx)

self.output_model = self.curr_layer

# TODO: "backprop" is work in progress
def backprop(self, target: List[float]) -> None:  # application of the chain rule to find derivative
"""
Gets executed from the method "forward_backprop". This method handels
the backpropagation of the Neural Network.

Returns
-------
None
"""
self.weight_change_cache = []
self.error_term_cache = []
self.logger.info("Backprop executed")
for idx, layer in reversed(list(enumerate(nn_architecture))):  # reversed because we go backwards
if not layer["layer_type"] == "input_layer":  # if we are in the input layer

# calculating the error term
if layer["layer_type"] == "output_layer":
temp_idx = "z" + str(idx)
d_a = self.activation_derivative(layer, self.layer_cache[temp_idx])
d_J = self.loss_derivative(y=target, y_hat=self.output_model)
error_term = np.array([np.multiply(d_a.flatten(), d_J.flatten())])
self.error_term_cache.append(error_term)

tmp_matrix_weight = np.asarray(self.weights[idx - 1])
tmp_bias_weight_t = np.array(tmp_matrix_weight.T[0])
self.bias_weight_tmp.append([tmp_bias_weight_t])
else:
temp_idx = "z" + str(idx)
layer_cache_tmp_drop_bias = np.delete(self.layer_cache[temp_idx], 0, 1)

d_a = self.activation_derivative(layer, layer_cache_tmp_drop_bias)

d_J = 0
for item in reversed(self.error_term_cache):
tmp_matrix_weight = np.asarray(self.weights[idx - 1])
self.bias_weight_tmp.append([tmp_matrix_weight.T[0]])

weights_tmp_drop_bias = np.delete(self.weights[idx], 0, 1)
d_J = d_J + np.dot(weights_tmp_drop_bias.T, item.T)

error_term = d_a.T * d_J
error_term = error_term.T
self.error_term_cache.append(error_term)

err_temp = error_term.T
temp_idx = "a" + str(idx - 1)
cache_tmp = self.layer_cache[temp_idx]
cache_tmp = np.delete(cache_tmp, 0, 1)  # delete bias
weight_change = err_temp * cache_tmp
self.weight_change_cache.append(weight_change)

# update weights
for idx in range(0, len(self.weight_change_cache)):  # reversed because we go backwards
curr_weight = self.weights[-idx - 1]
curr_weight = np.delete(curr_weight, 0, 1)  # delete bias
weight_change_tmp = self.weight_change_cache[idx]

total_weight_change = self.alpha * weight_change_tmp  # updating weight
curr_weight = curr_weight - total_weight_change
self.weights[-idx - 1] = curr_weight

# update bias
if layer["layer_type"] == "output_layer":
for i in range(0, len(self.bias_weight_tmp)):
tmp_weight_bias = np.asarray(self.bias_weight_tmp[i])
tmp_error_term_bias = np.asarray(self.error_term_cache[i])
self.bias_weight_tmp[i] = tmp_weight_bias - (self.alpha * tmp_error_term_bias)

# insert bias in weights
for i in range(0, len(self.weights)):
self.weights[i] = np.insert(self.weights[i], obj=0, values=self.bias_weight_tmp[i], axis=1)  # insert the weights for the biases

# TODO: "train" is work in progress
def train(self, how_often, epochs=20) -> None:
"""
Execute this method to start training your neural network.

Parameters
----------
how_often : int
gets handed over to communication. Is used to determine the frequently of updates from the training progress.
epochs : int
determines the epochs of training.

Returns
-------
None
"""
self.logger.info("Train-method executed")
for curr_epoch in range(epochs):
for idx, trainings_data in enumerate(x):

self.full_forward(trainings_data_with_bias)
self.backprop(self.y[idx])
self.communication(curr_epoch, idx, target=self.y[idx], data=trainings_data, how_often=how_often)

self.x_train_loss_history.append(curr_epoch)
self.y_train_loss_history.append(self.loss(y[idx], self.output_model))

def predict(self):
"""
Used for predicting with the neural network
"""
print("Predicting")
print("--------------------")

running = True
while(running):

pred_data = []
for i in range(0, self.nn_architecture[0]["layer_size"]):
tmp_input = input("Enter " + str(i) + " value: ")
pred_data.append(tmp_input)

self.full_forward(np.asarray([pred_data], dtype=float))
print("Predicted Output: ", self.output_model)
print(" ")

running = input('Enter "exit" if you want to exit. Else press "enter".')
if running == "exit" or running == "Exit":
running = False
else:
running = True

def visulize(self):
data = {"x": self.x_train_loss_history, "train": self.y_train_loss_history}
data = pd.DataFrame(data, columns=["x", "train"])

sns.set_style("darkgrid")
plt.figure(figsize=(12, 6))
sns.lineplot(x="x", y="train", data=data, label="train", color="orange")
plt.xlabel("Time In Epochs")
plt.ylabel("Loss")
plt.title("Loss over Time")
plt.show()

if __name__ == "__main__":
# data for nn and target
x = np.array([[1, 0]], dtype=float)
y = np.array([[0, 1]], dtype=float)

# nn_architecture is WITH input-layer and output-layer
nn_architecture = [{"layer_type": "input_layer", "layer_size": 2, "activation_function": "none"},
{"layer_type": "hidden_layer", "layer_size": 2, "activation_function": "sigmoid"},
{"layer_type": "output_layer", "layer_size": 2, "activation_function": "sigmoid"}]

weights_data = [np.array([[2, 0.15, 0.2], [2, 0.25, 0.3]], dtype=float), np.array([[4, 0.4, 0.45], [4, 0.5, 0.55]], dtype=float)]
weights_data = weights_data

#, custom_weights=True, custom_weights_data=weights_data
NeuralNetwork_Inst = NeuralNetwork(x, y, nn_architecture, 0.1, 5)
NeuralNetwork_Inst.train(how_often=100, epochs=500)
NeuralNetwork_Inst.visulize()

• Just FIY, check out numpy.hstack and numpy.vstack, you can append numpy arrays :) – IEatBagels Dec 12 '19 at 19:39
• @IEatBagels oh, I didn't know that thanks. – Lupos Dec 13 '19 at 14:40

# Separation of concerns

The NeuralNetwork class is quite complex at the moment, since it implements network handling (training, ...), logging and even visualization. The good news is, that there are already separate methods for them. My recommendation here would be to go one step further and move all the non-essential stuff (logging setup, visualization) out of the class. That will make the class much easier to maintain (and also to review). It will also very likely lead to greater flexibility, e.g. since the logging would not be hidden from the user.

# Internal functions

There are quite a few internal/helper methods in the class that are only supposed to be used by the class itself, e.g. in __init__. As per the PEP 8 style guide, their names should start with a single underscore (e.g. def _check_input_output_dimension(...) to mark them as "for internal use only" (there is no real private in Python). Following this convention makes it easier to tell the public and internal methods apart.

# Activation and derivatives

All the activation functions and their derivatives are stateless, i.e. they don't really need to be instance methods. Consider removing them from the class and provide them as callbacks when describing the network structure. For example:

# they could also live in your library, maybe with a bit of documentation
def sigmoid(x):
return 1 / (1 + np.exp(-x))

def sigmoid_derivative(x):
return sigmoid(x) * (1 - sigmoid(x))

# later or in an other file:

nn_architecture = [
{
"layer_type": "input_layer",
"layer_size": 2,
"activation_function": None
},
{
"layer_type": "hidden_layer",
"layer_size": 2,
"activation_function": {
"function": sigmoid,
"derivative": sigmoid_derivative
}
},
{
"layer_type": "output_layer",
"layer_size": 2,
"activation_function": {
"function": sigmoid,
"derivative": sigmoid_derivative
}
}
]


That will remove a lot of complexity from your implementation of various methods (e.g. activation_derivative and activate_neuron) and also makes it more extensible and flexible, since it's now up to the user to define new activation functions (and their derivative). Best practice implementations of the most common activation functions could still be part of your library, and you can even implement a helper function that does something like the following:

def get_activation(name):
if name == "sigmoid":
return {"function": sigmoid, "derivative": sigmoid_derivative}
elif name == "linear":
return {"function": linear, "derivative": linear_derivative}
elif ...:
...
# at the last line
except ValueError(f"No known activation function for name '{name}'")


or a dict

# a missing name would lead to a KeyError here, that maybe should be handled
# when used somewhere.
# also possible: implement get_activation from above using this dict,
# catch and transform the KeyError there
ACTIVATION = {
"sigmoid": {"function": sigmoid, "derivative": sigmoid_derivative},
"linear": {"function": linear, "derivative": linear_derivative},
...
}


This can also be hidden in your network, that if the user enters a string as it is now, the network class uses either of the two methods above and tries to determine which functions to use, while still providing the possibility to provide custom functions as well.

# Type annotations and documentation

From what I can see, there are a few cases where the type annotations don't seem to fit. E.g.

def relu_derivative(self, x: List[float]) -> List[float]:
x[x <= 0] = 0
x[x > 0] = 1
return x


This won't work with List[float], but is tailored to numpy arrays. I'd try to annotate the with np.ndarray, but the numpy developers don't seem to have settled on a best practice in that regard yet (see this GitHub issue). I don't use type annotations all to much, so maybe I'm wrong here. But they are not binding, so there is not a lot that can go wrong in that regard apart from confusing other programmers and some tools like mypy ;-)

Since you are otherwise following the numpydoc convention, a quick note on that regard: most numpy functions that can work both with Python types (lists, tuples, ...) and numpy arrays, define the input/output type to be array_like (see np.sin for example).

# Logging

Logging is a great functionality to have at hand, but there can be vastly different needs. My recommendation in that regard would be not to impose any kind of details on the user. There is simply no need to force a European date format on somebody from somewhere else or force them to have their log written to a file, especially if they can neither control the name nor the location the log file is written to. Simply allow the user to pass an (optional) logger when building the network, and work with that. What happens if no logger is provided is up to you. Either setting up a simple console logger or no logging at all are sensible defaults in my opinion.

# Tool support

It was already mentioned in a comment on the other answer, that there are quite a few typos in comments and method names (e.g. visulizevisualize). There are tools like codespell or language plugins for the IDE of your choice (e.g. Code Spell Checker for VS Code) that can help you in that regard.

There are also a lot of other tools in the Python ecosystem that can help you to keep a consistent code style and the like. A non-exhaustive list can be found at this answer here on Code Review Meta.

That's it for now. I would strongly recommend to implement at least some of these changes before bringing the class up for another round of review. Including them will make it much easier to judge the implementation of the core algorithms itself, since I'd reckon they are a lot easier to follow then.

• thanks you very much for your feedback. I'll implement your recommendations :) – Lupos Dec 13 '19 at 14:42
• How do I decide what to include in my class and what to "outsource"? e.g should I leave check_input_output_dimension() in or out and why? – Lupos Dec 13 '19 at 15:16
• And about the function with _ should I name all private function with _ or only helper function's? e.g. should I name forward(),full_forward, train() with _ or just helper functions like add_bias() with _? – Lupos Dec 13 '19 at 16:52
• Or should I name every function with _ which is a function the user is not supossed to run? – Lupos Dec 13 '19 at 16:57
• "[S]hould I name every function with _ which is a function the user is not supossed to run?" That's the idea. Most IDEs and documentation generation tools respect this convention as well, which greatly helps to reduce the amount of possible methods shown on autocomplete or in the API doc. – AlexV Dec 13 '19 at 17:29

A few points that stood out to me from a cursory glance:

• level_of_debugging doesn't seem to serve any purpose as an attribute. The __init__ function doesn't accept an argument to set it. The only other function which uses it, init_logging, takes it as an argument anyway. I think it should either
• be configurable via __init__, and init_logging should then use self.level_of_debugging instead of accepting an argument,
• or removed entirely and init_logging be simply called with logging.INFO directly.
• Instead of using path + "\\" + name and the like, fiddling with path separators manually, use pathlib.Path:

path = pathlib.Path.cwd()
name = "logs"
full_path = path / name


Even though it uses a Unix-like / for joining the paths, it will work fine on both Windows and POSIX systems. Similarly, use pathlib.Path.is_dir() and pathlib.Path.mkdir().

• If an activation function is not supported, a more appropriate action might to raise a TypeError or NotImplementedError instead of a generic Exception. From the explanation of TypeError:

This exception may be raised by user code to indicate that an attempted operation on an object is not supported, and is not meant to be. If an object is meant to support a given operation but has not yet provided an implementation, NotImplementedError is the proper exception to raise.

It's not a perfect fit, but it's a better fit than a generic Exception. More experienced Pythonistas can suggest better options.

• f-strings can, IMHO, give more readable code in the long run. Instead of lines like the following where you jump in and out of strings:

print("For iteration/trainings-example: #" + str(curr_epoch) + "/#"+ str(curr_trainingsdata))
print("Input: " + str(data))
print("Actual Output: " + str(target))
print("Predicted Output: " + str(self.output_model))
print("Loss: " + str(self.loss(y=target, y_hat=self.output_model)))
print("Value of last weight change: " + str(self.weight_change_cache[-1]))
print("\n")


You can have a cleaner form:

print(f"For iteration/trainings-example: #{curr_epoch)/#{curr_trainingsdata}")
print(f"Input: {data}")
print(f"Actual Output: {target}")
print(f"Predicted Output: {self.output_model}")
print(f"Loss: {self.loss(y=target, y_hat=self.output_model)}")
print(f"Value of last weight change: {self.weight_change_cache[-1]}")
print("\n")


Or, with sep to use a single print(), which is mostly a matter of taste (I think):

print(
f"For iteration/trainings-example: #{curr_epoch)/#{curr_trainingsdata}",
f"Input: {data}",
f"Actual Output: {target}",
f"Predicted Output: {self.output_model}",
f"Loss: {self.loss(y=target, y_hat=self.output_model)}",
f"Value of last weight change: {self.weight_change_cache[-1]}",
"\n",
sep="\n"
)


This last version is much more readable than the repeated prints and manual concatenation of strings.

• Okay, thank you very much for your feedback. I didn't know about the pathlib.Path libary and I also didn't know about the NotImplementedError. I'll fix these issues and the others too. Again thank you very much :) – Lupos Dec 11 '19 at 18:27
• @Lupos you're welcome, but I think you shouldn't have accepted the answer so soon (others might skip the question then). There are a few other places where improvements can be made, for example, using f-strings instead of "foo" + str(bar), typos (visulize instead of visualize), etc. – muru Dec 12 '19 at 2:49
• and what is the advantage of f-strings? – Lupos Dec 13 '19 at 16:54
• @Lupos I edited the answer – muru Dec 13 '19 at 17:42