View file src/colab/rnn_in_pytorch.py - Download

# -*- coding: utf-8 -*-
"""RNN_in_pytorch.ipynb

Automatically generated by Colab.

Original file is located at
    https://colab.research.google.com/drive/1kMR-QRzAXk8W0Dy2ZIFnRsN-szRTdfIs

https://www.kaggle.com/code/namanmanchanda/rnn-in-pytorch

Recurrent Neural Networks are a type of neural networks that are designed to work on sequence prediction models. RNNs can be used for text data, speech data, classification problems and generative models. Unlike ANNs, RNNs' prediction are based on the past prediction as well as the current input. RNNs are networks with loops in them allowing information to persist.

Each node of an RNN consists of 2 inputs:

- Memory unit
- Event unit

M(t-1) is the memory unit or the output of the previous prediction. E(t) is the current event or the information being provided at the present time. M(t) is the output of the current node or the output at the present time in the sequence.

1. Packages
"""

# Commented out IPython magic to ensure Python compatibility.
import numpy as np
import pandas as pd

import torch
import torch.nn as nn
import matplotlib.pyplot as plt
# %matplotlib inline

"""
2. Data definition

In this notebook, I'm going to train a very simple LSTM model, which is a type of RNN architecture to do time series prediction. Given some input data, it should be able to generate a prediction for the next step. I'll be using a Sin wave as an example as it's very easy to visualiase the behaviour of a sin wave.
"""

x = torch.linspace(0,799,800)
y = torch.sin(x*2*3.1416/40)

"""Plotting y"""

plt.figure(figsize=(12,4))
plt.xlim(-10,801)
plt.grid(True)
plt.xlabel("x")
plt.ylabel("sin")
plt.title("Sin plot")
plt.plot(y.numpy(),color='#8000ff')
plt.show()

"""
3. Batching the data

3.1 Splitting the data in train/test set
"""

test_size = 40
train_set = y[:-test_size]
test_set = y[-test_size:]

"""3.1.1 Plotting the training/testing set"""

plt.figure(figsize=(12,4))
plt.xlim(-10,801)
plt.grid(True)
plt.xlabel("x")
plt.ylabel("sin")
plt.title("Sin plot")
plt.plot(train_set.numpy(),color='#8000ff')
plt.plot(range(760,800),test_set.numpy(),color="#ff8000")
plt.show()

"""
3.2 Creating the batches of data

While working with LSTM models, we divide the training sequence into series of overlapping windows. The label used for comparison is the next value in the sequence.

For example if we have series of of 12 records and a window size of 3, we feed [x1, x2, x3] into the model, and compare the prediction to x4. Then we backdrop, update parameters, and feed [x2, x3, x4] into the model and compare the prediction to x5. To ease this process, I'm defining a function input_data(seq,ws) that created a list of (seq,labels) tuples. If ws is the window size, then the total number of (seq,labels) tuples will be len(series)-ws.
"""

def input_data(seq,ws):
    out = []
    L = len(seq)

    for i in range(L-ws):
        window = seq[i:i+ws]
        label = seq[i+ws:i+ws+1]
        out.append((window,label))

    return out

"""3.2.1 Calling the input_data function

The length of x = 800

The length of train_set = 800 - 40 = 760

The length of train_data = 760 - 40 - 720
"""

window_size = 40
train_data = input_data(train_set, window_size)
len(train_data)

"""3.2.2 Checking the 1st value from train_data"""

train_data[0]

"""
4. Defining the model

4.1 Model Class
"""

class LSTM(nn.Module):

    def __init__(self,input_size = 1, hidden_size = 50, out_size = 1):
        super().__init__()
        self.hidden_size = hidden_size
        self.lstm = nn.LSTM(input_size, hidden_size)
        self.linear = nn.Linear(hidden_size,out_size)
        self.hidden = (torch.zeros(1,1,hidden_size),torch.zeros(1,1,hidden_size))

    def forward(self,seq):
        lstm_out, self.hidden = self.lstm(seq.view(len(seq),1,-1), self.hidden)
        pred = self.linear(lstm_out.view(len(seq),-1))
        return pred[-1]

"""4.2 Model Instantiation"""

torch.manual_seed(42)
model = LSTM()
criterion = nn.MSELoss()
optimizer = torch.optim.SGD(model.parameters(), lr=0.01)
print(model)

"""
4.3 Training

During training, I'm visualising the prediction process for the test data on the go. It will give a better understanding of how the training is being carried out in each epoch. The training sequence is represented in purple while the predicted sequence in represented in orange.

"""

epochs = 10
future = 40

for i in range(epochs):

    for seq, y_train in train_data:
        optimizer.zero_grad()
        model.hidden = (torch.zeros(1,1,model.hidden_size),
                       torch.zeros(1,1,model.hidden_size))

        y_pred = model(seq)
        loss = criterion(y_pred, y_train)
        loss.backward()
        optimizer.step()

    print(f"Epoch {i} Loss: {loss.item()}")

    preds = train_set[-window_size:].tolist()
    for f in range(future):
        seq = torch.FloatTensor(preds[-window_size:])
        with torch.no_grad():
            model.hidden = (torch.zeros(1,1,model.hidden_size),
                           torch.zeros(1,1,model.hidden_size))
            preds.append(model(seq).item())

    loss = criterion(torch.tensor(preds[-window_size:]), y[760:])
    print(f"Performance on test range: {loss}")

    plt.figure(figsize=(12,4))
    plt.xlim(700,801)
    plt.grid(True)
    plt.plot(y.numpy(),color='#8000ff')
    plt.plot(range(760,800),preds[window_size:],color='#ff8000')
    plt.show()