Strategia di Momentum Settoriale per l'asset allocation tattica

Strategia di Momentum settoriale per l’asset allocation tattica

Sommario

SCRIVIMI SU TELEGRAM

Per informazioni, suggerimenti, collaborazioni...

Se è la prima volta che atterri su DataTrading, BENVENUTO!

Lascia che mi presenti. Sono Gianluca, ingegnere, trekker e appassionato di coding, finanza e trading. Leggi la mia storia.

Ho creato DataTrading per aiutare le altre persone ad utilizzare nuovi approcci e nuovi strumenti, ed applicarli correttamente al mondo del trading.

DataTrading vuole essere un punto di ritrovo per scambiare esperienze, opinioni ed idee.

TUTORIAL

Nell’articolo precedente abbiamo descritto un semplice portafoglio di allocazione statica con ribilanciamento periodico. In questo articolo implementiamo il backtest di una nota strategia dinamica di asset allocation tattica chiamata momentum settoriale .

Gran parte del codice è simile all’articolo precedente, ma introduciamo alcune nuove funzionalità. Aggiungiamo la classe DynamicUniverse, che ci consente di modificare la composizione degli asset nel tempo. Descriviamo inoltre le entità MomentumSignalSignalsCollection che  ci permettono di calcolare le serie temporali di momentum. Descriviamo anche come creare entità AlphaModel personalizzate per  generare specifici segnali da inserire nel processo di ottimizzazione del portafoglio.

Come per il precedente, questo articolo presuppone che abbiamo correttamente installato DataInvestor.

Segnaliamo che tutto il codice riportato di seguito può essere trovato all’interno del file /examples/momentum_taa.py nel repository GitHub di DataInvestor. Faremo riferimento a questo file nei passaggi del tutorial di seguito.

La prima cosa da fare è scaricare il file momentum_taa.py, utilizzare il codice completo riportato alla fine di questo articolo o creare un nuovo file vuoto e digitare manualmente il codice sottostante.

Iniziamo ad introdurre la logica della strategia e a scaricare i dati necessari da Yahoo Finance. Descriviamo successivamente tutto il codice momentum_taa.py e infine eseguiamo il backtest per visualizzare un “tearsheet” delle prestazioni.

Logica della strategia

La strategia di momentum settoriale del mercato statunitense è una strategia dinamica long-only di asset allocation tattica che tenta di superare la performance rispetto al semplice “buy & hold” sull’S&P500.

Alla fine di ogni mese la strategia calcola il momentum basato sul rendimento del periodo di holding (HPR) di tutti gli ETF settoriali prodotti da SPDR (i cui simboli ticker iniziano con il prefisso XL) e seleziona i primi N su cui investire per il mese successivo, dove N è solitamente compreso tra 3 e 6.

Nella seguente implementazione, il momentum HPR viene calcolato sui 126 giorni precedenti (circa sei mesi di giorni lavorativi) e vengono scelti i primi tre ETF di settore per il portafoglio. L’allocazione del portafoglio è pesata ugualmente tra ciascuno di questi tre ETF settoriali. Indipendentemente dalle variazioni del segnale, il portafoglio viene ribilanciato una volta al mese per pesare equamente le attività.

DataInvestor fornisce funzionalità automatiche per il calcolo dei valori del momentum di ciascun asset. Il seguente codice si concentra quindi sulla logica per determinare gli asset con momentum più elevato e quindi pesarli equamente.

Download dei dati

Il codice di momentum_taa.py usa i dati della “barra giornaliera” OHLC di Yahoo Finance. In particolare richiede i seguenti ETF settoreiali di SPDR: XLB, XLC, XLE, XLF, XLI, XLK, XLP, XLU, XLV e XLY. Ad esempio, XLK può essere scaricato qui. Scarichiamo quindi lo storico completo per ciascuno ETF e lo salviamo come file CSV nella directory definita in config.yaml.

Implementazione dello script di backtest

Gran parte del codice riportato di seguito è simile a quello delineato nell’articolo precedente. Per completezza descriviamo tutto il codice da zero, includendo il codice completo alla fine di questo articolo.

Nota: il seguente codice contiene commenti estesi per qualificare completamente i parametri e i tipi restituiti. Alcuni di questi commenti sono stati rimossi negli snippet man mano che si avanza nell’articolo per migliorare la leggibilità.

