乾貨丨時序數據庫DolphinDB歷史數據回放教程

一個量化策略在用於實際交易時，處理實時數據的程序一般爲事件驅動。而研發量化策略時，須要使用歷史數據進行回測，這時的程序一般不是事件驅動。所以同一個策略須要編寫兩套代碼，不只耗時並且容易出錯。在 DolphinDB database 中，用戶可將歷史數據按照時間順序以「實時數據」的方式導入流數據表中，這樣就可使用同一套代碼進行回測和實盤交易。

DolphinDB的流數據處理框架採用發佈-訂閱-消費的模式。數據生產者將實時數據繼續地以流的形式發佈給全部數據訂閱者。訂閱者收到消息之後，可以使用自定義函數或者DolphinDB內置的聚合引擎來處理消息。DolphinDB流數據接口支持多種語言的API，包括C++, C#, Java, 和Python等。用戶可使用這些API來編寫更加複雜的處理邏輯，更好地與實際生產環境相結合。有興趣的朋友能夠搜索「DolphinDB流數據教程」瞭解更多內容。

本文介紹replay和replayDS函數，而後使用金融數據展現數據回放的過程與應用場景。

1. 函數介紹

replay

replay(inputTables, outputTables, [dateColumn], [timeColumn], [replayRate], [parallelLevel=1])

replay函數的做用是將若干表或數據源同時回放到相應的輸出表中。用戶須要指定輸入的數據表或數據源、輸出表、日期列、時間列、回放速度以及並行度。

replay函數參數概念以下：

• inputTables: 單個表或包含若干表或數據源（見replayDS介紹）的元組。
• outputTables: 單個表或包含若干個表的元組，這些表一般爲流數據表。輸入表和輸出表的個數一致，且一一對應，每對輸入、輸出表的結構相同。
• dateColumn, timeColumn: string, 表示輸入表的日期和時間列，若不指定則默認第一列爲日期列。若輸入表中時間列同時包含日期和時間，須要將dateColumn和timeColumn設爲同一列。回放時，系統將根據dateColumn和timeColumn的設定，決定回放的最小時間精度。在此時間精度下，同一時刻的數據將在相同批次輸出。好比一張表同時有日期列和時間列，可是replay函數只設置了dateColumn，那麼同一天的全部數據會在一個批次輸出。
• replayRate: 整數, 表示每秒鐘回放的數據條數。因爲回放時同一個時刻數據在同一批次輸出，所以當replayRate小於一個批次的行數時，實際輸出的速率會大於replayRate。
• parallelLevel: 整數, 表示讀取數據的並行度。當源數據大小超過內存大小的時候，須要使用replayDS函數將源數據劃分爲若干個小的數據源，依次從磁盤中讀取數據並回放。指定多個讀取數據的線程數可提高數據讀取速度。

replayDS

replayDS(sqlObj, [dateColumn], [timeColumn], [timeRepartitionSchema])

replayDS函數能夠將輸入的SQL查詢轉化爲數據源，結合replay函數使用。其做用是根據輸入表的分區以及timeRepartitionSchema，將原始的SQL查詢按照時間順序拆分紅若干小的SQL查詢。

replayDS函數參數概念以下：

• sqlObj: SQL元代碼，表示回放的數據，如<select * from sourceTable>。
• dateColumn: 字符串, 表示日期列。若不指定，默認第一列爲日期列。replayDS函數默認日期列是數據源的一個分區列，並根據分區信息將原始SQL查詢拆分爲多個查詢。
• timeColumn: 字符串, 表示時間列，配合timeRepartitionSchema使用。
• timeRepartitionSchema: 時間類型向量，如08:00:00 .. 18:00:00。若同時指定了timeColumn, 則對SQL查詢在時間維度上進一步拆分。

單個內存表回放

單內存表回放只須要設置輸入表、輸出表、日期列、時間列和回放速度便可。

replay(inputTable, outputTable, `date, `time, 10)

使用data source的單表回放

當單錶行數過多時，能夠配合使用replayDS進行回放。首先使用replayDS生成data source，本例中指定了日期列和timeRepartitionColumn。回放調用與單個內存表回放類似，可是能夠指定回放的並行度。replay內部實現使用了pipeline框架，取數據和輸出分開執行。當輸入爲data source時，多塊數據能夠並行讀取，以免輸出線程等待的狀況。此例中並行度設置爲2，表示有兩個線程同時執行取數據的操做。

