PyTorch Time Series Forecasting

Time Series Forecasting Overview

Time series forecasting is the important task of predicting future values based on historical data. PyTorch provides powerful tools to build various time series forecasting models, from simple LSTM to complex Transformer architectures.

import torch
import torch.nn as nn
import torch.optim as optim
import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error, mean_absolute_error
from torch.utils.data import Dataset, DataLoader

Data Preprocessing

1. Time Series Dataset Class

class TimeSeriesDataset(Dataset):
    def __init__(self, data, sequence_length, prediction_length=1):
        """
        Time series dataset
        Args:
            data: Time series data (numpy array)
            sequence_length: Input sequence length
            prediction_length: Prediction length
        """
        self.data = data
        self.sequence_length = sequence_length
        self.prediction_length = prediction_length
        
    def __len__(self):
        return len(self.data) - self.sequence_length - self.prediction_length + 1
    
    def __getitem__(self, idx):
        # Input sequence
        x = self.data[idx:idx + self.sequence_length]
        # Target sequence
        y = self.data[idx + self.sequence_length:idx + self.sequence_length + self.prediction_length]
        
        return torch.FloatTensor(x), torch.FloatTensor(y)

def create_time_series_data(data, sequence_length=60, prediction_length=1, 
                           train_ratio=0.8, val_ratio=0.1):
    """Create time series dataset"""
    
    # Data normalization
    scaler = MinMaxScaler()
    scaled_data = scaler.fit_transform(data.reshape(-1, 1)).flatten()
    
    # Calculate split points
    total_len = len(scaled_data)
    train_len = int(total_len * train_ratio)
    val_len = int(total_len * val_ratio)
    
    # Split data
    train_data = scaled_data[:train_len]
    val_data = scaled_data[train_len:train_len + val_len]
    test_data = scaled_data[train_len + val_len:]
    
    # Create datasets
    train_dataset = TimeSeriesDataset(train_data, sequence_length, prediction_length)
    val_dataset = TimeSeriesDataset(val_data, sequence_length, prediction_length)
    test_dataset = TimeSeriesDataset(test_data, sequence_length, prediction_length)
    
    return train_dataset, val_dataset, test_dataset, scaler

# Generate example data
def generate_sine_wave_data(length=1000, frequency=0.1, noise_level=0.1):
    """Generate sine wave data"""
    t = np.linspace(0, length * frequency, length)
    data = np.sin(2 * np.pi * t) + noise_level * np.random.randn(length)
    return data

def generate_stock_like_data(length=1000, trend=0.001, volatility=0.02):
    """Generate stock-like data"""
    returns = np.random.normal(trend, volatility, length)
    prices = np.exp(np.cumsum(returns)) * 100
    return prices

# Example data
sine_data = generate_sine_wave_data(1000)
stock_data = generate_stock_like_data(1000)

# Create dataset
train_dataset, val_dataset, test_dataset, scaler = create_time_series_data(
    sine_data, sequence_length=60, prediction_length=1
)

print(f"Training set size: {len(train_dataset)}")
print(f"Validation set size: {len(val_dataset)}")
print(f"Test set size: {len(test_dataset)}")

2. Multivariate Time Series

class MultiVariateTimeSeriesDataset(Dataset):
    def __init__(self, data, sequence_length, prediction_length=1, target_column=0):
        """
        Multivariate time series dataset
        Args:
            data: Multivariate time series data (numpy array, shape: [time_steps, features])
            sequence_length: Input sequence length
            prediction_length: Prediction length
            target_column: Target variable column index
        """
        self.data = data
        self.sequence_length = sequence_length
        self.prediction_length = prediction_length
        self.target_column = target_column
        
    def __len__(self):
        return len(self.data) - self.sequence_length - self.prediction_length + 1
    
    def __getitem__(self, idx):
        # Input sequence (all features)
        x = self.data[idx:idx + self.sequence_length]
        # Target sequence (only target variable)
        y = self.data[idx + self.sequence_length:idx + self.sequence_length + self.prediction_length, 
                     self.target_column]
        
        return torch.FloatTensor(x), torch.FloatTensor(y)

