### Run TempoPFN Quick Start Script

Source: https://github.com/automl/tempopfn/blob/main/README.md

Executes the quick start script for TempoPFN. This is the recommended way to begin using the model after setup. Alternatively, a Jupyter Notebook version is available.

```bash
# 4. Run the Quick Start Script 
python examples/quick_start_tempo_pfn.py

# 5. Alternatively, you can run the Notebook version
jupyter notebook examples/quick_start_tempo_pfn.ipynb
```

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

### Set Up Python Environment and Install Dependencies

Source: https://github.com/automl/tempopfn/blob/main/README.md

Steps to create a virtual environment, activate it, set the PYTHONPATH, install PyTorch with CUDA support, and install TempoPFN and its development dependencies. Ensure your PyTorch installation matches your CUDA version.

```bash
# 3. Set up the environment
python3.12 -m venv venv & source venv/bin/activate
export PYTHONPATH=$PWD

# 4. Install PyTorch version matching your CUDA version
pip install torch --index-url https://download.pytorch.org/whl/cu128

# 5. Install dependencies
pip install .
pip install .[dev]
```

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

### Install Git LFS and Clone Repository

Source: https://github.com/automl/tempopfn/blob/main/README.md

Instructions for installing Git LFS and cloning the TempoPFN Hugging Face repository. Ensure Git LFS is installed before cloning to handle large model files.

```bash
# 1. Install Git LFS (if you haven't already)
# On Ubuntu: sudo apt-get install git-lfs
# On macOS: brew install git-lfs
git lfs install

# 2. Clone the Hugging Face repository
git clone https://huggingface.co/AutoML-org/TempoPFN
cd TempoPFN
```

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

### Setup Imports and CUDA Check

Source: https://github.com/automl/tempopfn/blob/main/examples/quick_start_tempo_pfn.ipynb

Initializes necessary libraries, checks for CUDA availability, and sets the device for computation. Ensures the repository root is correctly identified for loading configurations.

```python
from pathlib import Path

import numpy as np
import torch
from huggingface_hub import hf_hub_download

# Ensure CUDA is available
if not torch.cuda.is_available():
    raise RuntimeError("CUDA is required to run this demo. No CUDA device detected.")

device = torch.device("cuda:0")

# Resolve repository root to be robust to running from subdirectories (e.g., examples/)
repo_root = Path.cwd()
if not (repo_root / "configs").exists():
    repo_root = repo_root.parent

# Inline plotting
%matplotlib inline
```

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

### Train TempoPFN on Single GPU

Source: https://github.com/automl/tempopfn/blob/main/README.md

Use this command to start training on a single GPU for debugging purposes. Ensure the config file path is correct.

```bash
# Single-GPU (Debug)
torchrun --standalone --nproc_per_node=1 src/training/trainer_dist.py --config ./configs/train.yaml
```

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

### Train TempoPFN on Multi-GPU

Source: https://github.com/automl/tempopfn/blob/main/README.md

Use this command to start training on multiple GPUs (e.g., 8 GPUs) for faster model training. Adjust nproc_per_node based on your available GPUs.

```bash
# Multi-GPU (e.g., 8 GPUs)
torchrun --standalone --nproc_per_node=8 src/training/trainer_dist.py --config ./configs/train.yaml
```

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

### Environment Setup and Dataset Properties Loading

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Sets up environment variables and loads dataset properties from a JSON file. Includes error handling for file loading and defines lists of short and medium/long datasets.