inputDS = replayDS(<select * from inputTable>, `date, `time, 08:00:00.000 + (1..10) * 3600000)
replay(inputDS, outputTable, `date, `time, 1000, 2)

使用data source的多表回放

replay也支持多張表的同時回放，只須要將多張輸入表以元組的方式傳給replay,而且分別指定輸出表便可。這裏輸出表和輸入表應該一一對應，每一對都必須有相同的表結構。若是指定了日期列或時間列，那麼全部表中都應當有存在相應的列。

ds1 = replayDS(<select * from input1>, `date, `time, 08:00:00.000 + (1..10) * 3600000)
ds2 = replayDS(<select * from input2>, `date, `time, 08:00:00.000 + (1..10) * 3600000)
ds3 = replayDS(<select * from input3>, `date, `time, 08:00:00.000 + (1..10) * 3600000)
replay([ds1, ds2, ds3], [out1, out2, out3], `date, `time, 1000, 2)

取消回放

若是replay函數是經過submitJob調用，可使用getRecentJob獲取jobId，而後用cancelJob取消回放。

getRecentJobs()
cancelJob(jobid)

若是是直接調用，可在另一個GUI session中使用getConsoleJobs獲取jobId，而後使用cancelConsoleJob取消回聽任務。

getConsoleJobs()
cancelConsoleJob(jobId)

2．如何使用回放數據

回放的數據以流數據形式存在，咱們可使用如下三種方式來訂閱消費這些數據：

• 在DolphinDB中訂閱，使用DolphinDB腳本自定義回調函數來消費流數據。
• 在DolphinDB中訂閱，使用內置的流計算引擎來處理流數據，譬如時間序列聚合引擎、橫截面聚合引擎、異常檢測引擎等。DolphinDB內置的聚合引擎能夠對流數據進行實時聚合計算，使用簡便且性能優異。在3.2中，咱們使用橫截面聚合引擎來處理回放的數據，並計算ETF的內在價值。橫截面聚合引擎的具體用法參見DolphinDB的用戶手冊。
• 第三方客戶端經過DolphinDB的流數據API來訂閱和消費數據。

３. 金融示例

回放美國股市一天的level1交易數據，並計算ETF價值

本例中使用美國股市2007年8月17日的level1交易數據，利用replayDS進行數據回放，並經過DolphinDB內置的橫截面聚合引擎計算ETF價值。數據存放在分佈式數據庫dfs://TAQ的quotes表，下面是quotes表的結構以及數據預覽。

//加載數據庫中quotes表的數據，查看錶結構
quotes = database("dfs://TAQ").loadTable("quotes");
quotes.schema().colDefs;

name    typeString    typeInt
time    SECOND        10
symbol  SYMBOL        17
ofrsiz  INT           4
ofr     DOUBLE        16
mode    INT           4
mmid    SYMBOL        17
ex      CHAR          2
date    DATE          6
bidsize INT           4
bid     DOUBLE        16

//查看quotes表內前十行的數據
select top 10 * from quotes;

symbol    date         time       bid     ofr     bidsiz    ofrsiz    mode    ex    mmid
A         2007.08.17   04:15:06   0.01    0       10        0         12      80
A         2007.08.17   06:21:16   1       0       1         0         12      80
A         2007.08.17   06:21:44   0.01    0       10        0         12      80
A         2007.08.17   06:49:02   32.03   0       1         0         12      80
A         2007.08.17   06:49:02   32.03   32.78   1         1         12      80
A         2007.08.17   07:02:01   18.5    0       1         0         12      84
A         2007.08.17   07:02:01   18.5    45.25   1         1         12      84
A         2007.08.17   07:54:55   31.9    45.25   3         1         12      84
A         2007.08.17   08:00:00   31.9    40      3         2         12      84
A         2007.08.17   08:00:00   31.9    35.5    3         2         12      84

（1）對要進行回放的數據進行劃分。因爲一天的數據共有336,305,414條，一次性導入內存再回放會有較長延遲，還有可能致使內存溢出，所以先使用replayDS函數並指定參數timeRepartitionSchema，將數據按照時間戳分爲62個部分。

