实时计算高频因子

与其它流计算处理框架类似，DolphinDB 的流计算系统也包括消息发布、消息代理和消息订阅三个部分。消息发布端与消息订阅端均可为本地的数据节点、另一个数据节点，或第三方的 API（例如 Python, C++, Java, C# 等 API）。DolphinDB 数据节点上的流数据表（stream table）充当消息代理的角色，发布端向流数据表插入记录即实现了消息发布的功能。

高频因子的计算，以交易和报价数据作为主要的流数据输入，亦可配合使用其它数据。高频因子的计算结果通常输出到一个流数据表或内存表。

DolphinDB 流计算中使用的消息可以采用两种格式：表（table）和元组（tuple），由subscribeTable函数的 msgAsTable 参数指定。

本教程中所用的 csv 数据文件，可由此处下载，并存于文件夹 YOURDIR 中。

所谓无状态因子计算，即因子的计算不需要回溯历史数据，仅根据最新的一条消息即可生成因子。无状态因子计算场景下，建议使用表作为消息的格式，并使用 SQL 语句计算因子。

假设消息数据表包含以下列：symbol, date, time, askPrice1..askPrice5, bidPrice1..bidPrice5, askVolume1..askVolume5, bidVolume1..bidVolume5。其中 askPrice1..askPrice5 以及 askVolume1..askVolume5 为前5档卖方出价以及对应的量；bidPrice1..bidPrice5 以及 bidVolume1..bidVolume5 为前5档买方出价以及对应的量。计算以下两个状态因子：

• 因子1：根据每一条最新报价，计算因子 (askVolume1-bidVolume1)/(askVolume1+bidVolume1)。
• 因子2：对买档与卖档，分别计算最新5挡成交量加权累计之和，然后计算两者之比例。

上述无状态因子，可分别使用以下脚本计算：

• 因子1：

      select symbol,time(now()) as time, (askVolume1-bidVolume1)/(askVolume1+bidVolume1) as factorValue from msg

• 因子2：

      w = exp(-1 * 0..4)
      select symbol, time(now()) as time, 0.5*log(rowSum([bidVolume1,bidVolume2,bidVolume3,bidVolume4,bidVolume5]*w)/rowSum([askVolume1,askVolume2,askVolume3,askVolume4,askVolume5]*w)) as factorValue from msg

要实时计算上述无状态因子，把上述脚本封装成函数并在subscribeTable函数中指定其为 handler 参数（消息处理函数）即可。消息处理函数必须是单目函数，且唯一的参数就是本批次订阅到的消息。以因子2为例:

quotesData = loadText(yourDIR + "sampleQuotes.csv")

def factorHandler(mutable factor1, msg){
    w = exp(-1 * 0..4)
    x=select symbol, datetime(now()) as datetime, 0.5*log(rowSum([bidVolume1,bidVolume2,bidVolume3,bidVolume4,bidVolume5]*w)/rowSum([askVolume1,askVolume2,askVolume3,askVolume4,askVolume5]*w)) as factorValue from msg
    factor1.tableInsert(x)
}
share table(1000000:0, `symbol`datetime`factorValue, [SYMBOL,DATETIME,DOUBLE]) as factor1
x=quotesData.schema().colDefs
share(streamTable(1000000:0, x.name, x.typeString), "quotes")

subscribeTable(tableName="quotes", actionName="hfFactor", handler=factorHandler{factor1}, msgAsTable=true)

然后使用replay函数将样本数据 quotesData 写入流数据表 quotes 触发计算。本教程的所有例子均使用replay函数将历史数据进行回放以模拟实时数据。

replay(inputTables=quotesData, outputTables=quotes, dateColumn=`date, timeColumn=`time)

最后在数据表 factor1 中查看计算结果。本例计算结果中的时间为计算发生时间，也可以根据业务要求调整为数据时间，即直接取消息中的 date 与 time 列。

流数据计算中，有一类典型的计算是根据滑动的时间窗口计算聚合值，连续的两个窗口可以重叠或不重叠。这类计算统称为滑动窗口计算。DolphinDB 内置的时间序列聚合引擎用于解决滑动窗口计算。DolphinDB 针对某些聚合函数在流数据时序引擎中的使用进行了优化，在计算每个窗口时充分利用上一个窗口的计算结果，尽可能避免重复计算，显著提高运行速度。已优化的聚合函数包括：sum, sum2, min, max, first, last, avg, count, std, var, med, percentile, covar, corr, beta, wavg, wsum, prod, rank, stdp, varp, kurtosis, skew, ifirstNot, ilastNot, imax, imin, mad, maxPositiveStreak, se, slr。聚合引擎可以接受的指标除了单个聚合函数使用消息中的一个或两个列作为参数外，还可以进行两个方向的扩展：（1）多个聚合函数构成新的表达式；（2）聚合函数的参数可以是消息中多个字段的表达式。

下例使用时间序列聚合引擎计算分钟级K线。Trade 表是一个流数据表，这里为了方便，假定 Trade 表与订阅端在同一个节点上。实际使用中，两者也可处于不同的节点。

tradesData = loadText(yourDIR + "sampleTrades1.csv")

//定义流数据表Trade
x=tradesData.schema().colDefs
share streamTable(100:0, x.name, x.typeString) as Trade

//定义OHLC输出表
share streamTable(100:0, `datetime`symbol`open`high`low`close`volume`updatetime,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE,LONG,TIMESTAMP]) as OHLC

//定义实时聚合引擎：每分钟计算过去5分钟K线
tsAggrOHLC = createTimeSeriesAggregator(name="aggr_ohlc", windowSize=300000, step=60000, metrics=<[first(Price),max(Price),min(Price),last(Price),sum(Volume),now()]>, dummyTable=Trade, outputTable=OHLC, timeColumn=`Datetime, keyColumn=`Symbol)

//订阅流数据写入聚合引擎
subscribeTable(tableName="Trade", actionName="minuteOHLC1", offset=0, handler=append!{tsAggrOHLC}, msgAsTable=true)

replay(inputTables=tradesData, outputTables=Trade, dateColumn=`Datetime)

