Level 2 实时行情数据的流式实现

参考：https://gitee.com/dolphindb/Tutorials_CN/blob/master/Level-2_stock_data_processing.md#4-%E5%9F%BA%E4%BA%8E-level-2-%E5%AE%9E%E6%97%B6%E8%A1%8C%E6%83%85%E6%95%B0%E6%8D%AE%E7%9A%84%E6%B5%81%E5%BC%8F%

六mo神剑

450人浏览 · 2024-03-19 17:44:54

六mo神剑 · 2024-03-19 17:44:54 发布

4. 基于 Level 2 实时行情数据的流式实现
本章节介绍基于 DolphinDB 实时计算 Level 2 的行情数据。DolphinDB 的计算引擎不仅可以用于量化金融基于历史数据的批量开发计算，也大量用于生产环境的实时计算。

4.1 快照实时行情数据的高频因子流批一体实现
DolphinDB 的响应式状态引擎（Reactive State Engine），接收一个在历史数据上经过验证的 DolphinDB 因子代码，应用于实时行情数据，输出实时因子值，实现了高频因子流批一体的解决方案。具体教程请参考金融高频因子的流批统一计算：DolphinDB 响应式状态引擎介绍

如上3.1节中基于快照历史行情数据计算的时间加权订单斜率、成交价加权净委买比例、十档委买增额和十档买卖委托均价线性回归斜率因子表达式，将这些因子直接代入响应式状态引擎，以实现对流数据的实时因子计算。

在基于历史数据批量计算加权平均订单失衡率因子的表达式中，在进行因子标准化处理时，使用了迭代的算法（当前标准差很小时，使用上一标准化的因子值），这类因子在流式计算时需要改为流式引擎计算支持的表达式。而其它的高频因子计算中没有涉及迭代算法，可以直接代入流式表达式进行实时因子计算。

十档平均委卖订单斜率因子流式计算实现如下：

@state
def wavgSOIRStream(bidQty,askQty,lag=20){
	Imbalance_=rowWavg((bidQty-askQty)\(bidQty+askQty),
	 10 9 8 7 6 5 4 3 2 1)
	Imbalance= ffill(Imbalance_).nullFill(0)
	mean = mavg(prev(Imbalance), (lag-1), 2)
	std = mstdp(prev(Imbalance) *1000000, (lag-1),2) \ 1000000
	factorValue = conditionalIterate(std >= 0.0000001,
	(Imbalance - mean) \ std, cumlastNot)
	return ffill(factorValue).nullFill(0)
}
把批计算中的 iif(std >= 0.0000001,(Imbalance - mean) \ std, NULL)改为 conditionalIterate(std >= 0.0000001,(Imbalance - mean) \ std, cumlastNot)。

conditionalIterate 函数只适用于响应式状态引擎，通过条件迭代实现因子中的递归逻辑。假设该函数计算结果对应输出表的列为 factor，且迭代仅基于前一个值，对于第 k 条记录（k = 0, 1, 2 …），其计算逻辑为：

cond[k] == true：factor[k] = trueValue
cond[k] == false：factor[k] = falseIterFunc(factor)[k-1]
基于Level 2 行情快照数据的高频因子流式计算脚本如下：

metrics = array(ANY, 5)	
metrics[0]=<DateTime>
metrics[1] = <timeWeightedOrderSlope(BidPrice[0],BidOrderQty[0],OfferPrice[0],OfferOrderQty[0],20)>
metrics[2] =<level10_InferPriceTrend(BidPrice,OfferPrice,BidOrderQty,OfferOrderQty,60,20)>
metrics[3] =<traPriceWeightedNetBuyQuoteVolumeRatio(BidPrice[0],BidOrderQty[0],OfferPrice[0],OfferOrderQty[0],TotalValueTrade,TotalVolumeTrade,20)> 
metrics[4] =<wavgSOIRStream( BidOrderQty,OfferOrderQty,20)>

share streamTable(1:0, `SecurityID`DateTime`BidPrice`OfferPrice`BidOrderQty`OfferOrderQty`TotalValueTrade`TotalVolumeTrade, [STRING,TIMESTAMP,DOUBLE[],DOUBLE[],DOUBLE[],DOUBLE[],INT,INT]) as Streamdata

