Simulazione di trading ad alta frequenza con Analisi di flusso

Analisi di flusso di Azure supporta l'analisi avanzata tramite la combinazione di linguaggio SQL, funzioni definite dall'utente JavaScript e aggregazioni definite dall'utente. L'analisi avanzata comprende l'addestramento online e la valutazione dei modelli di apprendimento automatico, nonché la simulazione di processi con stato. Questo articolo descrive come eseguire la regressione lineare in un processo di Analisi di flusso di Azure che esegue il training continuo e l'assegnazione di punteggi in uno scenario di trading ad alta frequenza.

Prerequisiti

• Una sottoscrizione di Azure. Se non ne hai uno, crea un account gratuito.
• Un processo di Analisi di flusso di Azure.
• Uno spazio dei nomi di Hub eventi di Azure e un hub eventi.
• Familiarità con il linguaggio di query di Stream Analytics.
• (Facoltativo) Un account Power BI se si vuole visualizzare l'output.

Flusso di lavoro di trading ad alta frequenza

Il flusso logico del trading ad alta frequenza è:

1. Ottenere quotazioni in tempo reale da una borsa valori.
2. Creazione di un modello predittivo intorno alle quotazioni per anticipare il movimento dei prezzi.
3. L'inserimento di ordini di acquisto o vendita per fare soldi dalla previsione corretta dei movimenti di prezzo.

Questo scenario richiede:

• Feed di quotazioni in tempo reale.
• Modello predittivo in grado di operare sulle quotazioni in tempo reale.
• Simulazione di trading che dimostra il profitto o la perdita dell'algoritmo di trading.

Feed di quotazioni in tempo reale

Gli investitori Exchange (IEX) in precedenza offrivano offerte di offerta gratuita real-time bid e quotazioni ask utilizzando socket.io. È possibile scrivere una semplice applicazione console per ricevere quotazioni in tempo reale e inviarle ad Hub eventi di Azure come origine dati. Il codice seguente è uno scheletro del programma. Il codice omette la gestione degli errori per brevità. È anche necessario includere i SocketIoClientDotNet pacchetti NuGet e WindowsAzure.ServiceBus nel progetto.

using Quobject.SocketIoClientDotNet.Client;
using Microsoft.ServiceBus.Messaging;
var symbols = "msft,fb,amzn,goog";
var eventHubClient = EventHubClient.CreateFromConnectionString(connectionString, eventHubName);
var socket = IO.Socket("https://ws-api.iextrading.com/1.0/tops");
socket.On(Socket.EVENT_MESSAGE, (message) =>
{
    eventHubClient.Send(new EventData(Encoding.UTF8.GetBytes((string)message)));
});
socket.On(Socket.EVENT_CONNECT, () =>
{
    socket.Emit("subscribe", symbols);
});

Ecco alcuni eventi di esempio generati:

