Ai Time Series Forecasting

---




name: ai-time-series-forecasting
compatibility: opencode
completeness: 95
content-types:
- code
- guidance
- config
- do-dont
description: '"Provides Time series forecasting for price prediction and market analysis"'
license: MIT
maturity: stable
metadata:
  domain: trading
  output-format: code
  related-skills: ai-anomaly-detection, ai-explainable-ai, technical-cycle-analysis
  role: implementation
  scope: implementation
  triggers: ai time series forecasting, ai-time-series-forecasting, market, prediction,
    price
  archetypes:
  - tactical
  anti_triggers:
  - brainstorming
  - vague ideation
  - no risk management
  response_profile:
    verbosity: low
    directive_strength: high
    abstraction_level: operational
version: "1.0.0"




---




**Role:** Implement forecasting models that predict future price movements and market indicators

**Philosophy:** Markets exhibit both predictable patterns and random noise. Prioritize robustness, uncertainty quantification, and regime-aware forecasts over pure prediction accuracy.

## Key Principles

1. **Stationarity First**: Transform non-stationary price data to stationary returns or log-prices
2. **Multi-Horizon Forecasting**: Predict multiple time horizons (short, medium, long)
3. **Uncertainty Quantification**: Provide prediction intervals, not just point estimates
4. **Backtesting Validation**: Test forecasts in out-of-sample periods with transaction costs
5. **Ensemble Averaging**: Combine multiple models for more stable predictions

## Implementation Guidelines

### Structure
- Core logic: `ts_forecast/models.py` - Model implementations
- Feature builder: `ts_forecast/features.py` - Lag features, technical indicators
- Pipeline: `ts_forecast/pipeline.py` - Data preprocessing and forecasting
- Config: `config/ts_config.yaml` - Model parameters

### Patterns to Follow
- Use walk-forward validation for time series
- Include exogenous features (volume, volatility, macro data)
- Log-transform prices for multiplicative processes
- Normalize features per regime or per asset

## Adherence Checklist
Before completing your task, verify:
- [ ] Data transformed to stationary form (returns, not prices)
- [ ] Forecast includes uncertainty intervals
- [ ] Walk-forward validation used, not random train/test split
- [ ] Multiple horizons forecasted and evaluated separately
- [ ] Backtest considers transaction costs and slippage



## Code Examples

### AutoRegressive Model with ADF Stationarity Test

```python
import numpy as np
import pandas as pd
from statsmodels.tsa.stattools import adfuller
from statsmodels.tsa.arima.model import ARIMA
from statsmodels.tsa.statespace.sarimax import SARIMAX
from typing import Tuple, Optional

class TimeSeriesForecaster:
    """Time series forecasting with stationarity checks and ARIMA modeling."""
    
    def __init__(self, max_ar_order: int = 10, max_ma_order: int = 5):
        self.max_ar_order = max_ar_order
        self.max_ma_order = max_ma_order
        self.fitted_model = None
        self.is_stationary = False
        self.difference_order = 0
        
    def check_stationarity(self, series: np.ndarray, threshold: float = 0.05) -> bool:
        """Check if series is stationary using ADF test."""
        result = adfuller(series)
        p_value = result[1]
        return p_value < threshold
    
    def difference_series(self, series: np.ndarray, order: int = 1) -> np.ndarray:
        """Difference series to achieve stationarity."""
        diffed = series.copy()
        for _ in range(order):
            diffed = np.diff(diffed)
        return diffed
    
    def find_stationary_order(self, series: np.ndarray) -> int:
        """Find minimum differencing needed for stationarity."""
        for d in range(3):
            diffed = self.difference_series(series, d)
            if self.check_stationarity(diffed):
                return d
        return 2  # Maximum differencing
    
    def fit(self, prices: np.ndarray, order: Tuple[int, int, int] = None):
        """Fit ARIMA model to price series."""
        prices = np.asarray(prices)
        
        # Ensure stationarity
        if not self.check_stationarity(prices):
            self.difference_order = self.find_stationary_order(prices)
            prices = self.difference_series(prices, self.difference_order)
        
        # Auto-select order if not provided
        if order is None:
            order = self._select_order(prices)
        
        self.fitted_model = ARIMA(prices, order=order).fit()
        self.is_stationary = True
        
        return self
    
    def _select_order(self, series: np.ndarray) -> Tuple[int, int, int]:
        """Select ARIMA order using AIC criterion."""
        best_aic = np.inf
        best_order = (1, 0, 1)
        
        for p in range(self.max_ar_order + 1):
            for q in range(self.max_ma_order + 1):
                try:
                    model = ARIMA(series, order=(p, 0, q)).fit()
                    if model.aic < best_aic:
                        best_aic = model.aic
                        best_order = (p, 0, q)
                except:
                    continue
        
        return best_order
    
    def forecast(self, steps: int) -> Tuple[np.ndarray, np.ndarray]:
        """Forecast future values with prediction intervals."""
        if self.fitted_model is None:
            raise ValueError("Model not fitted")
        
        forecast_result = self.fitted_model.get_forecast(steps=steps)
        forecast_mean = forecast_result.predicted_mean.values
        forecast_std = forecast_result.se_mean.values
        
        # Revert differencing if applied
        if self.difference_order > 0:
            last_price = self._get_last_price()
            forecast_mean = self._revert_difference(forecast_mean, last_price)
        
        return forecast_mean, forecast_std
    
    def _get_last_price(self) -> float:
        """Get last observed price from training data."""
        # In practice, store this during fit
        return 100.0  # Placeholder
```