result = table(1000:0, `SecurityID`DateTime`TimeWeightedOrderSlope`Level10_InferPriceTrend`TraPriceWeightedNetBuyQuoteVolumeRatio`height_Imbalance, [STRING,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE,DOUBLE])
rse = createReactiveStateEngine(name="reactiveDemo", metrics =metrics, dummyTable=Streamdata, outputTable=result, keyColumn="SecurityID")
subscribeTable(tableName=`Streamdata, actionName="factors", handler=tableInsert{rse})
snapshotTB=select* from loadTable("dfs://TSDB_snapshot","snapshot") where date(TradeTime)=2022.04.14 and SecurityID in [`600000,`000001]
data1=select SecurityID,TradeTime as DateTime,BidPrice,OfferPrice,BidOrderQty,OfferOrderQty,TotalValueTrade,TotalVolumeTrade from snapshotTB 
Streamdata.append!(data1)
流式计算输出结果展示：


4.2 延时成交订单因子的流式实现
DolphinDB 内置的流计算引擎除了响应式状态引擎外，还有时间序列聚合引擎、横截面引擎、异常检测引擎和各种连接引擎。这些引擎均实现了数据表（table）的接口，因此使得多个引擎间的流水线处理变得异常简单。只要将后一个引擎作为前一个引擎的输出即可。引入流水线处理，可以解决更为复杂的因子计算问题。

本节通过流数据引擎的级联构建计算模型，以实现实时计算延时成交订单因子,详细代码见附件。

4.2.1 实现思路
整体计算流程如下图所示：


涉及到的流数据引擎有：左半等值连接引擎，响应式状态引擎，时间序列引擎。

流程说明：

左半等值连接引擎，把逐笔成交与逐笔委托数据关联，并计算延时成交订单的累计值和订单累计成交量；
用响应式状态引擎，计算每只股票延时订单因子和延时订单成交量；
用时间序列引擎 获取最新一分钟的延时订单因子和延时订单成交量，并输出结果。
各引擎间直接级联，无需通过中间表。详情可参考：用户手册流数据引擎主题。

4.2.2 实时计算延时成交订单因子
第一步：
tradeTable 和 entrustTable 分别为逐笔成交表和逐笔委托流表。通过创建两个左半等值连接引擎把两个表基于买卖订单进行等值关联。

metrics = [
<tradeTable.DateTime>,
<entrustTable.DateTime>,
<TradePrice>,
<TradeQty>,
<cumsum(iif((tradeTable.DateTime-entrustTable.DateTime)>60000,1,0)) as DelayedTraderflag>,
<Side>,
<cumsum(TradeQty)>]
lsjEngineBid=createLeftSemiJoinEngine("lsjEngineBid", tradeTable, entrustTable, lsjoutput, metrics,[[`SecurityID,`BidApplSeqNum],[`SecurityID,`ApplSeqNum]],50000000,true)
subscribeTable(tableName="tradeTable", actionName="Bid", offset=0, handler=appendForJoin{lsjEngineBid, true}, msgAsTable=true)
subscribeTable(tableName="entrustTable", actionName="Bid", offset=0, handler=appendForJoin{lsjEngineBid, false}, msgAsTable=true)

lsjEngineOffer=createLeftSemiJoinEngine("lsjEngineOffer", tradeTable, entrustTable, lsjoutput, metrics,[[`SecurityID,`OfferApplSeqNum],[`SecurityID,`ApplSeqNum]],50000000,true)
subscribeTable(tableName="tradeTable", actionName="Offer", offset=0, handler=appendForJoin{lsjEngineOffer, true}, msgAsTable=true)
subscribeTable(tableName="entrustTable", actionName="Offer", offset=0, handler=appendForJoin{lsjEngineOffer, false}, msgAsTable=true)
左半等值连接引擎返回一个左、右表关联后的表对象。对于左表每一条数据，都去匹配右表相同 matchingColumn 的数据，若无匹配的右表记录，则不输出。若匹配多条右表记录，则由 updateRightTable 参数决定连接右表的第一条记录还是最后一条记录。updateRightTable 为可选参数，默认为 false，表示右表存在多条相同 matchingColumn 的记录时，是保留第一条（false）还是最后一条记录（true）。这里订单成交时间需要与订单的最早委托时间关联，所以 updateRightTable 取默认值即可。

metrics 中计算订单的成交量、订单的累计成交量，成交时间和委托时间差是否大于1分钟的累计次数和订单的买卖方向等指标。

第二步：
通过响应式状态引擎，计算每一只股票的买卖延时订单数量和买卖延时订单成交量指标。

@state
def delayedTradeNum(bsFlag, flag, side){
      return iif(bsFlag==side && flag<=1, flag, 0).cumsum()

}
@state
def delayedTradeQty(bsFlag, flag, tradeQty, cumTradeQty, side){
        return iif(bsFlag==side && flag>1, tradeQty, iif(bsFlag==side && flag==1, cumTradeQty, 0)).cumsum()
}
metrics = array(ANY, 5)	
metrics[0]=<TradeTime>
metrics[1]=<delayedTradeNum(BuySellFlag,DelayedTradeFlag,"B")>
metrics[2]=<delayedTradeNum(BuySellFlag,DelayedTradeFlag,"S")>
metrics[3]=<delayedTradeQty(BuySellFlag,DelayedTradeFlag,TradeQty,cumTradeQty,"B")>
metrics[4]=<delayedTradeQty(BuySellFlag,DelayedTradeFlag,TradeQty,cumTradeQty,"S")>

secondrse = createReactiveStateEngine(name="reactiveDemo", metrics =metrics, dummyTable=lsjoutput, outputTable=RSEresult, keyColumn=["code"],filter=<TradePrice>0>)
subscribeTable(tableName=`lsjoutput, actionName="DelayedTrader", handler=tableInsert{secondrse})
DelayedTradeFlag 指标为订单延时成交累计记录。统计股票的延时订单数时，对每笔订单只需统计一次，计算股票的延时成交订单数只需计算 iif(bsFlag==side&&flag<=1,flag,0) 的累计值。

iif((BuySellFlag=="B")&&(DelayedTraderFlag>1),TradeQty,iif((BuySellFlag=="B")&&(DelayedTraderFlag==1),DelayedTraderFlag*cumTradeQty,0))
使用订单延时成交累计记录指标 DelayedTradeFlag 来计算延时订单的成交量。当首次记录为延时成交时，成交量为订单的累计成交量，当 DelayedTradeFlag>1 时，成交量只需加上当前的交易量。

第三步：
通过时间序列引擎引擎，计算每一分钟每只股票最新的买卖延时订单数量和买卖延时订单成交量指标。

tsengine = createTimeSeriesEngine(name="TSengine", windowSize=60000, step=60000, metrics=<[last(DelayedTradeBuyOrderNum),last(DelayedTradeSellOrderNum),last(DelayedTradeBuyOrderQty),last(DelayedTradeSellOrderQty)]>, dummyTable=RSEresult, outputTable=result, timeColumn=`TradeTime, useSystemTime=false, keyColumn=`code, garbageSize=50, useWindowStartTime=false)
subscribeTable(tableName="RSEresult", actionName="TSengine", offset=0, handler=append!{tsengine}, msgAsTable=true);