def create_multivariate_data(n_samples=1000, n_features=5):
    """Create multivariate time series data"""
    t = np.linspace(0, 10, n_samples)
    
    # Create correlated multivariate data
    data = np.zeros((n_samples, n_features))
    
    # Main trend
    trend = 0.1 * t + np.sin(0.5 * t)
    
    for i in range(n_features):
        # Each feature is related to main trend but with different phases and noise
        phase = i * np.pi / n_features
        noise = 0.1 * np.random.randn(n_samples)
        data[:, i] = trend + 0.5 * np.sin(t + phase) + noise
    
    return data

# Create multivariate data
multivar_data = create_multivariate_data(1000, 5)
multivar_dataset = MultiVariateTimeSeriesDataset(
    multivar_data, sequence_length=60, prediction_length=1, target_column=0
)

LSTM Time Series Models

1. Basic LSTM Model

class LSTMPredictor(nn.Module):
    def __init__(self, input_size=1, hidden_size=50, num_layers=2, 
                 output_size=1, dropout=0.2):
        super(LSTMPredictor, self).__init__()
        
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        
        # LSTM layer
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0
        )
        
        # Output layer
        self.fc = nn.Linear(hidden_size, output_size)
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, x):
        # LSTM forward pass
        lstm_out, (hidden, cell) = self.lstm(x)
        
        # Use the last timestep output
        last_output = lstm_out[:, -1, :]
        
        # Apply dropout and fully connected layer
        output = self.dropout(last_output)
        output = self.fc(output)
        
        return output

# Create model
model = LSTMPredictor(input_size=1, hidden_size=50, num_layers=2, output_size=1)

# Test model
sample_input = torch.randn(32, 60, 1)  # (batch_size, sequence_length, input_size)
sample_output = model(sample_input)
print(f"Input shape: {sample_input.shape}")
print(f"Output shape: {sample_output.shape}")

2. Bidirectional LSTM Model

class BiLSTMPredictor(nn.Module):
    def __init__(self, input_size=1, hidden_size=50, num_layers=2, 
                 output_size=1, dropout=0.2):
        super(BiLSTMPredictor, self).__init__()
        
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        
        # Bidirectional LSTM
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0,
            bidirectional=True
        )
        
        # Attention mechanism
        self.attention = nn.Linear(hidden_size * 2, 1)
        
        # Output layer
        self.fc = nn.Linear(hidden_size * 2, output_size)
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, x):
        # Bidirectional LSTM
        lstm_out, _ = self.lstm(x)  # (batch_size, seq_len, hidden_size * 2)
        
        # Attention mechanism
        attention_weights = torch.softmax(self.attention(lstm_out), dim=1)
        context_vector = torch.sum(attention_weights * lstm_out, dim=1)
        
        # Output
        output = self.dropout(context_vector)
        output = self.fc(output)
        
        return output

3. Multi-Step Prediction LSTM

class MultiStepLSTM(nn.Module):
    def __init__(self, input_size=1, hidden_size=50, num_layers=2, 
                 output_size=10, dropout=0.2):
        super(MultiStepLSTM, self).__init__()
        
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        self.output_size = output_size
        
        # Encoder LSTM
        self.encoder_lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0
        )
        
        # Decoder LSTM
        self.decoder_lstm = nn.LSTM(
            input_size=1,  # Decoder input dimension
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0
        )
        
        # Output layer
        self.fc = nn.Linear(hidden_size, 1)
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, x, target_length=None):
        batch_size = x.size(0)
        
        # Encoder
        encoder_out, (hidden, cell) = self.encoder_lstm(x)
        
        # Decoder
        if target_length is None:
            target_length = self.output_size
        
        decoder_input = torch.zeros(batch_size, 1, 1, device=x.device)
        decoder_hidden = (hidden, cell)
        
        outputs = []
        
        for _ in range(target_length):
            decoder_out, decoder_hidden = self.decoder_lstm(decoder_input, decoder_hidden)
            output = self.fc(self.dropout(decoder_out))
            outputs.append(output)
            
            # Use current output as next step input
            decoder_input = output
        
        # Concatenate all outputs
        outputs = torch.cat(outputs, dim=1)  # (batch_size, target_length, 1)
        
        return outputs.squeeze(-1)  # (batch_size, target_length)

GRU Time Series Models

1. GRU Predictor

