createThresholdEngine

createThresholdEngine(name, threshold, metrics, dummyTable, outputTable, thresholdColumn, [keyColumn], [timeColumn], [sessionBegin], [sessionEnd], [keyPurgeDaily=false], [forceTriggerSessionEndTime=0], [snapshotDir], [snapshotIntervalInMsgCount], [outputElapsedMicroseconds=false], [outputThreshold=false])

This function creates a threshold engine to implement aggregate calculations triggered by cumulative value thresholds. Each time the cumulative value of the threshold column (thresholdColumn) reaches a specific threshold (n * threshold), an aggregate calculation is triggered. Here, n is a positive integer that starts from 1 and increments by 1. It indicates which threshold boundary the current cumulative value has reached. The trigger boundaries are threshold, 2 * threshold, 3 * threshold, and so on.

• If keyColumn is specified, the aggregate calculation will be performed independently in each group.
• If sessionBegin and sessionEnd are specified, only data in the range [sessionBegin, sessionEnd] will be included in the calculation.

name is a STRING scalar that specifies the name of the engine. It is the only identifier of an engine on a data or compute node. It can have letter, number and "_" and must start with a letter.

threshold is a positive integer that specifies the threshold step size.

metrics is metacode scalar, vector, or a tuple that specifies the calculation formulas. For more information about metacode please refer to Metaprogramming.