```python
# Environment setup
os.environ["CUBLAS_WORKSPACE_CONFIG"] = ":4096:8"

# Use absolute path relative to the project root
_MODULE_DIR = Path.cwd().parent.parent  # Assumes notebook is in `examples/gift_eval/`
DATASET_PROPERTIES_PATH = _MODULE_DIR / "data" / "dataset_properties.json"

try:
    with open(DATASET_PROPERTIES_PATH) as f:
        DATASET_PROPERTIES = json.load(f)
except Exception as exc:  # pragma: no cover - logging path
    DATASET_PROPERTIES = {}
    logger.warning(
        "Could not load dataset properties from %s: %s. Domain and num_variates will fall back to defaults.",
        DATASET_PROPERTIES_PATH,
        exc,
    )

# Datasets
SHORT_DATASETS = (
    "m4_yearly",
    "m4_quarterly",
    "m4_monthly",
    "m4_weekly",
    "m4_daily",
    "m4_hourly",
    "electricity/15T",
    "electricity/H",
    "electricity/D",
    "electricity/W",
    "solar/10T",
    "solar/H",
    "solar/D",
    "solar/W",
    "hospital",
    "covid_deaths",
    "us_births/D",
    "us_births/M",
    "us_births/W",
    "saugeenday/D",
    "saugeenday/M",
    "saugeenday/W",
    "temperature_rain_with_missing",
    "kdd_cup_2018_with_missing/H",
    "kdd_cup_2018_with_missing/D",
    "car_parts_with_missing",
    "restaurant",
    "hierarchical_sales/D",
    "hierarchical_sales/W",
    "LOOP_SEATTLE/5T",
    "LOOP_SEATTLE/H",
    "LOOP_SEATTLE/D",
    "SZ_TAXI/15T",
    "SZ_TAXI/H",
    "M_DENSE/H",
    "M_DENSE/D",
    "ett1/15T",
    "ett1/H",
    "ett1/D",
    "ett1/W",
    "ett2/15T",
    "ett2/H",
    "ett2/D",
    "ett2/W",
    "jena_weather/10T",
    "jena_weather/H",
    "jena_weather/D",
    "bitbrains_fast_storage/5T",
    "bitbrains_fast_storage/H",
    "bitbrains_rnd/5T",
    "bitbrains_rnd/H",
    "bizitobs_application",
    "bizitobs_service",
    "bizitobs_l2c/5T",
    "bizitobs_l2c/H",
)

MED_LONG_DATASETS = (
    "electricity/15T",
    "electricity/H",
    "solar/10T",
    "solar/H",
    "kdd_cup_2018_with_missing/H",
    "LOOP_SEATTLE/5T",
    "LOOP_SEATTLE/H",
    "SZ_TAXI/15T",
    "M_DENSE/H",
    "ett1/15T",
    "ett1/H",
    "ett2/15T",
    "ett2/H",
    "jena_weather/10T",
    "jena_weather/H",
    "bitbrains_fast_storage/5T",
    "bitbrains_rnd/5T",
    "bizitobs_application",
    "bizitobs_service",
    "bizitobs_l2c/5T",
    "bizitobs_l2c/H",
)

# Preserve insertion order
ALL_DATASETS = list(dict.fromkeys(SHORT_DATASETS + MED_LONG_DATASETS))

# Evaluation terms
TERMS = ("short", "medium", "long")

# Pretty names mapping
PRETTY_NAMES = {
    "saugeenday": "saugeen",
    "temperature_rain_with_missing": "temperature_rain",
    "kdd_cup_2018_with_missing": "kdd_cup_2018",
    "car_parts_with_missing": "car_parts",
}

# Metrics
METRICS = (
    MSE(forecast_type="mean"),
    MSE(forecast_type=0.5),
    MAE(),
    MASE(),
    MAPE(),
    SMAPE(),
    MSIS(),
    RMSE(),
    NRMSE(),
    ND(),
    MeanWeightedSumQuantileLoss(quantile_levels=[0.1, 0.2, 0.3, 0.4, 0.5, 0.6, 0.7, 0.8, 0.9]),
)

# Standard metric names for CSV header
STANDARD_METRIC_NAMES = (
    "MSE[mean]",
    "MSE[0.5]",
    "MAE[0.5]",
    "MASE[0.5]",
    "MAPE[0.5]",
    "sMAPE[0.5]",
    "MSIS",
    "RMSE[mean]",
    "NRMSE[mean]",
    "ND[0.5]",
    "mean_weighted_sum_quantile_loss",
)

```

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

### Import Libraries for TempoPFN and GIFT-Eval

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Imports all necessary libraries for data handling, modeling, evaluation, and plotting. Assumes TempoPFN core code is installed or in PYTHONPATH. Configures logging and filters specific warnings.

```python
import csv
import glob
import json
import logging
import math
import os
import warnings
from collections.abc import Iterable, Iterator
from dataclasses import dataclass
from enum import Enum
from functools import cached_property
from pathlib import Path

# GluonTS and Data Handling
import datasets

# Plotting and Warnings
import matplotlib.pyplot as plt
import numpy as np
import pandas as pd
import pyarrow.compute as pc
import torch
import yaml
from dotenv import load_dotenv
from gluonts.dataset import DataEntry
from gluonts.dataset.common import ProcessDataEntry
from gluonts.dataset.split import TestData, TrainingDataset, split

# GluonTS Evaluation
from gluonts.ev.metrics import (
    MAE,
    MAPE,
    MASE,
    MSE,
    MSIS,
    ND,
    NRMSE,
    RMSE,
    SMAPE,
    MeanWeightedSumQuantileLoss,
)
from gluonts.itertools import Map
from gluonts.model.evaluation import evaluate_model
from gluonts.model.forecast import QuantileForecast
from gluonts.model.predictor import Predictor
from gluonts.time_feature import get_seasonality, norm_freq_str
from gluonts.transform import Transformation
from huggingface_hub import hf_hub_download
from linear_operator.utils.cholesky import NumericalWarning
from pandas.tseries.frequencies import to_offset

# --- TempoPFN Core Model Imports ---
# These are assumed to be installed or in the PYTHONPATH
from src.data.containers import BatchTimeSeriesContainer
from src.data.frequency import parse_frequency
from src.data.scalers import RobustScaler
from src.models.model import TimeSeriesModel
from src.utils.utils import device
from toolz import compose
from torch.nn.parallel import DistributedDataParallel as DDP

# --- Setup Logging ---
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")
logging.getLogger("matplotlib").setLevel(logging.WARNING)
logging.getLogger("matplotlib.font_manager").setLevel(logging.WARNING)
logging.getLogger("PIL").setLevel(logging.WARNING)
logger = logging.getLogger("gift_eval_runner")


# Filter out specific gluonts warnings
class WarningFilter(logging.Filter):
    def __init__(self, text_to_filter: str) -> None:
        super().__init__()
        self.text_to_filter = text_to_filter

    def filter(self, record: logging.LogRecord) -> bool:
        return self.text_to_filter not in record.getMessage()


gts_logger = logging.getLogger("gluonts.model.forecast")
gts_logger.addFilter(WarningFilter("The mean prediction is not stored in the forecast data"))

# Filter out numerical warnings
warnings.filterwarnings("ignore", category=NumericalWarning)
warnings.filterwarnings("ignore", category=FutureWarning)
warnings.filterwarnings("ignore", category=DeprecationWarning)

# Load environment variables (e.g., GIFT_EVAL_DATASET_STORAGE_PATH)
load_dotenv()
```

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

