dataset

sorix.utils.data.dataset

Dataset

Dataset(X, y=None, transform=None, target_transform=None)

Base class for all datasets in Sorix.

Inspired by PyTorch's Dataset API, it provides a standard way to wrap data and apply transformations during retrieval.

Parameters:

• X (Any) –

  Feature data (NumPy array, list, etc.).

• y (Any, default: None ) –

  Target data (optional).

• transform (Optional[Callable], default: None ) –

  A function/transform that takes in a sample and returns a transformed version.

• target_transform (Optional[Callable], default: None ) –

  A function/transform that takes in the target and transforms it.

Source code in sorix/utils/data/dataset.py

def __init__(
    self, 
    X: Any, 
    y: Any = None, 
    transform: Optional[Callable] = None, 
    target_transform: Optional[Callable] = None
):
    if y is not None and len(X) != len(y):
        raise ValueError(f"X and y must have the same length. Got len(X)={len(X)} and len(y)={len(y)}")
    self.X = X
    self.y = y
    self.transform = transform
    self.target_transform = target_transform

WalkForwardSplit

WalkForwardSplit(
    n_splits=5,
    train_size=None,
    val_size=None,
    gap=0,
    expanding=None,
)

Chronological (walk-forward) cross-validation splitter.

Unlike random k-fold, this splitter respects temporal order: the training window always precedes the validation window, preventing future data leakage. Use this for time-series datasets such as match histories or financial data.

There are two modes:

• Expanding window (default): the training set grows with each fold — all data before the validation window is used.
• Rolling window (expanding=False): the training set is a fixed-size sliding window of the last train_size samples.

Because train_size only means something for a rolling window, passing it selects rolling mode automatically. Passing both train_size and an explicit expanding=True is contradictory and raises ValueError.

Parameters:

• n_splits (int, default: 5 ) –

  Number of splits. Default: 5.

• train_size (int | None, default: None ) –

  Size of the rolling training window. Passing it switches to rolling mode. None keeps the expanding window, which uses all data before the validation window. Default: None.

• val_size (int | None, default: None ) –

  Number of validation samples per split. None auto-computes len(X) // (n_splits + 1). Default: None.

• gap (int, default: 0 ) –

  Number of samples to drop between the training and validation windows (e.g. to simulate a prediction lag). Default: 0.

• expanding (bool | None, default: None ) –

  Window mode. None (default) infers it from train_size. Pass False for a rolling window, True to force an expanding one.

Raises:

• ValueError –

  If train_size is combined with expanding=True, or if any size argument is out of range.

Note

split() yields slices of the data, not index arrays as scikit-learn's TimeSeriesSplit does. With a pandas object, pass df.to_numpy() or slice with .iloc yourself.

Example::

splitter = WalkForwardSplit(n_splits=5, val_size=50, gap=1)
for train_X, train_y, val_X, val_y in splitter.split(X, y):
    model.fit(train_X, train_y)
    preds = model.predict(val_X)

Source code in sorix/utils/data/dataset.py

def __init__(
    self,
    n_splits: int = 5,
    train_size: Optional[int] = None,
    val_size: Optional[int] = None,
    gap: int = 0,
    expanding: Optional[bool] = None,
) -> None:
    if n_splits < 1:
        raise ValueError("n_splits must be >= 1")
    if gap < 0:
        raise ValueError("gap must be >= 0")
    if train_size is not None and train_size < 1:
        raise ValueError(f"train_size must be >= 1, got {train_size}")
    if val_size is not None and val_size < 1:
        raise ValueError(f"val_size must be >= 1, got {val_size}")

    if expanding is None:
        # `train_size` is only meaningful for a rolling window, so supplying
        # it selects rolling mode.
        expanding = train_size is None
    elif expanding and train_size is not None:
        raise ValueError(
            "train_size only applies to a rolling window, but expanding=True "
            "was requested. Pass expanding=False for a fixed-size rolling "
            "window, or drop train_size to use an expanding window."
        )

    self.n_splits = n_splits
    self.train_size = train_size
    self.val_size = val_size
    self.gap = gap
    self.expanding = expanding

split

split(X, y=None)

Generate chronological train/validation splits.

Always yields exactly n_splits folds; if the data is too short to fit them all, it raises instead of silently returning fewer.

Parameters:

• X (Any) –

  Feature array with shape (n_samples, ...).

• y (Optional[Any], default: None ) –

  Target array with shape (n_samples,). Optional.

Yields:

• Tuple[Any, ...] –

  Tuple[Any, ...]: (train_X, train_y, val_X, val_y) if y is provided, otherwise (train_X, val_X).

Raises:

• ValueError –

  If y is shorter than X, or if len(X) cannot accommodate n_splits folds of val_size samples plus gap and at least one training sample.

Source code in sorix/utils/data/dataset.py

def split(
    self,
    X: Any,
    y: Optional[Any] = None,
) -> Iterator[Tuple[Any, ...]]:
    """
    Generate chronological train/validation splits.

    Always yields exactly ``n_splits`` folds; if the data is too short to fit
    them all, it raises instead of silently returning fewer.

    Args:
        X: Feature array with shape ``(n_samples, ...)``.
        y: Target array with shape ``(n_samples,)``. Optional.

    Yields:
        Tuple[Any, ...]: ``(train_X, train_y, val_X, val_y)`` if ``y`` is
            provided, otherwise ``(train_X, val_X)``.

    Raises:
        ValueError: If ``y`` is shorter than ``X``, or if ``len(X)`` cannot
            accommodate ``n_splits`` folds of ``val_size`` samples plus
            ``gap`` and at least one training sample.
    """
    n = len(X)
    if y is not None and len(y) != n:
        raise ValueError(
            f"X and y must have the same length. Got len(X)={n} and len(y)={len(y)}"
        )
    val_size = self.val_size if self.val_size is not None else max(1, n // (self.n_splits + 1))

    # The earliest fold validates on X[n - n_splits*val_size : ...], and needs
    # `gap` dropped samples plus >= 1 training sample before it.
    min_samples = self.n_splits * val_size + self.gap + 1
    if n < min_samples:
        raise ValueError(
            f"Not enough samples for {self.n_splits} chronological splits: "
            f"len(X)={n} but {min_samples} are required "
            f"(n_splits * val_size + gap + 1 = {self.n_splits} * {val_size} "
            f"+ {self.gap} + 1). Reduce n_splits, val_size or gap."
        )

    # Determine start indices for each validation fold
    val_starts = []
    for k in range(self.n_splits):
        val_end = n - (self.n_splits - 1 - k) * val_size
        val_start = val_end - val_size
        val_starts.append((val_start, val_end))

    for val_start, val_end in val_starts:
        train_end = val_start - self.gap

        if self.expanding or self.train_size is None:
            train_start = 0
        else:
            train_start = max(0, train_end - self.train_size)

        train_X = X[train_start:train_end]
        val_X = X[val_start:val_end]

        if y is not None:
            train_y = y[train_start:train_end]
            val_y = y[val_start:val_end]
            yield train_X, train_y, val_X, val_y
        else:
            yield train_X, val_X