Using Python With Machine Learning and Deep Learning ( RNN & LSTM )

What is Stock Price Prediction?

Stock Price Prediction is one of the most popular problems in the series of related problems regarding Time Series data. There, we will use the indicators recorded in the past of a stock and predict its price in the future, so that appropriate investment decisions can be made.

For stock investors who rely on the method of technical analysis (Technical Analysis), they often rely on the adjusted closing price of the stock to make decisions to buy/sell stocks. Therefore, in this project, we will build a model that predicts the adjusted closing price of Tesla company stock based on the indices recorded on stock trading days in the period from 06 /29/2010 to 03/17/2017.


Understand the Input and Output

Input / Output of this project is:

  • Input: Indexes of stocks in the previous 30 trading days.
  • Output: Adjusted Closing Price of the current day.
  • The Data Sources


    The time - series data
    File data source with columns: Date, Open, High, Low, Close, Volume, Adj Close.

    Download dataset in my repository: Here

    The Pipe Line


    Import necessary libraries and Read File


    EDA Dataset




    • Date: Transaction date.
    • Open: Opening price.
    • High: The highest price of the day.
    • Low: Lowest price of the day.
    • Close: The closing price.
    • Volume: Trading volume.
    • Adj Close: Corrective closing price.






    Ảnh 1
    Ảnh 1
    Ảnh 1


    # Visualization of Adjusted Closing Price of Tesla Stock plt.figure(figsize=(10, 5)) df['Adj Close'].plot() plt.title('Adjusted Closing Price of Tesla Stock') plt.xlabel('Date') plt.ylabel('Adj Closing Price') plt.show()
    # Visualization of Volume overday of Tesla Stock plt.figure(figsize=(10, 5)) df['Volume'].plot() plt.title('Volume over day of Tesla Stock') plt.xlabel('Date') plt.ylabel('Volume') plt.show()

    Data Preparation

    Use Slicing Windowing Method

    # Declare the Windowing function (used to create X, y pairs for time series data) def slicing_window(df, df_start_idx, df_end_idx, input_size, label_size, offset, label_name): features = [] # Declare a list to store X labels = [] # Declare a list to store y # If df_end_idx equals the last index of the data frame, need to move down by a window size if df_end_idx == None: df_end_idx = len(df) - label_size - offset df_start_idx = df_start_idx + input_size + offset # Iterate through each data sample for idx in range(df_start_idx, df_end_idx): feature_start_idx = idx - input_size - offset feature_end_idx = feature_start_idx + input_size label_start_idx = idx - 1 label_end_idx = label_start_idx + label_size feature = df[feature_start_idx:feature_end_idx] # Get X label = df[label_name][label_start_idx:label_end_idx] # Get y features.append(feature) labels.append(label) # Convert lists to np.ndarrays features = np.array(features) labels = np.array(labels) return features, labels

    Split to Train / Val / Test

    dataset_length = len(df) # Number of data samples in the DataFrame TRAIN_SIZE = 0.7 # Training set ratio VAL_SIZE = 0.2 # Validation set ratio # Convert ratios to indices TRAIN_END_IDX = int(TRAIN_SIZE * dataset_length) VAL_END_IDX = int(VAL_SIZE * dataset_length) + TRAIN_END_IDX

    Make a function call

    INPUT_SIZE = 30 LABEL_SIZE = 1 OFFSET = 1 BATCH_SIZE = 64 TARGET_NAME = 'Adj Close' # Initialize X, y for the training set X_train, y_train = slicing_window(df, df_start_idx=0, df_end_idx=TRAIN_END_IDX, input_size=INPUT_SIZE, label_size=LABEL_SIZE, offset=OFFSET, label_name=TARGET_NAME) # Initialize X, y for the validation set X_val, y_val = slicing_window(df, df_start_idx=TRAIN_END_IDX, df_end_idx=VAL_END_IDX, input_size=INPUT_SIZE, label_size=LABEL_SIZE, offset=OFFSET, label_name=TARGET_NAME) # Initialize X, y for the test set X_test, y_test = slicing_window(df, df_start_idx=VAL_END_IDX, df_end_idx=None, input_size=INPUT_SIZE, label_size=LABEL_SIZE, offset=OFFSET, label_name=TARGET_NAME)

    Create tf.data.Dataset
    for the convenience of training in Tensorflow

    # Initilize tf.data.Dataset train_ds = tf.data.Dataset.from_tensor_slices((X_train, y_train)).batch(BATCH_SIZE) val_ds = tf.data.Dataset.from_tensor_slices((X_val, y_val)).batch(BATCH_SIZE) test_ds = tf.data.Dataset.from_tensor_slices((X_test, y_test)).batch(BATCH_SIZE) # Configuring and Preparing Data AUTOTUNE = tf.data.AUTOTUNE train_ds = train_ds.cache().prefetch(buffer_size=AUTOTUNE) val_ds = val_ds.cache().prefetch(buffer_size=AUTOTUNE)

    Build Model

    # Normalization layer normalize_layer = tf.keras.layers.Normalization() normalize_layer.adapt(np.vstack((X_train, X_val, X_test))) # Build model def build_model(input_shape, output_size): input_layer = tf.keras.Input(shape=input_shape, name='input_layer') x = normalize_layer(input_layer) for n_unit in [128, 64]: rnn_x = tf.keras.layers.SimpleRNN(n_unit, return_sequences=True, kernel_initializer=tf.initializers.GlorotUniform(seed=RANDOM_SEED) )(x) lstm_x = tf.keras.layers.LSTM(n_unit, return_sequences=True, kernel_initializer=tf.initializers.GlorotUniform(seed=RANDOM_SEED) )(x) x = tf.concat([rnn_x, lstm_x], axis=-1) rnn_x = tf.keras.layers.SimpleRNN(n_unit, return_sequences=False, kernel_initializer=tf.initializers.GlorotUniform(seed=RANDOM_SEED) )(x) lstm_x = tf.keras.layers.LSTM(n_unit, return_sequences=False, kernel_initializer=tf.initializers.GlorotUniform(seed=RANDOM_SEED) )(x) x = tf.concat([rnn_x, lstm_x], axis=-1) x = tf.keras.layers.Dense(32, activation='relu', kernel_initializer=tf.initializers.GlorotUniform(seed=RANDOM_SEED), name='fc_layer_1' )(x) output_layer = tf.keras.layers.Dense(output_size, kernel_initializer=tf.initializers.GlorotUniform(seed=RANDOM_SEED), name='output_layer')(x) model = tf.keras.Model(input_layer, output_layer, name='combined_model') return model # Summary of model INPUT_SHAPE = X_train.shape[-2:] model = build_model(INPUT_SHAPE, LABEL_SIZE) model.summary()

    Setting other parameters for the model

    # hyperparameter values EPOCHS = 500 LR = 1e-3 # Some optimization for model model.compile( optimizer=tf.keras.optimizers.Adam(learning_rate=LR), # Use optimizer Adam loss=tf.keras.losses.MeanSquaredError(), # Use loss Mean Squared Error Function )

    Training Model

    history = model.fit(train_ds, validation_data=val_ds, epochs=EPOCHS)

    Model Evalutation

    # Define MAE def mae(y_true, y_pred): mae = np.mean(np.abs((y_true - y_pred))) return mae # Define MSE def mse(y_true, y_pred): mse = np.mean((y_true - y_pred) ** 2) return mse # Define RMSE def rmse(y_true, y_pred): rmse = np.sqrt(np.mean((y_true-y_pred)**2)) return rmse # Define MAPE def mape(y_true, y_pred): mape = np.mean(np.abs((y_true-y_pred) / y_true)) * 100 return mape # Print result y_test_pred = model.predict(X_test, verbose=0) print(f'RMSE: {rmse(y_test, y_test_pred)}') print(f'MAPE: {mape(y_test, y_test_pred)}') print(f'MSE: {mse(y_test, y_test_pred)}') print(f'MAE: {mae(y_test, y_test_pred)}')

    The result is:

    • RMSE: 7.08275569019287
    • MAPE: 2.698402709155248
    • MSE: 50.16542816695947
    • MAE: 5.680429943373632

    Visualization Result

    Visualize loss / accuracy results during training

    train_loss = history.history['loss'] # Read training loss information val_loss = history.history['val_loss'] # Read validation loss information plt.figure(figsize=(10, 5)) # Set the figure size plt.subplot(1, 2, 1) # Initialize the plot for training loss plt.xlabel('Epochs') # Display the x-axis label as 'Epochs' plt.ylabel('Loss') # Display the y-axis label as 'Loss' plt.title('Training loss') # Display the title of the current plot as 'Training Loss' plt.plot(train_loss, color='green') # Plot the training loss values over epochs (plotting in green) plt.subplot(1, 2, 2) # Initialize the plot for validation loss plt.xlabel('Epochs') # Display the x-axis label as 'Epochs' plt.ylabel('Loss') # Display the y-axis label as 'Loss' plt.title('Validation loss') # Display the title of the current plot as 'Validation loss' plt.plot(val_loss, color='orange') # Plot the validation loss values over epochs (plotting in orange) plt.show() # Display both small plots

    Visualize the predicted value compared to the actual value on the test set

    def plot_difference(y, pred): plt.figure(figsize=(20, 6)) times = range(len(y)) y_to_plot = y.flatten() pred_to_plot = pred.flatten() plt.plot(times, y_to_plot, color='steelblue', marker='o', label='True value') plt.plot(times, pred_to_plot, color='orangered', marker='X', label='Prediction') plt.title('Adj Closing Price per day') plt.xlabel('Date') plt.ylabel('Adj Close Price') plt.legend() plt.show() plot_difference(y_test[:300], model.predict(X_test[:300], verbose=0))

    Based on low-value MAE, MSE, RMSE and MAPE metrics, as well as training loss visualization, validation loss, and predicted versus true value visualization on the test set, it is possible concluded that the model performed well in predicting Tesla's stock price!

    View my full project: Here
    Previous Post Next Post
    Visits my social profile and get connected: