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],[includeImmediateExecution=false], [securitySubType='ConvertibleBond'], [priceScale=10000], [endTime])

Creates an order book engine. This engine generates order book snapshots in real time at a specified frequency, such as 1 second, 100 ms, or 30 ms, based on tick-by-tick order and trade data.

The output includes:

• Full-depth order book information
• Window-based statistics
• Cumulative statistics for the full trading day

This engine supports both real-time data streaming and historical tick-by-tick replay.

If the engine produces no output or very little output after you feed in data, see section 7.5 of the order book engine tutorial to troubleshoot the issue.

name: A STRING scalar that specifies the name of the order book engine. It can contain letters, digits, and underscores (_), but must start with a letter.

exchange: A STRING scalar that specifies the exchange and instrument type. Valid values:

• "XSHE" or "XSHESTOCK": SZSE stocks
• "XSHEBOND": SZSE bonds
• "XSHEFUND": SZSE funds
• "XSHG" or "XSHGSTOCK": SSE stocks
• "XSHGBOND": SSE bonds
• "XSHGFUND": SSE funds

orderbookDepth: A positive integer that specifies the maximum number of bid and ask levels displayed in the order book.

intervalInMilli: A positive integer that specifies the interval (in milliseconds) at which a snapshot is generated and output.

date: A DATE scalar that specifies the trading date. This parameter, together with the window's right boundary, forms the TIMESTAMP column in the output table.

startTime: A TIME scalar that specifies the start time for triggering output. The engine outputs only data after the aligned time specified by this parameter.

prevClose: A dictionary whose keys are STRING scalars or vectors that specify instrument symbols, and whose values are numeric values that specify the previous trading day's closing price for each instrument symbol. Note: If the value is a floating-point number, the engine treats it as the actual price and multiplies it by priceScale during processing to match the precision of priceColumn in the input table. If it is an integer, the engine assumes it already has the same precision as priceColumn and does not scale it further.

dummyTable: A table object whose schema matches that of the input stream table. It can contain data or be empty.

outputTable: A table object. If you specify userDefinedMetrics, the system defines the schema of the output table based on basic and depth specified by the genOutputColumnsForOBSnapshotEngine function, as well as the user-defined metrics. Otherwise, the system defines the table schema based on the outputColMap setting.

inputColMap: A dictionary that maps column names in the input table to the columns required by the engine for computation. Where:

• Each key is a string that identifies a fixed input column required by the engine. See the table below for the column names and their descriptions. Note that these keys are case-sensitive. You must specify all of them, but you can provide them in any order.
• Each value is a string that specifies a column name in the input table.

outputColMap (optional): A STRING vector that specifies the names of the columns to output. Column names are case-insensitive. For convenience, you can use the genOutputColumnsForOBSnapshotEngine function to generate the required output column names and assign the first element of the return value to outputColMap.

outputCodeMap (optional): A STRING vector that specifies instrument symbols, for example: "000803.SZ". If you specify this parameter, the engine outputs data only for the specified instruments.

outputIntervalOffsetMap (optional): A vector or dictionary that specifies the time offset, in milliseconds, for triggering calculation of each instrument in the output table.

• Calculation logic: The engine redefines the window boundaries for each instrument based on the offset. The first window automatically includes all tick-by-tick data from startTime to startTime + intervalInMilli + offset, ensuring that data generated during the offset period is included in the current snapshot calculation.
• Vector: The engine automatically splits the input instruments evenly based on the size of the vector. For example, outputIntervalOffsetMap = [400, 500] means the engine automatically splits the input instruments into two equal groups. It outputs one group's data after intervalInMilli + 400 (ms), and the other group's data after intervalInMilli + 500 (ms).
• Dictionary: Each key is a string that specifies an instrument symbol, and each value is an integer that specifies a time offset. For example, outputIntervalOffsetMap =dict(["127053.sz","123082.SZ"],[400, 500]) means the engine outputs data for 127053.sz after intervalInMilli + 400 (ms), and for 123082.SZ after intervalInMilli + 500 (ms).