Importazione dei moduli

La prima parte del codice Python importa i moduli necessari dall’interno della libreria DataInvestor:

				
					import operator

import pandas as pd
import pytz

from datainvestor.alpha_model.alpha_model import AlphaModel
from datainvestor.alpha_model.fixed_signals import FixedSignalsAlphaModel
from datainvestor.asset.equity import Equity
from datainvestor.asset.universe.dynamic import DynamicUniverse
from datainvestor.asset.universe.static import StaticUniverse
from datainvestor.signals.momentum import MomentumSignal
from datainvestor.signals.signals_collection import SignalsCollection
from datainvestor.data.backtest_data_handler import BacktestDataHandler
from datainvestor.data.daily_bar_csv import CSVDailyBarDataSource
from datainvestor.statistics.tearsheet import TearsheetStatistics
from datainvestor.trading.backtest import BacktestTradingSession
from datainvestor import settings
				
			

Per prima cosa importiamo operator, necessario per selezionare gli asset di momentum con le migliori prestazioni. Importiamo anche datainvestor.settings dato che dobbiamo interrogare la configurazione di DataInvestor per  conoscere la specifica directory dove sono memorizzati i file CSV. È possibile dire a DataInvestor di cercare i dati in una directory diversa tramite il file config.yaml, ma per gli scopi di questo articilo usiamo la direcotry di default.

Quindi importiamo Panda e Pytz, necessari per gestire la funzionalità di timestamp con i fusi orari.

Le restanti importazioni sono librerie che fanno parte di DataInvestor. Per prima cosa importiamo la classe base astratta AlphaModel, che è ustata per creare la nostra sottoclasse del modello alpha momentum. Importiamo anche FixedSignalsAlphaModel da usare per il benchmark S&P500.

Un AlphaModel di DataInvestor è una classe che emette una “previsione” o “segnale” adimensionale per un particolare  paniere di asset (noto come “universo degli asset”).

La classe  FixedSignalsAlphaModel è un semplice generatore di segnali che emette previsioni costanti, indipendentemente dal comportamento del mercato, per uno specifico paniere di asset. Dato che vogliamo confrontare la strategia  TAA momentum con il buy & hold dell’S&P500, abbiamo bisogno di un segnale fisso di 1.0 per l’ETF SPY, che rappresenta questo benchmark.

A questo punto importiamo  la classe Equity per dire al modulo che carica i dati storici che i file CSV rappresentano prezzi/volumi azionari.

Importiamo le classi StaticUniverse e DynamicUniverse. Il nostro ‘universo’ di asset momentum è costituito da un paniere che varia nel tempo. XLC (che rappresenta il settore delle “comunicazioni”) ed è stato aggiunto più recentemente rispetto agli altri nove ETF settoriali. Quindi dobbiamo usare la classe DynamicUniverse per dire a DataInvestor che un nuovo asset viene aggiunto durante il periodo di backtest. Usiamo la StaticUniverse  per il benchmark statico di buy & hold di SPY.

La differenza principale tra questo articolo e l’articolo precedente è l’introduzione delle classi MomentumSignal e SignalsCollection. Sono classi helper che calcolano il momentum su un paniere di asset, per uno specifico periodo di lookback. Descriviamo nel dettaglio la logica di queste classi in un paragrafo successivo.

Inoltre importiamo la  classe BacktestDataHandler, una classe helper che determina il modo in cui lo specifico universo di asset e le  sorgenti dati sono collegate.

Importiamo CSVDailyBarDataSource perché le specifiche sorgenti dati usate in questo articolo sono file CSV, al contrario di un database, un URL Web o un archivio di file (come HDF5).

Poiché desideriamo visualizzare i risultati del nostro backtest, importiamo la  classeTearsheetStatistics.

Infine, per legare tutto insieme, importiamo il BacktestTradingSession, che esegue il nostro “motore” di backtest pianificato tra i timestamp di inizio e di fine specificati.

Definizione del modello Momentum Alpha

Un’altra differenza con il precedente articolo è la creazione della logica AlphaModel tramite la classe TopNMomentumAlphaModel. In questa classe è implementata la maggior parte della logica della strategia:

				
					
