Kerasの時系列予測でgeneratorを使って大容量データを扱う 前編

import numpy as np
from keras.utils import Sequence
from keras.models import Sequential
from keras.layers import Dense, SimpleRNN

fit_generator(self, generator, epochs=1, 
              validation_data=None, shuffle=True)

predict_generator(self, generator)

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)