checkRestrict (optional): A Boolean value. The default value is true, which enables the price band mechanism. If you set it to false, the mechanism is disabled, and instrument trading restrictions are removed; in that case, generated snapshots for ChiNext instruments may be inaccurate.

maxPrice (optional): A dictionary. Each key is a string that specifies the instrument symbol, and each value is of type DOUBLE or INT and specifies the limit up price. Note: If maxPrice is a floating-point number, the engine treats it as the actual price and multiplies it by priceScale during processing to match the precision of priceColumn in the input table. If it is an integer, the engine assumes it already has the same precision as priceColumn and does not scale it further.

minPrice (optional): A dictionary. Each key is a string that specifies the instrument symbol, and each value is of type DOUBLE or INT and specifies the limit down price. Note: If minPrice is a floating-point number, the engine treats it as the actual price and multiplies it by priceScale during processing to match the precision of priceColumn in the input table. If it is an integer, the engine assumes it already has the same precision as priceColumn and does not scale it further.

userDefinedMetrics (optional): An unary function.

• The function takes a table as input. Each row specifies an instrument, and the columns contain the snapshot data for that instrument. If you specify outputColMap, the table contains the columns configured in outputColMap. Otherwise, it contains the basic columns plus bid/ask prices and quantities.
• The function returns a tuple. Each element in the tuple is a vector containing the results of the corresponding metric calculated for each instrument.

priceNullFill: A numeric value. This parameter is used to fill missing price levels in the bid/ask ladders in the output table. For example, after an instrument reaches the limit up price, all ask prices may be NULL. If you want the ask price to be output as 0 in this case, set priceNullFill=0.

triggerType (optional): A STRING scalar that specifies the trigger mode. Valid values:

• "mutual" (default): When the engine receives a new tick whose timestamp is greater than the right boundary of the current window, it triggers snapshot generation for all instruments whose snapshot has not yet been generated. The record that triggers the calculation is not included.
• "independent": When the engine receives a new tick whose timestamp is greater than the right boundary of the window, it triggers snapshot generation only for the corresponding instrument whose snapshot has not yet been generated. The record that triggers the calculation is not included.
• "perRow": When the engine receives any new tick, it triggers snapshot generation for the corresponding instrument and includes the triggering record in the calculation.

forceTriggerTime (optional): A non-negative integer, in milliseconds. In addition to normal snapshot generation triggers, some out-of-order data may fail to trigger snapshot generation and instead be cached in the engine. In this case, you can use this parameter to force the engine to generate snapshots from tick-by-tick data that has remained unprocessed for an extended period. Trigger rules:

1. If the timestamp (t) of the most recently received tick-by-tick data minus the timestamp (t0) of the last processed trade record is greater than or equal to forceTriggerTime, the engine triggers snapshot generation for the unprocessed record with the smallest sequence number and updates the timestamp of the last processed trade record to t1.
2. The engine repeats the preceding step. If t-t1 >= forceTriggerTime, it triggers snapshot generation for the unprocessed record with the smallest sequence number and updates the timestamp of the last processed trade record to t2. It continues until the difference between the two timestamps is less than forceTriggerTime, at which point snapshot generation stops.

precision (optional): An integer that specifies the number of decimal places. Valid range: [-1, 4].

• When precision = -1, the engine outputs price-related columns as integers. Note: The corresponding columns in the output table can be of either integer or decimal values.
• When precision is set to a value in [0, 4], it specifies the number of decimal places. All prices in the output table are rounded to the specified number of decimal places. Otherwise, the engine outputs the original results.

orderBySeq (optional): A tuple or Boolean value.