class TopNMomentumAlphaModel(AlphaModel):

    def __init__(self, signals, mom_lookback, mom_top_n, universe, data_handler):
        
        self.signals = signals
        self.mom_lookback = mom_lookback
        self.mom_top_n = mom_top_n
        self.universe = universe
        self.data_handler = data_handler

    def _highest_momentum_asset(self, dt):
        assets = self.signals['momentum'].assets

        # Calcola il momentum dei rendimento del periodo di holding per ciascun
        # asset, per il periodo di ricerca di momentum specificato.
        all_momenta = {
            asset: self.signals['momentum'](
                asset, self.mom_lookback
            ) for asset in assets
        }

        # Calcolo dell'elenco degli asset con le migliori prestazioni
        # in base al momentum, limitato dal numero di asset desiderate
        # da negoziare ogni mese
        return [
                asset[0] for asset in sorted(
                    all_momenta.items(),
                    key=operator.itemgetter(1),
                    reverse=True
                )
               ][:self.mom_top_n]

    def _generate_signals(self, dt, weights):
        top_assets = self._highest_momentum_asset(dt)
        for asset in top_assets:
            weights[asset] = 1.0 / self.mom_top_n
        return weights

    def __call__(self, dt):
        assets = self.universe.get_assets(dt)
        weights = {asset: 0.0 for asset in assets}

        # Genera pesi solo se l'ora corrente supera
        # il periodo di lookback del momentum
        if self.signals.warmup >= self.mom_lookback:
            weights = self._generate_signals(dt, weights)
        return weights

				
			

Possiamo vedere che la classe accetta cinque parametri.

Prende un’istanza SignalsCollection in modo da avere accesso a segnali di momentum precalcolati.

Richiede due parametri relativi al momentum: mom_lookback mom_top_n. Il primo è il numero intero di giorni lavorativi per calcolare il momentum del rendimento del periodo di  holding(HPR). Il secondo determina quanti asset scambiare in un determinato mese.

Richiede anche un’istanza Universe (in questo caso la nostra DynamicUniverse descritta sopra) e una DataHandler per ottenere l’accesso ai dati sui prezzi sottostanti.

Il primo metodo è _highest_momentum_asset. Ottiene i segnali di momentum su tutti gli asset correnti, per il particolare periodo di lookback specificato, e li inserisce in un dizionario indicizzato con il simbolo dell’asset. Quindi utilizza la funzione sorted integrata di Python, insieme al metodo operator.itemgetter e a una comprensione dell’elenco per creare un elenco ordinato inversamente degli asset con momentum più elevato, suddividendo questo elenco nei primi N asset.

Ad esempio, l’output di questo elenco potrebbe apparire come ['EQ:XLC', 'EQ:XLB', 'EQ:XLK']se questi fossero i primi tre asset di maggior momentum in questo periodo.

Il secondo metodo è _generate_signals. Chiama semplicemente il metodo _highest_momentum_asset e produce un dizionario weights con ciascun di questi asset pesato allo stesso modo.

Ad esempio, l’output potrebbe essere simile a {'EQ:XLC': 0.3333333, 'EQ:XLB': 0.3333333, 'EQ:XLK': 0.3333333}.

Il metodo finale, che trasforma the TopNMomentumAlphaModel in un callable, è __call__. Questo avvolge gli altri due metodi precedenti e genera questi pesi solo se a questo punto del backtest sono stati raccolti dati sufficienti per generare un segnale di momentum completo per ogni asset. Infine, viene restituito il dizionario weights.

Usiamo un’istanza di questa classe all’interno di if __name__ == "__main__":.

Durata del backtest

L’attività successiva consiste nel definire il punto di ingresso dello script e specificare il periodo temporale da considere nel backtest viene eseguito da e verso. Un’altra differenza tra questo articol e il precedente è la necessità di specificare un tempo di “burn in”. Questo differisce dall’ora di inizio del backtest:

				
					
if __name__ == "__main__":
    testing = False
    config_file = settings.DEFAULT_CONFIG_FILENAME
    config = settings.from_file(config_file, testing)

    # Durata del backtest
    start_dt = pd.Timestamp('1998-12-22 14:30:00', tz=pytz.UTC)
    burn_in_dt = pd.Timestamp('1999-12-22 14:30:00', tz=pytz.UTC)
    end_dt = pd.Timestamp('2020-12-31 23:59:00', tz=pytz.UTC)
				
			