### Multi-Horizon LSTM Forecaster

```python
import torch
import torch.nn as nn
from torch.utils.data import Dataset, DataLoader
import numpy as np
from typing import List, Tuple

class TimeSeriesDataset(Dataset):
    """PyTorch dataset for time series forecasting."""
    
    def __init__(self, data: np.ndarray, target: np.ndarray, 
                 input_window: int = 20, output_window: int = 5):
        self.data = data
        self.target = target
        self.input_window = input_window
        self.output_window = output_window
        self.n_samples = len(data) - input_window - output_window + 1
    
    def __len__(self):
        return self.n_samples
    
    def __getitem__(self, idx):
        x = self.data[idx:idx + self.input_window]
        y = self.target[idx + self.input_window:idx + self.input_window + self.output_window]
        return torch.FloatTensor(x), torch.FloatTensor(y)

class LSTMForecaster(nn.Module):
    """LSTM model for multi-step time series forecasting."""
    
    def __init__(self, input_dim: int = 1, hidden_dim: int = 64, 
                 num_layers: int = 2, output_dim: int = 5, dropout: float = 0.2):
        super(LSTMForecaster, self).__init__()
        
        self.hidden_dim = hidden_dim
        self.num_layers = num_layers
        
        self.lstm = nn.LSTM(
            input_size=input_dim,
            hidden_size=hidden_dim,
            num_layers=num_layers,
            batch_first=True,
            dropout=dropout if num_layers > 1 else 0
        )
        
        self.fc1 = nn.Linear(hidden_dim, hidden_dim // 2)
        self.relu = nn.ReLU()
        self.fc2 = nn.Linear(hidden_dim // 2, output_dim)
        self.dropout = nn.Dropout(dropout)
    
    def forward(self, x):
        # LSTM output
        lstm_out, (h_n, c_n) = self.lstm(x)
        
        # Use last hidden state
        out = lstm_out[:, -1, :]
        
        # Fully connected layers
        out = self.fc1(out)
        out = self.relu(out)
        out = self.dropout(out)
        out = self.fc2(out)
        
        return out

class MultiHorizonForecaster:
    """Wrapper for LSTM multi-horizon forecasting."""
    
    def __init__(self, input_window: int = 20, output_horizons: List[int] = [1, 5, 10, 20]):
        self.input_window = input_window
        self.output_horizons = output_horizons
        self.models = {}
        self.scaler = None
    
    def fit(self, returns: np.ndarray, epochs: int = 100, batch_size: int = 32):
        """Train models for each forecast horizon."""
        # Normalize returns
        mean = np.mean(returns)
        std = np.std(returns)
        normalized = (returns - mean) / std
        self.scaler = {'mean': mean, 'std': std}
        
        # Train separate model for each horizon
        for horizon in self.output_horizons:
            dataset = TimeSeriesDataset(
                normalized, normalized,
                input_window=self.input_window,
                output_window=horizon
            )
            
            model = LSTMForecaster(
                input_dim=1,
                hidden_dim=64,
                num_layers=2,
                output_dim=horizon
            )
            
            # Train model (simplified training loop)
            optimizer = torch.optim.Adam(model.parameters(), lr=0.001)
            criterion = nn.MSELoss()
            
            for epoch in range(epochs):
                for batch_x, batch_y in DataLoader(dataset, batch_size=batch_size, shuffle=True):
                    optimizer.zero_grad()
                    outputs = model(batch_x)
                    loss = criterion(outputs, batch_y)
                    loss.backward()
                    optimizer.step()
            
            self.models[horizon] = model
    
    def forecast(self, latest_data: np.ndarray) -> dict:
        """Forecast for all horizons."""
        if self.scaler is None or not self.models:
            raise ValueError("Model not fitted")
        
        # Normalize input
        normalized = (latest_data - self.scaler['mean']) / self.scaler['std']
        
        forecasts = {}
        for horizon, model in self.models.items():
            with torch.no_grad():
                input_tensor = torch.FloatTensor(normalized[-self.input_window:]).unsqueeze(0)
                prediction = model(input_tensor).numpy()[0]
                
                # Denormalize
                forecasts[horizon] = prediction * self.scaler['std'] + self.scaler['mean']
        
        return forecasts
```

