Dual-Ownership Reactive State Engine

In real financial scenarios, you often need to aggregate and calculate metrics based on tick-by-tick trades separately by buy order number and sell order number. This example aggregates trade values in tick-by-tick data by buy and sell order numbers, calculating the corresponding cumulative trading values.

// Define the table schema
colName = `SecurityID`TradeTime`TradePrice`TradeQty`TradeAmount`BuyNum`SellNum`TradeBSFlag
colType = [SYMBOL, TIMESTAMP, DOUBLE, LONG, DOUBLE, LONG, LONG, SYMBOL]
share(streamTable(1000:0, colName, colType), `tickStream)
go
colName = `SecurityID`BuyNum`SellNum`TradeTime`BuyTotalAmount`SellTotalAmount
colType = [SYMBOL, LONG, LONG, TIMESTAMP, DOUBLE, DOUBLE]
share(streamTable(1000:0, colName, colType), `processOrder)
go
// Define the engine
dors = createDualOwnershipReactiveStateEngine(
    name="test", 
    metrics1=[<TradeTime>, <cumsum(TradeAmount)>], 
    metrics2=<cumsum(TradeAmount)>, 
    dummyTable=tickStream, 
    outputTable=processOrder, 
    keyColumn1=`SecurityID`BuyNum, 
    keyColumn2=`SecurityID`SellNum
)
// Subscribe to the stream table
subscribeTable(tableName=`tickStream, actionName="test", msgAsTable=true, handler=tableInsert{dors})
// Create a function to generate mock data
def generateMockTradeData(n, startTime) {
    // Generate basic data
    securities = `000001.SZ`000002.SZ`600000.SH`600036.SH
    sides = `B`S
    // Generate random trade data
    securityList = take(securities, n)
    timeList = startTime + 1..n * 1000  // One data per second
    priceList = 10.0 + rand(100.0, n)  // Price between 10 and 110
    qtyList = rand(1000, n) + 100  // Quantity between 100 and 1100 shares
    amountList = priceList * qtyList  // Calculate trade amount
    buyNumList = rand(1000..10000, n)  // Buy order number
    sellNumList = rand(1000..10000, n)  // Sell order number
    bsFlagList = take(sides, n)  // Buy/Sell flag
    return table(
        securityList as SecurityID,
        timeList as TradeTime,
        priceList as TradePrice,
        qtyList as TradeQty,
        amountList as TradeAmount,
        buyNumList as BuyNum,
        sellNumList as SellNum,
        bsFlagList as TradeBSFlag
    )
}
// Generate and append mock data
mockData = generateMockTradeData(10, 2023.02.01T09:30:00.000)
tickStream.append!(mockData)
// Query the result
select * from processOrder

The output: