双分组响应式状态引擎

在金融实际场景中，常常需要基于逐笔成交数据分别按照买卖订单号聚合进行相应的计算。本例对逐笔成交数据中的成交金额分别按照买单号和卖单号进行聚合，计算相应的累计成交金额。

// 定义表结构
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

// 定义双分组响应式状态引擎
dors = createDualOwnershipReactiveStateEngine(
    name="test", 
    metrics1=[<TradeTime>, <cumsum(TradeAmount)>], 
    metrics2=<cumsum(TradeAmount)>, 
    dummyTable=tickStream, 
    outputTable=processOrder, 
    keyColumn1=`SecurityID`BuyNum, 
    keyColumn2=`SecurityID`SellNum
)

// 订阅流数据表
subscribeTable(tableName=`tickStream, actionName="test", msgAsTable=true, handler=tableInsert{dors})

// 创建模拟数据生成函数
def generateMockTradeData(n, startTime) {
    // 生成基础数据
    securities = `000001.SZ`000002.SZ`600000.SH`600036.SH
    sides = `B`S
    
    // 生成随机交易数据
    securityList = take(securities, n)
    timeList = startTime + 1..n * 1000  // 每秒一条数据
    priceList = 10.0 + rand(100.0, n)  // 价格在10-110之间
    qtyList = rand(1000, n) + 100  // 成交量100-1100股
    amountList = priceList * qtyList  // 计算成交金额
    buyNumList = rand(1000..10000, n)  // 买方订单号
    sellNumList = rand(1000..10000, n)  // 卖方订单号
    bsFlagList = take(sides, n)  // 买卖方向
    
    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
    )
}

// 生成并注入模拟数据
mockData = generateMockTradeData(10, 2023.02.01T09:30:00.000)
tickStream.append!(mockData)

// 查询结果
select * from processOrder

其结果示例如下：