处理 Level-2 行情数据实例

订单斜率因子衡量订单价差对订单量差的敏感度。其计算公式如下：

$logQuoteSlope_{t}$ 衡量 t 时刻的订单斜率。其中， $bid_{t}$ 和 $ask_{t}$ 分别表示买一价和卖一价， $bidQty_{t}$ 和 $askQty_{t}$ 分别表示买一量和卖一量。

 @state
def timeWeightedOrderSlope(bid,bidQty,ask,askQty,lag=20){
	return (log(iif(ask==0,bid,ask))-log(iif(bid==0,ask,bid)))\(log(askQty)-log(bidQty)).ffill().mavg(lag, 1).nullFill(0)
}

使用函数mavg计算过去 20 行的移动平均时间加权订单斜率，其中 @state 表示用户自定义的状态函数。状态算子计算时需要用到历史状态，DolphinDB 在流式计算中对自定义状态函数，通过增量的方式实现，性能有很大的提升。

订单失衡率（SOIR）因子衡量买卖委托量在总量中的不均衡程度。其计算公式如下：

表示 $t$ 时刻第 i 档的订单失衡率因子。其中 $bidQty_{i,t}$，$askQty_{i,t}$ 表示第 i 档买方委托量和卖方委托量。为了充分利用各档数据信息，计算各档位加权和订单失衡率因子时，根据买卖压力的影响力赋予不同档位相应的权重。通常，靠近交易价格的档位被赋予更高的权重。计算公式如下：

SOIR 反应盘口各档综合的买卖委托量不均衡程度。如果 SOIR 为正，则说明市场买压大于卖压，未来价格趋势上涨的概率较高。此外，SOIR 值越大，上涨的概率越高，反之亦然。

@state
def wavgSOIR(bidQty,askQty,lag=20){
	imbalance= rowWavg((bidQty - askQty)\(bidQty + askQty), 10 9 8 7 6 5 4 3 2 1).ffill().nullFill(0)
	mean = mavg(prev(imbalance), (lag-1), 2)
	std = mstdp(prev(imbalance) * 1000000, (lag-1), 2) \ 1000000
	return iif(std >= 0.0000001,(imbalance - mean) \ std, NULL).ffill().nullFill(0)
}

bidQty, askQty 为数组向量数据类型，分别为买方十档委托量和卖方十档委托数量。使用数组向量进行加减运算非常便捷，而使用rowWavg函数则可轻松计算加权平均值。在本例中，我们使用 rowWavg 函数计算各档加权平均的买卖委托量不均衡程度因子，即订单失衡率因子。最后对一段时间的指标进行移动标准化处理。

净委买比率是衡量一段时间内买盘相对强度的指标，计算公式为：

(盘口买一变化量－盘口卖一变化量)/ (abs（盘口买一变化量)+abs(盘口卖一变化量))×100%

当净委买比率为正值数值较大时，说明市场买盘强劲。当委比值为负值且数值较大时，说明市场卖盘强劲。数值从-100%逐步增加到+100%表示买盘逐渐增强，卖盘逐渐减弱；而从+100%逐步减少到-100%则表示买盘逐渐减弱，卖盘逐渐增强的过程。其计算公式如下：

$chg(bidQty_{i})$ 和 $chg(askQty_{i})$ 分别表示在 i 时刻盘口买一和卖一变化量，而 $avgPrice_{i}$ 表示在 i-1 时刻到 i 时刻成交的平均价格。

  @state
def traPriceWeightedNetBuyQuoteVolumeRatio(bid,bidQty,ask,askQty,TotalValTrd,TotalVolTrd,lag=20){
	prevbid = prev(bid)
	prevbidQty = prev(bidQty)
	prevask = prev(ask)
	prevaskQty = prev(askQty)
	bidchg = iif(round(bid-prevbid,2)>0, bidQty, iif(round(bid-prevbid,2)<0, -prevbidQty, bidQty-prevbidQty))
	offerchg = iif(iif(ask==0,iif(prevask>0,1,0), ask-prevask)>0, prevaskQty, iif(iif(prevask==0,
		iif(ask>0,-1,0), iif(ask>0,ask-prevask,1))<0, askQty, askQty-prevaskQty))
	avgprice = deltas(TotalValTrd)\deltas(TotalVolTrd)
	factorValue = (bidchg-offerchg)\(abs(bidchg)+abs(offerchg))*avgprice
	return nullFill(msum(factorValue,lag,1)\msum(avgprice,lag,1), 0)
}

