Strategia di Cointegrazione tra Asset con DataTrader

Strategia di Cointegrazione tra asset con DataTrader

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

Negli articoli precedenti è stato introdotto il concetto di cointegrazione . È stato mostrato come coppie di azioni o ETF cointegrate potrebbero portare a opportunità redditizie di trading mean-reverting.

Sono stati descrtitti due test specifici, il test Cointegrated Augmented Dickey-Fuller (CADF) e il test di Johansen, che possono aiutare a identificare statisticamente i portafogli cointegrati.

In questo articolo usiamo DataTrader per implementare una vera e propria strategia di trading basata su una (potenziale) relazione di cointegrazione tra un’azione e un ETF nel mercato delle materie prime.

L’analisi inizia formulando un’ipotesi su una relazione strutturale tra i prezzi di Alcoa Inc., grande produttore di alluminio, e il gas naturale statunitense. Questa relazione strutturale è verificata per la cointegrazione tramite il test CADF utilizzando Python. Vediamo che sebbene i prezzi appaiano parzialmente correlati, l’ipotesi nulla di nessuna relazione di cointegrazione non può essere rifiutata.

Quindi calcoliamo un hedge ratio statico tra le due serie e sviluppiamo una strategia di trading, prima di tutto per mostrare come questo tipo di strategia può essere implementata in DataTrader, indipendentemente dalla performance, e quindi valutare la performance di una coppia di asset leggermente correlati, ma non cointegrate.

Questa strategia è stata ispirata dalla famosa strategia di cointegrazione GLD-GDX di Ernie Chan [1] e da un post [2] del CEO di Quantopian, John Fawcett, che fa riferimento alla fusione dell’alluminio come potenziale per asset cointegrati.

Le ipotesi

Un insieme estremamente importante di processi nell’ingegneria chimica sono il processo Bayer e il processo Hall–Héroult. Sono i passaggi chiave nella fusione dell’alluminio dal minerale grezzo della bauxite, tramite la tecnica dell’elettrolisi.

L’elettrolisi richiede una notevole quantità di elettricità, gran parte della quale è generata da carbone, energia idroelettrica, nucleare o da turbine a gas a ciclo combinato (CCGT) . Quest’ultimo richiede il gas naturale come principale fonte di combustibile. Poiché l’acquisto di gas naturale per la fusione dell’alluminio è probabilmente un costo notevole per i produttori di alluminio, la loro redditività deriva in parte dal prezzo del gas naturale.

L’ipotesi da verificare è che il prezzo delle azioni di un grande produttore di alluminio, come Alcoa Inc. (ARNC) e quello di un ETF che rappresenta i prezzi del gas naturale negli Stati Uniti, come UNG, potrebbero essere cointegrati e quindi portare a una potenziale strategia mean-reverting di trading sistematico.

Test di cointegrazione in Python

Il tema della cointegrazione, è stato c’è descritto ed approfondito  su DataTrading.info nei seguenti articoli:

Per verificare l’ipotesi di cui sopra la procedura Cointegrated Augmented Dickey Fuller verrà eseguita su ARNC e UNG tramite Python. La procedura è stata descritta in modo approfondito negli articoli precedenti e quindi il codice verrà qui replicato con meno spiegazioni.

Il primo passo è importare la libreria yfinance di Python, per il download dei dati e la libreria statsmodels, per il test ADF. I dati giornalieri della barra di ARNC e UNG vengono scaricati per il periodo dall’11 novembre 2014 al 1 gennaio 2017. I dati vengono quindi impostati sui valori di chiusura rettificati (gestione di split/dividendi):

				
					import yfinance as yf

ARNC = yf.download('ARNC', start="2014-11-11", end="2017-01-01")
UNG = yf.download('UNG', start="2014-11-11", end="2017-01-01")

ARNC_adjcls = ARNC['Adj Close']
UNG_adjcls = UNG['Adj Close']
				
			

Di seguito viene visualizzato un grafico dei prezzi di ARNC (blu) e UNG (rosso) nel periodo:

				
					import matplotlib.pyplot as plt

plt.plot(ARNC_adjcls, 'b')
plt.plot(UNG_adjcls, 'r')
plt.show()
				
			
trading-algoritmico-datatrader-coint-arnc-ung-plot

Si può vedere che i prezzi di ARNC e UNG seguono uno schema sostanzialmente simile, che tende al ribasso per il 2015 e poi rimane piatto per il 2016. La visualizzazione di un grafico a dispersione fornirà un quadro più chiaro di qualsiasi potenziale correlazione:

				
					plt.scatter(ARNC_adjcls.values, UNG_adjcls.values)
plt.show()
				
			
trading-algoritmico-datatrader-coint-aluminium-scatterplot

Il grafico a dispersione è più ambiguo. C’è una leggera correlazione parziale positiva, come ci si aspetterebbe per un’azienda fortemente esposta ai prezzi del gas naturale, ma non è chiaro se ciò sia sufficiente per consentire un rapporto strutturale.

Effettuando una regressione lineare tra i due, dove UNG è la variabile indipendente, si ottiene un rapporto coefficiente di pendenza/copertura pari a 1.213. Il compito finale è eseguire il test ADF e determinare se esiste una relazione di cointegrazione strutturale. Si ottiene un coefficiente Dickey-Fuller pari a -2.5413 e un p-value di 0.3492.

Questa analisi mostra che non ci sono prove sufficienti per rifiutare l’ipotesi nulla di nessuna relazione di cointegrazione. Tuttavia, nonostante ciò, è istruttivo continuare ad attuare la strategia con il rapporto di copertura sopra calcolato, per due motivi:

  1. In primo luogo, qualsiasi altra potenziale analisi basata sulla cointegrazione, come derivata dal CADF o dal test di Johansen, può essere sottoposta a backtesting utilizzando il codice seguente, poiché è stato scritto per essere sufficientemente flessibile da far fronte a ampi portafogli di cointegrazione.
  2. In secondo luogo, è utile vedere come si comporta una strategia quando non ci sono prove sufficienti per rifiutare l’ipotesi nulla. Forse la coppia è ancora scambiabile anche se non è stata rilevata una relazione su questo piccolo set di dati.

Descriviamo ora il meccanismo della strategia di trading.

La strategia di trading

Al fine di generare effettivamente segnali di trading mean-reverting da uno “spread” dei prezzi da una combinazione lineare di ARNC e UNG, usiamo utilizzata una tecnica nota come Bande di Bollinger .

Le bande di Bollinger prevedono di considerare una media mobile semplice di una serie di prezzi e quindi la formazione di “bande” che circondano la serie che sono un multiplo scalare della deviazione standard mobile della serie di prezzi. Il periodo della finestra mobile è identico per la media mobile e per la deviazione standard. In sintesi le bande di bollinger sono una stima della volatilità storica di una serie di prezzi.

Per definizione, una serie mean-reverting si discosterà occasionalmente dalla sua media e per tornare ai suoi valori medi. Le bande di Bollinger forniscono un meccanismo per entrare e uscire dal mercato usando “soglie” di deviazione standard come segnali per entrare e uscire.

Per generare le operazioni, il primo compito è calcolare uno z-score/punteggio standard dell’ultimo prezzo di spread. Questo si ottiene prendendo l’ultimo prezzo di mercato del portafoglio , sottraendo la media mobile e dividendo per la deviazione standard mobile (come descritto sopra).

Una volta calcolato questo punteggio z, una posizione verrà aperta o chiusa alle seguenti condizioni:

  • \(z_{\text{score}} \lt -z_{\text{entry}}\) – Entrata lunga
  • \(z_{\text{score}} \gt +z_{\text{entry}}\) – Entrata breve
  • \(z_{\text{score}} \ge -z_{\text{exit}}\) – Lunga chiusura
  • \(z_{\text{score}} \le +z_{\text{exit}}\) – Chiusura corta

Dove \(z_{\text{score}}\) è l’ultimo prezzo di spread standardizzato, \(z_{\text{entry}}\) è la soglia di accesso al trading e \(z_{\text{exit}}\) è la soglia di uscita dal trading.

