基于期货快照行情数据计算合约 K 线以及主连行情

快照行情是时间间隔的数据切片。快照行情中的最高价和最低价数据反映的是从开盘到最新切片时刻的最高价和最低价，而不是切片间隔内的最高价和最低价。但是 K 线中最高价和最低价，则是计算窗口内的最高价和最低价。因此，我们不能简单的以快照行情的最高价和最低价来合成 K 线的最高价和最低价。

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

参数说明:

• deltasHighPrice: 同一个期货合约相邻两个切片快照的当日最高价（highPrice）的差。该差大于0，表示日内的最高价发生在该区间，此时区间内的最高价为该值
• highPrice: 快照行情的当日最高价
• lastPrice: 快照行情的当日最新价

注意，此算法为近似算法，计算窗口内的每个切片最后一条 LastPrice 价格有可能反映不出来整个时间窗口内的最高价和最低价。譬如上面的例子中，K 线窗口前一切片的最高价为 3.12， 根据算法最高价为 lastPrice 的最大值 3.11，但实际上很有可能存在一个成交价 3.12。要得到 K 线真实的最高价和最低价，只能通过逐笔成交数据来计算，而非现在的快照数据。

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

参数说明:

• deltasLowPrice: 同一个期货合约相邻两个切片快照的当日最低价（LowPrice）的差 。该差小于0，表示日内的最低价发生在该区间，此时区间内的最低价为该值
• lowPrice: 快照行情的当日最低价
• lastPrice: 快照行情的当日最新价

快照行情中的成交量、成交金额和成交笔数都是日累计求和值，所以在进行滚动窗口计算前，需要先求出两笔相邻快照的增量。使用DolphinDB 内置的 deltas 函数和 context by SQL 语句进行数据预处理，具体处理代码下文会详细介绍。

部分成交不活跃的期货，09:00:00 开始交易后一直没有成交，但是快照行情会正常推送。

针对开盘以后一直没有成交的计算窗口，本教程处理规则如下:

• OpenPrice, HighPrice, LowPrice, ClosePrice, Volume, Value 为 0
• PreClosePrice, ULimitPrice, LLimitPrice 为快照行情中对应的值

部分成交不活跃的期货，会出现盘中某些计算窗口内完全没有成交，但是快照行情会正常推送。

针对盘中没有成交的计算窗口，本教程处理规则如下:

• OpenPrice, HighPrice, LowPrice, ClosePrice 等于前一根 K 线的 ClosePrice
• Volume, Value 为 0
• PreClosePrice, ULimitPrice, LLimitPrice 等于前一根 K 线对应的值

由于不同期货品种的交易时间不同，需根据各期货品种的交易时段及交易代码，生成一个用于对齐分钟K线的对齐表。该对齐表将包含每个期货品种在其交易时段内的完整分钟数据。

原始快照行情数据的初步处理

以原始行情数据表为参数，对原始行情数据表按分钟频率进行处理，包括:

• 计算同一个期货的两笔相邻快照的最高价和最低价的变化幅度 ；
• 利用 deltas 函数计算相邻元素之差，后续一分钟 K 线中 Volume、Turnover 应为一分钟内的DeltasVolume、DeltasTurnover 之和。
• 计算同一个期货的两笔相邻快照的成交量、成交金额和成交笔数的增量，生成表 tempTB1 。

进行 N 分钟窗口和 N 分钟步长的聚合计算

针对处理后的快照行情 tempTB1 进行窗口为 N 分钟、步长为 N 分钟的滚动窗口计算，生成表 tempTB2。重要处理步骤如下:

• 调用了自定义函数 high 和 low 计算 K 线的最高价和最低价；
• 调用 DolphinDB 内置函数 interval，以 60 秒作为 duration 对数据基于日期时间进行降频处理。在此步骤时，针对盘中计算窗口缺失快照行情的情况，统一用0填充，后面会针对全部为 0 的 K 线填充数据进一步处理；
• 聚合计算变量包括交易所代码，期货代码，交易日，结算日以及降维之后的日期时间。
  注：

  对于 interval 函数具体用法与参数可参考文档 interval函数。参数中的 label字符串，表示将分组区间的哪一个边界作为 label 输出，可取值为 'left' 或 'right'。本教程中期货数据的分钟频数据根据分组区间右边界输出，可根据具体需求调整 label 参数。

对齐每日 K 线，填充缺失快照行情计算窗口

对于聚合计算后生成的分钟频表 tempTB2 中的期货代码进行标准化处理后，提取 tempTB2 表中期货代码，并将 tempTB2 的分钟频数据和每个期货对应品种的时间段对齐得到表 result 。

