my_rnn.py

my_rnn.py

class RecurrentNeuralNetwork():
    def __init__(self, rnn_units=10, rnn_activation="tanh", return_sequences=False, random_state=0):
        self.init_state = True   # 重みの初期化判定フラグ
        self.loss = None         # トレーニングデータのLoss
        self.val_loss = None     # テストデータのLoss
        self.acc = None          # トレーニングデータの正答率
        self.val_acc = None      # テストデータの正答率
        self.n_layers = 3  # 全レイヤーの数 - 1
        self.rnn_units = rnn_units  # 隠れ層のサイズ
        self.rnn_activation = rnn_activation  # 活性化関数の名前
        self.return_sequences = return_sequences
        np.random.seed(random_state)  # 乱数シード
        
        
    def fit(self, X, Y, batch_size, epochs, mu, validation_data, verbose):
        """
        # 学習の実施
        ## 引数の説明
        - X:  トレーニングサンプル
        - Y:  その教師データ
        - batch_size:  ミニバッチのサイズ
        - epochs:  エポック数
        - mu:  学習率
        - validation_data:  テストデータ。(X_test, y_test)のように。タプル形式で渡す。
        - verbose:  学習ログを出力するかどうか。0で出力しない。それ以外で出力する。
        """
        
        # サンプル数、特徴量数、クラス数
        n_samples, T, n_features = X.shape
        n_classes = Y.shape[-1]
        
        # テストデータ
        val_X = validation_data[0]
        val_Y = validation_data[1]
        
        # 重み＆学習ログ初期化
        if self.init_state:
            self.initialize_history()
            self.create_layers(n_features, n_classes)
            self.init_state = False
        
        # エポックループ
        for i_epoch in range(epochs):
            
            # トレーニング＆テストデータのLossと正答率を保存
            self.record_history(X, Y, val_X, val_Y)
            
            # ミニバッチを回す回数
            n_batch = int(np.floor(n_samples/batch_size))
            
            # ミニバッチループ
            for i_batch in range(n_batch):
                # トレーニングデータをバッチサイズ数分切り取る(端数があるので最後のループは以降全部選択)
                X_batch = X[batch_size*i_batch:batch_size*(i_batch+1+int(i_batch==n_batch-1)), :]
                Y_batch = Y[batch_size*i_batch:batch_size*(i_batch+1+int(i_batch==n_batch-1)), :]
                
                # バックプロパゲーション
                self.back_prop(X_batch, Y_batch)
                
                # 重み・閾値アップデート
                self.update_weights(mu)
            
            # 学習ログ出力
            if verbose!=0:
                self.print_latest_history(i_epoch, epochs)
                
                
    def create_layers(self, n_features, n_classes):
        """レイヤー作成"""
        self.rnn = RNN(units=self.rnn_units, input_dim=n_features, 
                       activation=self.rnn_activation, return_sequences=self.return_sequences)
        if self.return_sequences:
            self.dense = TimeSeriesDense(units=n_classes, input_dim=self.rnn_units)
        else:
            self.dense = Dense(units=n_classes, input_dim=self.rnn_units)
        self.act_func = Activation("linear")
        
        
    def forward_prop(self, X):
        """順伝播演算"""
        Phi = [np.array([])]*self.n_layers
        Z = [np.array([])]*(self.n_layers-1)
        
        Phi[0] = X
        Phi[1] = self.rnn.forward_prop(Phi[0])
        Z[0] = self.rnn.Z
        Z[1] = self.dense.forward_prop(Phi[1])
        Phi[2] = self.act_func.forward_prop(Z[1])
        
        return Z, Phi  # Lossを計算する時にPhi[-1]が必要なので返すようにする
    
    
    def back_prop(self, X, Y):
        """逆伝播演算"""
        Z, Phi = self.forward_prop(X)
        
        dPhi = [np.array([])]*self.n_layers
        dPhi[2] = -(Y - Phi[-1])
        Delta = self.act_func.back_prop(Z[1], dPhi[2])
        dPhi[1] = self.dense.back_prop(Phi[1], Delta)
        dPhi[0] = self.rnn.back_prop(dPhi[1])
        
        
    def update_weights(self, mu):
        """重み・閾値アップデート"""
        self.dense.update_weights(mu)
        self.rnn.update_weights(mu)
        
        
    def mean_square_error(self, X, Y):
        """平均2乗誤差"""
        Z, Phi = self.forward_prop(X)
        n_samples = Y.shape[0]
        return np.sum(0.5*np.power(Y - Phi[-1], 2))/n_samples # サンプル数で割って1サンプル当たりの平均値にする
    
    
    def initialize_history(self):
        """学習ログ初期化"""
        self.loss = np.array([])
        self.val_loss = np.array([])
        
        
    def record_history(self, X, Y, val_X, val_Y):
        """トレーニング＆テストデータのlossを保存"""
        self.loss = np.append(self.loss, self.mean_square_error(X, Y))
        self.val_loss = np.append(self.val_loss, self.mean_square_error(val_X, val_Y))
        
        
    def print_latest_history(self, i_epoch, epochs):
        """学習ログ(ヒストリーデータの最終値)を出力"""
        print("Epoch {0:d}/{1:d}".format(i_epoch + 1, epochs))
        print("- loss: {0:.4f} - val_loss: {1:.4f}".format(self.loss[-1], self.val_loss[-1]))
        
        
    def predict(self, X):
        """予測実施関数"""
        Z, Phi = self.forward_prop(X)
        return Phi[-1]