Dati

Per attuare questa strategia è necessario disporre di dati sui prezzi OHLCV giornalieri per le azioni e gli ETF nel periodo coperto da questo backtest:

Ticker Nome Periodo Collegamento
ARNC Arconic Inc. (precedente Alcoa Inc.) 11 novembre 2014 – 1 settembre 2016 Yahoo Finanza
UNG ETF sul gas naturale degli Stati Uniti 11 novembre 2014 – 1 settembre 2016 Yahoo Finanza

Se si desidera replicare i risultati, questi dati dovranno essere inseriti nella directory specificata dal file delle impostazioni di DataTrader.

Implementazione Python  con DataTrader

Nota: il codice completo di ciascuno di questi file Python è riportato alla fine dell’articolo.

Da notare inoltre che questa strategia contiene un implicito lookahead bias e quindi le sue prestazioni saranno grossolanamente aumentate rispetto a un’implementazione reale. Il lookahead bias si verifica a causa dell’uso del calcolo del hedge ratio sullo stesso campione di dati su cui viene simulata la strategia di trading. In un’implementazione reale saranno necessari due insiemi di dati separati per verificare che qualsiasi relazione strutturale persista out-of-sample.

L’implementazione della strategia è simile ad altre strategie di DataTrader. Implica la creazione di una sottoclasse di AbstractStrategy nel file coint_bollinger_strategy.py. Questa classe viene quindi utilizzata dal file coint_bollinger_backtest.py per simulare effettivamente il backtest.

Descriviamo inizialmente coint_bollinger_strategy.py. Importiamo la libreria NumPy, così come le librerie di DataTrader necessarie per la gestione di segnali e strategie. Usiamo PriceParser per regolare la gestione interna del meccanismo di memorizzazione dei prezzi di DataTrader, in modo da evitare errori di arrotondamento in virgola mobile.

Importiamo la classe deque – coda a doppia estremità – per memorizzare una finestra mobile dei prezzi di chiusura, necessaria per i calcoli della media mobile e della deviazione standard.

				
					
from collections import deque
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
				
			

Il passo successivo è definire la  sottoclasse CointegrationBollingerBandsStrategydi AbstractStrategy, che effettua la generazione dei segnali. Come per tutte le strategie, richiede un elenco di tickers da analizzare e negoziare e una coda events_queue dove posizionare gli oggetti SignalEvent.

Questa sottoclasse richiede parametri aggiuntivi. lookback è il numero intero di barre su cui eseguire i calcoli della media mobile mobile e della deviazione standard. weights è l’insieme dei rapporti di copertura fissi, cioè i componenti primari degli autovettori del test di Johansen, da utilizzare come “unità” di un portafoglio di attività di cointegrazione.

entry_z descrive il multiplo della soglia di ingresso dello z-score (ovvero il numero di deviazioni standard) per aprire un trade. exit_z è il numero corrispondente di deviazioni standard per uscire dall’operazione. base_quantity è il numero intero di “unità” del portafoglio da negoziare.

Inoltre la classe tiene traccia degli ultimi prezzi di tutti i ticker in un separato array self.latest_prices. La variabile self.port_mkt_value contiene una coda a doppia fine costituita dagli ultimi valori lookback dei prezzi di mercato di una “unità” del portafoglio. Il flag self.invested consente alla strategia stessa di tenere traccia se è “a mercato” o meno:

				
					
class CointegrationBollingerBandsStrategy(AbstractStrategy):
    """
    Requisiti:
    tickers - La lista dei simboli dei ticker
    events_queue - Manager del sistema della coda degli eventi
    lookback - Periodo di lookback per la media mobile e deviazione standard
    weights - Il vettore dei pesi che descrive le "unità" del portafoglio
    entry_z - La soglia z-score per entrare nel trade
    exit_z - La soglia z-score per uscire dal trade
    base_quantity - Numero di "unità" di un portafoglio che sono negoziate
    """
    def __init__(
        self, tickers, events_queue, 
        lookback, weights, entry_z, exit_z,
        base_quantity
    ):
        self.tickers = tickers
        self.events_queue = events_queue      
        self.lookback = lookback
        self.weights = weights
        self.entry_z = entry_z
        self.exit_z = exit_z
        self.qty = base_quantity
        self.time = None
        self.latest_prices = np.full(len(self.tickers), -1.0)
        self.port_mkt_val = deque(maxlen=self.lookback)
        self.invested = None
        self.bars_elapsed = 0
				
			

