Transform Timeseries

timeseries

CLASS	DESCRIPTION
TakensEmbedding	Takens time-delay embedding transformer.

FUNCTION	DESCRIPTION
takens_embedding	Compute Takens time-delay embedding of a time series.

Classes

TakensEmbedding(embedding_dim: int, delay: int = 1, original_length: int | None = None)

Bases: Transformer

Takens time-delay embedding transformer.

Applies Takens' theorem for phase space reconstruction by creating time-delayed embeddings of time series data.

The transformer can be used in pipelines with other transforms and is compatible with PyTorch's module system.

PARAMETER	DESCRIPTION
embedding_dim	Embedding dimension (number of delayed copies), must be >= 1. TYPE: int
delay	Time delay between consecutive embedded dimensions (in samples), must be >= 1. TYPE: int, optional, by default 1 DEFAULT: 1
original_length	Original time series length for inverse transformation. Required if inverse_transform will be used. TYPE: int | None, optional, by default None DEFAULT: None

ATTRIBUTE	DESCRIPTION
embedding_dim	The embedding dimension. TYPE: int
delay	The time delay in samples. TYPE: int
original_length	Original time series length if provided. TYPE: int | None

Examples:

Basic usage with univariate time series:

>>> import torch
>>> from spectre.transform import TakensEmbedding
>>> embedder = TakensEmbedding(embedding_dim=3, delay=1)
>>> X = torch.randn(32, 100)  # 32 samples, 100 timesteps each
>>> X_embedded = embedder.transform(X)
>>> X_embedded.shape
torch.Size([32, 98, 3])

Multivariate time series:

>>> X = torch.randn(16, 200, 4)  # 16 samples, 200 timesteps, 4 features
>>> embedder = TakensEmbedding(embedding_dim=5, delay=2)
>>> X_embedded = embedder.transform(X)
>>> X_embedded.shape  # 200 - (5-1)*2 = 192 points, 5*4 = 20 features
torch.Size([16, 192, 20])

With inverse transformation:

>>> embedder = TakensEmbedding(
...     embedding_dim=3,
...     delay=1,
...     original_length=100,
... )
>>> X = torch.randn(10, 100)
>>> X_embedded = embedder(X)
>>> X_reconstructed = embedder.inverse_transform(X_embedded)
>>> X_reconstructed.shape
torch.Size([10, 100])

Use in a pipeline

>>> from spectre.transform import CompositeTransformer, StandardizationTransformer
>>> pipeline = CompositeTransformer(
...     [
...         TakensEmbedding(embedding_dim=4, delay=2),
...         StandardizationTransformer(),
...     ]
... )
>>> X = torch.randn(20, 150)
>>> X_processed = pipeline(X)

METHOD	DESCRIPTION
transform	Apply Takens embedding to input time series.
inverse_transform	Reconstruct original time series from embedding.

Source code in spectre/transform/timeseries.py

def __init__(
    self,
    embedding_dim: int,
    delay: int = 1,
    original_length: int | None = None,
) -> None:
    super().__init__()

    if not isinstance(embedding_dim, int) or embedding_dim < 1:
        raise ValueError(f"embedding_dim must be >= 1, got {embedding_dim}.")

    if not isinstance(delay, int) or delay < 1:
        raise ValueError(f"delay must be >= 1, got {delay}.")

    if original_length is not None:
        if not isinstance(original_length, int) or original_length < 1:
            raise ValueError(
                f"original_length must be >= 1, got {original_length}."
            )

        min_length = (embedding_dim - 1) * delay + 1
        if original_length < min_length:
            raise ValueError(
                f"original_length ({original_length}) too short for "
                f"embedding_dim={embedding_dim} and delay={delay}. "
                f"Need at least {min_length}."
            )

    self.embedding_dim = embedding_dim
    self.delay = delay
    self.original_length = original_length

Functions

transform(X: torch.Tensor) -> torch.Tensor

Apply Takens embedding to input time series.

PARAMETER	DESCRIPTION
X	Input time series of shape (batch_size, n_timesteps) for univariate or (batch_size, n_timesteps, n_features) for multivariate. TYPE: Tensor

RETURNS	DESCRIPTION
Tensor	Embedded time series of shape (batch_size, n_embedded_points, embedding_dim) or (batch_size, n_embedded_points, embedding_dim * n_features).

Source code in spectre/transform/timeseries.py

def transform(self, X: torch.Tensor) -> torch.Tensor:
    """
    Apply Takens embedding to input time series.

    Parameters
    ----------
    X : torch.Tensor
        Input time series of shape (batch_size, n_timesteps) for univariate
        or (batch_size, n_timesteps, n_features) for multivariate.


    Returns
    -------
    torch.Tensor
        Embedded time series of shape (batch_size, n_embedded_points,
        embedding_dim) or (batch_size, n_embedded_points, embedding_dim *
        n_features).
    """
    return takens_embedding(X, self.embedding_dim, self.delay)

inverse_transform(X: torch.Tensor) -> torch.Tensor

Reconstruct original time series from embedding.