### Pack Data into BatchTimeSeriesContainer

Source: https://github.com/automl/tempopfn/blob/main/examples/quick_start_tempo_pfn.ipynb

Organizes the generated history and future values, along with start and frequency information, into a `BatchTimeSeriesContainer`. This container is optimized for model input.

```python
from src.data.containers import BatchTimeSeriesContainer

container = BatchTimeSeriesContainer(
    history_values=history_values.to(device),
    future_values=future_values.to(device),
    start=batch.start,
    frequency=batch.frequency,
)

container.batch_size, container.history_length, container.future_length
```

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

### Get Training Dataset Split

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Provides the training portion of the dataset, split by offsetting from the end based on prediction length and number of windows.

```python
    @property
    def training_dataset(self) -> TrainingDataset:
        training_dataset, _ = split(self.gluonts_dataset, offset=-self.prediction_length * (self.windows + 1))
        return training_dataset
```

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

### Get Validation Dataset Split

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Provides the validation portion of the dataset, split by offsetting from the end. This is typically used for hyperparameter tuning.

```python
    @property
    def validation_dataset(self) -> TrainingDataset:
        validation_dataset, _ = split(self.gluonts_dataset, offset=-self.prediction_length * self.windows)
        return validation_dataset
```

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

### Convert to Batch Container

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Converts a list of GluonTS test data entries into a BatchTimeSeriesContainer suitable for model input. It handles target reshaping, context length limiting, and extracts start dates and frequencies.

```python
    def _convert_to_batch_container(self, test_data_batch: list) -> BatchTimeSeriesContainer:
        """Convert gluonts test data to BatchTimeSeriesContainer."""

        batch_size = len(test_data_batch)
        history_values_list = []
        start_dates = []
        frequencies = []

        for entry in test_data_batch:
            target = entry["target"]

            if target.ndim == 1:
                target = target.reshape(-1, 1)
            else:
                target = target.T

            if self.max_context_length is not None and len(target) > self.max_context_length:
                target = target[-self.max_context_length :]

            history_values_list.append(target)
            start_dates.append(entry["start"].to_timestamp().to_datetime64())
            frequencies.append(parse_frequency(entry["freq"]))

        history_values_np = np.stack(history_values_list, axis=0)
        num_channels = history_values_np.shape[2]

        history_values = torch.tensor(history_values_np, dtype=torch.float32, device=device)

        future_values = torch.zeros(
            (batch_size, self.ds_prediction_length, num_channels),
            dtype=torch.float32,
            device=device,
        )

        return BatchTimeSeriesContainer(
            history_values=history_values,
            future_values=future_values,
            start=start_dates,
            frequency=frequencies,
        )
```

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

### Get Dataset Frequency

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Retrieves the frequency string of the dataset from the first data entry. Assumes all entries share the same frequency.

```python
    @cached_property
    def freq(self) -> str:
        return self.hf_dataset[0]["freq"]
```

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

### Get All Dataset Full Names

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Retrieves a list of all possible dataset full names used for validation. This function iterates through predefined dataset names and terms to construct comprehensive identifiers.

```python
def get_all_datasets_full_name() -> list[str]:
    """Get all possible dataset full names for validation."""

    terms = ["short", "medium", "long"]
    datasets_full_names: list[str] = []

    for name in ALL_DATASETS:
        for term in terms:
            if term in ["medium", "long"] and name not in MED_LONG_DATASETS:
                continue

            if "/" in name:
                ds_key, ds_freq = name.split("/")
                ds_key = ds_key.lower()
                ds_key = PRETTY_NAMES.get(ds_key, ds_key)
            else:
                ds_key = name.lower()
                ds_key = PRETTY_NAMES.get(ds_key, ds_key)
                ds_freq = DATASET_PROPERTIES.get(ds_key, {}).get("frequency")

            datasets_full_names.append(f"{ds_key}/{ds_freq if ds_freq else 'unknown'}/{term}")

    return datasets_full_names
```

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

### Configure TempoPFN Evaluation Parameters

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Set up parameters for the evaluation run, including model path, configuration file, Hugging Face repository details, dataset selection, and output directory. This snippet also includes a helper function to load YAML configuration files.

```python
# --- Parameters ---
model_path = None  # e.g., "/path/to/checkpoint.pth"; if None, download from HF
config_path = Path.cwd().parent.parent / "configs/example.yaml"
hf_repo_id = "AutoML-org/TempoPFN"
hf_filename = "models/checkpoint_38M.pth"

# --- Datasets and evaluation controls ---
# Use a small subset for testing, e.g., ["m4_weekly"]
datasets_arg = ["all"]  # list of dataset names or ["all"].
terms = ["short", "medium", "long"]
dataset_storage_path = os.getenv("GIFT_EVAL_DATASET_STORAGE_PATH")
max_windows = 20
batch_size = 64
max_context_length = 3072

# --- Output ---
after_each_dataset_flush = True  # write CSV as each dataset completes
model_name = "TempoPFN"
output_dir = Path.cwd().parent / "gift_eval_results" / model_name


# --- Helper Functions ---
def _load_yaml(path: str) -> dict:
    with open(path) as f:
        return yaml.safe_load(f)
```

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

### Download Model Checkpoint

Source: https://github.com/automl/tempopfn/blob/main/examples/quick_start_tempo_pfn.ipynb

Downloads the pretrained TempoPFN model checkpoint from the Hugging Face Hub. The checkpoint is essential for running inference.

