| import torch |
| import torch.nn as nn |
| import torch.nn.functional as F |
|
|
| import math |
|
|
| class PositionalEmbedding(nn.Module): |
| def __init__(self, d_model, max_len=5000): |
| super(PositionalEmbedding, self).__init__() |
| |
| pe = torch.zeros(max_len, d_model).float() |
| pe.require_grad = False |
|
|
| position = torch.arange(0, max_len).float().unsqueeze(1) |
| div_term = (torch.arange(0, d_model, 2).float() * -(math.log(10000.0) / d_model)).exp() |
|
|
| pe[:, 0::2] = torch.sin(position * div_term) |
| pe[:, 1::2] = torch.cos(position * div_term)[:, :pe[:, 1::2].shape[1]] |
|
|
| pe = pe.unsqueeze(0) |
| self.register_buffer('pe', pe) |
|
|
| def forward(self, x): |
| return self.pe[:, :x.size(1)] |
|
|
| class TokenEmbedding(nn.Module): |
| def __init__(self, c_in, d_model, kernel_size=3): |
| super(TokenEmbedding, self).__init__() |
| self.tokenConv = nn.Conv1d(in_channels=c_in, out_channels=d_model, |
| kernel_size=kernel_size, padding=int(kernel_size//2), padding_mode='circular') |
| for m in self.modules(): |
| if isinstance(m, nn.Conv1d): |
| nn.init.kaiming_normal_(m.weight,mode='fan_in',nonlinearity='leaky_relu') |
|
|
| def forward(self, x): |
| x = self.tokenConv(x.permute(0, 2, 1)).transpose(1,2) |
| return x |
|
|
| class FixedEmbedding(nn.Module): |
| def __init__(self, c_in, d_model): |
| super(FixedEmbedding, self).__init__() |
|
|
| w = torch.zeros(c_in, d_model).float() |
| w.require_grad = False |
|
|
| position = torch.arange(0, c_in).float().unsqueeze(1) |
| div_term = (torch.arange(0, d_model, 2).float() * -(math.log(10000.0) / d_model)).exp() |
|
|
| w[:, 0::2] = torch.sin(position * div_term) |
| w[:, 1::2] = torch.cos(position * div_term) |
|
|
| self.emb = nn.Embedding(c_in, d_model) |
| self.emb.weight = nn.Parameter(w, requires_grad=False) |
|
|
| def forward(self, x): |
| return self.emb(x).detach() |
|
|
| class TemporalEmbedding(nn.Module): |
| def __init__(self, d_model, embed_type='fixed', freq='h'): |
| super(TemporalEmbedding, self).__init__() |
|
|
| minute_size = 4; hour_size = 24 |
| weekday_size = 7; day_size = 32; month_size = 13 |
| self.freq = freq |
|
|
| Embed = FixedEmbedding if embed_type=='fixed' else nn.Embedding |
| |
| self.hour_embed = Embed(hour_size, 1) |
| self.weekday_embed = Embed(weekday_size, 1) |
| self.day_embed = Embed(day_size, 1) |
| self.month_embed = Embed(month_size, 1) |
| self.value_embedding = TokenEmbedding(c_in=4, d_model=d_model, kernel_size=1) |
|
|
| def forward(self, x): |
| x = x.long() |
|
|
| |
| hour_x = self.hour_embed(x[:,:,3]) |
| weekday_x = self.weekday_embed(x[:,:,2]) |
| day_x = self.day_embed(x[:,:,1]) |
| month_x = self.month_embed(x[:,:,0]) |
| tempFeature = torch.cat((hour_x, weekday_x, day_x, month_x), dim=2) |
| tempFeature = self.value_embedding(tempFeature) |
| return tempFeature |
|
|
| class TimeFeatureEmbedding(nn.Module): |
| def __init__(self, d_model, freq='h'): |
| super(TimeFeatureEmbedding, self).__init__() |
|
|
| freq_map = {'h':4, 't':5, 's':6, 'm':1, 'a':1, 'w':2, 'd':3, 'b':3} |
| d_inp = freq_map[freq] |
| self.embed = nn.Linear(d_inp, d_model) |
| |
| def forward(self, x): |
| return self.embed(x) |
|
|
| class DataEmbedding(nn.Module): |
| def __init__(self, c_in, d_model, embed_type='fixed', freq='h', dropout=0.1): |
| super(DataEmbedding, self).__init__() |
|
|
| self.value_embedding = TokenEmbedding(c_in=c_in, d_model=d_model) |
| |
| self.position_embedding = PositionalEmbedding(d_model=d_model) |
| self.temporal_embedding = TemporalEmbedding(d_model=d_model, embed_type=embed_type, freq=freq) |
|
|
| def forward(self, x, x_mark): |
| x = self.value_embedding(x) + self.position_embedding(x) |
| return torch.stack((x, self.temporal_embedding(x_mark))) |
|
|
| class DataEmbeddingInformer(nn.Module): |
| def __init__(self, c_in, d_model, embed_type='fixed', freq='h', dropout=0.1, args="None"): |
| super(DataEmbeddingInformer, self).__init__() |
|
|
| self.args = args |
| self.value_embedding = TokenEmbedding(c_in=c_in, d_model=d_model) |
| self.position_embedding = PositionalEmbedding(d_model=d_model) |
| self.temporal_embedding = TemporalEmbedding(d_model=d_model, embed_type=embed_type, freq=freq) if embed_type!='timeF' else TimeFeatureEmbedding(d_model=d_model, embed_type=embed_type, freq=freq) |
|
|
| self.dropout = nn.Dropout(p=dropout) |
|
|
| def forward(self, x, x_mark): |
| x = self.value_embedding(x) + self.position_embedding(x) + self.temporal_embedding(x_mark) |
| return self.dropout(x) |
| |
| class DataEmbeddingStand(nn.Module): |
| def __init__(self, c_in, d_model, embed_type='fixed', freq='h', dropout=0.1, args="None"): |
| super(DataEmbeddingStand, self).__init__() |
|
|
| self.args = args |
| self.value_embedding = TokenEmbedding(c_in=c_in, d_model=d_model) |
| self.position_embedding = PositionalEmbedding(d_model=d_model) |
|
|
| self.dropout = nn.Dropout(p=dropout) |
|
|
| def forward(self, x, x_mark): |
| x = self.value_embedding(x) + self.position_embedding(x) |
| return self.dropout(x) |