Backtesting-Event-Driven-Python-Trading-Algoritmico - Parte VIII

Motore di Backtesting con Python – Parte VIII (Backtest)

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
        
L’inizializzazione dell’oggetto Backtest richiede la directory CSV, l’elenco completo dei simboli da analizzare, il capitale iniziale, il periodo di “heartbreat” in millisecondi, la data e ora di inizio del backtest nonché degli oggetti DataHandler, ExecutionHandler, Portfolio e Strategy. Una coda viene utilizzata per contenere gli eventi. Vengono conteggiati i segnali, gli ordini e le esecuzioni:
            # 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()
        
Il primo metodo, _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)
        
Una volta completata la simulazione del backtest, è possibile visualizzare le prestazioni della strategia nel terminale / console python.
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)
        
L’ultimo metodo da implementare è il 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

Scroll to Top