In den beiden vorangegangenen Artikeln der Serie haben wir diskutiert, was ein ereignisgesteuertes Backtesting-System ist und die Klassenhierarchie für das Event-Objekt.
Einer unserer Ziele bei einem ereignisgesteuerten Handelssystem ist es, die Kopplung von Code zwischen dem Backtesting-Element und dem Live-Ausführungselement zu minimieren. Idealerweise wäre es optimal, die gleiche Signalgenerierungsmethodik und Portfoliomanagement-Komponenten sowohl für historische Tests als auch für Live-Handel zu verwenden. Damit dies funktioniert, müssen das Strategieobjekt, das die Signale generiert, und das Portfolioobjekt, das auf ihnen basierende Aufträge bereitstellt, eine identische Schnittstelle zu einem Marktfeed für historische und Live-Running verwenden.
Dies motiviert das Konzept einer Klassenhierarchie, die auf einem DataHandler-Objekt basiert, das allen Unterklassen eine Schnittstelle zur Bereitstellung von Marktdaten für die verbleibenden Komponenten im System bietet. Auf diese Weise kann jeder Unterklassendatenbehandler
Spezifische Beispiel-Unterklassen könnten HistoricCSVDataHandler, QuandlDataHandler, SecuritiesMasterDataHandler, InteractiveBrokersMarketFeedDataHandler usw. sein. In diesem Tutorial werden wir nur die Erstellung eines historischen CSV-Daten-Handlers betrachten, der Intraday-CSV-Daten für Aktien in einem Open-Low-High-Close-Volume-Open-Interest-Set von Balken laden wird. Dies kann dann verwendet werden, um
Die erste Aufgabe besteht darin, die notwendigen Bibliotheken zu importieren. Insbesondere werden wir Pandas und die abstrakten Basisklassen-Tools importieren. Da der DataHandler MarketEvents generiert, müssen wir auch importierenevent.pywie im vorherigen Tutorial beschrieben:
# data.py
import datetime
import os, os.path
import pandas as pd
von abc import ABCMeta, abstrakte Methode
von Ereignisimport MarketEvent Der DataHandler ist eine abstrakte Basisklasse (ABC), was bedeutet, dass es unmöglich ist, eine Instanz direkt zu instanzieren. Nur Unterklassen können instanziert werden. Die Begründung dafür ist, dass der ABC eine Schnittstelle bietet, an die alle nachfolgenden DataHandler-Unterklassen sich halten müssen, wodurch die Kompatibilität mit anderen Klassen gewährleistet wird, die mit ihnen kommunizieren.
Wir nutzen dieMetaklasseAußerdem verwenden wir den @abstractmethod Dekorator, um Python mitzuteilen, dass die Methode in Unterklassen überschrieben wird (dies ist identisch mit einer reinen virtuellen Methode in C++).
Die beiden Methoden von Interesse sind get_latest_bars und update_bars. Die erstere gibt die letzten N Bars aus dem aktuellen Herzschlagzeitstempel zurück, was für rollende Berechnungen in Strategy-Klassen nützlich ist. Die letztere Methode bietet einen
# 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()")
Mit dem angegebenen DataHandler ABC ist der nächste Schritt, einen Handler für historische CSV-Dateien zu erstellen.
Der Datenverarbeiter benötigt einige Parameter, nämlich eine Ereigniswarteschlange, auf der MarketEvent-Informationen geschoben werden können, den absoluten Pfad der CSV-Dateien und eine Liste von Symbolen.
# 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()
Es wird implizit versuchen, die Dateien mit dem Format
Einer der Vorteile der Verwendung von Pandas als Datenspeicher intern im HistoricCSVDataHandler besteht darin, dass die Indizes aller zu verfolgenden Symbole miteinander zusammengeführt werden können. Dies ermöglicht es, fehlende Datenpunkte vorwärts, rückwärts oder interpoliert in diesen Lücken zu polstern, so dass Tickers auf einer Balkenbasis verglichen werden können. Dies ist beispielsweise für Mittelumkehrstrategien notwendig. Beachten Sie die Verwendung von Union und Reindex-Methoden bei der Kombination der Indizes für alle Symbole:
# 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()
Die _get_new_bar-Methode erstellt einen Generator, um eine formatierte Version der Balkendaten bereitzustellen. Dies bedeutet, dass nachfolgende Aufrufe der Methode eine neue Balke ergeben, bis das Ende der Symboldaten erreicht ist:
# 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]])
Die erste zu implementierende abstrakte Methode von DataHandler ist get_latest_bars. Diese Methode liefert einfach eine Liste der letzten N Balken aus der neuesten_symbol_data-Struktur.
# 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:]
Die letzte Methode, update_bars, ist die zweite abstrakte Methode von DataHandler. Sie erzeugt einfach ein MarketEvent, das der Warteschlange hinzugefügt wird, wenn die neuesten Balken an die neuesten_symbol_data angehängt werden:
# 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())
So haben wir ein von DataHandler abgeleitetes Objekt, das von den übrigen Komponenten verwendet wird, um Marktdaten zu verfolgen.
Im nächsten Artikel werden wir die Strategy-Klassen-Hierarchie betrachten und beschreiben, wie eine Strategie entworfen werden kann, um mehrere Symbole zu verarbeiten, wodurch mehrere SignalEvents für das Portfolio-Objekt generiert werden.