Synthesize OHLC Bars and Continuous Contracts from Snapshot Data

Snapshot market data are time-interval data slices. The high and low prices in snapshot market data reflect the highest and lowest prices from the market open to the current snapshot timestamp, rather than the high and low within the snapshot interval. However, the high and low prices in an OHLC bar are the highest and lowest prices within the calculation window. Therefore, the high and low of a bar cannot be synthesized simply from the high and low prices in snapshot market data.

If the high (low) price in the snapshot data within the bar window is higher (lower) than the high (low) price in the snapshot data preceding the bar window, then the high (low) price in the snapshot data within the window is the bar's high (low). For example, if the last prices of the snapshots within the window are [3.0, 3.0, 3.11, 3.10], the high prices are [3.12, 3.12, 3.12, 3.12], and the high price of the snapshot immediately preceding the window is 3.11, then the high of the bar is 3.12. If the high price of the snapshot immediately preceding the window is 3.12, the bar's high (low) is calculated as follows: consistent with common industry practice, we use the maximum (minimum) of the last price within the window as the bar's high (low). The user-defined calculation function for the high price of a minute-level OHLC bar is as follows:

defg high(deltasHighPrice, highPrice, lastPrice){
	if(sum(deltasHighPrice)>0.000001){
		return max(highPrice)
	}
	else{
		return max(lastPrice)
	}
}

Parameter Description:

• deltasHighPrice: The difference between the day's high prices (highPrice) of two consecutive snapshot slices for the same futures contract. A value greater than 0 indicates that the intraday high occurred within this interval, in which case the interval's high price is this value (highPrice).

• highPrice: The day's high price from the snapshot market data.

• lastPrice: The last price from the snapshot market data.

Note that this is an approximation algorithm. The last price (lastPrice) of each snapshot within the calculation window may not fully capture the true high and low over the entire window. For example, in the case above, the high price of the snapshot immediately preceding the bar window is 3.12. According to the algorithm, the bar's high is the maximum lastPrice within the window, which is 3.11. However, in practice, a trade at 3.12 may well have occurred within the window. To obtain the true high and low prices of an OHLC bar, tick-by-tick trade data must be used instead of snapshot data.

Similarly, the user-defined function for calculating the low price of a minute-level OHLC bar is as follows:

defg low(deltasLowPrice, lowPrice, lastPrice){
	sumDeltas = sum(deltasLowPrice)
	if(sumDeltas<-0.000001 and sumDeltas!=NULL){
		return min(iif(lowPrice==0.0, NULL, lowPrice))
	}
	else{
		return min(lastPrice)
	}
}

Parameter Description:

• deltasLowPrice: The difference between the daily low prices (lowPrice) of two consecutive snapshot slices for the same futures contract. A value less than 0 indicates that the intraday low occurred within this interval, in which case the interval's low price is this value (lowPrice).

• lowPrice: The daily low price from the snapshot market data.

• lastPrice: The last price from the snapshot market data.

The trading volume, trading value, and number of trades in snapshot market data are all cumulative daily totals. Therefore, before performing rolling window calculations, the incremental difference between two consecutive snapshots must be derived first. The built-in deltas function and context by SQL clause in DolphinDB are used for data preprocessing. The specific implementation will be described in detail below.

For some thinly traded futures contracts, no trades may occur after trading begins at 09:00:00, although snapshot data continues to be pushed normally.

For calculation windows where no trades have occurred since the market open, this tutorial applies the following rules:

• OpenPrice, HighPrice, LowPrice, ClosePrice, Volume, Value are set to 0.

• PreClosePrice, ULimitPrice, LLimitPrice are set to the corresponding values from the snapshot market data.

For some thinly traded futures contracts, certain intraday calculation windows may contain no trades at all, although snapshot data continues to be pushed normally.

For calculation windows with no intraday trades, this tutorial applies the following rules:

• OpenPrice, HighPrice, LowPrice, ClosePrice are set to the ClosePrice of the previous OHLC bar

• Volume, Value are set to 0

• PreClosePrice, ULimitPrice, LLimitPrice are set to the corresponding values of the previous OHLC bar

Since trading hours vary across futures instruments, a minute bar alignment table is generated based on the trading sessions and contract codes of each futures instrument. The table contains the complete set of valid trading minutes for each futures instrument and is used to align minute bars.