• If you specify this parameter to a Boolean value, it indicates whether the engine processes data in ascending order of the values in the seqColumn column of the tick-by-tick data. When exchange = "XSHG" or "XSHGFUND", the default value is true. In all other cases, the default value is false.
  • When orderBySeq = true, the engine processes tick-by-tick data in sequence order and then calculates and outputs the results. For example, if the engine receives records with sequence numbers 1 and 3, it caches both because record 2 is missing. It does not calculate or output results until record 2 arrives. In this case, you can also specify forceTriggerTime to force the engine to calculate and output results.
  • When orderBySeq = false, the engine calculates and outputs results immediately each time it receives a record. In this case, you cannot set forceTriggerTime.
• If you specify this parameter to a tuple, use the form (BOOL, INTEGER, [STRING]), where:
  • The first element is a Boolean value that specifies whether to output results in sequence order for tick-by-tick data. Its effect is the same as described above.
  • The second element is a positive integer that specifies the time interval, in milliseconds, at which the amount of cached input data is recorded.
  • The third element is an optional STRING scalar that specifies the output log level. Valid values are "DEBUG" (default) and "INFO", which correspond to Debug-level and Info-level log output, respectively.

skipCrossedMarket (optional): A Boolean value that specifies whether to output results when the best ask and best bid cross.

• When skipCrossedMarket = true (default), if useSystemTime = false and the best ask and best bid cross, that is, the ask price <= the bid price, the engine does not output that result. If useSystemTime = true, the engine temporarily suppresses that snapshot when the best ask and best bid cross; if subsequently received data no longer shows a crossed market, the engine immediately outputs that snapshot.
• When skipCrossedMarket = false, the engine still outputs the result even if the best ask and best bid cross.

orderBookDetailDepth (optional): An integer that specifies the depth of the order book. The default value is 0, which means no output. This parameter must match the value of the orderBookDetailDepth column in outputColMap.

orderBookAsArray (optional): A Boolean value that specifies whether to output bid/ask prices and quantities as array vectors. The default value is false, in which case prices and quantities are output in multiple columns.

useSystemTime (optional): A Boolean value that specifies whether to use system time to trigger snapshot output.

• When useSystemTime = true, during trading hours, the engine uses the current system time to trigger snapshot output at the interval specified by intervalInMilli. In this case, the engine does not output data during the midday break ((11:30:00.000, 13:00:00.000]), and the timestamp of the first afternoon output window is 13:00:00.000+intervalInMilli. Note: If useSystemTime = true, you cannot specify forceTriggerTime. You can leave triggerType unspecified or set triggerType = "mutual".
• When useSystemTime = false (default), the engine triggers snapshot output based on event time.

independentForceTriggerTime (optional): A non-negative integer, in milliseconds, that specifies the interval for forcing snapshot output for groups that have not output a snapshot for an extended period. This parameter takes effect only when triggerType = "independent".

For example, assume that among the data processed so far, the timestamp of the most recent record is t. For each group, let ti denote the snapshot time that should have triggered an output based on the data processed for that group. If t - ti > independentForceTriggerTime, the engine triggers snapshot output for the corresponding group.

includeImmediateExecution (optional): A Boolean value that specifies whether immediate execution information is included in orderDetail statistics. The default value is false. This parameter takes effect only when generating SSE order books, and applies to stock and fund data (exchange set to "XSHG", "XSHGSTOCK", or "XSHGFUND").

securitySubType (optional): A case-insensitive STRING scalar that specifies the subtype of the instrument for which to generate the order book. It applies only when exchange is "XSHEBOND" or "XSHGBOND". Valid values:

• "ConvertibleBond" (default): For "XSHEBOND", the order book includes only bonds with symbol prefixes "123", "127", or "128". For "XSHGBOND", the order book includes only bonds with symbol prefixes "110", "111", "113", or "118".
• "All": Includes all bonds, with no symbol prefix restrictions.

priceScale (optional): A positive integer that specifies the scaling factor for the priceColumn column in the input table. The default value is 10000. When outputting results, the engine divides the calculated values by priceScale.

endTime (optional): A TIME scalar that specifies the upper bound of the last snapshot output window's right boundary. The engine outputs snapshot data for a window only if the window's right boundary timestamp is less than endTime.

You can set this parameter only when triggerType = "mutual". If you do not specify it, the default value of endTime is 15:00:00.000 when exchange = "XSHGBOND"; for all other values of exchange, the default value of endTime is 14:57:00.000.

To enable the snapshot mechanism, you must specify both snapshotDir and snapshotIntervalInMsgCount.

snapshotDir (optional): A STRING scalar that specifies the directory where engine snapshots are stored.

• The specified directory must already exist; otherwise, the system raises an exception.
• When you create a streaming engine with snapshotDir specified, the system checks whether snapshots exist in that directory. If a snapshot exists, the system loads it and restores the engine state.
• The system distinguishes snapshot files by engine name to allow multiple engines to use the same directory.
• An engine snapshot may use three filenames:
  • For temporary snapshot storage, the snapshot is named as .tmp.
  • After the snapshot is generated and flushed to disk, it is saved as .snapshot.
  • If a snapshot with the same name already exists, the older snapshot is automatically renamed to .old.

snapshotIntervalInMsgCount (optional): An integer that specifies how many records trigger one snapshot save for the streaming engine.

A table.

Before you run the code, download the file orderbookDemoInput.zip.

Example 1: This example shows how to build a high-frequency order book from historical tick-by-tick data. In this example, the engine is configured to compute and output a 10-level bid-ask order book for SZSE stocks every second. It demonstrates the complete data processing workflow, including data loading, engine creation, data insertion, and result verification.

// Log in to the DolphinDB server
login("admin", "123456")

// Release any existing engine
try { dropStreamEngine("demo") } catch(ex) { print(ex) }

// Create an output table, which will be passed to the outputTable parameter
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

// Create a dummy table to define the schema of the input table, which will be passed to the dummyTable parameter
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)

// Create a dictionary to define the meaning of each column in the input table, which will be passed to the inputColMap parameter
inputColMap = dict(`codeColumn`timeColumn`typeColumn`priceColumn`qtyColumn`buyOrderColumn`sellOrderColumn`sideColumn`msgTypeColumn`seqColumn, `SecurityID`Time`Type`Price`Qty`BuyNo`SellNo`BSFlag`SourceType`ApplSeqNum)

// Create a dictionary to define the previous day's closing prices, which will be passed to the prevClose parameter. prevClose does not affect any output columns other than the previous day's closing prices.
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])

// Create the engine to compute and output a 10-level bid-ask order book for SZSE stocks every second
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)

// Batch-insert tick-by-tick data of 10 stocks into the order book engine
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

Example 2: This example demonstrates price control and output filtering. Key parameters are set as follows:

• Use maxPrice and minPrice to set the limit up price and limit down price.
• Set priceNullFill = 0 to fill missing prices with 0.
• Use priceScale and precision to set the price scaling factor and the number of decimal places for output prices, respectively.
• Set skipCrossedMarket = false to allow output of crossed bid-ask results.

try { dropStreamEngine("demo") } catch(ex) { print(ex) }  
  
filePath = "./orderbookDemoInput.csv"  
  
// Create a dummy table to define the schema of the input table, which will be passed to the dummyTable parameter
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  
  
// Create a dictionary to define the meaning of each column in the input table, which will be passed to the inputColMap parameter
inputColMap = dict(`codeColumn`timeColumn`typeColumn`priceColumn`qtyColumn`buyOrderColumn`sellOrderColumn`sideColumn`msgTypeColumn`seqColumn, `SecurityID`Time`Type`Price`Qty`BuyNo`SellNo`BSFlag`SourceType`ApplSeqNum)  
  
// Create a dictionary to define the previous day's closing prices, which will be passed to the prevClose parameter
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])  
  
// Set the limit up price and limit down price (actual prices; the engine automatically multiplies them by priceScale)
maxPrice = dict(`000587.SZ`002694.SZ`002822.SZ, [1.83, 7.22, 6.71])  
minPrice = dict(`000587.SZ`002694.SZ`002822.SZ, [1.49, 5.90, 5.49])  
  
// Specify which stocks to output by symbol
outputCodeMap = `000587.SZ`002694.SZ`002822.SZ  
  
// Create the output table schema
outputColMap, outputTableSch = genOutputColumnsForOBSnapshotEngine(basic=true, time=false, depth=(10, true), tradeDetail=true, orderDetail=false, withdrawDetail=false, orderBookDetailDepth=0, prevDetail=false)  
  
// Create the engine and set price-related parameters and output filters
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,  
    outputCodeMap=outputCodeMap,  // Output only the specified stocks
    maxPrice=maxPrice,          // Set the limit up price
    minPrice=minPrice,          // Set the limit down price
    priceNullFill=0,            // Fill missing prices with 0
    priceScale=10000,           // Price scaling factor
    precision=2,                // Keep 2 decimal places for prices
    skipCrossedMarket=false     // Allow output with crossed bid and ask prices
)  
  
// Insert data
engine.append!(select * from loadText(filePath) order by Time)  
select top 10 * from outputTableSch where code in outputCodeMap, timestamp between 2022.01.10T13:15:01.000 and 2022.01.10T13:15:10.000

Example 3: This example demonstrates the trigger mechanism and time control. Key parameters are set as follows:

• Set triggerType = "independent" to enable independent trigger mode.
• Set forceTriggerTime to specify the interval for forced triggers.
• Set orderBySeq=true to ensure that data is processed in sequence order.
• Set independentForceTriggerTime to specify the forced trigger time for groups in independent trigger mode.

try { dropStreamEngine("demo") } catch(ex) { print(ex) }  
  
filePath = "./orderbookDemoInput.csv"  
  
// Create a dummy table to define the schema of the input table, which will be passed to the dummyTable parameter
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  
  
// Create a dictionary to define the meaning of each column in the input table, which will be passed to the inputColMap parameter
inputColMap = dict(`codeColumn`timeColumn`typeColumn`priceColumn`qtyColumn`buyOrderColumn`sellOrderColumn`sideColumn`msgTypeColumn`seqColumn, `SecurityID`Time`Type`Price`Qty`BuyNo`SellNo`BSFlag`SourceType`ApplSeqNum)  
  
// Create a dictionary to define the previous day's closing prices, which will be passed to the prevClose parameter
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])  
    
// Create the output table schema
outputColMap, outputTableSch = genOutputColumnsForOBSnapshotEngine(basic=true, time=false, depth=(10, true), tradeDetail=true, orderDetail=false, withdrawDetail=false, orderBookDetailDepth=0, prevDetail=false)  
  
// Create the engine and set the trigger mechanism and time control
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,  
    triggerType="independent",                      // Independent trigger mode
    forceTriggerTime=5000,                          // Force a trigger after 5 seconds
 orderBySeq=true, // Process data in ID order
    independentForceTriggerTime=3000                // Forced trigger time for independent groups
)  

// Insert data
engine.append!(select * from loadText(filePath) order by Time)  
select top 10 * from outputTableSch where timestamp between 2022.01.10T13:15:01.000 and 2022.01.10T13:15:10.000

Example 4: This example uses outputIntervalOffsetMap to specify different time offsets for specific stocks. Because outputIntervalOffsetMap cannot be used with triggerType = "independent" or triggerType = "perRow", this example sets triggerType = "mutual" (default).

try { dropStreamEngine("demo") } catch(ex) { print(ex) }  
  
filePath = "./orderbookDemoInput.csv"  
  
// Create a dummy table to define the schema of the input table, which will be passed to the dummyTable parameter
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  
  
// Create a dictionary to define the meaning of each column in the input table, which will be passed to the inputColMap parameter
inputColMap = dict(`codeColumn`timeColumn`typeColumn`priceColumn`qtyColumn`buyOrderColumn`sellOrderColumn`sideColumn`msgTypeColumn`seqColumn, `SecurityID`Time`Type`Price`Qty`BuyNo`SellNo`BSFlag`SourceType`ApplSeqNum)  
  
// Create a dictionary to define the previous day's closing prices, which will be passed to the prevClose parameter
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])  
  
// Set output time offsets as a dictionary to assign different output time offsets to different stocks
outputIntervalOffsetMap = dict(`000587.SZ`002694.SZ`002822.SZ, [100, 200, 300])  // Different stocks use offsets of 100 ms, 200 ms, and 300 ms
  
