PatchTST

https://github.com/yuqinie98/patchtst
Admin
PatchTST is an official implementation of a Transformer-based model for long-term time series...
Tokens:6,694
Snippets:20
Trust Score:7.7
License:Apache-2.0
Update:1 year ago
Show doc for...
Tokens:
Raw
### Install Requirements

Source: https://github.com/yuqinie98/patchtst/blob/main/README.md

Installs the necessary Python packages for the PatchTST project. This is typically the first step before running any training or fine-tuning scripts.

```bash
pip install -r requirements.txt
```

--------------------------------

### Single-Step Time Series Evaluation Example

Source: https://github.com/yuqinie98/patchtst/blob/main/PatchTST_supervised/Formers/Pyraformer/README.md

Example command for evaluating a single-step time series model. It specifies the data path, dataset name, and enables evaluation mode.

```python
python single_step_main.py -data_path data/elect/ -dataset elect -eval
```

--------------------------------

### Long-Range Time Series Evaluation Example

Source: https://github.com/yuqinie98/patchtst/blob/main/PatchTST_supervised/Formers/Pyraformer/README.md

Example command for evaluating a long-range time series model. It specifies dataset, input size, prediction steps, number of attention heads, and enables evaluation mode.

```python
python long_range_main.py -data ETTh1 -input_size 168 -predict_step 168 -n_head 6 -eval
```

--------------------------------

### Setup Learner for PatchTST

Source: https://context7.com/yuqinie98/patchtst/llms.txt

Configures the learner object for training the PatchTST model. This includes defining the loss function, callbacks (like RevInCB and PatchCB), and evaluation metrics (MSE, MAE).

```python
loss_func = torch.nn.MSELoss(reduction='mean')
cbs = [RevInCB(dls.vars, denorm=True), PatchCB(patch_len=args.patch_len, stride=args.stride)]
learner = Learner(dls, model, loss_func, lr=args.lr, cbs=cbs, metrics=[mse, mae])
```

--------------------------------

### Setup PatchTST Model for Linear Probing

Source: https://context7.com/yuqinie98/patchtst/llms.txt

Initializes the PatchTST model for linear probing. This involves setting specific arguments to indicate linear probing and no fine-tuning, then defining the model architecture with given parameters.

```python
args.is_linear_probe = 1
args.is_finetune = 0

model = PatchTST(
    c_in=dls.vars,
    target_dim=args.target_points,
    patch_len=args.patch_len,
    stride=args.stride,
    num_patch=num_patch,
    n_layers=args.n_layers,
    n_heads=args.n_heads,
    d_model=args.d_model,
    shared_embedding=True,
    d_ff=args.d_ff,
    dropout=args.dropout,
    head_dropout=args.head_dropout,
    head_type='prediction',
    res_attention=False
)
```

--------------------------------

### Run Weather Dataset Training (Supervised)

Source: https://github.com/yuqinie98/patchtst/blob/main/README.md

Executes the training script for the PatchTST/42 model on the weather dataset for multivariate forecasting. This command initiates the supervised learning process.

```bash
sh ./scripts/PatchTST/weather.sh
```

--------------------------------

### Pre-train PatchTST/64 Model (Self-Supervised)

Source: https://github.com/yuqinie98/patchtst/blob/main/README.md

Initiates the pre-training process for the PatchTST/64 model on the 'ettm1' dataset with a specified mask ratio. This script is used for the self-supervised learning part of the project.

```python
python patchtst_pretrain.py --dset ettm1 --mask_ratio 0.4
```

--------------------------------

### Pre-training PatchTST Model Configuration and Training

Source: https://context7.com/yuqinie98/patchtst/llms.txt

Configures and trains the PatchTST model for pre-training using masked patch prediction. It sets up data loaders, defines the model architecture with a pre-training head, initializes callbacks for RevIn and PatchMask, and utilizes a Learner for the training process. The pre-trained model weights are then saved.

