Simulação de negociação de alta frequência com Stream Analytics

O Azure Stream Analytics suporta análises avançadas através da combinação de linguagem SQL, funções definidas pelo utilizador (UDFs) e agregados definidos pelo utilizador (UDAs). Análises avançadas incluem formação e pontuação online em aprendizagem automática, e simulação de processos com estado. Este artigo descreve como realizar regressão linear num trabalho de Azure Stream Analytics que realiza formação contínua e pontuação num cenário de trading de alta frequência.

Pré-requisitos

• Uma assinatura do Azure. Se não tiver uma, crie uma conta gratuita.
• Uma tarefa do Azure Stream Analytics.
• Um espaço de nomes do Hubs de Eventos do Azure e um hub de eventos.
• Familiaridade com a Linguagem de Consultas de Análise de Fluxos.
• (Opcional) Uma conta Power BI se quiseres visualizar a saída.

Fluxo de trabalho de negociação de alta frequência

O fluxo lógico da negociação de alta frequência é:

1. Obter cotações em tempo real de uma bolsa de valores mobiliários.
2. Construir um modelo preditivo em torno das cotações para antecipar o movimento do preço.
3. Colocar ordens de compra ou venda para ganhar dinheiro com a previsão bem-sucedida dos movimentos de preço.

Este cenário exige:

• Um fluxo de cotações em tempo real.
• Um modelo preditivo que pode operar com as cotações em tempo real.
• Uma simulação de negociação que demonstra o lucro ou perda do algoritmo de negociação.

Feed de cotações em tempo real

A Investors Exchange (IEX) anteriormente oferecia cotações gratuitas em tempo real utilizando socket.io. Pode escrever um programa simples de consola para receber orçamentos em tempo real e enviá-los para o Hubs de Eventos do Azure como fonte de dados. O código seguinte é um esqueleto do programa. O código omite o tratamento de erros para simplificar. Também precisa de incluir os pacotes NuGet SocketIoClientDotNet e WindowsAzure.ServiceBus no seu projeto.

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);
});

Aqui estão alguns eventos de exemplo gerados:

{"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"}

Modelo preditivo para negociação de alta frequência

Para esta demonstração, o exemplo utiliza um modelo linear descrito em Order Imbalance Based Strategy in High Frequency Algorithmic Trading.

O desequilíbrio das ordens de volume (VOI) é uma função do preço e volume atuais de compra/venda, e do preço e volume de compra/venda a partir do último tick. O artigo identifica a correlação entre o VOI e os movimentos futuros dos preços. Constrói um modelo linear entre os últimos cinco valores de VOI e a variação de preço nos 10 ticks seguintes. O modelo treina com os dados do dia anterior com regressão linear.

O modelo treinado faz então previsões de variação de preço nas cotações do dia de negociação atual em tempo real. Quando o modelo prevê uma variação de preço suficientemente grande, executa uma operação. Dependendo da definição do limiar, uma única ação pode gerar milhares de operações durante um dia de negociação.

As secções seguintes mostram como expressar as operações de treino e previsão num trabalho de Azure Stream Analytics. A consulta completa é uma única instrução WITH composta por expressões de tabela comuns (CTEs) que formam uma cadeia de processamento:

Limpar e converter campos de introdução de citações

O primeiro CTE da consulta do Azure Stream Analytics converte os dados brutos de cotação do Event Hubs em colunas SQL corretamente tipadas. O DATEADD converte o tempo de época (milissegundos Unix) em data-hora. TRY_CAST coage os tipos de dados sem falhar a consulta. Converta os campos de entrada para os tipos de dados esperados para evitar comportamento inesperado na manipulação ou comparação dos campos.

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
),

Recuperar valores de tick anteriores com LAG

O próximo CTE na consulta do Azure Stream Analytics usa a função LAG para obter o preço de compra/venda e o volume do tick anterior para cada símbolo da ação. É arbitrariamente escolhido o valor de uma hora para DURAÇÃO LIMITE. Tendo em conta a frequência das cotações, pode encontrar a variação anterior recorrendo à hora anterior.

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
),