Il tempo di burn in è il momento in cui i pesi vengono effettivamente generati e inviati all’istanza di costruzione del portafoglio. Questo è necessario perché il segnale HPR momentum ha un periodo lookback. Se in un punto particolare del backtest esistono dati giornalieri insufficienti, abbiamo scelto di non generare segnali fino a quando i dati non saranno disponibili.

Nota: le date utilizzano la classe Timestamp di Pandas e richiedono la specifica del fuso orario tramite pytz.

Nota: un approccio alternativo consiste nell’utilizzare un periodo lookback a finestra espandibile, dove si usano tutti i dati attualmente disponibili per calcolare un segnale di momentum fino a quando non esistono dati sufficienti per utilizzare l’intero periodo lookback. In questo articolo abbiamo scelto di non utilizzare finestre espandibili.

Nota: in questa fase di sviluppo di DataInvestor, per la metodologia ‘buy & hold’ usata del benchmark è necessario specificare un’ora di inizio alle 14:30:00 UTC affinché il backtest sia effettuato correttamente.

Parametri del modello

Il modello di momentum top N richiede due parametri. Il primo è il numero totale di giorni lavorativi da utilizzare nel segnale di momentum HPR. In questo articolo abbiamo scelto 126, che corrisponde a circa sei mesi di giorni lavorativi. Il secondo parametro è il numero di asset da includere nel portafoglio (equipesato) per ogni mese.

				
					
    # Parametri del modello
    mom_lookback = 126  # Sei mesi di giorni di mercato aperto
    mom_top_n = 3  # Numero di asset da includere ad ogni ribilanciamento
				
			

Universo patrimoniale

Il prossimo passo è costruire l’universo dinamico degli asset per il backtest:

				
					
    # Costruzione dei simboli e degli asset necessari per il backtest. Si usa gli ETF
    # settoriali SPDR del mercato statunitense, tutti quelli che iniziato per XL
    strategy_symbols = ['XL%s' % sector for sector in "BCEFIKPUVY"]
    assets = ['EQ:%s' % symbol for symbol in strategy_symbols]

    # Poiché si tratta di un universo dinamico di asset (XLC viene aggiunto in seguito),
    # dobbiamo comunicare a DataInvestor quando XLC può essere incluso. A tale scopo si usa
    # un dizionario delle date degli asset
    asset_dates = {asset: start_dt for asset in assets}
    asset_dates['EQ:XLC'] = pd.Timestamp('2018-06-18 00:00:00', tz=pytz.UTC)
    strategy_universe = DynamicUniverse(asset_dates)
				
			

DataInvestor prevede che la  definizione degli asset siano precedute da un codice di due lettere che descrive il tipo di strumento. In questa simulazione gli asset devono essere precedute da EQ, che rappresenta strumenti simili alle azioni (azioni ordinarie ed ETF).

Il primo compito è creare un elenco assets che contiene tutti e dieci gli ETF usati nel backtest. L’attività successiva consiste nel costruire un dizionario asset_dates indicizzato con i simboli degli asset e contiene la data di inizio di ogni asset nell’universo dinamico. Tutti gli ETF tranne uno iniziano alla data di inizio del backtest, tuttavia XLC inizia il 18 giugno 2018 e la sua data di inizio viene modificata di conseguenza.

Il dizionario asset_dates viene passato all’istanza DynamicUniverse per creare l’universo di asset per il backtest della strategia.

Dati sui prezzi

L’attività successiva consiste nell’ottenere i dati  specifici per questi simboli e caricarli nel gestore dei dati:

				
					
    # Per evitare di caricare tutti i file CSV nella directory,
    # impostiamo l'origine dati in modo che carichiamo solo i simboli forniti
    csv_dir = config.CSV_DATA_DIR
    strategy_data_source = CSVDailyBarDataSource(csv_dir, Equity, csv_symbols=strategy_symbols)
    strategy_data_handler = BacktestDataHandler(strategy_universe, data_sources=[strategy_data_source])
				
			

Innanzitutto determiniamo la directory dei file CSV, come impostato nel file di configurazione, altrimenti per impostazione predefinita utilizziamo la sottodirectory data della stessa directory di momentum_taa.py. Quindi carichiamo i dati CSV e li colleghiamo all’universo degli asset tramite un BacktestDataHandler.

Segnali