Il seguente metodo _set_correct_time_and_price è simile a quello descritto nell’articolo relativo al filtro di Kalman con DataTrader. L’obiettivo di questo metodo è assicurarsi che l’array self.latest_prices sia popolato con gli ultimi valori di mercato di ciascun ticker. La strategia viene eseguita solo se questo array contiene un set completo di prezzi, tutti contenenti lo stesso timestamp (ovvero che rappresentano lo stesso timeframe di una barra).

La versione precedente di questo metodo prevede un array con dimensione fissa per gestire i prezzi di due ticker, mentre il codice seguente opera con un numero di ticker, necessario per la cointegrazione di portafogli che potrebbero contenere tre o più asset:

				
					    
    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:
            for i in range(0, len(self.tickers)):
                if event.ticker == self.tickers[i]:
                    self.latest_prices[i] = price
        else:
            self.time = event.time
            self.bars_elapsed += 1
            self.latest_prices = np.full(len(self.tickers), -1.0)
            for i in range(0, len(self.tickers)):
                if event.ticker == self.tickers[i]:
                    self.latest_prices[i] = price
				
			
go_long_units è un metodo di supporto che permette di andare long con l’appropriata quantità di “unità” di portafoglio, acquistando separatamente i singoli componenti nelle corrette quantità. Questo scopo è raggiunto cortocircuitando qualsiasi componente che ha un valore negativo  nell’array self.weights e andando long per qualsiasi componente che ha un valore positivo. Da notare che si moltiplica per il valore self.qty, che è il numero base di unità da negoziare per una “unità” di portafoglio:
				
					 
    def go_long_units(self):
        """
        Andiamo long con il numero appropriato di "unità" del portafoglio 
        per aprire una nuova posizione o chiudere una posizione short.
        """
        for i, ticker in enumerate(self.tickers):
            if self.weights[i] < 0.0:
                self.events_queue.put(SignalEvent(
                    ticker, "SLD",
                    int(floor(-1.0 * self.qty * self.weights[i])))
                )
            else:
                self.events_queue.put(SignalEvent(
                    ticker, "BOT",
                    int(floor(self.qty * self.weights[i])))
                )
				
			
Il metodo go_short_units è quasi identico al metodo precedente ad eccezione dell’inversione delle istruzioni long/short, in modo che le posizioni possano essere chiuse o andare short:
				
					
    def go_short_units(self):
        """
        Andare short del numero appropriato di "unità" del portafoglio 
        per aprire una nuova posizione o chiudere una posizione long.
        """
        for i, ticker in enumerate(self.tickers):
            if self.weights[i] < 0.0:
                self.events_queue.put(SignalEvent(
                    ticker, "BOT",
                    int(floor(-1.0 * self.qty * self.weights[i])))
                )
            else:
                self.events_queue.put(SignalEvent(
                    ticker, "SLD",
                    int(floor(self.qty * self.weights[i])))
                )
				
			

Il metodo zscore_trade prende l’ultimo z-score calcolato dal prezzo di mercato del portafoglio e lo utilizza per andare long, short o chiudere un trade. Il seguente codice racchiude la logica “Band di Bollinger” della strategia.

