ステップごとにイベント駆動型Pythonバックテストを行います。 パート2

前回の記事では、イベント指向のバックテストシステムとは何かについて説明し、そのために開発する必要があるクラス階層を整理しました。 今日は、このようなシステムが歴史的テストのコンテキストと取引所の「ライブ」ワークの両方で市場データをどのように使用するかについてお話します。

市場データを操作する

イベント指向のトレーディングシステムを作成する際のタスクの1つは、履歴データのテストと実際のトレーディングのコンテキストで、同じタスクに対して異なるコードを記述する必要性を最小限に抑えることです。 理想的には、これらの各ケースで統一された信号生成とポートフォリオ管理の方法論を使用する必要があります。 これを実現するには、取引シグナル（ Signals



）を生成するStrategy



オブジェクトと、それらに基づいて注文を生成するPortfolio



オブジェクトは、履歴テストとリアルタイム操作の両方のコンテキストで市場データにアクセスするために同じインターフェースを使用する必要があります。



この必要性から、 DataHandler



オブジェクトに基づくクラス階層の概念が出現しました。この概念は、市場データをシステムの他のコンポーネントに転送するためのインターフェースをサブクラスに提供します。 この構成では、サブクラスのハンドラーを単に「破棄」することができ、これは戦略およびポートフォリオ処理を担当するコンポーネントの作業には影響しません。



これらのサブクラスには、 HistoricCSVDataHandler



、 QuandlDataHandler



、 SecuritiesMasterDataHandler



、 InteractiveBrokersMarketFeedDataHandler



などがあります。 ここでは、バーの形式（低、高、終値、ボリューム取引量、OpenInterest建玉）で1日以内に対応する金融データのCSVファイルを読み込む履歴データを含むCSVハンドラーの作成のみを検討します。 このデータに基づいて、システムの各「ハートビート」（ハートビート）を使用して、さまざまな歪みを回避できるコンポーネントStrategy



およびPortfolio



を使用して詳細な分析を既に実行できます。



最初のステップは、必要なライブラリ、特にパンダと抽象基本クラスをインポートすることです。 DataHandler



はDataHandler



イベントを生成するため、 DataHandler



もインポートする必要があります。

 # data.py import datetime import os, os.path import pandas as pd from abc import ABCMeta, abstractmethod from event import MarketEvent
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

DataHandler



は抽象基本クラス（ABA）です。つまり、直接インスタンス化することはできません。 これは、サブクラスを使用してのみ実行できます。 この理由は、ABAが使用する必要のある基礎となるDataHandlerサブクラスのインターフェイスを提供し、相互作用可能な他のクラスとの互換性を提供することです。



Pythonが抽象基本クラスを処理していることを「理解」するために、 _metaclass_



プロパティを使用します。 @abstractmethod



デコレータは、メソッドがサブクラスでオーバーライドされること@abstractmethod



示します（C ++の完全仮想メソッドとまったく同じです）。



興味のある2つのメソッドはget_latest_bars



とupdate_bars



です。 最初のものは、システムの「ハートビート」の現在のタイムスタンプから最後のNバーを返します。これは、 Strategy



クラスの計算を実行するのに役立ちます。 後者の方法は、新しいデータ構造にバー情報を重ね合わせるための分析メカニズムを提供し、予測の歪みを完全に排除します。 クラスをインスタンス化しようとすると、例外がスローされます。

 # data.py class DataHandler(object): """ DataHandler —   ,       (       )  ()  DataHandler       (OLHCVI)     .       ,    ,     .             . """ __metaclass__ = ABCMeta @abstractmethod def get_latest_bars(self, symbol, N=1): """   N    latest_symbol  ,     . """ raise NotImplementedError("Should implement get_latest_bars()") @abstractmethod def update_bars(self): """            . """ raise NotImplementedError("Should implement update_bars()")
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

DataHandler



クラスを説明した後DataHandler



次のステップは、履歴CSVファイルのハンドラーを作成することです。 HistoricCSVDataHandler



は、多くのCSVファイル（金融商品ごとに1つ）をDataFrames



し、それらをパンダ用のDataFrames



フレームDataFrames



変換します。



ハンドラーには、 MarketEvent



