architecture inspired from https github comfcholletkeras https github comandersblldeeppy todo fix initialize model s layers xshape listxshape xshape0 self batchsize for layer in self layers layer setupxshape xshape layer shapexshape self nlayers lenself layers setup optimizer self optimizer setupself self initialized true logging infototal parameters s self nparams def findbpropentryself reshape vector to matrix pass neural network instance to an optimizer forward pass backward pass forward propagation for layer in self layers x layer forwardpassx return x def predictself xnone if not self initialized self setuplayersx shape y xbatch batchiteratorx self batchsize for xb in xbatch y appendself fpropxb return np concatenatey property def parametriclayersself for layer in self layers if hasattrlayer parameters yield layer property def parametersself calculate an error for given examples trainingphase self istraining if trainingphase temporally disable training some layers work differently while training e g dropout self istraining false if x is none and y is none ypred self predictself x score self metricself y ypred else ypred self predictx score self metricy ypred if trainingphase self istraining true return score property def istrainingself return self training istraining setter def istrainingself train self training train for layer in self layers if isinstancelayer phasemixin layer istraining train def shuffledatasetself returns the number of layers return self nlayers property def nparamsself architecture inspired from https github com fchollet keras https github com andersbll deeppy todo fix initialize model s layers setup optimizer find entry layer for back propagation reshape vector to matrix pass neural network instance to an optimizer forward pass backward pass forward propagation returns a list of all parameters calculate an error for given examples temporally disable training some layers work differently while training e g dropout shuffle rows in the dataset returns the number of layers return the number of trainable parameters
import logging

import numpy as np
from autograd import elementwise_grad

from mla.base import BaseEstimator
from mla.metrics.metrics import get_metric
from mla.neuralnet.layers import PhaseMixin
from mla.neuralnet.loss import get_loss
from mla.utils import batch_iterator

np.random.seed(9999)




class NeuralNet(BaseEstimator):
    fit_required = False

    def __init__(
        self, layers, optimizer, loss, max_epochs=10, batch_size=64, metric="mse", shuffle=False, verbose=True
    ):
        self.verbose = verbose
        self.shuffle = shuffle
        self.optimizer = optimizer

        self.loss = get_loss(loss)

        
        if loss == "categorical_crossentropy":
            self.loss_grad = lambda actual, predicted: -(actual - predicted)
        else:
            self.loss_grad = elementwise_grad(self.loss, 1)
        self.metric = get_metric(metric)
        self.layers = layers
        self.batch_size = batch_size
        self.max_epochs = max_epochs
        self._n_layers = 0
        self.log_metric = True if loss != metric else False
        self.metric_name = metric
        self.bprop_entry = self._find_bprop_entry()
        self.training = False
        self._initialized = False

    def _setup_layers(self, x_shape):
        
        x_shape = list(x_shape)
        x_shape[0] = self.batch_size

        for layer in self.layers:
            layer.setup(x_shape)
            x_shape = layer.shape(x_shape)

        self._n_layers = len(self.layers)
        
        self.optimizer.setup(self)
        self._initialized = True
        logging.info("Total parameters: %s" % self.n_params)

    def _find_bprop_entry(self):
        

        if len(self.layers) > 0 and not hasattr(self.layers[-1], "parameters"):
            return -1
        return len(self.layers)

    def fit(self, X, y=None):
        if not self._initialized:
            self._setup_layers(X.shape)

        if y.ndim == 1:
            
            y = y[:, np.newaxis]
        self._setup_input(X, y)

        self.is_training = True
        
        self.optimizer.optimize(self)
        self.is_training = False

    def update(self, X, y):
        
        y_pred = self.fprop(X)

        
        grad = self.loss_grad(y, y_pred)
        for layer in reversed(self.layers[: self.bprop_entry]):
            grad = layer.backward_pass(grad)
        return self.loss(y, y_pred)

    def fprop(self, X):
        
        for layer in self.layers:
            X = layer.forward_pass(X)
        return X

    def _predict(self, X=None):
        if not self._initialized:
            self._setup_layers(X.shape)

        y = []
        X_batch = batch_iterator(X, self.batch_size)
        for Xb in X_batch:
            y.append(self.fprop(Xb))
        return np.concatenate(y)

    @property
    def parametric_layers(self):
        for layer in self.layers:
            if hasattr(layer, "parameters"):
                yield layer

    @property
    def parameters(self):
        
        params = []
        for layer in self.parametric_layers:
            params.append(layer.parameters)
        return params

    def error(self, X=None, y=None):
        
        training_phase = self.is_training
        if training_phase:
            
            
            self.is_training = False
        if X is None and y is None:
            y_pred = self._predict(self.X)
            score = self.metric(self.y, y_pred)
        else:
            y_pred = self._predict(X)
            score = self.metric(y, y_pred)
        if training_phase:
            self.is_training = True
        return score

    @property
    def is_training(self):
        return self.training

    @is_training.setter
    def is_training(self, train):
        self.training = train
        for layer in self.layers:
            if isinstance(layer, PhaseMixin):
                layer.is_training = train

    def shuffle_dataset(self):
        
        n_samples = self.X.shape[0]
        indices = np.arange(n_samples)
        np.random.shuffle(indices)
        self.X = self.X.take(indices, axis=0)
        self.y = self.y.take(indices, axis=0)

    @property
    def n_layers(self):
        
        return self._n_layers

    @property
    def n_params(self):
        
        return sum([layer.parameters.n_params for layer in self.parametric_layers])

    def reset(self):
        self._initialized = False