ステップごとにイベント駆動型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éのブログでは、すでにトレーディングシステムの開発のさまざまな段階について検討しました。 このテーマに関するオンラインコースがあります。