```python
print("Downloading model checkpoint from Hugging Face Hub...")

CHECKPOINT_PATH = hf_hub_download(repo_id="AutoML-org/TempoPFN", filename="models/checkpoint_38M.pth")

print(f"Checkpoint is available at: {CHECKPOINT_PATH}")
```

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

### Generate Test Data Instances

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Creates test data instances by splitting the dataset and generating samples with specified prediction length, windows, and distance. This prepares data for final model evaluation.

```python
    @property
    def test_data(self) -> TestData:
        _, test_template = split(self.gluonts_dataset, offset=-self.prediction_length * self.windows)
        test_data = test_template.generate_instances(
            prediction_length=self.prediction_length,
            windows=self.windows,
            distance=self.prediction_length,
        )
        return test_data
```

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

### Get Dynamic Real Feature Dimension

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Calculates the dimension of the 'past_feat_dynamic_real' feature. Returns 0 if the feature is absent, 1 if it's a single vector, or its shape if it's multi-dimensional.

```python
    @cached_property
    def past_feat_dynamic_real_dim(self) -> int:
        if "past_feat_dynamic_real" not in self.hf_dataset[0]:
            return 0
        elif len((past_feat_dynamic_real := self.hf_dataset[0]["past_feat_dynamic_real"]).shape) > 1:
            return past_feat_dynamic_real.shape[0]
        else:
            return 1
```

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

### Load Model and Run Inference with bfloat16

Source: https://github.com/automl/tempopfn/blob/main/examples/quick_start_tempo_pfn.ipynb

Loads the TempoPFN model configuration and weights, then performs inference using bfloat16 precision on CUDA for enhanced performance. Handles potential scaling of predictions.

```python
import yaml
from src.models.model import TimeSeriesModel

with open(repo_root / "configs/example.yaml") as f:
    config = yaml.safe_load(f)

model = TimeSeriesModel(**config["TimeSeriesModel"]).to(device)
ckpt = torch.load(CHECKPOINT_PATH, map_location=device)
model.load_state_dict(ckpt["model_state_dict"])
model.eval()

# bfloat16 autocast on CUDA
with (
    torch.no_grad(),
    torch.autocast(device_type="cuda", dtype=torch.bfloat16, enabled=True),
):
    output = model(container)

preds = output["result"].to(torch.float32)
if hasattr(model, "scaler") and "scale_statistics" in output:
    preds = model.scaler.inverse_scale(preds, output["scale_statistics"])

preds.shape
```

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

### Load TimeSeriesModel from Path

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

A static method to load a TimeSeriesModel from a given path, including its state dictionary, using the provided configuration. It handles model instantiation, state loading, and sets the model to evaluation mode.

```python
    @staticmethod
    def _load_model_from_path(config: dict, model_path: str) -> TimeSeriesModel:
        try:
            model = TimeSeriesModel(**config["TimeSeriesModel"]).to(device)
            checkpoint = torch.load(model_path, map_location=device)
            model.load_state_dict(checkpoint["model_state_dict"])
            model.eval()
            logger.info(f"Successfully loaded model from {model_path}")
            return model
        except Exception as exc:  # pragma: no cover - logging path
            logger.error(f"Failed to load model from {model_path}: {exc}")
            raise
```

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

### TempoPFN Main Evaluation Loop

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Initializes the TimeSeriesPredictor and executes the evaluation loop across specified datasets. Handles model downloading from Hugging Face if a local path is not provided. Logs progress and errors during evaluation.

```python
logger.info("Starting evaluation for model: %s", model_name)

# 1. Build predictor from a checkpoint
resolved_model_path = None
if model_path:
    logger.info("Using local model checkpoint from `model_path`: %s", model_path)
    resolved_model_path = model_path
else:
    logger.info("Downloading model from Hugging Face Hub...")
    logger.info("  Repo: %s", hf_repo_id)
    logger.info("  File: %s", hf_filename)
    try:
        resolved_model_path = hf_hub_download(
            repo_id=hf_repo_id,
            filename=hf_filename,
        )
        logger.info("Download complete. Model path: %s", resolved_model_path)
    except Exception as e:
        logger.error("Failed to download model from Hugging Face Hub: %s", e)
        raise e

if not resolved_model_path or not Path(resolved_model_path).exists():
    raise FileNotFoundError(
        f"No model checkpoint found. Set `model_path` or check HF settings. Tried: {resolved_model_path}"
    )

assert Path(config_path).exists(), f"Config not found: {config_path}"
logger.info("Loading predictor from checkpoint: %s", resolved_model_path)

predictor = TimeSeriesPredictor.from_paths(
    model_path=resolved_model_path,
    config_path=config_path,
    ds_prediction_length=1,  # placeholder; set per dataset
    ds_freq="D",  # placeholder; set per dataset
    batch_size=batch_size,
    max_context_length=max_context_length,
)

# 2. Run evaluation loop
datasets_to_run = expand_datasets_arg(datasets_arg)
results_root = Path(output_dir)

for ds_name in datasets_to_run:
    try:
        items = evaluate_datasets(
            predictor=predictor,
            dataset=ds_name,
            dataset_storage_path=dataset_storage_path,
            terms=terms,
            max_windows=max_windows,
            batch_size=batch_size,
            max_context_length=max_context_length,
            create_plots=False,  # Set to True if you implement plotting
            max_plots_per_dataset=0,
        )
        write_results_to_disk(
            items=items,
            dataset_name=ds_name,
            output_dir=results_root,
            model_name=model_name,
            create_plots=False,
        )
        if after_each_dataset_flush:
            logger.info("Flushed results for %s", ds_name)
    except Exception as e:
        logger.error(f"FAILED evaluation for dataset: {ds_name}. Error: {e} !!!")
        logger.exception(e)
        continue  # Continue to the next dataset

print(f"\nEvaluation complete. See results under: {output_dir}")
```

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

