給定高頻交易資料以及報價資料，如何判斷每筆交易是由買方驅動或是賣方驅動，是進行高頻交易資料分析經常需要處理的問題。本文將介紹如何使用DolphinDB快速計算每筆交易的驅動方，只需不到2秒鐘即可對美國一天的level 1的高頻交易資料進行計算並存入資料庫。本文使用了非同時連線（asof join）以及map-reduce。

本文用到的資料是含有逐筆交易的交易表trade和買賣報價表nbbo。它們分別包含以下欄位：

trade

Symbol：股票程式碼

Time：時間

Trade_Volume：交易量

Trade_Price：交易價格

nbbo

Symbol：股票程式碼

Time：時間

Bid_Price：買方報價

Offer_Price：賣方報價

本文用到的資料都是從紐約證券交易所網站獲取，可以從下載。下載EQY_US_ALL_TRADE_20161024.gz和EQY_US_ALL_NBBO_20161024.gz兩個檔案，然後把它們解壓，儲存在/home/DolphinDB/Data目錄下，把兩個檔案的最後一行刪除，因為最後一行是用來標記檔案結尾的。

sed -i '$ d' EQY_US_ALL_TRADE_20161024
sed -i '$ d' EQY_US_ALL_NBBO_20161024

在DolphinDB中執行以下指令碼，把資料匯入到 DolphinDB database 中。本教程使用的是分散式資料庫，如果想使用記憶體資料庫，只需把dbPath修改為""，若要使用本地磁碟資料庫，只需把dbPath修改為磁碟目錄，比如“/home/DolphinDB/Data/EQY”。

DATA_DIR = "/home/DolphinDB/Data"
login("admin","123456")
dbPath= "dfs://EQY"
db = database(dbPath, SEQ, 16)

trade = loadTextEx(db, `trade, DATA_DIR + "/EQY_US_ALL_TRADE_20161024",'|')
nbbo = loadTextEx(db, `nbbo, DATA_DIR + "/EQY_US_ALL_NBBO_20161024",'|')

把分散式表載入到記憶體中：

db=database(dbPath);
trade = db.loadTable("trade")
nbbo = db.loadTable("nbbo")
 

通過map-reduce分散式計算框架，把結果儲存至分散式表中。分散式表的資料在物理上分佈在不同的節點，通過DolphinDB的分散式引擎，可以做統一查詢。

建立分散式表trade_side，用於儲存計算結果。用於儲存結果的表除了包含trade表中的欄位，還包含Bid_Price、Offer_Price和Side欄位。

model=select top 1 * from trade
model[`Bid_Price]=0.0
model[`Offer_Price]=0.0
model[`Side]='B'
if(existsTable(dbPath, "trade_side"))
	db.dropTable("trade_side")
db.createPartitionedTable(model, "trade_side", "Symbol")

判斷每筆交易由買方或賣方驅動，我們定義的演算法如下：如果交易價格小於買賣報價的平均價格，交易為賣方驅動，把Side設定為'S'；如果交易價格大於買賣報價的平均價格，交易為買方驅動，把Side設定為'B'。如果買方報價等於買賣報價的平均價格，則把Side設定為NULL。

def saveTradeSide(t){
	update t set Side = iif(Trade_Price<(Bid_Price + Offer_Price)*0.5, 'S',iif(Trade_Price>(Bid_Price + Offer_Price)*0.5, 'B',char()))
	update t set Side = NULL where Bid_Price >= Offer_Price or Bid_Price <= 0
	loadTable("dfs://EQY", "trade_side").append!(t)
	return t.size()
}

iif(condition, x, y)：iif是條件運算子。condition是條件向量，如果condition[i]為true，則返回x[i]，否則返回y[i]。

以下程式碼連線交易表trades和買賣報價表nbbo，sqlDS函式會根據輸入的SQL元程式碼建立資料來源。通過map-reduce函式mr把saveTradeSide應用到各個資料來源。

ds = sqlDS(<select trade.*, Bid_Price, Offer_Price from aj(trade,nbbo,`Symbol`Time) where Time between 09:30:00.000000000 : 15:59:59.999999999>)
mr(ds,saveTradeSide,+)

aj（asof join）是DolphinDB專門為時序資料設計的連線方式。由於成交和買賣報價的發生時間不可能完全一致，因此不能使用等值連線（equal join）。在上面的程式碼中，如果對同一支股票，表nbbo中沒有與表trade中Time匹配的行，asof join會在右表中取同一支股票該時刻之前最近的時間以匹配。

DolphinDB Database 提供了基於map-reduce和迭代的分散式演算法。使用者只需要指定分散式資料來源和核心函式，如map函式、reduce函式、final函式等，非常方便。DolphinDB的分散式應用無需編譯、打包或者部署，可以線上使用，大大提高了資料分析師的工作效率。trade表有8023只股票共2700萬條交易記錄，nbbo表有7800萬條記錄。如此龐大的資料量，使用分散式計算，僅需1秒多，效能極佳。

檢視IBM的前100條結果：

select top 100 Time, Exchange, Symbol, Trade_Volume, Trade_Price, Bid_Price, Offer_Price, Side from db.loadTable("trade_side") where Symbol=`IBM
Time	 Exchange	Symbol	Trade_Volume	Trade_Price	Bid_Price	Offer_Price	Side
09:30:00.105112000	80	IBM	900	150.4	150.12	150.97	'S'
09:30:00.105201000	80	IBM	900	150.4	150.12	150.97	'S'
09:30:00.105293000	80	IBM	400	150.4	150.12	150.97	'S'
09:30:00.105398000	80	IBM	119	150.4	150.12	150.97	'S'
09:30:00.105498000	80	IBM	81	150.4	150.12	150.97	'S'
09:30:00.432775000	80	IBM	100	150.49	150.49	150.97	'S'
09:30:00.452763000	90	IBM	200	150.49	150.49	150.97	'S'
09:30:00.480602000	84	IBM	100	150.49	150.49	150.73	'S'
09:30:00.480698000	84	IBM	100	150.49	150.49	150.73	'S'
09:30:00.563528000	78	IBM	55,940	150.58	150.49	150.73	'S'
09:30:00.577708000	90	IBM	100	150.59	150.49	150.95	'S'
09:30:00.578129000	78	IBM	40	150.65	150.49	150.95	'S'
09:30:00.578235000	78	IBM	60	150.69	150.2	150.9	'B'
09:30:00.584212000	80	IBM	89	150.5	150.2	150.9	'S'
09:30:00.600259000	80	IBM	1	150.5	150.2	150.9	'S'
...

如果資料量不大，可以通過SQL語句進行計算，直接線上使用，非常方便。

select trade.*, Bid_Price, Offer_Price,iif(Trade_Price<(Bid_Price + Offer_Price)*0.5, 'S',iif(Trade_Price>(Bid_Price + Offer_Price)*0.5, 'B',char())) as Side from aj(trade,nbbo,`Symbol`Time) where Time between 09:30:00.000000000 : 15:59:59.999999999 

歡迎訪問下載DolphinDB試用版

聯絡郵箱：[email protected]