以上代码，首先通过计算买卖一档前后价格的变化，得到盘口买卖一委托量的绝对变化量；然后利用净委买比率公式，计算过去一段时间的成交价加权净委买比例。

十档净委买增额因子指的是在有效十档范围内买方资金总体增加量，即所有买价变化量的总和，计算公式如下：

有效十档范围内表示不考虑已不在十档范围内的档位，即表示只考虑以下区间的档位：

@state
def level10_Diff(price, qty, buy, lag=20){
        prevPrice = price.prev()
        left, right = rowAlign(price, prevPrice, how=iif(buy, "bid", "ask"))
        qtyDiff = (qty.rowAt(left).nullFill(0) - qty.prev().rowAt(right).nullFill(0)) 
        amtDiff = rowSum(nullFill(price.rowAt(left), prevPrice.rowAt(right)) * qtyDiff)
        return msum(amtDiff, lag, 1).nullFill(0)
}

以上代码，首先通过行对齐函数rowAlign实现当前十档价格和前一个一十档价格进行行对齐，然后通过rowAt和nullFill函数分别获取对应档位的委托量和实现价格进行对齐，最后计算总的变化额。

十档买卖委托均价即为十档买卖委托额之和除以十档买卖委托量之和：

十档买卖委托均价线性回归斜率为十档买卖委托均价对时间t的线性回归的斜率。

@state
def level10_InferPriceTrend(bid, ask, bidQty, askQty, lag1=60, lag2=20){
	inferPrice = (rowSum(bid*bidQty)+rowSum(ask*askQty))\(rowSum(bidQty)+rowSum(askQty))
	price = iif(bid[0] <=0 or ask[0]<=0, NULL, inferPrice)
	return price.ffill().linearTimeTrend(lag1).at(1).nullFill(0).mavg(lag2, 1).nullFill(0)
}

以上代码，bid, ask, bidQty 和 askQty 均为数组向量数据类型，分别为买卖十档价格和十档委托数量。通过linearTimeTrend函数获取因子值对时间 t 的滑动线性回归斜率，该函数返回线性回归的截距和斜率。linearTimeTrend(price_,lag1)[1] 表示获取十档买卖委托均价对时间t的线性回归的斜率。

A 股 Level-2 行情快照数据一天的数据量超过 10G，因此金融量化工程师们非常关注 Level-2 行情快照数据的高频因子计算性能。本节将展示 5 个受到关注的高频因子的计算性能，包括：加权时间订单斜率、加权平均订单失衡率、成交价加权净委买比率、十档委买增额和十档买卖委托均价线性回归斜率。

timer {
	res=select SecurityID,DateTime,timeWeightedOrderSlope(bidPrice[0],bidOrderQty[0],OfferPrice[0],OfferOrderQty[0])  as TimeWeightedOrderSlope,
	level10_InferPriceTrend(bidPrice,OfferPrice,bidOrderQty,OfferOrderQty,60,20) as Level10_InferPriceTrend,
	level10_Diff(bidPrice, bidOrderQty, true,20) as Level10_Diff,
	traPriceWeightedNetBuyQuoteVolumeRatio(bidPrice[0],bidOrderQty[0],OfferPrice[0],OfferOrderQty[0],TotalValueTrade,
	totalVolumeTrade) as TraPriceWeightedNetBuyQuoteVolumeRatio,
	wavgSOIR( bidOrderQty,OfferOrderQty,20) as HeightImbalance
	from  loadTable(dbName,snapshotTBname) where date(DateTime)=idate context by SecurityID csort DateTime map 
}

• 计算结果展示

  
  