市場情報を公開するEvent Queue



、CSVファイルへの絶対パス、ツールのリストなど、いくつかのパラメーターが必要です。 クラスの初期化は次のようになります。

 # data.py class HistoricCSVDataHandler(DataHandler): """ HistoricCSVDataHandler    CSV-        «» ,    . """ def __init__(self, events, csv_dir, symbol_list): """       CSV-   . ,      'symbol.csv',  symbol —   . : events -  . csv_dir -      CSV-. symbol_list -   . """ self.events = events self.csv_dir = csv_dir self.symbol_list = symbol_list self.symbol_data = {} self.latest_symbol_data = {} self.continue_backtest = True self._open_convert_csv_files()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

彼は「SYMBOL.csv」形式のファイルを開こうとします。この形式では、SYMBOLは楽器のティッカーです。 ここで使用される形式は、提案されているDTN IQFeedデータプロバイダーと同じですが、他の形式で動作するように簡単に変更できます。 ファイルを開くことは、_open_convert_csv_filesメソッドによって処理されます。



パンダを使用してHistoricCSVDataHandler内にデータを保存することの利点の1つは、すべての監視対象機器のインデックスをマージできることです。 これにより、欠落したデータでさえ補間することができます。これは、機器のツールごとの比較に役立ちます（戦略平均復帰に必要な場合があります）。 インストゥルメントのインデックスを組み合わせる場合、 union



メソッドとreindex



メソッドが使用されます。

 # data.py def _open_convert_csv_files(self): """  CSV-  ,    pandas DataFrames   .    ,      DTN IQFeed,      . """ comb_index = None for s in self.symbol_list: #  CSV-   ,    self.symbol_data[s] = pd.io.parsers.read_csv( os.path.join(self.csv_dir, '%s.csv' % s), header=0, index_col=0, names=['datetime','open','low','high','close','volume','oi'] ) #    «»  if comb_index is None: comb_index = self.symbol_data[s].index else: comb_index.union(self.symbol_data[s].index) # Set the latest symbol_data to None self.latest_symbol_data[s] = [] # Reindex the dataframes for s in self.symbol_list: self.symbol_data[s] = self.symbol_data[s].reindex(index=comb_index, method='pad').iterrows()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

_get_new_bar



メソッドはジェネレーターを作成し、バーの形式のデータを作成します。 これは、後続のメソッド呼び出しが新しいバーになることを意味します（機器のデータ行の最後に到達するまで続きます）。

 # data.py def _get_new_bar(self, symbol): """     -  : (sybmbol, datetime, open, low, high, close, volume). """ for b in self.symbol_data[symbol]: yield tuple([symbol, datetime.datetime.strptime(b[0], '%Y-%m-%d %H:%M:%S'), b[1][0], b[1][1], b[1][2], b[1][3], b[1][4]])
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

DataHndler



から実装する最初の抽象メソッドはget_latest_bars



です。 latest_symbol_data



構造体の最後のNバーをリストするだけです。 N = 1に設定すると、現在のバーを取得できます。

 # data.py def get_latest_bars(self, symbol, N=1): """  N     latest_symbol,  Nk,   . """ try: bars_list = self.latest_symbol_data[symbol] except KeyError: print "That symbol is not available in the historical data set." else: return bars_list[-N:]
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

最後のメソッドはupdate_bars



、これはDataHandler



からの2番目の抽象メソッドDataHandler



。 最新のバーがlatest_symbol_data



追加されるとキューに入れられるイベント（ MarketEvent



）を生成します。

 # data.py def update_bars(self): """            . """ for s in self.symbol_list: try: bar = self._get_new_bar(s).next() except StopIteration: self.continue_backtest = False else: if bar is not None: self.latest_symbol_data[s].append(bar) self.events.put(MarketEvent())
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

したがって、 DataHandler



があります。これは、システムの他のコンポーネントが市場データを追跡するために使用する専用オブジェクトDataHandler



。 Stragety



、 Portfolio



およびExecutionHandler



オブジェクトが機能するには、現在の市場情報が必要であるため、ストレージの重複の可能性を回避するために中央で作業するのが理にかなっています。

情報から取引シグナルまで：戦略

Strategy