{"symbol":"MSFT","marketPercent":0.03246,"bidSize":100,"bidPrice":74.8,"askSize":300,"askPrice":74.83,"volume":70572,"lastSalePrice":74.825,"lastSaleSize":100,"lastSaleTime":1506953355123,"lastUpdated":1506953357170,"sector":"softwareservices","securityType":"commonstock"}
{"symbol":"GOOG","marketPercent":0.04825,"bidSize":114,"bidPrice":870,"askSize":0,"askPrice":0,"volume":11240,"lastSalePrice":959.47,"lastSaleSize":60,"lastSaleTime":1506953317571,"lastUpdated":1506953357633,"sector":"softwareservices","securityType":"commonstock"}
{"symbol":"MSFT","marketPercent":0.03244,"bidSize":100,"bidPrice":74.8,"askSize":100,"askPrice":74.83,"volume":70572,"lastSalePrice":74.825,"lastSaleSize":100,"lastSaleTime":1506953355123,"lastUpdated":1506953359118,"sector":"softwareservices","securityType":"commonstock"}
{"symbol":"FB","marketPercent":0.01211,"bidSize":100,"bidPrice":169.9,"askSize":100,"askPrice":170.67,"volume":39042,"lastSalePrice":170.67,"lastSaleSize":100,"lastSaleTime":1506953351912,"lastUpdated":1506953359641,"sector":"softwareservices","securityType":"commonstock"}
{"symbol":"GOOG","marketPercent":0.04795,"bidSize":100,"bidPrice":959.19,"askSize":0,"askPrice":0,"volume":11240,"lastSalePrice":959.47,"lastSaleSize":60,"lastSaleTime":1506953317571,"lastUpdated":1506953360949,"sector":"softwareservices","securityType":"commonstock"}
{"symbol":"FB","marketPercent":0.0121,"bidSize":100,"bidPrice":169.9,"askSize":100,"askPrice":170.7,"volume":39042,"lastSalePrice":170.67,"lastSaleSize":100,"lastSaleTime":1506953351912,"lastUpdated":1506953362205,"sector":"softwareservices","securityType":"commonstock"}
{"symbol":"GOOG","marketPercent":0.04795,"bidSize":114,"bidPrice":870,"askSize":0,"askPrice":0,"volume":11240,"lastSalePrice":959.47,"lastSaleSize":60,"lastSaleTime":1506953317571,"lastUpdated":1506953362629,"sector":"softwareservices","securityType":"commonstock"}

Modello predittivo per il trading ad alta frequenza

In questa dimostrazione, l'esempio utilizza un modello lineare descritto in Order Imbalance Based Strategy in High Frequency Algorithmic Trading.

Lo squilibrio dell'ordine dei volumi (VOI) è una funzione del prezzo corrente dell'offerta/richiesta e del volume e del prezzo dell'offerta/richiesta e del volume dall'ultimo tick. Il documento identifica la correlazione tra VOI e il movimento futuro dei prezzi. Crea un modello lineare tra i cinque valori VOI precedenti e la variazione del prezzo nei successivi 10 tick. Il modello viene addestrato sui dati del giorno precedente mediante regressione lineare.

Il modello addestrato effettua quindi in tempo reale previsioni delle variazioni di prezzo per le quotazioni nella giornata di trading corrente. Quando il modello prevede una variazione di prezzo sufficiente, esegue un commercio. A seconda dell'impostazione della soglia, un singolo titolo potrebbe generare migliaia di scambi durante una giornata di trading.

Le sezioni seguenti mostrano come esprimere le operazioni di training e previsione in un processo di Analisi di flusso di Azure. La query completa è una singola WITH istruzione composta da espressioni di tabella comuni (CTEs) che formano una pipeline:

Ripulire e convertire i campi di input delle citazioni

La prima CTE nella query di Analisi di flusso di Azure converte i dati grezzi delle quotazioni da Event Hubs in colonne SQL correttamente tipizzate. DATEADD converte il timestamp Unix (millisecondi) in data e ora. TRY_CAST converte i tipi di dati senza causare il fallimento della query. Convertire i campi di input nei tipi di dati previsti per evitare comportamenti imprevisti durante la manipolazione o il confronto dei campi.

WITH
typeconvertedquotes AS (
    /* convert all input fields to proper types */
    SELECT
        System.Timestamp AS lastUpdated,
        symbol,
        DATEADD(millisecond, CAST(lastSaleTime as bigint), '1970-01-01T00:00:00Z') AS lastSaleTime,
        TRY_CAST(bidSize as bigint) AS bidSize,
        TRY_CAST(bidPrice as float) AS bidPrice,
        TRY_CAST(askSize as bigint) AS askSize,
        TRY_CAST(askPrice as float) AS askPrice,
        TRY_CAST(volume as bigint) AS volume,
        TRY_CAST(lastSaleSize as bigint) AS lastSaleSize,
        TRY_CAST(lastSalePrice as float) AS lastSalePrice
    FROM quotes TIMESTAMP BY DATEADD(millisecond, CAST(lastUpdated as bigint), '1970-01-01T00:00:00Z')
),
timefilteredquotes AS (
    /* filter between 7am and 1pm PST, 14:00 to 20:00 UTC */
    /* clean up invalid data points */
	SELECT * FROM typeconvertedquotes
	WHERE DATEPART(hour, lastUpdated) >= 14 AND DATEPART(hour, lastUpdated) < 20 AND bidSize > 0 AND askSize > 0 AND bidPrice > 0 AND askPrice > 0
),