针对 result 进行如下处理，得到最终 K 线表 res :

• 针对盘中没有成交的计算窗口，OpenPrice, HighPrice, LowPrice, ClosePrice用前一根 K 线的 ClosePrice 填充 ，PreClosePrice 用前一根 K 线对应的值填充
• 删除盘中交易时间、收盘价为空的数据
• SortCo 数据全部大写处理

后复权方式的思路是以首日合约价格水平为基准，调整后续合约价格以保证连续性。

• adjustFactor 函数以 result 主力分钟频 K 线表，temp 主力日频表作为输入参数，并通过主力日频表获得需要更新的交易日。
• 首先设置后复权首日因子为1，并将首日与后续日期数据分开分别存入表 temImprove0 和表 temImprove1。将首日之后数据的表格与完整日期的表格 temImprove 进行左连接，将当日与前日对比，判断主连行情是否改变。
• 计算后复权因子（adjustFactor）:若主连行情未改变，后复权因子不进行调整；若主连行情发生改变，通过公式 backwardFactor（前一日复权因子）* forwardClosePrice（前一日主力合约收盘价）/ PreClosePrice（当日主力合约前一日收盘价）计算后复权因子
• 复权因子连乘:计算出当日复权因子之后，通过 cumprod 函数将复权因子进行连乘用于后续调整价格
• 在进行左连表格操作时，需要找回首次出现的主连行情代码。注意：首次复权值均为 1 ，因此对结果没有影响。比如，5 月 14 日左连至 5 月 13 日数据，但 5 月 14 日出现了 rr 品种，而 5 月 13 日没有该品种，在左连时 5 月 14 日的 rr 品种会丢失，因此需要对首日丢失的数据进行复权找回。

前复权方式的思路是以最新主力合约价格水平为基准，调整历史合约价格以保证连续性。

• 首先设置最后一日前复权因子为 1，并将最后一日与之前日期数据分开分别存入表 temImprove0 和表 temImprove1。将最后一日之前数据的表格与完整日期的表格 temImprove进行左连，将当日与后一日对比，判断主连行情是否改变。
• 计算前复权因子（adjustFactorRes）:若主连行情未改变，前复权因子不进行调整；若主连行情发生改变，通过公式 forwardFactor（后一日复权因子）* backwardPreClosePrice（后一日主力合约收盘价）/ ClosePrice（主力合约收盘价）计算前复权因子
• 复权因子连乘:计算出当日复权因子之后，通过 cumprod 函数将复权因子进行连乘用于后续调整价格
• 找回左连表格时，首次出现的主连行情代码。注意:首次复权都为1，因此不影响结果

以下示例中基于商品期货和国债期货交易频率与市场规模的特性，演示了商品期货和国债期货快照数据和分钟数据分区表的创建：

/ 引入模块文件
use futureMinK::createTable
// 商品期货快照数据数据库名
comdtySnapDbname = "dfs://comdtySnapDb"
// 商品期货快照数据数据表名
comdtySnapTbname = "comdtySnapTb"
// 商品期货分钟和主连数据库名
comdtyMinDbname = "dfs://comdtyMin"
// 商品期货分钟数据表名
comdtyMinTbname = "futurePrice1Min"
// 商品期货分钟数据主连行情数据表名
comdtyZlTbname = "futurePrice1MinZl"

// 国债期货快照数据数据库名
rtSnapDbname = "dfs://rtSnapDb"
// 国债期货快照数据数据表名
rtSnapTbname = "rtSnapTb"
// 国债期货分钟和主连数据库名
rtMinDbname = "dfs://rtMin"
// 国债期货分钟数据表名
rtMinTbname = "futurePrice1Min"
// 国债期货分钟数据主连行情数据表名
rtZlTbname = "futurePrice1MinZl"

// 根据指定参数创建相应库表
createComdtyFuturesDfs(comdtySnapDbname, comdtySnapTbname)
createComdtyFuturesMinDfs(comdtyMinDbname, comdtyMinTbname)
createComdtyFuturesZlDfs(comdtyMinDbname, comdtyZlTbname)

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

其中，以创建商品期货分钟 K 线持久化库表函数 createComdtyFuturesDfs 为例，需要传入两个参数，分别为：

建库建表的具体代码可参考 futureMinK 模块下的 createTable.dos 文件。

创建好库表之后，即可执行以下代码来构建流计算框架，并根据附件中提供的示例数据模拟实时计算过程。

// 引入模块文件
use futureMinK::streamFrame 
use futureMinK::tableSchema
use futureMinK::utils

// 商品期货原始流数据表名
rawComdtyStreamTbName = "comdtyStreamTb"
// 国债期货原始流数据表名
rawRtStreamTbName = "rtStreamTb"