Extracts the first embedded dimension and reconstructs the original time series length by using the most recent value at each time step. Note that this is an approximate inverse as the embedding loses information about the first (embedding_dim - 1) * delay points.

PARAMETER	DESCRIPTION
X	Embedded time series of shape (batch_size, n_embedded_points, embedding_dim) or (batch_size, n_embedded_points, embedding_dim * n_features). TYPE: Tensor

RETURNS	DESCRIPTION
Tensor	Reconstructed time series of shape (batch_size, original_length) or (batch_size, original_length, n_features).

Notes

This is not a true inverse because the embedding process loses information about early time points. The reconstruction extracts the most advanced time coordinate from each embedded vector.

Source code in spectre/transform/timeseries.py

def inverse_transform(self, X: torch.Tensor) -> torch.Tensor:
    """
    Reconstruct original time series from embedding.

    Extracts the first embedded dimension and reconstructs the original
    time series length by using the most recent value at each time step.
    Note that this is an approximate inverse as the embedding loses
    information about the first (embedding_dim - 1) * delay points.

    Parameters
    ----------
    X : torch.Tensor
        Embedded time series of shape (batch_size, n_embedded_points,
        embedding_dim) or (batch_size, n_embedded_points, embedding_dim *
        n_features).

    Returns
    -------
    torch.Tensor
        Reconstructed time series of shape (batch_size, original_length)
        or (batch_size, original_length, n_features).

    Notes
    -----
    This is not a true inverse because the embedding process loses
    information about early time points. The reconstruction extracts the
    most advanced time coordinate from each embedded vector.
    """
    if self.original_length is None:
        raise ValueError(
            "original_length must be provided at initialization to use "
            "inverse_transform."
        )

    batch_size = X.shape[0]
    n_embedded = X.shape[1]

    # Determine if this was univariate or multivariate
    # For univariate: shape is (batch, n_embedded, embedding_dim)
    # For multivariate: shape is (batch, n_embedded, embedding_dim *
    # n_features)
    embedding_features = X.shape[2]

    if embedding_features % self.embedding_dim != 0:
        raise ValueError(
            f"Embedded features ({embedding_features}) not divisible by "
            f"embedding_dim ({self.embedding_dim})."
        )

    n_features = embedding_features // self.embedding_dim

    # Reshape to separate embedding dimensions and features
    # (batch, n_embedded, embedding_dim, n_features)
    X_reshaped = X.reshape(batch_size, n_embedded, self.embedding_dim, n_features)

    # The first embedding dimension (index 0) contains x(t)
    # Extract it: (batch, n_embedded, n_features)
    reconstructed_core = X_reshaped[:, :, 0, :]

    # Pad the beginning to restore original length
    n_lost = self.original_length - n_embedded

    # Pad with the first value (simple strategy)
    if n_lost > 0:
        pad_value = reconstructed_core[:, 0:1, :].expand(
            batch_size, n_lost, n_features
        )
        reconstructed = torch.cat([pad_value, reconstructed_core], dim=1)
    else:
        reconstructed = reconstructed_core

    # For univariate case, squeeze the feature dimension
    if n_features == 1:
        reconstructed = reconstructed.squeeze(-1)

    return reconstructed

Functions

takens_embedding(X: torch.Tensor, embedding_dim: int, delay: int = 1) -> torch.Tensor

Compute Takens time-delay embedding of a time series.

Creates a phase space reconstruction by stacking time-delayed copies of the input. For a 1D time series, this creates an embedding_dim dimensional trajectory. For multivariate time series, each variable is embedded separately and concatenated.

The embedding transforms a time series \(x(t)\) into vectors: \([x(t), x(t-\tau), x(t-2\tau), ..., x(t-(m-1)\tau)]\) where \(\tau\) is the delay and \(m\) is the embedding dimension.

PARAMETER	DESCRIPTION
X	Input time series data of shape (batch_size, n_timesteps) for univariate or (batch_size, n_timesteps, n_features) for multivariate data. TYPE: Tensor
embedding_dim	Embedding dimension (number of delayed copies), must be >= 1. TYPE: int
delay	Time delay between consecutive embedded dimensions (in samples), must be >= 1. TYPE: int, optional, by default 1 DEFAULT: 1

RETURNS	DESCRIPTION
Tensor	Embedded time series of shape (batch_size, n_embedded_points, embedding_dim) for univariate or (batch_size, n_embedded_points, embedding_dim * n_features) for multivariate data, where: n_embedded_points = n_timesteps - (embedding_dim - 1) * delay

Examples:

Univariate time series embedding:

>>> import torch
>>> from spectre.transform import takens_embedding
>>> X = torch.tensor([[1.0, 2.0, 3.0, 4.0, 5.0]])
>>> embedded = takens_embedding(X, embedding_dim=3, delay=1)
>>> embedded.shape
torch.Size([1, 3, 3])
>>> embedded[0]
tensor([[3., 2., 1.],
        [4., 3., 2.],
        [5., 4., 3.]])

Multivariate time series:

>>> X = torch.randn(10, 100, 3)  # 10 samples, 100 timesteps, 3 features
>>> embedded = takens_embedding(X, embedding_dim=5, delay=2)
>>> embedded.shape  # 5 * 3 = 15 features in embedding
torch.Size([10, 92, 15])

