窄表响应式状态引擎

业务场景常需动态扩展计算指标。响应式状态引擎通常要求计算逻辑在初始化时完全确定；相比之下，窄表响应式状态引擎提供了动态添加指标的能力。指标更新后，无需重启引擎即可对后续数据流生效。

首先，使用窄表响应式状态引擎计算累计均价、累计最高价和累计成交量这三个指标，其计算代码如下：

dummy =  streamTable(1:0, `sym`date`time`price`qty, [STRING,DATE,TIME,DOUBLE, INT])
outputTable = streamTable(10000:0, `sym`date`time`metricNames`factorValue, [STRING,DATE,TIME,STRING,DOUBLE])
//这里定义计算指标，本例中定义了累计均价、累计最高价和累计成交量三个指标。用户可根据实际需要修改指标。
factor = [<date>, <time>, <cumavg(price)>, <cummax(price)>, <cumsum(qty)>]
Narrowtest = createNarrowReactiveStateEngine(name="narrowtest", metrics=factor, metricNames=["cumAvgPx","cumMaxPx","cumTotalQty"],dummyTable=dummy,outputTable=outputTable,keyColumn="sym")

//模拟数据
num = 5
tmp = table(take("A" + lpad(string(1..4),4,"0"),num) as sym, take(2023.09.01,num) as date, 2023.09.01 00:00:00+take(1..num, num).sort() as time, take(rand(100.0,num) join take(int(),30),num) as price, rand(50,num) as qty)
Narrowtest.append!(tmp)

上例通过 createNarrowReactiveStateEngine 引擎计算累计均价、累计最高价和累计成交量这三个指标，并将计算结果输出到窄表中，即三个指标的结果依次输出到同一列。其结果示例如下：

以新增累计最大成交量指标为例，其具体代码如下：

//新增累计最大成交量指标
metrics = [<cummax(qty)>]
addReactiveMetrics("narrowtest", "cumMaxQty", metrics)

//新增计算指标后，再次向引擎注入数据，则会输出更新后的指标计算结果
tmpNew = table("A0001" as sym, 2023.09.01 as date, 2023.09.01 00:00:06 as time, 88 as price, 666 as qty)
Narrowtest.append!(tmpNew)

新增计算指标后，引擎会对后续接收的新数据立即生效，其输出结果为：

DolphinDB 内置的流计算引擎包括响应式状态引擎，时间序列聚合引擎，横截面引擎和异常检测引擎等。这些引擎都以数据表作为输入和输出，因此解决复杂的因子计算问题时，可以将多个流计算引擎通过级联的方式合并成一个复杂的数据流拓扑。窄表响应式状态引擎适用于下游基于多因子窄表处理的场景，如下游因子计算依赖于不同因子的最新值。

本例首先通过窄表响应式状态引擎计算每只产品的累计均价和累计成交量，进一步使用响应式无状态引擎对其中一只产品的累计均价进行调整。

/**
    @ 若执行过示例 1，可使用以下语句清理计算环境
    dropStreamEngine("narrowtest")
 */
share streamTable(1:0, `sym`date`time`price`qty, [STRING,DATE,TIME,DOUBLE, INT]) as tickStream
share streamTable(1:0, `sym`metricNames`factorValue, [STRING,STRING,DOUBLE]) as tempResultStream
result = table(100:0, `sym`metricNames`factorValue, [STRING,STRING,DOUBLE])

// 创建 metrics 描述数据间的依赖关系
metrics = array(ANY, 0, 0)
metric = dict(STRING,ANY)
// 依赖关系 product_A:factorNew=product_A:factor1*product_A:factor2/(product_A:factor2+product_B:factor2)
metric["outputName"] = `product_A:`factorNew
metric["formula"] = <A*B\(B+C)>
metric["A"] = `product_A:`factor1
metric["B"] = `product_A:`factor2
metric["C"] = `product_B:`factor2
metrics.append!(metric)

rsle = createReactiveStatelessEngine("reactiveStateless", metrics, result)

//这里定义计算指标，本例中定义了累计均价和累计成交量两个指标。用户可根据实际需要修改指标。
factor = [<cumavg(price)>, <cumsum(qty)>]
Narrowtest = createNarrowReactiveStateEngine(name="narrowtest", metrics=factor, metricNames=["factor1","factor2"],dummyTable=tickStream,outputTable=tempResultStream,keyColumn="sym")
subscribeTable(tableName=`tickStream, actionName="factors", handler=tableInsert{Narrowtest})
subscribeTable(tableName=`tempResultStream, actionName="factorsNew", handler=tableInsert{rsle})

//模拟数据
num = 10
tmp = table(take(`product_A`product_B, num) as sym, take(2023.09.01, num) as date, 09:30:00+(1..num) as time, rand(10.0, num) as price, rand(1000, num) as qty)
tickStream.append!(tmp)

结果示例如下：