Narrow Reactive State Engine

Business scenarios often require you to dynamically add calculation metrics. The reactive state engine generally requires the calculation logic to be fully determined during initialization. In contrast, the narrow reactive state engine allows you to add metrics dynamically. After you update the metrics, the changes take effect for subsequent data streams without requiring an engine restart.

First, use the narrow reactive state engine to calculate the three metrics: cumulative average price, cumulative highest price, and cumulative trading volume. The calculation code is as follows:

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])
// Define the calculation metrics. In this example, three metrics are defined: cumulative average price, cumulative highest price, and cumulative trading volume. You can modify the metrics based on your needs.
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")
// Mock data
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)

The example above uses the narrow reactive state engine to calculate the three metrics: cumulative average price, cumulative highest price, and cumulative trading volume, and outputs the results to a narrow table, with the results of the three metrics output sequentially to the same column. An example of the results is shown below:

The following example adds a cumulative maximum trading volume metric:

// Add cumulative maximum trading volume metric.
metrics = [<cummax(qty)>]
addReactiveMetrics("narrowtest", "cumMaxQty", metrics)
// After adding the new metric, append data into the engine to output the updated metric calculation results.
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)

After adding themetric, the engine immediately applies it to subsequent data. The output:

DolphinDB's built-in streaming engines include the reactive state engine, time-series engine, cross-sectional engine, and anomaly detection engine, etc. All of these engines take tables as input and output, enabling the cascade of multiple streaming engines into a complex data-flow topology for complex factor calculations. The narrow reactive state engine is suitable for downstream processing scenarios that rely on multi-factor narrow tables. For example, when downstream factor calculations depend on the latest values of multiple factors.

First, use the narrow reactive state engine to calculate the cumulative average price and cumulative trading volume for each product. Then, use the reactive stateless engine to adjust the cumulative average price for a product.

/**
    @ If you have executed the code of example 1, use the following code to clean up the environment:
    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])
// Create metrics to describe the dependencies among the data
metrics = array(ANY, 0, 0)
metric = dict(STRING,ANY)
// Dependencies: 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)
// Define calculation metrics. In this example, we define cumulative average price and cumulative trading volume. You can modify the metrics as needed.
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})
// Mock data
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)

The output: