Keras 3 API 文件 / 資料載入 / 時間序列資料載入

時間序列資料載入

[原始碼]

timeseries_dataset_from_array 函數

keras.utils.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,
)

建立一個時間序列的滑動視窗資料集,時間序列以陣列形式提供。

此函數接收以相等間隔收集的資料點序列,以及時間序列參數,例如序列/視窗的長度、兩個序列/視窗之間的間距等,以產生時間序列輸入和目標的批次。

參數

  • data:包含連續資料點(時間步)的 NumPy 陣列或即時張量。軸 0 預期為時間維度。
  • targets:對應於 data 中時間步的目標。targets[i] 應該是與索引 i 開始的視窗相對應的目標(請參閱下面的範例 2)。如果沒有目標資料,則傳遞 None(在這種情況下,資料集將只產生輸入資料)。
  • sequence_length:輸出序列的長度(以時間步數為單位)。
  • sequence_stride:連續輸出序列之間的間隔。對於間隔 s,輸出樣本將從索引 data[i]data[i + s]data[i + 2 * s] 等開始。
  • sampling_rate:序列內連續個別時間步之間的間隔。對於速率 r,時間步 data[i], data[i + r], ... data[i + sequence_length] 用於建立樣本序列。
  • batch_size:每個批次中的時間序列樣本數量(最後一個批次可能除外)。如果為 None,則不會對資料進行批次處理(資料集將產生個別樣本)。
  • shuffle:是否打亂輸出樣本,或是按時間順序繪製它們。
  • seed:可選的整數;用於打亂的隨機種子。
  • start_index:可選的整數;早於(不包含)start_index 的資料點將不會在輸出序列中使用。這對於保留部分資料以進行測試或驗證很有用。
  • end_index:可選的整數;晚於(不包含)end_index 的資料點將不會在輸出序列中使用。這對於保留部分資料以進行測試或驗證很有用。

返回

一個 tf.data.Dataset 實例。如果傳遞了 targets,則資料集會產生元組 (batch_of_sequences, batch_of_targets)。如果沒有,則資料集僅產生 batch_of_sequences

範例 1

考慮索引 [0, 1, ... 98]。對於 sequence_length=10, sampling_rate=2, sequence_stride=3shuffle=False,資料集將產生由以下索引組成的序列批次

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]

在這種情況下,最後 2 個資料點將被捨棄,因為無法生成包含它們的完整序列(下一個序列將從索引 81 開始,因此它的最後一步將超過 98)。

範例 2:時間迴歸。

考慮一個形狀為 (steps,) 的純量值陣列 data。要生成一個使用過去 10 個時間步來預測下一個時間步的資料集,您可以使用

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

範例 3:用於多對多架構的時間迴歸。

考慮兩個形狀均為 (100,) 的純量值陣列 XY。生成的資料集應包含每個 20 個時間戳的樣本。樣本不應重疊。要生成一個使用目前時間戳來預測相應目標時間步的資料集,您可以使用

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

[原始碼]

pad_sequences 函數

keras.utils.pad_sequences(
    sequences, maxlen=None, dtype="int32", padding="pre", truncating="pre", value=0.0
)

將序列填充至相同長度。

此函數將序列列表(長度為 num_samples)(整數列表)轉換為形狀為 (num_samples, num_timesteps) 的 2D NumPy 陣列。num_timesteps 要么是提供的 maxlen 參數,要么是列表中最長序列的長度。

num_timesteps 短的序列會用 value 填充,直到它們長度達到 num_timesteps

長度超過 num_timesteps 的序列會被截斷,以符合所需的長度。

填充或截斷發生的位置分別由參數 paddingtruncating 決定。預先填充或從序列開頭移除值是預設行為。

>>> sequence = [[1], [2, 3], [4, 5, 6]]
>>> keras.utils.pad_sequences(sequence)
array([[0, 0, 1],
       [0, 2, 3],
       [4, 5, 6]], dtype=int32)
>>> keras.utils.pad_sequences(sequence, value=-1)
array([[-1, -1,  1],
       [-1,  2,  3],
       [ 4,  5,  6]], dtype=int32)
>>> keras.utils.pad_sequences(sequence, padding='post')
array([[1, 0, 0],
       [2, 3, 0],
       [4, 5, 6]], dtype=int32)
>>> keras.utils.pad_sequences(sequence, maxlen=2)
array([[0, 1],
       [2, 3],
       [5, 6]], dtype=int32)

參數

  • sequences:序列列表(每個序列都是一個整數列表)。
  • maxlen:可選的整數,所有序列的最大長度。如果未提供,則序列將填充至最長單個序列的長度。
  • dtype:(可選,預設為 "int32")。輸出序列的類型。要使用可變長度字串填充序列,您可以使用 object
  • padding:字串,"pre" 或 "post"(可選,預設為 "pre"):在每個序列之前或之後進行填充。
  • truncating:字串,"pre" 或 "post"(可選,預設為 "pre"):從大於 maxlen 的序列中移除值,可以從序列的開頭或結尾移除。
  • value:浮點數或字串,填充值。(可選,預設為 0.

返回

形狀為 (len(sequences), maxlen) 的 NumPy 陣列