一个量化策略在用于理论交易时，解决实时数据的程序通常为事件驱动。而研发量化策略时，须要应用历史数据进行回测，这时的程序通常不是事件驱动。因而同一个策略须要编写两套代码，不仅耗时而且容易出错。在 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 typeInttime SECOND 10symbol SYMBOL 17ofrsiz INT 4ofr DOUBLE 16mode INT 4mmid SYMBOL 17ex CHAR 2date DATE 6bidsize INT 4bid DOUBLE 16//查看quotes表内前十行的数据select top 10 * from quotes;symbol date time bid ofr bidsiz ofrsiz mode ex mmidA 2007.08.17 04:15:06 0.01 0 10 0 12 80A 2007.08.17 06:21:16 1 0 1 0 12 80A 2007.08.17 06:21:44 0.01 0 10 0 12 80A 2007.08.17 06:49:02 32.03 0 1 0 12 80A 2007.08.17 06:49:02 32.03 32.78 1 1 12 80A 2007.08.17 07:02:01 18.5 0 1 0 12 84A 2007.08.17 07:02:01 18.5 45.25 1 1 12 84A 2007.08.17 07:54:55 31.9 45.25 3 1 12 84A 2007.08.17 08:00:00 31.9 40 3 2 12 84A 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().colDefstrs = 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.1weights[`IBM] = 0.1weights[`MSFT] = 0.1weights[`NTES] = 0.1weights[`AMZN] = 0.1weights[`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 etf2019.06.04T16:40:18.476 14.7492019.06.04T16:40:19.476 14.7492019.06.04T16:40:20.477 14.7492019.06.04T16:40:21.477 22.0592019.06.04T16:40:22.477 22.0592019.06.04T16:40:23.477 34.0492019.06.04T16:40:24.477 34.0492019.06.04T16:40:25.477 284.2142019.06.04T16:40:26.477 284.2142019.06.04T16:40:27.477 285.682019.06.04T16:40:28.477 285.682019.06.04T16:40:29.478 285.512019.06.04T16:40:30.478 285.512019.06.04T16:40:31.478 285.512019.06.04T16:40:32.478 285.514. 性能测试
咱们在服务器上对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().colDefstrs = 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 outQuotesjobid = submitJob("replay_quotes","replay_quotes_stream", replay, [rds], [`outQuotes], `date, `time, , 4)在不设定回放速率（即以最快的速率回放），并且输出表没有任何订阅时，回放336,305,414条数据耗时仅须要90~110秒。