Se lo z-score è inferiore alla soglia negativa di ingresso, si crea una posizione long. Se lo z-score è maggiore della soglia positiva di ingresso, si crea una posizione short. Di conseguenza, se la strategia è già sul mercato e lo z-score è superiore alla soglia di uscita negativa, qualsiasi posizione long viene chiusa. Se la strategia è già sul mercato e lo z-score è inferiore alla soglia di uscita positiva, viene chiusa una posizione short:

				
					   
   def zscore_trade(self, zscore, event):
        """
        Determina il trade se la soglia dello zscore di 
        entrata o di uscita è stata superata.
        """
        # Se non siamo a mercato
        if self.invested is None:
            if zscore < -self.entry_z:
                # Entrata Long
                print("LONG: %s" % event.time)
                self.go_long_units()
                self.invested = "long"
            elif zscore > self.entry_z:
                # Entrata Short
                print("SHORT: %s" % event.time)
                self.go_short_units()
                self.invested = "short"
        # Se siamo a mercato
        if self.invested is not None:
            if self.invested == "long" and zscore >= -self.exit_z:
                print("CLOSING LONG: %s" % event.time)
                self.go_short_units()
                self.invested = None
            elif self.invested == "short" and zscore <= self.exit_z:
                print("CLOSING SHORT: %s" % event.time)
                self.go_long_units()
                self.invested = None

				
			

Infine, il  metodo calculate_signals assicura che l’array self.latest_prices sia completamente aggiornato ed effettua operazioni solo se tutti i ticket sono stati aggiornati con i prezzi più recenti. Se questi prezzi esistono, la coda  self.port_mkt_val viene aggiornata per contenere l’ultimo “valore di mercato” di un portafoglio di asset. Questo è semplicemente il prodotto scalare degli ultimi prezzi di ciascun componente e del relativo vettore dei pesi.

Lo z-score dell’ultimo valore di mercato dell’unità di portafoglio è calcolato sottraendo la media mobile e dividendo per la deviazione standard mobile. Il valore dello z-score è quindi inviato al precedente metodo zscore_trade per generare i segnali di trading:

				
					
    def calculate_signals(self, event):
        """
        Calcula i segnali della strategia.
        """
        if event.type == EventType.BAR:
            self._set_correct_time_and_price(event)

            # Operiamo sono se abbiamo tutti i prezzi
            if all(self.latest_prices > -1.0):
                # Calcoliamo il valore di mercato del portfolio tramite il prodotto
                # cartesiamo dei prezzi degli ETF e dei relativi pesi nel portafoglio
                self.port_mkt_val.append(
                    np.dot(self.latest_prices, self.weights)
                )
                # Se ci sono sufficienti dati per formare una completa finestra di ricerca, 
                # calcola lo zscore ed esegue le rispettive operazioni se le soglie vengono superate
                if self.bars_elapsed > self.lookback:
                    zscore = (self.port_mkt_val[-1] - np.mean(self.port_mkt_val)
                             ) / np.std(self.port_mkt_val)
                    self.zscore_trade(zscore, event)
				
			

Il file successivo è coint_bollinger_backtest.py, che incapsula la classe della strategia nella logica di backtesting. È estremamente simile a tutti gli altri file di backtest di DataTrader descritti sul sito. Mentre  il codice completo è riportato alla fine dell’articolo, il seguente snippet evidenzia all’aspetto importante, la creazione dell’oggettoCointegrationBollingerBandsStrategy.

L’array weights è hardcoded dai valori ottenuti dalla procedura CADF descritta in precedenza, mentre il  periodo di lookback è (arbitrariamente) impostato su 15 valori. Le soglie dello z-score di entrata e di uscita sono rispettivamente di 1,5 e 0,5 deviazioni standard. Dato che il capitale iniziale è fissato a 500.000 USD, le base_quantity delle azioni sono fissate a 10.000.

Questi valori possono essere tutti testati e ottimizzati, ad es. attraverso una procedura di ricerca nella griglia, se lo si desidera.

				
					
# Uso della strategia di trading Cointegration Bollinger Bands
weights = np.array([1.0, -1.213])
lookback = 15
entry_z = 1.5
exit_z = 0.5
base_quantity = 10000
strategy = CointegrationBollingerBandsStrategy(
    tickers, events_queue,
    lookback, weights,
    entry_z, exit_z, base_quantity
)
strategy = Strategies(strategy, DisplayStrategy())

				
			