sch = select name,typeString as type from quotes.schema().colDefs
trs = cutPoints(09:30:00.001..18:00:00.001, 60)
rds = replayDS(<select * from quotes>, `date, `time,  trs);

（2）定義輸出表outQuotes，通常爲流數據表。

share streamTable(100:0, sch.name,sch.type) as outQuotes

（3）定義股票權重字典weights以及聚合函數etfVal，用於計算ETF價值。在本例中，咱們僅計算AAPL、IBM、MSFT、NTES、AMZN、GOOG這幾隻股票的ETF價值。

defg etfVal(weights,sym, price) {
    return wsum(price, weights[sym])
}
weights = dict(STRING, DOUBLE)
weights[`AAPL] = 0.1
weights[`IBM] = 0.1
weights[`MSFT] = 0.1
weights[`NTES] = 0.1
weights[`AMZN] = 0.1
weights[`GOOG] = 0.5

（4）建立流聚合引擎，並訂閱數據回放的輸出表outQuotes。訂閱outQuotes表時，咱們指定了發佈表的過濾條件，只有symbol爲AAPL、IBM、MSFT、NTES、AMZN、GOOG的數據纔會發佈到橫截面聚合引擎，減小沒必要要的網絡開銷和數據傳輸。

setStreamTableFilterColumn(outQuotes, `symbol)
outputTable = table(1:0, `time`etf, [TIMESTAMP,DOUBLE])
tradesCrossAggregator=createCrossSectionalAggregator("etfvalue", <[etfVal{weights}(symbol, ofr)]>, quotes, outputTable, `symbol, `perBatch)
subscribeTable(,"outQuotes","tradesCrossAggregator",-1,append!{tradesCrossAggregator},true,,,,,`AAPL`IBM`MSFT`NTES`AMZN`GOOG)

（5）開始回放，設定每秒回放10萬條數據，聚合引擎則會實時地對回放的數據進行消費。

submitJob("replay_quotes", "replay_quotes_stream",  replay,  [rds],  [`outQuotes], `date, `time,100000,4)

（6）查看不一樣時間點下咱們選擇的股票的ETF價值。

//查看outputTable表內前15行的數據,其中第一列時間爲聚合計算髮生的時間
>select top 15 * from outputTable;

time                  　　etf
2019.06.04T16:40:18.476   14.749
2019.06.04T16:40:19.476	  14.749
2019.06.04T16:40:20.477	  14.749
2019.06.04T16:40:21.477	  22.059
2019.06.04T16:40:22.477	  22.059
2019.06.04T16:40:23.477	  34.049
2019.06.04T16:40:24.477	  34.049
2019.06.04T16:40:25.477	  284.214
2019.06.04T16:40:26.477	  284.214
2019.06.04T16:40:27.477	  285.68
2019.06.04T16:40:28.477	  285.68
2019.06.04T16:40:29.478	  285.51
2019.06.04T16:40:30.478	  285.51
2019.06.04T16:40:31.478	  285.51
2019.06.04T16:40:32.478	  285.51

4. 性能測試

咱們在服務器上對DolphinDB的數據回放功能進行了性能測試。服務器配置以下：

主機：DELL PowerEdge R730xd

CPU：Intel Xeon(R) CPU E5-2650 v4（24核 48線程 2.20GHz）

內存：512 GB （32GB × 16, 2666 MHz）

硬盤：17T HDD （1.7T × 10, 讀取速度222 MB/s，寫入速度210 MB/s）

網絡：萬兆以太網

測試腳本以下：

sch = select name,typeString as type from  quotes.schema().colDefs
trs = cutPoints(09:30:00.001..18:00:00.001,60)
rds = replayDS(<select * from quotes>, `date, `time,  trs);
share streamTable(100:0, sch.name,sch.type) as outQuotes
jobid = submitJob("replay_quotes","replay_quotes_stream",  replay,  [rds],  [`outQuotes], `date, `time, , 4)

在不設定回放速率（即以最快的速率回放），而且輸出表沒有任何訂閱時，回放336,305,414條數據耗時僅須要90~110秒。