```python
import argparse
import torch
from src.models.patchTST import PatchTST
from src.learner import Learner
from src.callback.patch_mask import PatchMaskCB
from src.callback.transforms import RevInCB
from datautils import get_dls

# Pre-training configuration
args = argparse.Namespace(
    dset='etth1',
    context_points=512,    # Long context window for pre-training
    target_points=96,
    batch_size=64,
    num_workers=4,
    scaler='standard',
    features='M',

    # Patch configuration
    patch_len=12,
    stride=12,

    # Model architecture
    n_layers=3,
    n_heads=16,
    d_model=128,
    d_ff=512,
    dropout=0.2,
    head_dropout=0.2,

    # Pre-training specific
    mask_ratio=0.4,           # Mask 40% of patches
    n_epochs_pretrain=100,
    lr=1e-4,
    revin=1,

    # Saving
    pretrained_model_id=1,
    model_type='based_model'
)

args.save_pretrained_model = f'patchtst_pretrained_cw{args.context_points}_patch{args.patch_len}_stride{args.stride}_epochs{args.n_epochs_pretrain}_mask{args.mask_ratio}_model{args.pretrained_model_id}'
args.save_path = f'saved_models/{args.dset}/masked_patchtst/{args.model_type}/'

# Load data
dls = get_dls(args)

# Calculate number of patches
num_patch = (max(args.context_points, args.patch_len) - args.patch_len) // args.stride + 1

# Initialize model for pre-training
model = PatchTST(
    c_in=dls.vars,
    target_dim=args.target_points,
    patch_len=args.patch_len,
    stride=args.stride,
    num_patch=num_patch,
    n_layers=args.n_layers,
    n_heads=args.n_heads,
    d_model=args.d_model,
    shared_embedding=True,
    d_ff=args.d_ff,
    dropout=args.dropout,
    head_dropout=args.head_dropout,
    act='relu',
    head_type='pretrain',      # Pretrain head for reconstruction
    res_attention=False
)

print(f'Model parameters: {sum(p.numel() for p in model.parameters() if p.requires_grad):,}')

# Setup callbacks
cbs = [
    RevInCB(dls.vars, denorm=False) if args.revin else None,
    PatchMaskCB(patch_len=args.patch_len, stride=args.stride, mask_ratio=args.mask_ratio)
]
cbs = [cb for cb in cbs if cb is not None]

# Initialize learner
loss_func = torch.nn.MSELoss(reduction='mean')
learner = Learner(dls, model, loss_func, lr=args.lr, cbs=cbs)

# Pre-train the model
learner.fit_one_cycle(n_epochs=args.n_epochs_pretrain, lr_max=args.lr)

# Save pre-trained model
torch.save(model.state_dict(), args.save_path + args.save_pretrained_model + '.pth')
print(f'Pre-training completed. Model saved to {args.save_path}')

```

--------------------------------

### Train PatchTST Model for Forecasting (Python)

Source: https://context7.com/yuqinie98/patchtst/llms.txt

This Python script configures and trains a PatchTST model for time series forecasting. It utilizes argparse for parameter handling and the Exp_Main class for experiment execution. The script sets up data paths, model hyperparameters, training parameters, and system configurations, then proceeds to train and test the model.

