Stream Analytics を使用した高周波取引シミュレーション

Azure Stream Analyticsでは、SQL 言語、JavaScript ユーザー定義関数 (UDF)、ユーザー定義集計 (UDA) の組み合わせによる高度な分析がサポートされています。 高度な分析には、オンライン機械学習のトレーニングとスコアリング、ステートフル プロセス シミュレーションが含まれます。 この記事では、高周波取引シナリオで継続的なトレーニングとスコアリングを行うAzure Stream Analytics ジョブで線形回帰を実行する方法について説明します。

Prerequisites

• Azure サブスクリプション。 お持ちでない場合は、 無料アカウントを作成してください。
• Azure Stream Analytics ジョブ。
• Azure Event Hubs名前空間とイベント ハブ。
• Stream Analytics クエリ言語に関する知識。
• (省略可能)出力を視覚化する場合は、Power BI アカウント。

高頻度取引ワークフロー

高周波取引の論理フローは次のとおりです。

1. 証券取引所からリアルタイムの相場情報を取得する。
2. 価格の動きを予測するために、見積もりの周りに予測モデルを構築します。
3. 価格の動きの成功予測から利益を得るために買い注文または売り注文を行います。

このシナリオでは、次のものが必要です。

• リアルタイムの見積もりフィード。
• リアルタイムの見積もりを操作できる予測モデル。
• 取引アルゴリズムの利益または損失を示す取引シミュレーション。

リアルタイム見積もりフィード

投資家Exchange(IEX)は、以前は無料のリアルタイムの入札を提供し、socket.io を使用して見積もりを求めました。 リアルタイム引用符を受け取り、データ ソースとしてAzure Event Hubsにプッシュする簡単なコンソール プログラムを記述できます。 次のコードは、プログラムのスケルトンです。 このコードでは、簡潔にするためにエラー処理が省略されています。 また、プロジェクトに SocketIoClientDotNet パッケージと WindowsAzure.ServiceBus NuGet パッケージを含める必要があります。

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

生成されるサンプル イベントを次に示します。

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

高周波取引の予測モデル

このデモでは、サンプルは高頻度アルゴリズム取引における注文不均衡に基づく戦略で説明されている線形モデルを使用します。

Volume Order Imbalance (VOI) は、現在の買い気配値/売り気配値/売買高と前回のティックの買い気配値/売り気配値/売買高との関数です。 この論文では、VOIと将来の価格変動の相関関係を特定しています。 直近 5 つの VOI 値と次の 10 ティックにおける価格変化の関係を表す線形モデルを構築します。 モデルは、前日のデータを線形回帰でトレーニングします。

その後、トレーニング済みのモデルは、現在の取引日の見積もりの価格変更予測をリアルタイムで行います。 モデルは、十分な大きな価格変化を予測すると、取引を実行します。 しきい値の設定によっては、1 つの株式が取引日に何千もの取引を生成する場合があります。

次のセクションでは、Azure Stream Analytics ジョブでトレーニング操作と予測操作を表現する方法を示します。 完全なクエリは、パイプラインを形成する共通テーブル式 (CTE) で構成される単一の WITH ステートメントです。

引用符入力フィールドを整理して変換

Azure Stream Analytics クエリの最初の CTE は、Event Hubs の生の見積もりデータを適切に型指定された SQL 列に変換します。 DATEADD はエポック時間 (Unix ミリ秒) を datetime に変換します。 TRY_CAST クエリに失敗することなくデータ型を強制します。 フィールドの操作や比較で予期しない動作が発生しないように、入力フィールドを予期されるデータ型にキャストします。

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

LAG を使用して以前のティック値を取得する

Azure Stream Analytics クエリの次の CTE では、LAG 関数を使用して、各銘柄について直前のティックからの買値/売値とサイズを取得します。 LIMIT DURATION 値の 1 時間が任意に選択されます。 クォート頻度が高いことを踏まえると、1 時間さかのぼれば直前のティックを確認できます。

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

出来高注文不均衡（VOI）を計算する