// Create the output table schema
outputColMap, outputTableSch = genOutputColumnsForOBSnapshotEngine(basic=true, time=false, depth=(10, true), tradeDetail=true, orderDetail=false, withdrawDetail=false, orderBookDetailDepth=0, prevDetail=false)  
  
// Create the engine and use the default triggerType="mutual" with outputIntervalOffsetMap
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,  
    outputIntervalOffsetMap=outputIntervalOffsetMap  // Set time offsets
)  
  
// Insert data
engine.append!(select * from loadText(filePath) order by Time)  
  
// View differences in output timestamps for different stocks
select * from outputTableSch where code in `000587.SZ`002694.SZ`002822.SZ, timestamp between 2022.01.10T13:15:00.000 and 2022.01.10T13:15:02.000 order by code, timestamp

Some results are shown below:

Example 5: Use outputColMap to specify the columns to output. The first element returned by genOutputColumnsForOBSnapshotEngine is outputColMap, and the second element determines the schema of outputTable.

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

filePath = "./orderbookDemoInput.csv"

// Create a dummy table to define the schema of the input table, which will be passed to the dummyTable parameter
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

// Create a dictionary to define the meaning of each column in the input table, which will be passed to the inputColMap parameter
inputColMap = dict(`codeColumn`timeColumn`typeColumn`priceColumn`qtyColumn`buyOrderColumn`sellOrderColumn`sideColumn`msgTypeColumn`seqColumn, `SecurityID`Time`Type`Price`Qty`BuyNo`SellNo`BSFlag`SourceType`ApplSeqNum)

// Create a dictionary to define the previous day's closing prices, which will be passed to the prevClose parameter. prevClose does not affect any output columns other than the previous day's closing prices.
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])

//Create outputColMap and outputTableSch to receive the return values of genOutputColumnsForOBSnapshotEngine. They are used to define outputColMap and the schema of outputTable, respectively.
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)

// Batch-insert tick-by-tick data of 10 stocks into the order book engine
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

Some results are shown below:

You can use genOutputColumnsForOBSnapshotEngine to output only the columns you need. Suppose you need to output only some columns from tradeDetail in the example, such as excluding the last four columns. You can process the return values of genOutputColumnsForOBSnapshotEngine as follows:

outputColMap = outputColMap[0 : (size(outputColMap) - 4)]
outputTableSch = outputTableSch[, 0 : (outputTableSch.columns() - 4)]
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)

Example 6: This example uses userDefinedMetrics to add user-defined metrics to the engine, enabling output of user-defined metrics in addition to standard order book data.

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

filePath = "./orderbookDemoInput.csv"

// Create a dummy table to define the schema of the input table, which will be passed to the dummyTable parameter
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

// Create a dictionary to define the meaning of each column in the input table, which will be passed to the inputColMap parameter
inputColMap = dict(`codeColumn`timeColumn`typeColumn`priceColumn`qtyColumn`buyOrderColumn`sellOrderColumn`sideColumn`msgTypeColumn`seqColumn`receiveTime, `SecurityID`Time`Type`Price`Qty`BuyNo`SellNo`BSFlag`SourceType`ApplSeqNum`ReceiveTime)

// Create a dictionary to define the previous day's closing prices, which will be passed to the prevClose parameter. prevClose does not affect any output columns other than the previous day's closing prices.
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])

//Define only outputColMap to receive the first return value of genOutputColumnsForOBSnapshotEngine. Because the metrics defined in userDefinedMetrics require order details and withdrawal details, you must call genOutputColumnsForOBSnapshotEngine with orderDetail=false and withdrawDetail=true
outputColMap = genOutputColumnsForOBSnapshotEngine(basic=true, time=false, depth=(10, true), tradeDetail=true, orderDetail=false, withdrawDetail=true, orderBookDetailDepth=0, prevDetail=false)[0]


//// Define user-defined metrics
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)
}

// Define the output table for the order book engine. It must include the basic columns, multiple levels of bid/ask prices and quantities (depth), and the output of the user-defined metrics in 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)

// Create the engine to generate a 10-level order book for SZSE stocks every second

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) // Set the ReceiveTime column

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

Some results are shown below:

Related function: genOutputColumnsForOBSnapshotEngine

Related tutorial: DolphinDB Order Book Engine: Build High-Frequency Order Books from Tick-by-Tick Data

basic (basic columns)

If neither outputColMap nor userDefinedMetrics is spexified, the engine outputs all columns except openPrice, maxPrice, and minPrice.

time (time columns)

depth (bid/ask levels)

If neither outputColMap nor userDefinedMetrics is specified, the engine outputs all columns in depth. There are two ways to store depth data. You can use only one of them:

• Output one column for each level. For example, if bidsPrice has 10 levels, the engine outputs 10 bidsPrice columns: bidsPrice1, bidsPrice2, …, bidsPrice10.

  Column Name	Type	Description
  bidsPrice	DOUBLE	Multiple columns, each storing a bid price at one level; the number of levels is determined by depth.
  bidsQty	LONG	Multiple columns, each storing the bid volume at one level; the number of levels is determined by depth.
  bidsCount	INT	Multiple columns, each storing the number of bid orders at one level; the number of levels is determined by depth.
  asksPrice	DOUBLE	Multiple columns, each storing an ask price at one level; the number of levels is determined by depth.
  asksQty	LONG	Multiple columns, each storing the ask volume at one level; the number of levels is determined by depth.
  asksCount	INT	Multiple columns, each storing the number of ask orders at one level; the number of levels is determined by depth.

• Output multiple levels into a single column whose type is an array vector. For example, if bidsPrice has 10 levels, the engine outputs those 10 levels as an array vector in a single column: bidsPriceList.

  Column Name	Type	Description
  bidsPriceList	DOUBLE[]	Array vector storing bid prices at multiple levels; the number of levels is determined by depth.
  bidsQtyList	LONG[]	Array vector storing bid volumes at multiple levels; the number of levels is determined by depth.
  bidsCountList	INT[]	Array vector storing the number of bid orders at multiple levels; the number of levels is determined by depth.
  asksPriceList	DOUBLE[]	Array vector storing ask prices at multiple levels; the number of levels is determined by depth.
  asksQtyList	LONG[]	Array vector storing ask volumes at multiple levels; the number of levels is determined by depth.
  asksCountList	INT[]	Array vector storing the number of ask orders at multiple levels; the number of levels is determined by depth.

tradeDetail

orderDetail

withdrawDetail (cancellation details)

Note:

• For SSE order books, if you set includeImmediateExecution = true, the engine outputs immediately executed orders in the order details. In this case, the withdrawBuyOrderQtyList and WithdrawSellOrderQtyList columns will include immediately executed quantities.
• When you call genOutputColumnsForOBSnapshotEngine, if withdrawDetail=true, it does not output the withdrawBuyOrderNoList and withdrawSellOrderNoList columns by default.

prevDetail (details of the previous trade)

orderBookDetailDepth

The engine outputs one column for each level, and each column is an array vector. For example, if buyQtyList has 20 levels, the engine outputs 20 buyQtyList fields: buyQtyList1, buyQtyList2, …, buyQtyList20. The same applies to buyValueList, sellQtyList, and sellValueList.

seqDetail

residualDetail (remaining order details)