```python
import argparse
import torch
from exp.exp_main import Exp_Main

# Configure model and training parameters
args = argparse.Namespace(
    random_seed=2021,
    is_training=1,
    model='PatchTST',
    model_id='weather_96_96',

    # Data configuration
    data='custom',
    root_path='./dataset/',
    data_path='weather.csv',
    features='M',  # M: multivariate->multivariate, S: univariate->univariate
    target='OT',
    freq='h',

    # Forecasting task
    seq_len=96,      # input sequence length
    label_len=48,    # start token length
    pred_len=96,     # prediction horizon

    # PatchTST hyperparameters
    patch_len=16,           # length of each patch
    stride=8,               # stride between patches
    padding_patch='end',    # padding strategy
    revin=1,                # RevIN normalization
    affine=0,               # RevIN affine parameters
    subtract_last=0,        # normalization mode
    decomposition=0,        # time series decomposition
    individual=0,           # individual head per channel

    # Model architecture
    enc_in=21,        # number of input channels
    dec_in=21,
    c_out=21,
    d_model=128,      # model dimension
    n_heads=16,       # attention heads
    e_layers=3,       # encoder layers
    d_layers=1,
    d_ff=256,         # feedforward dimension
    dropout=0.2,
    fc_dropout=0.2,
    head_dropout=0.0,

    # Training
    batch_size=128,
    learning_rate=0.0001,
    train_epochs=100,
    patience=10,
    lradj='type3',
    pct_start=0.3,
    use_amp=False,

    # System
    use_gpu=True,
    gpu=0,
    use_multi_gpu=False,
    devices='0,1',
    num_workers=10,
    checkpoints='./checkpoints/',
    test_flop=False
)

# Set random seeds
import random, numpy as np
random.seed(args.random_seed)
torch.manual_seed(args.random_seed)
np.random.seed(args.random_seed)

# Initialize experiment
exp = Exp_Main(args)

# Create experiment setting identifier
setting = f'{args.model_id}_{args.model}_{args.data}_ft{args.features}_sl{args.seq_len}_ll{args.label_len}_pl{args.pred_len}_dm{args.d_model}_nh{args.n_heads}_el{args.e_layers}'

# Train the model
print('>>>>>>>start training>>>>>>>>>>>>>>>>>>>>>>>>>>>')
model = exp.train(setting)

# Test the model
print('>>>>>>>testing<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<<')
exp.test(setting)

# Results are saved to result.txt with MSE, MAE, RSE metrics
# Model checkpoint saved to ./checkpoints/{setting}/checkpoint.pth

```

--------------------------------

### Fine-tuning Pre-trained PatchTST Model for Forecasting

Source: https://context7.com/yuqinie98/patchtst/llms.txt

Loads a pre-trained PatchTST model and fine-tunes it for a forecasting task on a new dataset. It initializes the model with a prediction head, transfers weights from the pre-trained model, and sets up callbacks for RevIn and patching. The fine-tuning process uses a Learner with gradual unfreezing, followed by evaluation on the test set.

```python
import torch
from src.models.patchTST import PatchTST
from src.learner import Learner, transfer_weights
from src.callback.patch_mask import PatchCB
from src.callback.transforms import RevInCB
from src.metrics import mse, mae
from datautils import get_dls
import argparse

# Fine-tuning configuration
args = argparse.Namespace(
    dset_finetune='ettm1',     # Target dataset (can differ from pre-train)
    context_points=512,
    target_points=96,
    batch_size=64,
    num_workers=4,
    features='M',

    # Same architecture as pre-training
    patch_len=12,
    stride=12,
    n_layers=3,
    n_heads=16,
    d_model=128,
    d_ff=256,
    dropout=0.2,
    head_dropout=0.2,
    revin=1,

    # Fine-tuning
    is_finetune=1,             # 1: full fine-tune, 0: linear probe
    is_linear_probe=0,
    n_epochs_finetune=20,
    lr=1e-4,

    # Load pre-trained weights
    pretrained_model='saved_models/etth1/masked_patchtst/based_model/patchtst_pretrained_cw512_patch12_stride12_epochs100_mask0.4_model1.pth'
)

# Load target dataset
args.dset = args.dset_finetune
dls = get_dls(args)

# Initialize model with prediction head
num_patch = (max(args.context_points, args.patch_len) - args.patch_len) // args.stride + 1

model = PatchTST(
    c_in=dls.vars,
    target_dim=args.target_points,
    patch_len=args.patch_len,
    stride=args.stride,
    num_patch=num_patch,
    n_layers=args.n_layers,
    n_heads=args.n_heads,
    d_model=args.d_model,
    shared_embedding=True,
    d_ff=args.d_ff,
    dropout=args.dropout,
    head_dropout=args.head_dropout,
    act='relu',
    head_type='prediction',    # Prediction head for forecasting
    res_attention=False
)

# Transfer pre-trained weights
model = transfer_weights(args.pretrained_model, model)
print('Pre-trained weights loaded successfully')

# Setup for fine-tuning
loss_func = torch.nn.MSELoss(reduction='mean')
cbs = [
    RevInCB(dls.vars, denorm=True) if args.revin else None,
    PatchCB(patch_len=args.patch_len, stride=args.stride)
]
cbs = [cb for cb in cbs if cb is not None]

learner = Learner(dls, model, loss_func, lr=args.lr, cbs=cbs, metrics=[mse, mae])

# Fine-tune with gradual unfreezing
learner.fine_tune(n_epochs=args.n_epochs_finetune, base_lr=args.lr, freeze_epochs=10)

# Evaluate on test set
test_results = learner.test(dls.test, scores=[mse, mae])
print(f'Test MSE: {test_results[2][0]:.4f}, MAE: {test_results[2][1]:.4f}')

```

