createOrderBookSnapshotEngine

createOrderBookSnapshotEngine(name, exchange, orderbookDepth, intervalInMilli, date, startTime, prevClose, dummyTable, outputTable, inputColMap, [outputColMap], [outputCodeMap], [snapshotDir], [snapshotIntervalInMsgCount], [raftGroup], [outputIntervalOffsetMap], [checkRestrict], [maxPrice], [minPrice], [userDefinedMetrics], [priceNullFill], [triggerType], [forceTriggerTime], [precision], [orderBySeq], [skipCrossedMarket=true], [orderBookDetailDepth=0], [orderBookAsArray=false], [useSystemTime=false],[independentForceTriggerTime])

该函数基于一个包含了逐笔委托和逐笔交易数据表，实时合成指定频率（1秒，100ms，30ms 等）的全档位盘口信息、窗口内统计信息、全天累计统计信息等订单薄（或者叫快照）数据。也可通过历史逐笔数据来合成需要的订单薄数据。

目前支持合成深交所股票、深交所可转债、深交所基金、上交所股票、上交所可转债、上交所基金的订单薄数据。请注意，可转债的规则遵循自2022年8月之后发布的新规定。

name 字符串标量，表示 orderbook 快照引擎的名称，可包含字母，数字和下划线，但必须以字母开头。

exchange 字符串标量，表示证券的类型。可选参数：

• “XSHE” 或 “XSHESTOCK”：深交所股票
• “XSHEBOND”：深交所可转债
• “XSHEFUND”：深交所基金
• “XSHG” 或 “XSHGSTOCK”：上交所股票
• “XSHGBOND”：上交所可转债
• “XSHGFUND”：上交所基金

orderbookDepth 正整数，表示 orderbook 中最多能够显示的买卖报价的档位。

intervalInMilli 正整数，表示触发输出数据的时间间隔，即合成快照的时间频率，单位为毫秒。

date DATE 类型标量，表示交易日期。该参数和窗口右边界组合成输出表的 TIMESTAMP 列。

startTime TIME 类型标量，表示触发输出数据的起始时刻。引擎只会输出该参数指定时刻（规整后）之后的数据。

prevClose 字典，其 key 为字符串标量或向量，表示股票代码；其 value 为数值类型，表示对应于股票代码的上一个交易日的收盘价格。

dummyTable 一个表对象，和订阅的流数据表的 schema 一致，可以含有数据，亦可为空表。

outputTable 一个表对象。若指定 userDefinedMetrics，outputTable 根据 genOutputColumnsForOBSnapshotEngine 函数指定的 basic、depth 以及用户自定义的指标来确定表结构。否则，outputTable 根据 outputColMap的设置来确定表结构。

inputColMap 字典，将输入表中列字段的名称映射为引擎计算所需要的列。其中：

• key 为字符串类型，表示引擎所需要的固定的输入字段。具体字段名和含义见下表。注意，这些 key 区分大小写，必须全部指定，但顺序可以不固定。
• value 为字符串类型，表示输入表中对应的列名称。

outputColMap 可选参数，字符串向量，用于指定需要输出的字段名称，不区分大小写。为方便指定 outputColMap，用户可通过 genOutputColumnsForOBSnapshotEngine 函数生成需要输出的字段名称，将返回值的第一个元素赋值给 outputColMap 即可。

outputCodeMap 可选参数，字符串向量，表示股票代码，例如：“000803.SZ”。指定该参数后，将只输出指定股票对应的数据。

outputIntervalOffsetMap 可选参数，向量或字典，用于指定输出表中股票被触发计算的时间偏移量。

• 当为向量时，向量中的元素表示不同的时间偏移量，单位为毫秒。引擎将根据输入的股票数量自动将这些股票均匀分配到这些偏移量上。例如，outputIntervalOffsetMap = [400, 500]，表示引擎将自动将输入的股票均匀分成两部分，经过intervalInMilli + 400(ms) 后触发其中一部分股票的数据输出；经过 intervalInMilli + 500(ms) 后触发另一部分股票的数据输出。
• 当为字典时，其 key 为字符串类型，表示股票代码；其 value 为整型，表示时间偏移量，单位为毫秒。例如：outputIntervalOffsetMap =dict(["127053.sz","123082.SZ"],[400, 500])，表示经过 intervalInMilli + 400(ms) 后触发 127053.sz 股票的数据输出；经过 intervalInMilli + 500(ms) 后触发 123082.SZ 股票的数据输出。

checkRestrict 可选参数，布尔值，默认值为 true，表示开启价格笼子机制；若设置为 false，则关闭笼子机制，此时会取消对股票交易的限制，创业板合成出的快照可能不正确。

