В предыдущих двух статьях серии мы обсуждали, что такое система обратного тестирования, основанная на событиях, и иерархию классов для объекта Event.
Одна из наших целей в системе торговли, основанной на событиях, заключается в минимизации дублирования кода между элементом обратного тестирования и элементом выполнения в режиме реального времени. В идеале было бы оптимально использовать одну и ту же методологию генерации сигналов и компоненты управления портфелем как для исторического тестирования, так и для реального времени торговли. Чтобы это сработало, объект Стратегия, генерирующий сигналы, и объект Портфель, предоставляющий ордера на основе них, должны использовать идентичный интерфейс для рынка как для исторического, так и для реального времени.
Это мотивирует концепцию иерархии классов, основанной на объекте DataHandler, который дает всем подклассам интерфейс для предоставления рыночных данных остальным компонентам в системе.
Конкретные примеры подклассов могут включать HistoricCSVDataHandler, QuandlDataHandler, SecuritiesMasterDataHandler, InteractiveBrokersMarketFeedDataHandler и т. Д. В этом учебном пособии мы рассмотрим только создание исторического обработчика данных CSV, который будет загружать внутридневные данные CSV для акций в набор баров Open-Low-High-Close-Volume-OpenInterest. Это затем может быть использовано для
Первая задача - импортировать необходимые библиотеки. В частности, мы собираемся импортировать панды и абстрактные инструменты базового класса.event.pyкак описано в предыдущем руководстве:
# data.py
import datetime
import os, os.path
import pandas as pd
из abc import ABCMeta, абстрактный метод
от импорта события MarketEvent DataHandler является абстрактным базовым классом (ABC), что означает, что невозможно инстанцировать экземпляр напрямую. Только подклассы могут быть инстанцированы. Основанием для этого является то, что ABC обеспечивает интерфейс, который должны соблюдать все последующие подклассы DataHandler, обеспечивая тем самым совместимость с другими классами, которые с ними общаются.
Мы используемметаклассКроме того, мы используем декоратор @abstractmethod, чтобы Python знал, что метод будет заменен в подклассах (это идентично чистому виртуальному методу в C++).
Два интересных метода - 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()")
С указанием DataHandler ABC следующим шагом является создание обработчика для исторических файлов CSV. В частности, HistoricCSVDataHandler возьмет несколько файлов CSV, по одному для каждого символа, и преобразует их в словарь панды DataFrames.
Обработчик данных требует нескольких параметров, а именно очереди событий, на которую можно нажать информацию 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 является то, что индексы всех отслеживаемых символов могут быть объединены вместе. Это позволяет прокладывать недостающие точки данных вперед, назад или интерполировать в этих пробелах, чтобы можно было сравнивать тикеры на баре с баром. Это необходимо, например, для стратегий обратного отсчета среднего значения. Обратите внимание на использование методов союза и повторного индекса при объединении индексов для всех символов:
# 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]])
Первым абстрактным методом из DataHandler, который будет реализован, является get_latest_bars. Этот метод просто предоставляет список последних N строк из структуры latest_symbol_data. Настройка 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, который добавляется в очередь, когда он добавляет последние строки к последним_символам_данным:
# 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, который используется остальными компонентами для отслеживания рыночных данных.
В следующей статье мы рассмотрим иерархию класса Strategy и опишем, как стратегия может быть разработана для обработки нескольких символов, тем самым генерируя несколько SignalEvents для объекта Portfolio.