gpu_info = !
nvidia-smi
gpu_info = '\n'.join(gpu_info)
if gpu_info.find('failed') >= 0:
print('Select the Runtime > "Change runtime type" menu to enable a GPU accelerator, ')
print('and then re-execute this cell.')
else:
print(gpu_info)
from psutil import virtual_memory
ram_gb = virtual_memory().total / 1e9
print('Your runtime has {:.1f} gigabytes of available RAM\n'.format(ram_gb))
if ram_gb < 20:
print('To enable a high-RAM runtime, select the Runtime > "Change runtime type"')
print('menu, and then select High-RAM in the Runtime shape dropdown. Then, ')
print('re-execute this cell.')
else:
print('You are using a high-RAM runtime!')
import pandas as pd
import numpy as np
import math
import [Link] as plt
import itertools
from [Link] import accuracy_score, precision_score, recall_score, f1_score, fbeta_score
from [Link] import roc_curve, roc_auc_score, precision_recall_curve, auc
from [Link] import classification_report, confusion_matrix, make_scorer
from sklearn.model_selection import GridSearchCV, TimeSeriesSplit
from [Link] import StandardScaler, MinMaxScaler
from [Link] import Sequential, save_model, load_model
from [Link] import LSTM, Dropout, Dense, GRU, RNN, SimpleRNN, Conv1D
from [Link] import LeakyReLU, Activation, Flatten, MaxPooling2D, MaxPooling1D
from [Link] import get_custom_objects
from [Link].scikit_learn import KerasClassifier
from [Link] import Adam,Nadam,SGD
from [Link] import BinaryCrossentropy
from [Link] import callbacks
import [Link] as ks
import tensorflow as tf
# Ignore warnings
import warnings
[Link]('ignore')
# Visualize the training and validation loss
def val_performance(plt, history, title):
[Link]([Link]['loss'], label='train loss')
[Link]([Link]['val_loss'], label='validation loss')
[Link]([Link]['accuracy'], label='train accuracy')
[Link]([Link]['val_accuracy'], label='validation accuracy')
[Link]()
[Link]('Model Training Vs Validation Performance for INDIA VIX (%s)' % title);
[Link]('epochs')
[Link]('Accuracy / Loss')
return
# plot ROC curve
def plot_roc_curve(plt, y_test, y_score, title):
fpr, tpr, _ = roc_curve(y_test, y_score)
auc_score = roc_auc_score(y_test, y_score)
[Link]()
lw = 2
[Link](fpr, tpr, color='darkorange',lw=lw, label='AUC = %.2f' % auc_score)
[Link]([0, 1], [0, 1], color='navy', lw=lw, linestyle='--', label='No Skill')
[Link]([0.0, 1.0])
[Link]([0.0, 1.0])
[Link]('False Positive Rate (FPR)')
[Link]('True Positive Rate (TPR)')
[Link]('ROC Curve for INDIA VIX (%s)' % title)
[Link]()
#[Link](loc="lower right")
[Link]()
return
def plot_confusion_matrix(y_test,y_pred, labels, normalize=False, title='Confusion Matrix', cmap=plt
cm = confusion_matrix(y_test, y_pred,labels=labels)
[Link](figsize=(4,4))
[Link](cm, interpolation='nearest', cmap=cmap)
[Link](title)
[Link]()
tick_marks = [Link](len(labels))
[Link](tick_marks, labels, rotation=0)
[Link](tick_marks, labels)
if normalize:
cm = [Link]('float') / [Link](axis=1)[:, [Link]]
cm = [Link](cm, decimals=2)
cm[[Link](cm)] = 0.0
thresh = [Link]() / 1.4
for i, j in [Link](range([Link][0]), range([Link][1])):
[Link](j, i, cm[i, j],
horizontalalignment="center",
color="white" if cm[i, j] > thresh else "black")
plt.tight_layout()
[Link]('True label')
[Link]('Predicted label')
[Link]()
return
def custom_range(start, stop, step, decimals=2):
# list([Link](0.05, 0.5, 0.05).round(2))
result = []
i = start
while(i <= stop):
[Link](round(i, decimals))
i += step
return result
# Define Models
################ LSTM ################
# [Link]
# [Link]
def create_model_lstm(neurons=10,
activation_h='tanh', activation_o='sigmoid',
recurrent_activation='sigmoid',
kernel_initializer='glorot_normal',
recurrent_initializer='orthogonal',
bias_initializer='zeros',
#kernel_constraint=maxnorm(3),
dropout_rate=0.0,
optimizer='Nadam',
learn_rate=0.001
):
model = Sequential()
# Input layer
[Link](LSTM(units=neurons,
activation=activation_h,
recurrent_activation=recurrent_activation,
kernel_initializer = kernel_initializer,
recurrent_initializer=recurrent_initializer,
bias_initializer=bias_initializer,
return_sequences=True,
input_shape=(X_train.shape[1], X_train.shape[2])))
[Link](Dropout(dropout_rate))
# 1st hidden layer
[Link](LSTM(units=neurons,
activation=activation_h,
kernel_initializer=kernel_initializer,
recurrent_initializer=recurrent_initializer,
bias_initializer=bias_initializer,
return_sequences=True))
[Link](Dropout(dropout_rate))
# 2nd hidden layer
[Link](LSTM(units=neurons,
activation=activation_h,
kernel_initializer=kernel_initializer,
recurrent_initializer=recurrent_initializer,
bias_initializer=bias_initializer,
return_sequences=False))
[Link](Dropout(dropout_rate))
#[Link](Dense(500,activation=activation))
# output layer
[Link](Dense(units=1,
activation=activation_o,
kernel_initializer=kernel_initializer,
bias_initializer=bias_initializer))
[Link](optimizer=optimizer, loss='binary_crossentropy', metrics=['accuracy'])
return model
# Fit Model with callback
def fit_model(model,data,
batch_size=10, epochs=10, title='DL',
class_weight=None, sample_weight=None):
X_train, y_train, X_test, y_test = tuple(data)
cb = [Link](monitor ='val_loss',
mode ='min',
verbose=1,
#baseline=0.4,
patience=50,
min_delta=0.001,
restore_best_weights = True)
history = [Link](X_train, y_train,
batch_size=batch_size,
epochs=epochs,
validation_split=0.2,
class_weight=class_weight,
sample_weight=sample_weight,
verbose=0,callbacks =[cb])
#save_model(model, title)
loss_epochs = [Link]([Link]['val_loss']) + 1
print('Minimum val loss at %d epochs' % loss_epochs)
accuracy_epochs = [Link]([Link]['val_accuracy']) + 1
print('Maximum val accuracy at %d epochs' % accuracy_epochs)
scores_train = [Link](X_train, y_train, verbose=0)
scores_test = [Link](X_test, y_test, verbose=0)
#print("Train Accuracy: %.2f%%" % (scores_train[1]*100))
#print("Test Accuracy: %.2f%%" % (scores_test[1]*100))
for key, val in zip(model.metrics_names, scores_train):
print ("Train: %s = %.2f%%" %(key, val*100))
for key, val in zip(model.metrics_names, scores_test):
print ("Test: %s = %.2f%%" %(key, val*100))
# y_score is the probability
y_score = [Link](X_test)
y_score = y_score.flatten()
y_pred = [Link](y_score<0.5, 0, 1)
plot_confusion_matrix(y_test, y_pred, labels, normalize=False)
#print(confusion_matrix(y_test, y_pred, labels=[0, 1]))
print(classification_report(y_test, y_pred))
val_performance(plt, history, title)
plot_roc_curve(plt,y_test,y_score,title)
return
# get data
path = '/content/drive/MyDrive/Colab Notebooks/ML/'
df = pd.read_csv(path+'indix vix cleaned [Link]',index_col='Date')
df['daysofweek'] = pd.to_datetime([Link]).weekday
target = 'y'
DOWN, UP = 0, 1
labels = [DOWN, UP]
display_labels = [DOWN, UP]
df[target] = [Link](df['Close'] > df['Close'].shift(1), UP, DOWN)
#[Link](columns=['Close_cboe_vix'],inplace=True)
#[Link](columns=['delta_cboe_vix','return_djia','return_v_nifty'],inplace=True)
[Link](inplace=True)
cols = list([Link])
[Link](target)
print(df[target].value_counts())
cols
X = []
y = []
n_timesteps = 5
n_features = len(cols)
#scaler = StandardScaler()
scaler = MinMaxScaler(feature_range=(-1,1))
index = int((len(df) + 1)*0.9)
data_train = df[cols].values[:index,:]
data_test = df[cols].values[index:,:]
# train
X_train_data = scaler.fit_transform(data_train)
X_train = []
y_train = []
for i in range(n_timesteps, X_train_data.shape[0]):
X_train.append(X_train_data[i-n_timesteps:i])
#y_train.append(df[target][i])
y_train = df[target].values[n_timesteps:index,]
X_train, y_train = [Link](X_train), [Link](y_train)
# test
X_test_data = [Link](data_test)
X_test = []
y_test = []
for i in range(n_timesteps, X_test_data.shape[0]):
X_test.append(X_test_data[i-n_timesteps:i])
#y_test.append(df[target][i])
y_test = df[target].values[index+n_timesteps:,]
X_test, y_test = [Link](X_test), [Link](y_test)
print('Dimention of X_train (3D):', X_train.shape)
print('Dimention of y_train (1D):', y_train.shape)
print('Dimention of X_test (3D):', X_test.shape)
print('Dimention of y_test (1D):', y_test.shape)
# list of hyperparameters
# [Link]
neurons = [x for x in range(10, 600, 20)]
dropout_rate = custom_range(0.05, 1, 0.05, decimals=2)
weight_constraint = [1, 2, 3, 4, 5]
# Define custom activation function
def gelu(x):
return 0.5 * x * (1 + [Link]([Link](2 / [Link]) * (x + 0.044715 * [Link](x, 3))))
# Add custom activation function to be used as string
get_custom_objects().update({'gelu': Activation(gelu)})
get_custom_objects().update({'lrelu': Activation(LeakyReLU(alpha=0.1))})
act_1 = ['softmax','softplus','softsign','sigmoid','hard_sigmoid']
act_2 = ['tanh','linear','relu','lrelu','selu','elu','gelu']
activation = [*act_1,*act_2]
optimizer = ['SGD', 'RMSprop', 'Adagrad', 'Adadelta', 'Adam', 'Adamax', 'Nadam']
learn_rate = [0.001, 0.01, 0.1, 0.2, 0.3]
#batch_size = [1, 2, 4, 8, 16, 32, 64, 128]
batch_size = [x for x in range(2, 129, 2)]
epochs = [10, 50, 100]
init_mode = ['uniform','lecun_uniform','normal','zero','once','glorot_normal']
init_mode = ['glorot_uniform','he_normal','he_uniform','orthogonal']
init_mode = [*init_mode_1, *init_mode_2]
model = KerasClassifier(build_fn=create_model_lstm, verbose=0)
param_grid = dict(neurons=[50],
dropout_rate=[0.05],
activation_h=['sigmoid'],
activation_o=['sigmoid'],
#recurrent_activation=['hard_sigmoid','sigmoid'],
optimizer=['Nadam'],
batch_size=[10],
epochs=[100],
kernel_initializer=['glorot_normal'],
recurrent_initializer=['orthogonal'],
bias_initializer=['zeros']
)
tscv_inner = TimeSeriesSplit(n_splits=2)
grid = GridSearchCV(estimator=model, param_grid=param_grid,
n_jobs=-1, cv=tscv_inner, verbose=1, refit=True)
grid_result = [Link](X_train, y_train)
for item in grid_result.best_params_.items():
print(item[0],':',item[1])
model = grid_result.best_estimator_
y_score = [Link](X_test)
y_score = y_score.flatten()
y_pred = [Link](y_score<0.5, 0, 1)
plot_confusion_matrix(y_test, y_pred, labels, normalize=False)
print(classification_report(y_test, y_pred))
plot_roc_curve(plt,y_test,y_score,title='LSTM')
# LSTM fits with callback
adam = Adam(learning_rate=0.0001,epsilon=1e-08,beta_1=0.9,beta_2=0.999,amsgrad=True)
nadam = Nadam(learning_rate=0.0001,epsilon=1e-07,beta_1=0.9,beta_2=0.99)
sgd = SGD(learning_rate=0.01,momentum=0.0,nesterov=False)
model = create_model_lstm(neurons=550,
activation_h='sigmoid', activation_o='sigmoid',
recurrent_activation='sigmoid',
optimizer=nadam,
dropout_rate=0.2,
kernel_initializer='glorot_normal',
recurrent_initializer='orthogonal',
bias_initializer='zeros'
)
#print([Link]())
batch_size, epochs, title = 10, 500, 'LSTM'
data = [X_train, y_train, X_test ,y_test]
fit_model(model,data,batch_size,epochs,title)
print([Link]())