Recuperare i valori di graduazione precedenti con LAG

Il CTE successivo nella query di Analisi di flusso di Azure usa la funzione LAG per ottenere il prezzo dell'offerta/richiesta e le dimensioni del segno di graduazione precedente per ogni simbolo azionario. Viene scelta arbitrariamente un'ora del valore LIMIT DURATION . Data la frequenza delle virgolette, è possibile trovare il segno di spunta precedente guardando indietro di un'ora.

shiftedquotes AS (
    /* get previous bid/ask price and size in order to calculate VOI */
	SELECT
		symbol,
		(bidPrice + askPrice)/2 AS midPrice,
		bidPrice,
		bidSize,
		askPrice,
		askSize,
		LAG(bidPrice) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS bidPricePrev,
		LAG(bidSize) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS bidSizePrev,
		LAG(askPrice) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS askPricePrev,
		LAG(askSize) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS askSizePrev
	FROM timefilteredquotes
),

Calcolare lo squilibrio del volume degli ordini (VOI)

Il CTE successivo calcola il valore VOI a partire dai dati bid/ask del tick corrente e di quello precedente. La query esclude i valori null nei casi in cui non esiste alcun tick precedente.

currentPriceAndVOI AS (
    /* calculate VOI */
	SELECT
		symbol,
		midPrice,
		(CASE WHEN (bidPrice < bidPricePrev) THEN 0
            ELSE (CASE WHEN (bidPrice = bidPricePrev) THEN (bidSize - bidSizePrev) ELSE bidSize END)
         END) -
        (CASE WHEN (askPrice < askPricePrev) THEN askSize
            ELSE (CASE WHEN (askPrice = askPricePrev) THEN (askSize - askSizePrev) ELSE 0 END)
         END) AS VOI
	FROM shiftedquotes
	WHERE
		bidPrice IS NOT NULL AND
		bidSize IS NOT NULL AND
		askPrice IS NOT NULL AND
		askSize IS NOT NULL AND
		bidPricePrev IS NOT NULL AND
		bidSizePrev IS NOT NULL AND
		askPricePrev IS NOT NULL AND
		askSizePrev IS NOT NULL
),

Creare sequenze di funzionalità per il training del modello

Il CTE successivo usa nuovamente LAG per creare una sequenza con 2 valori VOI consecutivi, seguiti da 10 valori consecutivi a metà prezzo. Queste sequenze formano i dati di training per il modello di regressione lineare.

shiftedPriceAndShiftedVOI AS (
    /* get 10 future prices and 2 previous VOIs */
    SELECT
		symbol,
		midPrice AS midPrice10,
		LAG(midPrice, 1) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS midPrice9,
		LAG(midPrice, 2) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS midPrice8,
		LAG(midPrice, 3) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS midPrice7,
		LAG(midPrice, 4) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS midPrice6,
		LAG(midPrice, 5) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS midPrice5,
		LAG(midPrice, 6) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS midPrice4,
		LAG(midPrice, 7) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS midPrice3,
		LAG(midPrice, 8) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS midPrice2,
		LAG(midPrice, 9) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS midPrice1,
		LAG(midPrice, 10) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS midPrice,
		LAG(VOI, 10) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS VOI1,
		LAG(VOI, 11) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS VOI2
	FROM currentPriceAndVOI
),

Modellare i dati in vettori di funzionalità

Il CTE successivo rimodella le sequenze di prezzo e VOI in vettori di caratteristiche per un modello lineare a due variabili, in cui i valori VOI sono le variabili indipendenti (x1, x2) e la variazione media futura del prezzo è la variabile dipendente (y). Gli eventi con dati incompleti vengono filtrati.

