Em dois artigos anteriores da série, discutimos o que é um sistema de backtesting orientado por eventos e a hierarquia de classes para o objeto Event.
Um dos nossos objetivos com um sistema de negociação baseado em eventos é minimizar a duplicação de código entre o elemento de backtesting e o elemento de execução ao vivo. Idealmente, seria ideal utilizar a mesma metodologia de geração de sinais e componentes de gerenciamento de carteira tanto para testes históricos quanto para negociação ao vivo. Para que isso funcione, o objeto Estratégia que gera os sinais e o objeto Portfólio que fornece ordens baseadas neles devem utilizar uma interface idêntica a um feed de mercado tanto para a execução histórica quanto para a execução ao vivo.
Isso motiva o conceito de uma hierarquia de classes baseada em um objeto DataHandler, que dá a todas as subclasses uma interface para fornecer dados de mercado aos componentes restantes dentro do sistema.
Exemplos específicos de subclasses podem incluir HistoricCSVDataHandler, QuandlDataHandler, SecuritiesMasterDataHandler, InteractiveBrokersMarketFeedDataHandler etc. Neste tutorial, vamos considerar apenas a criação de um processador de dados CSV histórico, que irá carregar dados CSV intradiários para ações em um conjunto de barras Open-Low-High-Close-Volume-OpenInterest. Isso pode então ser usado para
A primeira tarefa é importar as bibliotecas necessárias. Especificamente, vamos importar pandas e as ferramentas de classe base abstrata.event.pycomo descrito no tutorial anterior:
# data.py
import datetime
import os, os.path
import pandas as pd
de abc import ABCMeta, abstractmethod
de importação de evento MarketEvent O DataHandler é uma classe base abstrata (ABC), o que significa que é impossível instanciar uma instância diretamente. Apenas subclasses podem ser instanciadas. A razão para isso é que o ABC fornece uma interface que todas as subsequentes subclasses do DataHandler devem aderir, garantindo assim a compatibilidade com outras classes que se comunicam com elas.
Façamos uso dometaclasseAlém disso, usamos o decorador @abstractmethod para deixar Python saber que o método será substituído em subclasses (isso é idêntico a um método virtual puro em C++).
Os dois métodos de interesse são get_latest_bars e update_bars. O primeiro retorna as últimas N barras do carimbo de tempo do batimento cardíaco atual, o que é útil para cálculos rotativos necessários nas classes de Estratégia. O último método fornece um mecanismo de
# 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()")
Com o DataHandler ABC especificado, o próximo passo é criar um processador para arquivos CSV históricos.
O manipulador de dados requer alguns parâmetros, nomeadamente uma fila de eventos para a qual empurrar informações do MarketEvent, o caminho absoluto dos arquivos CSV e uma lista de símbolos.
# 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()
Ele tentará implícitamente abrir os arquivos com o formato de
Um dos benefícios de usar pandas como um armazém de dados internamente dentro do HistoricCSVDataHandler é que os índices de todos os símbolos a serem rastreados podem ser fundidos. Isso permite que os pontos de dados em falta sejam preenchidos para frente, para trás ou interpolados dentro dessas lacunas, de modo que os tickers possam ser comparados em uma base de barra a barra. Isso é necessário para estratégias de inversão da média, por exemplo. Observe o uso dos métodos de união e reindexação ao combinar os índices para todos os símbolos:
# 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()
O método _get_new_bar cria um gerador para fornecer uma versão formatada dos dados de barras. Isso significa que as chamadas subsequentes ao método produzirão uma nova barra até que o final dos dados do símbolo seja alcançado:
# 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]])
O primeiro método abstrato do DataHandler a ser implementado é get_latest_bars. Este método simplesmente fornece uma lista das últimas N barras da estrutura de dados latest_symbol_data.
# 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:]
O método final, update_bars, é o segundo método abstrato do DataHandler. Ele simplesmente gera um MarketEvent que é adicionado à fila ao anexar as últimas barras aos últimos_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())
Assim, temos um objeto derivado do DataHandler, que é usado pelos componentes restantes para acompanhar os dados de mercado. Os objetos Estratégia, Portfólio e ExecuçãoHandler exigem todos os dados de mercado atuais, portanto, faz sentido centralizá-los para evitar a duplicação de armazenamento.
No próximo artigo, consideraremos a hierarquia da classe Estratégia e descreveremos como uma estratégia pode ser projetada para lidar com múltiplos símbolos, gerando assim múltiplos eventos de sinal para o objeto Portfólio.