streamEngineParser

streamEngineParser(name, metrics, dummyTable, outputTable, keyColumn, [timeColumn], [useSystemTime=false], [snapshotDir], [snapshotIntervalInMsgCount], [raftGroup], [filter], [keepOrder], [keyPurgeFilter], [keyPurgeFreqInSecond], [triggeringPattern='perBatch'], [triggeringInterval=1000], [lastBatchOnly=false], [contextByColumn], [updateTime], [useWindowStartTime=false], [roundTime=true], [forceTriggerTime], [sessionBegin], [sessionEnd], [mergeSessionEnd=false], [closed='left'], [fill='none'], [garbageSize], [forceTriggerSessionEndTime], [keyPurgeFreqInSec=-1])

在实际场景下，用户的因子可能嵌套横截面、历史状态、时序窗口、表连接等多种复杂逻辑，需要多个计算引擎的级联才能实现。streamEngineParser 为只涉及横截面、历史状态、时序窗口三种逻辑嵌套的复杂因子，提供了一个统一的计算入口。用户无需编写复杂的级联代码，只需将嵌套因子按规则改写后传入streamEngineParser ，系统将自动解析各层嵌套涉及的计算逻辑，形成流计算引擎流水线，自动分发给对应的引擎进行计算。

为满足各逻辑对应引擎的功能，该函数集合了（日级）时序引擎、响应式状态引擎和横截面引擎的参数。需要注意的是，由于部分为公共参数（如 keyColumn），若公共参数在不同计算逻辑下需要指定的值不同（如时序窗口按天分组，状态逻辑要求按股票代码分组，则需要配置不同的 keyColumn），不能使用该函数，只能通过级联实现。

返回一个表对象，为解析后流水线上第一个流计算引擎返回的表，向该表写入数据意味着这些数据进入流水线进行计算。删除时，需要根据 streamEngineParser 解析出的引擎对应的名称（详见 name）单独进行删除。暂不支持通过一个函数删除全部引擎。

根据上文，若要使用 streamEngineParser ，必须根据业务场景提供的指标，分析业务中涉及的流计算引擎，将指标改写成 streamEngineParser 能够识别分解的 metrics。若 metrics 中包含了用户自定义函数，引擎对函数体内部的嵌套逻辑进一步进行解析。

改写规则如下：

1. 引擎分解标识

   • 若指标中的某一层嵌套逻辑要求计算截面数据，需将对应的计算函数改成以 “row开头” 的函数作为标识，若不添加 row 则该操作默认在响应式状态引擎处理。如：需要在截面计算排名，则需要将该层嵌套改写为 rowRank。

     目前支持的标识函数如下：

     "byRow", "rowAvg", "rowCount", "count", "rowDenseRank", "rowMax", "rowMin", "rowProd", "rowRank",
     "rowSize", "rowSkew", "rowStd", "rowStdp", "rowSum", "rowSum2", "rowVar", "rowVarp",
     "rowAnd", "rowOr", "rowXor", "rowKurtosis", "rowCorr", "rowCovar", "rowBeta", "rowWsum", "rowWavg"

     用户自定义函数或者 row 系列函数不支持的逻辑可以通过 byRow 高阶函数实现。

   • 若指标中的某一层嵌套逻辑涉及到时序聚合窗口的计算，必须在指标中以 rolling 作为标识进行改写。如：窗口为 3， 步长为 3，窗口内进行求和计算，计算列为 price，空值以前值填充，则可以写为 rolling(sum, price, 3, 3, 'ffill')。形式如下：

     rolling(func, funcArgs, window, [step=1], [fill='none'])

   • 分解时，不包含上述标识的内置函数均在响应式状态引擎中处理。

2. 中间引擎输入表结构

   在 streamEngineParser 中，用户只能通过参数 dummyTable 指定第一个引擎输入表的 schema ，中间引擎的 dummyTable 对用户而言是透明的， 为了能在 metrics 和部分参数中指定中间表的列名，streamEngineParser 提供了一套中间引擎 dummyTable 的命名规范：

   • 第一列为时间列。

   • 之后几列对应指定的 keyColumn。

   • 后续列为 metrics 对应的计算结果列，列名为：“col_0_, col_1_, col_2_...”。

   注：

   • 由于响应式状态引擎的输出表没有时间列，而时间序列聚合引擎以及横截面引擎的输出表需包含时间列，因此当流水线中涉及到响应式状态引擎相关操作时，其 metrics 会自动增加时间列作为下一个引擎的 dummyTable 的一列。若此时 useSystemTime = true，则其中间表的时间列的列名为 "datatime"。
   • 以下引擎的部分参数需要指定列名，如：
     • 横截面引擎：contextByColumn
     • 响应式状态引擎：filter, keyPurgeFilter
     若这些参数需要指定中间引擎的 dummyTable 的列名，则可以根据如上的命名规则给出的列名进行指定，如 "contextByColumn = col_0_"。若指定的列为 timeColumn 和 keyColumn，则也可以指定参数对应的列名。