• 计算耗时展示

  以下为基于快照数据计算一天全市场所有股票的 5 个因子的耗时，可以发现耗时基本与 CPU 数量成正比。

  CPU 数量	5个因子计算耗时
  4	35.51s
  8	19.27s
  16	11.91s
  32	8.11s

• 与 Python 计算性能的对比

  在 Python 中实现上面5个因子的计算逻辑。获取一天中两只股票的数据进行对比测试。

  下表为 DolphinDB 和 Python 计算性能的对比，可以发现，DolphinDB 显著优于 Python。代码见附件。

  因子名	Python 耗时	DolphinDB 耗时	耗时比
  timeWeightedOrderSlope	136.726ms	7.745ms	17.6
  wavgSOIR	432.329ms	11.805ms	36.6
  traPriceWeightedNetBuyQuoteVolumeRatio	144.154ms	8.339ms	17.3
  level10_Diff	34.120026s	13.025ms	2619.2
  level10_InferPriceTrend	1.267870s	20.324ms	73.7

单笔订单主动买入、卖出均价为单笔主买、主卖订单的成交均价价格之和除以订单数量。

$n$表示截至 t 时刻主买、主卖订单数量，$avgTradePrice$ 表示单笔订单主买、主卖的成交均价。

def singleOrderAveragePrice(buyNo,sellNo,tradePrice,tradeQty,BSFlag="B"){
	if(BSFlag=="B"){
		 totolMoney=groupby(sum,iif(buyNo>sellNo,tradePrice*tradeQty,0),buyNo).values()[1]
		 totolqty=groupby(sum,iif(buyNo>sellNo,tradeQty,0),buyNo).values()[1]
	}
	else{
		 totolMoney=groupby(sum,iif(buyNo<sellNo,tradePrice*tradeQty,0),sellNo).values()[1]
		 totolqty=groupby(sum,iif(buyNo<sellNo,tradeQty,0),sellNo).values()[1]
		}
	 return totolMoney\totolqty
}
res=select avg(singleOrderAveragePrice(BidApplSeqNum,OfferApplSeqNum,TradePrice,TradeQty,"B")) as ActBuySingleOrderAveragePriceFactor,
avg(singleOrderAveragePrice(BidApplSeqNum,OfferApplSeqNum,TradePrice,TradeQty,"S")) as ActSellSingleOrderAveragePriceFactor from 
tradeTB where  TradePrice>0 group by SecurityID cgroup by minute(DateTime) as minute order by  minute;

这里首先通过自定义函数实现单笔订单的成交均价，然后再计算单笔订单的成交均价。自定义函数中使用groupby函数分别计算单笔订单的成交金额和成交数量。

最后通过group by和cgroup by计算每只股票当日最新一分钟单笔订单主动买入、卖出均价。计算结果如下：

延时成交挂单数和成交量，一定程度上能反应大单或者机构成交情况。本节统计从下单到成交的时间间隔超过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()
}
///买方
///step 1,标记延时状态，计算笔订单的累计成交量
t1 = select SecurityID,DateTime,entrust.DateTime as time,"B" as bsFlag,tradeQty, cumsum(iif(DateTime-entrust.DateTime>60000,1,0)) as delayedTraderflag, cumsum(tradeQty) as cumTradeQty from lsj(trade, entrust, ['SecurityID', 'BidApplSeqNum'], ['SecurityID', 'ApplSeqNum']) where tradePrice>0 context by  SecurityID,BidApplSeqNum 
/////step 2,统计每只股票累计的延时成交订单数，和延时成交订单量
tt2 = select SecurityID,DateTime,delayedTradeNum(bsFlag, delayedTraderflag, "B") as DelayedTraderBuyOrderNum ,
 delayedTradeQty(bsFlag,delayedTraderflag, tradeQty, cumTradeQty, "B") as DelayedTraderBuyOrderQty from t1 context by SecurityID limit -1
 ///卖方
