Dans les deux articles précédents de la série, nous avons discuté de ce qu'est un système de backtesting basé sur des événements et de la hiérarchie de classe pour l'objet Event.
L'un de nos objectifs avec un système de négociation basé sur des événements est de minimiser la duplication du code entre l'élément de backtesting et l'élément d'exécution en direct. Idéalement, il serait optimal d'utiliser la même méthodologie de génération de signal et les mêmes composants de gestion de portefeuille pour les tests historiques et le trading en direct. Pour que cela fonctionne, l'objet Stratégie qui génère les signaux et l'objet Portfolio qui fournit des ordres basés sur eux doivent utiliser une interface identique à un flux de marché pour l'exécution historique et en direct.
Cela motive le concept d'une hiérarchie de classes basée sur un objet DataHandler, qui donne à toutes les sous-classes une interface pour fournir des données de marché aux composants restants du système.
Des sous-classes d'exemples spécifiques pourraient inclure HistoricCSVDataHandler, QuandlDataHandler, SecuritiesMasterDataHandler, InteractiveBrokersMarketFeedDataHandler, etc. Dans ce tutoriel, nous allons seulement considérer la création d'un gestionnaire de données CSV historique, qui chargera les données CSV intradiennes pour les actions dans un ensemble de barres Open-Low-High-Close-Volume-OpenInterest. Cela peut ensuite être utilisé pour
La première tâche est d'importer les bibliothèques nécessaires. spécifiquement nous allons importer les pandas et les outils de classe de base abstraits. puisque le DataHandler génère MarketEvents nous devons également importerevent.pycomme décrit dans le tutoriel précédent:
# data.py
import datetime
import os, os.path
import pandas as pd
de l'abc à l'import ABCMeta, méthode abstraite
depuis l'importation d'événement MarketEvent Le DataHandler est une classe de base abstraite (ABC), ce qui signifie qu'il est impossible d'instancier une instance directement. Seules les sous-classes peuvent être instanciées. La raison en est que le ABC fournit une interface à laquelle toutes les sous-classes DataHandler suivantes doivent adhérer, assurant ainsi la compatibilité avec d'autres classes qui communiquent avec elles.
Nous faisons usage de laméta-classeEn outre, nous utilisons le décorateur @abstractmethod pour faire savoir à Python que la méthode sera écrasée dans les sous-classes (c'est identique à une méthode virtuelle pure en C++).
Les deux méthodes d'intérêt sont get_latest_bars et update_bars. La première renvoie les dernières N barres de l'horodatage actuel du rythme cardiaque, ce qui est utile pour les calculs en rotation nécessaires dans les classes de stratégie. La dernière méthode fournit un mécanisme de flux de goutte à goutte pour placer des informations de barre sur une nouvelle structure de données qui interdit strictement le biais de la tête de recherche. Notez que des exceptions seront soulevées si une tentative d'instantiation de la classe se produit:
# 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()")
Avec le DataHandler ABC spécifié, l'étape suivante consiste à créer un gestionnaire pour les fichiers CSV historiques.
Le gestionnaire de données nécessite quelques paramètres, à savoir une file d'attente d'événements sur laquelle pousser les informations MarketEvent, le chemin absolu des fichiers CSV et une liste de symboles.
# 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()
Il tente implicitement d'ouvrir les fichiers avec le format de
L'un des avantages de l'utilisation de pandas comme entrepôt de données en interne dans le HistoricCSVDataHandler est que les indices de tous les symboles suivis peuvent être fusionnés. Cela permet de rembourser les points de données manquants vers l'avant, vers l'arrière ou d'interpoler dans ces espaces afin que les tickers puissent être comparés sur une base barre à barre. Ceci est nécessaire pour les stratégies de renversement de la moyenne, par exemple. Notez l'utilisation des méthodes d'union et de réindexation lors de la combinaison des indices pour tous les symboles:
# 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()
La méthode _get_new_bar crée un générateur pour fournir une version formatée des données de barre. Cela signifie que les appels ultérieurs à la méthode donneront une nouvelle barre jusqu'à ce que la fin des données de symbole soit atteinte:
# 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]])
La première méthode abstraite de DataHandler à implémenter est get_latest_bars. Cette méthode fournit simplement une liste des dernières N barres de la structure de données 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:]
La dernière méthode, update_bars, est la deuxième méthode abstraite de DataHandler. Elle génère simplement un MarketEvent qui est ajouté à la file d'attente lorsqu'elle ajoute les dernières barres aux dernières_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())
Ainsi, nous avons un objet dérivé de DataHandler, qui est utilisé par les autres composants pour suivre les données du marché.
Dans l'article suivant, nous examinerons la hiérarchie de la classe Strategy et décrirons comment une stratégie peut être conçue pour gérer plusieurs symboles, générant ainsi plusieurs SignalEvents pour l'objet Portfolio.