trainer

#!/usr/bin/env python3

import numpy as np
from Parser import Board, DataParser
import pickle
import argparse

"""Class that parses the arguments"""
class ArgumentParser:
    def __init__(self):
        self.parser = argparse.ArgumentParser(description="Chess neural network")
        self.parser.add_argument("dataset", help="The dataset to use for training, must be a csv (;) file with the following columns: FEN, RES as header")
        self.parser.add_argument("-t", "--eta", help="Learning rate", type=float, default=0.1)
        self.parser.add_argument("-e", "--epochs", help="Number of epochs", type=int, default=5000)
        self.parser.add_argument("-b", "--batch", help="Batch size", type=int, default=10)
        self.parser.add_argument("-l", "--layers", help="Layers of the network", type=int, nargs="+", default=[64, 32, 16, 8, 4])
        self.parser.add_argument("-m", "--model", help="Model to load", type=str, required=False)
        self.parser.add_argument("-o", "--output", help="Output file", type=str, required=False, default="model")

    def parse(self):
        return self.parser.parse_args()

"""Class representing the neural network"""
class Network:
    def __init__(self, layers):
        self.layers = layers
        self.nbLayers = len(layers)
        # Generate random biases, except for the first layer (input layer)
        self.bias = [np.random.randn(y, 1) for y in layers[1:]]
        # Generate random weights for each layer, except for the first one (input layer)
        # The weights are generated as a matrix of size (y, x) where y is the number of nodes in the current layer
        # and x is the number of nodes in the previous layer
        self.weights = [np.random.randn(y, x) for x, y in zip(layers[:-1], layers[1:])]

    """Function that train the network using gradient descent and backpropagation algorithm"""
    def train(self, parser: DataParser, epochs: int, learningRate: float, miniBatchSize: int, outputFile: str):
        print(f"Training network with {epochs} epochs, learning rate: {learningRate}, batch size: {miniBatchSize}, output file: {outputFile}")
        if epochs * miniBatchSize > parser.getNbrBoards():
            epochs = parser.getNbrBoards() // miniBatchSize
            print("Epochs reduced to {}".format(epochs))
        for _ in range(epochs): # For each epoch
            miniBatch = parser.makeRandomBatch(miniBatchSize) # Make a random batch of miniBatchSize boards
            self.updateMiniBatch(miniBatch, learningRate) # Update the weights and biases using the batch
        print("Training complete")
        print("Saving model...")
        self.saveToFile(outputFile) # Save the trained network to a file

    """Update the weights and biases using gradient descent and backpropagation algorithm on a batch of boards"""
    def updateMiniBatch(self, miniBatch, learningRate):
        # Initialize the biases and weights gradients
        gradientB = [np.zeros(b.shape) for b in self.bias]
        gradientW = [np.zeros(w.shape) for w in self.weights]
        for board in miniBatch:
            # Compute the gradients for the current board
            deltaGradientB, deltaGradientW = self.backPropagation(board)
            # Add the gradients to the total gradients
            gradientB = [gb + dgb for gb, dgb in zip(gradientB, deltaGradientB)]
            gradientW = [gw + dgw for gw, dgw in zip(gradientW, deltaGradientW)]
        # Update the weights and biases
        self.weights = [w - (learningRate / len(miniBatch)) * gw for w, gw in zip(self.weights, gradientW)]
        self.bias = [b - (learningRate / len(miniBatch)) * gb for b, gb in zip(self.bias, gradientB)]

    """Backpropagation algorithm for the network that returns the gradients
       for the cost functions for the the biases and weights
       The gradients are used to update the biases and weights"""
    def backPropagation(self, board: Board):
        # x is the input board
        # y is the expected output ==> board.res
        boardInt = board.toVector()
        currentLayer = 1
        # Initialize the biases and weights gradients
        gradientB = [np.zeros(b.shape) for b in self.bias]
        gradientW = [np.zeros(w.shape) for w in self.weights]
        zList = [] # List of the weighted sums of the layers
        nodeValues = [boardInt] # List of the activated values of the layers

        # Compute the output of the network
        nodeValue = self.feedForward(boardInt, zList, nodeValues)
        # Compute the error of the output layer
        delta = self.costDerivative(nodeValue, board.res) * self.sigmoidPrime(zList[-currentLayer])
        # Compute the gradients for the output layer
        gradientB[-currentLayer] = delta
        gradientW[-currentLayer] = np.dot(delta, nodeValues[-currentLayer - 1].transpose())

        # Compute the gradients for the hidden layers
        for currentLayer in range(2, self.nbLayers):
            z = zList[-currentLayer]
            sp = self.sigmoidPrime(z)
            delta = np.dot(self.weights[-currentLayer + 1].transpose(), delta) * sp
            gradientB[-currentLayer] = delta
            gradientW[-currentLayer] = np.dot(delta, (nodeValues[-currentLayer - 1]).transpose())
        return (gradientB, gradientW)

    """Feedforward algorithm for the network that returns the output of the network after one training example"""
    def feedForward(self, board, zList, nodeValues):
        for b, w in zip(self.bias, self.weights):
            z = np.dot(w, board) + b # Weighted sum of the weights and biases of the current layer
            zList.append(z) # Add the weighted sum to the list of weighted sums
            board = self.sigmoid(z) # Activated value
            nodeValues.append(board) # Add the activated value to the list of activated values
        return board

    """Return the vector of partial derivatives of the cost function for the output activations."""
    def costDerivative(self, outputActivations, y):
        for i in range(len(outputActivations)):
            outputActivations[i] -= y[i]
        return outputActivations

    """Activation function for the network / Used to put the value between 0 and 1"""
    def sigmoid(self, z):
        return 1.0 / (1.0 + np.exp(-z))

    """Derivative of the sigmoid function."""
    def sigmoidPrime(self, z):
        return self.sigmoid(z) * (1 - self.sigmoid(z))

    """Return the output of the network if ``a`` is input."""
    def feedForwardTest(self, a):
        for b, w in zip(self.bias, self.weights):
            a = self.sigmoid(np.dot(w, a)+b)
        return a

    """Save the model to a file using pickle"""
    def saveToFile(self, filename):
        with open(filename, "wb") as file:
            pickle.dump(self.layers, file)
            pickle.dump(self.weights, file)
            pickle.dump(self.bias, file)

    """Load the model from a file using pickle"""
    def loadModel(self, filename):
        with open(filename, "rb") as file:
            self.layers = pickle.load(file)
            self.weights = pickle.load(file)
            self.biais = pickle.load(file)

def main(parser, args):
    print("Initializing network...")
    print("Layers: {}".format(args.layers))
    network = Network(args.layers)
    if args.model:
        network.loadModel(args.model)
    print("Training...")
    network.train(parser, args.epochs, args.eta, args.batch, args.output)
    return 0

if __name__ == "__main__":
    args = ArgumentParser().parse()
    dataParser = DataParser(args.dataset)
    exit(main(dataParser, args))