이 시리즈의 이전 두 기사에서 우리는 이벤트 기반 백테스팅 시스템이 무엇인지 및 이벤트 객체의 클래스 계층에 대해 논의했습니다. 이 기사에서는 역사적 백테스팅 맥락과 라이브 거래 실행에서 시장 데이터가 어떻게 활용되는지 고려 할 것입니다.
이벤트 기반 거래 시스템에서의 우리의 목표 중 하나는 백테스팅 요소와 라이브 실행 요소 사이의 코드의 중복을 최소화하는 것입니다. 이상적으로 역사적 테스트와 라이브 거래 모두에 동일한 신호 생성 방법론과 포트폴리오 관리 구성 요소를 사용하는 것이 최적입니다. 이를 위해 신호를 생성하는 전략 객체와 이를 기반으로 주문을 제공하는 포트폴리오 객체가 역사 및 라이브 실행 모두에 대한 시장 피드에 동일한 인터페이스를 사용해야합니다.
이것은 데이터 핸들러 객체에 기반한 클래스 계층의 개념을 동원합니다. 이는 모든 하위 클래스에게 시스템 내의 나머지 구성 요소에 시장 데이터를 제공하는 인터페이스를 제공합니다. 이러한 방식으로 모든 하위 클래스 데이터 핸들러는 전략 또는 포트폴리오 계산에 영향을 미치지 않고
특정 예제 하위 클래스는 HistoricCSVDataHandler, QuandlDataHandler, SecuritiesMasterDataHandler, InteractiveBrokersMarketFeedDataHandler 등을 포함 할 수 있습니다. 이 튜토리얼에서는 우리는 단지 시스템 각 박동에 전략 및 포트폴리오 클래스에 데이터를
첫 번째 작업은 필요한 라이브러리를 수입하는 것입니다. 구체적으로 우리는 판다와 추상적인 기본 클래스 도구를 수입 할 것입니다. 데이터 핸들러가 MarketEvents를 생성하기 때문에 우리는 또한 수입해야합니다.event.py이전 튜토리얼에서 설명했듯이:
# data.py
import datetime
import os, os.path
import pandas as pd
abc에서 가져오기 ABCMeta, 추상 방법
이벤트 수입에서 MarketEvent 데이터 핸들러는 추상적인 기본 클래스 (ABC) 이며, 이는 인스턴스를 직접 인스턴스화하는 것이 불가능하다는 것을 의미합니다. 하위 클래스만 인스턴스 할 수 있습니다. 이러한 이유는 ABC가 모든 후속 데이터 핸들러 하위 클래스가 준수해야하는 인터페이스를 제공함으로써 그들과 통신하는 다른 클래스와 호환성을 보장하기 때문입니다.
우리는메타클래스Python이 ABC를 알 수 있도록 @abstractmethod decorator를 사용해서 Python이 하위 클래스에서 메소드를 덮어 쓸 수 있도록 합니다.
관심있는 두 가지 방법은 get_latest_bars 및 update_bars입니다. 전자는 현재 심장 박동 시간표에서 마지막 N 바를 반환하며, 이는 전략 클래스에 필요한 롤링 계산에 유용합니다. 후자는
# data.py
class DataHandler(object):
"""
DataHandler is an abstract base class providing an interface for
all subsequent (inherited) data handlers (both live and historic).
The goal of a (derived) DataHandler object is to output a generated
set of bars (OLHCVI) for each symbol requested.
This will replicate how a live strategy would function as current
market data would be sent "down the pipe". Thus a historic and live
system will be treated identically by the rest of the backtesting suite.
"""
__metaclass__ = ABCMeta
@abstractmethod
def get_latest_bars(self, symbol, N=1):
"""
Returns the last N bars from the latest_symbol list,
or fewer if less bars are available.
"""
raise NotImplementedError("Should implement get_latest_bars()")
@abstractmethod
def update_bars(self):
"""
Pushes the latest bar to the latest symbol structure
for all symbols in the symbol list.
"""
raise NotImplementedError("Should implement update_bars()")
데이터 핸들러 ABC를 지정하면 다음 단계는 역사 CSV 파일의 핸들러를 만드는 것입니다. 특히 HistoricCSVDataHandler는 여러 CSV 파일을, 각각의 기호에 한 개를 가져다가 팬다 데이터 프레임의 사전으로 변환합니다.
데이터 핸들러는 몇 가지 매개 변수, 즉 MarketEvent 정보를 푸시하는 이벤트 큐, CSV 파일의 절대 경로 및 기호 목록이 필요합니다. 클래스의 초기화:
# data.py
class HistoricCSVDataHandler(DataHandler):
"""
HistoricCSVDataHandler is designed to read CSV files for
each requested symbol from disk and provide an interface
to obtain the "latest" bar in a manner identical to a live
trading interface.
"""
def __init__(self, events, csv_dir, symbol_list):
"""
Initialises the historic data handler by requesting
the location of the CSV files and a list of symbols.
It will be assumed that all files are of the form
'symbol.csv', where symbol is a string in the list.
Parameters:
events - The Event Queue.
csv_dir - Absolute directory path to the CSV files.
symbol_list - A list of symbol strings.
"""
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()
그것은 암묵적으로
HistoricCSVDataHandler 내에서 pandas를 데이터 스토어로 사용하는 것의 장점 중 하나는 추적되는 모든 기호의 인덱스가 합쳐질 수 있다는 것입니다. 이것은 부족한 데이터 포인트를 앞으로, 뒤로 또는 이러한 격차 내에서
# data.py
def _open_convert_csv_files(self):
"""
Opens the CSV files from the data directory, converting
them into pandas DataFrames within a symbol dictionary.
For this handler it will be assumed that the data is
taken from DTN IQFeed. Thus its format will be respected.
"""
comb_index = None
for s in self.symbol_list:
# Load the CSV file with no header information, indexed on date
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']
)
# Combine the index to pad forward values
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):
"""
Returns the latest bar from the data feed as a tuple of
(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]])
데이터 핸들러에서 구현되는 첫 번째 추상 메소드는 get_latest_bars입니다. 이 메소드는 단순히 최신_신호_데이터 구조에서 마지막 N 바의 목록을 제공합니다. N=1을 설정하면 현재 바를 검색 할 수 있습니다.
# data.py
def get_latest_bars(self, symbol, N=1):
"""
Returns the last N bars from the latest_symbol list,
or N-k if less available.
"""
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의 두 번째 추상 메소드입니다. 그것은 단순히 MarketEvent을 생성하여 최신 바를 최신_symbol_data에 첨부하면서 대기열에 추가됩니다.
# data.py
def update_bars(self):
"""
Pushes the latest bar to the latest_symbol_data structure
for all symbols in the symbol list.
"""
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에서 파생된 객체를 가지고 있으며, 나머지 구성 요소가 시장 데이터를 추적하는 데 사용됩니다. 전략, 포트폴리오 및 실행 처리 객체 모두 현재 시장 데이터를 필요로하므로 저장 중복을 피하기 위해 중앙 집중시키는 것이 합리적입니다.
다음 기사에서는 전략 클래스 계층을 고려하고 여러 기호를 처리하기 위해 전략을 어떻게 설계할 수 있는지 설명하여 포트폴리오 객체에 대한 여러 신호 이벤트를 생성합니다.