Kerasでgeneratorを使った時系列予測のサンプルコードです。

keras_reccurent_generator.py

# -*- coding: utf-8 -*-
import numpy as np
from keras.utils import Sequence
from keras.models import Sequential
from keras.layers import Dense, SimpleRNN

class ReccurentTrainingGenerator(Sequence):
    """ Reccurent レイヤーを訓練するためのデータgeneratorクラス """
    def _resetindices(self):
        """バッチとして出力するデータのインデックスを乱数で生成する """
        self.num_called = 0 # 同一のエポック内で __getitem__　メソッドが呼び出された回数
        
        all_idx = np.random.permutation(np.arange(self.num_samples))
        remain_idx = np.random.choice(np.arange(self.num_samples),
                                      size=(self.steps_per_epoch*self.batch_size-len(all_idx)),
                                      replace=False)
        self.indices = np.hstack([all_idx, remain_idx]).reshape(self.steps_per_epoch, self.batch_size)
        
    def __init__(self, x_set, y_set, batch_size, timesteps, delay):
        """
        x_set     : 説明変数 (データ点数×特徴量数)のNumPy配列
        y_set     : 目的変数 (データ点数×1)のNumPy配列
        batch_size: バッチサイズ
        timesteps : どの程度過去からデータをReccurent層に与えるか
        delay     : 目的変数をどの程度遅らせるか
        """
        self.x = np.array(x_set)
        self.y = np.array(y_set)
        self.batch_size = batch_size
        self.steps = timesteps
        self.delay = delay
        
        self.num_samples = len(self.x)-timesteps-delay+1
        self.steps_per_epoch = int(np.ceil( self.num_samples / float(batch_size)))
        
        self._resetindices()
        
    def __len__(self):
        """ 1エポックあたりのステップ数を返す """
        return self.steps_per_epoch
        
    def __getitem__(self, idx):
        """ データをバッチにまとめて出力する """
        indices_temp = self.indices[idx]
        
        batch_x = np.array([self.x[i:i+self.steps] for i in indices_temp])
        batch_y = self.y[indices_temp+self.steps+self.delay-1]
        
        if self.num_called==(self.steps_per_epoch-1):
            self._resetindices() # 1エポック内の全てのバッチを返すと、データをシャッフルする
        else:
            self.num_called += 1
        
        return batch_x, batch_y

class ReccurentPredictingGenerator(Sequence):
    """ Reccurent レイヤーで予測するためのデータgeneratorクラス """ 
    def __init__(self, x_set, batch_size, timesteps):
        """
        x_set     : 説明変数 (データ点数×特徴量数)のNumPy配列
        batch_size: バッチサイズ
        timesteps : どの程度過去からデータをReccurent層に与えるか
        """
        self.x = np.array(x_set)
        self.batch_size = batch_size
        self.steps = timesteps
        
        self.num_samples = len(self.x)-timesteps+1
        self.steps_per_epoch = int(np.floor( self.num_samples / float(batch_size)))
        
    def __len__(self):
        """ 1エポックあたりのステップ数を返す """
        return self.steps_per_epoch
        
    def __getitem__(self, idx):
        """ データをバッチにまとめて出力する """
        start_idx = idx*self.batch_size
        batch_x = [self.x[start_idx+i : start_idx+i+self.steps] for i in range(self.batch_size)]
        return np.array(batch_x)


x_base = np.array([-1,-1,0,0,1,1,0,0], dtype=np.float32).reshape(-1, 1)
x_set = np.empty([0, 1], dtype=np.float32)

for i in range(10):
    x_set = np.vstack([x_set, x_base]) # 説明変数
    
y_set = x_set.copy() # 目的変数

timesteps = 5

RTG = ReccurentTrainingGenerator(x_set, y_set, batch_size=10, 
                                 timesteps=timesteps, delay=1)
# 学習用ジェネレータ

#%% ニューラルネットモデルを作成・学習
actfunc = "tanh"

model = Sequential()
model.add(SimpleRNN(10, activation=actfunc, 
                    batch_input_shape=(None, timesteps, 1)))
model.add(Dense(10, activation=actfunc))
model.add(Dense(1))

model.compile(optimizer='sgd', loss='mean_squared_error')

history = model.fit_generator(RTG, epochs=20, verbose=1) # 学習する

#%% 予測
x_test = np.array([-1,-1,0,0,1], dtype=np.float32).reshape(-1, 1)
# 検証データ

RPG  = ReccurentPredictingGenerator(x_test, batch_size=1, timesteps=timesteps)
# 予測用ジェネレータ

pred = model.predict_generator(RPG) # 予測する
print(pred)