1. TimeGPT Overview
TimeGPT is a foundation model for time series forecasting that leverages the transformer architecture and large-scale pre-training to achieve state-of-the-art performance across diverse time series domains.
Key Features of TimeGPT:
- Foundation Model: Pre-trained on massive time series datasets
- Zero-Shot Forecasting: No fine-tuning required for basic tasks
- Multi-Domain: Works across energy, finance, climate, and healthcare
- Scalable: Handles various sequence lengths and forecasting horizons
- Probabilistic: Provides uncertainty quantification
Model Specifications
TimeGPT Model Variants:
- TimeGPT-Small: 20M parameters, fast inference
- TimeGPT-Base: 46M parameters, balanced performance
- TimeGPT-Large: 200M parameters, high accuracy
- TimeGPT-XLarge: 710M parameters, state-of-the-art
2. Model Architecture
Transformer Encoder
TimeGPT uses a transformer encoder architecture adapted for time series data:
class TimeGPTEncoder(nn.Module):
def __init__(self, d_model=512, nhead=8, num_layers=6, d_ff=2048):
super().__init__()
self.d_model = d_model
self.embedding = nn.Linear(1, d_model)
self.pos_encoding = PositionalEncoding(d_model)
encoder_layer = nn.TransformerEncoderLayer(
d_model=d_model,
nhead=nhead,
dim_feedforward=d_ff,
dropout=0.1,
activation='gelu'
)
self.transformer = nn.TransformerEncoder(encoder_layer, num_layers)
def forward(self, x):
# Input shape: (batch_size, seq_len)
x = x.unsqueeze(-1) # Add feature dimension
x = self.embedding(x) * math.sqrt(self.d_model)
x = self.pos_encoding(x)
# Transformer expects (seq_len, batch_size, d_model)
x = x.transpose(0, 1)
output = self.transformer(x)
return output.transpose(0, 1)
Temporal Positional Encoding
Custom positional encoding for time series data:
$$PE(pos, 2i) = \sin\left(\frac{pos}{10000^{2i/d_{model}}}\right)$$
$$PE(pos, 2i+1) = \cos\left(\frac{pos}{10000^{2i/d_{model}}}\right)$$
class TemporalPositionalEncoding(nn.Module):
def __init__(self, d_model, max_len=5000):
super().__init__()
pe = torch.zeros(max_len, d_model)
position = torch.arange(0, max_len, dtype=torch.float).unsqueeze(1)
div_term = torch.exp(torch.arange(0, d_model, 2).float() *
(-math.log(10000.0) / d_model))
pe[:, 0::2] = torch.sin(position * div_term)
pe[:, 1::2] = torch.cos(position * div_term)
pe = pe.unsqueeze(0).transpose(0, 1)
self.register_buffer('pe', pe)
def forward(self, x):
return x + self.pe[:x.size(0), :]
3. Pre-training Strategy
Masked Language Modeling
TimeGPT uses masked language modeling adapted for time series:
$$\mathcal{L}_{\text{MLM}} = \mathbb{E}_{t}[\log p(x_t | x_{\setminus t}, \theta)]$$
class TimeSeriesMLM(nn.Module):
def __init__(self, encoder, vocab_size):
super().__init__()
self.encoder = encoder
self.mlm_head = nn.Linear(encoder.d_model, vocab_size)
self.mask_token_id = vocab_size
def forward(self, x, mask_prob=0.15):
batch_size, seq_len = x.shape
# Create masks
mask = torch.rand(batch_size, seq_len) < mask_prob
masked_x = x.clone()
masked_x[mask] = self.mask_token_id
# Encode
encoded = self.encoder(masked_x)
# Predict masked tokens
predictions = self.mlm_head(encoded)
return predictions, mask
Pre-training Data
Pre-training Dataset Composition:
- Energy Data: 40% - Electricity demand, renewable generation
- Financial Data: 25% - Stock prices, forex rates, crypto
- Climate Data: 20% - Temperature, precipitation, weather
- Healthcare Data: 10% - Vital signs, patient monitoring
- Other Domains: 5% - Miscellaneous time series
Training Configuration
Pre-training Hyperparameters:
# Training Configuration
batch_size = 32
learning_rate = 1e-4
warmup_steps = 10000
max_steps = 500000
gradient_accumulation_steps = 4
max_grad_norm = 1.0
# Model Configuration
d_model = 512
nhead = 8
num_layers = 6
d_ff = 2048
dropout = 0.1
# Data Configuration
max_seq_len = 512
min_seq_len = 64
vocab_size = 10000
4. Fine-tuning Techniques
Task-Specific Fine-tuning
Fine-tuning strategies for different forecasting tasks:
class TimeGPTForecaster(nn.Module):
def __init__(self, pretrained_encoder, forecast_horizon):
super().__init__()
self.encoder = pretrained_encoder
self.forecast_head = nn.Linear(
pretrained_encoder.d_model,
forecast_horizon
)
# Freeze early layers, fine-tune later layers
for param in self.encoder.parameters():
param.requires_grad = False
# Unfreeze last few layers
for param in self.encoder.transformer.layers[-2:].parameters():
param.requires_grad = True
def forward(self, x):
encoded = self.encoder(x)
# Use last token representation for forecasting
last_hidden = encoded[:, -1, :]
forecast = self.forecast_head(last_hidden)
return forecast
Adaptive Learning Rates
Different learning rates for different components:
def get_optimizer(model, lr_encoder=1e-5, lr_head=1e-3):
params_encoder = []
params_head = []
for name, param in model.named_parameters():
if param.requires_grad:
if 'forecast_head' in name:
params_head.append(param)
else:
params_encoder.append(param)
optimizer = torch.optim.AdamW([
{'params': params_encoder, 'lr': lr_encoder},
{'params': params_head, 'lr': lr_head}
])
return optimizer
Few-Shot Learning
For scenarios with limited data:
Few-Shot Fine-tuning Algorithm:
def few_shot_finetune(model, support_data, query_data, num_shots=5):
# Use support data for quick adaptation
optimizer = torch.optim.Adam(model.parameters(), lr=1e-4)
for epoch in range(10): # Quick adaptation
for batch in support_data:
optimizer.zero_grad()
predictions = model(batch['input'])
loss = F.mse_loss(predictions, batch['target'])
loss.backward()
optimizer.step()
# Evaluate on query data
model.eval()
with torch.no_grad():
query_predictions = model(query_data['input'])
query_loss = F.mse_loss(query_predictions, query_data['target'])
return query_loss.item()
5. Data Preparation
Data Preprocessing
class TimeSeriesPreprocessor:
def __init__(self, normalization='z-score', handle_missing='interpolate'):
self.normalization = normalization
self.handle_missing = handle_missing
self.stats = {}
def fit_transform(self, data):
"""Fit preprocessing parameters and transform data"""
processed_data = []
for ts in data:
# Handle missing values
if self.handle_missing == 'interpolate':
ts = ts.interpolate()
elif self.handle_missing == 'forward_fill':
ts = ts.fillna(method='ffill')
# Normalize
if self.normalization == 'z-score':
mean, std = ts.mean(), ts.std()
ts = (ts - mean) / std
self.stats['mean'] = mean
self.stats['std'] = std
elif self.normalization == 'minmax':
min_val, max_val = ts.min(), ts.max()
ts = (ts - min_val) / (max_val - min_val)
self.stats['min'] = min_val
self.stats['max'] = max_val
processed_data.append(ts.values)
return processed_data
def inverse_transform(self, data):
"""Inverse transform to original scale"""
if self.normalization == 'z-score':
return data * self.stats['std'] + self.stats['mean']
elif self.normalization == 'minmax':
return data * (self.stats['max'] - self.stats['min']) + self.stats['min']
return data
Data Augmentation
Time Series Augmentation Techniques:
class TimeSeriesAugmentation:
@staticmethod
def add_noise(x, noise_level=0.1):
noise = torch.randn_like(x) * noise_level
return x + noise
@staticmethod
def time_warp(x, sigma=0.2):
"""Apply time warping to time series"""
seq_len = x.shape[-1]
tt = torch.arange(seq_len, dtype=torch.float32)
tt_new = tt + torch.randn(1) * sigma * seq_len
return F.interpolate(x.unsqueeze(0), size=seq_len, mode='linear').squeeze(0)
@staticmethod
def magnitude_warp(x, sigma=0.2):
"""Apply magnitude warping"""
curve = torch.randn_like(x) * sigma + 1.0
return x * curve
@staticmethod
def window_slice(x, reduce_ratio=0.9):
"""Randomly slice a window from the time series"""
seq_len = x.shape[-1]
target_len = int(seq_len * reduce_ratio)
start = torch.randint(0, seq_len - target_len, (1,))
return x[..., start:start+target_len]
6. Training Pipeline
Training Loop
def train_timegpt(model, train_loader, val_loader, num_epochs=100):
optimizer = get_optimizer(model)
scheduler = get_cosine_scheduler(optimizer, num_epochs)
criterion = nn.MSELoss()
best_val_loss = float('inf')
for epoch in range(num_epochs):
# Training phase
model.train()
train_loss = 0.0
for batch in train_loader:
optimizer.zero_grad()
# Forward pass
predictions = model(batch['input'])
loss = criterion(predictions, batch['target'])
# Backward pass
loss.backward()
torch.nn.utils.clip_grad_norm_(model.parameters(), max_grad_norm=1.0)
optimizer.step()
train_loss += loss.item()
# Validation phase
model.eval()
val_loss = 0.0
with torch.no_grad():
for batch in val_loader:
predictions = model(batch['input'])
loss = criterion(predictions, batch['target'])
val_loss += loss.item()
# Update learning rate
scheduler.step()
# Save best model
if val_loss < best_val_loss:
best_val_loss = val_loss
torch.save(model.state_dict(), 'best_timegpt.pth')
print(f'Epoch {epoch+1}/{num_epochs}: '
f'Train Loss: {train_loss/len(train_loader):.4f}, '
f'Val Loss: {val_loss/len(val_loader):.4f}')
Evaluation Metrics
Comprehensive Evaluation:
def evaluate_timegpt(model, test_loader):
model.eval()
metrics = {
'mae': 0.0,
'rmse': 0.0,
'mape': 0.0,
'smape': 0.0,
'mase': 0.0
}
with torch.no_grad():
for batch in test_loader:
predictions = model(batch['input'])
targets = batch['target']
# Calculate metrics
mae = F.l1_loss(predictions, targets)
mse = F.mse_loss(predictions, targets)
rmse = torch.sqrt(mse)
mape = torch.mean(torch.abs((targets - predictions) / targets)) * 100
smape = torch.mean(2 * torch.abs(targets - predictions) /
(torch.abs(targets) + torch.abs(predictions))) * 100
metrics['mae'] += mae.item()
metrics['rmse'] += rmse.item()
metrics['mape'] += mape.item()
metrics['smape'] += smape.item()
# Average metrics
for key in metrics:
metrics[key] /= len(test_loader)
return metrics
7. Inference Implementation
Single-Step Forecasting
def single_step_forecast(model, input_sequence, horizon=1):
"""Generate single-step forecasts"""
model.eval()
with torch.no_grad():
# Prepare input
input_tensor = torch.FloatTensor(input_sequence).unsqueeze(0)
# Generate forecast
forecast = model(input_tensor)
return forecast.squeeze().numpy()
Multi-Step Forecasting
def multi_step_forecast(model, input_sequence, horizon,
strategy='autoregressive'):
"""Generate multi-step forecasts"""
model.eval()
forecasts = []
current_input = input_sequence.copy()
with torch.no_grad():
for step in range(horizon):
# Prepare input tensor
input_tensor = torch.FloatTensor(current_input).unsqueeze(0)
# Generate next step
next_step = model(input_tensor)
forecasts.append(next_step.item())
# Update input for next iteration
if strategy == 'autoregressive':
current_input = np.append(current_input[1:], next_step.item())
elif strategy == 'direct':
# For direct strategy, we would use a different model
pass
return np.array(forecasts)
Uncertainty Quantification
Monte Carlo Dropout for Uncertainty:
def forecast_with_uncertainty(model, input_sequence, horizon, num_samples=100):
"""Generate forecasts with uncertainty estimates"""
model.train() # Enable dropout for uncertainty estimation
forecasts = []
for _ in range(num_samples):
with torch.no_grad():
input_tensor = torch.FloatTensor(input_sequence).unsqueeze(0)
forecast = model(input_tensor)
forecasts.append(forecast.squeeze().numpy())
forecasts = np.array(forecasts)
# Calculate statistics
mean_forecast = np.mean(forecasts, axis=0)
std_forecast = np.std(forecasts, axis=0)
# Prediction intervals
lower_bound = np.percentile(forecasts, 2.5, axis=0)
upper_bound = np.percentile(forecasts, 97.5, axis=0)
return {
'mean': mean_forecast,
'std': std_forecast,
'lower_bound': lower_bound,
'upper_bound': upper_bound,
'samples': forecasts
}
8. Performance Optimization
Model Optimization
Key Optimization Techniques:
- Mixed Precision Training: Use FP16 for faster training
- Gradient Checkpointing: Reduce memory usage
- Dynamic Batching: Optimize batch sizes
- Model Parallelism: Distribute large models
# Mixed Precision Training
from torch.cuda.amp import autocast, GradScaler
scaler = GradScaler()
for batch in train_loader:
optimizer.zero_grad()
with autocast():
predictions = model(batch['input'])
loss = criterion(predictions, batch['target'])
scaler.scale(loss).backward()
scaler.step(optimizer)
scaler.update()
Inference Optimization
Optimization Strategies:
# Model Quantization
import torch.quantization as quantization
def quantize_model(model, calibration_data):
model.eval()
model.qconfig = quantization.get_default_qconfig('fbgemm')
# Prepare model for quantization
model_prepared = quantization.prepare(model)
# Calibrate with sample data
with torch.no_grad():
for batch in calibration_data:
model_prepared(batch)
# Convert to quantized model
quantized_model = quantization.convert(model_prepared)
return quantized_model
# TensorRT Optimization
def optimize_with_tensorrt(model, input_shape):
import torch_tensorrt
# Compile model for TensorRT
trt_model = torch_tensorrt.compile(
model,
inputs=[torch.randn(input_shape)],
enabled_precisions={torch.float, torch.half}
)
return trt_model
9. Deployment Considerations
Model Serving
class TimeGPTServer:
def __init__(self, model_path, device='cpu'):
self.device = device
self.model = self.load_model(model_path)
self.preprocessor = TimeSeriesPreprocessor()
def load_model(self, model_path):
model = TimeGPTForecaster(
pretrained_encoder=TimeGPTEncoder(),
forecast_horizon=24
)
model.load_state_dict(torch.load(model_path, map_location=self.device))
model.to(self.device)
model.eval()
return model
def predict(self, input_data, horizon=24):
# Preprocess input
processed_data = self.preprocessor.transform([input_data])[0]
# Generate forecast
forecast = single_step_forecast(
self.model, processed_data, horizon
)
# Inverse transform
forecast = self.preprocessor.inverse_transform(forecast)
return forecast
API Implementation
FastAPI Service:
from fastapi import FastAPI, HTTPException
from pydantic import BaseModel
import numpy as np
app = FastAPI()
server = TimeGPTServer('models/timegpt_best.pth')
class ForecastRequest(BaseModel):
data: list
horizon: int = 24
include_uncertainty: bool = False
@app.post("/forecast")
async def forecast(request: ForecastRequest):
try:
if request.include_uncertainty:
result = server.predict_with_uncertainty(
request.data, request.horizon
)
else:
result = server.predict(request.data, request.horizon)
return {"forecast": result.tolist()}
except Exception as e:
raise HTTPException(status_code=500, detail=str(e))
@app.get("/health")
async def health_check():
return {"status": "healthy"}
Monitoring and Logging
Production Monitoring:
- Model Performance: Track prediction accuracy
- Inference Latency: Monitor response times
- Resource Usage: CPU, memory, GPU utilization
- Data Drift: Monitor input distribution changes
- Error Rates: Track prediction failures