In questa sezione di momentum_taa.py creiamo la classe helper MomentumSignal, passando l’universo su cui desideriamo calcolare il momentum, così come l’intervallo di lookback a cui siamo interessati. Poiché siamo interessati solo a un periodo di lookback, includiamo solo un elenco  con singolo elemento [mom_lookback].

Il SignalsCollection viene usato per raccogliere i vari segnali che potremmo utilizzare all’interno delle nostre istanze del modello alpha. Per gli scopi di questo artico usiamo solo segnali momentum, quindi li passiamo come dizionario con un elemento singolo:

				
					
    # Genera i segnali (in questo caso il momentum basato sul rendimento
    # del periodo di holding) usati nel modello alfa del momentum top-N
    momentum = MomentumSignal(start_dt, strategy_universe, lookbacks=[mom_lookback])
    signals = SignalsCollection({'momentum': momentum}, strategy_data_handler)
				
			

Modello Alpha e Backtest

Il prossimo set di codice crea un’istanza del modello alpha di momentum top N definito in precedenza, crea l’istanza di backtest e la esegue:

				
					

    # Genera l'istanza del modello alfa per il modello alfa del momentum top-N
    strategy_alpha_model = TopNMomentumAlphaModel(
        signals, mom_lookback, mom_top_n, strategy_universe, strategy_data_handler
    )

    # Costruzione del backtest della strategia e l'esegue
    strategy_backtest = BacktestTradingSession(
        start_dt,
        end_dt,
        strategy_universe,
        strategy_alpha_model,
        signals=signals,
        rebalance='end_of_month',
        long_only=True,
        cash_buffer_percentage=0.01,
        burn_in_dt=burn_in_dt,
        data_handler=strategy_data_handler
    )
    strategy_backtest.run()
				
			

Da notare l’aggiunta di entrambi gli argomenti signals burn_in_dt, che non sono richiesti nell’articolo precedente .

Innanzitutto inizializziamo TopNMomentumAlphaModel con i parametri che sono stati definiti sopra.

Successivamente inizializziamo il BacktestTradingSession con le date di inizio, burn-in e fine, l’universo dinamico degli asset, il raccoglitore dei segnali e lo specifico modello alpha sopra definito.

Inoltre forniamo un argomento rebalance. Questo indica al motore di backtesting con quale frequenza vogliamo rieseguire la generazione del segnale e la logica di costruzione del portafoglio. Dato che è disaccoppiato dai timestamp effettivo dei dati, dobbiamo esplicitare la frequenza di ribilanciamento.

Per questa particolare strategia, eseguiamo la generazione del segnale e la costruzione del portafoglio una volta al mese alla chiusura dell’ultimo giorno di negoziazione. In questo modo se nel corso del mese le allocazioni percentuali effettive di ciascun ETF settoriale si sono discostate dalle loro allocazioni equipesate, sono creati ordini di “ribilanciamento” per acquistare/vendere una specifica quantità di EFT al fine di allineare le correnti allocazioni con quelle desiderate.

L’argomento long_only speficia al backtester che questa strategia va prevedere solo posizioni “long” (ovvero, non prevede vendite allo scoperto).

Abbiamo anche bisogno di un “buffer di cassa” nel conto per gestire il caso in cui viene generato un ordine target per una particolare quantità di azioni (alla chiusura del mercato) ma quando viene eseguito alla successiva apertura del mercato l’ordine risulta essere troppo costoso da eseguire con la liquidità disponibile nel conto.

Questo scenario si verifica se il prezzo delle azioni per questo asset è aumentato notevolmente durante la notte. Questo buffer di cassa tenta di tenere conto di tali gap. Manteniamo in contati questa percentuale (approssimativa) del capitale del conto per coprire questa eventualità. In questo esempio lo impostiamo all’1% del patrimonio netto, ma potrebbe essere necessario modificare questo valore se si esegue il backtest su azioni particolarmente volatili.

Infine diciamo al motore di backtest quale gestore dati utilizzare ed eseguiamo il backtest con il metodo run.

In questa fase, una volta terminata l’esecuzione del backtest, l’istanza strategy_backtest contiene tutti i valori calcolati dal backtest (come la curva equity, la curva drawdown e varie altre statistiche).

Benchmark

