I am posting my code below. This class returns LSTM models. It takes, X_train, X_Test, Y_Train, Y_Test, num of units in layer etc and other parameters.
class lstm_model(object):
def __init__(self, name, X_Train, X_Test, Y_Train, Y_Test,
mask,layer1_units, layer1_act, go_backwards,
merge_mode,lr, epochs, batch_size, model = 0, model_hist = 0,
return_sequences = True):
self.name = name
self.X_Train = X_Train
self.X_Test = X_Test
self.Y_Train = Y_Train
self.Y_Test = Y_Test
self.mask = mask
self.layer1_units = layer1_units
self.layer1_act = layer1_act
self.return_sequences = return_sequences
self.go_backwards = go_backwards
self.merge_mode = merge_mode
self.lr = lr
self.epochs = epochs
self.batch_size = batch_size
def model(self):
self.model = keras.Sequential()
if self.mask == True:
self.model.add(keras.layers.Masking(mask_value=0, input_shape=(self.X_Train.shape[1],
self.X_Train.shape[2])))
self.model.add(keras.layers.Bidirectional(keras.layers.LSTM(self.layer1_units,
return_sequences=self.return_sequences,
go_backwards= self.go_backwards,
activation = self.layer1_act),
#input_shape=(input_shape[0], input_shape[1]),
merge_mode = self.merge_mode))
else:
self.model.add(keras.layers.Bidirectional(keras.layers.LSTM(self.layer1_units,
return_sequences=self.return_sequences,
go_backwards= self.go_backwards,
activation = self.layer1_act),
input_shape=(self.X_Train.shape[1],
self.X_Train.shape[2]),
merge_mode = self.merge_mode))
if self.return_sequences == True:
self.model.add(keras.layers.TimeDistributed(keras.layers.Dense(3, activation=tf.nn.softmax)))
else:
self.model.add(keras.layers.Dense(3, activation=tf.nn.softmax))
adam = keras.optimizers.Adam(lr=self.lr, beta_1=0.9, beta_2=0.999, epsilon=None, decay=0.0, amsgrad=True)
self.model.compile(loss='sparse_categorical_crossentropy', optimizer = adam, metrics = ['accuracy'])
# define the checkpoint to store and retreive on later stages
filepath = "/Volumes/SEAGATE GEN/CSV Files/Converted/Test_Train/Keras/"
filename = self.name
join(filepath, filename)
checkpoint = keras.callbacks.ModelCheckpoint(join(filepath, filename),
monitor='loss',
verbose=1,
save_best_only=True,
mode='min')
self.callbacks_list = [checkpoint]
return self.model
def model_fit(self):
if self.return_sequences == True:
self.model_hist = self.model.fit(X_Train, Y_Train_Ori,
#validation_split = 0.1 ,
epochs=self.epochs,batch_size = self.batch_size,
verbose=1, callbacks = self.callbacks_list)
else:
self.model_hist = self.model.fit(X_Train, Y_Train_Ori_seq,
validation_data= (X_Test, Y_Test_Ori_seq),
#validation_split = 0.1 ,
epochs=self.epochs,batch_size = self.batch_size,
verbose=1, callbacks = self.callbacks_list)
return self.model_hist
def conf_mat(self):
if self.return_sequences == True:
self.Test_len = [3198,3357,3597,3804,3237,3118,3599,2847,3999,3198,3198,3198,3198,3260,3995,2718,3995,3832]
yhat = self.model.predict_classes(X_Test, verbose=1)
padding = np.zeros([72000,1])
for i in range(len(Test_len)):
padding[4000*(i):4000*(i+1) - self.Test_len[i],0] = 0
padding[4000*(i+1) - self.Test_len[i]: 4000*(i+1),0] = 1
self.act_pred = pd.DataFrame()
self.act_pred['Actual'] = Y_Test_Ori.reshape(-1)
self.act_pred['Predicted'] = yhat.reshape(-1)
self.act_pred['Padding'] = padding.reshape(-1)
self.act_pred = self.act_pred.loc[self.act_pred.Padding == 1].reset_index(drop=True)
print("\n\n{:.<23s}{:15f}".format('Accuracy', accuracy_score(self.act_pred.Actual, self.act_pred.Predicted)))
#print("{:.<23s}{:15f}".format('Recall', recall_score(self.act_pred.Actual, self.act_pred.Predicted)))
#print("{:.<23s}{:15f}".format('Precision', precision_score(self.act_pred.Actual, self.act_pred.Predicted)))
#print("{:.<23s}{:15f}".format('F1_Score', f1_score(self.act_pred.Actual, self.act_pred.Predicted)))
return confusion_matrix(self.act_pred.Actual, self.act_pred.Predicted)
else:
yhat = self.model.predict_classes(X_Test, verbose=1)
return confusion_matrix(Y_Test_Ori_seq, yhat)
def plot(self):
if self.return_sequences == False:
result = pd.DataFrame()
result[self.name + '_loss'] = self.model_hist.history['loss']
result[self.name + '_acc'] = self.model_hist.history['acc']
result[self.name + '_val_loss'] = self.model_hist.history['val_loss']
result[self.name + '_val_acc'] = self.model_hist.history['val_acc']
result.plot()
plt.show()
else:
result = pd.DataFrame()
result[self.name + '_loss'] = self.model_hist.history['loss']
result[self.name + '_acc'] = self.model_hist.history['acc']
result.plot()
plt.show()
def Sequence_conf_mat(self):
if self.return_sequences == True:
p=0
Test_len_cum = []
for i in self.Test_len:
p += i
Test_len_cum.append(p)
Actual = []
Predict = []
for i in range(len(Test_len_cum)):
if i == 0:
Actual.append(self.act_pred.Actual.values[0 : Test_len_cum[i]])
Predict.append(self.act_pred.Predicted.values[0 : Test_len_cum[i]])
else:
Actual.append(self.act_pred.Actual.values[Test_len_cum[i-1] : Test_len_cum[i]])
Predict.append(self.act_pred.Predicted.values[Test_len_cum[i-1] : Test_len_cum[i]])
Actual_seq = [int(np.mean(i)) for i in Actual]
Pred_seq = [stats.mode(i)[0][0] for i in Predict]
Mat = confusion_matrix(Actual_seq, Pred_seq)
pred_dict = []
for i in Predict:
dict1 = {0:0,1:0,2:0}
for j in i.tolist():
if j == 0:
dict1[0]+=1
elif j == 1:
dict1[1]+=1
else:
dict1[2]+=1
pred_dict.append(list(dict1.values()))
df2 = pd.DataFrame(pred_dict, columns = ['Count_0', 'Count_1', 'Count_2'])
df = pd.DataFrame()
df['Actual'] = Actual_seq
df['Predicted'] = Pred_seq
df['Sequence_Length'] = Test_len
df = pd.concat([df,df2], axis =1)
return Mat, df
else:
return print('\nAttribute: return_sequences needs to be true to enable this function call')
I have created a lstm_model() class instance and defined it. Here is the code:
I have used these parameters:
lstm_Trial = lstm_model('Class_Trial', X_Train = X_Train, X_Test = X_Test, Y_Train = Y_Train_Ori,
Y_Test = Y_Test_Ori, mask = True, layer1_units = 1 ,
layer1_act = 'tanh', go_backwards = False, return_sequences = True,
merge_mode = 'concat', lr = 0.01, epochs = 2, batch_size = 200)
- layer1_units = 40
- layer1_act = 'tanh'
- go_backwards = False
- return_sequences = True
- merge_mode = 'concat'
- lr = 0.01
- epochs = 2
- batch_size = 200
The model() method from the class returns the model. I am going to use this lstm model in the build_fn attribute of KerasClassifier().