Per eseguire il backtest è necessaria un’installazione funzionante di DataTrader e i due file sopra descritti devono essere collocati nella stessa directory. Supponendo la disponibilità dei dati ARNC e UNG, il backtest verrà eseguito digitando il seguente comando nel terminale:

				
					$ python coint_bollinger_backtest.py

				
			

Riceverai il seguente output (estratto):

				
					..
..
Backtest complete.
Sharpe Ratio: 1.22071888063
Max Drawdown: 0.0701967400339
Max Drawdown Pct: 0.0701967400339

				
			

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 . Non tengono conto delle differenze di commissione per gli ETF, ma sono ragionevolmente rappresentative di ciò che potrebbe essere ottenuto da una vera strategia di trading.

Tearsheet

trading-algoritmico-datatrader-coint-aluminium-tearsheet

Ricordiamo ancora una volta che questa strategia contiene un implicito lookahead bias dovuto al fatto che la procedura CADF è stata eseguita sullo stesso campione di dati della strategia di trading.

Con questo in mente, la strategia pubblica uno Sharpe Ratio di 1,22, con un drawdown giornaliero massimo del 7,02%. La maggior parte dei guadagni della strategia si verifica in un solo mese entro gennaio 2015, dopodiché la strategia si comporta male. Rimane in drawdown per tutto il 2016. Ciò non sorprende poiché non è stata trovata alcuna relazione di cointegrazione statisticamente significativa tra ARNC e UNG durante il periodo studiato, almeno utilizzando la procedura del test ADF.

Per migliorare questa strategia si potrebbe approfondire l’economia del processo di fusione dell’alluminio. Ad esempio, sebbene vi sia una chiara necessità di elettricità per eseguire il processo di elettrolisi, questa energia può essere derivata da molte fonti, tra cui idroelettrica, carbone, nucleare e, probabilmente in futuro, eolica e solare. Potrebbe essere preso in considerazione un portafoglio più completo che includa il prezzo dell’alluminio, i grandi produttori di alluminio e gli ETF che rappresentano diverse fonti di energia.

Riferimenti

Codice completo

Per il codice completo riportato in questo articolo, utilizzando il framework open-source di backtesting event-driven DataTrader si può consultare il seguente repository di github:
https://github.com/datatrading-info/DataTrader

				
					# coint_bollinger_strategy.py

from collections import deque
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