--------------------------------

### Import PatchTST Backbone (Python)

Source: https://context7.com/yuqinie98/patchtst/llms.txt

This Python snippet demonstrates how to import the PatchTST backbone model from the layers module. It is a foundational step for utilizing the PatchTST architecture within a PyTorch environment.

```python
import torch
import torch.nn as nn
from layers.PatchTST_backbone import PatchTST_backbone

```

--------------------------------

### Load Pre-trained Weights for PatchTST

Source: https://context7.com/yuqinie98/patchtst/llms.txt

Loads pre-trained weights into the initialized PatchTST model. This step is crucial for transfer learning scenarios where a model is fine-tuned on a new dataset.

```python
model = transfer_weights(args.pretrained_model, model)
```

--------------------------------

### Python TimeSeriesDataset for Data Loading and Preprocessing

Source: https://context7.com/yuqinie98/patchtst/llms.txt

Implements a custom PyTorch Dataset for time series data, handling feature selection, scaling using StandardScaler, and creating input/target sequences for forecasting tasks. It supports different dataset splits (train, val, test) and configurations for sequence length, label length, and prediction length. Dependencies include pandas, numpy, torch, and sklearn.

```python
import pandas as pd
import numpy as np
from torch.utils.data import Dataset, DataLoader
from sklearn.preprocessing import StandardScaler

class TimeSeriesDataset(Dataset):
    """ETT (Electricity Transformer Temperature) dataset loader"""
    def __init__(self, root_path='./dataset/', data_path='ETTh1.csv',
                 flag='train', seq_len=96, label_len=48, pred_len=96,
                 features='M', target='OT', scale=True):

        self.seq_len = seq_len      # Input sequence length
        self.label_len = label_len  # Start token length
        self.pred_len = pred_len    # Prediction length
        self.features = features    # M, S, or MS
        self.target = target        # Target variable name
        self.scale = scale

        # Dataset splits: 12 months train, 4 months val, 4 months test
        type_map = {'train': 0, 'val': 1, 'test': 2}
        self.set_type = type_map[flag]

        # Load and preprocess data
        df_raw = pd.read_csv(f"{root_path}/{data_path}")

        # Define borders for train/val/test
        border1s = [0, 12*30*24 - seq_len, 12*30*24 + 4*30*24 - seq_len]
        border2s = [12*30*24, 12*30*24 + 4*30*24, 12*30*24 + 8*30*24]
        border1, border2 = border1s[self.set_type], border2s[self.set_type]

        # Select features
        if features == 'M' or features == 'MS':
            df_data = df_raw[df_raw.columns[1:]]  # All columns except date
        elif features == 'S':
            df_data = df_raw[[target]]  # Single target column

        # Standardize using training set statistics
        if scale:
            self.scaler = StandardScaler()
            train_data = df_data[border1s[0]:border2s[0]]
            self.scaler.fit(train_data.values)
            data = self.scaler.transform(df_data.values)
        else:
            data = df_data.values

        self.data_x = data[border1:border2]
        self.data_y = data[border1:border2]

    def __getitem__(self, index):
        s_begin = index
        s_end = s_begin + self.seq_len
        r_begin = s_end - self.label_len
        r_end = r_begin + self.label_len + self.pred_len

        seq_x = self.data_x[s_begin:s_end]       # Input: [96, 7]
        seq_y = self.data_y[r_begin:r_end]       # Target: [144, 7]

        return seq_x, seq_y

    def __len__(self):
        return len(self.data_x) - self.seq_len - self.pred_len + 1

# Create dataloaders
train_dataset = TimeSeriesDataset(flag='train', seq_len=96, pred_len=96)
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True, num_workers=4)

# Iterate through batches
for batch_x, batch_y in train_loader:
    print(f"Input shape: {batch_x.shape}")   # [32, 96, 7]
    print(f"Target shape: {batch_y.shape}")  # [32, 144, 7]
    break

```