///step 1,标记延时状态，计算笔订单的累计成交量
t1 = select SecurityID,DateTime,"S" as bsFlag,tradeQty, cumsum(iif(DateTime-entrust.DateTime>60000,1,0)) as delayedTraderflag, cumsum(tradeQty) as cumTradeQty from lsj(trade, entrust, ['SecurityID', 'OfferApplSeqNum'], ['SecurityID', 'ApplSeqNum']) where tradePrice>0 context by SecurityID, OfferApplSeqNum
/////step 2,统计每只股票累计的延时成交订单数，和延时成交订单量
t3 = select SecurityID,DateTime,delayedTradeNum(bsFlag, delayedTraderflag, "S") as DelayedTraderSellOrderNum ,
 delayedTradeQty(bsFlag,delayedTraderflag , tradeQty, cumTradeQty, "S") as DelayedTraderSellOrderQty from t1 context  by SecurityID limit -1
 t2=lsj(t2,t3,`SecurityID`DateTime)

下单信息记录在逐笔委托表里，如果统计下单到成交之间的时间间隔，则需要把逐笔成交表和逐笔委托表进行关联。这里首先通过左半连接(lsj) 返回逐笔成交表中所有与逐笔委托表匹配的记录，如果逐笔委托表中有多条匹配记录（如上交所的下单和撤单记录），lsj 将会取第一条（下单时的订单记录）匹配记录。因此，lsj 可以把订单委托下单的时间以及下单量准确关联到成交记录中。

计算股票延时成交订单因子的步骤为，首先根据成交表的买卖单号与委托表的订单委托号建立连接，并计算改订单的累计延时成交次数和订单的累计成交量；其次通过自定义函数，计算股票的延时成交订单数以及延时成交的订单量。

委托买卖金额是对一段时间内委托订单的买卖方向的资金总量进行统计。由于深交所的市价订单的价格在逐笔委托表里的价格不是一个成交价格（一般标记为0）。因此统计计算委托金额时，我们需要在逐笔成交记录中找到最近的成交价格来作为其近似值。

defg calcSZOrderValue(side,price,orderQty,tradePrice,orderType,bsFlag){
	price_=iif(orderType =="50",price,NULL).nullFill(tradePrice)
	return sum(iif(side==bsFlag,price_*orderQty,NULL)).nullFill(0)
}
res=select calcSZOrderValue(side,price,orderQty,tradePrice,orderType,"B") as BuyOrderValue, 
calcSZOrderValue(side,price,orderQty,tradePrice,orderType,"S") as SellOrderValue
from aj(entrustTB,tradeTB,`SecurityID`ApplSeqNum) group by SecurityID,bar(DateTime,1m) as DateTime

这里通过aj （asof join）把逐笔成交里的最新价格关联到逐笔成交中；对市价订单，以获取最新的成交价格作为当前市价委托单的委托价格，最后计算股票每分钟内的买卖委托金额。

深交所的撤单标记记录在逐笔成交表中，撤单时的价格一般标记为0，而委托价格记录在逐笔委托表中，因此计算买卖方撤单金额时需要将逐笔成交表和逐笔委托表进行关联。

trade = select SecurityID,DateTime, max(BidApplSeqNum,OfferApplSeqNum) as ApplSeqNum,TradeQty from tradeTB where  ExecType=52
entrust = select  SecurityID,ApplSeqNum, Price,Side from entrustTB
res = select  sum(iif(side=="B",Price*TradeQty,NULL)).nullFill(0) as buywithdrawOrderValue,
sum(iif(side=="S",Price*TradeQty,NULL)).nullFill(0) as sellwithdrawOrderValue 
from lsj(trade,entrust,`SecurityID`ApplSeqNum) group by SecurityID,bar(DateTime,1m) as DateTime

这里通过 lj（left join)分别把买卖撤单的委托价格关联到撤单信息表中，然后计算每只股票每分钟的买卖撤单金额。

整体计算流程如下图所示：

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

流程说明：

• 左半等值连接引擎，把逐笔成交与逐笔委托数据关联，并计算延时成交订单的累计值和订单累计成交量；
• 用响应式状态引擎，计算每只股票延时订单因子和延时订单成交量；
• 用时间序列引擎 获取最新一分钟的延时订单因子和延时订单成交量，并输出结果。

各引擎间直接级联，无需通过中间表。

• 第一步：

  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);