• It can use one or more built-in or user-defined aggregate functions (which must be defined by the defg keyword) such as <[sum(volume), avg(price)]>, or expressions of aggregate functions such as as <[avg(price1)-avg(price2)]>, or aggregate functions involving multiple columns such as <[std(price1-price2)]>.
• You can specify functions that return multiple values for metrics, such as <func(price) as `col1`col2> (it's optional to specify the column names).

dummyTable is a table object whose schema must be the same as the input stream table. Whether dummyTable contains data does not matter.

outputTable is the output table for calculation results. It can be an in-memory table or a DFS table. Before using the createThresholdEngine function, you must predefine the output table as an empty table and specify the column names and data types. After the threshold engine triggers a calculation, it inserts the output results into this table in column-position order.

The output table columns are ordered as follows:

keyColumn (optional) is a STRING scalar or vector that specifies the name of the grouping column(s). If it is specified, the engine conducts the calculations within each group. For example, group the data by stock symbol and apply moving aggregation functions to each stock.

timeColumn (optional) is a STRING scalar or vector that specifies the time column(s) of the subscribed stream table. When useSystemTime = false, it must be specified.

Note: If timeColumn is a vector, it must have a date element (of DATE type) and a time element (of TIME, SECOND or NANOTIME type). In this case, the first column in outputTable must take the data type of concatDateTime(date, time).

sessionBegin (optional) can be a scalar or vector of type SECOND, TIME or NANOTIME corresponding to the data type of the time column, specifying the starting time of each session. If it is a vector, it must be increasing.

sessionEnd (optional) can be a scalar or vector of type SECOND, TIME or NANOTIME corresponding to the data type of the time column, specifying the end time of each session. Specify sessionEnd as 00:00:00 to indicate the beginning of the next day (i.e., 24:00:00 of the current day).

keyPurgeDaily (optional) is a BOOL scalar. Specifies whether, when the engine receives the first batch of data containing a new date, it first clears all previously saved groups before processing that batch. The default value is false, which means the engine does not clear the previous day's groups.

forceTriggerSessionEndTime (optional) is a non-negative integer. The unit of forceTriggerSessionEndTime is consistent with the precision of timeColumn. It indicates the waiting time to force trigger calculation in the window containing the sessionEnd, if it ends without calculation. The default value is 0, indicating the calculation will not be triggered in this way.

snapshotDir (optional) is a string indicating the directory where the streaming engine snapshot is saved. The directory must already exist, otherwise an exception is thrown. If snapshotDir is specified, the system checks whether a snapshot already exists in the directory when creating a streaming engine. If it exists, the snapshot will be loaded to restore the engine state. Multiple streaming engines can share a directory where the snapshot files are named as the engine names.

The file extension of a snapshot can be:

• <engineName>.tmp: temporary snapshot
  • <engineName>.snapshot: a snapshot that is generated and flushed to disk
  • <engineName>.old: if a snapshot with the same name already exists, the previous snapshot is renamed to <engineName>.old.

snapshotIntervalInMsgCount (optional) is a positive integer indicating the number of messages to receive before the next snapshot is saved.

outputElapsedMicroseconds (optional) is a BOOLEAN value. The default value is false. It determines whether to output the elapsed time (in microseconds) from the time the calculation is triggered to the output of result for each window.

outputThreshold (optional) is a BOOLEAN value. It determines whether to output the threshold window values and the cumulative sums of thresholdColumn.

A table object. Data written to this table is ingested into the engine for aggregation.

Example 1

// data preparation
n = 1000000;
sampleDate = 2019.11.07;
symbols = `600519`000001`600000`601766;
trade = table(take(sampleDate, n) as date, 
	(09:30:00.000 + rand(7200000, n/2)).sort!() join (13:00:00.000 + rand(7200000, n/2)).sort!() as time, 
	rand(symbols, n) as symbol, 
	100+cumsum(rand(0.02, n)-0.01) as price, 
	rand(1000, n) as volume)

// create the dummyTable and outputTable	
share(streamTable(10:0,`date`time`symbol`price`volumn,[DATE, TIME, SYMBOL, DOUBLE, DOUBLE]), `trades);
share(table(1:0, `timestamp`symbol`open`high`low`close, [TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE]), `outputTable);
go
// create the threshold engine
thresholdEngine = createThresholdEngine(name="demo", threshold=1000000, metrics=<[first(price), max(price), min(price), last(price)]>, dummyTable=trades, outputTable=outputTable, thresholdColumn=`volumn, keyColumn=`symbol, timeColumn=[`date, `time]);

// insert data
thresholdEngine.append!(trade);

// query the result
select * from outputTable

Some of results are as follows:

Description of the outputTable definition in this example:

When the engine is created, timeColumn, keyColumn, and metrics are specified, while outputThreshold=true and outputElapsedMicroseconds=true are not specified. Therefore, the columns of outputTable correspond to the following items in order:

Example 2 Restrict calculations to specified trading sessions with sessionBegin and sessionEnd

// Prepare data
t = table(
    2024.01.02 2024.01.02 2024.01.02 2024.01.02 2024.01.02 as date,
    09:20:00.000 09:31:00.000 10:15:00.000 11:35:00.000 13:05:00.000 as time,
    `A`A`A`A`A as sym,
    10.1 10.2 10.3 10.4 10.5 as price,
    200 500 600 700 800 as volume
)

// Create dummyTable and outputTable
share(streamTable(10:0, `date`time`sym`price`volume, [DATE, TIME, SYMBOL, DOUBLE, DOUBLE]), `trades1)
share(table(10:0, `timestamp`sym`sumVol, [TIMESTAMP, SYMBOL, DOUBLE]), `outputTable1)
go
// Create threshold engine
engine = createThresholdEngine(
    name="thresholdDemo1",
    threshold=1000,
    metrics=<[sum(volume)]>,
    dummyTable=trades1,
    outputTable=outputTable1,
    thresholdColumn=`volume,
    keyColumn=`sym,
    timeColumn=[`date, `time],
    sessionBegin=09:30:00.000 13:00:00.000,
    sessionEnd=11:30:00.000 15:00:00.000
)

// Insert data
engine.append!(t)

// Query the results
select * from outputTable1

Example 3 Control whether group states are cleared across trading days with keyPurgeDaily

// Prepare data
t = table(
    2024.01.02 2024.01.03 2024.01.03 as date,
    14:59:00.000 09:31:00.000 09:31:01.000 as time,
    `A`A`A as sym,
    10.1 10.2 10.3 as price,
    600 500 700 as volume
)

// Create dummyTable and outputTable
share(streamTable(10:0, `date`time`sym`price`volume, [DATE, TIME, SYMBOL, DOUBLE, DOUBLE]), `trades)
share(table(10:0, `timestamp`sym`sumVol, [TIMESTAMP, SYMBOL, DOUBLE]), `output1)
share(table(10:0, `timestamp`sym`sumVol, [TIMESTAMP, SYMBOL, DOUBLE]), `output2)
go

// keyPurgeDaily=true: clear the previous day's group state upon receiving data of a new date
engine1 = createThresholdEngine(
    name="thresholdDemo2_true",
    threshold=1000,
    metrics=<[sum(volume)]>,
    dummyTable=trades,
    outputTable=output1,
    thresholdColumn=`volume,
    keyColumn=`sym,
    timeColumn=[`date, `time],
    keyPurgeDaily=true
)

// keyPurgeDaily=false: keep the previous day's group state upon receiving data of a new date
engine2 = createThresholdEngine(
    name="thresholdDemo2_false",
    threshold=1000,
    metrics=<[sum(volume)]>,
    dummyTable=trades,
    outputTable=output2,
    thresholdColumn=`volume,
    keyColumn=`sym,
    timeColumn=[`date, `time],
    keyPurgeDaily=false
)

// Insert data
engine1.append!(t)
engine2.append!(t)

// Query the results
select * from output1

select * from output2

Example 4 Output the elapsed time of each calculation with outputElapsedMicroseconds

// Prepare data
t = table(
    2024.01.02 2024.01.02 2024.01.02 as date,
    09:31:00.000 09:32:00.000 09:33:00.000 as time,
    `A`A`A as sym,
    10.1 10.2 10.3 as price,
    300 400 500 as volume
)

// Create dummyTable and outputTable

share(streamTable(10:0, `date`time`sym`price`volume, [DATE, TIME, SYMBOL, DOUBLE, DOUBLE]), `trades)
share(table(10:0, `timestamp`sym`elapsed`avgPrice`sumVol, [TIMESTAMP, SYMBOL, LONG, DOUBLE, DOUBLE]), `outputTable)
go
// Create threshold engine
engine = createThresholdEngine(
    name="thresholdDemo4",
    threshold=1000,
    metrics=<[avg(price), sum(volume)]>,
    dummyTable=trades,
    outputTable=outputTable,
    thresholdColumn=`volume,
    keyColumn=`sym,
    timeColumn=[`date, `time],
    outputElapsedMicroseconds=true
)

// Insert data
engine.append!(t)

// Query the results
select * from outputTable

// Prepare data
t = table(
    2024.01.02 2024.01.02 2024.01.02 as date,
    09:31:00.000 09:32:00.000 09:33:00.000 as time,
    `A`A`A as sym,
    10.1 10.2 10.3 as price,
    300 400 500 as volume
)

// Create dummyTable and outputTable
share(streamTable(10:0, `date`time`sym`price`volume, [DATE, TIME, SYMBOL, DOUBLE, DOUBLE]), `trades)
share(table(10:0, `timestamp`sym`thresholdValue`cumThreshold`lastPrice, [TIMESTAMP, SYMBOL, LONG, DOUBLE, DOUBLE]), `outputTable)
go

// Create threshold engine
engine = createThresholdEngine(
    name="thresholdDemo5",
    threshold=1000,
    metrics=<[last(price)]>,
    dummyTable=trades,
    outputTable=outputTable,
    thresholdColumn=`volume,
    keyColumn=`sym,
    timeColumn=[`date, `time],
    outputThreshold=true
)

// Insert data
engine.append!(t)

// Query the results
select * from outputTable