creates a dataset of sliding windows over a timeseries provided as array this function takes in a sequence of datapoints gathered at equal intervals along with time series parameters such as length of the sequenceswindows spacing between two sequencewindows etc to produce batches of timeseries inputs and targets args data numpy array or eager tensor containing consecutive data points timesteps axis 0 is expected to be the time dimension targets targets corresponding to timesteps in data targetsi should be the target corresponding to the window that starts at index i see example 2 below pass none if you don t have target data in this case the dataset will only yield the input data sequencelength length of the output sequences in number of timesteps sequencestride period between successive output sequences for stride s output samples would start at index datai datai s datai 2 s etc samplingrate period between successive individual timesteps within sequences for rate r timesteps datai datai r datai sequencelength are used for creating a sample sequence batchsize number of timeseries samples in each batch except maybe the last one if none the data will not be batched the dataset will yield individual samples shuffle whether to shuffle output samples or instead draw them in chronological order seed optional int random seed for shuffling startindex optional int data points earlier exclusive than startindex will not be used in the output sequences this is useful to reserve part of the data for test or validation endindex optional int data points later exclusive than endindex will not be used in the output sequences this is useful to reserve part of the data for test or validation returns a tf data dataset instance if targets was passed the dataset yields tuple batchofsequences batchoftargets if not the dataset yields only batchofsequences example 1 consider indices 0 1 98 with sequencelength10 samplingrate2 sequencestride3 shufflefalse the dataset will yield batches of sequences composed of the following indices first sequence 0 2 4 6 8 10 12 14 16 18 second sequence 3 5 7 9 11 13 15 17 19 21 third sequence 6 8 10 12 14 16 18 20 22 24 last sequence 78 80 82 84 86 88 90 92 94 96 in this case the last 2 data points are discarded since no full sequence can be generated to include them the next sequence would have started at index 81 and thus its last step would have gone over 98 example 2 temporal regression consider an array data of scalar values of shape steps to generate a dataset that uses the past 10 timesteps to predict the next timestep you would use python inputdata data 10 targets data10 dataset timeseriesdatasetfromarray inputdata targets sequencelength10 for batch in dataset inputs targets batch assert np arrayequalinputs0 data 10 first sequence steps 09 corresponding target step 10 assert np arrayequaltargets0 data10 break example 3 temporal regression for manytomany architectures consider two arrays of scalar values x and y both of shape 100 the resulting dataset should consist samples with 20 timestamps each the samples should not overlap to generate a dataset that uses the current timestamp to predict the corresponding target timestep you would use python x np arange100 y x2 samplelength 20 inputdataset timeseriesdatasetfromarray x none sequencelengthsamplelength sequencestridesamplelength targetdataset timeseriesdatasetfromarray y none sequencelengthsamplelength sequencestridesamplelength for batch in zipinputdataset targetdataset inputs targets batch assert np arrayequalinputs0 x samplelength second sample equals output timestamps 2040 assert np arrayequaltargets1 ysamplelength 2samplelength break validate strides determine the lowest dtype to store start positions to lower memory usage generate start positions for each initial window position generates indices of the window elements shuffle locally at each iteration creates a dataset of sliding windows over a timeseries provided as array this function takes in a sequence of data points gathered at equal intervals along with time series parameters such as length of the sequences windows spacing between two sequence windows etc to produce batches of timeseries inputs and targets args data numpy array or eager tensor containing consecutive data points timesteps axis 0 is expected to be the time dimension targets targets corresponding to timesteps in data targets i should be the target corresponding to the window that starts at index i see example 2 below pass none if you don t have target data in this case the dataset will only yield the input data sequence_length length of the output sequences in number of timesteps sequence_stride period between successive output sequences for stride s output samples would start at index data i data i s data i 2 s etc sampling_rate period between successive individual timesteps within sequences for rate r timesteps data i data i r data i sequence_length are used for creating a sample sequence batch_size number of timeseries samples in each batch except maybe the last one if none the data will not be batched the dataset will yield individual samples shuffle whether to shuffle output samples or instead draw them in chronological order seed optional int random seed for shuffling start_index optional int data points earlier exclusive than start_index will not be used in the output sequences this is useful to reserve part of the data for test or validation end_index optional int data points later exclusive than end_index will not be used in the output sequences this is useful to reserve part of the data for test or validation returns a tf data dataset instance if targets was passed the dataset yields tuple batch_of_sequences batch_of_targets if not the dataset yields only batch_of_sequences example 1 consider indices 0 1 98 with sequence_length 10 sampling_rate 2 sequence_stride 3 shuffle false the dataset will yield batches of sequences composed of the following indices first sequence 0 2 4 6 8 10 12 14 16 18 second sequence 3 5 7 9 11 13 15 17 19 21 third sequence 6 8 10 12 14 16 18 20 22 24 last sequence 78 80 82 84 86 88 90 92 94 96 in this case the last 2 data points are discarded since no full sequence can be generated to include them the next sequence would have started at index 81 and thus its last step would have gone over 98 example 2 temporal regression consider an array data of scalar values of shape steps to generate a dataset that uses the past 10 timesteps to predict the next timestep you would use python input_data data 10 targets data 10 dataset timeseries_dataset_from_array input_data targets sequence_length 10 for batch in dataset inputs targets batch assert np array_equal inputs 0 data 10 first sequence steps 0 9 corresponding target step 10 assert np array_equal targets 0 data 10 break example 3 temporal regression for many to many architectures consider two arrays of scalar values x and y both of shape 100 the resulting dataset should consist samples with 20 timestamps each the samples should not overlap to generate a dataset that uses the current timestamp to predict the corresponding target timestep you would use python x np arange 100 y x 2 sample_length 20 input_dataset timeseries_dataset_from_array x none sequence_length sample_length sequence_stride sample_length target_dataset timeseries_dataset_from_array y none sequence_length sample_length sequence_stride sample_length for batch in zip input_dataset target_dataset inputs targets batch assert np array_equal inputs 0 x sample_length second sample equals output timestamps 20 40 assert np array_equal targets 1 y sample_length 2 sample_length break validate strides determine the lowest dtype to store start positions to lower memory usage generate start positions for each initial window position generates indices of the window elements shuffle locally at each iteration
import numpy as np

from keras.api_export import keras_export
from keras.utils.module_utils import tensorflow as tf


@keras_export(
    [
        "keras.utils.timeseries_dataset_from_array",
        "keras.preprocessing.timeseries_dataset_from_array",
    ]
)
def timeseries_dataset_from_array(
    data,
    targets,
    sequence_length,
    sequence_stride=1,
    sampling_rate=1,
    batch_size=128,
    shuffle=False,
    seed=None,
    start_index=None,
    end_index=None,
):
    
    if start_index:
        if start_index < 0:
            raise ValueError(
                "`start_index` must be 0 or greater. Received: "
                f"start_index={start_index}"
            )
        if start_index >= len(data):
            raise ValueError(
                "`start_index` must be lower than the length of the "
                f"data. Received: start_index={start_index}, for data "
                f"of length {len(data)}"
            )
    if end_index:
        if start_index and end_index <= start_index:
            raise ValueError(
                "`end_index` must be higher than `start_index`. "
                f"Received: start_index={start_index}, and "
                f"end_index={end_index} "
            )
        if end_index >= len(data):
            raise ValueError(
                "`end_index` must be lower than the length of the "
                f"data. Received: end_index={end_index}, for data of "
                f"length {len(data)}"
            )
        if end_index <= 0:
            raise ValueError(
                "`end_index` must be higher than 0. "
                f"Received: end_index={end_index}"
            )

    
    if sampling_rate <= 0:
        raise ValueError(
            "`sampling_rate` must be higher than 0. Received: "
            f"sampling_rate={sampling_rate}"
        )
    if sampling_rate >= len(data):
        raise ValueError(
            "`sampling_rate` must be lower than the length of the "
            f"data. Received: sampling_rate={sampling_rate}, for data "
            f"of length {len(data)}"
        )
    if sequence_stride <= 0:
        raise ValueError(
            "`sequence_stride` must be higher than 0. Received: "
            f"sequence_stride={sequence_stride}"
        )
    if sequence_stride >= len(data):
        raise ValueError(
            "`sequence_stride` must be lower than the length of the "
            f"data. Received: sequence_stride={sequence_stride}, for "
            f"data of length {len(data)}"
        )

    if start_index is None:
        start_index = 0
    if end_index is None:
        end_index = len(data)

    
    
    num_seqs = end_index - start_index - (sequence_length - 1) * sampling_rate
    if targets is not None:
        num_seqs = min(num_seqs, len(targets))
    if num_seqs < 2147483647:
        index_dtype = "int32"
    else:
        index_dtype = "int64"

    
    start_positions = np.arange(0, num_seqs, sequence_stride, dtype=index_dtype)
    if shuffle:
        if seed is None:
            seed = np.random.randint(1e6)
        rng = np.random.RandomState(seed)
        rng.shuffle(start_positions)

    sequence_length = tf.cast(sequence_length, dtype=index_dtype)
    sampling_rate = tf.cast(sampling_rate, dtype=index_dtype)

    positions_ds = tf.data.Dataset.from_tensors(start_positions).repeat()

    
    indices = tf.data.Dataset.zip(
        (tf.data.Dataset.range(len(start_positions)), positions_ds)
    ).map(
        lambda i, positions: tf.range(
            positions[i],
            positions[i] + sequence_length * sampling_rate,
            sampling_rate,
        ),
        num_parallel_calls=tf.data.AUTOTUNE,
    )

    dataset = sequences_from_indices(data, indices, start_index, end_index)
    if targets is not None:
        indices = tf.data.Dataset.zip(
            (tf.data.Dataset.range(len(start_positions)), positions_ds)
        ).map(
            lambda i, positions: positions[i],
            num_parallel_calls=tf.data.AUTOTUNE,
        )
        target_ds = sequences_from_indices(
            targets, indices, start_index, end_index
        )
        dataset = tf.data.Dataset.zip((dataset, target_ds))
    dataset = dataset.prefetch(tf.data.AUTOTUNE)
    if batch_size is not None:
        if shuffle:
            
            dataset = dataset.shuffle(buffer_size=batch_size * 8, seed=seed)
        dataset = dataset.batch(batch_size)
    else:
        if shuffle:
            dataset = dataset.shuffle(buffer_size=1024, seed=seed)
    return dataset


def sequences_from_indices(array, indices_ds, start_index, end_index):
    dataset = tf.data.Dataset.from_tensors(array[start_index:end_index])
    dataset = tf.data.Dataset.zip((dataset.repeat(), indices_ds)).map(
        lambda steps, inds: tf.gather(steps, inds),
        num_parallel_calls=tf.data.AUTOTUNE,
    )
    return dataset