<s> """<STR_LIT>""" <EOL> import numpy as np <EOL> from sklearn . base import BaseEstimator , ClassifierMixin <EOL> from sklearn . metrics import roc_auc_score <EOL> from progressbar import Bar , ETA , Percentage , ProgressBar , RotatingMarker <EOL> from preprocessing . aux import delay_preds , delay_preds_2d <EOL> from utils . nn import buildNN <EOL> class NeuralNet ( BaseEstimator , ClassifierMixin ) : <EOL> """<STR_LIT>""" <EOL> def __init__ ( self , ensemble , architecture , training_params , <EOL> partsTrain = <NUM_LIT:1> , partsTest = <NUM_LIT:1> , <EOL> delay = <NUM_LIT> , skip = <NUM_LIT:100> , jump = None , subsample = <NUM_LIT:1> , <EOL> smallEpochs = <NUM_LIT:2> , majorEpochs = <NUM_LIT:20> , checkEveryEpochs = <NUM_LIT:2> , <EOL> verbose = True ) : <EOL> """<STR_LIT>""" <EOL> self . delay = delay <EOL> self . skip = skip <EOL> self . jump = jump <EOL> self . partsTrain = partsTrain <EOL> self . partsTest = partsTest <EOL> self . smallEpochs = smallEpochs <EOL> self . majorEpochs = majorEpochs <EOL> self . checkEveryEpochs = checkEveryEpochs <EOL> self . verbose = verbose <EOL> self . mdlNr = <NUM_LIT:0> <EOL> self . subsample = subsample <EOL> self . architecture = architecture <EOL> self . ensemble = ensemble <EOL> self . training_params = training_params <EOL> def fit ( self , X , y , Xtest = None , ytest = None ) : <EOL> """<STR_LIT>""" <EOL> input_dim = X . shape [ <NUM_LIT:1> ] <EOL> layers = [ i . keys ( ) [ <NUM_LIT:0> ] for i in self . architecture ] <EOL> self . isCNN = '<STR_LIT>' in layers <EOL> self . isRecurrent = '<STR_LIT>' in layers or '<STR_LIT>' in layers <EOL> if self . isCNN : <EOL> self . addDelay = delay_preds <EOL> self . training_params [ '<STR_LIT>' ] = self . delay // self . skip <EOL> elif self . isRecurrent : <EOL> self . addDelay = delay_preds_2d <EOL> else : <EOL> input_dim *= self . delay / self . skip <EOL> input_dim = int ( input_dim ) <EOL> self . addDelay = delay_preds <EOL> self . model = buildNN ( self . architecture , self . training_params , input_dim ) <EOL> widgets = [ '<STR_LIT>' , Percentage ( ) , '<STR_LIT:U+0020>' , Bar ( marker = RotatingMarker ( ) ) , <EOL> '<STR_LIT:U+0020>' , ETA ( ) , '<STR_LIT:U+0020>' ] <EOL> pbar = ProgressBar ( widgets = widgets , maxval = self . majorEpochs ) <EOL> pbar . start ( ) <EOL> for majorEpoch in range ( self . majorEpochs ) : <EOL> startingPoint = majorEpoch % self . partsTrain or self . mdlNr % self . partsTrain <EOL> if self . jump is not None : <EOL> trainData = self . addDelay ( X , delay = self . delay , skip = self . skip , <EOL> subsample = self . partsTrain , start = startingPoint , jump = self . jump ) <EOL> else : <EOL> trainData = self . addDelay ( X , delay = self . delay , skip = self . skip , <EOL> subsample = self . partsTrain , start = startingPoint ) <EOL> if self . isCNN : <EOL> trainData = trainData . reshape ( ( trainData . shape [ <NUM_LIT:0> ] , <NUM_LIT:1> , trainData . shape [ <NUM_LIT:1> ] , <NUM_LIT:1> ) ) <EOL> targets = y [ startingPoint : : self . partsTrain ] <EOL> trainData = trainData [ : : self . subsample ] <EOL> targets = targets [ : : self . subsample ] <EOL> self . model . fit ( trainData , targets , nb_epoch = self . smallEpochs , <EOL> batch_size = <NUM_LIT> , verbose = <NUM_LIT:0> , show_accuracy = True ) <EOL> trainData = None <EOL> pbar . update ( majorEpoch ) <EOL> if self . verbose and majorEpoch % self . checkEveryEpochs == <NUM_LIT:0> : <EOL> print ( "<STR_LIT>" % ( self . smallEpochs * ( majorEpoch + <NUM_LIT:1> ) ) ) <EOL> if Xtest is not None and ytest is not None : <EOL> pred = self . _predict_proba_train ( Xtest ) <EOL> score = np . mean ( roc_auc_score ( ytest [ <NUM_LIT:0> : : self . partsTest ] , pred ) ) <EOL> print ( "<STR_LIT>" % ( score ) ) <EOL> pred = None <EOL> if self . verbose : <EOL> print ( '<STR_LIT>' % ( self . smallEpochs * ( majorEpoch + <NUM_LIT:1> ) ) ) <EOL> def predict_proba ( self , X ) : <EOL> """<STR_LIT>""" <EOL> pred = [ ] <EOL> for part in range ( self . partsTest ) : <EOL> start = part * len ( X ) // self . partsTest - self . delay * ( part > <NUM_LIT:0> ) <EOL> stop = ( part + <NUM_LIT:1> ) * len ( X ) // self . partsTest <EOL> testData = self . addDelay ( X [ slice ( start , stop ) ] , delay = self . delay , skip = self . skip , <EOL> jump = self . jump ) [ self . delay * ( part > <NUM_LIT:0> ) : ] <EOL> if self . isCNN : <EOL> testData = testData . reshape ( ( testData . shape [ <NUM_LIT:0> ] , <NUM_LIT:1> , testData . shape [ <NUM_LIT:1> ] , <NUM_LIT:1> ) ) <EOL> pred . append ( self . model . predict_proba ( testData , batch_size = <NUM_LIT> , verbose = <NUM_LIT:0> ) ) <EOL> testData = None <EOL> pred = np . concatenate ( pred ) <EOL> return pred <EOL> def _predict_proba_train ( self , X ) : <EOL> """<STR_LIT>""" <EOL> testData = self . addDelay ( X , delay = self . delay , skip = self . skip , subsample = self . partsTest , start = <NUM_LIT:0> , jump = self . jump ) <EOL> if self . isCNN : <EOL> testData = testData . reshape ( ( testData . shape [ <NUM_LIT:0> ] , <NUM_LIT:1> , testData . shape [ <NUM_LIT:1> ] , <NUM_LIT:1> ) ) <EOL> pred = self . model . predict_proba ( testData , batch_size = <NUM_LIT> , verbose = <NUM_LIT:0> ) <EOL> testData = None <EOL> return pred </s>
