eason
79db6e5c96
新增锂电池剩余使用寿命预测模型RUL-Mamba，包含以下主要组件： 1. 添加Mamba模块作为核心时序建模组件 2. 实现特征注意力网络(FAN)和门控残差网络(GRN) 3. 新增数据预处理和归一化层 4. 添加模型训练和评估脚本 5. 补充README文档说明使用方法 6. 添加可视化辅助工具Helper_Plot.py 7. 实现多种时间序列处理层(Embedding、AutoCorrelation等) 8. 添加配置文件requirements.txt 9. 补充测试数据集TJU battery data
694 lines
27 KiB
Python
694 lines
27 KiB
Python
import torch
|
||
import torch.nn as nn
|
||
import torch.nn.functional as F
|
||
import math
|
||
import numpy as np
|
||
import numpy as np
|
||
import math
|
||
from math import sqrt
|
||
|
||
|
||
class PositionalEmbedding(nn.Module):
|
||
def __init__(self, d_model, max_len=5000):
|
||
super(PositionalEmbedding, self).__init__()
|
||
# Compute the positional encodings once in log space.
|
||
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 = 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):
|
||
super(TokenEmbedding, self).__init__()
|
||
padding = 1 if torch.__version__ >= '1.5.0' else 2
|
||
self.tokenConv = nn.Conv1d(in_channels=c_in, out_channels=d_model,
|
||
kernel_size=3, padding=padding, padding_mode='circular', bias=False)
|
||
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
|
||
|
||
Embed = FixedEmbedding if embed_type == 'fixed' else nn.Embedding
|
||
if freq == 't':
|
||
self.minute_embed = Embed(minute_size, d_model)
|
||
self.hour_embed = Embed(hour_size, d_model)
|
||
self.weekday_embed = Embed(weekday_size, d_model)
|
||
self.day_embed = Embed(day_size, d_model)
|
||
self.month_embed = Embed(month_size, d_model)
|
||
|
||
def forward(self, x):
|
||
x = x.long()
|
||
minute_x = self.minute_embed(x[:, :, 4]) if hasattr(
|
||
self, 'minute_embed') else 0.
|
||
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])
|
||
|
||
return hour_x + weekday_x + day_x + month_x + minute_x
|
||
|
||
|
||
class TimeFeatureEmbedding(nn.Module):
|
||
def __init__(self, d_model, embed_type='timeF', 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, bias=False)
|
||
|
||
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) 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):
|
||
if x_mark is None:
|
||
x = self.value_embedding(x) + self.position_embedding(x)
|
||
else:
|
||
x = self.value_embedding(
|
||
x) + self.temporal_embedding(x_mark) + self.position_embedding(x)
|
||
return self.dropout(x)
|
||
|
||
|
||
class DataEmbedding_inverted(nn.Module):
|
||
def __init__(self, c_in, d_model, embed_type='fixed', freq='h', dropout=0.1):
|
||
super(DataEmbedding_inverted, self).__init__()
|
||
self.value_embedding = nn.Linear(c_in, d_model)
|
||
self.dropout = nn.Dropout(p=dropout)
|
||
|
||
def forward(self, x, x_mark):
|
||
x = x.permute(0, 2, 1)
|
||
# x: [Batch Variate Time]
|
||
if x_mark is None:
|
||
x = self.value_embedding(x)
|
||
else:
|
||
x = self.value_embedding(torch.cat([x, x_mark.permute(0, 2, 1)], 1))
|
||
# x: [Batch Variate d_model]
|
||
return self.dropout(x)
|
||
|
||
|
||
class DataEmbedding_wo_pos(nn.Module):
|
||
def __init__(self, c_in, d_model, embed_type='fixed', freq='h', dropout=0.1):
|
||
super(DataEmbedding_wo_pos, 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) 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):
|
||
if x_mark is None:
|
||
x = self.value_embedding(x)
|
||
else:
|
||
x = self.value_embedding(x) + self.temporal_embedding(x_mark)
|
||
return self.dropout(x)
|
||
|
||
|
||
class PatchEmbedding(nn.Module):
|
||
def __init__(self, d_model, patch_len, stride, padding, dropout):
|
||
super(PatchEmbedding, self).__init__()
|
||
# Patching
|
||
self.patch_len = patch_len
|
||
self.stride = stride
|
||
self.padding_patch_layer = nn.ReplicationPad1d((0, padding))
|
||
|
||
# Backbone, Input encoding: projection of feature vectors onto a d-dim vector space
|
||
self.value_embedding = nn.Linear(patch_len, d_model, bias=False)
|
||
|
||
# Positional embedding
|
||
self.position_embedding = PositionalEmbedding(d_model)
|
||
|
||
# Residual dropout
|
||
self.dropout = nn.Dropout(dropout)
|
||
|
||
def forward(self, x):
|
||
# do patching
|
||
n_vars = x.shape[1]
|
||
x = self.padding_patch_layer(x)
|
||
x = x.unfold(dimension=-1, size=self.patch_len, step=self.stride)
|
||
x = torch.reshape(x, (x.shape[0] * x.shape[1], x.shape[2], x.shape[3]))
|
||
# Input encoding
|
||
x = self.value_embedding(x) + self.position_embedding(x)
|
||
return self.dropout(x), n_vars
|
||
|
||
class AutoCorrelation(nn.Module):
|
||
"""
|
||
AutoCorrelation Mechanism with the following two phases:
|
||
(1) period-based dependencies discovery
|
||
(2) time delay aggregation
|
||
This block can replace the self-attention family mechanism seamlessly.
|
||
"""
|
||
|
||
def __init__(self, mask_flag=True, factor=1, scale=None, attention_dropout=0.1, output_attention=False):
|
||
super(AutoCorrelation, self).__init__()
|
||
self.factor = factor
|
||
self.scale = scale
|
||
self.mask_flag = mask_flag
|
||
self.output_attention = output_attention
|
||
self.dropout = nn.Dropout(attention_dropout)
|
||
|
||
def time_delay_agg_training(self, values, corr):
|
||
"""
|
||
SpeedUp version of Autocorrelation (a batch-normalization style design)
|
||
This is for the training phase.
|
||
"""
|
||
head = values.shape[1]
|
||
channel = values.shape[2]
|
||
length = values.shape[3]
|
||
# find top k
|
||
top_k = int(self.factor * math.log(length))
|
||
mean_value = torch.mean(torch.mean(corr, dim=1), dim=1)
|
||
index = torch.topk(torch.mean(mean_value, dim=0), top_k, dim=-1)[1]
|
||
weights = torch.stack([mean_value[:, index[i]] for i in range(top_k)], dim=-1)
|
||
# update corr
|
||
tmp_corr = torch.softmax(weights, dim=-1)
|
||
# aggregation
|
||
tmp_values = values
|
||
delays_agg = torch.zeros_like(values).float()
|
||
for i in range(top_k):
|
||
pattern = torch.roll(tmp_values, -int(index[i]), -1)
|
||
delays_agg = delays_agg + pattern * \
|
||
(tmp_corr[:, i].unsqueeze(1).unsqueeze(1).unsqueeze(1).repeat(1, head, channel, length))
|
||
return delays_agg
|
||
|
||
def time_delay_agg_inference(self, values, corr):
|
||
"""
|
||
SpeedUp version of Autocorrelation (a batch-normalization style design)
|
||
This is for the inference phase.
|
||
"""
|
||
batch = values.shape[0]
|
||
head = values.shape[1]
|
||
channel = values.shape[2]
|
||
length = values.shape[3]
|
||
# index init
|
||
init_index = torch.arange(length).unsqueeze(0).unsqueeze(0).unsqueeze(0).repeat(batch, head, channel, 1).to(values.device)
|
||
# find top k
|
||
top_k = int(self.factor * math.log(length))
|
||
mean_value = torch.mean(torch.mean(corr, dim=1), dim=1)
|
||
weights, delay = torch.topk(mean_value, top_k, dim=-1)
|
||
# update corr
|
||
tmp_corr = torch.softmax(weights, dim=-1)
|
||
# aggregation
|
||
tmp_values = values.repeat(1, 1, 1, 2)
|
||
delays_agg = torch.zeros_like(values).float()
|
||
for i in range(top_k):
|
||
tmp_delay = init_index + delay[:, i].unsqueeze(1).unsqueeze(1).unsqueeze(1).repeat(1, head, channel, length)
|
||
pattern = torch.gather(tmp_values, dim=-1, index=tmp_delay)
|
||
delays_agg = delays_agg + pattern * \
|
||
(tmp_corr[:, i].unsqueeze(1).unsqueeze(1).unsqueeze(1).repeat(1, head, channel, length))
|
||
return delays_agg
|
||
|
||
def time_delay_agg_full(self, values, corr):
|
||
"""
|
||
Standard version of Autocorrelation
|
||
"""
|
||
batch = values.shape[0]
|
||
head = values.shape[1]
|
||
channel = values.shape[2]
|
||
length = values.shape[3]
|
||
# index init
|
||
init_index = torch.arange(length).unsqueeze(0).unsqueeze(0).unsqueeze(0).repeat(batch, head, channel, 1).to(values.device)
|
||
# find top k
|
||
top_k = int(self.factor * math.log(length))
|
||
weights, delay = torch.topk(corr, top_k, dim=-1)
|
||
# update corr
|
||
tmp_corr = torch.softmax(weights, dim=-1)
|
||
# aggregation
|
||
tmp_values = values.repeat(1, 1, 1, 2)
|
||
delays_agg = torch.zeros_like(values).float()
|
||
for i in range(top_k):
|
||
tmp_delay = init_index + delay[..., i].unsqueeze(-1)
|
||
pattern = torch.gather(tmp_values, dim=-1, index=tmp_delay)
|
||
delays_agg = delays_agg + pattern * (tmp_corr[..., i].unsqueeze(-1))
|
||
return delays_agg
|
||
|
||
def forward(self, queries, keys, values, attn_mask):
|
||
B, L, H, E = queries.shape
|
||
_, S, _, D = values.shape
|
||
if L > S:
|
||
zeros = torch.zeros_like(queries[:, :(L - S), :]).float()
|
||
values = torch.cat([values, zeros], dim=1)
|
||
keys = torch.cat([keys, zeros], dim=1)
|
||
else:
|
||
values = values[:, :L, :, :]
|
||
keys = keys[:, :L, :, :]
|
||
|
||
# period-based dependencies
|
||
q_fft = torch.fft.rfft(queries.permute(0, 2, 3, 1).contiguous(), dim=-1)
|
||
k_fft = torch.fft.rfft(keys.permute(0, 2, 3, 1).contiguous(), dim=-1)
|
||
res = q_fft * torch.conj(k_fft)
|
||
corr = torch.fft.irfft(res, dim=-1)
|
||
|
||
# time delay agg
|
||
if self.training:
|
||
V = self.time_delay_agg_training(values.permute(0, 2, 3, 1).contiguous(), corr).permute(0, 3, 1, 2)
|
||
else:
|
||
V = self.time_delay_agg_inference(values.permute(0, 2, 3, 1).contiguous(), corr).permute(0, 3, 1, 2)
|
||
|
||
if self.output_attention:
|
||
return (V.contiguous(), corr.permute(0, 3, 1, 2))
|
||
else:
|
||
return (V.contiguous(), None)
|
||
|
||
|
||
class AutoCorrelationLayer(nn.Module):
|
||
def __init__(self, correlation, d_model, n_heads, d_keys=None,
|
||
d_values=None):
|
||
super(AutoCorrelationLayer, self).__init__()
|
||
|
||
d_keys = d_keys or (d_model // n_heads)
|
||
d_values = d_values or (d_model // n_heads)
|
||
|
||
self.inner_correlation = correlation
|
||
self.query_projection = nn.Linear(d_model, d_keys * n_heads)
|
||
self.key_projection = nn.Linear(d_model, d_keys * n_heads)
|
||
self.value_projection = nn.Linear(d_model, d_values * n_heads)
|
||
self.out_projection = nn.Linear(d_values * n_heads, d_model)
|
||
self.n_heads = n_heads
|
||
|
||
def forward(self, queries, keys, values, attn_mask):
|
||
B, L, _ = queries.shape
|
||
_, S, _ = keys.shape
|
||
H = self.n_heads
|
||
|
||
queries = self.query_projection(queries).view(B, L, H, -1)
|
||
keys = self.key_projection(keys).view(B, S, H, -1)
|
||
values = self.value_projection(values).view(B, S, H, -1)
|
||
|
||
out, attn = self.inner_correlation(
|
||
queries,
|
||
keys,
|
||
values,
|
||
attn_mask
|
||
)
|
||
out = out.view(B, L, -1)
|
||
|
||
return self.out_projection(out), attn
|
||
|
||
class my_Layernorm(nn.Module):
|
||
"""
|
||
Special designed layernorm for the seasonal part
|
||
"""
|
||
|
||
def __init__(self, channels):
|
||
super(my_Layernorm, self).__init__()
|
||
self.layernorm = nn.LayerNorm(channels)
|
||
|
||
def forward(self, x):
|
||
x_hat = self.layernorm(x)
|
||
bias = torch.mean(x_hat, dim=1).unsqueeze(1).repeat(1, x.shape[1], 1)
|
||
return x_hat - bias
|
||
|
||
|
||
class moving_avg(nn.Module):
|
||
"""
|
||
Moving average block to highlight the trend of time series
|
||
"""
|
||
|
||
def __init__(self, kernel_size, stride):
|
||
super(moving_avg, self).__init__()
|
||
self.kernel_size = kernel_size
|
||
self.avg = nn.AvgPool1d(kernel_size=kernel_size, stride=stride, padding=0)
|
||
|
||
def forward(self, x):
|
||
# padding on the both ends of time series
|
||
front = x[:, 0:1, :].repeat(1, (self.kernel_size - 1) // 2, 1)
|
||
end = x[:, -1:, :].repeat(1, (self.kernel_size - 1) // 2, 1)
|
||
x = torch.cat([front, x, end], dim=1)
|
||
x = self.avg(x.permute(0, 2, 1))
|
||
x = x.permute(0, 2, 1)
|
||
return x
|
||
|
||
|
||
class series_decomp(nn.Module):
|
||
"""
|
||
Series decomposition block
|
||
"""
|
||
|
||
def __init__(self, kernel_size):
|
||
super(series_decomp, self).__init__()
|
||
self.moving_avg = moving_avg(kernel_size, stride=1)
|
||
|
||
def forward(self, x):
|
||
moving_mean = self.moving_avg(x)
|
||
res = x - moving_mean
|
||
return res, moving_mean
|
||
|
||
|
||
class series_decomp_multi(nn.Module):
|
||
"""
|
||
Multiple Series decomposition block from FEDformer
|
||
"""
|
||
|
||
def __init__(self, kernel_size):
|
||
super(series_decomp_multi, self).__init__()
|
||
self.kernel_size = kernel_size
|
||
self.series_decomp = [series_decomp(kernel) for kernel in kernel_size]
|
||
|
||
def forward(self, x):
|
||
moving_mean = []
|
||
res = []
|
||
for func in self.series_decomp:
|
||
sea, moving_avg = func(x)
|
||
moving_mean.append(moving_avg)
|
||
res.append(sea)
|
||
|
||
sea = sum(res) / len(res)
|
||
moving_mean = sum(moving_mean) / len(moving_mean)
|
||
return sea, moving_mean
|
||
|
||
|
||
class EncoderLayer(nn.Module):
|
||
"""
|
||
Autoformer encoder layer with the progressive decomposition architecture
|
||
"""
|
||
|
||
def __init__(self, attention, d_model, d_ff=None, moving_avg=25, dropout=0.1, activation="relu"):
|
||
super(EncoderLayer, self).__init__()
|
||
d_ff = d_ff or 4 * d_model
|
||
self.attention = attention
|
||
self.conv1 = nn.Conv1d(in_channels=d_model, out_channels=d_ff, kernel_size=1, bias=False)
|
||
self.conv2 = nn.Conv1d(in_channels=d_ff, out_channels=d_model, kernel_size=1, bias=False)
|
||
self.decomp1 = series_decomp(moving_avg)
|
||
self.decomp2 = series_decomp(moving_avg)
|
||
self.dropout = nn.Dropout(dropout)
|
||
self.activation = F.relu if activation == "relu" else F.gelu
|
||
|
||
def forward(self, x, attn_mask=None):
|
||
new_x, attn = self.attention(
|
||
x, x, x,
|
||
attn_mask=attn_mask
|
||
)
|
||
x = x + self.dropout(new_x)
|
||
x, _ = self.decomp1(x)
|
||
y = x
|
||
y = self.dropout(self.activation(self.conv1(y.transpose(-1, 1))))
|
||
y = self.dropout(self.conv2(y).transpose(-1, 1))
|
||
res, _ = self.decomp2(x + y)
|
||
return res, attn
|
||
|
||
|
||
class Encoder(nn.Module):
|
||
"""
|
||
Autoformer encoder
|
||
"""
|
||
|
||
def __init__(self, attn_layers, conv_layers=None, norm_layer=None):
|
||
super(Encoder, self).__init__()
|
||
self.attn_layers = nn.ModuleList(attn_layers)
|
||
self.conv_layers = nn.ModuleList(conv_layers) if conv_layers is not None else None
|
||
self.norm = norm_layer
|
||
|
||
def forward(self, x, attn_mask=None):
|
||
attns = []
|
||
if self.conv_layers is not None:
|
||
for attn_layer, conv_layer in zip(self.attn_layers, self.conv_layers):
|
||
x, attn = attn_layer(x, attn_mask=attn_mask)
|
||
x = conv_layer(x)
|
||
attns.append(attn)
|
||
x, attn = self.attn_layers[-1](x)
|
||
attns.append(attn)
|
||
else:
|
||
for attn_layer in self.attn_layers:
|
||
x, attn = attn_layer(x, attn_mask=attn_mask)
|
||
attns.append(attn)
|
||
|
||
if self.norm is not None:
|
||
x = self.norm(x)
|
||
|
||
return x, attns
|
||
|
||
|
||
class DecoderLayer(nn.Module):
|
||
"""
|
||
Autoformer decoder layer with the progressive decomposition architecture
|
||
"""
|
||
|
||
def __init__(self, self_attention, cross_attention, d_model, c_out, d_ff=None,
|
||
moving_avg=25, dropout=0.1, activation="relu"):
|
||
super(DecoderLayer, self).__init__()
|
||
d_ff = d_ff or 4 * d_model
|
||
self.self_attention = self_attention
|
||
self.cross_attention = cross_attention
|
||
self.conv1 = nn.Conv1d(in_channels=d_model, out_channels=d_ff, kernel_size=1, bias=False)
|
||
self.conv2 = nn.Conv1d(in_channels=d_ff, out_channels=d_model, kernel_size=1, bias=False)
|
||
self.decomp1 = series_decomp(moving_avg)
|
||
self.decomp2 = series_decomp(moving_avg)
|
||
self.decomp3 = series_decomp(moving_avg)
|
||
self.dropout = nn.Dropout(dropout)
|
||
self.projection = nn.Conv1d(in_channels=d_model, out_channels=c_out, kernel_size=3, stride=1, padding=1,
|
||
padding_mode='circular', bias=False)
|
||
self.activation = F.relu if activation == "relu" else F.gelu
|
||
|
||
def forward(self, x, cross, x_mask=None, cross_mask=None):
|
||
x = x + self.dropout(self.self_attention(
|
||
x, x, x,
|
||
attn_mask=x_mask
|
||
)[0])
|
||
x, trend1 = self.decomp1(x)
|
||
x = x + self.dropout(self.cross_attention(
|
||
x, cross, cross,
|
||
attn_mask=cross_mask
|
||
)[0])
|
||
x, trend2 = self.decomp2(x)
|
||
y = x
|
||
y = self.dropout(self.activation(self.conv1(y.transpose(-1, 1))))
|
||
y = self.dropout(self.conv2(y).transpose(-1, 1))
|
||
x, trend3 = self.decomp3(x + y)
|
||
|
||
residual_trend = trend1 + trend2 + trend3
|
||
residual_trend = self.projection(residual_trend.permute(0, 2, 1)).transpose(1, 2)
|
||
return x, residual_trend
|
||
|
||
|
||
class Decoder(nn.Module):
|
||
"""
|
||
Autoformer encoder
|
||
"""
|
||
|
||
def __init__(self, layers, norm_layer=None, projection=None):
|
||
super(Decoder, self).__init__()
|
||
self.layers = nn.ModuleList(layers)
|
||
self.norm = norm_layer
|
||
self.projection = projection
|
||
|
||
def forward(self, x, cross, x_mask=None, cross_mask=None, trend=None):
|
||
for layer in self.layers:
|
||
x, residual_trend = layer(x, cross, x_mask=x_mask, cross_mask=cross_mask)
|
||
trend = trend + residual_trend
|
||
|
||
if self.norm is not None:
|
||
x = self.norm(x)
|
||
|
||
if self.projection is not None:
|
||
x = self.projection(x)
|
||
return x, trend
|
||
|
||
|
||
class Autoformer(nn.Module):
|
||
"""
|
||
Autoformer is the first method to achieve the series-wise connection,
|
||
with inherent O(LlogL) complexity
|
||
Paper link: https://openreview.net/pdf?id=I55UqU-M11y
|
||
"""
|
||
|
||
def __init__(self, task_name='short_term_forecast',seq_len=96, label_len=48, pred_len=96, enc_in=7, dec_in=7, c_out=1, e_layers=2, d_layers=1, n_heads=8,factor=3,
|
||
d_model=16, d_ff=32, des='Exp', expand=2, d_conv=4, top_k=5, embed='timeF',freq='h', dropout=0.1,num_kernels=6,
|
||
moving_avg=25,channel_independence=1, decomp_method='moving_avg', use_norm=1,
|
||
version='fourier', mode_select='random', modes=32, activation='gelu',seasonal_patterns='Monthly',
|
||
inverse=False, mask_rate=0.25, anomaly_ratio=0.25,output_attention=False,down_sampling_layers=0, down_sampling_window=1, down_sampling_method=None,
|
||
seg_len=48, num_workers=0, itr=1, train_epochs=100, batch_size=32, patience=3, learning_rate=0.0001, loss='MSE',
|
||
lradj='type1', use_amp=False, use_gpu=True, gpu=0, use_multi_gpu=False, devices='0,1,2,3', p_hidden_dims=[128, 128],
|
||
p_hidden_layers=2, use_dtw=False, augmentation_ratio=0, seed=2, jitter=False, scaling=False, permutation=False,
|
||
randompermutation=False, magwarp=False, timewarp=False, windowslice=False, windowwarp=False, rotation=False,
|
||
spawner=False, dtwwarp=False, shapedtwwarp=False, wdba=False, discdtw=False, discsdtw=False, extra_tag='',
|
||
**kwargs):
|
||
super(Autoformer, self).__init__()
|
||
self.task_name = task_name
|
||
self.seq_len = seq_len
|
||
self.label_len = label_len
|
||
self.pred_len = pred_len
|
||
self.output_attention = output_attention
|
||
|
||
# Decomp
|
||
kernel_size = moving_avg
|
||
self.decomp = series_decomp(kernel_size)
|
||
|
||
# Embedding
|
||
self.enc_embedding = DataEmbedding_wo_pos(enc_in, d_model, embed, freq, dropout)
|
||
# Encoder
|
||
self.encoder = Encoder(
|
||
[
|
||
EncoderLayer(
|
||
AutoCorrelationLayer(
|
||
AutoCorrelation(False, factor, attention_dropout=dropout,
|
||
output_attention=output_attention),
|
||
d_model, n_heads),
|
||
d_model,
|
||
d_ff,
|
||
moving_avg=moving_avg,
|
||
dropout=dropout,
|
||
activation=activation
|
||
) for l in range(e_layers)
|
||
],
|
||
norm_layer=my_Layernorm(d_model)
|
||
)
|
||
# Decoder
|
||
if self.task_name == 'long_term_forecast' or self.task_name == 'short_term_forecast':
|
||
self.dec_embedding = DataEmbedding_wo_pos(dec_in, d_model, embed, freq,dropout)
|
||
self.decoder = Decoder(
|
||
[
|
||
DecoderLayer(
|
||
AutoCorrelationLayer(
|
||
AutoCorrelation(True, factor, attention_dropout=dropout,
|
||
output_attention=False),
|
||
d_model, n_heads),
|
||
AutoCorrelationLayer(
|
||
AutoCorrelation(False, factor, attention_dropout=dropout,
|
||
output_attention=False),
|
||
d_model, n_heads),
|
||
d_model,
|
||
c_out,
|
||
d_ff,
|
||
moving_avg=moving_avg,
|
||
dropout=dropout,
|
||
activation=activation,
|
||
)
|
||
for l in range(d_layers)
|
||
],
|
||
norm_layer=my_Layernorm(d_model),
|
||
projection=nn.Linear(d_model, c_out, bias=True)
|
||
)
|
||
|
||
self.projection_final = nn.Linear(pred_len*enc_in, pred_len*c_out, bias=True)
|
||
def forward(self, x_enc, x_mark_enc, x_dec, x_mark_dec):
|
||
# decomp init
|
||
mean = torch.mean(x_enc, dim=1).unsqueeze(
|
||
1).repeat(1, self.pred_len, 1)
|
||
zeros = torch.zeros([x_dec.shape[0], self.pred_len,
|
||
x_dec.shape[2]], device=x_enc.device)
|
||
seasonal_init, trend_init = self.decomp(x_enc)
|
||
# decoder input
|
||
if self.label_len == 0:
|
||
trend_init = trend_init
|
||
seasonal_init = seasonal_init
|
||
else:
|
||
trend_init = torch.cat([trend_init[:, -self.label_len:, :], mean], dim=1)
|
||
seasonal_init = torch.cat([seasonal_init[:, -self.label_len:, :], zeros], dim=1)
|
||
|
||
# enc
|
||
enc_out = self.enc_embedding(x_enc, x_mark_enc)
|
||
enc_out, attns = self.encoder(enc_out, attn_mask=None)
|
||
# dec
|
||
dec_out = self.dec_embedding(seasonal_init, x_mark_dec)
|
||
seasonal_part, trend_part = self.decoder(dec_out, enc_out, x_mask=None, cross_mask=None,trend=trend_init)
|
||
# final
|
||
dec_out = trend_part + seasonal_part
|
||
dec_out=dec_out[:, -self.pred_len:, :]
|
||
dec_out=self.projection_final(dec_out.view(dec_out.shape[0], -1))
|
||
return dec_out
|
||
|
||
from pytorch_forecasting.models import BaseModel
|
||
from typing import Dict
|
||
|
||
class AutoFormerNetModel(BaseModel):
|
||
def __init__(self,seq_len=24, label_len=0, pred_len=1, enc_in=7, dec_in=7, c_out=1, e_layers=2, d_layers=1, factor=3,
|
||
d_model=16, d_ff=32, des='Exp', itr=1, top_k=5,embed='timeF',freq='h', dropout=0.1,num_kernels=6, **kwargs):
|
||
# saves arguments in signature to `.hparams` attribute, mandatory call - do not skip this
|
||
self.save_hyperparameters()
|
||
# pass additional arguments to BaseModel.__init__, mandatory call - do not skip this
|
||
super().__init__(**kwargs)
|
||
self.network = Autoformer(
|
||
seq_len=seq_len, label_len=label_len, pred_len=pred_len, enc_in=enc_in, dec_in=dec_in, c_out=c_out, e_layers=e_layers, d_layers=d_layers, factor=factor,
|
||
d_model=d_model, d_ff=d_ff, des=des, itr=itr, top_k=top_k, embed=embed, freq=freq, dropout=dropout, num_kernels=num_kernels
|
||
)
|
||
self.label_len=label_len
|
||
|
||
# 修改，锂电池预测
|
||
def forward(self, x: Dict[str, torch.Tensor]) -> Dict[str, torch.Tensor]:
|
||
|
||
x_enc = x["encoder_cont"][:,:,:-1] # torch.Size([100, 10, 9])
|
||
x_dec = torch.cat([x["encoder_cont"][:, -self.label_len:, :-1], x["decoder_cont"][:,:,:-1]],
|
||
dim=1) # torch.Size([100, 11, 9])
|
||
# 输出
|
||
prediction = self.network(x_enc=x_enc,x_mark_enc=None,x_dec=x_dec,x_mark_dec=None)
|
||
# 输出rescale， rescale predictions into target space
|
||
prediction = self.transform_output(prediction, target_scale=x["target_scale"])
|
||
|
||
# 返回一个字典，包含输出结果（prediction）
|
||
return self.to_network_output(prediction=prediction)
|
||
|
||
|
||
if __name__=='__main__':
|
||
N,L,C=100,96,15
|
||
label_len = 16
|
||
c_out = 1
|
||
pred_len=16
|
||
x_enc=torch.ones((N,L,C))
|
||
x_mark_enc=torch.ones((N, L, 4))
|
||
x_dec = torch.ones((N, pred_len, C))
|
||
x_mark_dec=torch.ones((N, pred_len, 4))
|
||
model=Autoformer(seq_len=L, enc_in=C, dec_in=C, label_len = label_len, pred_len=pred_len, c_out=1) # pred_len 被限制了
|
||
out=model(x_enc=x_enc, x_mark_enc=x_mark_enc, x_dec=x_dec, x_mark_dec=x_mark_dec)
|
||
print(out.shape) |