try{
	// 先删除持久化流表
	unsubAndDropAll(rawComdtyStreamTbName)
    unsubAndDropAll(rawRtStreamTbName)
	// 再创建相应持久化流表
	enableTableShareAndPersistence(table=createComdtyRawTable(), tableName=rawComdtyStreamTbName, cacheSize=10000)
    enableTableShareAndPersistence(table=createRTRawTable(), tableName=rawRtStreamTbName, cacheSize=10000)
}catch(ex){
	print(ex)

// 创建商品期货快照数据流计算框架
nMin = 1 //分钟频率 (例：1 表示生成 1 分钟 K 线)
dbname = comdtyMinDbname // 分钟数据最终导入的数据库名
tbname = comdtyMinTbname // 分钟数据最终导入的数据表名
futureMinK::streamFrame::buildComdtyFrame(nMin, rawComdtyStreamTbName, dbname, tbname)


// 创建国债期货快照数据流计算框架
nMin = 1 //分钟频率 (例：1 表示生成 1 分钟 K 线)
dbname = rtMinDbname // 分钟数据最终导入的数据库名
tbname = rtMinTbname // 分钟数据最终导入的数据表名
futureMinK::streamFrame::buildRtFrame(nMin, rawRtStreamTbName, dbname, tbname)

// 商品期货数据文件存放路径
comdtyPath = "/dolphindb/data/comdty_tickdata.csv"
// 获取流表表结构
sche = schema(objByName(rawComdtyStreamTbName,true)).colDefs
// 将示例数据导入内存表
comdty_tickdata = loadText(comdtyPath,,sche)
// 回放示例数据到流表中
replay(comdty_tickdata, objByName(rawComdtyStreamTbName,true), `trd_ts, `trd_ts, -1, false)

// 国债期货数据文件存放路径
rtPath = "/dolphindb/data/rt_tickdata.csv"
// 获取流表表结构
sche = schema(objByName(rawRtStreamTbName,true)).colDefs
// 将示例数据导入内存表
rtTickdata = loadText(rtPath,,sche)
// 回放示例数据到流表中
replay(rtTickdata, objByName(rawRtStreamTbName,true), `trd_ts, `trd_ts, -1, false)

其中，以创建商品期货快照数据流计算引擎函数 buildComdtyFrame 为例，需要传入四个参数，分别为：

关于流计算框架的详细搭建步骤，可参考本文中第三章，具体代码可参考 futureMinK 模块下的 streamFrame.dos 文件。

本部分以示例数据为例，展示如何基于历史期货快照数据计算分钟 K 线

use futureMinK::batchFrame

// 商品期货历史快照数据 K 线计算
dataType = "cmdty" // 
dbname = comdtySnapDbname// 存放历史原始数据的数据库名
tbname = comdtySnapTbname // 存放历史原始数据的数据表名
// 将示例数据导入库表中
pt = loadTable(dbname, tbname)
pt.tableInsert(comdty_tickdata)

startDate = 2024.10.10 // 计算开始时间 
endDate = 2024.10.10 // 计算结束时间 
nMin = 1 //分钟频率 (例：1 表示生成 1分钟 K 线)
res  = getFuturesKMin(dataType, dbname, tbname, startDate, endDate, nMin)

// 国债期货历史快照数据 K 线计算
dataType = "rt" // 
dbname = rtSnapDbname// 存放历史原始数据的数据库名
tbname = rtSnapTbname // 存放历史原始数据的数据表名
// 将示例数据导入库表中
pt = loadTable(dbname, tbname)
pt.tableInsert(rtTickdata)

startDate = 2024.10.08 // 计算开始时间 
endDate = 2024.10.08 // 计算结束时间 
nMin = 1 //分钟频率 (例：1 表示生成 1分钟 K 线)
res  = getFuturesKMin(dataType, dbname, tbname, startDate, endDate, nMin)

其中，计算期货快照分钟K线数据函数 getFuturesKMin 需要传入六个参数，分别为：

关于历史期货快照数据分钟 K 线的详细计算步骤，可参考本文中第一章，具体代码可参考 futureMinK 模块下的 batchFrame.dos 文件。

本部分以示例数据为例，展示如何基于历史期货快照数据合成主连行情

// 商品期货主连行情
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)

// 国债期货主连行情
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)

其中，以历史商品期货快照数据主连行情合成函数 comdtyZlFuturesKMin 为例，需要传入七个参数，分别为：

关于历史期货快照数据主连行情合成的详细计算步骤，可参考本文中第二章，具体代码可参考 futureMinK 模块下的 batchFrame.dos 文件。