LSTM
class LSTMModel(nn.Module):
def __init__(self, input_size, hidden_size, num_layers, output_window, dropout):
super(LSTMModel, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
self.lstm = nn.LSTM(input_size, hidden_size, num_layers, batch_first=True)
self.selu = nn.SELU()
self.dropout = nn.Dropout(dropout)
self.fc = nn.Linear(hidden_size, output_window)
def forward(self, x):
batch_size = x.size(0)
h0 = Variable(torch.zeros(self.num_layers, batch_size, self.hidden_size)).to(x.device)
c0 = Variable(torch.zeros(self.num_layers, batch_size, self.hidden_size)).to(x.device)
x, _ = self.lstm(x, (h0, c0))
x = x[:, -1, :]
x = self.dropout(self.selu(x))
x = self.fc(x)
return x.unsqueeze(-1)
GRU
class GRUModel(nn.Module):
def __init__(self, input_size, hidden_size, num_layers, output_window, dropout):
super(GRUModel, self).__init__()
self.hidden_size = hidden_size
self.num_layers = num_layers
self.gru = nn.GRU(input_size, hidden_size, num_layers, batch_first=True)
self.selu = nn.SeLU()
self.dropout = nn.Dropout(dropout)
self.fc = nn.Linear(hidden_size, output_window)
def forward(self, x):
h0 = Variable(torch.zeros(self.num_layers, x.size(0), self.hidden_size)).to(x.device)
x, _ = self.gru(x, h0)
x = x[:, -1, :]
x = self.dropout(self.selu(x))
x = self.fc(x)
return x.unsqueeze(-1)
CNN
TCN
"""
https://github.com/hyliush/deep-time-series/blob/master/models/TCN.py
"""
class Chomp1d(nn.Module):
def __init__(self, chomp_size):
super(Chomp1d, self).__init__()
self.chomp_size = chomp_size
def forward(self, x):
return x[:, :, :-self.chomp_size].contiguous()
class TemporalBlock(nn.Module):
def __init__(self, n_inputs, n_outputs, kernel_size, stride, dilation, padding, dropout=0.2):
super(TemporalBlock, self).__init__()
self.conv1 = nn.utils.parametrizations.weight_norm(nn.Conv1d(n_inputs, n_outputs, kernel_size,
stride=stride, padding=padding, dilation=dilation))
self.chomp1 = Chomp1d(padding)
self.dropout1 = nn.Dropout(dropout)
self.conv2 = nn.utils.parametrizations.weight_norm(nn.Conv1d(n_outputs, n_outputs, kernel_size,
stride=stride, padding=padding, dilation=dilation))
self.chomp2 = Chomp1d(padding)
self.dropout2 = nn.Dropout(dropout)
self.net = nn.Sequential(self.conv1, self.chomp1, self.dropout1,
self.conv2, self.chomp2, self.dropout2)
self.downsample = nn.Conv1d(n_inputs, n_outputs, 1) if n_inputs != n_outputs else None
self.init_weights()
def init_weights(self):
self.conv1.weight.data.normal_(0, 0.01)
self.conv2.weight.data.normal_(0, 0.01)
if self.downsample is not None:
self.downsample.weight.data.normal_(0, 0.01)
def forward(self, x):
out = self.net(x)
res = x if self.downsample is None else self.downsample(x)
return out + res
class TCN(nn.Module):
def __init__(self, input_size, hidden_size, num_layers, kernel_size, output_window, dropout=0.2):
super(TCN, self).__init__()
self.output_window = output_window
num_channels = [hidden_size] * num_layers
layers = []
num_levels = len(num_channels)
for i in range(num_levels):
dilation_size = 2 ** i
in_channels = input_size if i == 0 else num_channels[i-1]
out_channels = num_channels[i]
layers += [TemporalBlock(in_channels, out_channels, kernel_size, stride=1, dilation=dilation_size,
padding=(kernel_size-1) * dilation_size, dropout=dropout)]
self.network = nn.Sequential(*layers)
self.out_proj = nn.Linear(hidden_size, self.output_window)
def forward(self, x):
x = x.transpose(1, 2)
x = self.network(x)
x = x[:, :, -1]
x = self.out_proj(x)
x = x.view(-1, self.output_window)
return x.unsqueeze(-1)
