乾貨丨如何使用時序數據庫快速計算買方或賣方驅動交易

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

本文用到的數據是含有逐筆交易的交易表trade和買賣報價表nbbo。它們分別包含如下字段：

trade

Symbol：股票代碼

Time：時間

Trade_Volume：交易量

Trade_Price：交易價格

nbbo

Symbol：股票代碼

Time：時間

Bid_Price：買方報價

Offer_Price：賣方報價

本文用到的數據都是從紐約證券交易所網站獲取，能夠從NYSE的ftp下載。下載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提供了基於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',i