### Walk-Forward Validation

```python
from sklearn.model_selection import TimeSeriesSplit
import numpy as np

class WalkForwardValidator:
    """Perform walk-forward validation for time series forecasting."""
    
    def __init__(self, n_splits: int = 5, test_size: int = 100):
        self.n_splits = n_splits
        self.test_size = test_size
    
    def split(self, data_length: int):
        """Generate train/test indices for walk-forward validation."""
        train_start = 0
        
        for i in range(self.n_splits):
            train_end = train_start + data_length // (self.n_splits + 1)
            test_start = train_end
            test_end = test_start + self.test_size
            
            if test_end > data_length:
                break
            
            yield np.arange(train_start, train_end), np.arange(test_start, test_end)
            
            train_start = test_start + 1
    
    def evaluate(self, forecaster, prices: np.ndarray, metrics: list = ['mae', 'mse']):
        """Evaluate forecaster using walk-forward validation."""
        prices = np.asarray(prices)
        returns = np.diff(np.log(prices))
        
        all_predictions = []
        all_actuals = []
        
        for train_idx, test_idx in self.split(len(returns)):
            train_returns = returns[train_idx]
            test_returns = returns[test_idx]
            
            # Fit model on training data
            forecaster.fit(train_returns)
            
            # Forecast on test data
            predictions = forecaster.forecast(test_returns[:forecaster.input_window])
            
            all_predictions.extend(predictions)
            all_actuals.extend(test_returns[forecaster.input_window:])
        
        # Calculate metrics
        results = {}
        for metric in metrics:
            if metric == 'mae':
                results['mae'] = np.mean(np.abs(np.array(all_predictions) - np.array(all_actuals)))
            elif metric == 'mse':
                results['mse'] = np.mean((np.array(all_predictions) - np.array(all_actuals)) ** 2)
        
        return results
```

---

---

## Constraints

### MUST DO
- Validate input feature distributions against training data baselines; flag drift exceeding 2 standard deviations
- Implement model versioning with reproducibility tags — every prediction must be traceable to the exact model artifact and config
- Include confidence intervals or probability estimates alongside all point predictions, never return raw scores without context
- Log all model inputs, outputs, and metadata to enable post-hoc analysis of prediction failures
- Implement feature computation consistently between training and inference — use the same transformation pipeline for both

### MUST NOT DO
- Do not train models on look-ahead biased features (e.g., using future prices or events in training data)
- Avoid deploying a new model version without shadow-testing against the current production model first
- Never retrain a model on a data window that includes regime changes without explicit regime-aware validation
- Do not use accuracy as the primary metric for imbalanced datasets — use precision/recall, F1, or AUC-ROC
- Avoid hardcoding feature names; load them from a schema or config file to prevent mismatches between training and inference


## Live References

> Authoritative documentation links for this skill's domain. The model follows markdown links at load time to resolve external references and inline content.

- [What is Time Series Forecasting?](https://databricks.com/blog/what-time-series-forecasting)
- [Prophet - Facebook Time Series Tool](https://facebook.github.io/prophet/)
- [ARIMA Model Guide](https://www.investopedia.com/terms/a/arima.asp)
- [LSTM for Time Series Prediction](https://www.tensorflow.org/tutorials/structured_data/time_series)
- [Time Series Forecasting Best Practices](https://otexts.com/fpp3/)