streamEngineParser

When parsing a metric, the parser follows the following rules:

1. It looks for function signifiers indicating the engine type

   • If a metric involves cross-sectional computation, the corresponding function must be a function signifier (often a function with the row- prefix). For example, to compute the sorted order of elements in each row, the rowRank function must be used to implement this layer of logic.

     Supported function signifiers for cross-sectional computation include:

     "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"

     For logic not expressible with row- functions or must be expressed with user-defined functions, you can use the byRow higher-order function.

   • If a metric involves windowing aggregation over time series, use the rolling higher-order function. For example, to calculate a rolling 3-element sum stepped by 3 elements on the "price" column using forward filling, specify rolling(sum,price,3,3,`ffill`).

     Syntax of rolling:

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

   • Built-in functions without the above function signifiers will be processed by the reactive state engine.

2. Input table schema for intermediate engines

   streamEngineParser allows you to specify the schema for only the input table of the first parsed engine through the dummyTable parameter. The intermediate processing steps are abstracted away from the user. The stream engine parser uses specific naming conventions for the dummyTables of intermediate engines. If you need to reference a column from an intermediate engine's dummyTable in metrics or other parameters, follow these naming conventions:

   • The first column of any intermediate dummyTable is the time column.
   • The next columns are the ones specified by keyColumn.
   • After that come columns holding the calculation results of metrics. These metric columns are named "col_0_, col_1_, col_2_,..." and so on.

metrics is metacode indicating the metrics to be computed by the parser. Tuples are supported. For more information, see Functional Metaprogramming.

• metrics can include built-in or user-defined functions, like <[sum(qty), avg(price)]>; expressions, like <[avg(price1)-avg(price2)]>; computation on columns, like <[std(price1-price2)]>.

• Functions that return multiple values are also supported, such as <func(price) as col1 col2> (column names are optional).

The metrics are typically complex, with multiple layers of computation logic. The parser analyzes each layer of logic and assigns it to the appropriate streaming engine (see Details).

outputTable holds the computation results from the engine pipeline. It can be an in-memory table or a DFS table, but must be empty before calling streamEngineParser. The column names and data types in outputTable must be specified upfront. The schema of outputTable depends on the last engine in the pipeline:

• For a time-series engine, the columns are ordered as:

  1. Time column

     1. The column type is either TIMESTAMP (when useSystemTime = true) or same type as timeColumn (when useSystemTime = false).

     2. It displays window start (when useWindowStartTime = true) or end times (when useWindowStartTime = false).

  2. Key column(s), if keyColumn is specified. They are ordered the same as in keyColumn.

  3. Metric computation result column(s).

• For a reactive state engine, the columns are ordered as:

  1. Key column(s), ordered the same as in keyColumn.

  2. Metric computation result column(s).

• For a cross-sectional engine:

  • If contextByColumn is not specified, the columns of the output table are ordered as:

    1. A TIMESTAMP column with system timestamps for each computation (or the corresponding values from the timeColumn, if this parameter is specified).

    2. Metric computation result column(s). The data types must be consistent with the metric computation results.

  • If contextByColumn is specified, the columns of the output table are ordered as:

    1. A TIMESTAMP column with system timestamps for each computation (or the corresponding values from the timeColumn, if this parameter is specified).

    2. The column specified in contextByColumn.

    3. Metric computation result column(s). The data types must be consistent with the metric computation results.

keyColumn is a string scalar or vector.

• For time-series engines and reactive state engines, keyColumn indicates the columns to group data by. If specified, computations are done per group, such as aggregations by stock.

• For cross-sectional engines, keyColumn indicates the column holding the unique keys. The engine will use only the latest record per key for each computation.

timeColumn specifies the time column in the stream table to which the engine subscribes. It is required when useSystemTime = false.

• For time-series engines and reactive state engines, timeColumn is can be a string scalar or a 2-element string vector. The vector specifies a DATE and a TIME/SECOND/NANOTIME column. The engine will concatenate the date and time values into a single time value for the output table. See concatDateTime() for details on the concatenated data type.

• For cross-sectional engines, timeColumn is a string. The specified column must be of TIMESTAMP type.

useSystemTime indicates how the engine handles time when processing data.

• For cross-sectional engines, useSystemTime is an optional boolean parameter. If true (default), the engine will use the system time when each calculation starts and output it in the first column of the output table. If false, it will use the time values from the specified timeColumn instead.

• For time-series engines, useSystemTime is an optional parameter specifying how data is windowed for processing:

  • When useSystemTime = true, the engine will regularly window the streaming data at fixed time intervals for calculations according to the ingestion time (local system time with millisecond precision, independent of any time columns in the stream table) of each record. As long as a window contains data, the calculation will be performed automatically when the window ends. The first column in output table indicates the timestamps when the calculation occurred. Note: When useSystemTime = true, timeColumn must be left empty.

  • useSystemTime = false (default): the engine will window the streaming data according to the timeColumn in the stream table. The calculation for a window is triggered by the first record after the previous window. Please note that the record which triggers the calculation will not participate in this calculation.

snapshotDir 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.

Below are the parameters unique to a specific type of streaming engine:

• reactive state engines (createReactiveStateEngine): filter, keepOrder, keyPurgeFilter, keyPurgeFreqInSecond

• cross-sectional engines (createCrossSectionalEngine): triggeringPattern, triggeringInterval, lastBatchOnly, contextByColumn

• time-series engines (createTimeSeriesEngine): updateTime, useWindowStartTime, roundTime, forceTriggerTime, closed, fill, garbageSize, keyPurgeFreqInSec

• daily time-series engines (createDailyTimeSeriesEngine): sessionBegin, sessionEnd, mergeSessionEnd, closed, fill, garbageSize, forceTriggerSessionEndTime, keyPurgeFreqInSec

This example implements Formula #1 from World Quant's 101 Alpha Factors on streaming data. The rank function performs a cross-sectional operation and is processed with a cross-sectional engine. rank's parameters are computed using a reactive state engine which outputs to the cross-sectional engine.

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)

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

//create a cross-sectional engine to calculate the rank of each stock
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)
}

//create a reactive state engine which outputs to the cross-sectional engine
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")

The computation can also be done by using the streamEngineParser:

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

//construc metrics
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")