maxPrice 可选参数，字典。其 key 为字符串类型，表示证券代码；其 value 为 DOUBLE 类型，表示涨停价格。

minPrice 可选参数，字典。其 key 为字符串类型，表示证券代码；其 value 为 DOUBLE 类型，表示跌停价格。

userDefinedMetrics 可选参数，一元函数。

• 函数的入参是一个表，表的每一行是一个标的，每一列是标的对应的快照。若指定 outputColMap，则表中数据为 outputColMap设置的字段；否则表中数据为基础字段和报价/询价的价格和数量。

• 函数的返回值是一个元组，元组的每一个元素是一个常规向量，代表每个标的经对应指标计算后的结果。

priceNullFill 一个数字。该参数值用于填充输出表中多档买/卖价格中缺失档位的价格。例如涨停后卖单的价格都为 NULL，此时如有指定卖单价格为 0 的需求，可以设置 priceNullFill=0。

triggerType 可选参数，字符串标量，表示触发方式。可选值为：

• “mutual”：新收到任意一条逐笔数据，当其时间戳大于窗口右边界时，就会触发所有股票未计算的快照合成输出。触发计算的数据并不会参与该次计算。
• “independent”：新收到任意一条逐笔数据，当其时间戳大于窗口右边界时，只触发该条数据对应股票未计算的快照合成输出。触发计算的数据并不会参与该次计算。
• “perRow”：每收到一行逐笔数据都会触发该条数据对应股票计算并输出，触发计算的数据并参与该次计算。

forceTriggerTime 可选参数，是非负整数，单位为毫秒。除了正常触发合成快照外，有时会出现一些数据因乱序不能触发合成快照，而是被缓存在引擎中。在这种情况下，可通过该参数设置强制触发引擎中长时间未处理的逐笔数据合成快照。触发规则为：

1. 最新收到的逐笔数据的时间戳（t ）减去最后一条已经处理的交易数据的时间戳（t0 ）大于或等于 forceTriggerTime，则触发未处理的数据中序号最小的那条数据合成快照，并更新已经处理的交易数据的时间戳（t1）。
2. 重复上述操作，若判断 t-t1 >= forceTriggerTime，则触发未处理的数据中序号最小的那条数据合成快照，并更新已经处理的交易数据的时间戳（t2）；直到两个时间戳的差值小于 forceTriggerTime 时，停止触发快照合成。

precision 可选参数，非负整数，表示小数位数，取值范围为[0,4]。若指定该参数，则输出表中所有的价格将按照指定的位数进行四舍五入。否则，按照原始结果输出。

orderBySeq 可选参数，布尔值或元组

• 指定该参数值为布尔值时，表示是否按照逐笔数据中的 seqColumn 列中值的大小顺序处理数据。当 exchange="XSHG" 或 "XSHGFUND" 时，默认值为 true；在其他情况下，默认值为 false。
  • 当 orderBySeq=true 时，会根据逐笔数据的序号有序处理数据，并计算输出。例如引擎先后收到序号为1，3的数据，因为缺少序号为2的数据，引擎先将1和3的数据缓存，待收到序号2的数据时，再进行计算输出。此时也可以指定 forceTriggerTime 强制触发计算输出。
  • 当 orderBySeq=false 时，每接收到一条数据都会立即进行计算并输出，此时不可设置 forceTriggerTime。
• 指定该参数值为元组时，以（BOOL, INTEGER, [STRING]）的形式表示，其中：
  • 第一个元素为布尔值，表示是否按照逐笔数据的序号有序输出结果的布尔值，作用同上。
  • 第二个元素为正整数，表示记录缓存输入数据量的时间间隔，单位为毫秒（ms）。
  • 第三个元素为指定输出日志级别的字符串标量，可选参数，其可用值为 DEBUG （默认值）和 INFO，分别对应 Debug 和 Info 级别的日志输出。

skipCrossedMarket可选参数， 布尔值，表示是否输出买一卖一价交叉的计算结果。

• 当 skipCrossedMarket=true（默认值）时，若设置 useSystemTime = false，若出现买一卖一价交叉，即卖一价<=买一价，则不输出该条结果；若设置 useSystemTime = true，则出现买一卖一交叉时，暂时不输出该条快照，若接下来收到的数据不再有交叉，则立即触发该条快照输出。
• 当 skipCrossedMarket=false 时，若出现买一卖一价交叉，仍然输出该条结果。

