Hi there,
I’d like to do an anomaly detection on a univariate time series, but how to do it with a batch training? At the moment I’m using a workaround, but it is very slow…
class Encoder(nn.Module):
def __init__(self, input_length=118, features=1, compressed_dimension=8, batch_size=10):
super(Encoder, self).__init__()
self.input_length, self.features, self.batch_size = input_length, features, batch_size
self.compressed_dimension, self.hidden_dimension = compressed_dimension, 2*compressed_dimension
self.lstm_enc1 = nn.LSTM(
input_size=self.features,
hidden_size=self.hidden_dimension,
dropout=0.1,
batch_first=True
)
self.lstm_enc2 = nn.LSTM(
input_size=self.hidden_dimension,
hidden_size=self.compressed_dimension,
dropout=0.1,
batch_first=True
)
def forward(self, x):
size = len(x)
x = x.reshape((size, self.input_length, self.features)) #Batch_Size, Sequence Length, Features
x, (_, _) = self.lstm_enc1(x)
x, (hidden, _) = self.lstm_enc2(x)
x = hidden.reshape((size, self.features, self.compressed_dimension))
return x
class Decoder(nn.Module):
def __init__(self, input_length=118, features=1, compressed_dimension=8, batch_size=10):
super(Decoder, self).__init__()
self.input_length, self.features, self.batch_size = input_length, features, batch_size
self.compressed_dimension, self.hidden_dimension = compressed_dimension, 2*compressed_dimension
self.lstm_dec1 = nn.LSTM(
input_size=self.compressed_dimension,
hidden_size=self.compressed_dimension,
dropout=0.1,
batch_first=True
)
self.lstm_dec2 = nn.LSTM(
input_size=self.compressed_dimension,
hidden_size=self.hidden_dimension,
dropout=0.1,
batch_first=True
)
self.dense_layer = nn.Linear(self.hidden_dimension,self.features)
def batch_wise(self, x):# 10, 118, 16
temp = []
for i in range(x.shape[0]):
temp.append(self.dense_layer(x[i]))
return torch.stack(temp)
def forward(self, x):
size = len(x)
x = x.repeat(self.batch_size, self.input_length,self.features)
x = x.reshape((size, self.input_length, self.compressed_dimension))
x, (_, _) = self.lstm_dec1(x)
x, (_, _) = self.lstm_dec2(x)
x = x.reshape((size, self.input_length, self.hidden_dimension))
x = self.batch_wise(x)
x = x.reshape((size, self.features, self.input_length))
return x
class LAE(nn.Module):
def __init__(self):
super(LAE, self).__init__()
self.encoder = Encoder()
self.decoder = Decoder()
def forward(self, x):
x = self.encoder(x)
x = self.decoder(x)
return x
Does anyone know a better way to do a batch training with a LSTM Autoencoder? I’ve seen in Keras it seems to be no problem.
Thank you for your help 