Siamo inoltre interessati a confrontare questo risultato con un’altra strategia di “benchmark”, che è semplicemente un “buy & hold” dell’ETF SPY. In questo modo possiamo vedere cosa è successo durante la durata del backtest se avessimo comprato e mantenuto in portafoglio solo un ETF che rappresenta l’S&P500 statunitense durante lo stesso periodo di tempo.

Per creare un benchmark eseguiamo un altro backtest completo ma con un diverso modello alpha e un  diverso programma di ribilanciamento:

				
					   
   # Costruzione degli asset del benchmark (buy & hold SPY)
    benchmark_symbols = ['SPY']
    benchmark_assets = ['EQ:SPY']
    benchmark_universe = StaticUniverse(benchmark_assets)
    benchmark_data_source = CSVDailyBarDataSource(csv_dir, Equity, csv_symbols=benchmark_symbols)
    benchmark_data_handler = BacktestDataHandler(benchmark_universe, data_sources=[benchmark_data_source])

    # Costruzione di un modello Alpha di benchmark che fornisca un'allocazione
    # statica al 100% all'ETF SPY, senza ribilanciamento
    benchmark_alpha_model = FixedSignalsAlphaModel({'EQ:SPY': 1.0})
    benchmark_backtest = BacktestTradingSession(
        burn_in_dt,
        end_dt,
        benchmark_universe,
        benchmark_alpha_model,
        rebalance='buy_and_hold',
        long_only=True,
        cash_buffer_percentage=0.01,
        data_handler=benchmark_data_handler
    )
    benchmark_backtest.run()
				
			

Notiamo che FixedSignalsAlphaModel prevede un’assegnazione del 100% a SPY per il benchmark. Utilizza anche il programma  di ribilanciamento 'buy_and_hold'. Quest’ultimo informa il backtester di generare un singolo segnale all’inizio del backtest per andare completamente long SPY. Non ci sono ulteriori trade effettuati per il benchmark.

I restanti parametri sono simili ed eseguiamo ancora una volta il backtest del benchmark con il metodo run.

Nota: la “data di inizio” è stata sostituita con la precedente burn_in_dt, perché è la prima data in cui sono calcolate le posizioni per la strategia momentum. Per confrontare ‘mele con mele’ è necessario far iniziare anche il benchmark da questa data.

Report Tearsheet

Tutti i risultati del backtest sono memorizzati in strategy_backtest benchmark_backtest. Vogliamo visualizzare questi risultati e produrre alcune statistiche di base. Per questo possiamo usare un ‘Tearsheet’:

				
					   
    # Output delle performance
    tearsheet = TearsheetStatistics(
        strategy_equity=strategy_backtest.get_equity_curve(),
        benchmark_equity=benchmark_backtest.get_equity_curve(),
        title='US Sector Momentum - Top 3 Sectors'
    )
    tearsheet.plot_results()
				
			

La classe TearsheetStatistics è istanziata con la curva equity del backtest della strategia, la curva equity del backtest del benchmark (opzionale) e uno specifico titolo. Infine i risultati sono tracciati (usando Matplotlib) attraverso il metodo plot_results.

Esecuzione del backtest

Per eseguire il codice assicuriamoci che l’ambiente virtuale Python, come Anaconda o virtualenv, sia attivato. Quindi digitiamo quanto segue nella stessa directory di momentum_taa.py:

				
					$ python momentum_taa.py
				
			

Alla fine dell’esecuzione del backtest viene mostrato un tearsheet che raffigurante i risultati. Dovrebbe essere simile al seguente:

trading-algoritmico-datainvestor_momentum_taa_backtest

Codice completo

				
					import operator

import pandas as pd
import pytz

from datainvestor.alpha_model.alpha_model import AlphaModel
from datainvestor.alpha_model.fixed_signals import FixedSignalsAlphaModel
from datainvestor.asset.equity import Equity
from datainvestor.asset.universe.dynamic import DynamicUniverse
from datainvestor.asset.universe.static import StaticUniverse
from datainvestor.signals.momentum import MomentumSignal
from datainvestor.signals.signals_collection import SignalsCollection
from datainvestor.data.backtest_data_handler import BacktestDataHandler
from datainvestor.data.daily_bar_csv import CSVDailyBarDataSource
from datainvestor.statistics.tearsheet import TearsheetStatistics
from datainvestor.trading.backtest import BacktestTradingSession
from datainvestor import settings