class GRUPredictor(nn.Module):
    def __init__(self, input_size=1, hidden_size=50, num_layers=2, 
                 output_size=1, dropout=0.2):
        super(GRUPredictor, self).__init__()
        
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        
        # GRU layer
        self.gru = nn.GRU(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0
        )
        
        # Output layer
        self.fc = nn.Linear(hidden_size, output_size)
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, x):
        # GRU forward pass
        gru_out, hidden = self.gru(x)
        
        # Use the last timestep output
        last_output = gru_out[:, -1, :]
        
        # Output
        output = self.dropout(last_output)
        output = self.fc(output)
        
        return output

Transformer Time Series Models

1. Time Series Transformer

class TimeSeriesTransformer(nn.Module):
    def __init__(self, input_size=1, d_model=64, nhead=8, num_layers=6, 
                 output_size=1, max_seq_length=1000, dropout=0.1):
        super(TimeSeriesTransformer, self).__init__()
        
        self.d_model = d_model
        self.input_projection = nn.Linear(input_size, d_model)
        
        # Positional encoding
        self.pos_encoding = PositionalEncoding(d_model, max_seq_length)
        
        # Transformer encoder
        encoder_layer = nn.TransformerEncoderLayer(
            d_model=d_model,
            nhead=nhead,
            dim_feedforward=d_model * 4,
            dropout=dropout,
            batch_first=True
        )
        self.transformer = nn.TransformerEncoder(encoder_layer, num_layers)
        
        # Output layer
        self.output_projection = nn.Linear(d_model, output_size)
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, x):
        # Input projection
        x = self.input_projection(x) * math.sqrt(self.d_model)
        
        # Positional encoding
        x = self.pos_encoding(x)
        
        # Transformer encoding
        transformer_out = self.transformer(x)
        
        # Use the last timestep output
        last_output = transformer_out[:, -1, :]
        
        # Output projection
        output = self.dropout(last_output)
        output = self.output_projection(output)
        
        return output

class PositionalEncoding(nn.Module):
    def __init__(self, d_model, max_len=5000):
        super(PositionalEncoding, self).__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(1), :].transpose(0, 1)

Training Framework

1. Time Series Trainer

class TimeSeriesTrainer:
    def __init__(self, model, train_loader, val_loader, device, learning_rate=0.001):
        self.model = model.to(device)
        self.train_loader = train_loader
        self.val_loader = val_loader
        self.device = device
        
        # Loss function and optimizer
        self.criterion = nn.MSELoss()
        self.optimizer = optim.Adam(model.parameters(), lr=learning_rate)
        self.scheduler = optim.lr_scheduler.ReduceLROnPlateau(
            self.optimizer, mode='min', patience=10, factor=0.5
        )
        
        # Training history
        self.train_losses = []
        self.val_losses = []
        self.best_val_loss = float('inf')
        
    def train_epoch(self):
        """Train one epoch"""
        self.model.train()
        total_loss = 0
        
        for batch_idx, (data, target) in enumerate(self.train_loader):
            data, target = data.to(self.device), target.to(self.device)
            
            self.optimizer.zero_grad()
            
            # Forward pass
            output = self.model(data)
            
            # Ensure output and target shapes match
            if output.dim() != target.dim():
                if target.dim() == 1:
                    target = target.unsqueeze(-1)
                elif output.dim() == 1:
                    output = output.unsqueeze(-1)
            
            loss = self.criterion(output, target)
            
            # Backward pass
            loss.backward()
            
            # Gradient clipping
            torch.nn.utils.clip_grad_norm_(self.model.parameters(), max_norm=1.0)
            
            self.optimizer.step()
            
            total_loss += loss.item()
        
        return total_loss / len(self.train_loader)
    
    def validate_epoch(self):
        """Validate one epoch"""
        self.model.eval()
        total_loss = 0
        
        with torch.no_grad():
            for data, target in self.val_loader:
                data, target = data.to(self.device), target.to(self.device)
                
                output = self.model(data)
                
                # Ensure shapes match
                if output.dim() != target.dim():
                    if target.dim() == 1:
                        target = target.unsqueeze(-1)
                    elif output.dim() == 1:
                        output = output.unsqueeze(-1)
                
                loss = self.criterion(output, target)
                total_loss += loss.item()
        
        return total_loss / len(self.val_loader)
    
    def train(self, num_epochs):
        """Complete training process"""
        print(f"Starting training, {num_epochs} epochs")
        
        for epoch in range(num_epochs):
            # Train
            train_loss = self.train_epoch()
            
            # Validate
            val_loss = self.validate_epoch()
            
            # Update learning rate
            self.scheduler.step(val_loss)
            
            # Record history
            self.train_losses.append(train_loss)
            self.val_losses.append(val_loss)
            
            # Save best model
            if val_loss < self.best_val_loss:
                self.best_val_loss = val_loss
                torch.save(self.model.state_dict(), 'best_model.pth')
            
            # Print progress
            if (epoch + 1) % 10 == 0:
                print(f'Epoch {epoch+1}/{num_epochs}:')
                print(f'  Train Loss: {train_loss:.6f}')
                print(f'  Val Loss: {val_loss:.6f}')
                print(f'  LR: {self.optimizer.param_groups[0]["lr"]:.8f}')
        
        print(f'Training complete! Best validation loss: {self.best_val_loss:.6f}')
        
        return self.train_losses, self.val_losses