class CointegrationBollingerBandsStrategy(AbstractStrategy):
    """
    Requisiti:
    tickers - La lista dei simboli dei ticker
    events_queue - Manager del sistema della coda degli eventi
    lookback - Periodo di lookback per la media mobile e deviazione standard
    weights - Il vettore dei pesi che descrive le "unità" del portafoglio
    entry_z - La soglia z-score per entrare nel trade
    exit_z - La soglia z-score per uscire dal trade
    base_quantity - Numero di "unità" di un portafoglio che sono negoziate
    """
    def __init__(
        self, tickers, events_queue,
        lookback, weights, entry_z, exit_z,
        base_quantity
    ):
        self.tickers = tickers
        self.events_queue = events_queue
        self.lookback = lookback
        self.weights = weights
        self.entry_z = entry_z
        self.exit_z = exit_z
        self.qty = base_quantity
        self.time = None
        self.latest_prices = np.full(len(self.tickers), -1.0)
        self.port_mkt_val = deque(maxlen=self.lookback)
        self.invested = None
        self.bars_elapsed = 0

    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:
            for i in range(0, len(self.tickers)):
                if event.ticker == self.tickers[i]:
                    self.latest_prices[i] = price
        else:
            self.time = event.time
            self.bars_elapsed += 1
            self.latest_prices = np.full(len(self.tickers), -1.0)
            for i in range(0, len(self.tickers)):
                if event.ticker == self.tickers[i]:
                    self.latest_prices[i] = price

    def go_long_units(self):
        """
        Andiamo long con il numero appropriato di "unità" del portafoglio
        per aprire una nuova posizione o chiudere una posizione short.
        """
        for i, ticker in enumerate(self.tickers):
            if self.weights[i] < 0.0:
                self.events_queue.put(SignalEvent(
                    ticker, "SLD",
                    int(floor(-1.0 * self.qty * self.weights[i])))
                )
            else:
                self.events_queue.put(SignalEvent(
                    ticker, "BOT",
                    int(floor(self.qty * self.weights[i])))
                )

    def go_short_units(self):
        """
        Andare short del numero appropriato di "unità" del portafoglio
        per aprire una nuova posizione o chiudere una posizione long.
        """
        for i, ticker in enumerate(self.tickers):
            if self.weights[i] < 0.0:
                self.events_queue.put(SignalEvent(
                    ticker, "BOT",
                    int(floor(-1.0 * self.qty * self.weights[i])))
                )
            else:
                self.events_queue.put(SignalEvent(
                    ticker, "SLD",
                    int(floor(self.qty * self.weights[i])))
                )

    def zscore_trade(self, zscore, event):
        """
        Determina il trade se la soglia dello zscore di
        entrata o di uscita è stata superata.
        """
        # Se non siamo a mercato
        if self.invested is None:
            if zscore < -self.entry_z:
                # Entrata Long
                print("LONG: %s" % event.time)
                self.go_long_units()
                self.invested = "long"
            elif zscore > self.entry_z:
                # Entrata Short
                print("SHORT: %s" % event.time)
                self.go_short_units()
                self.invested = "short"
        # Se siamo a mercato
        if self.invested is not None:
            if self.invested == "long" and zscore >= -self.exit_z:
                print("CLOSING LONG: %s" % event.time)
                self.go_short_units()
                self.invested = None
            elif self.invested == "short" and zscore <= self.exit_z:
                print("CLOSING SHORT: %s" % event.time)
                self.go_long_units()
                self.invested = None

    def calculate_signals(self, event):
        """
        Calcula i segnali della strategia.
        """
        if event.type == EventType.BAR:
            self._set_correct_time_and_price(event)

            # Operiamo sono se abbiamo tutti i prezzi
            if all(self.latest_prices > -1.0):
                # Calcoliamo il valore di mercato del portfolio tramite il prodotto
                # cartesiamo dei prezzi degli ETF e dei relativi pesi nel portafoglio
                self.port_mkt_val.append(
                    np.dot(self.latest_prices, self.weights)
                )
                # Se ci sono sufficienti dati per formare una completa finestra di ricerca,
                # calcola lo zscore ed esegue le rispettive operazioni se le soglie vengono superate
                if self.bars_elapsed > self.lookback:
                    zscore = (self.port_mkt_val[-1] - np.mean(self.port_mkt_val)
                             ) / np.std(self.port_mkt_val)
                    self.zscore_trade(zscore, event)
				
			
				
					# coint_bollinger_backtest.py


import datetime

import click
import numpy as np

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 coint_bollinger_strategy import CointegrationBollingerBandsStrategy


def run(config, testing, tickers, filename):

    # Impostazione delle variabili necessarie per il backtest
    events_queue = queue.Queue()
    csv_dir = config.CSV_DATA_DIR
    initial_equity = PriceParser.parse(500000.00)

    # Uso del manager Yahoo Daily Price
    start_date = datetime.datetime(2015, 1, 1)
    end_date = datetime.datetime(2016, 9, 1)
    price_handler = YahooDailyCsvBarPriceHandler(
        csv_dir, events_queue, tickers,
        start_date=start_date, end_date=end_date
    )

    # Uso della strategia Cointegration Bollinger Bands
    weights = np.array([1.0, -1.213])
    lookback = 15
    entry_z = 1.5
    exit_z = 0.5
    base_quantity = 10000
    strategy = CointegrationBollingerBandsStrategy(
        tickers, events_queue,
        lookback, weights,
        entry_z, exit_z, base_quantity
    )
    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
    title = ["aluminum Smelting Strategy - ARNC/UNG"]
    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 = settings.DEFAULT_CONFIG_FILENAME
    testing = False
    tickers = 'ARNC,UNG'
    filename = None
    main(config, testing, tickers, filename)

				
			
Torna su