In questo articolo spieghiamo come implementiamo il backtest di una strategia con il Filtro di Kalman usando il framework Python per il trading basato su eventi DataTrader. La strategia si fonda sui concetti matematici dei modelli spaziali statali e dei filtri di Kalman. Applichiamo questi concetti utilizzando la libreria Python pykalman su una coppia di ETF per regolare dinamicamente un rapporto di copertura come base per una strategia di tipo mean-reverting.
La strategia che vogliamo implementare è stata introdotta da Ernest Chan (2012) [1] e testata da Aidan O’Mahony su Quantopian [2]. Utilizziamo il nostro framework open source DataTrader, sviluppato in Python, per implementare questa strategia. Con DataTrader gestiamo tracciamento delle posizioni, gestione del portafoglio e acquisizione dei dati, mentre ci concentriamo esclusivamente sul codice che genera i segnali di trading.
La strategia con il filtro di Kalman
Applichiamo una strategia di pairs-trading a una coppia di Exchange Traded Funds (ETF) che replicano i rendimenti delle obbligazioni del Tesoro USA di durata diversa. I due ETF sono:
Il nostro obiettivo consiste nel costruire una strategia di ritorno alla media usando questa coppia di ETF come strumenti sottostanti.
Lo “spread” sintetico tra TLT e IEI rappresenta la serie storica su cui vogliamo operare, andando long o short in modo sistematico. Utilizziamo il filtro di Kalman per tracciare dinamicamente il rapporto di copertura tra i due ETF, mantenendo stazionario lo spread (e quindi il comportamento mean-reverting).
Per stabilire regole di trading efficaci dobbiamo capire quando lo spread si discosta eccessivamente dal suo valore atteso. Come definiamo questo “troppo lontano”? Potremmo usare soglie fisse, ma dovremmo determinarle empiricamente. Questo introdurrebbe un ulteriore parametro da ottimizzare, aumentando il rischio di overfitting nel modello.
Adottiamo un approccio privo di parametri predefiniti, basandoci su multipli della deviazione standard dello spread da usare come soglie operative. Per semplicità impostiamo il coefficiente di moltiplicazione pari a uno.
Entriamo long sullo spread se l’errore di previsione scende sotto la deviazione standard negativa. Al contrario entriamo short se supera quella positiva. Le regole di uscita coincidono con l’opposto delle rispettive regole di entrata.
Il rapporto di copertura dinamico è dato da un componente del vettore degli stati nascosti al tempo t, \(\theta_t\), che chiamiamo \(\theta^0_t\). Questo rappresenta la pendenza “beta” ben nota nella regressione lineare.
Le regole della strategia
Quando andiamo long sullo spread acquistiamo N unità di TLT e vendiamo \(\lfloor{ \theta^0_t N \rfloor}\) unità di IEI, dove \(\lfloor{ x \rfloor}\) è il più grande intero non superiore a x. Questo arrotondamento è necessario poiché dobbiamo trattare un numero intero di quote ETF. L’operazione short è semplicemente l’opposto, con N che determina la dimensione totale della posizione.
Indichiamo con \(e_t\) l’errore di previsione (o errore residuo) al tempo t, mentre \(Q_t\) rappresenta la varianza di questa previsione nello stesso istante.
Nel dettaglio, la strategia si articola nelle seguenti regole:
- \(e_t \lt -\sqrt{Q_t}\) – Entrata Long: acquistiamo N azioni di TLT e vendiamo \(\lfloor{ \theta^0_t N \rfloor}\) azioni di IEI
- \(e_t \ge -\sqrt{Q_t}\) – Uscita Long: chiudiamo tutte le posizioni long su TLT e IEI
- \(e_t \gt \sqrt{Q_t}\) – Entrata Short: vendiamo N azioni di TLT e acquistiamo \(\lfloor{ \theta^0_t N \rfloor}\) azioni di IEI
- \(e_t \le \sqrt{Q_t}\) – Uscita Short: chiudiamo tutte le posizioni short su TLT e IEI
Il filtro di Kalman ci permette di calcolare \(\theta_t\), \(e_t\) e \(Q_t\). \(\theta_t\) rappresenta la pendenza e l’intercetta della regressione lineare tra TLT e IEI al tempo t, stimate dinamicamente. L’errore \(e_t = y_t – \hat{y}_t\) misura la differenza tra il valore attuale di TLT e la sua previsione secondo il filtro. \(Q_t\) rappresenta la varianza della previsione, mentre \(\sqrt{Q_t}\) è la sua deviazione standard.
La strategia prevede le seguenti fasi operative:
- Acquisiamo le barre giornaliere OHLCV per TLT e IEI.
- Usiamo il filtro di Kalman in modo ricorsivo per stimare il prezzo odierno di TLT partendo dalle osservazioni di IEI del giorno precedente.
- Calcoliamo la differenza tra la stima di Kalman e il valore effettivo di TLT, chiamata errore di previsione, per valutare la deviazione dello spread attuale dal valore atteso.
- Andiamo long se il valore si discosta troppo verso il basso, oppure short se si discosta troppo verso l’alto rispetto alla previsione.
- Chiudiamo le posizioni long e short quando il valore dello spread ritorna al suo valore previsto.
Scaricare i Dati
Per applicare questa strategia con il filtro di Kalman abbiamo bisogno dei dati OHLCV di prezzo per il periodo incluso nel backtest. In particolare dobbiamo scaricare i seguenti file:
- TLT – Per il periodo dal 3 agosto 2009 al 1 agosto 2016 (link)
- IEI – Per il periodo dal 3 agosto 2009 al 1 agosto 2016 (link).
Se vogliamo replicare i risultati, dobbiamo inserire questi dati nella directory indicata dal file di configurazione di DataTrader.
Implementazione Python con DataTrader
Poiché DataTrader gestisce il tracciamento delle posizioni, il portafoglio, i dati e gli ordini, scriviamo solo il codice relativo alla classe Strategy.
La classe Strategy comunica con il PortfolioHandler attraverso la coda degli eventi, usando oggetti SignalEvent. Dobbiamo anche importare la classe astratta base AbstractStrategy.
Ricordiamo che nella versione attuale di DataTrader dobbiamo importare la classe PriceParser. Questa classe ci consente di moltiplicare tutti i prezzi per un grande multiplo (\(10^8\)) per operare sempre con numeri interi nel tracciamento delle posizioni. In questo modo evitiamo problemi di arrotondamento in virgola mobile che si accumulano nei backtest estesi. Infine, dividiamo ogni risultato per PriceParser.PRICE_MULTIPLIER per ottenere il valore corretto.
from math import floor
import numpy as np
from datatrader.price_parser import PriceParser
from datatrader.event import (SignalEvent, EventType)
from datatrader.strategy.base import AbstractStrategy
Per testare la strategia con il filtro di Kalman dobbiamo creare la classe KalmanPairsTradingStrategy. Il compito di questa classe è determinare quando creare oggetti SignalEvent in base ai messaggi ricevuti dai dati di mercato BarEvent, cioè barre OHLCV giornaliere di TLT e IEI da Yahoo Finance.
Ci sono molti modi diversi per organizzare questa classe. Per facilitare la descrizione del codice, abbiamo codificato tutti i parametri della classe. In particolare abbiamo corretto il valore di \(\delta=10^{-4}\) e di \(v_t=10^{-3}\). Questi rappresentano il rumore del sistema e la varianza del rumore delle misurazioni nel modello del filtro di Kalman. Questi potrebbero anche essere implementati come argomenti di una parola chiave nel costruttore __init__ della classe. Un tale approccio consentirebbe una semplice ottimizzazione dei parametri.
Inizializzazione della Strategia
Il primo compito è impostare i parametri time e invested in modo che siano uguali a None, poiché verranno aggiornati man mano che i dati di mercato vengono accettati e vengono generati segnali di trading. latest_prices è una doppia matrice che contiene gli ultimi prezzi di TLT e IEI, utilizzata per comodità all’interno della classe.
Il successivo set di parametri si riferiscono al filtro di Kalman e sono spiegati in modo approfondito nel corso dedicato ai modelli quantitativi delle serie temporali. Il set finale di parametri include days, utilizzato per tracciare quanti giorni sono trascorsi, nonché qty e cur_hedge_qty, utilizzato per tracciare le quantità assolute di ETF da acquistare sia per il lato long che per quello short. In questo caso abbiamo impostato 2.000 unità su un patrimonio netto di 100.000 USD.
class KalmanPairsTradingStrategy(AbstractStrategy):
"""
Requisiti:
tickers - Lista dei simboli dei ticker
events_queue - Manager del sistema della coda degli eventi
short_window - numero di barre per la moving average di breve periodo
long_window - numero di barre per la moving average di lungo periodo
"""
def __init__(
self, tickers, events_queue
):
self.tickers = tickers
self.events_queue = events_queue
self.time = None
self.latest_prices = np.array([-1.0, -1.0])
self.invested = None
self.delta = 1e-4
self.wt = self.delta / (1 - self.delta) * np.eye(2)
self.vt = 1e-3
self.theta = np.zeros(2)
self.P = np.zeros((2, 2))
self.R = None
self.C = None
self.days = 0
self.qty = 2000
self.cur_hedge_qty = self.qty
Il metodo successivo _set_correct_time_and_price è un metodo “di appoggio” utilizzato per garantire che il filtro Kalman abbia tutte le corrette informazioni sui prezzi disponibili al momento giusto. Ciò è necessario perché in un sistema di backtest basato su eventi come DataTrader le informazioni sul mercato arrivano in sequenza.
Gestione degli eventi
Potremmo trovarci nello scenario dove in giornata K abbiamo ricevuto un prezzo per IEI, ma non TFT. Quindi dobbiamo aspettare fino a che entrambi gli eventi Market di TFT e IEI siano entranti nel ciclo di backtest, attraverso la coda degli eventi. Nel trading dal vivo questo non è un problema poiché questi eventi arriveranno quasi rispetto al periodo di trading di pochi giorni. Tuttavia, in un backtest basato sugli eventi, dobbiamo attendere l’arrivo di entrambi i prezzi prima di aggiornare i valori del filtro di Kalman.
Il codice controlla essenzialmente se l’evento successivo è relativo al giorno corrente. In tal caso, il prezzo corretto viene aggiunto all’elenco latest_price di TLT e IEI. Se abbiamo un nuovo giorno, si ripristina l’elenco latest_price e i prezzi corretti sono di nuovo aggiunti.
In futuro questo tipo di metodo di “pulizia” sarà probabilmente inserito nel codice base di DataTrader, riducendo la necessità di scrivere codice “boilerplate”, ma per ora dobbiamo prevederlo all’interno della nostra strategia.
def _set_correct_time_and_price(self, event):
"""
Impostazione del corretto prezzo e timestamp dell'evento
estratto in ordine dalla coda degli eventi.
"""
# Impostazione della prima istanza di time
if self.time is None:
self.time = event.time
# Correzione degli ultimi prezzi, che dipendono dall'ordine in cui
# arrivano gli eventi delle barre di mercato
price = event.adj_close_price/PriceParser.PRICE_MULTIPLIER
if event.time == self.time:
if event.ticker == self.tickers[0]:
self.latest_prices[0] = price
else:
self.latest_prices[1] = price
else:
self.time = event.time
self.days += 1
self.latest_prices = np.array([-1.0, -1.0])
if event.ticker == self.tickers[0]:
self.latest_prices[0] = price
else:
self.latest_prices[1] = price
Calcolo dei segnali
Il nucleo della strategia con il filtro di Kalman risiede nel metodo calculate_signals. Per prima cosa impostiamo i tempi e i prezzi corretti (come descritto in precedenza). Quindi controlliamo di avere entrambi i prezzi di TLT e IEI, a quel punto possiamo considerare nuovi segnali di trading.
y è impostato pari all’ultimo prezzo di IEI, mentre F è la matrice di osservazione contenente l’ultimo prezzo di TLT, nonché come variabile unitaria per rappresentare l’intercetta della regressione lineare. Il filtro di Kalman viene successivamente aggiornato con questi ultimi prezzi. Infine calcoliamo l’errore di previsione \(e_t\) e la deviazione standard delle previsioni, \(\sqrt{Q_t}\). Esaminiamo questo codice passo dopo passo, poiché sembra un po’ complicato.
Il primo compito è calcolare il valore scalare y e la matrice di osservazione F, che contengono rispettivamente i prezzi di IEI e e TLT. Calcoliamo la matrice varianza-covarianza R o la impostiamo come una matrice di zeri nel caso non sia stata ancora inizializzata. Successivamente calcoliamo la nuova previsione per l’osservazione yhat così come l’errore di previsione et.
Calcoliamo quindi la varianza delle previsioni di osservazione Qt e la deviazione standard sqrt_Qt. Usiamo le regole di aggiornamento descritte dal modello dello spazio degli stati per ottenere la distribuzione a posteriori degli stati theta, che contiene il rapporto di copertura / pendenza tra i due prezzi:
def calculate_signals(self, event):
"""
Calculo dei segnali della stategia con il filtro di Kalman.
"""
if event.type == EventType.BAR:
self._set_correct_time_and_price(event)
# Opera solo se abbiamo entrambe le osservazioni
if all(self.latest_prices > -1.0):
# Creare la matrice di osservazione degli ultimi prezzi di
# TLT e il valore dell'intercetta nonché il
# valore scalare dell'ultimo prezzo di IEI
F = np.asarray([self.latest_prices[0], 1.0]).reshape((1, 2))
y = self.latest_prices[1]
# Il valore a priori degli stati \theta_t è una distribuzione
# gaussiana multivariata con media a_t e varianza-covarianza R_t
if self.R is not None:
self.R = self.C + self.wt
else:
self.R = np.zeros((2, 2))
# Calcola l'aggiornamento del filtro di Kalman
# ----------------------------------
# Calcola la previsione di una nuova osservazione
# e il relativo errore di previsione
yhat = F.dot(self.theta)
et = y - yhat
# Q_t è la varianza della previsione delle osservazioni
# e sqrt{Q_t} è la deviazione standard delle previsioni
Qt = F.dot(self.R).dot(F.T) + self.vt
sqrt_Qt = np.sqrt(Qt)
# Il valore a posteriori degli stati \theta_t ha una
# distribuzione gaussiana multivariata con
# media m_t e varianza-covarianza C_t
At = self.R.dot(F.T) / Qt
self.theta = self.theta + At.flatten() * et
self.C = self.R - At * F.dot(self.R)
Infine generiamo i segnali di trading in base ai valori di \(e_t\) e \(\sqrt{Q_t}\). Per fare ciò dobbiamo verificare qual è lo stato delle posizioni – “long”, “short” o “nessuna”.
Da notare che regolare l’attuale quantità di copertura cur_hedge_qty quando andiamo long o short come come la pendenza \(\theta^0_t\) si adegua costantemente nel tempo:
..
# Opera solo se i giorni sono maggiori del
# periodo di "riscaldamento"
if self.days > 1:
# Se non siamo a mercato...
if self.invested is None:
if et < -sqrt_Qt:
# Entrata Long
print("LONG: %s" % event.time)
self.cur_hedge_qty = int(floor(self.qty*self.theta[0]))
self.events_queue.put(SignalEvent(self.tickers[1], "BOT", self.qty))
self.events_queue.put(SignalEvent(self.tickers[0], "SLD", self.cur_hedge_qty))
self.invested = "long"
elif et > sqrt_Qt:
# Entrata Short
print("SHORT: %s" % event.time)
self.cur_hedge_qty = int(floor(self.qty*self.theta[0]))
self.events_queue.put(SignalEvent(self.tickers[1], "SLD", self.qty))
self.events_queue.put(SignalEvent(self.tickers[0], "BOT", self.cur_hedge_qty))
self.invested = "short"
# Se siamo a mercato...
if self.invested is not None:
if self.invested == "long" and et > -sqrt_Qt:
print("CLOSING LONG: %s" % event.time)
self.events_queue.put(SignalEvent(self.tickers[1], "SLD", self.qty))
self.events_queue.put(SignalEvent(self.tickers[0], "BOT", self.cur_hedge_qty))
self.invested = None
elif self.invested == "short" and et < sqrt_Qt:
print("CLOSING SHORT: %s" % event.time)
self.events_queue.put(SignalEvent(self.tickers[1], "BOT", self.qty))
self.events_queue.put(SignalEvent(self.tickers[0], "SLD", self.cur_hedge_qty))
self.invested = None
Esecuzione del Backtest
Questo è tutto il codice necessario per l’oggetto Strategy. Abbiamo anche bisogno di creare un file di backtest per incapsulare tutta la logica della strategia con il filtro di Kalman e la scelta delle classi da usare. Questa specifica versione è molto simile a quelle presenti nella directory examples e sostituisce il capitale di 500.000 USD con 100.000 USD.
Si sostituisce inoltre il FixedPositionSizer con il NaivePositionSizer. Quest’ultimo viene utilizzato per accettare “ingenuamente” le quantità assolute di quote ETF da negoziare così come suggerite nella classe KalmanPairsTradingStrategy. In un ambiente reale sarebbe necessario adeguarlo a seconda degli obiettivi di gestione del rischio del portafoglio.
Ecco il codice completo per kalman_datatrader_backtest.py:
import datetime
from datatrader import settings
from datatrader.compat import queue
from datatrader.price_parser import PriceParser
from datatrader.price_handler.yahoo_daily_csv_bar import YahooDailyCsvBarPriceHandler
from datatrader.strategy.base import Strategies # DisplayStrategy
from datatrader.position_sizer.naive import NaivePositionSizer
from datatrader.risk_manager.example import ExampleRiskManager
from datatrader.portfolio_handler import PortfolioHandler
from datatrader.compliance.example import ExampleCompliance
from datatrader.execution_handler.ib_simulated import IBSimulatedExecutionHandler
from datatrader.statistics.tearsheet import TearsheetStatistics
from datatrader.trading_session import TradingSession
from kalman_datatrader_strategy import KalmanPairsTradingStrategy
def run(config, testing, tickers, filename):
# Impostazione delle variabili necessarie per il backtest
# Informazioni sul Backtest
title = ['Kalman Filter Example on TLT and IEI: 100x300']
initial_equity = 100000.00
start_date = datetime.datetime(2009, 8, 1)
end_date = datetime.datetime(2016, 8, 1)
events_queue = queue.Queue()
# Uso del Manager dei Prezzi di Yahoo Daily
price_handler = YahooDailyCsvBarPriceHandler(config.CSV_DATA_DIR, events_queue,
tickers, start_date=start_date, end_date=end_date)
# Uso della strategia KalmanPairsTrading
strategy = KalmanPairsTradingStrategy(tickers, events_queue)
strategy = Strategies(strategy)
# Uso di un Position Sizer standard
position_sizer = NaivePositionSizer()
# Uso di Manager di Risk di esempio
risk_manager = ExampleRiskManager()
# Use del Manager di Portfolio di default
portfolio_handler = PortfolioHandler(
PriceParser.parse(initial_equity), events_queue, price_handler,
position_sizer, risk_manager
)
# Uso del componente ExampleCompliance
compliance = ExampleCompliance(config)
# Uso un Manager di Esecuzione che simula IB
execution_handler = IBSimulatedExecutionHandler(
events_queue, price_handler, compliance
)
# Uso delle statistiche di default
statistics = TearsheetStatistics(
config, portfolio_handler, title=""
)
# Settaggio del backtest
backtest = TradingSession(
config, strategy, tickers,
initial_equity, start_date, end_date, events_queue,
price_handler=price_handler,
portfolio_handler=portfolio_handler,
compliance=compliance,
position_sizer=position_sizer,
execution_handler=execution_handler,
risk_manager=risk_manager,
statistics=statistics,
sentiment_handler=None,
title=title, benchmark=None
)
results = backtest.start_trading(testing=testing)
statistics.save(filename)
return results
def main(config, testing, tickers, filename):
tickers = tickers.split(",")
config = settings.from_file(config, testing)
run(config, testing, tickers, filename)
if __name__ == "__main__":
config = './datatrader.yml'
testing = False
tickers = 'TLT,IEI'
filename = None
main(config, testing, tickers, filename)
Se DataTrader è installato correttamente e i dati sono stati scaricati da Yahoo Finance, il codice può essere eseguito tramite il seguente comando da terminale:
$ python kalman_datatrader_backtest.py
Grazie ad alcuni miglioramenti, il codice è ottimizzato per lavorare con dati rapresentati come barre OHLCV ed esegue il backtesting in tempi più rapidi.
Risultati della strategia
Una delle funzionalità implementate in DataTrader è quella del “tearsheet” che permette la produzione di un report che raccoglie le statistiche e le performance della strategia.
Il tearsheet è usato principalmente come report “one pager” di una strategia di trading. La classe TearsheetStatistics nella base di codice DataTrader replica molte delle statistiche presenti in un tipico report sulle prestazioni della strategia.
I primi due grafici rappresentano rispettivamente la curva equity e la percentuale di drawdown. Sotto questo ci sono i pannelli delle prestazioni mensili e annuali. Infine vengono presentate la curva del patrimonio netto, le statistiche a livello di trade e temporali:
La curva di equity inizia relativamente piatta per il primo anno della strategia, ma si intensifica rapidamente durante il 2011. Durante il 2012 la strategia diventa significativamente più volatile rimanendo “sott’acqua” fino al 2015 e raggiungendo una percentuale massima di drawdown del 15,79%. La performance aumenta gradualmente dal drawdown alla fine del 2013 fino al 2016.
Questa strategia con il filtro di Kalman ha un CAGR dell’6,93% con uno Sharpe Ratio di 0,55. Ha anche una durata massima di drawdown di 923 giorni – quasi tre anni! Si noti che questa strategia viene eseguita al lordo dei costi di transazione, quindi la performance reale sarebbe probabilmente peggiore.
Conclusioni
Abbiamo descritto un primo approccio ad una strategia con filtro di Kalman per il pairs trading. E’ necessario molto lavoro di ricerca per trasformare questa in una strategia redditizia da implementarla in un trading dal vivo. Le potenziali aree di miglioramenti includono:
- Ottimizzazione dei parametri – Variazione dei parametri del filtro Kalman tramite la ricerca nella griglia di convalida incrociata o una qualche forma di ottimizzazione di apprendimento automatico. Tuttavia, questo introduce la netta possibilità di overfitting ai dati storici.
- Selezione degli asset – La scelta di coppie di ETF aggiuntive o alternative aiuterebbe ad aggiungere diversificazione al portafoglio, ma aumenta la complessità della strategia così come il numero di operazioni (e quindi i costi di transazione).
Nei prossimi articoli considereremo come eseguire queste procedure per varie strategie di trading.
Riferimenti
- [1] Chan, E. P. (2013) Algorithmic Trading: Winning Strategies and their Rationale, Wiley
- [2] O’Mahony, A. (2014) Ernie Chan’s EWA/EWC pair trade with Kalman filter, https://www.quantopian.com/posts/ernie-chans-ewa-slash-ewc-pair-trade-with-kalman-filter
Il codice completo presentato in questo articolo, basato sul framework di trading quantitativo event-driven DataTrader, è disponibile nel seguente repository GitHub: https://github.com/tradingquant-it/DataTrader.”