# Usage example
device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')

# Create data loaders
train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False)

# Create model
model = LSTMPredictor(input_size=1, hidden_size=50, num_layers=2, output_size=1)

# Create trainer
trainer = TimeSeriesTrainer(model, train_loader, val_loader, device)

# Train model
train_losses, val_losses = trainer.train(num_epochs=100)

Model Evaluation

1. Prediction and Evaluation

def evaluate_model(model, test_loader, scaler, device):
    """Evaluate model performance"""
    model.eval()
    predictions = []
    actuals = []
    
    with torch.no_grad():
        for data, target in test_loader:
            data, target = data.to(device), target.to(device)
            
            output = model(data)
            
            # Convert to CPU and add to lists
            predictions.extend(output.cpu().numpy())
            actuals.extend(target.cpu().numpy())
    
    # Convert to numpy arrays
    predictions = np.array(predictions)
    actuals = np.array(actuals)
    
    # Inverse normalization
    predictions = scaler.inverse_transform(predictions.reshape(-1, 1)).flatten()
    actuals = scaler.inverse_transform(actuals.reshape(-1, 1)).flatten()
    
    # Calculate metrics
    mse = mean_squared_error(actuals, predictions)
    mae = mean_absolute_error(actuals, predictions)
    rmse = np.sqrt(mse)
    
    # Calculate MAPE
    mape = np.mean(np.abs((actuals - predictions) / actuals)) * 100
    
    print(f"Evaluation results:")
    print(f"  MSE: {mse:.6f}")
    print(f"  MAE: {mae:.6f}")
    print(f"  RMSE: {rmse:.6f}")
    print(f"  MAPE: {mape:.2f}%")
    
    return predictions, actuals, {
        'mse': mse, 'mae': mae, 'rmse': rmse, 'mape': mape
    }

# Evaluate model
test_loader = DataLoader(test_dataset, batch_size=32, shuffle=False)
model.load_state_dict(torch.load('best_model.pth'))

predictions, actuals, metrics = evaluate_model(model, test_loader, scaler, device)

2. Visualize Results

def plot_predictions(actuals, predictions, title="Time Series Prediction Results"):
    """Visualize prediction results"""
    plt.figure(figsize=(15, 6))
    
    # Only show first 200 points for clarity
    n_points = min(200, len(actuals))
    
    plt.plot(range(n_points), actuals[:n_points], label='Actual', color='blue', alpha=0.7)
    plt.plot(range(n_points), predictions[:n_points], label='Predicted', color='red', alpha=0.7)
    
    plt.title(title)
    plt.xlabel('Time Step')
    plt.ylabel('Value')
    plt.legend()
    plt.grid(True, alpha=0.3)
    plt.tight_layout()
    plt.show()