--------------------------------

### Fine-tune PatchTST Model (Self-Supervised)

Source: https://github.com/yuqinie98/patchtst/blob/main/README.md

Performs the fine-tuning step for a pre-trained PatchTST model on a specified dataset. This script supports both linear probing and fine-tuning the entire network.

```python
python patchtst_finetune.py --dset ettm1 --pretrained_model <model_name>
```

--------------------------------

### Run LongExp Forecasting with Shell Script (Bash)

Source: https://context7.com/yuqinie98/patchtst/llms.txt

This bash script automates the training and evaluation of the PatchTST model for long-term time series forecasting across different prediction horizons. It sets up necessary directories, configures parameters like sequence length and model name, and iterates through various prediction lengths, logging the output for each run.

```bash
#!/bin/bash

# Setup directories
mkdir -p ./logs ./logs/LongForecasting

# Configuration
seq_len=96
pred_len=96
model_name=PatchTST
data_path=./dataset/weather.csv

# Train and evaluate for multiple prediction horizons
for pred_len in 96 192 336 720
do
    python -u run_longExp.py \
      --random_seed 2021 \
      --is_training 1 \
      --root_path ./dataset/ \
      --data_path weather.csv \
      --model_id weather_$seq_len'_'$pred_len \
      --model $model_name \
      --data custom \
      --features M \
      --seq_len $seq_len \
      --pred_len $pred_len \
      --enc_in 21 \
      --e_layers 3 \
      --n_heads 16 \
      --d_model 128 \
      --d_ff 256 \
      --dropout 0.2 \
      --fc_dropout 0.2 \
      --head_dropout 0 \
      --patch_len 16 \
      --stride 8 \
      --des 'Exp' \
      --train_epochs 100 \
      --lradj 'type3' \
      --itr 1 \
      --batch_size 128 \
      --learning_rate 0.0001 \
      >logs/LongForecasting/$model_name'_weather_'$seq_len'_'$pred_len.log
done

# View results
cat result.txt

```

--------------------------------

### Evaluate PatchTST Model on Test Data

Source: https://context7.com/yuqinie98/patchtst/llms.txt

Evaluates the trained PatchTST model on the test dataset. It calculates specified metrics such as Mean Squared Error (MSE) and Mean Absolute Error (MAE) and prints the results.

```python
test_results = learner.test(dls.test, scores=[mse, mae])
print(f'Linear probe - Test MSE: {test_results[2][0]:.4f}, MAE: {test_results[2][1]:.4f}')
```

--------------------------------

### PatchTST Model Implementation in Python

Source: https://context7.com/yuqinie98/patchtst/llms.txt

The PatchTSTModel class implements a transformer-based architecture for time series forecasting. It processes input time series by patching, transforming, and reconstructing, with support for configurable parameters like patch length, stride, and transformer layers. Dependencies include PyTorch and argparse.