查看结果：

select top 10 * from OHLC

上面的例子中，流数据表 Trade 的消息直接作为聚合引擎的输入。某些场景下，需要对流数据表中的数据进行预处理后再输入到聚合引擎。若 Trade 表中的 Volume 是从开盘到当前的累计交易量，下例中定义函数calcVolume作为数据订阅的消息处理函数（即 handler 参数），将累计交易量转化为当前交易量之后，再输入聚合引擎。使用字典 dictVol 保存每只股票上一条消息中的累计交易量，以计算当前交易量。由于若 handler 参数是函数时，必须仅有一个参数，即本批次订阅到的消息，所以calcVolume必须包括 msg 这样一个代表本批次订阅数据的参数，且在subscribeTable函数中使用时，使用“部分应用”将其它参数固化。

def calcVolume(mutable dictVolume, mutable tsAggrOHLC, msg){
	t = select Symbol, DateTime, Price, Volume from msg context by Symbol limit -1 
	update t set prevVolume = dictVolume[Symbol]
	dictVolume[t.Symbol] = t.Volume
	tsAggrOHLC.append!(t.update!("Volume", <Volume - prevVolume>).dropColumns!("prevVolume"))
}

tradesData = loadText(yourDIR + "sampleTrades2.csv")

//定义流数据表Trade
x=tradesData.schema().colDefs
share streamTable(100:0, x.name, x.typeString) as Trade

//定义OHLC输出表
share streamTable(100:0, `datetime`symbol`open`high`low`close`volume`updatetime,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE,LONG,TIMESTAMP]) as OHLC

//定义实时聚合引擎：每分钟计算过去5分钟K线
tsAggrOHLC = createTimeSeriesAggregator(name="aggr_ohlc", windowSize=300000, step=60000, metrics=<[first(Price),max(Price),min(Price),last(Price),sum(Volume),now()]>, dummyTable=Trade, outputTable=OHLC, timeColumn=`Datetime, keyColumn=`Symbol)

//订阅流数据写入聚合引擎
dictVol = dict(STRING, DOUBLE)
subscribeTable(tableName="Trade", actionName="minuteOHLC2", offset=0, handler=calcVolume{dictVol,tsAggrOHLC}, msgAsTable=true)

replay(inputTables=tradesData, outputTables=Trade, dateColumn=`Datetime)

最后查看结果：

select top 10 * from OHLC 

响应式状态引擎支持一部分优化过的序列处理函数，因此除了通过定义函数calcVolume来将累计交易量转化为当前交易量以外，可以通过响应式状态引擎和时间序列聚合引擎串联来实现同样的效果：

tradesData = loadText(yourDIR + "sampleTrades2.csv")

//定义流数据表Trade
x=tradesData.schema().colDefs
share streamTable(100:0, x.name, x.typeString) as Trade

//定义OHLC输出表
share streamTable(100:0, `datetime`symbol`open`high`low`close`volume`updatetime,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE,LONG,TIMESTAMP]) as OHLC

//定义实时聚合引擎：每分钟计算过去5分钟K线
tsAggrOHLC = createTimeSeriesAggregator(name="aggr_ohlc", windowSize=300000, step=60000, metrics=<[first(Price),max(Price),min(Price),last(Price),sum(Volume),now()]>, dummyTable=Trade, outputTable=OHLC, timeColumn=`Datetime, keyColumn=`Symbol)

//定义响应式状态引擎：预处理Volume数据
rsAggrOHLC = createReactiveStateEngine(name="calc_vol", metrics=<[Datetime, Price, deltas(Volume)]>, dummyTable=Trade, outputTable=tsAggrOHLC, keyColumn=`Symbol)

//订阅流数据写入聚合引擎
subscribeTable(tableName="Trade", actionName="minuteOHLC2", offset=0, handler=append!{rsAggrOHLC}, msgAsTable=true)