def plot_training_history(train_losses, val_losses):
    """Visualize training history"""
    plt.figure(figsize=(12, 4))
    
    plt.subplot(1, 2, 1)
    plt.plot(train_losses, label='Training Loss')
    plt.plot(val_losses, label='Validation Loss')
    plt.title('Training History')
    plt.xlabel('Epoch')
    plt.ylabel('Loss')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    plt.subplot(1, 2, 2)
    plt.plot(train_losses, label='Training Loss')
    plt.plot(val_losses, label='Validation Loss')
    plt.title('Training History (Log Scale)')
    plt.xlabel('Epoch')
    plt.ylabel('Loss (log scale)')
    plt.yscale('log')
    plt.legend()
    plt.grid(True, alpha=0.3)
    
    plt.tight_layout()
    plt.show()

# Visualize results
plot_predictions(actuals, predictions)
plot_training_history(train_losses, val_losses)

Advanced Techniques

1. Attention Mechanism

class AttentionLSTM(nn.Module):
    def __init__(self, input_size=1, hidden_size=50, num_layers=2, 
                 output_size=1, dropout=0.2):
        super(AttentionLSTM, self).__init__()
        
        self.hidden_size = hidden_size
        self.num_layers = num_layers
        
        # LSTM layer
        self.lstm = nn.LSTM(
            input_size=input_size,
            hidden_size=hidden_size,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0
        )
        
        # Attention mechanism
        self.attention = nn.Sequential(
            nn.Linear(hidden_size, hidden_size),
            nn.Tanh(),
            nn.Linear(hidden_size, 1)
        )
        
        # Output layer
        self.fc = nn.Linear(hidden_size, output_size)
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, x):
        # LSTM output
        lstm_out, _ = self.lstm(x)  # (batch_size, seq_len, hidden_size)
        
        # Compute attention weights
        attention_scores = self.attention(lstm_out)  # (batch_size, seq_len, 1)
        attention_weights = torch.softmax(attention_scores, dim=1)
        
        # Weighted sum
        context_vector = torch.sum(attention_weights * lstm_out, dim=1)  # (batch_size, hidden_size)
        
        # Output
        output = self.dropout(context_vector)
        output = self.fc(output)
        
        return output, attention_weights

2. Residual Connection

class ResidualLSTM(nn.Module):
    def __init__(self, input_size=1, hidden_size=50, num_layers=4, 
                 output_size=1, dropout=0.2):
        super(ResidualLSTM, self).__init__()
        
        self.input_projection = nn.Linear(input_size, hidden_size)
        
        # Multiple LSTM layers with residual connections
        self.lstm_layers = nn.ModuleList()
        for i in range(num_layers):
            self.lstm_layers.append(
                nn.LSTM(hidden_size, hidden_size, 1, batch_first=True, dropout=0)
            )
        
        self.layer_norms = nn.ModuleList([
            nn.LayerNorm(hidden_size) for _ in range(num_layers)
        ])
        
        self.dropout = nn.Dropout(dropout)
        self.fc = nn.Linear(hidden_size, output_size)
        
    def forward(self, x):
        # Input projection
        x = self.input_projection(x)
        
        # Through multiple LSTM layers
        for i, (lstm, norm) in enumerate(zip(self.lstm_layers, self.layer_norms)):
            residual = x
            lstm_out, _ = lstm(x)
            
            # Residual connection and layer normalization
            x = norm(lstm_out + residual)
            x = self.dropout(x)
        
        # Output
        last_output = x[:, -1, :]
        output = self.fc(last_output)
        
        return output

3. Multi-Scale Feature Extraction

class MultiScaleLSTM(nn.Module):
    def __init__(self, input_size=1, hidden_size=50, output_size=1, 
                 scales=[1, 3, 5], dropout=0.2):
        super(MultiScaleLSTM, self).__init__()
        
        self.scales = scales
        
        # Different scale LSTMs
        self.lstm_layers = nn.ModuleList()
        for scale in scales:
            self.lstm_layers.append(
                nn.LSTM(input_size, hidden_size, 2, batch_first=True, dropout=dropout)
            )
        
        # Fusion layer
        self.fusion = nn.Linear(len(scales) * hidden_size, hidden_size)
        self.fc = nn.Linear(hidden_size, output_size)
        self.dropout = nn.Dropout(dropout)
        
    def forward(self, x):
        scale_outputs = []
        
        for i, (scale, lstm) in enumerate(zip(self.scales, self.lstm_layers)):
            # Multi-scale sampling
            if scale > 1:
                # Simple downsampling
                sampled_x = x[:, ::scale, :]
            else:
                sampled_x = x
            
            # LSTM processing
            lstm_out, _ = lstm(sampled_x)
            last_output = lstm_out[:, -1, :]
            scale_outputs.append(last_output)
        
        # Fuse features from different scales
        fused = torch.cat(scale_outputs, dim=1)
        fused = self.fusion(fused)
        fused = torch.relu(fused)
        fused = self.dropout(fused)
        
        # Output
        output = self.fc(fused)
        
        return output

