Nella lezione sui modelli di Markov nascosti abbiamo descritto la loro applicazione per indicizzare i dati sui rendimenti come meccanismo per scoprire i “regimi di mercato” latenti. Abbiamo analizzato i rendimenti dell’S&P500 usando le librerie statistiche Python e identificato periodi di diversa volatilità, utilizzando sia modelli a due stati che a tre stati. In questo articolo, implementiamo la gestione del rischio con i modelli di Markov nascosti per identificare il regime di mercato, e effettuiamo il backtest di una strategia tramite il framework DataTrader.
L’obbiettivo consiste nel decidere di non effettuare operazioni quando prevediamo regimi di elevata volatilità. In questo modo vogliamo eliminare i trade non redditizi e, possibilmente, ridurre la volatilità della strategia, aumentando così il suo Sharpe ratio. Per raggiungere questo obiettivo abbiamo apportato alcune piccole modifiche al codice di DataTrader, disponibile nella sua pagina Github.
Abbiamo abbinato l’identificazione del regime di mercato a una strategia di trend-following a breve termine molto semplice, basata su regole di crossover della media mobile. La strategia in sé è relativamente irrilevante per questo articolo, poiché ci concentriamo soprattutto sull’implementazione della logica di gestione del rischio con i modelli di Markov nascosti, identificando il regime di mercato.
Poiché DataTrader è scritto in Python, per questo articolo utilizziamo una libreria Python che ci fornisce un’implementazione pronta all’uso del Modello di Markov Nascosto. La libreria che usiamo si chiama hmmlearn.
Gestione del rischio con i modelli di Markov nascosti
Se non conosci i modelli di Markov nascosti e/o non sai come possiamo usarli come strumento per la gestione del rischio, vale la pena consultare le seguenti lezioni del corso di finanza quantitativa:
- I Modelli di Markov nascosti nel trading quantitativo
- Regime di Mercato con i Modelli di Markov nascosti
Nella prima lezione spieghiamo i concetti matematici e statistici alla base del modello, mentre nel secondo utilizziamo Python per adattare un HMM ai rendimenti dell’S&P500.
I modelli di Markov nascosti rappresentano una classe di modelli stocastici dello spazio degli stati. Presuppongono l’esistenza di stati “nascosti” o “latenti” che non possiamo osservare direttamente. Tuttavia, questi stati influenzano i valori che osserviamo, chiamati osservazioni. Uno degli obiettivi principali del modello consiste nell’inferire lo stato attuale a partire dalle osservazioni note.
Nel trading quantitativo, traduciamo questo problema nell’identificazione di regimi di mercato latenti, come variazioni normative o periodi di alta volatilità. In questo contesto, consideriamo i rendimenti dei dati di mercato come osservazioni indirettamente influenzate dai regimi nascosti. Adattando un modello di Markov nascosto ai rendimenti, possiamo “prevedere” nuovi stati di regime, da usare come filtro nella gestione del rischio.
La strategia di trading
Abbiamo scelto una strategia di trading estremamente semplice per facilitarne la comprensione. L’aspetto centrale che vogliamo approfondire riguarda la gestione del rischio con i modelli di Markov nascosti.
La strategia di trend-following a breve termine che adottiamo è il classico crossover di medie mobili. Le regole sono le seguenti:
- A ogni barra calcoliamo le medie mobili semplici (SMA) a 10 e 30 giorni
- Se la SMA a 10 giorni supera la SMA a 30 giorni e non siamo a mercato, entriamo long
- Se la SMA a 30 giorni supera la SMA a 10 giorni e siamo a mercato, chiudiamo la posizione
Questa strategia, con questi parametri, non si rivela particolarmente efficace sui prezzi dell’indice S&P500. Otteniamo risultati simili a un approccio buy-and-hold sull’ETF SPY nello stesso periodo.
Tuttavia, se combiniamo la strategia con un filtro di gestione del rischio, otteniamo risultati migliori. Possiamo infatti eliminare i trade che si verificano in periodi di elevata volatilità, in cui le strategie di trend-following tendono a fallire.
Costruiamo il filtro addestrando un modello di Markov nascosto sui dati S&P500 dal 29 gennaio 1993 (data iniziale disponibile per SPY) al 31 dicembre 2004. Successivamente serializziamo il modello (tramite pickle di Python) e lo utilizziamo nella sottoclasse RiskManager di DataTrader.
Il gestore del rischio verifica, per ogni operazione inviata, se lo stato corrente corrisponde a un regime di bassa o alta volatilità. Se la volatilità è bassa, eseguiamo tutte le operazioni long. Se la volatilità è alta, chiudiamo qualsiasi operazione aperta al segnale di uscita e blocchiamo ogni nuovo trade long prima dell’esecuzione.
In questo modo otteniamo l’effetto desiderato: eliminare i trade trend-following in fasi di alta volatilità, quando è più probabile che risultino in perdita per via dell’errata identificazione del “trend”.
Set di Dati Storici
Effettuiamo il backtest della strategia dal 1 gennaio 2005 al 31 dicembre 2014, senza riaddestrare l’HMM. Questo implica che usiamo il modello in modalità out-of-sample, su dati diversi da quelli di addestramento.
Per testare la strategia di gestione del rischio con i modelli di Markov nascosti abbiamo bisogno dei dati OHLCV giornalieri dell’ETF SPY, sia per il periodo di addestramento dell’HMM che per il periodo di backtest:
| Ticker | Nome | Periodo |
|---|---|---|
| SPY | SPDR S&P 500 ETF | 29 gennaio 1993 – 31 dicembre 2014 |
Se vogliamo replicare i risultati, dobbiamo inserire questi dati nella directory indicata dal file delle impostazioni di DataTrader.
Implementazione Python
Calcolo dei rendimenti con DataTrader
Per effettuare la gestione del rischio con i modelli di Markov nascosti dobbiamo calcolare e memorizzare i rendimenti dei prezzi di chiusura dell’ETF SPY. Finora abbiamo memorizzato solo i prezzi. La sede naturale per archiviare i rendimenti è all’interno della sottoclasse PriceHandler. Perciò abbiamo aggiunto questa funzionalità al framework DataTrader.
Si è trattato di un aggiornamento relativamente semplice, composto da due modifiche principali. Per prima cosa abbiamo introdotto un flag booleano calc_adj_returns nell’inizializzazione della classe. Quando lo impostiamo su True, calcoliamo e memorizziamo i rendimenti. Il valore predefinito del flag è False, così minimizziamo l’impatto sul resto del codice client.
La seconda modifica consiste nel sovrascrivere il metodo “virtuale” _store_event presente nella classe AbstractBarPriceHandler con il nuovo codice nella classe YahooDailyCsvBarPriceHandler.
Il codice verifica se calc_adj_returns è uguale a True. In tal caso, memorizziamo i prezzi di chiusura attuali e precedenti tramite il PriceParser, calcoliamo i rendimenti percentuali e li aggiungiamo alla lista adj_close_returns. Il RegimeHMMRiskManager utilizza in seguito questa lista per prevedere lo stato corrente del regime.
def _store_event(self, event):
"""
Memorizza il prezzo di chiusura e di chiusura aggiustata per ogni evento
"""
ticker = event.ticker
# Se il flag calc_adj_returns è True, calcola e memorizza
# in un elenco tutta la lista dei rendimenti percentuali
# del prezzo di chiusura aggiustata
# TODO: Aumentare la velocità
if self.calc_adj_returns:
prev_adj_close = self.tickers[ticker]["adj_close"] / float(PriceParser.PRICE_MULTIPLIER)
cur_adj_close = event.adj_close_price / float(PriceParser.PRICE_MULTIPLIER)
self.tickers[ticker][
"adj_close_ret"
] = cur_adj_close / prev_adj_close - 1.0
self.adj_close_returns.append(self.tickers[ticker]["adj_close_ret"])
self.tickers[ticker]["close"] = event.close_price
self.tickers[ticker]["adj_close"] = event.adj_close_price
self.tickers[ticker]["timestamp"] = event.time
Questa modifica è disponibile nell’ultima versione di DataTrader, che (come sempre) possiamo trovare alla pagina Github.
Implementazione per identificare il regime
Ora concentriamo l’attenzione sull’implementazione del filtro per il regime di mercato e sulla strategia di trend following a breve termine che utilizziamo per eseguire il backtest. Per eseguire questa strategia, usiamo quattro differenti file. Riportiamo il listato completo di ciascun file alla fine di questo articolo. In questo modo chi desidera implementare un metodo simile può replicare direttamente i risultati.
Nel primo file adattiamo un modello gaussiano di Markov nascosto a un ampio periodo dei rendimenti dell’S&P500. Nel secondo inseriamo la logica per effettuare il trend-following di breve termine. Il terzo file fornisce il filtro del regime di mercato tramite un oggetto per la gestione del rischio. Nell’ultimo file colleghiamo insieme tutti questi moduli in una procedura di backtest per la gestione del rischio con i modelli di Markov.
Addestrare il modello di Markov nascosto
Prima di creare un filtro per rilevare il regime del mercato, adattiamo il modello di Markov nascosto a un dataset dei rendimenti. A tale scopo usiamo la libreria Python hmmlearn. La sua API è estremamente semplice e ci consente di adattare e memorizzare facilmente il modello per un uso successivo.
Iniziamo importando le librerie necessarie. Utilizziamo pickle per serializzare il modello da usare nel gestore del rischio per il rilevamento del regime. Con warnings eliminiamo gli avvisi di deprecazione generati da Scikit-Learn tramite le chiamate API di hmmlearn. Usiamo GaussianHMM di hmmlearn come base del nostro modello. Infine importiamo Matplotlib e Seaborn per visualizzare i grafici degli stati nascosti all’interno del campione, fondamentali per eseguire un “controllo di integrità” sul comportamento dei modelli.
# regime_hmm_train.py
import datetime
import pickle
import warnings
from hmmlearn.hmm import GaussianHMM
from matplotlib import cm, pyplot as plt
from matplotlib.dates import YearLocator, MonthLocator
import numpy as np
import pandas as pd
import seaborn as sns
La funzione obtain_prices_df legge il file CSV dei dati SPY scaricati da Yahoo Finance e importa i dati in un Pandas DataFrame. Quindi calcola i rendimenti percentuali dei prezzi di chiusura rettificati e tronca la data di fine al periodo di addestramento desiderato. Il calcolo dei rendimenti percentuali introduce valori NaN nel DataFrame, che vengono quindi eliminati:
def obtain_prices_df(csv_filepath, end_date):
"""
Legge i prezzi dal file CSV e li carica in un Dataframe,
filtra per data di fine e calcola i rendimenti percentuali.
"""
df = pd.read_csv(
csv_filepath, header=0,
names=[
"Date", "Open", "High", "Low",
"Close", "Volume", "Adj Close"
],
index_col="Date", parse_dates=True
)
df["Returns"] = df["Adj Close"].pct_change()
df = df[:end_date.strftime("%Y-%m-%d")]
df.dropna(inplace=True)
return df
Grafico degli stati
La seguente funzione, plot_in_sample_hidden_states, non è strettamente necessaria ai fini dell’addestramento. Abbiamo modificato il file tutorial, presente nella documentazione di hmmlearn.
Il codice prende i dati del modello e il dataframe dei prezzi e crea un grafico per ogni stato nascosto generato dal modello. Ogni grafico mostra il prezzo di chiusura mascherato da quel particolare stato/regime nascosto. Questo è utile per verificare se l’HMM sta producendo stati “sani”:
def plot_in_sample_hidden_states(hmm_model, df):
"""
Traccia il grafico dei prezzi di chiusura rettificati
mascherati dagli stati nascosti nel campione come
meccanismo per comprendere i regimi di mercato.
"""
# Array con gli stati nascosti previsti
hidden_states = hmm_model.predict(df["Returns"])
# Crea il grafico formattato correttamente
fig, axs = plt.subplots(
hmm_model.n_components,
sharex=True, sharey=True
)
colours = cm.rainbow(
np.linspace(0, 1, hmm_model.n_components)
)
for i, (ax, colour) in enumerate(zip(axs, colours)):
mask = hidden_states == i
ax.plot_date(
df.index[mask],
df["Adj Close"][mask],
".", linestyle='none',
c=colour
)
ax.set_title("Hidden State #%s" % i)
ax.xaxis.set_major_locator(YearLocator())
ax.xaxis.set_minor_locator(MonthLocator())
ax.grid(True)
plt.show()
La seguente figura riporta l’output di questa specifica funzione:
Notiamo come il rilevamento del regime cattura in gran parte periodi di “tendenza” e periodi altamente volatili. In particolare la maggior parte del 2008 si trova nel Hidden State #1.
Script per eseguire l’addestramento
Questo script è legato insieme nella funzione __main__. In primo luogo, tutti gli avvisi vengono ignorati. A rigor di logica questo non l’approccio corretto, ma in questo caso ci sono molti avvisi di deprecazione generati da Scikit-Learn che oscurano l’output desiderato dello script.
Successivamente apriamo il file CSV e si crea la variabile rets tramite il comando np.column_stack perché hmmlearn richiede una matrice di oggetti Series di pandas, nonostante si tratti di un modello univariato (agisce solo sui rendimenti). A tale scopo dobbiamo usare le funzionalità messe a disposizione dalla libreria NumPy.
L’oggetto GaussianHMM richiede di specificare il numero di stati tramite il parametro n_components. In questo articolo usiamo due stati, ma si può facilmente testare lo stesso algoritmo con tre stati. Inoltre usiamo una matrice di covarianza completa, anziché una versione diagonale e usiamo il parametro n_iter per definire il numero di iterazioni per l’algoritmo Expectation-Maximisation.
Addestriamo il modello e visualizziamo l’output dell’algoritmo e il grafico degli Hidden State dei prezzi di chiusura rettificati. Infine il modello è serializzato in pickle_path, pronto per essere usato nel gestore del rischio per rilevamento del regime:
if __name__ == "__main__":
# Nasconde gli avvisi di deprecazione per sklearn
warnings.filterwarnings("ignore")
# Crea il dataframe SPY dal file CSV di Yahoo Finance e
# formatta correttamente i rendimente per l'uso nell'HMM
csv_filepath = "/path/to/your/data/SPY.csv"
pickle_path = "/path/to/your/model/hmm_model_spy.pkl"
end_date = datetime.datetime(2004, 12, 31)
spy = obtain_prices_df(csv_filepath, end_date)
rets = np.column_stack([spy["Returns"]])
# Crea il Gaussian Hidden markov Model e lo adatta ai
# dati dei rendimenti di SPY, visualizzando il punteggio
hmm_model = GaussianHMM(
n_components=2, covariance_type="full", n_iter=1000
).fit(rets)
print("Model Score:", hmm_model.score(rets))
# Grafico dei valori di chiusura degli stati nascosti del campione
plot_in_sample_hidden_states(hmm_model, spy)
print("Pickling HMM model...")
pickle.dump(hmm_model, open(pickle_path, "wb"))
print("...HMM model pickled.")
Strategia Trend Following a breve termine
La fase successiva del processo consiste nel creare la classe Strategy che implementa la logica trend following a breve termine, filtrata successivamente dal modulo RiskManager.
Come per tutte le strategie sviluppate all’interno di DataTrader è necessario importare alcune classi specifiche, tra cui PriceParser, SignalEvente la classe base AbstractStrategy. Questa strategia è simile a molte descritte negli articoli precedenti quindi non descriviamo i singoli passaggi di importazione delle librerie:
# regime_hmm_strategy.py
from collections import deque
import numpy as np
from datatrader.price_parser import PriceParser
from datatrader.event import SignalEvent, EventType
from datatrader.strategy.base import AbstractStrategy
In realtà la sottoclasse MovingAverageCrossStrategy sottoclasse è già stata usata in uno degli esempi precedenti. Tuttavia la replichiamo per completezza. La strategia utilizza due code a doppia estremità, disponibili nel modulo deque, per fornire le finestre scorrevoli sui dati sui prezzi. Questo serve per calcolare le medie mobili semplici che formano la logica del trend-following a breve termine:
class MovingAverageCrossStrategy(AbstractStrategy):
"""
Requisiti:
tickers - La lista dei simboli dei ticker
events_queue - Il manager della coda degli eventi
short_window - Periodo di lookback per la media mobile breve
long_window - Periodo di lookback per la media mobile lunga
"""
def __init__(
self, tickers,
events_queue, base_quantity,
short_window=10, long_window=30
):
self.tickers = tickers
self.events_queue = events_queue
self.base_quantity = base_quantity
self.short_window = short_window
self.long_window = long_window
self.bars = 0
self.invested = False
self.sw_bars = deque(maxlen=self.short_window)
self.lw_bars = deque(maxlen=self.long_window)
Calcolo dei Segnali
Nel framework di backtesting event-driven DataTrader tutte le sottoclassi derivate da AbstractStrategy prevedono l’uso del metodo calculate_signals per generare oggetti SignalEvent. Per la nostra strategia il metodo verifica innanzitutto se l’evento è una barra OHLCV. Ad esempio, potrebbe essere un SentimentEvent (come in altre strategie ) e quindi è necessario prevedere un controllo. Aggiungiamo i prezzi più recenti alle code delle finestre mobili in modo da aggiornare le SMA.
Se ci sono abbastanza barre per eseguire le medie mobili, queste sono entrambe calcolate. Una volta che questi valori sono presenti, eseguiamo le regole di trading sopra descritte. Nel caso la SMA della finestra breve supera la SMA della finestra lunga e la strategia non è già a mercato, generiamo una posizione long di base_quantity azioni. Altrimenti, se la SMA della finestra lunga supera la SMA della finestra breve la posizione viene chiusa se siamo a mercato:
def calculate_signals(self, event):
# Applica SMA al primo ticker
ticker = self.tickers[0]
if event.type == EventType.BAR and event.ticker == ticker:
# Aggiunge l'ultimo prezzo di chiusura ai dati
# delle finestre corta e lunga
price = event.adj_close_price / PriceParser.PRICE_MULTIPLIER
self.lw_bars.append(price)
if self.bars > self.long_window - self.short_window:
self.sw_bars.append(price)
# Sono presenti abbastanza barre per il trading
if self.bars > self.long_window:
# Calcola le medie mobili semplici
short_sma = np.mean(self.sw_bars)
long_sma = np.mean(self.lw_bars)
# Segnali di trading basati sulla media mobile incrociata
if short_sma > long_sma and not self.invested:
print("LONG: %s" % event.time)
signal = SignalEvent(ticker, "BOT", self.base_quantity)
self.events_queue.put(signal)
self.invested = True
elif short_sma < long_sma and self.invested:
print("SHORT: %s" % event.time)
signal = SignalEvent(ticker, "SLD", self.base_quantity)
self.events_queue.put(signal)
self.invested = False
self.bars += 1
Rilevamento del regime nel RiskManager
In questo articolo vediamo come creare un oggetto come sottoclasse di AbstractRiskManager. E’ il primo utilizzo importante della gestione del rischio con i modelli di Markov nascosti applicata separatamente a una strategia di trading fino ad oggi sul sito tradingquant.it. Come indicato sopra, l’obiettivo di questo oggetto è quello di filtrare le operazioni trend-following a breve termine quando si trovano in un regime ad alta volatilità non desiderato.
Tutte le sottoclassi di tipo RiskManager richiedono l’accesso ad un OrderEvent perchè devono poter di eliminare, modificare o creare ordini a seconda dei vincoli di rischio del portafoglio:
# regime_hmm_risk_manager.py
import numpy as np
from datatrader.event import OrderEvent
from datatrader.price_parser import PriceParser
from datatrader.risk_manager.base import AbstractRiskManager
Creiamo la classe RegimeHMMRiskManager che richiede semplicemente la lettura del file del modello HMM deserializzato. Inoltre dobbiamo tenere traccia se la strategia è “investita” o meno, poiché l’oggetto Strategy non è a conoscenza se i suoi segnali sono stati effettivamente eseguiti:
class RegimeHMMRiskManager(AbstractRiskManager):
"""
Utilizza un modello Hidden Markov precedentemente adattato
come meccanismo di rilevamento del regime. Il gestore del
rischio ignora gli ordini che si verificano durante
un regime non desiderato.
Ciò spiega anche il fatto che un'operazione può essere
a cavallo di due regimi separati. Se un ordine di chiusura
viene ricevuto nel regime non desiderato e l'ordine è aperto,
verrà chiuso, ma non verranno generati nuovi ordini fino
al raggiungimento del regime desiderato.
"""
def __init__(self, hmm_model):
self.hmm_model = hmm_model
self.invested = False
Creiamo un metodo helper, determine_regime, che usa l’oggetto price_handler e l’evento l’ sized_order per ottenere l’elenco completo dei rendimenti dei prezzi di chiusura calcolati da DataTrader (per i dettagli vedere il codice nella sezione precedente). Quindi usiamo il metodo predict dell’oggetto GaussianHMM per produrre una serie di stati di regime previsti. Prendiamo il valore più recente e lo usiamo come “stato nascosto” o regime corrente:
def determine_regime(self, price_handler, sized_order):
"""
Determina il probabile regime effettuando una previsione sui rendimenti
dei prezzi di chiusura nell'oggetto PriceHandler e quindi prende
il valore intero finale come "stato del regime nascosto"
"""
returns = np.column_stack(
[np.array(price_handler.adj_close_returns)]
)
hidden_state = self.hmm_model.predict(returns)[-1]
return hidden_state
Gestione degli ordini
Il metodo refine_orders è obbligatori in tutte le sottoclassi derivate da AbstractRiskManager. In questo caso eseguiamo il metodo determine_regime per determinare lo stato del regime. Creiamo infine il corretto oggetto l’ OrderEvent, che sarà modificato successivamente:
def refine_orders(self, portfolio, sized_order):
"""
Utilizza il modello di Markov nascosto con i rendimenti percentuali
per determinare il regime corrente, 0 per desiderabile o 1 per
indesiderabile. Ingressi Long seguiti solo in regime 0, operazioni
di chiusura sono consentite in regime 1.
"""
# Determinare il regime previsto HMM come un intero
# uguale a 0 (desiderabile) o 1 (indesiderabile)
price_handler = portfolio.price_handler
regime = self.determine_regime(
price_handler, sized_order
)
action = sized_order.action
# Crea l'evento dell'ordine, indipendentemente dal regime. Sarà
# restituito solo se le condizioni corrette sono soddisfatte.
order_event = OrderEvent(
sized_order.ticker,
sized_order.action,
sized_order.quantity
)
..
..
Nella seconda metà del metodo implementiamo la logica per gestire il rischio a seguito del rilevamento del regime. Inseriamo un blocco condizionale che verifica quale stato di regime abbiamo individuato.
Se siamo in uno stato di bassa volatilità #0, controlliamo se l’ordine è di tipo “BOT” o “SLD”. Nel caso di un ordine “BOT” (long), restituiamo un OrderEvent e aggiorniamo lo stato “invested”. Se invece l’ordine è “SLD” (chiudi), chiudiamo la posizione se è aperta, altrimenti annulliamo l’ordine.
Inoltre, se prevediamo un regime di alta volatilità #1, verifichiamo quale ordine abbiamo creato. Non vogliamo posizioni long in questo regime di mercato. Consentiamo invece di chiudere una posizione solo se abbiamo precedentemente aperto una posizione long, altrimenti annulliamo l’ordine.
Questo ci permette di non generare mai una nuova posizione long quando siamo in regime #1. Tuttavia, possiamo chiudere una posizione long precedentemente aperta nel regime n. 1.
Un approccio alternativo potrebbe essere chiudere immediatamente qualsiasi posizione long aperta appena entriamo nel regime #1. Lasciamo questo come esercizio per il lettore!
..
..
# Se abbiamo un regime desiderato, permettiamo gli ordini di acquisto e di
# vendita normalmente per una strategia di trend following di solo lungo
if regime == 0:
if action == "BOT":
self.invested = True
return [order_event]
elif action == "SLD":
if self.invested == True:
self.invested = False
return [order_event]
else:
return []
# Se abbiamo un regime non desiderato, non permetiamo ordini di
# acquisto e permettiamo solo di chiudere posizioni aperte se la
# strategia è già a mercato (da un precedenete regime desiderato)
elif regime == 1:
if action == "BOT":
self.invested = False
return []
elif action == "SLD":
if self.invested == True:
self.invested = False
return [order_event]
else:
return []
Esecuzione del Backtest
Questo conclude il codice RegimeHMMRiskManager. Non ci resta che collegare insieme i tre script/moduli precedenti tramite un oggetto Backtest. Il codice completo per questo script può essere trovato, come per il resto dei moduli, alla fine di questo articolo.
In regime_hmm_backtest.py importiamo le classi ExampleRiskManager e RegimeHMMRiskManager. Questo ci permette con un semplice “switch out” dei gestori del rischio per verificare i risultati di differenti backtest:
# regime_hmm_backtest.py
..
..
from datatrader.risk_manager.example import ExampleRiskManager
..
..
from regime_hmm_strategy import MovingAverageCrossStrategy
from regime_hmm_risk_manager import RegimeHMMRiskManager
Nella funzione run iniziamo con specificare il percorso di del file pickle con i dati necessari per la deserializzazione del l modello HMM. Successivamente specifichiamo il gestore dei dati dei prezzi. Impostiamo il flag calc_adj_return su true, in modo che il gestore dei prezzi calcoli e memorizzi l’array dei rendimenti.
In questa fase configuriamo la MovingAverageCrossStrategy con una finestra breve di 10 giorni, una finestra lunga di 30 giorni e una quantità base di azioni SPY pari a 10.000 unità.
Infine deserializziamo il hmm_model tramite pickle e creiamo un’istanza di risk_manager. Il resto dello script è estremamente simile ad altri backtest descritti negli articoli precedente, quindi riportiamo il codice completo solo alla fine dell’articolo.
Il metodo Run
È semplice “cambiare” i gestori del rischio commentando la riga RegimeHMMRiskManager, sostituendola con la riga ExampleRiskManager e quindi rieseguire il backtest:
def run(config, testing, tickers, filename):
# Impostazione delle variabili necessarie al backtest
pickle_path = "/path/to/your/model/hmm_model_spy.pkl"
..
..
# uso del Use Yahoo Daily Price Handler
start_date = datetime.datetime(2005, 1, 1)
end_date = datetime.datetime(2014, 12, 31)
price_handler = YahooDailyCsvBarPriceHandler(
csv_dir, events_queue, tickers,
start_date=start_date, end_date=end_date,
calc_adj_returns=True
)
# Uso della strategia Moving Average Crossover
base_quantity = 10000
strategy = MovingAverageCrossStrategy(
tickers, events_queue, base_quantity,
short_window=10, long_window=30
)
strategy = Strategies(strategy, DisplayStrategy())
..
..
# Uso del Risk Manager di determinazione del regime HMM
hmm_model = pickle.load(open(pickle_path, "rb"))
risk_manager = RegimeHMMRiskManager(hmm_model)
# Uso di un Risk Manager di esempio
#risk_manager = ExampleRiskManager()
Per eseguire il backtest è necessario aprire il Terminale e digitare quanto segue:
$ python regime_hmm_backtest.py --tickers=SPY
L’output è il seguente:
..
..
---------------------------------
Backtest complete.
Sharpe Ratio: 0.518857928421
Max Drawdown: 0.356537705234
Max Drawdown Pct: 0.356537705234
Risultati della strategia
Costi di transazione
I risultati della strategia qui presentati sono al netto dei costi di transazione. I costi sono simulati utilizzando i prezzi fissi delle azioni statunitensi di Interactive Brokers per le azioni del Nord America. Sono ragionevolmente rappresentativi di ciò che potrebbe essere ottenuto in una vera strategia di trading.
Nessun filtro di rilevamento del regime
La strategia è progettata per catturare i trend a breve termine dell’ETF SPY. Otteniamo uno Sharpe Ratio di 0,37, cioè stiamo assumendo una notevole quantità di volatilità per generare pochi rendimenti. In effetti il benchmark ha uno Sharpe ratio quasi identico. Il massimo drawdown giornaliero è leggermente superiore al benchmark, ma produce un leggero aumento del CAGR al 6,41% rispetto al 5,62%.
In sostanza, la strategia si comporta come il benchmark buy-and-hold. Questo è prevedibile dato che le medie mobili sono un indicatore di ritardo e, nonostante abbia effettuato 41 operazioni, non evita i grandi movimenti al ribasso. La domanda principale è se un filtro di regime migliorerà la strategia o meno.
Filtro di rilevamento del regime HMM
Notiamo che applichiamo il filtro del regime in modalità out-of-sample. Cioè, non utilizziamo nessun dato dei rendimenti impiegato nel backtest per addestrare il modello Hidden Markov.
Conclusioni
La strategia basata sul filtro del regime produce risultati sensibilmente diversi. In particolare, riduciamo il drawdown giornaliero massimo della strategia a circa il 24%, rispetto al 56% del benchmark. Si tratta di un’ottima riduzione del “rischio”. Tuttavia, lo Sharpe ratio, pari a 0,48, non cresce in modo significativo rispetto al valore precedente, poiché la strategia continua a subire l’elevata volatilità necessaria per ottenere quei rendimenti.
Il CAGR migliora solo lievemente, raggiungendo il 6,88% rispetto al 6,41% della strategia precedente, ma riusciamo a ridurre leggermente il rischio.
Un aspetto più problematico riguarda il numero di operazioni, che scende da 41 a 31. Anche se eliminiamo grandi movimenti al ribasso (e quindi vantaggiosi), questo significa che la strategia effettua meno “scommesse positive” e, di conseguenza, possiede una minore validità statistica.
Inoltre, non eseguiamo alcuna operazione dall’inizio del 2008 alla metà del 2009. Pertanto, restiamo effettivamente al di sotto del precedente high watermark durante questo periodo. Il vantaggio principale, naturalmente, è che evitiamo perdite quando molti altri ne avrebbero subite!
Per portare “live” questa strategia, probabilmente dobbiamo prevedere un addestramento periodico del modello Hidden Markov, dato che le probabilità di transizione degli stati non si mantengono stazionarie nel tempo. In sostanza, l’HMM può prevedere le transizioni di stato solo in base alle distribuzioni precedenti dei rendimenti ricevuti come input. Se la distribuzione cambia (per esempio a causa di un nuovo contesto normativo), dobbiamo riadattare il modello per coglierne il nuovo comportamento. La frequenza con cui compiamo questa operazione resta, ovviamente, un interessante tema per ricerche future!
Codice Completo
Il codice completo della gestione del rischio con i modelli di Markov nascosti presentata in questo articolo, basato sul framework di trading quantitativo event-driven DataTrader, è disponibile nel seguente repository GitHub: https://github.com/tradingquant-it/DataTrader.”
# regime_hmm_train.py
import datetime
import pickle
import warnings
from hmmlearn.hmm import GaussianHMM
from matplotlib import cm, pyplot as plt
from matplotlib.dates import YearLocator, MonthLocator
import numpy as np
import pandas as pd
import seaborn as sns
def obtain_prices_df(csv_filepath, end_date):
"""
Legge i prezzi dal file CSV e li carica in un Dataframe,
filtra per data di fine e calcola i rendimenti percentuali.
"""
df = pd.read_csv(
csv_filepath, header=0,
names=[
"Date", "Open", "High", "Low",
"Close", "Volume", "Adj Close"
],
index_col="Date", parse_dates=True
)
df["Returns"] = df["Adj Close"].pct_change()
df = df[:end_date.strftime("%Y-%m-%d")]
df.dropna(inplace=True)
return df
def plot_in_sample_hidden_states(hmm_model, df):
"""
Traccia il grafico dei prezzi di chiusura rettificati
mascherati dagli stati nascosti nel campione come
meccanismo per comprendere i regimi di mercato.
"""
# Array con gli stati nascosti previsti
hidden_states = hmm_model.predict(df["Returns"])
# Crea il grafico formattato correttamente
fig, axs = plt.subplots(
hmm_model.n_components,
sharex=True, sharey=True
)
colours = cm.rainbow(
np.linspace(0, 1, hmm_model.n_components)
)
for i, (ax, colour) in enumerate(zip(axs, colours)):
mask = hidden_states == i
ax.plot_date(
df.index[mask],
df["Adj Close"][mask],
".", linestyle='none',
c=colour
)
ax.set_title("Hidden State #%s" % i)
ax.xaxis.set_major_locator(YearLocator())
ax.xaxis.set_minor_locator(MonthLocator())
ax.grid(True)
plt.show()
if __name__ == "__main__":
# Nasconde gli avvisi di deprecazione per sklearn
warnings.filterwarnings("ignore")
# Crea il dataframe SPY dal file CSV di Yahoo Finance e
# formatta correttamente i rendimente per l'uso nell'HMM
csv_filepath = "/path/to/your/data/SPY.csv"
pickle_path = "/path/to/your/model/hmm_model_spy.pkl"
end_date = datetime.datetime(2004, 12, 31)
spy = obtain_prices_df(csv_filepath, end_date)
rets = np.column_stack([spy["Returns"]])
# Crea il Gaussian Hidden markov Model e lo adatta ai
# dati dei rendimenti di SPY, visualizzando il punteggio
hmm_model = GaussianHMM(
n_components=2, covariance_type="full", n_iter=1000
).fit(rets)
print("Model Score:", hmm_model.score(rets))
# Grafico dei valori di chiusura degli stati nascosti del campione
plot_in_sample_hidden_states(hmm_model, spy)
print("Pickling HMM model...")
pickle.dump(hmm_model, open(pickle_path, "wb"))
print("...HMM model pickled.")
# regime_hmm_strategy.py
from collections import deque
import numpy as np
from datatrader.price_parser import PriceParser
from datatrader.event import SignalEvent, EventType
from datatrader.strategy.base import AbstractStrategy
class MovingAverageCrossStrategy(AbstractStrategy):
"""
Requisiti:
tickers - La lista dei simboli dei ticker
events_queue - Il manager della coda degli eventi
short_window - Periodo di lookback per la media mobile breve
long_window - Periodo di lookback per la media mobile lunga
"""
def __init__(
self, tickers,
events_queue, base_quantity,
short_window=10, long_window=30
):
self.tickers = tickers
self.events_queue = events_queue
self.base_quantity = base_quantity
self.short_window = short_window
self.long_window = long_window
self.bars = 0
self.invested = False
self.sw_bars = deque(maxlen=self.short_window)
self.lw_bars = deque(maxlen=self.long_window)
def calculate_signals(self, event):
# Applica SMA al primo ticker
ticker = self.tickers[0]
if event.type == EventType.BAR and event.ticker == ticker:
# Aggiunge l'ultimo prezzo di chiusura ai dati
# delle finestre corta e lunga
price = event.adj_close_price / PriceParser.PRICE_MULTIPLIER
self.lw_bars.append(price)
if self.bars > self.long_window - self.short_window:
self.sw_bars.append(price)
# Sono presenti abbastanza barre per il trading
if self.bars > self.long_window:
# Calcola le medie mobili semplici
short_sma = np.mean(self.sw_bars)
long_sma = np.mean(self.lw_bars)
# Segnali di trading basati sulla media mobile incrociata
if short_sma > long_sma and not self.invested:
print("LONG: %s" % event.time)
signal = SignalEvent(ticker, "BOT", self.base_quantity)
self.events_queue.put(signal)
self.invested = True
elif short_sma < long_sma and self.invested:
print("SHORT: %s" % event.time)
signal = SignalEvent(ticker, "SLD", self.base_quantity)
self.events_queue.put(signal)
self.invested = False
self.bars += 1
# regime_hmm_risk_manager.py
import numpy as np
from datatrader.event import OrderEvent
from datatrader.price_parser import PriceParser
from datatrader.risk_manager.base import AbstractRiskManager
class RegimeHMMRiskManager(AbstractRiskManager):
"""
Utilizza un modello Hidden Markov precedentemente adattato
come meccanismo di rilevamento del regime. Il gestore del
rischio ignora gli ordini che si verificano durante
un regime non desiderato.
Ciò spiega anche il fatto che un'operazione può essere
a cavallo di due regimi separati. Se un ordine di chiusura
viene ricevuto nel regime non desiderato e l'ordine è aperto,
verrà chiuso, ma non verranno generati nuovi ordini fino
al raggiungimento del regime desiderato.
"""
def __init__(self, hmm_model):
self.hmm_model = hmm_model
self.invested = False
def determine_regime(self, price_handler, sized_order):
"""
Determina il probabile regime effettuando una previsione sui rendimenti
dei prezzi di chiusura nell'oggetto PriceHandler e quindi prende
il valore intero finale come "stato del regime nascosto"
"""
returns = np.column_stack(
[np.array(price_handler.adj_close_returns)]
)
hidden_state = self.hmm_model.predict(returns)[-1]
return hidden_state
def refine_orders(self, portfolio, sized_order):
"""
Utilizza il modello di Markov nascosto con i rendimenti percentuali
per determinare il regime corrente, 0 per desiderabile o 1 per
indesiderabile. Ingressi Long seguiti solo in regime 0, operazioni
di chiusura sono consentite in regime 1.
"""
# Determinare il regime previsto HMM come un intero
# uguale a 0 (desiderabile) o 1 (indesiderabile)
price_handler = portfolio.price_handler
regime = self.determine_regime(
price_handler, sized_order
)
action = sized_order.action
# Crea l'evento dell'ordine, indipendentemente dal regime. Sarà
# restituito solo se le condizioni corrette sono soddisfatte.
order_event = OrderEvent(
sized_order.ticker,
sized_order.action,
sized_order.quantity
)
# Se abbiamo un regime desiderato, permettiamo gli ordini di acquisto e di
# vendita normalmente per una strategia di trend following di solo lungo
if regime == 0:
if action == "BOT":
self.invested = True
return [order_event]
elif action == "SLD":
if self.invested == True:
self.invested = False
return [order_event]
else:
return []
# Se abbiamo un regime non desiderato, non permetiamo ordini di
# acquisto e permettiamo solo di chiudere posizioni aperte se la
# strategia è già a mercato (da un precedenete regime desiderato)
elif regime == 1:
if action == "BOT":
self.invested = False
return []
elif action == "SLD":
if self.invested == True:
self.invested = False
return [order_event]
else:
return []
# regime_hmm_backtest.py
import click
import datetime
import pickle
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
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 regime_hmm_strategy import MovingAverageCrossStrategy
from regime_hmm_risk_manager import RegimeHMMRiskManager
def run(config, testing, tickers, filename):
# Impostazione delle variabili necessarie al backtest
pickle_path = "/path/to/your/model/hmm_model_spy.pkl"
events_queue = queue.Queue()
csv_dir = config.CSV_DATA_DIR
initial_equity = PriceParser.parse(500000.00)
# uso del Use Yahoo Daily Price Handler
start_date = datetime.datetime(2005, 1, 1)
end_date = datetime.datetime(2014, 12, 31)
price_handler = YahooDailyCsvBarPriceHandler(
csv_dir, events_queue, tickers,
start_date=start_date, end_date=end_date,
calc_adj_returns=True
)
# Uso della strategia Moving Average Crossover
base_quantity = 10000
strategy = MovingAverageCrossStrategy(
tickers, events_queue, base_quantity,
short_window=10, long_window=30
)
strategy = Strategies(strategy)
# Uso di un Position Sizer standard
position_sizer = NaivePositionSizer()
# Uso del Risk Manager di determinazione del regime HMM
hmm_model = pickle.load(open(pickle_path, "rb"))
risk_manager = RegimeHMMRiskManager(hmm_model)
# Uso di un Risk Manager 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
title = ["Trend Following Regime Detection with HMM"]
statistics = TearsheetStatistics(
config, portfolio_handler, title,
benchmark="SPY"
)
# 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='SPY'
)
results = backtest.start_trading(testing=testing)
statistics.save(filename)
return results
@click.command()
@click.option('--config', default=settings.DEFAULT_CONFIG_FILENAME, help='Config filename')
@click.option('--testing/--no-testing', default=False, help='Enable testing mode')
@click.option('--tickers', default='SPY', help='Tickers (use comma)')
@click.option('--filename', default='', help='Pickle (.pkl) statistics filename')
def main(config, testing, tickers, filename):
tickers = tickers.split(",")
config = settings.from_file(config, testing)
run(config, testing, tickers, filename)
if __name__ == "__main__":
main()
