"""
TimeSeriesDataset -- sliding-window train/val/test splits for univariate,
non-exogenous forecasting.
Guarantees temporal ordering, computes scaling parameters on the train split
only (no leakage), and can pad the test targets with NaN for iterative /
autoregressive forecasting.
"""
from __future__ import annotations
from typing import Literal, Optional, Tuple
import numpy as np
[docs]
class TimeSeriesDataset:
"""Create lookback/lookahead windows for forecasting.
Parameters
----------
series : ndarray, shape (n_timesteps,)
Univariate time series.
lookback : int
Number of past steps used as input X.
lookahead : int
Number of future steps to predict (Y).
test_size : int
Number of final timesteps reserved for testing.
val_size : int, optional
Validation length; defaults to ``2 * test_size``.
scaling : {'none', 'minmax', 'standard', 'log'}, default='none'
Scaling fitted on the train split only.
"""
def __init__(
self,
series: np.ndarray,
lookback: int,
lookahead: int,
test_size: int,
val_size: Optional[int] = None,
scaling: Literal["none", "minmax", "standard", "log"] = "none",
):
series = np.asarray(series).flatten()
if series.ndim != 1:
raise ValueError("series must be 1-dimensional")
if lookback < 1:
raise ValueError("lookback must be >= 1")
if lookahead < 1:
raise ValueError("lookahead must be >= 1")
if test_size < 1:
raise ValueError("test_size must be >= 1")
if val_size is None:
val_size = 2 * test_size
if val_size < 1:
raise ValueError("val_size must be >= 1")
if scaling not in ("none", "minmax", "standard", "log"):
raise ValueError(
f"scaling must be one of ['none','minmax','standard','log'], got '{scaling}'"
)
if scaling == "log" and np.any(series <= 0):
raise ValueError("Log scaling requires all data to be positive (> 0)")
min_required = lookback + lookahead + test_size + val_size
if len(series) < min_required:
raise ValueError(
f"Series too short. Need at least {min_required} timesteps, got {len(series)}"
)
self.series_ = series
self.scaling_ = scaling
self.lookback_ = lookback
self.lookahead_ = lookahead
self.test_size_ = test_size
self.val_size_ = val_size
# Split BEFORE scaling so scaling params come from train data only
test_series = series[-test_size:]
train_val_series = series[:-test_size]
val_series = train_val_series[-val_size:]
train_series = train_val_series[:-val_size]
self._compute_scaling_params(train_series)
self.train_series_ = self._apply_scaling(train_series)
self.val_series_ = self._apply_scaling(val_series)
self.test_series_ = self._apply_scaling(test_series)
self.full_train_series_ = self._apply_scaling(train_val_series)
self.series_scaled_ = self._apply_scaling(series)
self.train_end_idx_ = len(self.train_series_)
self.val_start_idx_ = self.train_end_idx_
self.val_end_idx_ = self.val_start_idx_ + len(self.val_series_)
self.test_start_idx_ = self.val_end_idx_
# ------------------------------------------------------------- scaling
def _compute_scaling_params(self, train_data: np.ndarray) -> None:
if self.scaling_ == "minmax":
self.scale_min_ = np.min(train_data)
self.scale_max_ = np.max(train_data)
if self.scale_max_ == self.scale_min_:
raise ValueError("Cannot minmax-scale: all training values identical")
elif self.scaling_ == "standard":
self.scale_mean_ = np.mean(train_data)
self.scale_std_ = np.std(train_data)
if self.scale_std_ == 0:
raise ValueError("Cannot standard-scale: training data has zero variance")
def _apply_scaling(self, data: np.ndarray) -> np.ndarray:
if self.scaling_ == "minmax":
return (data - self.scale_min_) / (self.scale_max_ - self.scale_min_)
if self.scaling_ == "standard":
return (data - self.scale_mean_) / self.scale_std_
if self.scaling_ == "log":
return np.log(data)
return data.copy()
[docs]
def inverse_scaling(self, data: np.ndarray) -> np.ndarray:
"""Map scaled values (e.g. predictions) back to the original scale."""
if self.scaling_ == "minmax":
return data * (self.scale_max_ - self.scale_min_) + self.scale_min_
if self.scaling_ == "standard":
return data * self.scale_std_ + self.scale_mean_
if self.scaling_ == "log":
return np.exp(data)
return data.copy()
# ------------------------------------------------------------- windows
def _create_sliding_windows(self, data: np.ndarray) -> Tuple[np.ndarray, np.ndarray]:
X_list, y_list = [], []
for i in range(len(data) - self.lookback_ - self.lookahead_ + 1):
X_list.append(data[i : i + self.lookback_])
y_list.append(data[i + self.lookback_ : i + self.lookback_ + self.lookahead_])
if not X_list:
return (
np.array([]).reshape(0, self.lookback_),
np.array([]).reshape(0, self.lookahead_),
)
return np.array(X_list), np.array(y_list)
[docs]
def get_train_data(self) -> Tuple[np.ndarray, np.ndarray]:
"""Windows over the train split (for hyperparameter tuning)."""
return self._create_sliding_windows(self.train_series_)
[docs]
def get_val_data(self) -> Tuple[np.ndarray, np.ndarray]:
"""Windows over the validation split, with lookback context from train."""
context_size = self.lookback_
if len(self.train_series_) >= context_size:
combined = np.concatenate([self.train_series_[-context_size:], self.val_series_])
else:
combined = self.val_series_
X, y = self._create_sliding_windows(combined)
n_context_windows = max(0, context_size - self.lookahead_)
if n_context_windows > 0 and len(X) > n_context_windows:
X, y = X[n_context_windows:], y[n_context_windows:]
return X, y
[docs]
def get_full_train_data(self) -> Tuple[np.ndarray, np.ndarray]:
"""Windows over train+val (for final model training)."""
return self._create_sliding_windows(self.full_train_series_)
[docs]
def get_test_data(self, use_predictions: bool = False) -> Tuple[np.ndarray, np.ndarray]:
"""Windows over the test split, with train+val context.
Targets that run past the end of the series are padded with NaN so
callers can implement iterative forecasting.
"""
context_size = self.lookback_
if len(self.full_train_series_) >= context_size:
combined = np.concatenate([self.full_train_series_[-context_size:], self.test_series_])
else:
combined = self.test_series_
X_list, y_list = [], []
max_start = len(combined) - self.lookback_
for i in range(max_start + 1):
X = combined[i : i + self.lookback_]
y_start = i + self.lookback_
y_end = y_start + self.lookahead_
if y_end <= len(combined):
y = combined[y_start:y_end]
else:
available = combined[y_start : len(combined)]
padding = np.full(y_end - len(combined), np.nan)
y = np.concatenate([available, padding])
X_list.append(X)
y_list.append(y)
if not X_list:
return (
np.array([]).reshape(0, self.lookback_),
np.array([]).reshape(0, self.lookahead_),
)
X, y = np.array(X_list), np.array(y_list)
n_context_windows = max(0, context_size - self.lookahead_)
if n_context_windows > 0 and len(X) > n_context_windows:
X, y = X[n_context_windows:], y[n_context_windows:]
return X, y
[docs]
def get_info(self) -> dict:
"""Summary of split sizes, ranges and scaling parameters."""
info = {
"series_length": len(self.series_),
"lookback": self.lookback_,
"lookahead": self.lookahead_,
"test_size": self.test_size_,
"val_size": self.val_size_,
"train_length": len(self.train_series_),
"val_length": len(self.val_series_),
"full_train_length": len(self.full_train_series_),
"scaling": self.scaling_,
}
if self.scaling_ == "minmax":
info["scale_min"] = self.scale_min_
info["scale_max"] = self.scale_max_
elif self.scaling_ == "standard":
info["scale_mean"] = self.scale_mean_
info["scale_std"] = self.scale_std_
return info
def __repr__(self) -> str:
info = self.get_info()
return (
f"TimeSeriesDataset(series_length={info['series_length']}, "
f"lookback={info['lookback']}, lookahead={info['lookahead']}, "
f"train={info['train_length']}, val={info['val_length']}, "
f"test={info['test_size']})"
)