オブジェクトは、マーケットデータの処理に関連するすべての計算をカプセル化して、 Portfolio



オブジェクトへの助言信号を作成します。 イベント指向のバックテスターの開発のこの段階では、テクニカル分析で使用されるインジケーターやフィルターの概念はありません。 それらを実装するには、別のデータ構造を作成できますが、これはすでにこの記事の範囲外です。



戦略の階層は​​比較的単純ですSignalEvents



オブジェクトを作成するための単一の仮想メソッドを持つ抽象基本クラスで構成されます。 戦略階層を作成するには、NumPy、pandas、Queue、abstract base tools、SignalEventをインポートする必要があります。

 # strategy.py import datetime import numpy as np import pandas as pd import Queue from abc import ABCMeta, abstractmethod from event import SignalEvent
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

抽象基本クラスStrategy



は、仮想メソッドcalculate_signals



定義しcalculate_signals



。 市場データの更新に基づいてSignalEvent



オブジェクトの作成を処理するために使用されます。

 # strategy.py class Strategy(object): """ Strategy —   ,     ()    .    Strategy             (OLHCVI),   DataHandler.          ,        —  Strategy      ,     . """ __metaclass__ = ABCMeta @abstractmethod def calculate_signals(self): """      . """ raise NotImplementedError("Should implement calculate_signals()")
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

抽象基本クラスStrategy



の定義Strategy



非常に単純です。 Strategy



オブジェクトでサブクラスを使用する最初の例は、買いと保留の戦略を使用し、対応するBuyAndHoldStrategy



クラスを作成することBuyAndHoldStrategy



。 彼は特定の日に特定の株式を買い、その地位を保持します。 したがって、共有ごとに1つの信号のみが生成されます。



コンストラクター（ __init__



）は、barsマーケットデータハンドラーとイベントイベントキューオブジェクトの存在を必要とします。

 # strategy.py class BuyAndHoldStrategy(Strategy): """   ,              .       Strategy      . """ def __init__(self, bars, events): """   buy and hold. : bars -  DataHandler,      events -   . """ self.bars = bars self.symbol_list = self.bars.symbol_list self.events = events #        ,   True self.bought = self._calculate_initial_bought()
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

BuyAndHoldStrategy



戦略を初期化すると、 bought



ディクショナリには、Falseに設定された各インストゥルメントのキーのセットが含まれます。 特定の楽器が購入されると（ロングポジションが開かれる）、キーはTrueポジションに転送されます。 これにより、 Strategy



オブジェクトはポジションが開いているかどうかを理解できます。

 # strategy.py def _calculate_initial_bought(self): """     bought    False. """ bought = {} for s in self.symbol_list: bought[s] = False return bought
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

このメソッドにcalculate_signals



、仮想メソッドcalculate_signals



実装されています。 このメソッドはリスト内のすべての楽器を調べ、バーハンドラーから最後のバーを取得します。 次に、ツールが「購入」されたかどうか（それによって市場にいるかどうか）をチェックし、 SignalEvent



シグナルオブジェクトがSignalEvent



ます。 次に、それはイベントキューに配置され、購入した辞書は対応する情報で更新されます（購入したツールの場合はTrue）。

 # strategy.py def calculate_signals(self, event): """  "Buy and Hold"     .  ,          . : event -  MarketEvent. """ if event.type == 'MARKET': for s in self.symbol_list: bars = self.bars.get_latest_bars(s, N=1) if bars is not None and bars != []: if self.bought[s] == False: # (Symbol, Datetime, Type = LONG, SHORT or EXIT) signal = SignalEvent(bars[0][0], bars[0][1], 'LONG') self.events.put(signal) self.bought[s] = True
      
      

        
        
        
      

    
        
        
        
      
      

        
        
        
      

    
     

これは非常に単純な戦略ですが、イベント指向の戦略の階層の性質を実証するには十分です。 次の記事では、ペア取引など、より複雑な戦略を検討します。 また、次の記事では、ポジション（損益、PnL）ごとに損益を追跡するPortfolio



階層の作成について説明しPortfolio



。



継続するには...



PSHabréのブログでは、すでにトレーディングシステムの開発のさまざまな段階について検討しました。 このテーマに関するオンラインコースがあります。