root/lang/python/tremolo/tremolo/indications/neuralnet.py @ 32712

#!/usr/bin/env python
#-*- coding:utf-8 -*-

import sys, os
from foundation import Indication, Indications

try:
        from pybrain.tools.shortcuts import buildNetwork
        from pybrain.datasets import SupervisedDataSet
        from pybrain.structure  import RecurrentNetwork, LinearLayer, SigmoidLayer
        from pybrain.supervised import RPropMinusTrainer
except: 
        raise ImportError("this module needs pybrain.") 

class FeedForwardNetwork (Indication):
        """
        隠れレイヤの関数にシグモイド関数、学習に逆誤差伝搬法を用いた、
        単純ニューラルネットワーク型のクラスです.
        与えられた指標群のデータから、対象となる指標の値を推定します.
        """
        @classmethod
        def isRealIndication(cls): return True
        @classmethod
        def getType(cls): return float
        
        def __init__(self, indications, target, network=None,
                     hiddenLayers=10, dim=1, verbose=True, parent=None):
                """
                オブジェクトのコンストラクタです.
                
                indications  : 対象となるIndicationsインスタンス
                target       : 学習先となる指標
                network      : 使用するネットワーク.指定されない場合は新規に作成される
                hiddenLayers : 隠れレイヤに用いるシグモイドユニットの数
                dim          : 隠れレイヤの数
                verbose      : 詳細なログを出力するか.デフォルトはTrue
                parent       : 親となるオブジェクト
                """
                Indication.__init__(self, None, parent)
                
                if isinstance(indications, Indications):
                        self.__indications = indications
                else: raise TypeError("the attribute 'indications' is not valid.")
                if issubclass(target.__class__, Indication):
                        self.__target = target
                else: raise TypeError("the attribute 'target' is not valid.")
                
                assert hiddenLayers > 0
                assert dim > 0
                self.__hiddenLayers = hiddenLayers
                self.__dim = dim
                self.__verbose = verbose
                
                if network: self.__network = network
                else:
                        args = [len(self.indications),]
                        args += [hiddenLayers for i in xrange(dim)]
                        args += [1,]
                        kwargs = {"outputbias":False, "recurrent":False}
                        self.__network = buildNetwork(*args, **kwargs)
                
                self.__dataset = SupervisedDataSet(len(self.indications), 1)
                self.__trainer = RPropMinusTrainer(self.__network, verbose=verbose)
        
        def getVerbose(self): return self.__verbose
        def setVerbose(self, v):
                self.__verbose = v
                self.__trainer.verbose = v
        verbose = property(getVerbose, setVerbose)
        
        def getIndications(self): return self.__indications
        def setIndications(self, i):
                if isinstance(i, Indications):
                        self.__indications = i
                else: raise AttributeError
        indications = property(getIndications, setIndications)
        
        def getTarget(self): return self.__target
        def setTarget(self, t):
                if issubclass(t.__class__, Indication):
                        self.__target = t
                else: raise AttributeError
        target = property(getTarget, setTarget)
        
        def getNetwork(self): return self.__network
        network = property(getNetwork)
        
        def getHiddenLayers(self): return self.__hiddenlayers
        hiddenlayers = property(getHiddenLayers)
        
        def getDim(self): return self.__dim
        dim = property(getDim)
        
        def getDataset(self): return self.__dataset
        dataset = property(getDataset)
        
        def getTrainer(self): return self.__trainer
        trainer = property(getTrainer)
        
        def getDescription(self): return "NeuralNet"
        
        def addSamples(self, start, end):
                """
                学習に必要なサンプルを指定されたtの範囲で収集します.
                """
                for t in xrange(start, end):
                        input_list  = [c.getScalar(t) for c in self.indications]
                        output_list = [self.target.getScalar(t),]
                        self.__dataset.addSample(input_list, output_list)
        
        def clearSamples(self):
                """
                サンプルを消去します.
                """
                self.__dataset = SupervisedDataSet(len(self.indications), 1)
        
        def train(self):
                """
                与えられたサンプルを元に、ネットワークの学習を行います.
                """
                self.__trainer.trainOnDataset(self.__dataset)
        
        def trainEpochs(self, epochs=2):
                """
                指定された回数分だけ学習を行います.
                """
                self.__trainer.trainOnDataset(self.__dataset, epochs)
                
        def trainUntilConvergence(self, validationProportion=0.25):
                """
                ネットワークが指定された誤り率以下になるまで学習を繰り返します.
                """
                vp = validationProportion
                self.__trainer.trainUntilConvergence(dataset=self.__dataset,
                                                     validationProportion = vp)
        
        def evaluate(self, t):
                args = [c.getScalar(t) for c in self.indications]
                return self.network.activate(args)[0]


