conditionalIterate

conditionalIterate(cond, trueValue, falseIterFunc)

This function can only be used as a state function in the reactive state engine.

cond is a conditional expression or a function with BOOLEAN return values. It must contain fields from the input table. Constants/constant expressions are not supported.

trueValue is the calculation formula.

falseIterFunc is the function for iteration, whose only parameter is the column from the output table. Currently, only the following functions are supported (use partial application to specify functions with multiple parameters):

• Moving functions: tmove, tmavg, tmmax, tmmin, tmsum, mavg, mmax, mmin, mcount, msum;

• Cumulative window functions: cumlastNot, cumfirstNot;

• Order-sensitive functions: ffill, move.

If cond returns true, the calculation of trueValue is triggered. If cond returns false, falseIterFunc is called for iteration.

• As the iterations are performed based on the historical data, the output for the current record is calculated based on the historical results in the output table and X.

• When calculating with time-based moving windows, windows are determined by the current timestamp T, i.e., (T - window, T).

• The data type of the result column is determined by the result of trueValue. Therefore, the data type of the result of trueValue and falseIterFunc must be compatible. Data type compatibility rules:

  • INT, SHORT, LONG and CHAR are compatible;

  • FLOAT and DOUBLE are compatible;

  • STRING and SYMBOL are compatible.

Supposing the iteration is based only on the previous result, for the k-th (k ∈ N+) record, the calculation logic is (where the column "factor" holds the results):

• cond[k] == true: factor[k] = trueValue

• cond[k] == false: factor[k] = falseIterFunc(factor)[k-1]

Note: If falseIterFunc is a window function, the iteration is based on multiple previous results.

Example 1. Understand the calculation logic of conditionalIterate with the following example.

trade = table(take("A", 10) as sym,  take(1 3 6, 10) as val0,  take(10, 10) as val1)
trade

Define a reactive state stream engine and group the data by "sym" for calculation.

• If val0 > 5 returns true, the formula is factor[k]=trueValue. Output the value of val1.

• If val0 > 5 returns false, the corresponding formula is factor[k]=falseIterFunc(factor)[k-1]. When k=3, then the corresponding val0=1, val1=10, factor=[NULL, NULL, 10], the result is msum([NULL, NULL, 10], 3)[2]=10. Similarly, when k=4, the corresponding val0=3, val1=10, factor=[NULL, NULL, 10, 10], the result is msum([NULL, NULL, 10, 10], 3)[3]=20.

inputTable = streamTable(1:0, `sym`val0`val1, [SYMBOL, INT, INT])
outputTable = table(100:0, `sym`factor, [STRING, DOUBLE])
rse = createReactiveStateEngine(name="rsTest", metrics=<conditionalIterate(val0 > 5, val1, msum{, 3})>, dummyTable=inputTable, outputTable=outputTable, keyColumn="sym")
rse.append!(trade)
select * from outputTable

Example 2. The calculation logic of "factor":

def factor(TotalVolumeTrade, TotalValueTrade, HighPx, LowPx){
    factorValue = iif(TotalVolumeTrade < 1500000, pow(HighPx*LowPx, 0.5)-(TotalVolumeTrade/TotalValueTrade), mavg(factor(TotalVolumeTrade, TotalValueTrade, HighPx, LowPx), 3))
    return factorValue
}

The calculation of factor involves recursion. You can use function conditionalIterate to implement the factor in a reactive state engine with the following script:

@state
def factor1(TotalVolumeTrade, TotalValueTrade, HighPx, LowPx){
   factorValue = conditionalIterate(TotalVolumeTrade < 1500000, (pow(HighPx*LowPx, 0.5)-(TotalVolumeTrade/TotalValueTrade)), mavg{,3})
   return factorValue
}

SecurityID =  ["000001.SZ","000001.SZ","000001.SZ","000001.SZ","000001.SZ","000001.SZ","000001.SZ","000001.SZ","000001.SZ","000001.SZ"]$SYMBOL
Date = [2022.04.01,2022.04.01,2022.04.01,2022.04.01,2022.04.01,2022.04.01,2022.04.01,2022.04.01,2022.04.01,2022.04.01]
Time = [09:30:00.000,09:30:03.000,09:30:06.000,09:30:09.000,09:30:12.000,09:30:15.000,09:30:18.000,09:30:21.000,09:30:24.000,09:30:27.000]
TotalVolumeTrade = [844800,1035700,1240100,1304500,1457800,1522400,1550900,1663800,1692100,1767100]
TotalValueTrade = [12982101,15908020,19038479,20022525,22363886,23349799,23784950,25506625.75,25937850.75,27080561.75]
HighPx = [15.37,15.37,15.37,15.37,15.37,15.37,15.37,15.37,15.37,15.37]
LowPx = [15.3,15.3,15.29,15.28,15.24,15.24,15.24,15.22,15.22,15.22]
trade = table(SecurityID, Date, Time, TotalVolumeTrade, TotalValueTrade, HighPx, LowPx)

result = table(1:0, `SecurityID`Date`Time`Factor, `SYMBOL`DATE`TIME`DOUBLE)

factor=[<Date>,<Time>, <factor1(TotalVolumeTrade, TotalValueTrade, HighPx, LowPx)>]
rse = createReactiveStateEngine(name="rsTest", metrics=factor, dummyTable=trade, outputTable=result, keyColumn="SecurityID")
rse.append!(trade)

trade1 = select *,  (pow(HighPx*LowPx, 0.5)-(TotalVolumeTrade/TotalValueTrade)) as Factor0 from trade
select * from lj(trade1, result, `SecurityID`Date`Time)
dropStreamEngine("rsTest")

The following is part of the result:

For records in blue box, data in column Factor is the results of trueValue, which are the same as the output in Factor0. For records in red box, data in column Factor is the results of falseIterFunc, and each of them is the average of the previous three results.

Related functions: stateIterate