Siamo ora in grado di creare la classe gerarchica Backtest. L’oggetto Backtest incapsula la logica di gestione degli eventi ed essenzialmente lega insieme tutte le altre classi che abbiamo descritto negli articoli precedenti.
L’oggetto Backtest è progettato per eseguire un sistema guidato da eventi annidati in un ciclo while per gestire gli eventi inseriti nell’oggetto EventQueue. Il ciclo while esterno è noto come il “heartbeat loop” e decide la risoluzione temporale del sistema di backtesting. In un live ambiente questo valore sarà un numero positivo, ad esempio 600 secondi (ogni dieci minuti). Così i dati di mercato e le posizioni verranno aggiornati solo in questo lasso di tempo.
Per il backtester qui descritto il “heartbeat” può essere impostato a zero, indipendentemente dalla frequenza della strategia, poiché i dati sono già disponibili in virtù del fatto che sono storici!
Il Motore di Backtest
Possiamo eseguire il backtest a qualsiasi velocità desideriamo, poiché il sistema guidato dagli eventi è agnostico dalla disponibilità temporale dei dati storici, a condizione che abbiano un timestamp associato. Quindi ho solo incluso per dimostrare come funzionerebbe un motore di trading live. Il ciclo esterno così termina una volta che DataHandler lo comunica all’oggetto Backtest, utilizzando un attributo booleano continue_backtest.
Il ciclo while interno elabora effettivamente i segnali e li invia al componente corretto a seconda del tipo di evento. Pertanto, la coda degli eventi viene continuamente popolata e spopolata da eventi. Questo è ciò che significa avere un sistema guidato dagli eventi.
Il primo compito è importare le librerie necessarie. Importiamo pprint (“pretty-print”), perché vogliamo visualizzare le statistiche in modo semplice per l’output:
# backtest.py
import datetime
import pprint
import queue
import time
# backtest.py
class Backtest(object):
"""
Racchiude le impostazioni e i componenti per l'esecuzione
un backtest basato sugli eventi.
"""
def __init__(self, csv_dir, symbol_list, initial_capital,
heartbeat, start_date, data_handler,
execution_handler, portfolio, strategy ):
"""
Inizializza il backtest.
Parametri:
csv_dir - Il percorso della directory dei dati CSV.
symbol_list - L'elenco dei simboli.
intial_capital - Il capitale iniziale del portafoglio.
heartbeat - il "battito cardiaco" del backtest in secondi
data_inizio - La data e ora di inizio della strategia.
data_handler - (Classe) Gestisce il feed di dati di mercato.
execution_handler - (Classe) Gestisce gli ordini / esecuzioni per i trade.
portfolio - (Classe) Tiene traccia del portafoglio attuale e delle posizioni precedenti.
strategy - (Classe) Genera segnali basati sui dati di mercato.
"""
self.csv_dir = csv_dir
self.symbol_list = symbol_list
self.initial_capital = initial_capital
self.heartbeat = heartbeat
self.start_date = start_date
self.data_handler_cls = data_handler
self.execution_handler_cls = execution_handler
self.portfolio_cls = portfolio
self.strategy_cls = strategy
self.events = queue.Queue()
self.signals = 0
self.orders = 0
self.fills = 0
self.num_strats = 1
self._generate_trading_instances()
_generate_trading_instances, collega tutti gli oggetti di trading (Data- Handler, Strategy, Portfolio and ExecutionHandler) a vari componenti interni:
# backtest.py
def _generate_trading_instances(self):
"""
Genera le istanze degli componenti del backtest a partire dalle loro classi.
"""
print("Creating DataHandler, Strategy, Portfolio and ExecutionHandler")
self.data_handler = self.data_handler_cls(self.events,
self.csv_dir,
self.symbol_list)
self.strategy = self.strategy_cls(self.data_handler,
self.events)
self.portfolio = self.portfolio_cls(self.data_handler,
self.events,
self.start_date,
self.initial_capital)
self.execution_handler = self.execution_handler_cls(self.events)
Il metodo _run_backtest è dove viene effettuata la gestione di segnali all’interno del motore di backtest.
Come descritto negli articoli precedenti, ci sono due cicli while, uno annidato all’interno dell’altro.
Il ciclo esterno tiene traccia del “battito” del sistema, mentre il ciclo interno controlla se c’è un evento in coda e agisce su di esso chiamando il metodo appropriato sull’oggetto necessario.
Per un MarketEvent, si chiede all’oggetto Strategy di ricalcolare nuovi segnali, mentre all’ oggetto Portfolio
si chiede di reindicizzare l’ora.
Se si riceve un SignalEvent, si comunica al Portfolio di gestire il nuovo segnale e convertirlo in un insieme di OrderEvents, se necessario.
Nel caso di ricezione di un OrderEvent, si invia l’ordine all’ExecutionHandler in modo che venga trasmesso al broker (se è attivo il live trading).
Infine, se viene ricevuto un “FillEvent”, si aggiorna il Portfolio in modo da essere a allineato con le nuove posizioni:
# backtest.py
def _run_backtest(self):
"""
Esecuzione del backtest.
"""
i = 0
while True:
i += 1
print(i)
# Aggiornamento dei dati di mercato
if self.data_handler.continue_backtest == True:
self.data_handler.update_bars()
else:
break
# Gestione degli eventi
while True:
try:
event = self.events.get(False)
except queue.Empty:
break
else:
if event is not None:
if event.type == 'MARKET':
self.strategy.calculate_signals(event)
self.portfolio.update_timeindex(event)
elif event.type == 'SIGNAL':
self.signals += 1
self.portfolio.update_signal(event)
elif event.type == 'ORDER':
self.orders += 1
self.execution_handler.execute_order(event)
elif event.type == 'FILL':
self.fills += 1
self.portfolio.update_fill(event)
time.sleep(self.heartbeat)
Viene creata la curva di equity dal Dataframe pandas e vengono visualizzate le statistiche di riepilogo, così come il conteggio di Segnali, Ordini ed Eseguiti:
# backtest.py
def _output_performance(self):
"""
Stampa delle performance della strategia dai risultati del backtest.
"""
self.portfolio.create_equity_curve_dataframe()
print("Creating summary stats...")
stats = self.portfolio.output_summary_stats()
print("Creating equity curve...")
print(self.portfolio.equity_curve.tail(10))
pprint.pprint(stats)
print("Signals: %s" % self.signals)
print("Orders: %s" % self.orders)
print("Fills: %s" % self.fills)
simulate_trading. Esso richiama semplicemente in ordine i 2 metodi descritti precedentemente:
# backtest.py
def simulate_trading(self):
"""
Simula il backtest e stampa le performance del portafoglio.
"""
self._run_backtest()
self._output_performance()
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