次の CTE は、現在のティックと直前のティックの買値/売値データから VOI 値を産出します。 このクエリは、以前のティックが存在しない場合、NULL 値を除外します。

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

モデル トレーニング用の機能シーケンスを構築する

次の CTE では、 LAG を再び使用して、2 つの連続する VOI 値を含むシーケンスを作成し、その後に 10 個の連続する中間価格値を作成します。 これらのシーケンスは、線形回帰モデルのトレーニング データを形成します。

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

データを特徴ベクトルに整形する

次の CTE では、価格と VOI シーケンスが 2 変数線形モデルの特徴ベクトルに整形されます。VOI 値は独立変数 (x1、x2) で、平均将来の価格変化は従属変数 (y) です。 不完全なデータを含むイベントは除外されます。

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

SUM と AVG を使用して線形回帰モデルをトレーニングする

Azure Stream Analyticsには線形回帰関数が組み込まれていないため、クエリでは SUM および AVG 集計を使用して、2 変数線形回帰モデルの係数 (a、b1、b2) を計算します。 このモデルは、24 時間の移動ウィンドウを使用して毎日再学習を行います。

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

前日のモデルで現在の引用符をスコア付けする

クエリでは、前日のトレーニング済みの線形回帰モデルを使用して現在のイベントをスコア付けするために、モデル係数と引用符を結合します。 クエリでは、 JOIN を使用する代わりに UNION を使用して、モデル イベントとクォート イベントを 1 つのストリームに結合します。 その後、 LAG を使用して前日のモデルとイベントをペアリングするため、正確に 1 つの一致が得られます。 週末のため、クエリは 3 日間 (72 時間) を振り返ります。 単純な JOIN を使用すると、すべての見積もりイベントに対して 3 つのモデルが取得されます。

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

予測からトレードシグナルを生成する

最終的な CTE では、線形回帰式 (a + b1 * x1 + b2 * x2) を適用して、予想される将来の価格変化 (efpc) を計算し、±0.02 しきい値に基づいて売買シグナルを生成します。 10のトレード値は買いです。 取引値が -10 の場合は売りです。

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

シミュレーションを使用して取引戦略をテストする

取引シグナルを生成した後、トレーディング戦略が実際の取引なしでどれだけ効果的であるかをテストします。

このテストでは、1 分ごとにホップするホッピング ウィンドウを持つ UDA を使用します。 日付と HAVING 句のグループ化により、ウィンドウは同じ日に属するイベントのみを考慮します。 2 日間にわたるホッピング ウィンドウの場合、 GROUP BY 日付はグループ化を前日と現在の日に分けます。 HAVING 句は、当日で終了するが前日をグループ化の基準とする期間を抽出します。

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
)

JavaScript UDA は、 init 関数内のすべてのアキュムレータを初期化し、ウィンドウに追加されたすべてのイベントで状態遷移を計算し、ウィンドウの最後にシミュレーション結果を返します。 シミュレーションでは、1 回の取引につき 10 株の株式を保持またはショートします。 トランザクション コストはフラットな $8です。 次の表は、UDA が実行する 4 つの取引アクションを示しています。

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

最後に、視覚化のためにPower BI ダッシュボードに出力します。

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

まとめ

この記事では、Azure Stream Analyticsで中程度に複雑なクエリを使用して、現実的な高周波取引モデルを実装する方法について説明します。 Azure Stream Analyticsには組み込みの線形回帰関数が含まれていないため、モデルでは 5 つではなく 2 つの入力変数が使用されます。 ただし、より高度なアルゴリズムを JavaScript UDA としてより高い次元で実装することもできます。

Visual Studio Code 用の Azure Stream Analytics ツールを使用して、クエリの開発、テスト、デバッグを行うことで、JavaScript UDA 以外のほとんどのクエリをテストおよびデバッグできます。

関連するコンテンツ

• Azure Stream Analyticsの概要
• Azure Stream Analytics クエリ言語リファレンス
• JavaScript ユーザー定義集計 (Azure Stream Analytics)
• Azure Stream Analytics から Power BI への出力