Reference

Utilities¶

Data loading and ESN-shaped splitting, seeded RNG construction, and the echo-state-property diagnostic. load_file and prepare_esn_data are re-exported at resdag.utils for convenience; their canonical home is resdag.utils.data. The ESP index, esp_index, is likewise re-exported at resdag.utils and at the top level, so all three of these work:

from resdag import esp_index
from resdag.utils import esp_index
from resdag.utils.states import esp_index  # canonical home

utils ¶

Utility Functions¶

This module provides utility functions for data loading, preparation, and general operations in resdag.

Submodules

data Data loading and preparation utilities for ESN training. states Reservoir-state diagnostics (the Echo State Property index).

FUNCTION	DESCRIPTION
`seed_everything`	Seed Python, NumPy, and torch (CPU + CUDA) for reproducible runs.
`resolve_device`	Resolve a device spec (`'auto'`/str/`torch.device`) to a `torch.device`.
`create_rng`	Create a NumPy random number generator with optional seed.
`create_torch_generator`	Create a torch `Generator` for reproducible weight/bias draws.
`coerce_seed_to_int`	Reduce an int/`torch.Generator`/None seed to a plain int (or None).
`SeedLike`	Type alias for accepted seed inputs (`int \| torch.Generator \| None`).
`DeviceLike`	Type alias for accepted device specs (`str \| torch.device \| None`).
`load_file`	Load a time series from disk (re-exported from `resdag.data`).
`prepare_esn_data`	Split a series into ESN train/forecast segments (re-exported from `resdag.data`).
`normalize_data`	Normalize a series and return the fitted statistics (re-exported from `resdag.data`).
`denormalize_data`	Invert a normalization back to the original scale (re-exported from `resdag.data`).
`lorenz, rossler, henon, mackey_glass, narma, sine`	Canonical reservoir-computing dataset generators (re-exported from `resdag.datasets`).
`esp_index`	Echo State Property index — the library's signature stability diagnostic (re-exported from `resdag.utils.states`).

Examples:

>>> import resdag as rd
>>> data = rd.utils.load_file("timeseries.csv")
>>> warmup, train, target, f_warmup, val = rd.utils.prepare_esn_data(
...     data, warmup_steps=100, train_steps=500, val_steps=200
... )

>>> series = rd.utils.lorenz(2000)  # (1, 2000, 3) chaotic benchmark series

>>> from resdag.utils import esp_index  # signature stability diagnostic

See Also

resdag.utils.data : Data loading and preparation. resdag.utils.states : Echo State Property diagnostics.

create_rng ¶

create_rng(seed: int | Generator | None = None) -> Generator

Create a NumPy random number generator.

This is the NumPy-RNG-for-graph-topology helper: it is threaded into the graph builders and named initializers, which draw exclusively from NumPy. For whole-program reproducibility (torch + NumPy + random at once) use :func:seed_everything instead; for a torch :class:~torch.Generator for weight/bias draws use :func:create_torch_generator.

PARAMETER DESCRIPTION

seed

If int, used as seed for a new Generator. If Generator, returned as-is. If None, the seed is derived from torch's global RNG so that torch.manual_seed propagates to the generator. This keeps graph-based topologies (which draw from NumPy) reproducible under the same torch global seed, matching the behaviour of matrix topologies that draw from torch directly.

TYPE: int, np.random.Generator, or None DEFAULT: None

RETURNS	DESCRIPTION
`Generator`	A NumPy random number generator.

Notes

When seed is None the numpy seed is sampled from torch's global RNG via torch.randint. This advances the torch global RNG state, but ties every NumPy draw to torch.manual_seed for end-to-end reproducibility.

See Also

seed_everything : Seed torch, NumPy, and random in one call. create_torch_generator : Build a torch Generator for weight draws.

Source code in src/resdag/utils/general.py

def create_rng(seed: int | np.random.Generator | None = None) -> np.random.Generator:
    """Create a NumPy random number generator.

    This is the NumPy-RNG-for-graph-topology helper: it is threaded into the
    graph builders and named initializers, which draw exclusively from NumPy.
    For whole-program reproducibility (torch + NumPy + ``random`` at once) use
    :func:`seed_everything` instead; for a torch :class:`~torch.Generator` for
    weight/bias draws use :func:`create_torch_generator`.

    Parameters
    ----------
    seed : int, np.random.Generator, or None
        If int, used as seed for a new Generator.
        If Generator, returned as-is.
        If None, the seed is derived from torch's global RNG so that
        ``torch.manual_seed`` propagates to the generator. This keeps
        graph-based topologies (which draw from NumPy) reproducible under
        the same torch global seed, matching the behaviour of matrix
        topologies that draw from torch directly.

    Returns
    -------
    np.random.Generator
        A NumPy random number generator.

    Notes
    -----
    When ``seed is None`` the numpy seed is sampled from torch's global RNG
    via ``torch.randint``. This advances the torch global RNG state, but ties
    every NumPy draw to ``torch.manual_seed`` for end-to-end reproducibility.

    See Also
    --------
    seed_everything : Seed torch, NumPy, and ``random`` in one call.
    create_torch_generator : Build a torch ``Generator`` for weight draws.
    """
    if isinstance(seed, np.random.Generator):
        return seed
    if seed is None:
        # Derive the numpy seed from torch's global RNG so that
        # torch.manual_seed(...) makes graph generators reproducible.
        seed = int(torch.randint(0, 2**63 - 1, (1,)).item())
    return np.random.default_rng(seed)

create_torch_generator ¶

create_torch_generator(seed: SeedLike = None, device: device | str | None = None) -> Generator

Create a torch :class:~torch.Generator for reproducible weight draws.

Used for the default nn.init weight/bias draws inside reservoir cells so that a single seed deterministically fixes every parameter without touching (and without depending on the prior state of) torch's global RNG.

