import torch
import torch.optim as optim
import torch.nn as nn
import numpy as np
import pandas as pd
from torch.utils.data import DataLoader, TensorDataset
from sklearn.preprocessing import MinMaxScaler
import matplotlib.pyplot as plt
torch.manual_seed(0)
device = torch.device("cuda:0" if torch.cuda.is_available()
else "cpu")
seq_length = 7
data_dim = 6
hidden_dim = 10
output_dim = 1
learning_rate = 0.01
epochs = 500
batch_size = 100
def build_dataset(data, seq_len):
dataX = []
dataY = []
for i in range(len(data)-seq_len):
x = data[i:i+seq_len, :]
y = data[i+seq_len, [-1]]
dataX.append(x)
dataY.append(y)
return np.array(dataX), np.array(dataY)
df = pd.read_csv(r-")
df = df[::-1]
df = df[['Date','Close/Last','Volume','Open','High','Low']]
train_size = int(len(df)*0.7)
train_set = df[0:train_size]
test_set = df[train_size-seq_length:]
scaler_x = MinMaxScaler()
scaler_x.fit(train_set.iloc[:,:-1])
train_set.iloc[:,:-1] = scaler_x.transform(train_set.iloc[:,:-1])
test_set.iloc[:,:-1] =scaler_x.transform(test_set.iloc[:,:-1])
scaler_y = MinMaxScaler()
scaler_y.fit(train_set.iloc[:,[-1]])
trainX, trainY = build_dataset(np.array(train_set), seq_length)
testX, testY = build_dataset(np.array(test_set), seq_length)
trainX_tensor = torch.LongTensor(trainX).to(device)
trainY_tensor = torch.LongTensor(trainY).to(device)
testX_tensor = torch.LongTensor(testX).to(device)
testY_tensor = torch.LongTensor(testY).to(device)
dataset = TensorDataset(trainX_tensor, trainY_tensor)
dataloader = DataLoader(dataset,
batch_size=batch_size,
shuffle=False,
drop_last=True)
class LSTM(nn.Module):
def __init__(self, input_dim, hidden_dim, seq_len, output_dim, layers):
super(LSTM, self).__init__()
self.hidden_dim = hidden_dim
self.seq_len = seq_len
self.output_dim = output_dim
self.layers = layers
self.lstm = nn.LSTM(input_dim, hidden_dim, num_layers=layers,
batch_first=True)
self.fc = nn.Linear(hidden_dim, output_dim, bias=True)
def reset_hidden_state(self):
self.hidden = (
torch.zeros(self.layers, self.seq_len, self.hidden_dim),
torch.zeros(self.layers, self.seq_len, self.hidden_dim)
)
def forward(self, x):
x, _status = self.lstm(x)
x = self.fc(x[:,-1])
return x;
LSTM = LSTM(data_dim, hidden_dim, seq_length, output_dim, 1).to(device)