orderBookDetailDepth 可选参数，整型标量，表示 orderbook 明细的深度。默认值为0，表示不输出。该参数必须和 outputColMap 中 orderBookDetailDepth 字段的值保持一致。

orderBookAsArray 可选参数，布尔值，表示是否以数组向量形式输出报价/询价的价格和数量。默认值为 false，价格和数量将以多列形式输出。

注意，若指定 userDefinedMetrics，价格和数量的输出形式由 orderBookAsArray 确定。否则，价格和数量的输出形式由 outputColMap 确定，该参数失效。

independentForceTriggerTime 可选参数，非负整数，单位为毫秒。该参数用于设置在长时间未触发快照输出的分组中，强制输出快照的时间间隔。仅当设置 triggerType=“independent” 时，该参数才会生效。

举例说明：假设在当前已处理的数据中，最新一条数据的时间为 t。各个分组内已处理的应被触发输出的快照时间为 ti。如果满足 t-ti>independentForceTriggerTime，则会触发输出相应分组的快照。

运行代码前，先下载 ../data/orderbookDemoInput.zip 文件。

// 登录
login("admin", "123456")

// 释放已有的引擎
try { dropStreamEngine("demo") } catch(ex) { print(ex) }

// 创建引擎参数 outputTable，即指定输出表
suffix = string(1..10)
colNames = `SecurityID`timestamp`lastAppSeqNum`tradingPhaseCode`modified`turnover`volume`tradeNum`totalTurnover`totalVolume`totalTradeNum`lastPx`highPx`lowPx`ask`bid`askVol`bidVol`preClosePx`invalid  join ("bids" + suffix) join ("bidVolumes" + suffix) join ("bidOrderNums" + suffix) join ("asks" + suffix)  join ("askVolumes" + suffix) join ("askOrderNums" + suffix) 
colTypes = [SYMBOL,TIMESTAMP,LONG,INT,BOOL,DOUBLE,LONG,INT,DOUBLE,LONG,INT,DOUBLE,DOUBLE,DOUBLE,DOUBLE,DOUBLE,LONG,LONG,DOUBLE,BOOL] join take(DOUBLE, 10) join take(LONG, 10) join take(INT, 10) join take(DOUBLE, 10) join take(LONG, 10) join take(INT, 10) 
share table(10000000:0, colNames, colTypes) as outTable

// 创建引擎参数 dummyTable，即指定输入表的表结构
colNames = `SecurityID`Date`Time`SecurityIDSource`SecurityType`Index`SourceType`Type`Price`Qty`BSFlag`BuyNo`SellNo`ApplSeqNum`ChannelNo
colTypes = [SYMBOL, DATE, TIME, SYMBOL, SYMBOL, LONG, INT, INT, LONG, LONG, INT, LONG, LONG, LONG, INT]
dummyOrderStream = table(1:0, colNames, colTypes)

// 创建引擎参数 inputColMap，即指定输入表各字段的含义
inputColMap = dict(`codeColumn`timeColumn`typeColumn`priceColumn`qtyColumn`buyOrderColumn`sellOrderColumn`sideColumn`msgTypeColumn`seqColumn, `SecurityID`Time`Type`Price`Qty`BuyNo`SellNo`BSFlag`SourceType`ApplSeqNum)

// 创建引擎参数 prevClose，即昨日收盘价，prevClose 不影响最终的输出结果中除昨日收盘价以外的其他字段
prevClose = dict(`000587.SZ`002694.SZ`002822.SZ`000683.SZ`301063.SZ`300459.SZ`300057.SZ`300593.SZ`301035.SZ`300765.SZ, [1.66, 6.56, 6.10, 8.47, 38.10, 5.34, 9.14, 48.81, 60.04, 16.52])

// 定义引擎，每1s计算输出深交所股票10档买卖盘口
engine = createOrderBookSnapshotEngine(name="demo", exchange="XSHE", orderbookDepth=10, intervalInMilli = 1000, date=2022.01.10, startTime=09:15:00.000, prevClose=prevClose, dummyTable=dummyOrderStream, outputTable=outTable, inputColMap=inputColMap)

filePath = "./orderbookDemoInput.csv"
colNames = `SecurityID`Date`Time`SecurityIDSource`SecurityType`Index`SourceType`Type`Price`Qty`BSFlag`BuyNo`SellNo`ApplSeqNum`ChannelNo
colTypes = [SYMBOL, DATE, TIME, SYMBOL, SYMBOL, LONG, INT, INT, LONG, LONG, INT, LONG, LONG, LONG, INT]
orderTrade = table(1:0, colNames, colTypes)
orderTrade.append!(select * from loadText(filePath) order by Time)