```python
import torch
import torch.nn as nn
import argparse

class PatchTST_backbone(nn.Module):
    def __init__(self,
            c_in,
            context_window,
            target_window,
            patch_len,
            stride,
            max_seq_len,
            n_layers=3,
            d_model=128,
            n_heads=16,
            d_ff=256,
            dropout=0.0,
            fc_dropout=0.0,
            head_dropout=0.0,
            revin=True,
            affine=False,
            individual=False):
        super().__init__()
        
        self.revin = revin
        if revin:
            self.revin_layer = RevIN(c_in, affine=affine, subtract_last=False)
            
        self.patch_embedding = nn.Linear(patch_len, d_model)
        self.patch_folding = nn.Linear(d_model, patch_len)

        self.encoder = nn.ModuleList([
            nn.TransformerEncoderLayer(d_model=d_model,
                                        nhead=n_heads,
                                        dim_feedforward=d_ff,
                                        dropout=dropout,
                                        activation='gelu',
                                        batch_first=True,
                                        norm_first=True)
            for _ in range(n_layers)
        ])
        self.dropout = nn.Dropout(dropout)
        self.fc_dropout = nn.Dropout(fc_dropout)
        self.fc_out = nn.Linear(d_model, target_window)
        self.fc_norm = nn.LayerNorm(d_model)
        if head_dropout:
            self.head_dropout = nn.Dropout(head_dropout)
        else:
            self.head_dropout = None
        self.projection = nn.Linear(d_model, target_window, bias=True)

    def forward(self, x):
        # x: [Batch, Input_length, Channels]
        B, T, N = x.size()
        if self.revin:
            x = self.revin_layer(x, 'norm')

        # patch and positional encode
        x = x.unfold(dimension=1, size=self.patch_size, step=self.stride)
        x = x.permute(0, 1, 2, 3) # B, #patches, N, patch_len
        x = self.patch_embedding(x) # B, #patches, N, d_model
        x = x.reshape(B, -1, N * self.d_model) # B, #patches, N*d_model

        # Transformer
        for layer in self.encoder:
            x = layer(x)
        
        x = self.fc_norm(x)
        x = self.fc_out(x)
        x = self.projection(x)
        
        # Output
        x = x.permute(0, 2, 1) # B, N, T
        if self.revin:
            x = self.revin_layer(x, 'denorm')
        return x

class PatchTSTModel(nn.Module):
    def __init__(self, configs):
        super().__init__()

        # Model configuration
        c_in = configs.enc_in              # 21 channels for weather data
        context_window = configs.seq_len   # 96 time steps
        target_window = configs.pred_len   # 96 time steps forecast
        patch_len = configs.patch_len      # 16
        stride = configs.stride            # 8

        # Initialize backbone
        self.model = PatchTST_backbone(
            c_in=c_in,
            context_window=context_window,
            target_window=target_window,
            patch_len=patch_len,
            stride=stride,
            max_seq_len=1024,
            n_layers=configs.e_layers,
            d_model=configs.d_model,
            n_heads=configs.n_heads,
            d_ff=configs.d_ff,
            dropout=configs.dropout,
            fc_dropout=configs.fc_dropout,
            head_dropout=configs.head_dropout,
            revin=configs.revin,
            affine=configs.affine,
            individual=configs.individual
        )

    def forward(self, x):
        # x: [Batch, Input_length, Channels] -> [32, 96, 21]
        x = x.permute(0, 2, 1)    # [Batch, Channels, Input_length]
        x = self.model(x)          # [Batch, Channels, Pred_length]
        x = x.permute(0, 2, 1)    # [Batch, Pred_length, Channels]
        return x                   # [32, 96, 21]

# Example usage
configs = argparse.Namespace(
    enc_in=21, seq_len=96, pred_len=96,
    patch_len=16, stride=8, e_layers=3,
    d_model=128, n_heads=16, d_ff=256,
    dropout=0.2, fc_dropout=0.2, head_dropout=0.0,
    revin=1, affine=0, individual=0
)

model = PatchTSTModel(configs)
batch_input = torch.randn(32, 96, 21)  # 32 samples, 96 timesteps, 21 features
forecast = model(batch_input)           # Output: [32, 96, 21]
print(f"Forecast shape: {forecast.shape}")  # torch.Size([32, 96, 21])
```

--------------------------------

### Train PatchTST Model using Linear Probing

Source: https://context7.com/yuqinie98/patchtst/llms.txt

Trains only the prediction head of the PatchTST model using the linear probing technique. This method freezes the backbone weights and trains the final layers for the specific downstream task.

```python
learner.linear_probe(n_epochs=args.n_epochs_finetune, base_lr=args.lr)
```

--------------------------------

### Python Forecasting Evaluation Metrics Calculation

Source: https://context7.com/yuqinie98/patchtst/llms.txt

