Given a simple multivariate time series problem
l = [ list(range(1000)),list(range(1000,2000)),list(range(2000,3000)), list(range(3000,4000)), list(range(4000,5000))]
df = pd.DataFrame(l).T
df.columns = ['feat_1', 'feat_2', 'feat_3', 'feat_4', 'feat_5']
target_sensor = 'feat_5'
If We want to predict the value of target_sensor in t+forecast_lead time steps
forecast_lead = 15
print('\nforecast_lead', forecast_lead)
target = f"{target_sensor}_TARGET{forecast_lead}"
features = list(df.columns.difference([target]))
df[target] = df[target_sensor].shift(-forecast_lead)
df = df.iloc[:-forecast_lead]
The input preparation is based on this torch.Dataset class:
class My_Dataset(Dataset):
def __init__(self, dataframe, target, features, sequence_length):
self.features = features #list of columns
self.target = target #str target col name
self.sequence_length = sequence_length #history we want to use
self.X = torch.tensor(dataframe[features].values).float() #to tensor
self.y = torch.tensor(dataframe[target].values).float() #to tensor
def __len__(self):
return self.X.shape[0]
def __getitem__(self, i):
if i >= self.sequence_length - 1:
i_start = i - self.sequence_length + 1
x = self.X[i_start:(i + 1), :]
else:
padding = self.X[0].repeat(self.sequence_length - i - 1, 1)
x = self.X[0:(i + 1), :]
x = torch.cat((padding, x), 0)
return x, self.y[i]
With the following Dataloader (batch_size =1 ) for simplicity.
train_loader = DataLoader(train_dataset, batch_size=1, shuffle=False,num_workers=1)
test_loader = DataLoader(test_dataset, batch_size=1, shuffle=False, num_workers=1)
The Model I am using is this:
class LSTM_Multivariate_Time_Series_Regression(nn.Module):
def __init__(self, num_features, hidden_size):
super().__init__()
self.num_feture= num_features # this is the number of features
self.hidden_size = hidden_size
self.num_layers = 1 # OR MORE THAN 1 HERE
self.lstm = nn.LSTM(input_size=num_features,
hidden_size=hidden_size,
batch_first=True,
num_layers=self.num_layers
)
self.linear = nn.Linear(in_features=self.hidden_size,
out_features=1)
In the forward pass if NUM_LAYER = 1
def forward(self, x):
lstm_output , (hn, cn) = self.lstm(x)
out = self.linear(hn[0]) # First dim of Hn is num_layers, which is set to 1 above.
In the forward pass, before passing through the Linear Layer IF SELF.NUM_LAYER > 1, I suppose following options are available
Use the last hidden state
hn[-1]Use the concatenation of all hidden state
#Docs WITH BATCH FIRST = TRUE : lstm_output tensor of shape: (BATCH_SIZE , SEQ_LENGHT, HIDDEN_SIZE) #docs WITH BATCH FIRST h_n: tensor of shape (NUM_LAYER, BATCH_SIZE, HIDDEN_SIZE) containing the final hidden state for each element in the batch
.1
out = self.linear(hn[-1]).flatten()
return out
Is a correct solution to add final Linear layer to the lstm layer this way?