modelInput AS (
    /* create feature vector, x being VOI, y being delta price */
	SELECT
		symbol,
		(midPrice1 + midPrice2 + midPrice3 + midPrice4 + midPrice5 + midPrice6 + midPrice7 + midPrice8 + midPrice9 + midPrice10)/10.0 - midPrice AS y,
		VOI1 AS x1,
		VOI2 AS x2
	FROM shiftedPriceAndShiftedVOI
	WHERE
		midPrice1 IS NOT NULL AND
		midPrice2 IS NOT NULL AND
		midPrice3 IS NOT NULL AND
		midPrice4 IS NOT NULL AND
		midPrice5 IS NOT NULL AND
		midPrice6 IS NOT NULL AND
		midPrice7 IS NOT NULL AND
		midPrice8 IS NOT NULL AND
		midPrice9 IS NOT NULL AND
		midPrice10 IS NOT NULL AND
		midPrice IS NOT NULL AND
		VOI1 IS NOT NULL AND
		VOI2 IS NOT NULL
),

Addestra il modello di regressione lineare con SUM e AVG

Poiché Analisi di flusso di Azure non dispone di una funzione di regressione lineare predefinita, la query usa SUM e AVG aggregazioni per calcolare i coefficienti (a, b1, b2) per il modello di regressione lineare a due variabili. Il modello viene riaddestrato ogni giorno utilizzando una finestra temporale fissa di 24 ore.

modelagg AS (
    /* get aggregates for linear regression calculation,
     http://faculty.cas.usf.edu/mbrannick/regression/Reg2IV.html */
	SELECT
		symbol,
		SUM(x1 * x1) AS x1x1,
		SUM(x2 * x2) AS x2x2,
		SUM(x1 * y) AS x1y,
		SUM(x2 * y) AS x2y,
		SUM(x1 * x2) AS x1x2,
		AVG(y) AS avgy,
		AVG(x1) AS avgx1,
		AVG(x2) AS avgx2
	FROM modelInput
	GROUP BY symbol, TumblingWindow(hour, 24, -4)
),
modelparambs AS (
    /* calculate b1 and b2 for the linear model */
	SELECT
		symbol,
		(x2x2 * x1y - x1x2 * x2y)/(x1x1 * x2x2 - x1x2 * x1x2) AS b1,
		(x1x1 * x2y - x1x2 * x1y)/(x1x1 * x2x2 - x1x2 * x1x2) AS b2,
		avgy,
		avgx1,
		avgx2
	FROM modelagg
),
model AS (
    /* calculate a for the linear model */
	SELECT
		symbol,
		avgy - b1 * avgx1 - b2 * avgx2 AS a,
		b1,
		b2
	FROM modelparambs
),

Valutare le quotazioni correnti usando il modello del giorno precedente

Per usare il modello di regressione lineare sottoposto a training del giorno precedente per assegnare un punteggio all'evento corrente, la query unisce le virgolette con i coefficienti del modello. Anziché usare JOIN, la query usa UNION per combinare gli eventi del modello e gli eventi di virgolette in un singolo flusso. Usa quindi LAG per associare gli eventi al modello del giorno precedente, in modo da ottenere esattamente una corrispondenza. A causa del weekend, la query considera i tre giorni precedenti (72 ore). Se si usasse un JOIN diretto, si otterrebbero tre modelli per ogni evento di quotazione.

shiftedVOI AS (
    /* get two consecutive VOIs */
	SELECT
		symbol,
		midPrice,
		VOI AS VOI1,		
		LAG(VOI, 1) OVER (PARTITION BY symbol LIMIT DURATION(hour, 1)) AS VOI2
	FROM currentPriceAndVOI
),
VOIAndModel AS (
    /* combine VOIs and models */
	SELECT
		'voi' AS type,
		symbol,
		midPrice,
		VOI1,
		VOI2,
        0.0 AS a,
        0.0 AS b1,
        0.0 AS b2
	FROM shiftedVOI
	UNION
	SELECT
		'model' AS type,
		symbol,
        0.0 AS midPrice,
        0 AS VOI1,
        0 AS VOI2,
		a,
		b1,
		b2
	FROM model
),
VOIANDModelJoined AS (
    /* match VOIs with the latest model within 3 days (72 hours, to take the weekend into account) */
	SELECT
		symbol,
		midPrice,
		VOI1 as x1,
		VOI2 as x2,
		LAG(a, 1) OVER (PARTITION BY symbol LIMIT DURATION(hour, 72) WHEN type = 'model') AS a,
		LAG(b1, 1) OVER (PARTITION BY symbol LIMIT DURATION(hour, 72) WHEN type = 'model') AS b1,
		LAG(b2, 1) OVER (PARTITION BY symbol LIMIT DURATION(hour, 72) WHEN type = 'model') AS b2
	FROM VOIAndModel
	WHERE type = 'voi'
),