class TopNMomentumAlphaModel(AlphaModel):

    def __init__(
            self, signals, mom_lookback, mom_top_n, universe, data_handler
    ):
        """
        Inizializzazione del TopNMomentumAlphaModel

        Parameters
        ----------
        signals : `SignalsCollection`
            L'entità per l'interfacciamento con vari segnali
            precalcolati. In questo caso vogliamo usare 'momentum'.
        mom_lookback : `integer`
            Il numero di giorni lavorativi su cui calcolare
            il momentum lookback.
        mom_top_n : `integer`
            Il numero di asset da includere nel portafoglio,
            in ordine discendente dal momentum più alto.
        universe : `Universe`
            L'insieme di asset utilizzati per la generazione dei signali.
        data_handler : `DataHandler`
            L'interfaccia per i dati CSV.

        Returns
        -------
        None
        """
        self.signals = signals
        self.mom_lookback = mom_lookback
        self.mom_top_n = mom_top_n
        self.universe = universe
        self.data_handler = data_handler

    def _highest_momentum_asset(
            self, dt
    ):
        """
        Calcola l'elenco ordinato degli asset con momentum più performanti
        limitati ai "Top N", per una determinata data e ora.

        Parameters
        ----------
        dt : `pd.Timestamp`
            La data e l'ora per la quale devono essere calcolati
            gli asset con momentum più elevato.

        Returns
        -------
        `list[str]`
            Elenco ordinato degli asset con momentum più
            performante limitato ai "Top N".
        """
        assets = self.signals['momentum'].assets

        # Calcola il momentum dei rendimento del periodo di holding per ciascun
        # asset, per il periodo di ricerca di momentum specificato.
        all_momenta = {
            asset: self.signals['momentum'](
                asset, self.mom_lookback
            ) for asset in assets
        }

        # Calcolo dell'elenco degli asset con le migliori prestazioni
        # in base al momentum, limitato dal numero di asset desiderate
        # da negoziare ogni mese
        return [
                asset[0] for asset in sorted(
                    all_momenta.items(),
                    key=operator.itemgetter(1),
                    reverse=True
                )
               ][:self.mom_top_n]

    def _generate_signals(
            self, dt, weights
    ):
        """
        Calcola il momentum più performante per ciascun asset,
        quindi assegna 1/N del peso del segnale a ciascuno
        di questi asset.

        Parameters
        ----------
        dt : `pd.Timestamp`
            Data/ora per la quale devono essere calcolati
            i pesi del segnale.
        weights : `dict{str: float}`
            Il dizionario dei pesi dei segnali.

        Returns
        -------
        `dict{str: float}`
            Il dizionario dei pesi dei segnali appena creato.
        """
        top_assets = self._highest_momentum_asset(dt)
        for asset in top_assets:
            weights[asset] = 1.0 / self.mom_top_n
        return weights

    def __call__(
            self, dt
    ):
        """
        Calculates the signal weights for the top N
        momentum alpha model, assuming that there is
        sufficient data to begin calculating momentum
        on  e desired assets.
        Calcola i pesi del segnale per il modello alfa dei top N momentum,
        presupponendo che vi siano dati sufficienti per iniziare a
        calcolare il momentum per gli asset desiderati.

        Parameters
        ----------
        dt : `pd.Timestamp`
            Data/ora per la quale devono essere calcolati
            i pesi del segnale.

        Returns
        -------
        `dict{str: float}`
            Il dizionario dei pesi dei segnali appena creato.
        """
        assets = self.universe.get_assets(dt)
        weights = {asset: 0.0 for asset in assets}

        # Genera pesi solo se l'ora corrente supera
        # il periodo di lookback del momentum
        if self.signals.warmup >= self.mom_lookback:
            weights = self._generate_signals(dt, weights)
        return weights