For example, the SSE 50 Index, CSI 300 Index, CSI 500 Index, and CSI 1000 Index futures trade from 9:30 to 11:30 and from 13:00 to 15:15. To calculate 1-minute-level OHLC bars, the table includes data for each minute within the two trading sessions (9:30-11:30 and 13:00-15:15) for these index futures.

Note:

For detailed steps on generating the alignment table, refer to the appendix.

Preliminary Processing of Raw Snapshot Market Data

Taking the raw market data table as input, the following preprocessing steps are applied to the raw snapshot data at the minute level:

• Calculate the change in high and low prices between two consecutive snapshots for the same futures contract.

• Use the deltas function to compute the difference between adjacent elements for each metric.

• Calculate the incremental changes in trading volume, trading value, and number of trades between two consecutive snapshots of the same futures contract, and generate table tempTB1. The Volume and Turnover in the resulting minute bar are the sum of these incremental values within the one-minute window.

Perform N-Minute Window and N-Minute Step Aggregation

Perform a rolling window calculation on the processed snapshot data table tempTB1 using a window of N minutes and a step of N minutes to generate table tempTB2. Key processing steps are as follows:

• Call the user-defined functions high and low to calculate the high and low prices of the OHLC bar.

• Call the DolphinDB built-in function interval to downsample the data by datetime with a duration of 60 seconds. At this step, any missing snapshot data within intraday calculation windows is uniformly filled with 0. Further processing will be applied later to bars where all fields are 0.

• The aggregation variables include exchange code, futures contract code, trading date, settlement date, and the datetime after downsampling.

  Note:

  For the specific usage and parameters of the interval function, refer to the documentation interval. The label parameter specifies which boundary of the grouping interval to use as the label for output. The value can be 'left' or 'right'. In this tutorial, the minute-level futures data is output using the right boundary of the grouping interval. The label parameter can be adjusted according to specific requirements.

Align daily OHLC bars and fill missing snapshot calculation windows

After standardizing the futures contract codes in the minute-level table tempTB2 generated from the aggregation calculation, extract the contract codes from tempTB2 and align the minute-level data in tempTB2 with the trading sessions of each corresponding futures instrument to obtain the table result.

Apply the following processing steps to the result to obtain the final bar table res:

• For calculation windows with no intraday trades, OpenPrice, HighPrice, LowPrice, and ClosePrice are filled with the ClosePrice of the preceding bar, and PreClosePrice is filled with the corresponding value of the preceding bar.

• Delete records where the intraday trading time or closing price is null.

• Convert all SortCo data to uppercase.

The backward adjustment approach takes the first day's contract price level as the benchmark and adjusts subsequent contract prices to ensure continuity.

• The function takes the main contract minute-level OHLC bar table (result) and the temp main contract daily table as input parameters, and uses the daily table to identify the trading days that require updates.

• First, set the backward adjustment factor for the first day to 1, and separate the data of the first day and subsequent days into tables temImprove0 and temImprove1, respectively. Perform a left join between the table containing data after the first day and the full-date table temImprove, then compare the current day with the previous day to determine whether the main contract has changed.

• Calculate the backward adjustment factor (adjustFactor): If the main contract has not changed, the backward adjustment factor remains unchanged; if it has changed, compute the backward adjustment factor using the formula: backwardFactor (previous day's adjustment factor) * forwardClosePrice (closing price of previous day's main contract) / PreClosePrice (previous day's closing price of the current day's main contract).

• Cumulative multiplication of backward adjustment factors: After computing the adjustment factor for the current day, use the cumprod function to cumulatively multiply the adjustment factors for subsequent price adjustments.

• During the left join operation, the contract code of the first occurrence of the continuous contract must be retrieved. Note: The initial adjustment factor values are all 1, so they do not affect the result. For example, when the data for May 14 is left-joined to the data for May 13, a new instrument “rr” appears on May 14 that did not exist on May 13. As a result, the “rr” records for May 14 are lost during the left join. Therefore, the data lost on the first occurrence date must be recovered and assigned a backwardadjustment factor of 1.

The forward adjustment approach takes the latest main contract price level as the benchmark and adjusts historical contract prices to ensure continuity.

• First, set the forward adjustment factor for the last day to 1, and separate the data of the last day and previous days into tables temImprove0 and temImprove1, respectively. Perform a left join between the table containing data before the last day and the full-date table temImprove, then compare the current day with the next day to determine whether the main contract has changed.

