DeepTime，是一個結合使用元學習的深度時間指數模型。通過使用元學習公式來預測未來，以應對時間序列中的常見問題（協變量偏移和條件分布偏移——非平穩）。該模型是時間序列預測的元學習公式協同作用的一個很好的例子。
DeepTime架構
DeepTime元件
DeepTime中有三種類型的層:
- 嶺回歸
- 多層感覺機(MLP)
- 随機傅裡葉特征
讓我們看看這些層在做什麼:
嶺回歸
多層感覺機(MLP)
這些是在神經網絡(nn)中使用的線性回歸公式。然後使用了一個ReLU函數激活。這些層非常适合将時間指數映射到該時間指數的時間序列值。公式如下:
随機的傅裡葉層
随機傅裡葉允許mlp學習高頻模式。盡管随機傅裡葉層需要為每個任務和資料集找到不同的超參數(隻是為了不過度拟合或不足拟合)，但作者通過将各種傅裡葉基函數與各種尺度參數相結合來限制這種計算。
DeepTIME架構
在每個任務中，選擇一個時間序列，然後将其分為主幹視窗(綠色)和預測視窗(藍色)兩部分。然後，然後他們通過兩個彼此共享資訊并與元參數關聯的元模型。 在上圖描述的架構上訓練模型後，計算損失函數并嘗試将其最小化。
其他時間序列預測模型的差別
DeepTIME是一個時間指數模型，就像Prophet，高斯過程等，而最近比較突出的模型如N-HiTS, Autoformer, DeepAR, Informer等都是曆史價值模型。
當我們說時間序列的時間指數模型時，确切的意思是預測絕對随時間變化(它考慮了目前的時間指數特征)。另一方面，曆史價值模型使用以前的事件來預測未來。這個公式能讓你更清楚。：）
它包含了元學習公式，這意味着這個模型可以學會如何學習。由于它是一個時間指數模型，可以在元學習中表現出更好的樣本效率。
它采用直接多步估計(DMS)的方法（DMS模型一次直接預測幾個資料點）。另外通過多步疊代（IMS），它隻預測下一個值，然後使用它來預測下一個資料點，這與ARIMA、DeepAR等相同。
元學習給時間序列預測帶來了什麼？
- 更好的任務泛化
- 符合附近時間步長遵循局部平穩分布的假設。
- 還包含了相似的時間點将具有相似的特征的假設。
模型如何預測
在每一次訓練時，将資料分為兩個視窗(通過使用第一個視窗預測第二個視窗)。這裡為了簡單起見使用PyTorch Lightning簡化訓練過程。
import numpy as np
import gin
import pytorch_lightning as pl
from models import get_model
import random
import torch
import torch.nn.functional as F
from torch import optim
import math
from utils import Checkpoint, default_device, to_tensor
@gin.configurable
class DeepTimeTrainer(pl.LightningModule):
def __init__(self,
lr,
lambda_lr,
weight_decay,
warmup_epochs,
random_seed,
T_max,
eta_min,
dim_size,
datetime_feats,
):
gin.parse_config_file('/home/reza/Projects/PL_DeepTime/DeepTime/config/config.gin')
super(DeepTimeTrainer, self).__init__()
self.lr = lr
self.lambda_lr = lambda_lr
self.weight_decay = weight_decay
self.warmup_epochs = warmup_epochs
self.random_seed = random_seed
self.lr = lr
self.lambda_lr = lambda_lr
self.weight_decay = weight_decay
self.T_max = T_max
self.warmup_epochs = warmup_epochs
self.eta_min = eta_min
self.model = get_model(
model_type='deeptime',
dim_size=dim_size,
datetime_feats=datetime_feats
)
def on_fit_start(self):
torch.manual_seed(self.random_seed)
np.random.seed(self.random_seed)
random.seed(self.random_seed)
def training_step(self, batch, batch_idx):
x, y, x_time, y_time = map(to_tensor, batch)
forecast = self.model(x, x_time, y_time)
if isinstance(forecast, tuple):
# for models which require reconstruction + forecast loss
loss = F.mse_loss(forecast[0], x) + \
F.mse_loss(forecast[1], y)
else:
loss = F.mse_loss(forecast, y)
self.log('train_loss', loss, prog_bar=True, on_epoch=True)
return {'loss': loss, 'train_loss': loss, }
def training_epoch_end(self, outputs):
avg_train_loss = torch.stack([x["train_loss"] for x in outputs]).mean()
self.log('avg_train_loss', avg_train_loss, on_epoch=True, sync_dist=True)
def validation_step(self, batch, batch_idx):
x, y, x_time, y_time = map(to_tensor, batch)
forecast = self.model(x, x_time, y_time)
if isinstance(forecast, tuple):
# for models which require reconstruction + forecast loss
loss = F.mse_loss(forecast[0], x) + \
F.mse_loss(forecast[1], y)
else:
loss = F.mse_loss(forecast, y)
self.log('val_loss', loss, prog_bar=True, on_epoch=True)
return {'val_loss': loss}
def validation_epoch_end(self, outputs):
return outputs
def test_step(self, batch, batch_idx):
x, y, x_time, y_time = map(to_tensor, batch)
forecast = self.model(x, x_time, y_time)
if isinstance(forecast, tuple):
# for models which require reconstruction + forecast loss
loss = F.mse_loss(forecast[0], x) + \
F.mse_loss(forecast[1], y)
else:
loss = F.mse_loss(forecast, y)
self.log('test_loss', loss, prog_bar=True, on_epoch=True)
return {'test_loss': loss}
def test_epoch_end(self, outputs):
return outputs
@gin.configurable
def configure_optimizers(self):
group1 = [] # lambda
group2 = [] # no decay
group3 = [] # decay
no_decay_list = ('bias', 'norm',)
for param_name, param in self.model.named_parameters():
if '_lambda' in param_name:
group1.append(param)
elif any([mod in param_name for mod in no_decay_list]):
group2.append(param)
else:
group3.append(param)
optimizer = optim.Adam([
{'params': group1, 'weight_decay': 0, 'lr': self.lambda_lr, 'scheduler': 'cosine_annealing'},
{'params': group2, 'weight_decay': 0, 'scheduler': 'cosine_annealing_with_linear_warmup'},
{'params': group3, 'scheduler': 'cosine_annealing_with_linear_warmup'}
], lr=self.lr, weight_decay=self.weight_decay)
scheduler_fns = []
for param_group in optimizer.param_groups:
scheduler = param_group['scheduler']
if scheduler == 'none':
fn = lambda T_cur: 1
elif scheduler == 'cosine_annealing':
lr = eta_max = param_group['lr']
fn = lambda T_cur: (self.eta_min + 0.5 * (eta_max - self.eta_min) * (
1.0 + math.cos(
(T_cur - self.warmup_epochs) / (self.T_max - self.warmup_epochs) * math.pi))) / lr
elif scheduler == 'cosine_annealing_with_linear_warmup':
lr = eta_max = param_group['lr']
fn = lambda T_cur: T_cur / self.warmup_epochs if T_cur < self.warmup_epochs else (self.eta_min + 0.5 * (
eta_max - self.eta_min) * (1.0 + math.cos(
(T_cur - self.warmup_epochs) / (self.T_max - self.warmup_epochs) * math.pi))) / lr
else:
raise ValueError(f'No such scheduler, {scheduler}')
scheduler_fns.append(fn)
scheduler = optim.lr_scheduler.LambdaLR(optimizer, lr_lambda=scheduler_fns)
return {'optimizer': optimizer, 'lr_scheduler': scheduler}
def forward(self, batch, z_0=None):
z_0 = None
Y = batch['Y'].to(default_device)
sample_mask = batch['sample_mask'].to(default_device)
available_mask = batch['available_mask'].to(default_device)
# Forecasting
forecasting_mask = available_mask.clone()
if self.n_time_out > 0:
forecasting_mask[:, 0, -self.n_time_out:] = 0
Y, Y_hat, z = self.model(Y=Y, mask=forecasting_mask, idxs=None, z_0=z_0)
if self.n_time_out > 0:
Y = Y[:, :, -self.n_time_out:]
Y_hat = Y_hat[:, :, -self.n_time_out:]
sample_mask = sample_mask[:, :, -self.n_time_out:]
return Y, Y_hat, sample_mask, z 作者在合成資料集和真實世界資料集上進行了廣泛的實驗，表明DeepTime具有極具競争力的性能，在基于MSE的多元預測基準的24個實驗中，有20個獲得了最先進的結果。
作者：Reza Yazdanfar