Calculates common forecasting metrics including Mean Squared Error (MSE), Mean Absolute Error (MAE), Root Mean Squared Error (RMSE), Root Relative Squared Error (RSE), and Correlation coefficient (CORR). This function takes predicted values and ground truth as NumPy arrays and returns a dictionary of these metrics. Dependencies include numpy.

```python
import numpy as np

def evaluate_forecast(predictions, ground_truth):
    """Calculate comprehensive forecasting metrics"""

    # Mean Squared Error
    mse = np.mean((predictions - ground_truth) ** 2)

    # Mean Absolute Error
    mae = np.mean(np.abs(predictions - ground_truth))

    # Root Mean Squared Error
    rmse = np.sqrt(mse)

    # Root Relative Squared Error
    rse = np.sqrt(np.sum((ground_truth - predictions) ** 2)) /
          np.sqrt(np.sum((ground_truth - ground_truth.mean()) ** 2))

    # Correlation coefficient
    u = ((ground_truth - ground_truth.mean(0)) *
         (predictions - predictions.mean(0))).sum(0)
    d = np.sqrt(((ground_truth - ground_truth.mean(0)) ** 2 *
                 (predictions - predictions.mean(0)) ** 2).sum(0))
    corr = 0.01 * (u / (d + 1e-12)).mean()

    return {
        'MSE': mse,
        'MAE': mae,
        'RMSE': rmse,
        'RSE': rse,
        'CORR': corr
    }

# Example evaluation
predictions = np.random.randn(1000, 96, 7)  # [samples, time, features]
ground_truth = np.random.randn(1000, 96, 7)

metrics = evaluate_forecast(predictions, ground_truth)
print(f"MSE: {metrics['MSE']:.4f}")
print(f"MAE: {metrics['MAE']:.4f}")
print(f"RSE: {metrics['RSE']:.4f}")

```

--------------------------------

### RevIN Layer for Time Series Normalization in Python

Source: https://context7.com/yuqinie98/patchtst/llms.txt

The RevIN (Reversible Instance Normalization) class implements a normalization layer specifically designed for non-stationary time series data. It supports both forward normalization ('norm') and inverse denormalization ('denorm') operations, with optional affine transformations. Dependencies include PyTorch.

```python
import torch
import torch.nn as nn

class RevIN(nn.Module):
    """Reversible Instance Normalization for non-stationary time series"""
    def __init__(self, num_features: int, eps=1e-5, affine=True, subtract_last=False):
        super().__init__()
        self.num_features = num_features
        self.eps = eps
        self.affine = affine
        self.subtract_last = subtract_last

        if self.affine:
            self.affine_weight = nn.Parameter(torch.ones(num_features))
            self.affine_bias = nn.Parameter(torch.zeros(num_features))

    def forward(self, x, mode: str):
        # x: [Batch, Length, Features]
        if mode == 'norm':
            # Compute statistics
            if self.subtract_last:
                self.last = x[:, -1, :].unsqueeze(1)
                x = x - self.last
            else:
                self.mean = torch.mean(x, dim=1, keepdim=True).detach()
                x = x - self.mean

            self.stdev = torch.sqrt(
                torch.var(x, dim=1, keepdim=True, unbiased=False) + self.eps
            ).detach()
            x = x / self.stdev

            if self.affine:
                x = x * self.affine_weight + self.affine_bias

        elif mode == 'denorm':
            # Reverse normalization
            if self.affine:
                x = (x - self.affine_bias) / (self.affine_weight + self.eps**2)
            x = x * self.stdev
            x = x + (self.last if self.subtract_last else self.mean)

        return x

# Example: Normalizing time series data
revin = RevIN(num_features=7, affine=True, subtract_last=False)
input_data = torch.randn(32, 96, 7)  # [batch, length, features]

# Normalize before model
normalized = revin(input_data, mode='norm')
print(f"Normalized mean: {normalized.mean():.6f}")  # Close to 0

# Process through model (placeholder)
output = normalized  # model(normalized)

# Denormalize after model
denormalized = revin(output, mode='denorm')
print(f"Original range restored: {denormalized.std():.4f}")
```
Tokens:
Raw