Generare segnali commerciali da stime

Le CTE finali calcolano la variazione del prezzo futura prevista (efpc) applicando la formula di regressione lineare (a + b1 * x1 + b2 * x2) e quindi generano segnali di acquisto/vendita in base a una soglia di ±0,02. Un valore di scambio pari a 10 indica un acquisto. Un valore di scambio pari a -10 corrisponde a vendere.

prediction AS (
    /* make prediction if there is a model */
	SELECT
		symbol,
		midPrice,
		a + b1 * x1 + b2 * x2 AS efpc
	FROM VOIANDModelJoined
	WHERE
		a IS NOT NULL AND
		b1 IS NOT NULL AND
		b2 IS NOT NULL AND
        x1 IS NOT NULL AND
        x2 IS NOT NULL
),
tradeSignal AS (
    /* generate buy/sell signals */
	SELECT
        DateAdd(hour, -7, System.Timestamp) AS time,
		symbol,		
		midPrice,
        efpc,
		CASE WHEN (efpc > 0.02) THEN 10 ELSE (CASE WHEN (efpc < -0.02) THEN -10 ELSE 0 END) END AS trade,
		DATETIMEFROMPARTS(DATEPART(year, System.Timestamp), DATEPART(month, System.Timestamp), DATEPART(day, System.Timestamp), 0, 0, 0, 0) as date
	FROM prediction
),

Testare la strategia di trading con una simulazione

Dopo aver generato i segnali di trading, verifica quanto è efficace la strategia di trading senza fare trading reale.

Questo test usa un'UDA con una finestra mobile che avanza ogni minuto. Il raggruppamento in base alla data e alla clausola HAVING garantisce che la finestra conti solo gli eventi che appartengono allo stesso giorno. Per una finestra di salto tra due giorni, la data GROUP BY separa il raggruppamento nel giorno precedente e nel giorno corrente. La clausola HAVING esclude le finestre che terminano nel giorno corrente ma sono raggruppate nel giorno precedente.

simulation AS
(
    /* perform trade simulation for the past 7 hours to cover an entire trading day, and generate output every minute */
	SELECT
        DateAdd(hour, -7, System.Timestamp) AS time,
		symbol,
		date,
		uda.TradeSimulation(tradeSignal) AS s
	FROM tradeSignal
	GROUP BY HoppingWindow(minute, 420, 1), symbol, date
	Having DateDiff(day, date, time) < 1 AND DATEPART(hour, time) < 13
)

L'UDA JavaScript inizializza tutti gli accumulatori nella funzione init, calcola la transizione di stato a ogni evento aggiunto alla finestra e restituisce i risultati della simulazione alla fine della finestra. La simulazione detiene o vende allo scoperto 10 azioni di un titolo per ogni operazione. Il costo della transazione è di $8 fissi. La tabella seguente illustra le quattro azioni di trading eseguite dall'UDA:

function main() {
	var TRADE_COST = 8.0;
	var SHARES = 10;
	this.init = function () {
		this.own = false;
		this.pos = 0;
		this.pnl = 0.0;
		this.tradeCosts = 0.0;
		this.buyPrice = 0.0;
		this.sellPrice = 0.0;
		this.buySize = 0;
		this.sellSize = 0;
		this.buyTotal = 0.0;
		this.sellTotal = 0.0;
	}
	this.accumulate = function (tradeSignal, timestamp) {
		if(!this.own && tradeSignal.trade == 10) {
		  // Buy to open
		  this.own = true;
		  this.pos = 1;
		  this.buyPrice = tradeSignal.midprice;
		  this.tradeCosts += TRADE_COST;
		  this.buySize += SHARES;
		  this.buyTotal += SHARES * tradeSignal.midprice;
		} else if(!this.own && tradeSignal.trade == -10) {
		  // Sell to open
		  this.own = true;
		  this.pos = -1
		  this.sellPrice = tradeSignal.midprice;
		  this.tradeCosts += TRADE_COST;
		  this.sellSize += SHARES;
		  this.sellTotal += SHARES * tradeSignal.midprice;
		} else if(this.own && this.pos == 1 && tradeSignal.trade == -10) {
		  // Sell to close
		  this.own = false;
		  this.pos = 0;
		  this.sellPrice = tradeSignal.midprice;
		  this.tradeCosts += TRADE_COST;
		  this.pnl += (this.sellPrice - this.buyPrice)*SHARES - 2*TRADE_COST;
		  this.sellSize += SHARES;
		  this.sellTotal += SHARES * tradeSignal.midprice;
		  // Sell to open
		  this.own = true;
		  this.pos = -1;
		  this.sellPrice = tradeSignal.midprice;
		  this.tradeCosts += TRADE_COST;
		  this.sellSize += SHARES;		  
		  this.sellTotal += SHARES * tradeSignal.midprice;
		} else if(this.own && this.pos == -1 && tradeSignal.trade == 10) {
		  // Buy to close
		  this.own = false;
		  this.pos = 0;
		  this.buyPrice = tradeSignal.midprice;
		  this.tradeCosts += TRADE_COST;
		  this.pnl += (this.sellPrice - this.buyPrice)*SHARES - 2*TRADE_COST;
		  this.buySize += SHARES;
		  this.buyTotal += SHARES * tradeSignal.midprice;
		  // Buy to open
		  this.own = true;
		  this.pos = 1;
		  this.buyPrice = tradeSignal.midprice;
		  this.tradeCosts += TRADE_COST;
		  this.buySize += SHARES;		  
		  this.buyTotal += SHARES * tradeSignal.midprice;
		}
	}
	this.computeResult = function () {
		var result = {
			"pnl": this.pnl,
			"buySize": this.buySize,
			"sellSize": this.sellSize,
			"buyTotal": this.buyTotal,
			"sellTotal": this.sellTotal,
			"tradeCost": this.tradeCost
			};
		return result;
	}
}

Infine, inviare l'output al dashboard Power BI per la visualizzazione.

SELECT * INTO tradeSignalDashboard FROM tradeSignal /* output tradeSignal to PBI */
SELECT
    symbol,
    time,
    date,
    TRY_CAST(s.pnl as float) AS pnl,
    TRY_CAST(s.buySize as bigint) AS buySize,
    TRY_CAST(s.sellSize as bigint) AS sellSize,
    TRY_CAST(s.buyTotal as float) AS buyTotal,
    TRY_CAST(s.sellTotal as float) AS sellTotal
    INTO pnlDashboard
FROM simulation /* output trade simulation to PBI */

Sommario

Questo articolo illustra come implementare un modello di trading realistico ad alta frequenza con una query moderatamente complessa in Analisi di flusso di Azure. Il modello usa due variabili di input anziché cinque perché Analisi di flusso di Azure non include una funzione di regressione lineare predefinita. Tuttavia, è anche possibile implementare algoritmi più sofisticati di dimensionalità superiore come UDA JavaScript.

È possibile testare ed eseguire il debug della maggior parte della query, diversa dall'UDA JavaScript, usando Analisi di flusso di Azure strumenti per Visual Studio Code per lo sviluppo di query, il test e il debug.

Contenuti correlati

• Panoramica di Analisi di flusso di Azure
• Informazioni di riferimento sul linguaggio di query di Analisi di flusso di Azure
• Funzioni di aggregazione JavaScript definite dall'utente in Analisi di flusso di Azure
• Analisi di flusso di Azure output in Power BI