WindTDF 行情插件最佳实践指南

为保证后续用于持久化的 enableTableShareAndPersistence 函数能够正常执行，需要节点启动之前在配置文件中（单节点：dolohindb.cfg，集群：cluster.cfg）指定配置参数 persistenceDir ，配置参考功能配置。

创建连接 handle

由于万得 TD 系统在连接行情系统后无法修改订阅，因此在连接前需要预先设置所有数据的订阅。WindTDF 插件的基本使用流程为：创建 handle、进行数据订阅、连接。这里先创建连接 handle，再进行后续的操作。

handle = WindTDF::createHandle([HOST], [PORT], [USERNAME], [PASSWORD])

获取表结构

调用 WindTDF::getSchema 方法可以获取行情数据各个表的表结构。运行以下代码可以获取快照、逐笔委托、逐笔成交三种表结构。注意，万得 TD 系统已经将数据进行了初步的处理和对齐，因此 WindTDF 插件不同市场的同种类型的数据表结构相同，比如上海市场的快照结构与深圳市场相同。

snapshotSchema = WindTDF::getSchema(`snapshot);
orderSchema = WindTDF::getSchema(`order);
tradeSchema = WindTDF::getSchema(`trade);

创建持久化流数据表

通过插件得到行情数据的表结构后，使用该表结构创建三种数据品类的持久化流数据表。

cacheSize = 1000000
snapshot_sh = streamTable(10000:0, snapshotSchema[`name], snapshotSchema[`typeString]);
enableTableShareAndPersistence(table=snapshot_sh,tableName=`snapshot_sh_s,cacheSize=cacheSize, preCache=cacheSize)
snapshot_sz = streamTable(10000:0, snapshotSchema[`name], snapshotSchema[`typeString]);
enableTableShareAndPersistence(table=snapshot_sz,tableName=`snapshot_sz_s,cacheSize=cacheSize, preCache=cacheSize)

order_sh = streamTable(10000:0, orderSchema[`name], orderSchema[`typeString]);
enableTableShareAndPersistence(table=order_sh,tableName=`order_sh_s,cacheSize=cacheSize, preCache=cacheSize)
order_sz = streamTable(10000:0, orderSchema[`name], orderSchema[`typeString]);
enableTableShareAndPersistence(table=order_sz,tableName=`order_sz_s,cacheSize=cacheSize, preCache=cacheSize)

trade_sh = streamTable(10000:0, tradeSchema[`name], tradeSchema[`typeString]);
enableTableShareAndPersistence(table=trade_sh,tableName=`trade_sh_s,cacheSize=cacheSize, preCache=cacheSize)
trade_sz = streamTable(10000:0, tradeSchema[`name], tradeSchema[`typeString]);
enableTableShareAndPersistence(table=trade_sz,tableName=`trade_sz_s,cacheSize=cacheSize, preCache=cacheSize)

代码第一行的 cacheSize 变量控制了在建表时预分配内存的大小、以及流数据表可占用的最大内存，其单位是行，设置较大的 cacheSize 可以降低出现峰值时延的频率。cacheSize 的具体大小可以根据实际的可使用的内存大小决定。具体优化原理可参考 DolphinDB 流计算时延统计与性能优化。

创建分布式数据库

为将行情数据存入分布式数据库，需要根据之前得到的行情数据表结构来创建分布式库表，分区规则参考自基于 DolphinDB 存储金融数据的分区方案最佳实践，使用按天分区 + 按股票代码 HASH25 分区的方式。

dbName = "dfs://WindTDF_quote"
shTransactionTbName = "transactionTable_sh"
shOrderTbName = "orderTable_sh"
shSnapshotTbName = "snapshotTable_sh"

szTransactionTbName = "transactionTable_sz"
szOrderTbName = "orderTable_sz"
szSnapshotTbName = "snapshotTable_sz"

dbDate = database(, partitionType=VALUE, partitionScheme=2024.01.01..2025.01.01)
dbID = database(, partitionType=HASH, partitionScheme=[SYMBOL, 25])
db = database(directory=dbName, partitionType=COMPO, partitionScheme=[dbDate, dbID],engine='TSDB',atomic='CHUNK')

tbSchema = table(1:0, snapshotSchema.name, snapshotSchema.typeInt)
db.createPartitionedTable(table=tbSchema,tableName=shSnapshotTbName,partitionColumns=`nActionDay`szWindCode,sortColumns=`szWindCode`nTime)
shSnapshot = loadTable(db, shSnapshotTbName)
db.createPartitionedTable(table=tbSchema,tableName=szSnapshotTbName,partitionColumns=`nActionDay`szWindCode,sortColumns=`szWindCode`nTime)
szSnapshot = loadTable(db, szSnapshotTbName)

tbSchema = table(1:0, orderSchema.name, orderSchema.typeInt)
db.createPartitionedTable(table=tbSchema,tableName=shOrderTbName,partitionColumns=`nActionDay`szWindCode,sortColumns=`szWindCode`nTime)
shOrder = loadTable(db, shOrderTbName)
db.createPartitionedTable(table=tbSchema,tableName=szOrderTbName,partitionColumns=`nActionDay`szWindCode,sortColumns=`szWindCode`nTime)
szOrder = loadTable(db, szOrderTbName)

tbSchema = table(1:0, tradeSchema.name, tradeSchema.typeInt)
db.createPartitionedTable(table=tbSchema,tableName=shTransactionTbName,partitionColumns=`nActionDay`szWindCode,sortColumns=`szWindCode`nTime)
shTrade = loadTable(db, shTransactionTbName)
db.createPartitionedTable(table=tbSchema,tableName=szTransactionTbName,partitionColumns=`nActionDay`szWindCode,sortColumns=`szWindCode`nTime)
szTrade = loadTable(db, szTransactionTbName)

订阅行情数据到流数据表

使用 WindTDF::subscribe 函数进行 WindTDF 行情订阅。

WindTDF::subscribe(handle, snapshot_sh_s, "SH-2-0", "snapshot");
WindTDF::subscribe(handle, snapshot_sz_s, "SZ-2-0", "snapshot");

WindTDF::subscribe(handle, order_sh_s, "SH-2-0", "order");
WindTDF::subscribe(handle, order_sz_s, "SZ-2-0", "order");

WindTDF::subscribe(handle, trade_sh_s, "SH-2-0", "trade");
WindTDF::subscribe(handle, trade_sz_s, "SZ-2-0", "trade");

有关 WindTDF 插件对数据品类的支持情况和相关订阅参数，参考附录 2。

订阅流表写入分布式库

订阅 2.2.1 节中的三个持久化流数据表，将增量数据实时写入分布式数据库。

def handleInsert(tb, mutable msg) {
        tableInsert(tb, msg)
}

subscribeTable(tableName="snapshot_sh_s", actionName="snapshotTableInsert_sh", offset=-1, handler=handleInsert{shSnapshot}, msgAsTable=true, batchSize=20000, throttle=1, reconnect=true)
subscribeTable(tableName="snapshot_sz_s", actionName="snapshotTableInsert_sz", offset=-1, handler=handleInsert{szSnapshot}, msgAsTable=true, batchSize=20000, throttle=1, reconnect=true)

subscribeTable(tableName="order_sh_s", actionName="orderTableInsert_sh", offset=-1, handler=handleInsert{shOrder}, msgAsTable=true, batchSize=20000, throttle=1, reconnect=true)
subscribeTable(tableName="order_sz_s", actionName="orderTableInsert_sz", offset=-1, handler=handleInsert{szOrder}, msgAsTable=true, batchSize=20000, throttle=1, reconnect=true)

subscribeTable(tableName="trade_sh_s", actionName="tradeTableInsert_sh", offset=-1, handler=handleInsert{shTrade}, msgAsTable=true, batchSize=20000, throttle=1, reconnect=true)
subscribeTable(tableName="trade_sz_s", actionName="tradeTableInsert_sz", offset=-1, handler=handleInsert{szTrade}, msgAsTable=true, batchSize=20000, throttle=1, reconnect=true)

• handleInsert 自定义函数会往将数据写入分布式数据库。
• 通过调整 subscribeTable 函数中的 batchSize 和 throttle 参数可以控制写入分布式数据库的频率，达到其中一个设置的条件，才会向分布式数据库写入一次。
  • batchSize=20000 表示当未处理消息的数量达到 20000 时，handler 才会处理消息。
  • throttle=1 表示继上次 handler 处理消息之后，若未处理消息的数量还没有达到 20000 ，但是时间间隔 1s 后也会处理消息。

启动 WindTDF 连接，开启数据写入

在订阅后，使用函数 WindTDF::connect 连接 WindTDF，行情数据将进入流数据表。

WindTDF::connect(handle)