Calcular o desequilíbrio de ordem de volume (VOI)

O próximo CTE calcula o valor do VOI a partir dos dados de bid/ask do tick atual e anterior. A consulta exclui os valores nulos nos casos em que não existe um tick anterior.

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
),

Construir sequências de características para treino de modelos

O próximo CTE usa novamente o LAG para criar uma sequência com 2 valores consecutivos de VOI, seguidos de 10 valores consecutivos de preço médio. Estas sequências formam os dados de treino para o modelo de regressão linear.

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
),

Remodelar dados em vetores de características

A próxima CTE remodela as sequências de preços e VOI em vetores de características para um modelo linear de duas variáveis, onde os valores de VOI são as variáveis independentes (x1, x2) e a variação média futura do preço é a variável dependente (y). Eventos com dados incompletos são filtrados.

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
),

Treine o modelo de regressão linear com SUM e AVG

Como Azure Stream Analytics não tem uma função de regressão linear incorporada, a consulta utiliza agregados SUM e AVG para calcular os coeficientes (a, b1, b2) para o modelo de regressão linear de duas variáveis. O modelo é treinado diariamente usando uma janela de tumbling de 24 horas.

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
),

Avaliar as cotações atuais usando o modelo do dia anterior

Para usar o modelo de regressão linear treinado do dia anterior para pontuar o evento atual, a consulta junta as aspas aos coeficientes do modelo. Em vez de usar JOIN, a consulta utiliza UNION para combinar eventos de modelo e citar eventos num único fluxo. Depois, usa LAG para emparelhar os eventos com o modelo do dia anterior, de modo que obténs exatamente uma correspondência. Devido ao fim de semana, a consulta retrocede três dias (72 horas). Se fosse usado um JOIN simples, obter-se-iam três modelos para cada evento de cotação.

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'
),

Gerar sinais comerciais a partir de previsões

Os CTEs finais calculam a alteração futura esperada do preço (EFPC) aplicando a fórmula de regressão linear (a + b1 * x1 + b2 * x2) e depois geram sinais de compra/venda com base num limiar de ±0,02. Um valor de negociação de 10 corresponde a compra. Um valor de transação de -10 corresponde a uma venda.

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
),

Teste a estratégia de negociação com uma simulação

Depois de gerar os sinais de negociação, teste a eficácia da estratégia sem negociar realmente.

Este teste usa um UDA com uma janela de salto que salta a cada minuto. O agrupamento na data e a cláusula HAVING garantem que a janela só contabilize eventos que pertencem ao mesmo dia. Para uma janela com salto que abrange dois dias, a data no GROUP BY separa o agrupamento no dia anterior e no dia atual. A cláusula HAVING filtra as janelas que terminam no dia atual mas se agrupam no dia anterior.

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
)

O JavaScript UDA inicializa todos os acumuladores na init função, calcula a transição de estado com cada evento adicionado à janela e devolve os resultados da simulação no final da janela. A simulação detém ou vende a descoberto 10 ações em cada operação. O custo de transação é fixo $8. A tabela seguinte mostra as quatro ações de negociação que a UDA realiza:

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;
	}
}

Por fim, exporte os resultados para o painel do Power BI para visualização.

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 */

Resumo

Este artigo mostra como implementar um modelo realista de trading de alta frequência com uma consulta moderadamente complexa no Azure Stream Analytics. O modelo usa duas variáveis de entrada em vez de cinco porque o Azure Stream Analytics não inclui uma função de regressão linear incorporada. No entanto, também pode implementar algoritmos mais sofisticados com dimensões mais elevadas como UDAs em JavaScript.

Pode testar e depurar a maioria das consultas, exceto o JavaScript UDA, usando ferramentas Azure Stream Analytics para Visual Studio Code para desenvolvimento, testes e depuração de consultas.

Conteúdo relacionado

• Azure Stream Analytics visão geral
• Referência da Linguagem de Consulta do Azure Stream Analytics
• Agregados JavaScript definidos pelo utilizador no Azure Stream Analytics
• saída do Azure Stream Analytics para o Power BI