class RecurrentNetwork (Indication):
        """
        RNNを用いたニューラルネットワーク型のクラスです.
        与えられた指標群のデータから、対象となる指標の値を推定します.
        """
        @classmethod
        def isRealIndication(cls): return True
        @classmethod
        def getType(cls): return float
        
        def __init__(self, indications, target, network=None,
                     hiddenLayers=10, dim=1, verbose=True, parent=None):
                """
                オブジェクトのコンストラクタです.
                
                indications  : 対象となるIndicationsインスタンス
                target       : 学習先となる指標
                network      : 使用するネットワーク.指定されない場合は新規に作成される
                hiddenLayers : 隠れレイヤに用いるシグモイドユニットの数
                dim          : 隠れレイヤの数
                parent       : 親となるオブジェクト
                """
                Indication.__init__(self, None, parent)
                if isinstance(indications, Indications):
                        self.__indications = indications
                else: raise TypeError("the attribute 'indications' is not valid.")
                if issubclass(target.__class__, Indication):
                        self.__target = target
                else: raise TypeError("the attribute 'target' is not valid.")
                
                assert hiddenLayers > 0
                assert dim > 0
                self.__hiddenLayers = hiddenLayers
                self.__dim = dim
                self.__verbose = verbose
                
                if network: self.__network = network
                else:
                        args = [len(self.indications),]
                        args += [hiddenLayers for i in xrange(dim)]
                        args += [1,]
                        kwargs = {"outputbias":False, "recurrent":False}
                        rnn = buildNetwork(*args, **kwargs)
                        self.__network = rnn
                
                self.__dataset = SupervisedDataSet(len(self.indications), 1)
                self.__trainer = RPropMinusTrainer(self.__network,
                                                   self.__dataset, verbose=verbose)
        
        def getVerbose(self): return self.__verbose
        def setVerbose(self, v):
                self.__verbose = v
                self.__trainer.verbose = v
        verbose = property(getVerbose, setVerbose)
        
        def getIndications(self): return self.__indications
        def setIndications(self, i):
                if isinstance(i, Indications):
                        self.__indications = i
                else: raise AttributeError
        indications = property(getIndications, setIndications)
        
        def getTarget(self): return self.__target
        def setTarget(self, t):
                if issubclass(t.__class__, Indication):
                        self.__target = t
                else: raise AttributeError
        target = property(getTarget, setTarget)
        
        def getNetwork(self): return self.__network
        network = property(getNetwork)
        
        def getHiddenLayers(self): return self.__hiddenlayers
        hiddenlayers = property(getHiddenLayers)
        
        def getDim(self): return self.__dim
        dim = property(getDim)
        
        def getDataset(self): return self.__dataset
        dataset = property(getDataset)
        
        def getTrainer(self): return self.__trainer
        trainer = property(getTrainer)
        
        def getDescription(self): return "ReccurentNeuralNet"
        
        def addSamples(self, start, end):
                """
                学習に必要なサンプルを指定されたtの範囲で収集します.
                """
                for t in xrange(start, end):
                        input_list  = [c.getScalar(t) for c in self.indications]
                        output_list = [self.target.getScalar(t),]
                        self.__dataset.addSample(input_list, output_list)
        
        def clearSamples(self):
                """
                サンプルを消去します.
                """
                self.__dataset = SupervisedDataSet(len(self.indications), 1)
        
        def train(self):
                """
                与えられたサンプルを元に、ネットワークの学習を行います.
                """
                self.__trainer.trainOnDataset(self.__dataset)
        
        def trainEpochs(self, epochs=2):
                """
                指定された回数分だけ学習を行います.
                """
                self.__trainer.trainOnDataset(self.__dataset, epochs)
        
        def trainUntilConvergence(self, validationProportion=0.25):
                """
                ネットワークが指定された誤り率以下になるまで学習を繰り返します.
                """
                vp = validationProportion
                self.__trainer.trainUntilConvergence(dataset=self.__dataset,
                                                     validationProportion = vp)
        
        def evaluate(self, t):
                args = [c.getScalar(t) for c in self.indications]
                return self.network.activate(args)[0]