Practical Application Examples

1. Stock Price Prediction

def create_stock_prediction_pipeline():
    """Create stock prediction pipeline"""
    
    # Generate simulated stock price data
    stock_data = generate_stock_like_data(2000, trend=0.0005, volatility=0.02)
    
    # Create dataset
    train_dataset, val_dataset, test_dataset, scaler = create_time_series_data(
        stock_data, sequence_length=60, prediction_length=1
    )
    
    # Create data loaders
    train_loader = DataLoader(train_dataset, batch_size=64, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=64, shuffle=False)
    test_loader = DataLoader(test_dataset, batch_size=64, shuffle=False)
    
    # Create model
    model = AttentionLSTM(input_size=1, hidden_size=64, num_layers=3, output_size=1)
    
    # Train
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    trainer = TimeSeriesTrainer(model, train_loader, val_loader, device, learning_rate=0.001)
    
    train_losses, val_losses = trainer.train(num_epochs=100)
    
    # Evaluate
    model.load_state_dict(torch.load('best_model.pth'))
    predictions, actuals, metrics = evaluate_model(model, test_loader, scaler, device)
    
    return model, predictions, actuals, metrics

# Run stock price prediction example
# model, predictions, actuals, metrics = create_stock_prediction_pipeline()

2. Multi-Step Prediction

def multi_step_prediction_example():
    """Multi-step prediction example"""
    
    # Create multi-step prediction dataset
    class MultiStepDataset(Dataset):
        def __init__(self, data, input_length, output_length):
            self.data = data
            self.input_length = input_length
            self.output_length = output_length
            
        def __len__(self):
            return len(self.data) - self.input_length - self.output_length + 1
        
        def __getitem__(self, idx):
            x = self.data[idx:idx + self.input_length]
            y = self.data[idx + self.input_length:idx + self.input_length + self.output_length]
            return torch.FloatTensor(x), torch.FloatTensor(y)
    
    # Generate data
    data = generate_sine_wave_data(1000)
    scaler = MinMaxScaler()
    scaled_data = scaler.fit_transform(data.reshape(-1, 1)).flatten()
    
    # Create dataset
    dataset = MultiStepDataset(scaled_data, input_length=60, output_length=10)
    train_size = int(0.8 * len(dataset))
    val_size = len(dataset) - train_size
    
    train_dataset, val_dataset = torch.utils.data.random_split(dataset, [train_size, val_size])
    
    train_loader = DataLoader(train_dataset, batch_size=32, shuffle=True)
    val_loader = DataLoader(val_dataset, batch_size=32, shuffle=False)
    
    # Create multi-step prediction model
    model = MultiStepLSTM(input_size=1, hidden_size=64, num_layers=2, output_size=10)
    
    # Train
    device = torch.device('cuda' if torch.cuda.is_available() else 'cpu')
    trainer = TimeSeriesTrainer(model, train_loader, val_loader, device)
    
    train_losses, val_losses = trainer.train(num_epochs=50)
    
    return model, scaler

# Run multi-step prediction example
# multi_step_model, multi_step_scaler = multi_step_prediction_example()

Summary

Time series forecasting is an important application area of PyTorch. This chapter introduced:

1. Data Preprocessing: Time series dataset creation and preprocessing techniques
2. Classic Models: Recurrent neural network models like LSTM, GRU
3. Modern Architectures: Attention mechanism models like Transformer
4. Training Framework: Complete training, validation, and evaluation process
5. Advanced Techniques: Attention mechanisms, residual connections, multi-scale feature extraction
6. Practical Applications: Specific cases like stock price prediction, multi-step forecasting

Mastering these techniques will help you succeed in time series related tasks such as financial forecasting, demand forecasting, and anomaly detection!