Nei due articoli precedenti della serie abbiamo introdotto i concetti base di un sistema di backtesting basato sugli eventi e la gerarchia di classi per l’oggetto Event. In questo articolo vediamo come vengono utilizzati i dati di mercato, sia in un contesto storico di backtesting sia per l’esecuzione del live trading.
Uno dei nostri obiettivi con un sistema di trading basato sugli eventi è di minimizzare la duplicazione del codice tra l’elemento di backtesting e l’elemento di esecuzione live. Idealmente, è ottimale utilizzare la stessa metodologia di generazione del segnale e le stesse componenti di gestione del portafoglio sia per i test storici che per trading reale. Affinché questo funzioni, l’oggetto Strategy, che genera i segnali, e l’oggetto Portfolio, che fornisce gli ordini basati su di essi, devono utilizzare un’identica interfaccia verso un feed di dati finanziari, sia per la versione di backtesting che per quella live.
I Dati di Mercato
Questo requisito motiva la necessità di una gerarchia di classi basata sull’oggetto DataHandler, che l’implementa un’interfaccia, disponibile a tutte le sottoclassi, per fornire i dati di mercato alle rimanenti componenti del sistema. In questo modo, si può intercambiare qualsiasi sottoclasse di “fornitura” di dati finanziari senza influenzare la strategia o il calcolo del portafoglio.
Esempi di sottoclassi specifiche possono includere HistoricCSVDataHandler, QuandlDataHandler, SecuritiesMasterDataHandler, InteractiveBrokersMarketFeedDataHandler ecc. In questo tutorial descriviamo solamente la creazione di un gestore CSV di dati storici, che caricherà da un CSV i dati intraday per le azioni nel formato Open-Low-High-Close- Volume-OpenInterest. Questo può quindi essere usato per alimentare con i dati “candela-per-candela” le classi Strategy e Portfolio per ogni heartbeat (o impulso) del sistema, evitando così i bias di look-ahead.
Il primo compito è importare le librerie necessarie. Nello specifico, si includono Pandas e gli strumenti astratti della classe base. Dato che DataHandler genera MarketEvents, si importa anche event.py come descritto nel tutorial precedente:
# data.py
import datetime
import os, os.path
import pandas as pd
from abc import ABCMeta, abstractmethod
from event import MarketEvent
La classe DataHandler è una classe base astratta (ABC), cioè è impossibile istanziare direttamente un’istanza. Possono essere istanziate solamente le sottoclassi. Con questo approccio la classe ABC fornisce un’interfaccia che tutte le successive sottoclassi di DataHandler devono rispettare, garantendo in tal modo la compatibilità con altre classi che comunicano con esse.
Facciamo uso della proprietà __metaclass__ per far sapere a Python che questa è una classe ABC. Inoltre usiamo il decoratore @abstractmethod per far sapere a Python che il metodo verrà sovrascritto dalle sottoclassi (questo è identico a un metodo virtuale puro di C++).
I due metodi fondamentali sono get_latest_bars e update_bars. Il primo restituisce le ultime barre N a partire dal timestamp dall’attuale “impulso”, necessarie per far eseguire le elaborazioni previste nelle classi Strategy. Il secondo metodo fornisce un meccanismo di “alimentazione a goccia” per posizionare le informazioni OLHCV su una nuova struttura dati in modo da evitare la distorsione lookahead. Si noti che verranno sollevate eccezioni se si verifica un tentativo di istanziazione della classe:
# data.py
class DataHandler(object):
"""
DataHandler è una classe base astratta che fornisce un'interfaccia per
tutti i successivi gestori di dati (ereditati) (sia live che storici).
L'obiettivo di un oggetto (derivato da) DataHandler è generare un
set di barre (OLHCVI) per ogni simbolo richiesto.
Questo replicherà il modo in cui una strategia live funzionerebbe quando nuovi
i dati di mercato sarebbero inviati "giù per il tubo". Questo permette a sistemi
live e a sistemi con dati storici di essere trattati allo stesso modo dal resto
della suite di backtest.
"""
__metaclass__ = ABCMeta
@abstractmethod
def get_latest_bar(self, symbol):
"""
Restituisce l'ultima barra dalla lista latest_symbol.
"""
raise NotImplementedError("Should implement get_latest_bar()")
@abstractmethod
def get_latest_bars(self, symbol, N=1):
"""
Restituisce le ultime N barre dalla lista di barre
per il simbolo, o meno se sono disponibili poche barre
"""
raise NotImplementedError("Should implement get_latest_bars()")
def get_latest_bar(self, symbol):
"""
Restituisce l'ultima barra dalla lista latest_symbol.
"""
raise NotImplementedError("Should implement get_latest_bar()")
@abstractmethod
def get_latest_bar_datetime(self, symbol):
"""
Restituisce un oggetto datetime di Python per l'ultima barra.
"""
raise NotImplementedError("Should implement get_latest_bar_datetime()")\
@abstractmethod
def get_latest_bar_value(self, symbol, val_type):
"""
Restituisce un elemento tra Open, High, Low, Close, Volume o Adj_Close
from the last bar.
"""
raise NotImplementedError("Should implement get_latest_bar_value()")
@abstractmethod
def get_latest_bars_values(self, symbol, val_type, N=1):
"""
Restituisce i valori delle ultime N barre dalla lista
latest_symbol, o N-k se non meno disponibili.
"""
raise NotImplementedError("Should implement get_latest_bars_values()")
@abstractmethod
def update_bars(self):
"""
Inserisce la barra più recente nella struttura delle barre per
tutti i simboli della lista di simboli.
"""
raise NotImplementedError("Should implement update_bars()")
Dopo aver definito la classe DataHandler, il passo successivo è creare un gestore per i file CSV di dati storici. In particolare, HistoricCSVDataHandler prenderà più file CSV, uno per ciascun simbolo, e li convertirà in un DataFrame di Panda.
Il gestore dati richiede alcuni parametri, ovvero una coda di eventi su cui inviare informazioni di MarketEvent, il percorso assoluto dei file CSV e un elenco di simboli.
Di seguito l’inizializzazione della classe:
# data.py
class HistoricCSVDataHandler(DataHandler):
"""
HistoricCSVDataHandler è progettato per leggere dal disco
fisso un file CSV per ogni simbolo richiesto e fornire
un'interfaccia per ottenere la barra "più recente" in un
modo identico a un'interfaccia di live trading.
"""
def __init__(self, events, csv_dir, symbol_list):
"""
Inizializza il gestore dei dati storici richiedendo
la posizione dei file CSV e un elenco di simboli.
Si presume che tutti i file abbiano la forma
"symbol.csv", dove symbol è una stringa dell'elenco.
Parametri:
events - la coda degli eventi.
csv_dir - percorso assoluto della directory dei file CSV.
symbol_list - Un elenco di stringhe di simboli.
"""
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()
Questa funzione prevede quindi di aprire i file nel formato “SYMBOL.csv” dove il SYMBOL è il simbolo del ticker. Il formato dei file corrisponde a quello fornito da DTN IQFeed, ma si può facilmente modificare per gestire formati di dati aggiuntivi. L’apertura dei file è gestita dal seguente metodo _open_convert_csv_files.
Uno dei vantaggi dell’uso della libreria Pandas come archivio all’interno di HistoricCSVDataHandler è la possibilità di unire gli indici di tutti i simboli tracciati. Ciò consente di correggere i punti di dati mancanti in avanti, indietro o interpolati all’interno di questi spazi, in modo tale che i ticker possano essere confrontati “candela-per-candela”. Questo è necessario, ad esempio, per strategie di mean-reverting. Si noti l’uso dei metodi union e reindex quando si combina gli indici di tutti i simboli:
# data.py
def _open_convert_csv_files(self):
"""
Apre i file CSV dalla directory dei dati, convertendoli
in DataFrame pandas all'interno di un dizionario di simboli.
Per questo gestore si assumerà che i dati siano
tratto da DTN IQFeed. Così il suo formato sarà rispettato.
"""
comb_index = None
for s in self.symbol_list:
# Carica il file CSV senza nomi delle colonne, indicizzati per data
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','adj_close']
)
# Combina l'indice per riempire i valori successivi
if comb_index is None:
comb_index = self.symbol_data[s].index
else:
comb_index.union(self.symbol_data[s].index)
# Imposta il più recente symbol_data a None
self.latest_symbol_data[s] = []
# Indicizza nuovamente i dataframes
for s in self.symbol_list:
self.symbol_data[s] = self.symbol_data[s].reindex(index=comb_index, method='pad').iterrows()
Il metodo _get_new_bar crea un generatore python per fornire una versione formattata dei dati OLCHV. Questo significa che le successive chiamate al metodo genereranno una nuova barra fino al raggiungimento della fine dei dati del simbolo:
# data.py
def _get_new_bar(self, symbol):
"""
Restituisce l'ultima barra dal feed di dati come una tupla di
(sybmbol, datetime, open, low, high, close, volume).
"""
for b in self.symbol_data[symbol]:
yield b
Di seguito l’implementazione dei metodi astratti di DataHandler. Questi metodi forniscono varie forme di accesso alle barre acquisite. Dipende dalla fonte di acquisizione dati e dalla struttura dati in cui viene acquisita:
# data.py
def get_latest_bar(self, symbol):
"""
Restituisce l'ultima barra dalla lista latest_symbol.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list[-1]
def get_latest_bars(self, symbol, N=1):
"""
Restituisce le ultime N barre dall'elenco latest_symbol
o N-k se non sono tutte disponibili.
"""
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:]
def get_latest_bar_datetime(self, symbol):
"""
Restituisce un oggetto datetime di Python per l'ultima barra.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return bars_list[-1][0]
def get_latest_bar_value(self, symbol, val_type):
"""
Restituisce un elemento tra Open, High, Low, Close, Volume o Adj_Close
from the last bar.
"""
try:
bars_list = self.latest_symbol_data[symbol]
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return getattr(bars_list[-1][1], val_type)
def get_latest_bars_values(self, symbol, val_type, N=1):
"""
Restituisce i valori delle ultime N barre dalla lista
latest_symbol, o N-k se non meno disponibili.
"""
try:
bars_list = self.get_latest_bars(symbol, N)
except KeyError:
print("That symbol is not available in the historical data set.")
raise
else:
return np.array([getattr(b[1], val_type) for b in bars_list])
L’ultimo metodo astratto, update_bars, genera semplicemente un MarketEvent che viene aggiunto alla coda, e aggiunge le ultime barre a latest_symbol_data:
# data.py
def update_bars(self):
"""
Inserisce l'ultima barra nella struttura latest_symbol_data
per tutti i simboli nell'elenco dei simboli.
"""
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())
A questo punto abbiamo implementato un oggetto derivato da DataHandler, che viene utilizzato dai restanti componenti per tenere traccia dei dati di mercato. Gli oggetti Strategy, Portfolio ed ExecutionHandler richiedono i dati di mercato aggiornati, quindi ha senso centralizzare questa gestione al fine di evitare la duplicazione del codice e di possibili bug.
Nel prossimo articolo vedremo la gerarchia della classe Strategy e descriviamo come una strategia può essere progettata per gestire più simboli, generando così più SignalEvents per l’oggetto Portfolio.
Per il codice completo riportato in questo articolo, utilizzando il modulo di backtesting event-driven DataBacktest si può consultare il seguente repository di github:
https://github.com/datatrading-info/DataBacktest