PARAMETER	DESCRIPTION
`seed`	If `int`, seeds a fresh generator on `device`. If `torch.Generator`, it is returned as-is (the caller's generator is threaded straight through, so successive draws advance one stream). If `None`, the generator is seeded from torch's global RNG so that `torch.manual_seed` still propagates — matching :func:`create_rng`. TYPE: `int, torch.Generator, or None` DEFAULT: `None`
`device`	Device for the new generator. Ignored when `seed` is already a :class:`torch.Generator`. TYPE: `torch.device, str, or None` DEFAULT: `None`

RETURNS	DESCRIPTION
`Generator`	A generator suitable for passing to `nn.init.*` via `generator=`.

Notes

When seed is None the integer seed is sampled from torch's global RNG via torch.randint, advancing the global state but tying the draws to torch.manual_seed for end-to-end reproducibility.

Source code in src/resdag/utils/general.py

def create_torch_generator(
    seed: SeedLike = None,
    device: torch.device | str | None = None,
) -> torch.Generator:
    """Create a torch :class:`~torch.Generator` for reproducible weight draws.

    Used for the default ``nn.init`` weight/bias draws inside reservoir cells so
    that a single ``seed`` deterministically fixes every parameter without
    touching (and without depending on the prior state of) torch's global RNG.

    Parameters
    ----------
    seed : int, torch.Generator, or None
        If ``int``, seeds a fresh generator on ``device``.
        If ``torch.Generator``, it is returned as-is (the caller's generator is
        threaded straight through, so successive draws advance one stream).
        If ``None``, the generator is seeded from torch's global RNG so that
        ``torch.manual_seed`` still propagates — matching :func:`create_rng`.

    device : torch.device, str, or None
        Device for the new generator.  Ignored when ``seed`` is already a
        :class:`torch.Generator`.

    Returns
    -------
    torch.Generator
        A generator suitable for passing to ``nn.init.*`` via ``generator=``.

    Notes
    -----
    When ``seed is None`` the integer seed is sampled from torch's global RNG
    via ``torch.randint``, advancing the global state but tying the draws to
    ``torch.manual_seed`` for end-to-end reproducibility.
    """
    if isinstance(seed, torch.Generator):
        return seed
    generator = torch.Generator(device=device if device is not None else "cpu")
    if seed is None:
        seed = int(torch.randint(0, 2**63 - 1, (1,)).item())
    generator.manual_seed(int(seed))
    return generator

coerce_seed_to_int ¶

coerce_seed_to_int(seed: SeedLike) -> int | None

Reduce a torch/int/None seed to a plain int seed (or None).

The NumPy-backed graph builders and named initializers thread an integer (or None) seed down to :func:create_rng. A :class:torch.Generator cannot be passed there directly, so this helper extracts a deterministic integer from its initial seed — two generators created with the same manual_seed therefore yield the same int, keeping the topology a pure function of the generator.

PARAMETER	DESCRIPTION
`seed`	Seed in any of the accepted forms. TYPE: `int, torch.Generator, or None`

RETURNS	DESCRIPTION
`int or None`	`seed` itself if it was an `int`; the generator's `initial_seed()` if it was a :class:`torch.Generator`; `None` if it was `None`.

Source code in src/resdag/utils/general.py

def coerce_seed_to_int(seed: SeedLike) -> int | None:
    """Reduce a torch/int/None seed to a plain ``int`` seed (or ``None``).

    The NumPy-backed graph builders and named initializers thread an integer
    (or ``None``) seed down to :func:`create_rng`.  A :class:`torch.Generator`
    cannot be passed there directly, so this helper extracts a deterministic
    integer from its initial seed — two generators created with the same
    ``manual_seed`` therefore yield the same int, keeping the topology a pure
    function of the generator.

    Parameters
    ----------
    seed : int, torch.Generator, or None
        Seed in any of the accepted forms.

    Returns
    -------
    int or None
        ``seed`` itself if it was an ``int``; the generator's
        ``initial_seed()`` if it was a :class:`torch.Generator`; ``None`` if it
        was ``None``.
    """
    if seed is None or isinstance(seed, int):
        return seed
    if isinstance(seed, torch.Generator):
        # initial_seed() is the value the generator was seeded with; identical
        # generators (same manual_seed) therefore map to the same int.
        return int(seed.initial_seed())
    raise TypeError(
        f"Invalid seed type: {type(seed).__name__}. Expected int, torch.Generator, or None."
    )

data ¶

Data Loading and Preparation (compatibility shim)¶

.. deprecated:: 0.7.0 The canonical home for data preparation and file I/O is now :mod:resdag.data. This module re-exports those symbols unchanged for back-compat, so from resdag.utils.data import prepare_esn_data (etc.) keeps working. New code should import from :mod:resdag.data.

The re-exported callables are the same function objects as their :mod:resdag.data counterparts, so identity checks hold::

resdag.utils.data.prepare_esn_data is resdag.data.prepare_esn_data  # True

The canonical dataset generators (:func:lorenz, :func:rossler, …) still live here and are re-exported by :mod:resdag.datasets.

File I/O Functions

load_file Auto-detect format and load time series data. load_csv, load_npy, load_npz, load_nc Format-specific loaders. save_csv, save_npy, save_npz, save_nc Save data in respective formats. list_files List files matching a pattern.

Data Preparation Functions

prepare_esn_data Split time series into warmup, train, target, f_warmup, val. normalize_data Normalize data using various methods. denormalize_data Invert a normalization, mapping data back to its original scale. load_and_prepare Load and prepare data in one step.

Canonical Dataset Generators

lorenz, rossler Chaotic continuous-time attractors (RK4 integration). henon Chaotic discrete-time map. mackey_glass Chaotic delay-differential-equation series. narma Nonlinear system-identification benchmark (input/output series). sine Simple periodic signal for sanity checks.

Each generator returns a (1, T, D) tensor, is seedable via create_rng, supports a transient/discard period, and supports optional normalization.

Examples:

Loading and preparing data (canonical home is :mod:resdag.data):

>>> from resdag.data import load_file, prepare_esn_data
>>> data = load_file("lorenz.csv")
>>> warmup, train, target, f_warmup, val = prepare_esn_data(
...     data, warmup_steps=100, train_steps=500, val_steps=200, normalize="minmax"
... )

Generating a canonical benchmark series:

>>> from resdag.datasets import lorenz
>>> series = lorenz(2000)  # (1, 2000, 3) — chaotic Lorenz-63 trajectory

See Also

resdag.data : Canonical home for data preparation, file I/O, and streaming windows. resdag.datasets : Canonical benchmark time-series generators.

load_file ¶

load_file(path: PathLike, dtype: dtype | None = None, **kwargs: Any) -> Tensor

Load time series data from a file, detecting format from extension.

PARAMETER	DESCRIPTION
`path`	Path to the data file. Supported extensions: .csv, .npy, .npz, .nc TYPE: `str or Path`
`dtype`	Desired tensor dtype. If None, uses `torch.get_default_dtype()`. TYPE: `dtype` DEFAULT: `None`
`**kwargs`	Additional arguments passed to the specific loader (e.g., key for .npz). TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`Tensor`	Data tensor with shape (B, T, D).

RAISES	DESCRIPTION
`ValueError`	If the file extension is not supported.

Source code in src/resdag/data/io.py

def load_file(path: PathLike, dtype: torch.dtype | None = None, **kwargs: Any) -> torch.Tensor:
    """Load time series data from a file, detecting format from extension.

    Parameters
    ----------
    path : str or Path
        Path to the data file. Supported extensions: .csv, .npy, .npz, .nc
    dtype : torch.dtype, optional
        Desired tensor dtype. If None, uses ``torch.get_default_dtype()``.
    **kwargs
        Additional arguments passed to the specific loader (e.g., key for .npz).

    Returns
    -------
    torch.Tensor
        Data tensor with shape (B, T, D).

    Raises
    ------
    ValueError
        If the file extension is not supported.
    """
    path = Path(path)
    suffix = path.suffix.lower()

    loaders: dict[str, Callable[..., torch.Tensor]] = {
        ".csv": load_csv,
        ".npy": load_npy,
        ".npz": load_npz,
        ".nc": load_nc,
    }

    if suffix not in loaders:
        raise ValueError(
            f"Unsupported file extension '{suffix}' for '{path}'. Supported: {list(loaders.keys())}"
        )

    return loaders[suffix](path, dtype=dtype, **kwargs)

load_csv ¶

load_csv(path: PathLike, delimiter: str = ',', dtype: dtype | None = None) -> Tensor

Load time series data from a CSV file.

PARAMETER	DESCRIPTION
`path`	Path to the CSV file. TYPE: `str or Path`
`delimiter`	Field delimiter. TYPE: `str` DEFAULT: `","`
`dtype`	Desired tensor dtype. If None, uses `torch.get_default_dtype()`. TYPE: `dtype` DEFAULT: `None`

RETURNS	DESCRIPTION
`Tensor`	Data tensor with shape (1, T, D).

Notes

Expects headerless CSV with numeric values only.

The file is read with ndmin=2 so the on-disk grid maps directly to (T, D) and the (timesteps, features) convention is preserved without ambiguity:

a single-row, N-column file -> (1, N) -> (1, 1, N) (one timestep of N features);
a single-column, T-row file -> (T, 1) -> (1, T, 1) (T timesteps of one feature).

Without ndmin=2 np.loadtxt collapses a single row to a 1D array, which would silently transpose the time and feature axes.

Source code in src/resdag/data/io.py

def load_csv(
    path: PathLike, delimiter: str = ",", dtype: torch.dtype | None = None
) -> torch.Tensor:
    """Load time series data from a CSV file.

    Parameters
    ----------
    path : str or Path
        Path to the CSV file.
    delimiter : str, default=","
        Field delimiter.
    dtype : torch.dtype, optional
        Desired tensor dtype. If None, uses ``torch.get_default_dtype()``.

    Returns
    -------
    torch.Tensor
        Data tensor with shape (1, T, D).

    Notes
    -----
    Expects headerless CSV with numeric values only.

    The file is read with ``ndmin=2`` so the on-disk grid maps directly to
    ``(T, D)`` and the ``(timesteps, features)`` convention is preserved without
    ambiguity:

    - a single-row, ``N``-column file -> ``(1, N)`` -> ``(1, 1, N)``
      (one timestep of ``N`` features);
    - a single-column, ``T``-row file -> ``(T, 1)`` -> ``(1, T, 1)``
      (``T`` timesteps of one feature).

    Without ``ndmin=2`` ``np.loadtxt`` collapses a single row to a 1D array,
    which would silently transpose the time and feature axes.
    """
    dtype = dtype or torch.get_default_dtype()
    np_dtype = _torch_to_numpy_dtype(dtype)
    data = np.loadtxt(path, delimiter=delimiter, dtype=np_dtype, ndmin=2)
    data = _ensure_3d(data, f"CSV file '{path}'")
    return torch.from_numpy(data)

load_npy ¶

load_npy(path: PathLike, dtype: dtype | None = None) -> Tensor

Load time series data from a NumPy .npy file.

PARAMETER	DESCRIPTION
`path`	Path to the .npy file. TYPE: `str or Path`
`dtype`	Desired tensor dtype. If None, uses `torch.get_default_dtype()`. TYPE: `dtype` DEFAULT: `None`

RETURNS	DESCRIPTION
`Tensor`	Data tensor with shape (B, T, D).

WARNS	DESCRIPTION
`UserWarning`	If the stored array is 1D `(T,)`. It is interpreted as a univariate series and reshaped to `(1, T, 1)`; store data as 2D `(T, D)` to make the shape unambiguous and suppress the warning.

Source code in src/resdag/data/io.py

def load_npy(path: PathLike, dtype: torch.dtype | None = None) -> torch.Tensor:
    """Load time series data from a NumPy .npy file.

    Parameters
    ----------
    path : str or Path
        Path to the .npy file.
    dtype : torch.dtype, optional
        Desired tensor dtype. If None, uses ``torch.get_default_dtype()``.

    Returns
    -------
    torch.Tensor
        Data tensor with shape (B, T, D).

    Warns
    -----
    UserWarning
        If the stored array is 1D ``(T,)``. It is interpreted as a univariate
        series and reshaped to ``(1, T, 1)``; store data as 2D ``(T, D)`` to
        make the shape unambiguous and suppress the warning.
    """
    dtype = dtype or torch.get_default_dtype()
    np_dtype = _torch_to_numpy_dtype(dtype)
    data = np.load(path).astype(np_dtype)
    _warn_if_1d(data, f"NPY file '{path}'")
    data = _ensure_3d(data, f"NPY file '{path}'")
    return torch.from_numpy(data)

load_npz ¶

load_npz(path: PathLike, key: str = 'data', dtype: dtype | None = None) -> Tensor

Load time series data from a NumPy .npz file.

PARAMETER	DESCRIPTION
`path`	Path to the .npz file. TYPE: `str or Path`
`key`	Key to access the data array in the .npz file. TYPE: `str` DEFAULT: `"data"`
`dtype`	Desired tensor dtype. If None, uses `torch.get_default_dtype()`. TYPE: `dtype` DEFAULT: `None`

RETURNS	DESCRIPTION
`Tensor`	Data tensor with shape (B, T, D).

RAISES	DESCRIPTION
`KeyError`	If the specified key is not found in the file.

WARNS	DESCRIPTION
`UserWarning`	If the stored array is 1D `(T,)`. It is interpreted as a univariate series and reshaped to `(1, T, 1)`; store data as 2D `(T, D)` to make the shape unambiguous and suppress the warning.

Source code in src/resdag/data/io.py

def load_npz(path: PathLike, key: str = "data", dtype: torch.dtype | None = None) -> torch.Tensor:
    """Load time series data from a NumPy .npz file.

    Parameters
    ----------
    path : str or Path
        Path to the .npz file.
    key : str, default="data"
        Key to access the data array in the .npz file.
    dtype : torch.dtype, optional
        Desired tensor dtype. If None, uses ``torch.get_default_dtype()``.

    Returns
    -------
    torch.Tensor
        Data tensor with shape (B, T, D).

    Raises
    ------
    KeyError
        If the specified key is not found in the file.

    Warns
    -----
    UserWarning
        If the stored array is 1D ``(T,)``. It is interpreted as a univariate
        series and reshaped to ``(1, T, 1)``; store data as 2D ``(T, D)`` to
        make the shape unambiguous and suppress the warning.
    """
    dtype = dtype or torch.get_default_dtype()
    np_dtype = _torch_to_numpy_dtype(dtype)
    data_dict = np.load(path)
    if key not in data_dict:
        available = list(data_dict.keys())
        raise KeyError(f"Key '{key}' not found in '{path}'. Available keys: {available}")

    data = data_dict[key].astype(np_dtype)
    _warn_if_1d(data, f"NPZ file '{path}' (key '{key}')")
    data = _ensure_3d(data, f"NPZ file '{path}'")
    return torch.from_numpy(data)

load_nc ¶

load_nc(path: PathLike, dtype: dtype | None = None) -> Tensor

Load time series data from a NetCDF (.nc) file.

PARAMETER	DESCRIPTION
`path`	Path to the .nc file. TYPE: `str or Path`
`dtype`	Desired tensor dtype. If None, uses `torch.get_default_dtype()`. TYPE: `dtype` DEFAULT: `None`

RETURNS	DESCRIPTION
`Tensor`	Data tensor with shape (B, T, D).

WARNS	DESCRIPTION
`UserWarning`	If the stored array is 1D `(T,)`. It is interpreted as a univariate series and reshaped to `(1, T, 1)`; store data as 2D `(T, D)` to make the shape unambiguous and suppress the warning.

Notes

Requires xarray to be installed.

Source code in src/resdag/data/io.py

def load_nc(path: PathLike, dtype: torch.dtype | None = None) -> torch.Tensor:
    """Load time series data from a NetCDF (.nc) file.

    Parameters
    ----------
    path : str or Path
        Path to the .nc file.
    dtype : torch.dtype, optional
        Desired tensor dtype. If None, uses ``torch.get_default_dtype()``.

    Returns
    -------
    torch.Tensor
        Data tensor with shape (B, T, D).

    Warns
    -----
    UserWarning
        If the stored array is 1D ``(T,)``. It is interpreted as a univariate
        series and reshaped to ``(1, T, 1)``; store data as 2D ``(T, D)`` to
        make the shape unambiguous and suppress the warning.

    Notes
    -----
    Requires xarray to be installed.
    """
    try:
        import xarray as xr
    except ImportError as e:
        raise ImportError(
            "xarray is required to load NetCDF files. Install with: pip install xarray netcdf4"
        ) from e

    dtype = dtype or torch.get_default_dtype()
    np_dtype = _torch_to_numpy_dtype(dtype)
    data = xr.open_dataarray(path).to_numpy().astype(np_dtype)
    _warn_if_1d(data, f"NetCDF file '{path}'")
    data = _ensure_3d(data, f"NetCDF file '{path}'")
    return torch.from_numpy(data)

save_csv ¶

save_csv(data: Tensor, path: PathLike, delimiter: str = ',') -> None

Save data to a CSV file.

PARAMETER	DESCRIPTION
`data`	Data to save. Will be squeezed to 2D (T, D) if 3D with B=1. TYPE: `Tensor`
`path`	Path to save the CSV file. TYPE: `str or Path`
`delimiter`	Field delimiter. TYPE: `str` DEFAULT: `","`

RAISES	DESCRIPTION
`ValueError`	If data cannot be represented as 2D.

Source code in src/resdag/data/io.py

def save_csv(data: torch.Tensor, path: PathLike, delimiter: str = ",") -> None:
    """Save data to a CSV file.

    Parameters
    ----------
    data : torch.Tensor
        Data to save. Will be squeezed to 2D (T, D) if 3D with B=1.
    path : str or Path
        Path to save the CSV file.
    delimiter : str, default=","
        Field delimiter.

    Raises
    ------
    ValueError
        If data cannot be represented as 2D.
    """
    arr = data.detach().cpu().numpy()

    # Squeeze batch dim if B=1
    if arr.ndim == 3 and arr.shape[0] == 1:
        arr = arr.squeeze(0)

    if arr.ndim != 2:
        raise ValueError(f"CSV saving requires 2D data, got shape {arr.shape}")

    np.savetxt(path, arr, delimiter=delimiter)

save_npy ¶

save_npy(data: Tensor, path: PathLike) -> None

Save data to a NumPy .npy file.

PARAMETER	DESCRIPTION
`data`	Data to save. TYPE: `Tensor`
`path`	Path to save the .npy file. TYPE: `str or Path`

Source code in src/resdag/data/io.py

def save_npy(data: torch.Tensor, path: PathLike) -> None:
    """Save data to a NumPy .npy file.

    Parameters
    ----------
    data : torch.Tensor
        Data to save.
    path : str or Path
        Path to save the .npy file.
    """
    np.save(path, data.detach().cpu().numpy())

save_npz ¶

save_npz(data: Tensor, path: PathLike, key: str = 'data') -> None

Save data to a NumPy .npz file.

PARAMETER	DESCRIPTION
`data`	Data to save. TYPE: `Tensor`
`path`	Path to save the .npz file. TYPE: `str or Path`
`key`	Key to use when storing the data. TYPE: `str` DEFAULT: `"data"`

Source code in src/resdag/data/io.py

def save_npz(data: torch.Tensor, path: PathLike, key: str = "data") -> None:
    """Save data to a NumPy .npz file.

    Parameters
    ----------
    data : torch.Tensor
        Data to save.
    path : str or Path
        Path to save the .npz file.
    key : str, default="data"
        Key to use when storing the data.
    """
    # ``key`` is dynamic, so the array is passed via ``**``. numpy 2.4's typed
    # stub for ``savez`` added an ``allow_pickle`` keyword that a str-keyed ``**``
    # mapping cannot satisfy; bind through a generically-typed reference so the
    # call type-checks identically on every supported numpy (>=2.0) without
    # changing the runtime call.
    savez: Callable[..., None] = np.savez
    savez(path, **{key: data.detach().cpu().numpy()})

save_nc ¶

save_nc(data: Tensor, path: PathLike) -> None

Save data to a NetCDF (.nc) file.

PARAMETER	DESCRIPTION
`data`	Data to save. TYPE: `Tensor`
`path`	Path to save the .nc file. TYPE: `str or Path`

Notes

Requires xarray to be installed.

Source code in src/resdag/data/io.py

def save_nc(data: torch.Tensor, path: PathLike) -> None:
    """Save data to a NetCDF (.nc) file.

    Parameters
    ----------
    data : torch.Tensor
        Data to save.
    path : str or Path
        Path to save the .nc file.

    Notes
    -----
    Requires xarray to be installed.
    """
    try:
        import xarray as xr
    except ImportError as e:
        raise ImportError(
            "xarray is required to save NetCDF files. Install with: pip install xarray netcdf4"
        ) from e

    arr = data.detach().cpu().numpy()
    xr.DataArray(arr).to_netcdf(path)

list_files ¶

list_files(directory: PathLike, extensions: list[str] | None = None) -> list[Path]

List files in a directory, optionally filtering by extension.

PARAMETER	DESCRIPTION
`directory`	Path to the directory. TYPE: `str or Path`
`extensions`	List of extensions to filter (e.g., [".csv", ".npy"]). If None, returns all files. TYPE: `list of str` DEFAULT: `None`

RETURNS	DESCRIPTION
`list[Path]`	List of file paths (excluding directories).

Source code in src/resdag/data/io.py

def list_files(directory: PathLike, extensions: list[str] | None = None) -> list[Path]:
    """List files in a directory, optionally filtering by extension.

    Parameters
    ----------
    directory : str or Path
        Path to the directory.
    extensions : list of str, optional
        List of extensions to filter (e.g., [".csv", ".npy"]).
        If None, returns all files.

    Returns
    -------
    list[Path]
        List of file paths (excluding directories).
    """
    directory = Path(directory)
    files = [f for f in directory.iterdir() if f.is_file()]

    if extensions:
        # Normalize extensions to lowercase with leading dot
        exts = {e.lower() if e.startswith(".") else f".{e.lower()}" for e in extensions}
        files = [f for f in files if f.suffix.lower() in exts]

    return sorted(files)

prepare_esn_data ¶

prepare_esn_data(data: Tensor, warmup_steps: int, train_steps: int, val_steps: int | None = None, discard_steps: int = 0, normalize: bool = False, norm_method: NormMethod = 'minmax', return_stats: bool = False) -> ESNDataSplits | ESNDataSplitsWithStats

Prepare time series data for ESN training and forecasting.

Splits data into segments appropriate for ESN workflows: 1. Warmup: Initial steps for reservoir state synchronization 2. Train: Training input data 3. Target: Training targets (train data shifted by 1 step) 4. Forecast warmup: Last warmup_steps of training for forecast initialization 5. Validation: Held-out data for testing, starting one step after the train window so it aligns with the (purely autoregressive) forecast.

Data layout (train_end = warmup_steps + train_steps)::

[discard][warmup][------- train -------][s][------ val ------]

s is data[train_end]: the autoregressive seam — the final training target and the value the forecast warmup's last output predicts (the seed the forecast consumes). It is intentionally excluded from val, so forecast(forecast_warmup, horizon=val_steps) lines up element-for-element with val (no manual shift).

PARAMETER	DESCRIPTION
`data`	Input time series of shape (B, T, D). TYPE: `Tensor`
`warmup_steps`	Number of steps for reservoir warmup/synchronization. TYPE: `int`
`train_steps`	Number of training steps (after warmup). TYPE: `int`
`val_steps`	Number of validation steps. If None, uses all remaining data. TYPE: `int` DEFAULT: `None`
`discard_steps`	Number of initial steps to discard (e.g., initial transients). TYPE: `int` DEFAULT: `0`
`normalize`	Whether to normalize data. If True, statistics are computed from training data and applied to all splits globally. TYPE: `bool` DEFAULT: `False`
`norm_method`	Normalization method if normalize=True. TYPE: `str` DEFAULT: `"minmax"`
`return_stats`	If True, append the fitted normalization statistics as a 6th return element so the normalize -> forecast -> report loop can be closed via :func:`denormalize_data`. The element is the `stats` dict when `normalize=True` and `None` when `normalize=False` (nothing was fitted). When False (the default), the legacy 5-tuple is returned unchanged. TYPE: `bool` DEFAULT: `False`

RETURNS	DESCRIPTION
`warmup`	Warmup data, shape (B, warmup_steps, D). TYPE: `Tensor`
`train`	Training input, shape (B, train_steps, D). TYPE: `Tensor`
`target`	Training target (shifted by 1), shape (B, train_steps, D). TYPE: `Tensor`
`forecast_warmup`	Last warmup_steps of training for forecast init, shape (B, warmup_steps, D). TYPE: `Tensor`
`val`	Validation data, shape (B, val_steps, D). Starts at `data[train_end + 1]` so it aligns directly with an autoregressive `forecast` of the same horizon (the `data[train_end]` seam is the forecast's seed, not a target). TYPE: `Tensor`
`stats`	Only returned when `return_stats=True`. The fitted normalization statistics (as returned by :func:`normalize_data`) when `normalize=True`, otherwise `None`. Pass to :func:`denormalize_data` to map forecasts back to the original scale. TYPE: `dict or None`

RAISES	DESCRIPTION
`ValueError`	If `warmup_steps` or `train_steps` is not strictly positive, if `discard_steps` is negative, if `val_steps` is negative, or if the data is too short for the requested splits.

WARNS	DESCRIPTION
`UserWarning`	If the resulting validation window is empty (`val_steps=0` or the data is exhausted by warmup/train/seam), meaning no held-out validation data remains.

See Also

denormalize_data : Inverts the normalization using the returned stats.

Examples:

>>> data = torch.randn(1, 1000, 3)  # (batch=1, time=1000, features=3)
>>> warmup, train, target, f_warmup, val = prepare_esn_data(
...     data, warmup_steps=100, train_steps=500, val_steps=200
... )
>>> print(warmup.shape)   # (1, 100, 3)
>>> print(train.shape)    # (1, 500, 3)
>>> print(target.shape)   # (1, 500, 3)
>>> print(f_warmup.shape) # (1, 100, 3)
>>> print(val.shape)      # (1, 200, 3)

Close the round-trip by keeping the fitted stats and inverting later:

>>> warmup, train, target, f_warmup, val, stats = prepare_esn_data(
...     data, warmup_steps=100, train_steps=500, val_steps=200,
...     normalize=True, norm_method="minmax", return_stats=True,
... )
>>> val_physical = denormalize_data(val, "minmax", stats)  # original units

Source code in src/resdag/data/prepare.py

def prepare_esn_data(
    data: torch.Tensor,
    warmup_steps: int,
    train_steps: int,
    val_steps: int | None = None,
    discard_steps: int = 0,
    normalize: bool = False,
    norm_method: NormMethod = "minmax",
    return_stats: bool = False,
) -> ESNDataSplits | ESNDataSplitsWithStats:
    """Prepare time series data for ESN training and forecasting.

    Splits data into segments appropriate for ESN workflows:
    1. Warmup: Initial steps for reservoir state synchronization
    2. Train: Training input data
    3. Target: Training targets (train data shifted by 1 step)
    4. Forecast warmup: Last warmup_steps of training for forecast initialization
    5. Validation: Held-out data for testing, starting one step after the train
       window so it aligns with the (purely autoregressive) forecast.

    Data layout (``train_end = warmup_steps + train_steps``)::

        [discard][warmup][------- train -------][s][------ val ------]

    ``s`` is ``data[train_end]``: the autoregressive seam — the final training
    target and the value the forecast warmup's last output predicts (the seed
    the forecast consumes). It is intentionally excluded from ``val``, so
    ``forecast(forecast_warmup, horizon=val_steps)`` lines up element-for-element
    with ``val`` (no manual shift).


    Parameters
    ----------
    data : torch.Tensor
        Input time series of shape (B, T, D).
    warmup_steps : int
        Number of steps for reservoir warmup/synchronization.
    train_steps : int
        Number of training steps (after warmup).
    val_steps : int, optional
        Number of validation steps. If None, uses all remaining data.
    discard_steps : int, default=0
        Number of initial steps to discard (e.g., initial transients).
    normalize : bool, default=False
        Whether to normalize data. If True, statistics are computed from
        training data and applied to all splits globally.
    norm_method : str, default="minmax"
        Normalization method if normalize=True.
    return_stats : bool, default=False
        If True, append the fitted normalization statistics as a 6th return
        element so the normalize -> forecast -> report loop can be closed via
        :func:`denormalize_data`. The element is the ``stats`` dict when
        ``normalize=True`` and ``None`` when ``normalize=False`` (nothing was
        fitted). When False (the default), the legacy 5-tuple is returned
        unchanged.

    Returns
    -------
    warmup : torch.Tensor
        Warmup data, shape (B, warmup_steps, D).
    train : torch.Tensor
        Training input, shape (B, train_steps, D).
    target : torch.Tensor
        Training target (shifted by 1), shape (B, train_steps, D).
    forecast_warmup : torch.Tensor
        Last warmup_steps of training for forecast init, shape (B, warmup_steps, D).
    val : torch.Tensor
        Validation data, shape (B, val_steps, D). Starts at ``data[train_end + 1]``
        so it aligns directly with an autoregressive ``forecast`` of the same
        horizon (the ``data[train_end]`` seam is the forecast's seed, not a target).
    stats : dict or None
        Only returned when ``return_stats=True``. The fitted normalization
        statistics (as returned by :func:`normalize_data`) when ``normalize=True``,
        otherwise ``None``. Pass to :func:`denormalize_data` to map forecasts
        back to the original scale.

    Raises
    ------
    ValueError
        If ``warmup_steps`` or ``train_steps`` is not strictly positive, if
        ``discard_steps`` is negative, if ``val_steps`` is negative, or if the
        data is too short for the requested splits.

    Warns
    -----
    UserWarning
        If the resulting validation window is empty (``val_steps=0`` or the data
        is exhausted by warmup/train/seam), meaning no held-out validation data
        remains.

    See Also
    --------
    denormalize_data : Inverts the normalization using the returned ``stats``.

    Examples
    --------
    >>> data = torch.randn(1, 1000, 3)  # (batch=1, time=1000, features=3)
    >>> warmup, train, target, f_warmup, val = prepare_esn_data(
    ...     data, warmup_steps=100, train_steps=500, val_steps=200
    ... )
    >>> print(warmup.shape)   # (1, 100, 3)
    >>> print(train.shape)    # (1, 500, 3)
    >>> print(target.shape)   # (1, 500, 3)
    >>> print(f_warmup.shape) # (1, 100, 3)
    >>> print(val.shape)      # (1, 200, 3)

    Close the round-trip by keeping the fitted stats and inverting later:

    >>> warmup, train, target, f_warmup, val, stats = prepare_esn_data(
    ...     data, warmup_steps=100, train_steps=500, val_steps=200,
    ...     normalize=True, norm_method="minmax", return_stats=True,
    ... )
    >>> val_physical = denormalize_data(val, "minmax", stats)  # original units
    """
    _, timesteps, _ = data.shape

    # Validate step counts before any negative slicing can silently produce
    # plausible-but-wrong / empty splits.
    if warmup_steps <= 0:
        raise ValueError(f"warmup_steps must be a positive integer, got {warmup_steps}")
    if train_steps <= 0:
        raise ValueError(f"train_steps must be a positive integer, got {train_steps}")
    if discard_steps < 0:
        raise ValueError(f"discard_steps must be non-negative, got {discard_steps}")
    if val_steps is not None and val_steps < 0:
        raise ValueError(f"val_steps must be non-negative or None, got {val_steps}")

    # Validate discard_steps
    if discard_steps >= timesteps:
        raise ValueError(
            f"discard_steps ({discard_steps}) must be less than data length ({timesteps})"
        )

    # Trim initial steps
    data = data[:, discard_steps:, :]
    timesteps = data.shape[1]

    # Calculate required length
    train_end = warmup_steps + train_steps

    if train_end >= timesteps:
        raise ValueError(
            f"warmup_steps + train_steps ({train_end}) exceeds "
            f"available data length ({timesteps}) after discarding {discard_steps} steps"
        )

    # Determine validation length. The autoregressive forecast consumes
    # ``data[train_end]`` as its bootstrap seed (the value the forecast warmup's
    # final output predicts) and only emits predictions from ``data[train_end + 1]``
    # onward, so the validation window starts one step after ``train_end``.
    val_start = train_end + 1
    if val_steps is None:
        val_steps = timesteps - val_start
        if val_steps == 0:
            warnings.warn(
                "Validation window is empty: warmup + train + the autoregressive "
                f"seam consume all {timesteps} available steps (after discarding "
                f"{discard_steps}), leaving no held-out data for val. Reduce "
                "warmup_steps/train_steps or supply more data if you need a val split.",
                UserWarning,
                stacklevel=2,
            )
    else:
        required = val_start + val_steps
        if required > timesteps:
            raise ValueError(
                f"Required data ({required} = warmup + train + 1 seam + val) "
                f"exceeds available length ({timesteps}). The +1 is the "
                f"autoregressive seam step (data[train_end]) the forecast "
                f"consumes as its seed before producing val_steps predictions."
            )
        if val_steps == 0:
            warnings.warn(
                "val_steps=0 produces an empty validation split. Pass a positive "
                "val_steps (or None to use all remaining data) if you need a val "
                "window.",
                UserWarning,
                stacklevel=2,
            )

    # Split data
    warmup = data[:, :warmup_steps, :]
    train = data[:, warmup_steps:train_end, :]
    target = data[:, warmup_steps + 1 : train_end + 1, :]
    forecast_warmup = train[:, -warmup_steps:, :]
    val = data[:, val_start : val_start + val_steps, :]

    # Optional normalization (compute stats from training data only)
    stats: dict[str, torch.Tensor] | None = None
    if normalize:
        stats = _compute_stats(train, norm_method)
        warmup = _apply_norm(warmup, norm_method, stats)
        train = _apply_norm(train, norm_method, stats)
        target = _apply_norm(target, norm_method, stats)
        forecast_warmup = _apply_norm(forecast_warmup, norm_method, stats)
        val = _apply_norm(val, norm_method, stats)

    if return_stats:
        return warmup, train, target, forecast_warmup, val, stats
    return warmup, train, target, forecast_warmup, val

normalize_data ¶

normalize_data(data: Tensor, method: NormMethod = 'minmax', stats: dict[str, Tensor] | None = None) -> tuple[Tensor, dict[str, Tensor]]

Normalize time series data globally.

Computes statistics across all batches and timesteps, applying the same normalization to the entire dataset. This is the correct approach when batches contain trajectories from the same dynamical system.

PARAMETER	DESCRIPTION
`data`	Data tensor of shape (B, T, D). TYPE: `Tensor`
`method`	Normalization method: - "minmax": Scale to [-1, 1] range - "standard": Zero mean, unit variance - "noncentered": Scale by max absolute value (preserves zero) - "meanpreserving": Scale deviations to [-1, 1], then restore mean TYPE: `(minmax, standard, noncentered, meanpreserving)` DEFAULT: `"minmax"`
`stats`	Pre-computed statistics for normalization. If provided, these are used instead of computing from data. TYPE: `dict` DEFAULT: `None`

RETURNS	DESCRIPTION
`normalized`	Normalized data with same shape as input. TYPE: `Tensor`
`stats`	Statistics used for normalization (for applying to other data). TYPE: `dict`

Examples:

>>> data = torch.randn(1, 100, 3)
>>> normalized, stats = normalize_data(data, method="minmax")
>>> # Apply same normalization to new data
>>> new_normalized, _ = normalize_data(new_data, method="minmax", stats=stats)

Source code in src/resdag/data/prepare.py

def normalize_data(
    data: torch.Tensor,
    method: NormMethod = "minmax",
    stats: dict[str, torch.Tensor] | None = None,
) -> tuple[torch.Tensor, dict[str, torch.Tensor]]:
    """Normalize time series data globally.

    Computes statistics across all batches and timesteps, applying the same
    normalization to the entire dataset. This is the correct approach when
    batches contain trajectories from the same dynamical system.

    Parameters
    ----------
    data : torch.Tensor
        Data tensor of shape (B, T, D).
    method : {"minmax", "standard", "noncentered", "meanpreserving"}
        Normalization method:
        - "minmax": Scale to [-1, 1] range
        - "standard": Zero mean, unit variance
        - "noncentered": Scale by max absolute value (preserves zero)
        - "meanpreserving": Scale deviations to [-1, 1], then restore mean
    stats : dict, optional
        Pre-computed statistics for normalization. If provided, these are used
        instead of computing from data.

    Returns
    -------
    normalized : torch.Tensor
        Normalized data with same shape as input.
    stats : dict
        Statistics used for normalization (for applying to other data).

    Examples
    --------
    >>> data = torch.randn(1, 100, 3)
    >>> normalized, stats = normalize_data(data, method="minmax")
    >>> # Apply same normalization to new data
    >>> new_normalized, _ = normalize_data(new_data, method="minmax", stats=stats)
    """
    if stats is None:
        stats = _compute_stats(data, method)

    normalized = _apply_norm(data, method, stats)
    return normalized, stats

load_and_prepare ¶

load_and_prepare(paths: str | list[str], warmup_steps: int, train_steps: int, val_steps: int | None = None, discard_steps: int = 0, normalize: bool = False, norm_method: NormMethod = 'minmax', return_stats: bool = False, dtype: dtype | None = None, **load_kwargs: Any) -> ESNDataSplits | ESNDataSplitsWithStats

Load data from file(s) and prepare for ESN training.

Convenience function that combines loading and preparation. If multiple paths are provided, data is concatenated along the batch dimension.

PARAMETER	DESCRIPTION
`paths`	Path(s) to data file(s). Supports .csv, .npy, .npz, .nc TYPE: `str or list of str`
`warmup_steps`	Number of warmup steps. TYPE: `int`
`train_steps`	Number of training steps. TYPE: `int`
`val_steps`	Number of validation steps. TYPE: `int` DEFAULT: `None`
`discard_steps`	Initial steps to discard. TYPE: `int` DEFAULT: `0`
`normalize`	Whether to normalize data. TYPE: `bool` DEFAULT: `False`
`norm_method`	Normalization method. TYPE: `str` DEFAULT: `"minmax"`
`return_stats`	If True, append the fitted normalization statistics as a 6th return element (`None` when `normalize=False`). Forwarded to :func:`prepare_esn_data`; see its docstring for details. TYPE: `bool` DEFAULT: `False`
`dtype`	Desired tensor dtype. If None, uses `torch.get_default_dtype()`. TYPE: `dtype` DEFAULT: `None`
`**load_kwargs`	Additional arguments passed to load_file (e.g., key for .npz). TYPE: `Any` DEFAULT: `{}`

RETURNS	DESCRIPTION
`ESNDataSplits or ESNDataSplitsWithStats`	Tuple of (warmup, train, target, forecast_warmup, val) tensors, with a trailing `stats` element when `return_stats=True`.

See Also

denormalize_data : Inverts the normalization using the returned stats.

Examples:

>>> splits = load_and_prepare(
...     "timeseries.csv",
...     warmup_steps=100,
...     train_steps=500,
...     val_steps=200,
...     normalize=True,
... )
>>> warmup, train, target, f_warmup, val = splits

Source code in src/resdag/data/prepare.py

def load_and_prepare(
    paths: str | list[str],
    warmup_steps: int,
    train_steps: int,
    val_steps: int | None = None,
    discard_steps: int = 0,
    normalize: bool = False,
    norm_method: NormMethod = "minmax",
    return_stats: bool = False,
    dtype: torch.dtype | None = None,
    **load_kwargs: Any,
) -> ESNDataSplits | ESNDataSplitsWithStats:
    """Load data from file(s) and prepare for ESN training.

    Convenience function that combines loading and preparation.
    If multiple paths are provided, data is concatenated along the batch dimension.

    Parameters
    ----------
    paths : str or list of str
        Path(s) to data file(s). Supports .csv, .npy, .npz, .nc
    warmup_steps : int
        Number of warmup steps.
    train_steps : int
        Number of training steps.
    val_steps : int, optional
        Number of validation steps.
    discard_steps : int, default=0
        Initial steps to discard.
    normalize : bool, default=False
        Whether to normalize data.
    norm_method : str, default="minmax"
        Normalization method.
    return_stats : bool, default=False
        If True, append the fitted normalization statistics as a 6th return
        element (``None`` when ``normalize=False``). Forwarded to
        :func:`prepare_esn_data`; see its docstring for details.
    dtype : torch.dtype, optional
        Desired tensor dtype. If None, uses ``torch.get_default_dtype()``.
    **load_kwargs
        Additional arguments passed to load_file (e.g., key for .npz).

    Returns
    -------
    ESNDataSplits or ESNDataSplitsWithStats
        Tuple of (warmup, train, target, forecast_warmup, val) tensors, with a
        trailing ``stats`` element when ``return_stats=True``.

    See Also
    --------
    denormalize_data : Inverts the normalization using the returned ``stats``.

    Examples
    --------
    >>> splits = load_and_prepare(
    ...     "timeseries.csv",
    ...     warmup_steps=100,
    ...     train_steps=500,
    ...     val_steps=200,
    ...     normalize=True,
    ... )
    >>> warmup, train, target, f_warmup, val = splits
    """
    from .io import load_file

    # Handle single path or list of paths
    if isinstance(paths, (str, type(None))):
        paths = [paths] if paths else []

    if not paths:
        raise ValueError("At least one data path must be provided")

    # Load and concatenate along batch dimension
    tensors = [load_file(p, dtype=dtype, **load_kwargs) for p in paths]
    data = torch.cat(tensors, dim=0) if len(tensors) > 1 else tensors[0]

    return prepare_esn_data(
        data,
        warmup_steps=warmup_steps,
        train_steps=train_steps,
        val_steps=val_steps,
        discard_steps=discard_steps,
        normalize=normalize,
        norm_method=norm_method,
        return_stats=return_stats,
    )

states ¶

State management and analysis utilities for reservoir layers.

esp_index ¶

esp_index(model: 'ESNModel', feedback_seq: Tensor, *driving_seqs: Tensor, history: bool = False, iterations: int = 10, transient: int = 0, window: int | None = None, relative: bool = False, chunk_size: int | None = None, verbose: bool = True) -> Union[dict[str, list[Tensor]], Tuple[dict[str, list[Tensor]], dict[str, list[Tensor]]]]

Compute the Echo State Property (ESP) index for reservoir layers.

The ESP index measures how quickly trajectories from different initial states converge when driven by the same input. All :class:~resdag.layers.reservoirs.base_reservoir.BaseReservoirLayer reservoirs are discovered automatically, so the diagnostic applies equally to ESN, NG-RC, and any future custom-cell reservoir.

Random restarts are drawn through each reservoir's public :meth:~resdag.layers.reservoirs.base_reservoir.BaseReservoirLayer.set_random_state, which samples a standard-normal state — the same convention as :meth:~resdag.core.ESNModel.set_random_reservoir_states. The index is therefore directly comparable to states produced elsewhere in the library.

PARAMETER	DESCRIPTION
`model`	Model containing reservoir layers. TYPE: `ESNModel`
`feedback_seq`	Feedback sequence, shape `(batch, timesteps, features)`. TYPE: `Tensor`
`*driving_seqs`	Optional driving sequences in model input order. TYPE: `Tensor` DEFAULT: `()`
`history`	If True, return full distance history over time. TYPE: `bool` DEFAULT: `False`
`iterations`	Number of random initial states to average over. TYPE: `int` DEFAULT: `10`
`transient`	Timesteps to discard from sequence start. TYPE: `int` DEFAULT: `0`
`window`	Length of the trailing window over which the index is averaged. By default (`None`) the index is the mean over every post-transient timestep, which for a healthy reservoir is dominated by the early convergence transient rather than the asymptotic distance. Setting `window=k` averages only the last `k` post-transient steps, so the index reflects the asymptotic regime — a far more diagnostic statistic. Must be a positive integer no larger than the number of post-transient timesteps. Does not affect the returned `history` tensors, which always span the full post-transient sequence. TYPE: `int or None` DEFAULT: `None`
`relative`	If True, normalise each per-step distance by the corresponding base-state norm, `\|\|base - rand\|\| / (\|\|base\|\| + eps)`, yielding a scale-free relative distance that is comparable across reservoir sizes and spectral radii. If False (default), the raw absolute distance is used. The normalisation is applied to both the returned index and the `history` tensors. TYPE: `bool` DEFAULT: `False`
`chunk_size`	Maximum number of random restarts run together in a single batched forward pass. All restarts are statistically independent and differ only in their initial state, so they are folded into the batch dimension and evaluated in one forward pass instead of `iterations` sequential passes (typically a ~3x speed-up). A large `reservoir_size * iterations` product can exhaust device memory, however; set `chunk_size` to cap the per-pass batch at `batch * chunk_size` and process the restarts in chunks. `None` (default) runs all `iterations` restarts in one pass, transparently falling back to chunking (halving the chunk each time) if a CUDA out-of-memory error is raised. TYPE: `int or None` DEFAULT: `None`
`verbose`	Print progress and skip messages. TYPE: `bool` DEFAULT: `True`

RETURNS	DESCRIPTION
`dict or tuple`	If `history=False`: dict mapping layer names to single-element lists containing the ESP index scalar tensor. If `history=True`: tuple of (ESP indices dict, history dict). History tensors have shape `(iterations, timesteps, batch)` and span the full post-transient sequence regardless of `window`. Reservoirs whose ESP is undefined (no randomisable state) are omitted from both dictionaries.

RAISES	DESCRIPTION
`ValueError`	If `transient` is not smaller than the number of timesteps, if `window` is not a positive integer no larger than the number of post-transient timesteps, if the model contains no reservoir layers, or if no reservoir has a well-defined ESP.

Notes

For LINEAR systems (identity activation), input does NOT affect ESP because the input contribution cancels out in the state difference. This is mathematically correct behavior.

State is never mutated by direct assignment. The zero base orbit is set via res.reset_state(batch_size=...) and the random restarts via res.set_random_state(), so each cell's own shape contract (2-D for ESN, 3-D delay buffer for NG-RC) is honoured and validated by the public state-management API.

The caller's reservoir states are saved before the computation and restored afterwards (even on error), so calling esp_index never leaves the model's reservoirs in a mutated (random) state.

Batched restarts. The iterations random restarts are statistically independent and differ only in their initial state, so they are folded into the batch dimension: the feedback (and any drivers) are tiled iterations times, one random initial state is composed per restart into a single (batch * iterations, ...) state, and the whole ensemble is evaluated in one forward pass rather than iterations sequential passes. The random restarts are drawn in exactly the same order as the sequential loop (outer over restarts, inner over reservoirs), so the initial states are bit-identical to the sequential computation. The reservoir outputs — and hence the returned index — match the sequential value to floating-point rounding (typically < 1e-6 relative), the only difference being the reassociation of a wider batched matmul; the result is fully deterministic under a fixed seed. (If a reservoir cell injects per-step state noise in training mode, noise > 0, the batched noise draws differ from the sequential ones; the index remains a valid, seed-deterministic stochastic diagnostic but is no longer numerically identical. Run the model in eval mode, the default for a diagnostic, to disable noise entirely.)

Examples:

Score only the asymptotic regime with a scale-free statistic::

idx = esp_index(model, feedback, window=50, relative=True)

See Also

resdag.core.ESNModel.set_random_reservoir_states : Source of the standard-normal restart convention used here.

Source code in src/resdag/utils/states/esp_index.py

def esp_index(
    model: "ESNModel",
    feedback_seq: torch.Tensor,
    *driving_seqs: torch.Tensor,
    history: bool = False,
    iterations: int = 10,
    transient: int = 0,
    window: int | None = None,
    relative: bool = False,
    chunk_size: int | None = None,
    verbose: bool = True,
) -> Union[
    dict[str, list[torch.Tensor]],
    Tuple[dict[str, list[torch.Tensor]], dict[str, list[torch.Tensor]]],
]:
    """
    Compute the Echo State Property (ESP) index for reservoir layers.

    The ESP index measures how quickly trajectories from different initial
    states converge when driven by the same input.  All
    :class:`~resdag.layers.reservoirs.base_reservoir.BaseReservoirLayer`
    reservoirs are discovered automatically, so the diagnostic applies equally
    to ESN, NG-RC, and any future custom-cell reservoir.

    Random restarts are drawn through each reservoir's public
    :meth:`~resdag.layers.reservoirs.base_reservoir.BaseReservoirLayer.set_random_state`,
    which samples a **standard-normal** state — the same convention as
    :meth:`~resdag.core.ESNModel.set_random_reservoir_states`.  The index is
    therefore directly comparable to states produced elsewhere in the library.

    Parameters
    ----------
    model : ESNModel
        Model containing reservoir layers.
    feedback_seq : torch.Tensor
        Feedback sequence, shape ``(batch, timesteps, features)``.
    *driving_seqs : torch.Tensor
        Optional driving sequences in model input order.
    history : bool, default=False
        If True, return full distance history over time.
    iterations : int, default=10
        Number of random initial states to average over.
    transient : int, default=0
        Timesteps to discard from sequence start.
    window : int or None, default=None
        Length of the trailing window over which the index is averaged.  By
        default (``None``) the index is the mean over *every* post-transient
        timestep, which for a healthy reservoir is dominated by the early
        convergence transient rather than the asymptotic distance.  Setting
        ``window=k`` averages only the **last** ``k`` post-transient steps, so
        the index reflects the asymptotic regime — a far more diagnostic
        statistic.  Must be a positive integer no larger than the number of
        post-transient timesteps.  Does not affect the returned ``history``
        tensors, which always span the full post-transient sequence.
    relative : bool, default=False
        If True, normalise each per-step distance by the corresponding
        base-state norm, ``||base - rand|| / (||base|| + eps)``, yielding a
        scale-free relative distance that is comparable across reservoir sizes
        and spectral radii.  If False (default), the raw absolute distance is
        used.  The normalisation is applied to both the returned index and the
        ``history`` tensors.
    chunk_size : int or None, default=None
        Maximum number of random restarts run together in a single batched
        forward pass.  All restarts are statistically independent and differ
        only in their initial state, so they are folded into the batch
        dimension and evaluated in one forward pass instead of ``iterations``
        sequential passes (typically a ~3x speed-up).  A large
        ``reservoir_size * iterations`` product can exhaust device memory,
        however; set ``chunk_size`` to cap the per-pass batch at
        ``batch * chunk_size`` and process the restarts in chunks.  ``None``
        (default) runs all ``iterations`` restarts in one pass, transparently
        falling back to chunking (halving the chunk each time) if a CUDA
        out-of-memory error is raised.
    verbose : bool, default=True
        Print progress and skip messages.

    Returns
    -------
    dict or tuple
        If ``history=False``: dict mapping layer names to single-element
        lists containing the ESP index scalar tensor.
        If ``history=True``: tuple of (ESP indices dict, history dict).
        History tensors have shape ``(iterations, timesteps, batch)`` and span
        the full post-transient sequence regardless of ``window``.

        Reservoirs whose ESP is undefined (no randomisable state) are omitted
        from both dictionaries.

    Raises
    ------
    ValueError
        If ``transient`` is not smaller than the number of timesteps, if
        ``window`` is not a positive integer no larger than the number of
        post-transient timesteps, if the model contains no reservoir layers,
        or if no reservoir has a well-defined ESP.

    Notes
    -----
    For LINEAR systems (identity activation), input does NOT affect ESP
    because the input contribution cancels out in the state difference.
    This is mathematically correct behavior.

    State is never mutated by direct assignment.  The zero base orbit is set
    via ``res.reset_state(batch_size=...)`` and the random restarts via
    ``res.set_random_state()``, so each cell's own shape contract (2-D for
    ESN, 3-D delay buffer for NG-RC) is honoured and validated by the public
    state-management API.

    The caller's reservoir states are saved before the computation and
    restored afterwards (even on error), so calling ``esp_index`` never leaves
    the model's reservoirs in a mutated (random) state.

    **Batched restarts.**  The ``iterations`` random restarts are statistically
    independent and differ only in their initial state, so they are folded into
    the batch dimension: the feedback (and any drivers) are tiled ``iterations``
    times, one random initial state is composed per restart into a single
    ``(batch * iterations, ...)`` state, and the whole ensemble is evaluated in
    **one** forward pass rather than ``iterations`` sequential passes.  The
    random restarts are drawn in exactly the same order as the sequential loop
    (outer over restarts, inner over reservoirs), so the *initial states* are
    bit-identical to the sequential computation.  The reservoir *outputs* — and
    hence the returned index — match the sequential value to floating-point
    rounding (typically ``< 1e-6`` relative), the only difference being the
    reassociation of a wider batched matmul; the result is fully deterministic
    under a fixed seed.  (If a reservoir cell injects per-step state noise in
    training mode, ``noise > 0``, the batched noise draws differ from the
    sequential ones; the index remains a valid, seed-deterministic stochastic
    diagnostic but is no longer numerically identical.  Run the model in eval
    mode, the default for a diagnostic, to disable noise entirely.)

    Examples
    --------
    Score only the asymptotic regime with a scale-free statistic::

        idx = esp_index(model, feedback, window=50, relative=True)

    See Also
    --------
    resdag.core.ESNModel.set_random_reservoir_states :
        Source of the standard-normal restart convention used here.
    """
    from resdag.layers.reservoirs.base_reservoir import BaseReservoirLayer

    device = feedback_seq.device
    dtype = feedback_seq.dtype
    batch_size, total_timesteps, _ = feedback_seq.shape

    if transient >= total_timesteps:
        raise ValueError(f"transient ({transient}) >= timesteps ({total_timesteps})")

    timesteps = total_timesteps - transient

    if window is not None and not (0 < window <= timesteps):
        raise ValueError(
            f"window ({window}) must be a positive integer no larger than the "
            f"number of post-transient timesteps ({timesteps})."
        )

    if chunk_size is not None and chunk_size <= 0:
        raise ValueError(f"chunk_size ({chunk_size}) must be a positive integer or None.")

    inputs = (feedback_seq,) + driving_seqs

    # Save the caller's reservoir states *before* any probing/mutation so they
    # can be restored on the way out (including on error).  ``esp_index`` is a
    # read-only diagnostic; it must not leave the model's reservoirs in the
    # random state it uses internally.  ``_has_well_defined_esp`` and the base
    # orbit below both mutate state, so the snapshot has to be taken here first.
    # ``get_reservoir_states`` omits reservoirs whose state is ``None`` (never
    # warmed up), so those names are tracked separately and reset back to
    # ``None`` on restore instead of being left holding an internal state.
    saved_states = model.get_reservoir_states()
    reservoirs_with_none_state = [
        module
        for module in model.modules()
        if isinstance(module, BaseReservoirLayer) and module.state is None
    ]

    try:
        # Discover every reservoir layer (cell-agnostic): any
        # BaseReservoirLayer, not just ESNLayer.  This includes NGReservoirLayer and
        # any custom-cell layer.
        all_reservoirs: list[tuple[str, "BaseReservoirLayer"]] = []
        for name, module in model.named_modules():
            if isinstance(module, BaseReservoirLayer):
                all_reservoirs.append((name, module))

        if not all_reservoirs:
            raise ValueError("No reservoir layers found in model.")

        # Partition reservoirs into those with a well-defined ESP (a non-empty,
        # randomisable state) and those without.  A reservoir with no per-sample
        # state — e.g. a ``k=1`` NG-RC delay buffer of size 0 — has no notion of
        # "different initial states converging", so its ESP is undefined.
        reservoirs: list[tuple[str, "BaseReservoirLayer"]] = []
        for name, res in all_reservoirs:
            if _has_well_defined_esp(res, batch_size):
                reservoirs.append((name, res))
            elif verbose:
                label = name or type(res).__name__
                print(
                    f"Skipping reservoir '{label}': ESP is undefined "
                    f"(reservoir carries no randomisable state)."
                )

        if not reservoirs:
            raise ValueError(
                "No reservoir with a well-defined ESP found in model "
                "(every reservoir carries an empty state)."
            )

        # Run base orbit from the zero initial state.  Routed through the
        # public, validating, cloning API instead of assigning ``res.state``.
        for _, res in reservoirs:
            res.reset_state(batch_size=batch_size)

        base_states = _run_and_collect(model, reservoirs, inputs)
        if transient > 0:
            base_states = {name: s[:, transient:, :] for name, s in base_states.items()}

        # Evaluate all random restarts.  They are statistically independent and
        # differ only in their initial state, so they fold into the batch
        # dimension and run in one (or a few chunked) forward passes instead of
        # ``iterations`` sequential ones.
        random_states_by_iter = _collect_random_restarts(
            model,
            reservoirs,
            inputs,
            batch_size=batch_size,
            iterations=iterations,
            transient=transient,
            device=device,
            dtype=dtype,
            chunk_size=chunk_size,
            verbose=verbose,
        )

        esp_sums = {name: torch.tensor(0.0, device=device, dtype=dtype) for name, _ in reservoirs}
        if history:
            esp_history = {
                name: torch.zeros(iterations, timesteps, batch_size, device=device, dtype=dtype)
                for name, _ in reservoirs
            }

        # Reduce per restart.  Splitting the batched outputs back into
        # per-restart blocks and reducing them one at a time reproduces the
        # sequential arithmetic exactly (same per-iteration mean, then mean over
        # iterations), so ``window`` / ``relative`` / ``history`` behave
        # identically to the sequential path.
        for i in range(iterations):
            random_states = random_states_by_iter[i]
            for name, base in base_states.items():
                rand = random_states[name]
                # Distance at each (batch, timestep): norm over the feature
                # dimension.  Works for any output rank the reservoir produces.
                dist = torch.norm(base - rand, dim=-1)  # (batch, timesteps)

                if relative:
                    # Scale-free relative distance: divide by the base-state
                    # norm so the index is comparable across sizes / radii.
                    base_norm = torch.norm(base, dim=-1)  # (batch, timesteps)
                    dist = dist / (base_norm + 1e-12)

                # Mean distance for this restart.  With ``window`` set, average
                # only over the trailing window (the asymptotic regime);
                # otherwise over every post-transient timestep.
                dist_for_index = dist[:, -window:] if window is not None else dist
                esp_sums[name] += dist_for_index.mean()

                if history:
                    # Store the full post-transient sequence regardless of
                    # window: (batch, timesteps) -> row i of
                    # (iterations, timesteps, batch).
                    esp_history[name][i] = dist.T  # Transpose to (timesteps, batch)

        # Average over iterations
        esp_indices = {name: [val / iterations] for name, val in esp_sums.items()}

        if history:
            # Wrap history values in lists for API consistency
            esp_history_out = {name: [h] for name, h in esp_history.items()}
            return esp_indices, esp_history_out

        return esp_indices
    finally:
        # Restore the caller's reservoir states, whatever happened above.  Use
        # ``strict=False`` because reservoirs skipped as ESP-undefined (or any
        # that had no state to begin with) are simply absent from the snapshot.
        model.set_reservoir_states(saved_states, strict=False)
        # Reservoirs that started with a ``None`` state (never warmed up) are
        # absent from the snapshot, so ``set_reservoir_states`` would leave them
        # holding an internal state.  Reset them back to ``None`` to fully
        # restore the caller's pre-call condition.
        for module in reservoirs_with_none_state:
            module.reset_state()

DataLoader path¶

TimeSeriesWindowDataset and make_dataloader (both in resdag.data, and re-exported at the top level) are the windowed-DataLoader training path — see Work · Streaming & DataLoaders for the worked loops. The time-series generators in resdag.datasets (lorenz, rossler, henon, mackey_glass, narma, sine) each return a (1, n_timesteps, features) tensor and take n_timesteps as their first positional argument.

data ¶

Data Pipeline: Preparation, I/O, and Streaming Windows¶

Canonical public home for resdag's data plumbing. Three layers live here:

Whole-tensor preparation — split a long trajectory into the fixed warmup/train/target/f_warmup/val tensors the algebraic (:class:resdag.training.ESNTrainer) path consumes, with optional normalization.

File I/O — load and save time series from CSV / NPY / NPZ / NetCDF, always returning the (B, T, D) convention.

Streaming / SGD windows — slice a long trajectory (or many) into fixed-length, batchable windows behind a standard :class:torch.utils.data.DataLoader, for the SGD and streaming-algebraic paths.

CLASS	DESCRIPTION
`TimeSeriesWindowDataset`	Sliding-window `Dataset` over a `(T, D)` tensor, a `(B, T, D)` tensor, or a list of variable-length trajectories. Yields `(input_window, target_window, washout)` triples that never straddle a trajectory boundary.

FUNCTION	DESCRIPTION
`make_dataloader`	Wrap :class:`TimeSeriesWindowDataset` in a `DataLoader` whose batches are `(B, window_len, D)` and feed straight into `ESNLayer` / `ESNModel`.
`prepare_esn_data`	Split a series into warmup, train, target, f_warmup, val.
`normalize_data, denormalize_data`	Normalize a series (and invert it) using various methods.
`load_and_prepare`	Load and prepare data from file(s) in one step.
`load_file, load_csv, load_npy, load_npz, load_nc`	Auto-detecting and format-specific loaders.
`save_csv, save_npy, save_npz, save_nc, list_files`	Savers and a directory listing helper.

Examples:

Whole-tensor preparation for the algebraic path:

>>> from resdag.data import load_file, prepare_esn_data
>>> data = load_file("lorenz.csv")
>>> warmup, train, target, f_warmup, val = prepare_esn_data(
...     data, warmup_steps=100, train_steps=500, val_steps=200, normalize="minmax"
... )

Streaming windows for the SGD path:

>>> import torch
>>> from resdag.data import make_dataloader
>>> series = torch.randn(2000, 3)
>>> loader = make_dataloader(series, batch_size=16, window_len=200, washout=50)
>>> for x, y, washout in loader:
...     model.reset_reservoirs()
...     pred = model(x)
...     loss = mse(pred[:, washout:], y[:, washout:])

See Also

resdag.datasets : Canonical benchmark time-series generators. resdag.layers.IncrementalRidgeReadoutLayer : Streaming ridge readout for the loader.

TimeSeriesWindowDataset ¶

TimeSeriesWindowDataset(series: Tensor | Sequence[Tensor], window_len: int, horizon: int = 1, stride: int = 1, washout: int = 0, targets: Tensor | Sequence[Tensor] | None = None, stats: dict[str, Tensor] | None = None, norm_method: NormMethod | None = None)

Bases: Dataset[WindowItem]

Sliding-window :class:~torch.utils.data.Dataset over time-series data.

Slices one or more trajectories into fixed-length windows suitable for SGD (or streaming-algebraic) training of a reservoir model. Each item is an (input_window, target_window, washout) triple where input_window and target_window both have shape (window_len, D):

Forecasting (default, no external targets): the target is the input shifted forward by horizon steps, so window i covers source steps [start, start + window_len) for the input and [start + horizon, start + horizon + window_len) for the target.
Regression (external targets given): the target is the matching window sliced from targets at the same source positions as the input (horizon is ignored — supply pre-aligned targets).

Windows are generated per trajectory and never straddle a trajectory boundary, so feeding the loader's batches with a per-batch reset_reservoirs() (or relying on :attr:~resdag.layers.BaseReservoirLayer.detach_state_between_calls) keeps reservoir state from leaking across unrelated trajectories. The first washout steps of every window are reported in the returned triple so the caller can exclude them from the loss while the reservoir synchronises.

PARAMETER	DESCRIPTION
`series`	Source data. A `(T, D)` tensor (one trajectory), a `(B, T, D)` tensor (`B` trajectories), or a list of `(T, D)` tensors of varying length (ragged trajectories). TYPE: `torch.Tensor or sequence of torch.Tensor`
`window_len`	Number of timesteps in each input/target window. Must be positive. TYPE: `int`
`horizon`	Forecast offset: the target window is the input window shifted forward by `horizon` steps. Ignored when `targets` is provided. Must be non-negative. TYPE: `int` DEFAULT: `1`
`stride`	Step between successive window start positions within a trajectory. Must be positive. TYPE: `int` DEFAULT: `1`
`washout`	Number of leading steps in each window the caller should exclude from the loss (transient reservoir warmup). Reported per item; must satisfy `0 <= washout < window_len`. TYPE: `int` DEFAULT: `0`
`targets`	External regression targets aligned step-for-step with `series`. Must have the same form and per-trajectory length as `series` (the feature dimension may differ). When given, the dataset returns input/target windows sliced from the same source positions and `horizon` is ignored. TYPE: `torch.Tensor or sequence of torch.Tensor` DEFAULT: `None`
`stats`	Pre-computed normalization statistics from :func:`resdag.data.normalize_data`. When given, every window is normalized on access with the matching method. Requires `norm_method`. TYPE: `dict` DEFAULT: `None`
`norm_method`	Normalization method matching `stats`. Required when `stats` is given, ignored otherwise. TYPE: `(minmax, standard, noncentered, meanpreserving)` DEFAULT: `"minmax"`

ATTRIBUTE	DESCRIPTION
`window_len`	Window length. TYPE: `int`
`horizon`	Forecast offset (forecasting mode only). TYPE: `int`
`stride`	Window start spacing. TYPE: `int`
`washout`	Leading steps to exclude from the loss. TYPE: `int`
`feature_dim`	Input feature dimension `D`. TYPE: `int`
`target_dim`	Target feature dimension (equals `feature_dim` in forecasting mode). TYPE: `int`

RAISES	DESCRIPTION
`ValueError`	For invalid `window_len`/`horizon`/`stride`/`washout`, mismatched `targets`, or an incomplete `stats`/`norm_method` pair.

Examples:

Forecasting windows over a single (T, D) series:

>>> import torch
>>> from resdag.data import TimeSeriesWindowDataset
>>> series = torch.randn(1000, 3)
>>> ds = TimeSeriesWindowDataset(series, window_len=200, horizon=1, washout=50)
>>> x, y, washout = ds[0]
>>> x.shape, y.shape, washout
(torch.Size([200, 3]), torch.Size([200, 3]), 50)

See Also

make_dataloader : Wrap this dataset in a batched DataLoader. resdag.layers.IncrementalRidgeReadoutLayer : Algebraic over-DataLoader fit.

Source code in src/resdag/data/dataset.py

def __init__(
    self,
    series: torch.Tensor | Sequence[torch.Tensor],
    window_len: int,
    horizon: int = 1,
    stride: int = 1,
    washout: int = 0,
    targets: torch.Tensor | Sequence[torch.Tensor] | None = None,
    stats: dict[str, torch.Tensor] | None = None,
    norm_method: NormMethod | None = None,
) -> None:
    if window_len <= 0:
        raise ValueError(f"window_len must be positive, got {window_len}.")
    if horizon < 0:
        raise ValueError(f"horizon must be non-negative, got {horizon}.")
    if stride <= 0:
        raise ValueError(f"stride must be positive, got {stride}.")
    if not 0 <= washout < window_len:
        raise ValueError(
            f"washout must satisfy 0 <= washout < window_len, got washout={washout}, "
            f"window_len={window_len}."
        )
    if (stats is None) != (norm_method is None):
        raise ValueError(
            "stats and norm_method must be provided together; got "
            f"stats={'set' if stats is not None else None}, norm_method={norm_method!r}."
        )

    self.window_len = window_len
    self.horizon = horizon
    self.stride = stride
    self.washout = washout
    self._stats = stats
    self._norm_method = norm_method

    self._inputs = _as_trajectory_list(series)
    self.feature_dim = self._inputs[0].shape[-1]

    self._has_external_targets = targets is not None
    if targets is not None:
        self._targets: list[torch.Tensor] = _as_trajectory_list(targets)
        if len(self._targets) != len(self._inputs):
            raise ValueError(
                f"series has {len(self._inputs)} trajectories but targets has "
                f"{len(self._targets)}; they must match."
            )
        for i, (x_traj, y_traj) in enumerate(zip(self._inputs, self._targets)):
            if x_traj.shape[0] != y_traj.shape[0]:
                raise ValueError(
                    f"trajectory {i}: series length ({x_traj.shape[0]}) and targets "
                    f"length ({y_traj.shape[0]}) must match."
                )
        self.target_dim = self._targets[0].shape[-1]
        # Targets are pre-aligned; no forecast offset is applied.
        self._offset = 0
    else:
        self._targets = self._inputs
        self.target_dim = self.feature_dim
        self._offset = horizon

    # Precompute (trajectory index, window start) for every valid window so
    # __getitem__ is O(1) and windows never cross a trajectory boundary.
    # A window of source span [start, start + window_len) plus the target
    # offset must fit inside the trajectory: start + window_len + offset <= T.
    self._index: list[tuple[int, int]] = []
    for t_idx, traj in enumerate(self._inputs):
        max_start = traj.shape[0] - self.window_len - self._offset
        start = 0
        while start <= max_start:
            self._index.append((t_idx, start))
            start += self.stride

make_dataloader ¶

make_dataloader(series: Tensor | Sequence[Tensor], batch_size: int, window_len: int, horizon: int = 1, stride: int = 1, washout: int = 0, targets: Tensor | Sequence[Tensor] | None = None, stats: dict[str, Tensor] | None = None, norm_method: NormMethod | None = None, shuffle: bool = False, drop_last: bool = False, num_workers: int = 0, **dataloader_kwargs: object) -> DataLoader[WindowItem]

Build a windowed :class:~torch.utils.data.DataLoader over series.

Convenience wrapper that constructs a :class:TimeSeriesWindowDataset from series and the windowing parameters, then returns a :class:~torch.utils.data.DataLoader whose batches are (inputs, targets, washout) with inputs/targets of shape (B, window_len, D) — ready to feed straight into :class:~resdag.layers.ESNLayer / :class:~resdag.core.ESNModel.forward.

PARAMETER	DESCRIPTION
`series`	Source data; see :class:`TimeSeriesWindowDataset`. TYPE: `torch.Tensor or sequence of torch.Tensor`
`batch_size`	Number of windows per batch. TYPE: `int`
`window_len`	Window length (timesteps per input/target window). TYPE: `int`
`horizon`	Forecast offset (ignored when `targets` is given). TYPE: `int` DEFAULT: `1`
`stride`	Spacing between window starts within a trajectory. TYPE: `int` DEFAULT: `1`
`washout`	Leading steps per window to exclude from the loss. TYPE: `int` DEFAULT: `0`
`targets`	External regression targets aligned with `series`. TYPE: `torch.Tensor or sequence of torch.Tensor` DEFAULT: `None`
`stats`	Pre-computed normalization statistics (with `norm_method`). TYPE: `dict` DEFAULT: `None`
`norm_method`	Normalization method matching `stats`. TYPE: `str` DEFAULT: `None`
`shuffle`	Whether to shuffle window order each epoch. TYPE: `bool` DEFAULT: `False`
`drop_last`	Whether to drop the final, smaller-than-`batch_size` batch. TYPE: `bool` DEFAULT: `False`
`num_workers`	Number of worker processes for data loading. TYPE: `int` DEFAULT: `0`
`**dataloader_kwargs`	Extra keyword arguments forwarded to :class:`~torch.utils.data.DataLoader` (e.g. `pin_memory`, `generator`). `collate_fn` is supplied here and must not be overridden. TYPE: `object` DEFAULT: `{}`

RETURNS	DESCRIPTION
`DataLoader`	Loader yielding `(inputs, targets, washout)` batches.

Examples:

>>> import torch
>>> from resdag.data import make_dataloader
>>> series = torch.randn(1000, 3)
>>> loader = make_dataloader(series, batch_size=16, window_len=200, washout=50)
>>> x, y, washout = next(iter(loader))
>>> x.shape  # (B, window_len, D)
torch.Size([16, 200, 3])

See Also

TimeSeriesWindowDataset : The underlying windowing dataset.

Source code in src/resdag/data/dataset.py

def make_dataloader(
    series: torch.Tensor | Sequence[torch.Tensor],
    batch_size: int,
    window_len: int,
    horizon: int = 1,
    stride: int = 1,
    washout: int = 0,
    targets: torch.Tensor | Sequence[torch.Tensor] | None = None,
    stats: dict[str, torch.Tensor] | None = None,
    norm_method: NormMethod | None = None,
    shuffle: bool = False,
    drop_last: bool = False,
    num_workers: int = 0,
    **dataloader_kwargs: object,
) -> DataLoader[WindowItem]:
    """Build a windowed :class:`~torch.utils.data.DataLoader` over ``series``.

    Convenience wrapper that constructs a :class:`TimeSeriesWindowDataset` from
    ``series`` and the windowing parameters, then returns a
    :class:`~torch.utils.data.DataLoader` whose batches are
    ``(inputs, targets, washout)`` with ``inputs``/``targets`` of shape
    ``(B, window_len, D)`` — ready to feed straight into
    :class:`~resdag.layers.ESNLayer` / :class:`~resdag.core.ESNModel.forward`.

    Parameters
    ----------
    series : torch.Tensor or sequence of torch.Tensor
        Source data; see :class:`TimeSeriesWindowDataset`.
    batch_size : int
        Number of windows per batch.
    window_len : int
        Window length (timesteps per input/target window).
    horizon : int, default=1
        Forecast offset (ignored when ``targets`` is given).
    stride : int, default=1
        Spacing between window starts within a trajectory.
    washout : int, default=0
        Leading steps per window to exclude from the loss.
    targets : torch.Tensor or sequence of torch.Tensor, optional
        External regression targets aligned with ``series``.
    stats : dict, optional
        Pre-computed normalization statistics (with ``norm_method``).
    norm_method : str, optional
        Normalization method matching ``stats``.
    shuffle : bool, default=False
        Whether to shuffle window order each epoch.
    drop_last : bool, default=False
        Whether to drop the final, smaller-than-``batch_size`` batch.
    num_workers : int, default=0
        Number of worker processes for data loading.
    **dataloader_kwargs : object
        Extra keyword arguments forwarded to :class:`~torch.utils.data.DataLoader`
        (e.g. ``pin_memory``, ``generator``). ``collate_fn`` is supplied here and
        must not be overridden.

    Returns
    -------
    torch.utils.data.DataLoader
        Loader yielding ``(inputs, targets, washout)`` batches.

    Examples
    --------
    >>> import torch
    >>> from resdag.data import make_dataloader
    >>> series = torch.randn(1000, 3)
    >>> loader = make_dataloader(series, batch_size=16, window_len=200, washout=50)
    >>> x, y, washout = next(iter(loader))
    >>> x.shape  # (B, window_len, D)
    torch.Size([16, 200, 3])

    See Also
    --------
    TimeSeriesWindowDataset : The underlying windowing dataset.
    """
    dataset = TimeSeriesWindowDataset(
        series,
        window_len=window_len,
        horizon=horizon,
        stride=stride,
        washout=washout,
        targets=targets,
        stats=stats,
        norm_method=norm_method,
    )
    return DataLoader(
        dataset,
        batch_size=batch_size,
        shuffle=shuffle,
        drop_last=drop_last,
        num_workers=num_workers,
        collate_fn=_collate_windows,
        **dataloader_kwargs,  # type: ignore[arg-type]
    )