// 10支股票的逐笔数据批量注入快照合成引擎
engine.append!(orderTrade)
select count(*) from outTable where SecurityID="300593.SZ", timestamp between 2022.01.10T13:15:01.000 and 2022.01.10T13:15:10.000

//output: 10

try { dropStreamEngine("demo") } catch(ex) { print(ex) }

filePath = "./orderbookDemoInput.csv"

// 创建引擎参数 dummyTable，即指定输入表的表结构
colNames = `SecurityID`Date`Time`SecurityIDSource`SecurityType`Index`SourceType`Type`Price`Qty`BSFlag`BuyNo`SellNo`ApplSeqNum`ChannelNo
colTypes = [SYMBOL, DATE, TIME, SYMBOL, SYMBOL, LONG, INT, INT, LONG, LONG, INT, LONG, LONG, LONG, INT]
share table(1:0, colNames, colTypes) as dummyOrderStream

// 创建引擎参数 inputColMap，即指定输入表各字段的含义
inputColMap = dict(`codeColumn`timeColumn`typeColumn`priceColumn`qtyColumn`buyOrderColumn`sellOrderColumn`sideColumn`msgTypeColumn`seqColumn, `SecurityID`Time`Type`Price`Qty`BuyNo`SellNo`BSFlag`SourceType`ApplSeqNum)

// 创建引擎参数 prevClose，即昨日收盘价，prevClose 不影响最终的输出结果中除昨日收盘价以外的其他字段
prevClose = dict(`000587.SZ`002694.SZ`002822.SZ`000683.SZ`301063.SZ`300459.SZ`300057.SZ`300593.SZ`301035.SZ`300765.SZ, [1.66, 6.56, 6.10, 8.47, 38.10, 5.34, 9.14, 48.81, 60.04, 16.52])

//创建使用 outputColMap 和 outputTableSch 接收 genOutputColumnsForOBSnapshotEngine 的返回值。它们分别用于确定 outputColMap 和 outputTable
outputColMap, outputTableSch = genOutputColumnsForOBSnapshotEngine(basic=true, time=false, depth=(10, true), tradeDetail=true, orderDetail=false, withdrawDetail=false, orderBookDetailDepth=0, prevDetail=false)

engine = createOrderBookSnapshotEngine(name="demo", exchange="XSHE", orderbookDepth=10, intervalInMilli = 1000, date=2022.01.10, startTime=09:15:00.000, prevClose=prevClose, dummyTable=dummyOrderStream, outputTable=outputTableSch, inputColMap=inputColMap, outputColMap=outputColMap, orderBookAsArray=true)

// 10支股票的逐笔数据批量注入快照合成引擎
engine.append!(select * from loadText(filePath) order by Time)
select top 10 * from outputTableSch where code="300593.SZ", timestamp between 2022.01.10T13:15:01.000 and 2022.01.10T13:15:10.000

部分结果展示如下：

try { dropStreamEngine("demo") } catch(ex) { print(ex) }

filePath = "./orderbookDemoInput.csv"

// 创建引擎参数 dummyTable，即指定输入表的表结构
colNames = `SecurityID`Date`Time`SecurityIDSource`SecurityType`Index`SourceType`Type`Price`Qty`BSFlag`BuyNo`SellNo`ApplSeqNum`ChannelNo`ReceiveTime
colTypes = [SYMBOL, DATE, TIME, SYMBOL, SYMBOL, LONG, INT, INT, LONG, LONG, INT, LONG, LONG, LONG, INT, NANOTIMESTAMP]
share table(1:0, colNames, colTypes) as dummyOrderStream

// 创建引擎参数 inputColMap，即指定输入表各字段的含义
inputColMap = dict(`codeColumn`timeColumn`typeColumn`priceColumn`qtyColumn`buyOrderColumn`sellOrderColumn`sideColumn`msgTypeColumn`seqColumn`receiveTime, `SecurityID`Time`Type`Price`Qty`BuyNo`SellNo`BSFlag`SourceType`ApplSeqNum`ReceiveTime)

// 创建引擎参数 prevClose，即昨日收盘价，prevClose 不影响最终的输出结果中除昨日收盘价以外的其他字段
prevClose = dict(`000587.SZ`002694.SZ`002822.SZ`000683.SZ`301063.SZ`300459.SZ`300057.SZ`300593.SZ`301035.SZ`300765.SZ, [1.66, 6.56, 6.10, 8.47, 38.10, 5.34, 9.14, 48.81, 60.04, 16.52])

//此处只定义 outputColMap 接收 genOutputColumnsForOBSnapshotEngine 的第一个返回值。因 userDefinedMetrics 定义的指标中需要使用委托明细和回撤明细，因此 genOutputColumnsForOBSnapshotEngine  需要指定 orderDetail=false, withdrawDetail=true
outputColMap = genOutputColumnsForOBSnapshotEngine(basic=true, time=false, depth=(10, true), tradeDetail=true, orderDetail=false, withdrawDetail=true, orderBookDetailDepth=0, prevDetail=false)[0]


//// 定义用户自定义因子
def userDefinedFunc(t){
        AvgBuyDuration = rowAvg(t.TradeMDTimeList-t.TradeOrderBuyNoTimeList).int()
        AvgSellDuration = rowAvg(t.TradeMDTimeList-t.TradeOrderSellNoTimeList).int()        
        BuyWithdrawQty = rowSum(t.WithdrawBuyQtyList)
        SellWithdrawQty = rowSum(t.WithdrawSellQtyList)
        return (AvgBuyDuration, AvgSellDuration, BuyWithdrawQty, SellWithdrawQty)
}

// 定义 orderbook 引擎的输出表，需要包含基础字段（basic），报价/询价的多档价格和数量（depth），userDefinedMetrics 自定义指标的输出结果
outputTableSch = genOutputColumnsForOBSnapshotEngine(basic=true, time=false, depth=(10, true), tradeDetail=false, orderDetail=false, withdrawDetail=false, orderBookDetailDepth=0, prevDetail=false)[1]
colNames = outputTableSch.schema().colDefs.name join (`AvgBuyDuration`AvgSellDuration`BuyWithdrawQty`SellWithdrawQty)
colTypes = outputTableSch.schema().colDefs.typeString join (`INT`INT`INT`INT) 
outputTable = table(1:0, colNames, colTypes)

// 创建引擎，每1s计算输出深交所股票10档 orderbook 

engine = createOrderBookSnapshotEngine(name="demo", exchange="XSHE", orderbookDepth=10, intervalInMilli = 1000, date=2022.01.10, startTime=09:30:00.000, prevClose=prevClose, dummyTable=dummyOrderStream, outputTable=outputTable, inputColMap=inputColMap, outputColMap=outputColMap, orderBookAsArray=true, userDefinedMetrics=userDefinedFunc)

t = select * from loadText(filePath) order by Time
update t set ReceiveTime = now(true) // 构造接收时间列

getStreamEngine("demo").append!(t)

select top 10 * from outputTable where code="300593.SZ", timestamp between 2022.01.10T13:15:01.000 and 2022.01.10T13:15:10.000

部分结果展示如下：

相关函数：genOutputColumnsForOBSnapshotEngine

相关教程：基于逐笔数据合成高频 Orderbook：DolphinDB Orderbook 引擎

basic（基础字段）

不指定 outputColMap 和 userDefinedMetrics 时，输出除 openPrice，maxPrice，minPrice 外的字段。

time（时间字段）

depth（报价/询价档位）

不指定 outputColMap 和 userDefinedMetrics 时，会输出 depth 中的所有字段。提供两种方式存储 depth 数据。两种方式只能选其一：

• 每个档位输出一列。比如 bidsPrice 有10个档位，则会输出10个 bidsPrice 字段：bidsPrice1、bidsPrice2、…、bidsPrice10。

• 多个档位输出到一个列字段，该列的类型为数组向量。比如 bidsPrice 有10个档位，则会将这10个档位以数组向量的形式输出到一个字段：bidsPriceList。

tradeDetail（成交明细）

orderDetail （委托明细）

withdrawDetail（撤单明细）

prevDetail（上一笔交易明细）

orderBookDetailDepth（ orderbook 明细档位）

每个档位输出一列，该列的类型是数组向量。比如 buyQtyList 有20个档位，则会输出20个 buyQtyList 字段：buyQtyList1、buyQtyList2、…、buyQtyList20。buyValueList，sellQtyList 和 sellValueList 同理。

seqDetail（sequence 明细）

residualDetail（剩余委托明细）

注意：

• 卖方字段（ResidualAskPriceList, ResidualAskQtyList, ResidualAskTimeList, ResidualAskApplSeqNumList）中每个档位的信息按照价格和交易时间升序排序。若价格和交易时间相同，则按数据进入引擎的时间升序排序。

• 买方字段（ResidualBidPriceList, ResidualBidQtyList, ResidualBidTimeList, ResidualBidApplSeqNumList）中每个档位的信息按价格降序和交易时间升序排序，若价格和交易时间相同，则按数据进入引擎的时间升序排序。