### Initialize TimeSeriesPredictor

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Initializes the TimeSeriesPredictor with a TimeSeriesModel, configuration, and dataset-specific parameters. It sets up the model for evaluation and initializes a scaler.

```python
class TimeSeriesPredictor(Predictor):
    """Unified predictor for TimeSeriesModel supporting flexible construction."""

    def __init__(
        self,
        model: TimeSeriesModel,
        config: dict,
        ds_prediction_length: int,
        ds_freq: str,
        batch_size: int = 32,
        max_context_length: int | None = None,
        debug: bool = False,
    ) -> None:
        # Dataset-specific context (can be updated per dataset/term)
        self.ds_prediction_length = ds_prediction_length
        self.ds_freq = ds_freq
        self.batch_size = batch_size
        self.max_context_length = max_context_length
        self.debug = debug

        # Persistent model/config (unwrap DDP if needed)
        self.model = model.module if isinstance(model, DDP) else model
        self.model.eval()
        self.config = config

        # Initialize scaler (using same type as model)
        scaler_type = self.config.get("TimeSeriesModel", {}).get("scaler", "custom_robust")
        epsilon = self.config.get("TimeSeriesModel", {}).get("epsilon", 1e-3)
        if scaler_type == "custom_robust":
            self.scaler = RobustScaler(epsilon=epsilon)
        else:
            raise ValueError(f"Unsupported scaler type: {scaler_type}")
```

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

### Configure Local Cache for Triton and TorchInductor

Source: https://github.com/automl/tempopfn/blob/main/README.md

Optimizes inference performance by routing Triton and TorchInductor caches to a local directory, preventing potential slowdowns on network filesystems. This is particularly useful for repeated runs or when using shared storage.

```bash
LOCAL_CACHE_BASE="${TMPDIR:-/tmp}/tsf-$(date +%s)"
mkdir -p "${LOCAL_CACHE_BASE}/triton" "${LOCAL_CACHE_BASE}/torchinductor"
export TRITON_CACHE_DIR="${LOCAL_CACHE_BASE}/triton"
export TORCHINDUCTOR_CACHE_DIR="${LOCAL_CACHE_BASE}/torchinductor"

python examples/quick_start_tempo_pfn.py
```

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

### Custom Implementation Forward Method Return Value

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

The custom implementation accepts and forwards the 'initial_state' to the attention layer, and its forward method signature reflects this capability.

```python
def forward(
    self,
    hidden_states: torch.Tensor,
    attention_mask: Optional[torch.Tensor] = None,
    past_key_values: Optional[Union[Cache, List[torch.FloatTensor]]] = None,
    use_cache: Optional[bool] = False,
    output_attentions: Optional[bool] = False,
    initial_state: Optional[torch.FloatTensor] = None,  # ← ADDED
    **kwargs: Unpack[Dict],
) -> Tuple[
    torch.FloatTensor, Optional[Tuple[torch.FloatTensor, torch.FloatTensor]]
]:
    # ...
    hidden_states, attentions, past_key_values = self.attn(
        # ...
        initial_state=initial_state,  # ← Passed through
        **kwargs,
    )
    # ...
    return outputs  # Returns (hidden_states, attentions, past_key_values)
```

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

### Custom Implementation Forward Method Signature

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

The custom implementation adds an 'initial_state' parameter to the forward method, enabling external control of the recurrent state for layer-to-layer propagation.

```python
def forward(
    self,
    hidden_states: torch.Tensor,
    attention_mask: Optional[torch.Tensor] = None,
    past_key_values: Optional[Cache] = None,
    initial_state: Optional[torch.Tensor] = None,  # ← ADDED
    use_cache: Optional[bool] = False,
    output_attentions: Optional[bool] = False,
    **kwargs: Unpack[Dict],
) -> Tuple[torch.Tensor, Optional[torch.Tensor], Optional[Cache]]:
```

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

### Learnable Initial Hidden States Initialization

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

Initializes a list of learnable parameters, one for each encoder layer. These parameters are used as initial hidden states and are trainable during the model's optimization process. They are initialized with small random values.

```python
num_initial_hidden_states = self.num_encoder_layers
self.initial_hidden_state = nn.ParameterList(
    [
        nn.Parameter(
            torch.randn(
                1, self.encoder_config["num_heads"], head_k_dim, head_v_dim
            )
            / head_k_dim,
            requires_grad=True,
        )
        for _ in range(num_initial_hidden_states)
    ]
)
```

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

### Create Predictor from File Paths

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

A class method to create a TimeSeriesPredictor by loading the model and configuration from specified file paths. This is useful for initializing the predictor when model artifacts are stored on disk.

```python
    @classmethod
    def from_paths(
        cls,
        model_path: str,
        config_path: str,
        ds_prediction_length: int,
        ds_freq: str,
        batch_size: int = 32,
        max_context_length: int | None = None,
        debug: bool = False,
    ) -> "TimeSeriesPredictor":
        with open(config_path) as f:
            config = yaml.safe_load(f)
        model = cls._load_model_from_path(config=config, model_path=model_path)
        return cls(
            model=model,
            config=config,
            ds_prediction_length=ds_prediction_length,
            ds_freq=ds_freq,
            batch_size=batch_size,
            max_context_length=max_context_length,
            debug=debug,
        )
```

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