• Calculate the forward adjustment factor (adjustFactorRes): If the main contract has not changed, the forward adjustment factor remains unchanged; if it has changed, compute the forward adjustment factor using the formula: forwardFactor (next day's adjustment factor) * backwardPreClosePrice (closing price of next day's main contract) / ClosePrice (closing price of the current main contract).

• Cumulative multiplication of forward adjustment factors: After computing the adjustment factor for the current day, use the cumprod function to cumulatively multiply the adjustment factors for subsequent price adjustments.

• Retrieve the main contract code appearing for the first time in the left join. Note: The initial adjustment factors are all 1, so they do not affect the result.

The following example demonstrates the creation of partitioned tables for snapshot data and minute-level data of commodity futures and treasury bond futures, based on the characteristics of their respective trading frequencies and market sizes:

// Import module file
use FuturesOLHC::createTable
// Database name for commodity futures snapshot data
comdtySnapDbname = "dfs://comdtySnapDb"
// Table name for commodity futures snapshot data
comdtySnapTbname = "comdtySnapTb"
// Database name for commodity futures minute-level data and continuous contract data
comdtyMinDbname = "dfs://comdtyMin"
// Table name for commodity futures minute-level data
comdtyMinTbname = "futurePrice1Min"
// Table name for commodity futures minute-level continuous contract data
comdtyZlTbname = "futurePrice1MinZl"

// Database name for treasury bond futures snapshot data
rtSnapDbname = "dfs://rtSnapDb"
// Table name for treasury bond futures snapshot data
rtSnapTbname = "rtSnapTb"
// Database name for treasury bond futures minute-level and continuous contract data
rtMinDbname = "dfs://rtMin"
// Table name for treasury bond futures minute-level data
rtMinTbname = "futurePrice1Min"
// Table name for treasury bond futures minute-level continuous contract market data
rtZlTbname = "futurePrice1MinZl"

// Create corresponding databases and tables based on specified parameters
createComdtyFuturesDfs(comdtySnapDbname, comdtySnapTbname)
createComdtyFuturesMinDfs(comdtyMinDbname, comdtyMinTbname)
createComdtyFuturesZlDfs(comdtyMinDbname, comdtyZlTbname)

createRtFuturesDfs(rtSnapDbname, rtSnapTbname)
createRtFuturesMinDfs(rtMinDbname, rtMinTbname)
createRtFuturesZlDfs(rtMinDbname, rtZlTbname)

Take the createComdtyFuturesDfs function, which creates partitioned tables for commodity futures minute bars, as an example. It requires two parameters:

For the specific code to create databases and tables, refer to the createTable.dos file in the FuturesOLHC module.

After creating the databases and tables, execute the following code to build the streaming engine and simulate the real-time computation process using the sample data provided in the attachment.

// Import module file
use FuturesOLHC::streamFrame 
use FuturesOLHC::tableSchema
use FuturesOLHC::utils

// Raw commodity futures stream table name
rawComdtyStreamTbName = "comdtyStreamTb"
// Raw treasury bond futures stream table name
rawRtStreamTbName = "rtStreamTb"

try{
	// First, drop the persisted stream table
	unsubAndDropAll(rawComdtyStreamTbName)
    unsubAndDropAll(rawRtStreamTbName)
	// Then create the corresponding persisted stream table
	enableTableShareAndPersistence(table=createComdtyRawTable(), tableName=rawComdtyStreamTbName, cacheSize=10000)
    enableTableShareAndPersistence(table=createRTRawTable(), tableName=rawRtStreamTbName, cacheSize=10000)
}catch(ex){
	print(ex)
}

// Create the streaming engine for commodity futures snapshot data
nMin = 1 //Minute frequency (e.g., 1 indicates generating 1-minute-level OHLC bars)
dbname = comdtyMinDbname // Name of the target database for importing minute-level data
tbname = comdtyMinTbname // Name of the target table for importing minute-level data
FuturesOLHC::streamFrame::buildComdtyFrame(nMin, rawComdtyStreamTbName, dbname, tbname)