name 是一个字符串，表示指定流水线内引擎名称的前缀，可包含字母，数字和下划线，但必须以字母开头。比如指定 name = "test"，则流水线内部对应的引擎名称为 "test0", "test1", "test2"... 其中，数字代表 metrics 分解出的流数据引擎的级联顺序。

dummyTable 一个表对象，和订阅的流数据表的 schema 一致，可以含有数据，亦可为空表。

若要开启快照机制 （snapshot），必须指定 snapshotDir 与 snapshotIntervalInMsgCount。启用快照机制之后，系统若出现异常，可及时将流数据引擎恢复到最新的快照状态。

raftGroup 是流数据高可用订阅端 raft 组的 ID (大于1的整数，由流数据高可用相关的配置项 streamingRaftGroups 指定)。设置该参数表示开启计算引擎高可用。在 leader 节点创建流数据引擎后，会同步在 follower 节点创建该引擎。每次保存的 snapshot 也会同步到 follower。当 raft 组的 leader 节点宕机时，会自动切换新 leader 节点重新订阅流数据表。请注意，若要指定 raftGroup，必须同时指定 snapshotDir。

该流水线引擎只能在 raft 组的 leader 节点创建。

各引擎特有的参数如下：

• 响应式状态引擎 createReactiveStateEngine: filter, keepOrder, keyPurgeFilter, keyPurgeFreqInSecond

• 横截面引擎参数 createCrossSectionalEngine: triggeringPattern, triggeringInterval, lastBatchOnly, contextByColumn

• 时间序列聚合引擎 createTimeSeriesEngine: updateTime, useWindowStartTime, roundTime, forceTriggerTime, closed, fill, garbageSize, keyPurgeFreqInSec

• 日级时间序列聚合引擎 createDailyTimeSeriesEngine: sessionBegin, sessionEnd, mergeSessionEnd, closed, fill, garbageSize, forceTriggerSessionEndTime, keyPurgeFreqInSec

下面的例子是 World Quant 101 个 Alpha 因子中的1号因子公式的流数据实现。rank 函数是一个横截面操作。rank 的参数部分用响应式状态引擎实现。rank 函数本身用横截面引擎实现。横截面引擎作为状态引擎的输出。

n = 100
sym = rand(`aaa`bbb`ccc, n)
time = 2021.01.01T13:30:10.008 + 1..n
maxIndex=rand(100.0, n)
data = table(sym as sym, time as time, maxIndex as maxIndex)

rank(Ts_ArgMax(SignedPower((returns<0?stddev(returns,20):close), 2), 5))-0.5

// 创建横截面引擎，计算每个股票的rank
dummy = table(1:0, `sym`time`maxIndex, [SYMBOL, TIMESTAMP, DOUBLE])
resultTable = streamTable(10000:0, `time`sym`factor1, [TIMESTAMP, SYMBOL, DOUBLE])
ccsRank = createCrossSectionalAggregator(name="alpha1CCS", metrics=<[sym, rank(maxIndex, percent=true) - 0.5]>,  dummyTable=dummy, outputTable=resultTable,  keyColumn=`sym, triggeringPattern='keyCount', triggeringInterval=3000, timeColumn=`time, useSystemTime=false)

@state
def wqAlpha1TS(close){
    ret = ratios(close) - 1
    v = iif(ret < 0, mstd(ret, 20), close)
    return mimax(signum(v)*v*v, 5)
}

// 创建响应式状态引擎，输出到前面的横截面引擎ccsRank
input = table(1:0, `sym`time`close, [SYMBOL, TIMESTAMP, DOUBLE])
rse = createReactiveStateEngine(name="alpha1", metrics=<[time, wqAlpha1TS(close)]>, dummyTable=input, outputTable=ccsRank, keyColumn="sym")
rse.append!(data)

dropStreamEngine("alpha1CCS")
dropStreamEngine("alpha1")

上述操作可以通过直接构建流水线引擎进行替代：

input = table(1:0, `sym`time`close, [SYMBOL, TIMESTAMP, DOUBLE])
resultTable = streamTable(10000:0, `time`sym`factor1, [TIMESTAMP, SYMBOL, DOUBLE])

// 构建计算因子
metrics=<[sym, rowRank(wqAlpha1TS(close), percent=true)- 0.5]>

streamEngine=streamEngineParser(name=`alpha1_parser, metrics=metrics, dummyTable=input, outputTable=resultTable, keyColumn=`sym, timeColumn=`time, triggeringPattern='keyCount', triggeringInterval=3000)
streamEngine.append!(data)

dropStreamEngine("alpha1_parser0")
dropStreamEngine("alpha1_parser1")