### Custom Implementation Return Values

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

Shows the return signature of the custom implementation, which returns the output, the recurrent state, and past key-values. This allows for state propagation.

```python
return o, recurrent_state, past_key_values  # Returns (output, recurrent_state, past_key_values)
```

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

### Visualize Predictions and Intervals

Source: https://github.com/automl/tempopfn/blob/main/examples/quick_start_tempo_pfn.ipynb

Generates plots for each sample in the batch, showing the historical data, ground truth future values, predicted median, and a prediction interval. This helps in visually assessing the model's performance.

```python
import matplotlib.pyplot as plt

plt.set_loglevel("error")

# preds: [B, P, N, Q] for quantiles (univariate -> N=1)
preds_np = preds.cpu().numpy()

batch_size = preds_np.shape[0]
prediction_length = preds_np.shape[1]
num_quantiles = preds_np.shape[-1]

for i in range(batch_size):
    fig, ax = plt.subplots(figsize=(12, 4))

    history = container.history_values[i, :, 0].detach().cpu().numpy()
    future = container.future_values[i, :, 0].detach().cpu().numpy()

    # Time axes
    hist_t = np.arange(len(history))
    fut_t = np.arange(len(history), len(history) + len(future))

    # Plot history and ground truth future
    ax.plot(hist_t, history, label="History", color="black")
    ax.plot(fut_t, future, label="Ground Truth", color="blue")

    # Plot quantiles
    median_idx = num_quantiles // 2
    ax.plot(
        fut_t,
        preds_np[i, :, 0, median_idx],
        label="Prediction (Median)",
        color="orange",
        linestyle="--",
    )
    if num_quantiles >= 3:
        ax.fill_between(
            fut_t,
            preds_np[i, :, 0, 0],
            preds_np[i, :, 0, -1],
            color="orange",
            alpha=0.2,
            label="Prediction Interval",
        )

    ax.axvline(x=len(history), color="k", linestyle=":", alpha=0.7)
    ax.set_xlabel("Time Steps")
    ax.set_ylabel("Value")
    ax.set_title(f"Sample {i + 1}")
    ax.legend()
    ax.grid(True, alpha=0.3)
    plt.show()
```

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

### Ensure Results CSV Exists

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Creates a CSV file for results if it doesn't exist, including necessary headers. This function is used to initialize the results file before writing any data.

```python
def _ensure_results_csv(csv_file_path: Path) -> None:
    if not csv_file_path.exists():
        csv_file_path.parent.mkdir(parents=True, exist_ok=True)
        with open(csv_file_path, "w", newline="") as csvfile:
            writer = csv.writer(csvfile)
            header = (
                ["dataset", "model"] 
                + [f"eval_metrics/{name}" for name in STANDARD_METRIC_NAMES]
                + ["domain", "num_variates"]
            )
            writer.writerow(header)
```

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

### Convert Model Output to Forecasts

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Transforms the model's raw output into QuantileForecast objects. It handles unscaling predictions and determining whether the output represents quantiles or a single median forecast.

```python
    def _convert_to_forecasts(
        self,
        model_output: dict,
        test_data_batch: list,
        batch_container: BatchTimeSeriesContainer,
    ) -> list[QuantileForecast]:
        """Convert model predictions to QuantileForecast objects."""

        predictions = model_output["result"]
        scale_statistics = model_output["scale_statistics"]

        if predictions.ndim == 4:
            predictions_unscaled = self.scaler.inverse_scale(predictions, scale_statistics)
            is_quantile = True
            quantile_levels = self.model.quantiles
        else:
            predictions_unscaled = self.scaler.inverse_scale(predictions, scale_statistics)
            is_quantile = False
            quantile_levels = [0.5]

        forecasts: list[QuantileForecast] = []
        for idx, entry in enumerate(test_data_batch):
            history_length = int(batch_container.history_values.shape[1])
            start_date = entry["start"]
            forecast_start = start_date + history_length

            if is_quantile:
                pred_array = predictions_unscaled[idx].cpu().numpy()

                if pred_array.shape[1] == 1:
                    pred_array = pred_array.squeeze(1)
                    forecast_arrays = pred_array.T
                else:
                    forecast_arrays = pred_array.transpose(2, 0, 1)

```

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

### Custom Implementation Usage of initial_state in Chunk Mode

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

The custom implementation prioritizes the externally provided 'initial_state' over 'recurrent_state' from 'past_key_values' in chunk mode, facilitating state propagation.

```python
if mode == "chunk":
    o, recurrent_state = chunk_gated_delta_product(
        q=q, k=k, v=v, g=g, beta=beta,
        initial_state=initial_state,  # ← Uses external initial_state if provided
        output_final_state=output_attentions,
        cu_seqlens=cu_seqlens,
        num_householder=self.num_householder,
        use_qk_l2norm_in_kernel=True,
    )
```

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

### Official FLA Forward Method Signature

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

The official FLA implementation's forward method does not include an 'initial_state' parameter. It relies solely on 'recurrent_state' from 'past_key_values'.

```python
def forward(
    self,
    hidden_states: torch.Tensor,
    attention_mask: torch.Tensor | None = None,
    past_key_values: Cache | None = None,
    use_cache: bool | None = False,
    output_attentions: bool | None = False,
    **kwargs: Unpack[dict],
) -> tuple[torch.Tensor, torch.Tensor | None, Cache | None]:
```

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