Larger delay:

>>> X = torch.randn(1, 50)
>>> embedded = takens_embedding(X, embedding_dim=4, delay=3)
>>> embedded.shape  # 50 - (4-1)*3 = 41 points
torch.Size([1, 41, 4])

Source code in spectre/transform/timeseries.py

def takens_embedding(
    X: torch.Tensor, embedding_dim: int, delay: int = 1
) -> torch.Tensor:
    """
    Compute Takens time-delay embedding of a time series.

    Creates a phase space reconstruction by stacking time-delayed copies of the
    input. For a 1D time series, this creates an embedding_dim dimensional
    trajectory. For multivariate time series, each variable is embedded
    separately and concatenated.

    The embedding transforms a time series $x(t)$ into vectors:
    $[x(t), x(t-\\tau), x(t-2\\tau), ..., x(t-(m-1)\\tau)]$
    where $\\tau$ is the delay and $m$ is the embedding dimension.

    Parameters
    ----------
    X : torch.Tensor
        Input time series data of shape (batch_size, n_timesteps) for univariate
        or (batch_size, n_timesteps, n_features) for multivariate data.

    embedding_dim : int
        Embedding dimension (number of delayed copies), must be >= 1.

    delay : int, optional, by default 1
        Time delay between consecutive embedded dimensions (in samples),
        must be >= 1.

    Returns
    -------
    torch.Tensor
        Embedded time series of shape (batch_size, n_embedded_points,
        embedding_dim) for univariate or (batch_size, n_embedded_points,
        embedding_dim * n_features) for multivariate data, where:
        `n_embedded_points = n_timesteps - (embedding_dim - 1) * delay`

    Examples
    --------
    Univariate time series embedding:

    >>> import torch
    >>> from spectre.transform import takens_embedding
    >>> X = torch.tensor([[1.0, 2.0, 3.0, 4.0, 5.0]])
    >>> embedded = takens_embedding(X, embedding_dim=3, delay=1)
    >>> embedded.shape
    torch.Size([1, 3, 3])
    >>> embedded[0]
    tensor([[3., 2., 1.],
            [4., 3., 2.],
            [5., 4., 3.]])

    Multivariate time series:

    >>> X = torch.randn(10, 100, 3)  # 10 samples, 100 timesteps, 3 features
    >>> embedded = takens_embedding(X, embedding_dim=5, delay=2)
    >>> embedded.shape  # 5 * 3 = 15 features in embedding
    torch.Size([10, 92, 15])

    Larger delay:

    >>> X = torch.randn(1, 50)
    >>> embedded = takens_embedding(X, embedding_dim=4, delay=3)
    >>> embedded.shape  # 50 - (4-1)*3 = 41 points
    torch.Size([1, 41, 4])
    """
    if not isinstance(X, torch.Tensor):
        raise TypeError(f"X must be torch.Tensor, got {type(X).__name__}.")

    if X.dim() not in [2, 3]:
        raise ValueError(
            f"X must be 2D (batch_size, n_timesteps) or 3D "
            f"(batch_size, n_timesteps, n_features), got shape {X.shape}."
        )

    if not isinstance(embedding_dim, int) or embedding_dim < 1:
        raise ValueError(f"embedding_dim must be >= 1, got {embedding_dim}.")

    if not isinstance(delay, int) or delay < 1:
        raise ValueError(f"delay must be >= 1, got {delay}.")

    # Handle both 2D and 3D inputs
    if X.dim() == 2:
        # Univariate: (batch_size, n_timesteps) -> (batch_size, n_timesteps, 1)
        X = X.unsqueeze(-1)
        squeeze_output = True
    else:
        squeeze_output = False

    batch_size, n_timesteps, n_features = X.shape

    # Check if time series is long enough
    min_length = (embedding_dim - 1) * delay + 1
    if n_timesteps < min_length:
        raise ValueError(
            f"Time series too short for embedding. Need at least "
            f"{min_length} timesteps for embedding_dim={embedding_dim} "
            f"and delay={delay}, got {n_timesteps}."
        )

    # Number of points in embedded space
    n_embedded = n_timesteps - (embedding_dim - 1) * delay

    # Create delay embeddings for each feature
    delayed_series = []

    for i in range(embedding_dim):
        start_idx = (embedding_dim - 1 - i) * delay
        end_idx = start_idx + n_embedded
        delayed_series.append(X[:, start_idx:end_idx, :])

    # Stack along the last dimension: (batch, n_embedded, embedding_dim,
    # n_features)
    embedded = torch.stack(delayed_series, dim=2)

    # Reshape to (batch, n_embedded, embedding_dim * n_features)
    embedded = embedded.reshape(batch_size, n_embedded, embedding_dim * n_features)

    # For univariate case with n_features=1, optionally squeeze
    if squeeze_output and n_features == 1:
        # Result is (batch, n_embedded, embedding_dim)
        embedded = embedded.squeeze(-1)
        if embedding_dim == 1:
            embedded = embedded.unsqueeze(-1)

    return embedded