// Create the streaming engine for the treasury bond futures snapshot data
nMin = 1 //Minute frequency (e.g., 1 indicates generating 1-minute OHLC bars)
dbname = rtMinDbname // Name of the target database for importing minute-level data
tbname = rtMinTbname // Name of the target table for importing minute-level data
FuturesOLHC::streamFrame::buildRtFrame(nMin, rawRtStreamTbName, dbname, tbname)

// Path to commodity futures data files
comdtyPath = "/dolphindb/data/comdty_tickdata.csv"
// Get the stream table schema
sche = schema(objByName(rawComdtyStreamTbName,true)).colDefs
// Load sample data into an in-memory table
comdty_tickdata = loadText(comdtyPath,,sche)
// Replay sample data into the stream table
replay(comdty_tickdata, objByName(rawComdtyStreamTbName,true), `trd_ts, `trd_ts, -1, false)

// Path to treasury bond futures data files
rtPath = "/dolphindb/data/rt_tickdata.csv"
// Get the stream table schema
sche = schema(objByName(rawRtStreamTbName,true)).colDefs
// Load sample data into an in-memory table
rtTickdata = loadText(rtPath,,sche)
// Replay sample data into the stream table
replay(rtTickdata, objByName(rawRtStreamTbName,true), `trd_ts, `trd_ts, -1, false)

Take the buildComdtyFrame function, which creates a stream processing engine for commodity futures snapshot data, as an example. It takes four parameters:

For detailed steps on building the streaming engine, refer to Chapter 3. For the specific code, refer to the streamFrame.dos file in the FuturesOLHC module.

This section uses sample data to demonstrate how to compute minute-level OHLC bars from historical futures snapshot data.

use FuturesOLHC::batchFrame

// Compute OHLC bars from historical commodity futures snapshot data
dataType = "cmdty" 
dbname = comdtySnapDbname // Database name for storing historical raw data
tbname = comdtySnapTbname // Table name for storing historical raw data
// Load sample data into the database table
pt = loadTable(dbname, tbname)
pt.tableInsert(comdty_tickdata)

startDate = 2024.10.10 // Calculation start date
endDate = 2024.10.10 // Calculation end date
nMin = 1 //Minute frequency (e.g., 1 indicates generating 1-minute OHLC bars)
res  = getFuturesKMin(dataType, dbname, tbname, startDate, endDate, nMin)

// Compute OHLC bars from historical treasury bond futures snapshot data
dataType = "rt" 
dbname = rtSnapDbname // Database name for storing historical raw data
tbname = rtSnapTbname // Table name for storing historical raw data
// Load sample data into the database table
pt = loadTable(dbname, tbname)
pt.tableInsert(rtTickdata)

startDate = 2024.10.08 // Calculation start date
endDate = 2024.10.08 // Calculation end date
nMin = 1 //Minute frequency (e.g., 1 indicates generating 1-minute OHLC bars)
res  = getFuturesKMin(dataType, dbname, tbname, startDate, endDate, nMin)

The function getFuturesKMin for computing minute-level OHLC bars from futures snapshot data takes six parameters:

For detailed calculation steps for generating minute-level OHLC bars from historical futures snapshot data, refer to Chapter 1; the specific code can be found in the FuturesOLHC module under the batchFrame.dos file.

This section uses sample data to demonstrate how to synthesize continuous contract market data from historical futures snapshot data.

// Commodity Futures Continuous Contract Market Data
minDbname = comdtyMinDbname
minTbname = comdtyMinTbname
zlDbname = comdtyMinDbname
zlTbname = comdtyZlTbname
startDate = 2024.10.10
endDate = 2024.10.10
nMin = 1
comdtyZL = comdtyZlFuturesKMin(minDbname, minTbname, zlDbname, zlTbname, startDate, endDate, nMin)

// Treasury Bond Futures Continuous Contract Market Data
minDbname = rtMinDbname
minTbname = rtMinTbname
zlDbname = rtMinDbname
zlTbname = rtZlTbname
startDate = 2024.10.08
endDate = 2024.10.08
nMin = 1
rtZL = rtZlFuturesKMin(minDbname, minTbname, zlDbname, zlTbname, startDate, endDate, nMin)

Here, taking the function comdtyZlFuturesKMin for synthesizing continuous contract market data from historical commodity futures snapshot data as an example, seven parameters are required:

For detailed calculation steps for synthesizing continuous contract market data from historical futures snapshot data, refer to Chapter 2; the specific code can be found in the FuturesOLHC module under the batchFrame.dos file.