### Dataset Class for Time Series Data

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Manages loading, preprocessing, and splitting of time series datasets. It handles frequency normalization, prediction length calculation, and conversion to univariate format if specified.

```python
class Dataset:
    def __init__(
        self,
        name: str,
        term: Term | str = Term.SHORT,
        to_univariate: bool = False,
        storage_path: str = None,
        max_windows: int | None = None,
    ):
        storage_path = Path(storage_path)
        self.hf_dataset = datasets.load_from_disk(str(storage_path / name)).with_format("numpy")
        process = ProcessDataEntry(
            self.freq,
            one_dim_target=self.target_dim == 1,
        )

        self.gluonts_dataset = Map(compose(process, itemize_start), self.hf_dataset)
        if to_univariate:
            self.gluonts_dataset = MultivariateToUnivariate("target").apply(self.gluonts_dataset)

        self.term = Term(term)
        self.name = name
        self.max_windows = max_windows if max_windows is not None else MAX_WINDOW
```

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

### Create Predictor from Model Instance

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

A class method to construct a TimeSeriesPredictor directly from an existing TimeSeriesModel instance and its configuration. Simplifies predictor instantiation when the model is already loaded.

```python
    @classmethod
    def from_model(
        cls,
        model: TimeSeriesModel,
        config: dict,
        ds_prediction_length: int,
        ds_freq: str,
        batch_size: int = 32,
        max_context_length: int | None = None,
        debug: bool = False,
    ) -> "TimeSeriesPredictor":
        return cls(
            model=model,
            config=config,
            ds_prediction_length=ds_prediction_length,
            ds_freq=ds_freq,
            batch_size=batch_size,
            max_context_length=max_context_length,
            debug=debug,
        )
```

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

### Construct Evaluation Data

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Builds datasets and associated metadata for evaluation. It handles different dataset terms (short, medium, long) and probes dimensionality to determine if univariate conversion is needed.

```python
def construct_evaluation_data(
    dataset_name: str,
    dataset_storage_path: str,
    terms: list[str] | None = None,
    max_windows: int | None = None,
) -> list[tuple[Dataset, DatasetMetadata]]:
    """Build datasets and rich metadata per term for a dataset name."""
    # Avoid mutable default argument
    if terms is None:
        terms = ["short", "medium", "long"]

    sub_datasets: list[tuple[Dataset, DatasetMetadata]] = []

    if "/" in dataset_name:
        ds_key, ds_freq = dataset_name.split("/")
        ds_key = ds_key.lower()
        ds_key = PRETTY_NAMES.get(ds_key, ds_key)
    else:
        ds_key = dataset_name.lower()
        ds_key = PRETTY_NAMES.get(ds_key, ds_key)
        ds_freq = DATASET_PROPERTIES.get(ds_key, {}).get("frequency")

    for term in terms:
        # Skip medium/long terms for datasets that don't support them
        if (term == "medium" or term == "long") and dataset_name not in MED_LONG_DATASETS:
            continue

        # Probe once to determine dimensionality
        probe_dataset = Dataset(
            name=dataset_name,
            term=term,
            to_univariate=False,
            storage_path=dataset_storage_path,
            max_windows=max_windows,
        )

        to_univariate = probe_dataset.target_dim > 1

        dataset = Dataset(
            name=dataset_name,
            term=term,
            to_univariate=to_univariate,
            storage_path=dataset_storage_path,
            max_windows=max_windows,
        )

        # Compute metadata
        season_length = get_seasonality(dataset.freq)
        actual_freq = ds_freq if ds_freq else dataset.freq

        metadata = DatasetMetadata(
            full_name=f"{ds_key}/{actual_freq}/{term}",
            key=ds_key,
            freq=actual_freq,
            term=term,
            season_length=season_length,
            target_dim=probe_dataset.target_dim,
            to_univariate=to_univariate,
            prediction_length=dataset.prediction_length,
            windows=dataset.windows,
        )

        sub_datasets.append((dataset, metadata))

    return sub_datasets
```

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

### Calculate Number of Windows

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Determines the number of windows for training/testing based on the test split ratio, minimum series length, prediction length, and a maximum window limit. Ensures at least one window.

```python
    @cached_property
    def windows(self) -> int:
        if "m4" in self.name:
            return 1
        w = math.ceil(TEST_SPLIT * self._min_series_length / self.prediction_length)
        return min(max(1, w), self.max_windows)
```

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

### State Propagation in State Weaving

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

This code illustrates the state propagation step in the state weaving mechanism. It shows how the final hidden state of a layer is added to the learnable initial state of the next layer to enable bidirectional information flow.

```python
H_0^{i+1} = H_0^{i+1} + H_t^i
```

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

### Official FLA Usage of initial_state in Chunk Mode

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

In chunk mode, the official FLA implementation uses 'recurrent_state' directly from 'past_key_values' as the initial state.

```python
if mode == 'chunk':
    o, recurrent_state = chunk_gated_delta_product(
        q=q, k=k, v=v, g=g, beta=beta,
        initial_state=recurrent_state,  # ← Only from past_key_values
        output_final_state=use_cache,
        cu_seqlens=cu_seqlens,
        num_householder=self.num_householder,
        use_qk_l2norm_in_kernel=True,
    )
```

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

### Official FLA Return Values

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

Shows the return signature of the official FLA implementation, which returns the output, None for the recurrent state, and past key-values.