replay(inputTables=tradesData, outputTables=Trade, dateColumn=`Datetime)

最后查看结果：

select top 10 * from OHLC 

有状态的因子，即因子的计算不仅用到当前数据，还会用到历史数据。实现状态因子的计算，一般包括这几个步骤：（1）保存本批次的消息数据到历史记录；（2）根据更新后的历史记录，计算因子，（3）将因子计算结果写入输出表中。如有必要，删除未来不再需要的的历史记录。

由于 DolphinDB 的消息处理函数必须是单目函数，且唯一的参数就是当前的消息。要保存历史状态并且可以在消息处理函数中引用它，可以使用部分应用，定义一个多个参数的消息处理函数，其中一个参数用于接收消息，其它所有参数被固化，用于保存历史状态。这些固化参数只对消息处理函数可见，不受其他应用的影响。

历史状态可保存在内存表，字典或分区内存表中。本节的三个例子分别演示如何使用这三种方法来保存历史状态并计算因子。

一些较为复杂的因子可能需要使用流水线处理（或者链式处理）。一种常见场景为：先根据输入的行情数据使用时间序列聚合引擎生成分钟级K线，然后根据分钟级K线生成状态因子。可将消息处理函数的输出指向另一个流数据表，以实现流水线处理。

下例演示因子计算的流水线处理。首先计算每只股票分钟级的K线，然后根据最近的10个K线记录计算前面5个时间段的资金净流入量与后面5个时间段的资金净流入量的比例。

tradesData = loadText(yourDIR + "sampleTrades1.csv")

defg factorMoneyFlowRatio(x){
	n = x.size()
	if(n < 9) return double()
	else return x.subarray((n-9):(n-5)).sum()\x.subarray((n-4):n).sum()
}

def factorHandler2(mutable historyDict, mutable factors, msg){
	netAmount = exec volume * iif(close>=open, 1, -1) from msg
	dictUpdate!(historyDict, append!, msg.symbol, netAmount, x->array(x.type(), 0, 500).append!(x))
	syms = msg.symbol.distinct()
	cnt = syms.size()
	v = array(DOUBLE, cnt)
	for(i in 0:cnt){
	    v[i] = factorMoneyFlowRatio(historyDict[syms[i]])
	}
	factors.tableInsert([take(now(), cnt), syms, v])
}

//定义流数据表Trade
x=tradesData.schema().colDefs
share streamTable(100:0, x.name, x.typeString) as Trade

//定义OHLC输出表
share streamTable(100:0, `datetime`symbol`open`high`low`close`volume`updatetime,[TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE,LONG,TIMESTAMP]) as OHLC

//定义实时聚合引擎：每分钟计算过去1分钟K线
tsAggrOHLC = createTimeSeriesAggregator(name="aggr_ohlc", windowSize=60000, step=60000, metrics=<[first(Price),max(Price),min(Price),last(Price),sum(Volume),now()]>, dummyTable=Trade, outputTable=OHLC, timeColumn=`Datetime, keyColumn=`Symbol)

//订阅流数据写入聚合引擎
subscribeTable(tableName="Trade", actionName="minuteOHLC3", offset=0, handler=append!{tsAggrOHLC}, msgAsTable=true)

//订阅流数据表OHLC，计算指标，并输出到流数据表factors
dictHistory = dict(STRING, ANY)
share streamTable(100000:0, `timestamp`symbol`factor, [TIMESTAMP,SYMBOL,DOUBLE]) as factors
subscribeTable(tableName="OHLC", actionName="calcMoneyFlowRatio", offset=0, handler=factorHandler2{dictHistory,factors}, msgAsTable=true)

replay(inputTables=tradesData, outputTables=Trade, dateColumn=`Datetime)

查看结果：

select top 10 * from factors where isValid(factor)

使用流水线计算因子时，需要合理安排消息的执行线程。最理想的情况是每一个环节由不同的线程来完成，以降低延迟。流数据消息的默认执行模式是为每一个线程分配一个消息队列，订阅的消息进入了不同的队列，就由相应的线程来执行。那如何为一个订阅指定执行线程或队列呢？可以使用函数subscribeTable的可选参数 hash 。如果订阅的执行线程总数为N，那么分配的线程序号（从0开始）为 N 除以 hash 的余数。订阅的执行线程总数可以通过配置变量 subExecutors 来设置，默认值为1。

本教程中使用了单机模式的 DolphinDB Server。配置文件（dolphindb.cfg）的内容如下：

mode=single
workerNum=4
maxPubConnections=8
subPort=20001
persistenceDir=dbCache
subThrottle=1
subExecutors=2

• 单机模式下，本节点既是发布端又是订阅端。作为发布端的必需参数是 maxPubConnections；作为订阅端的必需参数是 subPort，指定订阅线程监听的端口号。
• 通常情况下，流数据表的数据量会随时间不断增长。为了避免内存被耗尽，生产环境下建议设定配置参数 persistenceDir 以启用流数据表持久化。
• 鉴于“有状态因子计算”一节使用了毫秒级别的 throttle，需要配置 subThrottle = 1。
• 鉴于“因子计算流水线”和“提高计算效率的方法”提到使用订阅系统的线程池和全局线程池，配置 subExecutors 和 workerNum。