if __name__ == "__main__":
    testing = False
    config_file = settings.DEFAULT_CONFIG_FILENAME
    config = settings.from_file(config_file, testing)

    # Durata del backtest
    start_dt = pd.Timestamp('1998-12-22 14:30:00', tz=pytz.UTC)
    burn_in_dt = pd.Timestamp('1999-12-22 14:30:00', tz=pytz.UTC)
    end_dt = pd.Timestamp('2020-12-31 23:59:00', tz=pytz.UTC)

    # Parametri del modello
    mom_lookback = 126  # Sei mesi di giorni di mercato aperto
    mom_top_n = 3  # Numero di asset da includere ad ogni ribilanciamento

    # Costruzione dei simboli e degli asset necessari per il backtest. Si usa gli ETF
    # settoriali SPDR del mercato statunitense, tutti quelli che iniziato per XL
    strategy_symbols = ['XL%s' % sector for sector in "BCEFIKPUVY"]
    assets = ['EQ:%s' % symbol for symbol in strategy_symbols]

    # Poiché si tratta di un universo dinamico di asset (XLC viene aggiunto in seguito),
    # dobbiamo comunicare a DataInvestor quando XLC può essere incluso. A tale scopo si usa
    # un dizionario delle date degli asset
    asset_dates = {asset: start_dt for asset in assets}
    asset_dates['EQ:XLC'] = pd.Timestamp('2018-06-18 00:00:00', tz=pytz.UTC)
    strategy_universe = DynamicUniverse(asset_dates)

    # Per evitare di caricare tutti i file CSV nella directory,
    # impostiamo l'origine dati in modo che carichiamo solo i simboli forniti
    csv_dir = config.CSV_DATA_DIR
    strategy_data_source = CSVDailyBarDataSource(csv_dir, Equity, csv_symbols=strategy_symbols)
    strategy_data_handler = BacktestDataHandler(strategy_universe, data_sources=[strategy_data_source])

    # Genera i segnali (in questo caso il momentum basato sul rendimento
    # del periodo di holding) usati nel modello alfa del momentum top-N
    momentum = MomentumSignal(start_dt, strategy_universe, lookbacks=[mom_lookback])
    signals = SignalsCollection({'momentum': momentum}, strategy_data_handler)

    # Genera l'istanza del modello alfa per il modello alfa del momentum top-N
    strategy_alpha_model = TopNMomentumAlphaModel(
        signals, mom_lookback, mom_top_n, strategy_universe, strategy_data_handler
    )

    # Costruzione del backtest della strategia e l'esegue
    strategy_backtest = BacktestTradingSession(
        start_dt,
        end_dt,
        strategy_universe,
        strategy_alpha_model,
        signals=signals,
        rebalance='end_of_month',
        long_only=True,
        cash_buffer_percentage=0.01,
        burn_in_dt=burn_in_dt,
        data_handler=strategy_data_handler
    )
    strategy_backtest.run()

    # Costruzione degli asset del benchmark (buy & hold SPY)
    benchmark_symbols = ['SPY']
    benchmark_assets = ['EQ:SPY']
    benchmark_universe = StaticUniverse(benchmark_assets)
    benchmark_data_source = CSVDailyBarDataSource(csv_dir, Equity, csv_symbols=benchmark_symbols)
    benchmark_data_handler = BacktestDataHandler(benchmark_universe, data_sources=[benchmark_data_source])

    # Costruzione di un modello Alpha di benchmark che fornisca un'allocazione
    # statica al 100% all'ETF SPY, senza ribilanciamento
    benchmark_alpha_model = FixedSignalsAlphaModel({'EQ:SPY': 1.0})
    benchmark_backtest = BacktestTradingSession(
        burn_in_dt,
        end_dt,
        benchmark_universe,
        benchmark_alpha_model,
        rebalance='buy_and_hold',
        long_only=True,
        cash_buffer_percentage=0.01,
        data_handler=benchmark_data_handler
    )
    benchmark_backtest.run()

    # Output delle performance
    tearsheet = TearsheetStatistics(
        strategy_equity=strategy_backtest.get_equity_curve(),
        benchmark_equity=benchmark_backtest.get_equity_curve(),
        title='US Sector Momentum - Top 3 Sectors'
    )
    tearsheet.plot_results()
				
			

Riepilogo

Abbiamo descritto come usare DataInvestor per eseguire backtest su  una strategia momentum per un asset allocation dinamica con ribilanciamenti mensili, comunemente implementata tramite ETF a basso costo.

Lo script è progettato per essere facilmente modificato in modo da poter applicare ribilanciamenti alternativi, in ordine di pesi degli asset e frequenza giornalieria. DataInvestor gestisce il ribilanciamento giornaliero e settimanale, come implementato in questo codice.

Se hai domande sullo script o in generale su DataInvestor, non esitare a inviarci un’e-mail all’indirizzo [email protected].

Torna su