```python
return o, None, past_key_values  # Returns (output, None, past_key_values)
```

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

### Write Evaluation Results to Disk

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

Writes evaluation metrics for a given dataset to a CSV file and optionally saves generated plots. It ensures the output directory and CSV file exist, then appends the results for each evaluation item.

```python
def write_results_to_disk(
    items: list[EvaluationItem],
    dataset_name: str,
    output_dir: Path,
    model_name: str,
    create_plots: bool,
) -> None:
    output_dir = output_dir / dataset_name
    output_dir.mkdir(parents=True, exist_ok=True)
    output_csv_path = output_dir / "results.csv"
    _ensure_results_csv(output_csv_path)

    with open(output_csv_path, "a", newline="") as csvfile:
        writer = csv.writer(csvfile)
        for item in items:
            md: DatasetMetadata = item.dataset_metadata
            metric_values: list[float | None] = []
            for metric_name in STANDARD_METRIC_NAMES:
                value = item.metrics.get(metric_name, None)
                if value is None:
                    metric_values.append(None)
                else:
                    if hasattr(value, "__len__") and not isinstance(value, (str, bytes)) and len(value) == 1:
                        value = value[0]
                    elif hasattr(value, "item"):
                        value = value.item()
                    metric_values.append(value)

            ds_key = md.key.lower()
            props = DATASET_PROPERTIES.get(ds_key, {})
            domain = props.get("domain", "unknown")
            num_variates = props.get("num_variates", 1 if md.to_univariate else md.target_dim)

            row = [md.full_name, model_name] + metric_values + [domain, num_variates]
            writer.writerow(row)

            if create_plots and item.figures and plt is not None:
                plots_dir = output_dir / "plots" / md.key / md.term
                plots_dir.mkdir(parents=True, exist_ok=True)
                for fig, filename in item.figures:
                    filepath = plots_dir / filename
                    fig.savefig(filepath, dpi=300, bbox_inches="tight")
                    plt.close(fig)

    logger.info(
        "Evaluation complete for dataset '%s'. Results saved to %s",
        dataset_name,
        output_csv_path,
    )
    if create_plots:
        logger.info("Plots saved under %s", output_dir / "plots")
```

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

### Generate Forecasts

Source: https://github.com/automl/tempopfn/blob/main/examples/gift_eval/gift_eval_submission.ipynb

The main predict method that takes test data input and generates forecasts. It handles input formatting and groups time series by effective length for efficient batch processing.

```python
    def predict(self, test_data_input) -> Iterator[QuantileForecast]:
        """Generate forecasts for the test data."""

        if hasattr(test_data_input, "__iter__") and not isinstance(test_data_input, list):
            test_data_input = list(test_data_input)
        logger.debug(f"Processing {len(test_data_input)} time series")

        # Group series by their effective length (after optional truncation),
        # then process each uniform-length group in sub-batches up to batch_size.
        def _effective_length(entry) -> int:
            target = entry["target"]
            if target.ndim == 1:
                seq_len = len(target)
            else:
                # target shape is [num_channels, seq_len]
                seq_len = target.shape[1]
            if self.max_context_length is not None:
                seq_len = min(seq_len, self.max_context_length)
            return seq_len
```

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

### Generate Synthetic Sine Wave Data

Source: https://github.com/automl/tempopfn/blob/main/examples/quick_start_tempo_pfn.ipynb

Generates synthetic time series data using a sine wave pattern. This includes defining parameters for the generator and creating a batch of time series values.

```python
from src.synthetic_generation.generator_params import SineWaveGeneratorParams
from src.synthetic_generation.sine_waves.sine_wave_generator_wrapper import \
    SineWaveGeneratorWrapper

batch_size = 3
total_length = 1024
seed = 2025

sine_params = SineWaveGeneratorParams(global_seed=seed, length=total_length)
wrapper = SineWaveGeneratorWrapper(sine_params)

batch = wrapper.generate_batch(batch_size=batch_size, seed=seed)
values = torch.from_numpy(batch.values).to(torch.float32)
if values.ndim == 2:
    values = values.unsqueeze(-1)  # [B, S, 1]

future_length = 256
history_values = values[:, :-future_length, :]
future_values = values[:, -future_length:, :]

print("History:", history_values.shape, "Future:", future_values.shape)
```

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

### Official FLA Forward Method Return Value

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

The official FLA implementation's forward method returns hidden states and attentions, but does not explicitly handle or return an 'initial_state'.

```python
def forward(
    self,
    hidden_states: torch.Tensor,
    attention_mask: torch.Tensor | None = None,
    past_key_values: Cache | list[torch.FloatTensor] | None = None,
    use_cache: bool | None = False,
    output_attentions: bool | None = False,
    **kwargs: Unpack[dict],
) -> tuple[torch.FloatTensor, tuple[torch.FloatTensor, torch.FloatTensor] | None]:
    # ...
    return outputs  # Returns (hidden_states, attentions, past_key_values)
```

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

### Encoder Wrapper Return Values

Source: https://github.com/automl/tempopfn/blob/main/src/models/gated_deltaproduct/README.md

Illustrates the return values from the GatedDeltaProductEncoder, which includes the output and the last hidden state. This hidden state is crucial for state propagation between layers in the TimeSeriesModel.

```python
x, last_hidden_state, _ = self.encoder_layer(
    x, output_attentions=True, initial_state=initial_state
)
return x, last_hidden_state  # ← Returns hidden state for weaving
```