0% found this document useful (0 votes)
17 views10 pages

LSTM Model Training for INDIA VIX Analysis

The document outlines a Python script for training a Long Short-Term Memory (LSTM) model using TensorFlow and Keras to predict stock market trends based on historical data. It includes sections for checking GPU availability, loading and preprocessing data, defining model architecture, fitting the model, and evaluating performance using metrics like accuracy and ROC curves. The script also incorporates hyperparameter tuning using GridSearchCV for optimizing model parameters.

Uploaded by

Shreya Parekh
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd
0% found this document useful (0 votes)
17 views10 pages

LSTM Model Training for INDIA VIX Analysis

The document outlines a Python script for training a Long Short-Term Memory (LSTM) model using TensorFlow and Keras to predict stock market trends based on historical data. It includes sections for checking GPU availability, loading and preprocessing data, defining model architecture, fitting the model, and evaluating performance using metrics like accuracy and ROC curves. The script also incorporates hyperparameter tuning using GridSearchCV for optimizing model parameters.

Uploaded by

Shreya Parekh
Copyright
© All Rights Reserved
We take content rights seriously. If you suspect this is your content, claim it here.
Available Formats
Download as PDF, TXT or read online on Scribd

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]())

You might also like