实时选取外汇行情多价源最优价

def cleanEnvironment(){
    try{ unsubscribeTable(tableName = `HKFEMTStreamTable, actionName = "joinLeft") }catch(ex){ print(ex) }
    try{ dropStreamTable(`HKFEMTStreamTable)}catch(ex){ print(ex) }
    try{ unsubscribeTable(tableName = "CFETSTStreamTable", actionName = "joinRight")}catch(ex){ print(ex) }
    try{ dropStreamTable(`CFETSTStreamTable)}catch(ex){ print(ex) }
    try{ dropStreamTable(`bestPriceStreamTable)}catch(ex){ print(ex) }
    try{ dropStreamEngine(`bestPricEngine)}catch(ex){ print(ex) }
    try{ dropStreamEngine(`fillPrevEngine)}catch(ex){ print(ex) }
}
cleanEnvironment()

def createStreamTableFunc()
{
    // Define streamTables that receive original market data: HKFEMTStreamTable and CFETSTStreamTable
    colName = `market`currencyPair`time`bid`bidVolume
 colType = [`SYMBOL,`SYMBOL,`TIMESTAMP,"DOUBLE","INT"]
    HKFEMTTemp = streamTable(20000000:0,colName,colType)
    try{ enableTableShareAndPersistence(table = HKFEMTTemp, tableName = "HKFEMTStreamTable", asynWrite = true, compress = true, cacheSize = 20000000, retentionMinutes = 1440, flushMode = 0, preCache = 10000) }
 catch(ex){ print(ex) }
    CFETSTemp = streamTable(20000000:0,colName,colType)
    try{ enableTableShareAndPersistence(table = CFETSTemp, tableName = "CFETSTStreamTable", asynWrite = true, compress = true, cacheSize = 20000000, retentionMinutes = 1440, flushMode = 0, preCache = 10000) }
 catch(ex){ print(ex) }

    // Define streamTables that save results: bestPriceTempStreamTable
    colName = `currencyPair`time`market`bestPrice`leftBid`rightBid`delay
 colType = [`SYMBOL,`TIMESTAMP,`INT,`DOUBLE,`DOUBLE,`DOUBLE,`DOUBLE]
    bestPriceTemp = streamTable(20000000:0,colName,colType)
    try{ enableTableShareAndPersistence(table = bestPriceTemp, tableName = "bestPriceStreamTable", asynWrite = true, compress = true, cacheSize = 20000000, retentionMinutes = 1440, flushMode = 0, preCache = 10000) }
 catch(ex){ print(ex) }
}
createStreamTableFunc()
go

• go 语句的作用是对代码分段进行解析和执行
• enableTableShareAndPersistence 共享流数据表并持久化到磁盘。上述代码的主要作用是调用该函数创建三个共享的异步持久化流数据表。其中两个是 HKFEMTStreamTable 和 CFETSTStreamTable，它们分别用于接收和发布香港外汇市场和中国外汇交易中心的实时行情数据；另一个是 bestPriceStreamTable，它用于接收和发布引擎处理完毕的最优价结果。

在本案例中，自定义一个函数用于实现最优价和最优价市场的选取，该函数输入参数的数据类型除 threshold 外都是向量，对应原始行情数据表中相关的列；输入参数 threshold 对应模拟数据生成的各个货币对的阈值，其数据类型是一个字典，它的 Key 是货币对代码，Value 是该货币对的阈值。这个函数后续以参数的方式配置到等值连接引擎中。

def bestPriceAndMarket(leftBid, rightBid, leftMarket, rightMarket, currencyPair, threshold)
{
    price = iif(leftBid > threshold[currencyPair] and rightBid > threshold[currencyPair],NULL,iif((leftBid > threshold[currencyPair]),rightBid,iif(rightBid > threshold[currencyPair],leftBid,max(leftBid,rightBid))))
    market = iif(leftBid > threshold[currencyPair] and rightBid > threshold[currencyPair],NULL,iif((leftBid > threshold[currencyPair]),2,iif(rightBid > threshold[currencyPair],1,iif(leftBid > rightBid,1,2))))
    return market, price
}

• iif(cond, trueResult, falseResult)：如果满足条件 condition，则返回 trueResult，否则返回 falseResult。它等效于对每个元素分别运行 if...else 语句
• 为了方便后续数据填充，市场的数据类型转为整形，引擎左表市场返回 1，引擎右表市场返回 2

本案例使用等值连接引擎和响应式状态引擎串联实现，等值连接引擎的输出是响应式状态引擎的输入。代码如下：

fillPrevSch = table(1:0,`time`currencyPair`market`bestPrice`leftBid`rightBid`inTime,[`TIMESTAMP,`SYMBOL,`INT,`DOUBLE,`DOUBLE,`DOUBLE,`NANOTIMESTAMP])
// create reactiveStateEngine
fillPrevEngine = createReactiveStateEngine(name = "fillPrevEngine", metrics = [<Time>,<ffill(Market)>,<ffill(BestPrice)>,<leftBid>,<rightBid>,<(now(true) - inTime) \ 1000000>], dummyTable = fillPrevSch, outputTable = bestPriceStreamTable, keyColumn="CurrencyPair")

HKFEMT_Table_left = table(1:0,`market`currencyPair`time`bid`bidVolume`inTime,[`SYMBOL,`SYMBOL,`TIMESTAMP,`DOUBLE,`INT,`NANOTIMESTAMP])
CFETS_Table_right = table(1:0,`market`currencyPair`time`bid`bidVolume`inTime,[`SYMBOL,`SYMBOL,`TIMESTAMP,`DOUBLE,`INT,`NANOTIMESTAMP])

// create equiJoinEngine
bestPricEejEngine = createEquiJoinEngine(name = "bestPricEngine", leftTable = HKFEMT_Table_left, rightTable = CFETS_Table_right, outputTable = fillPrevEngine, metrics = [<bestPriceAndMarket(HKFEMT_Table_left.bid,CFETS_Table_right.bid,HKFEMT_Table_left.market,CFETS_Table_right.market,currencyPair,threshold) as `market`bestPrice>,<HKFEMT_Table_left.bid>,<CFETS_Table_right.bid>,<max(HKFEMT_Table_left.inTime,CFETS_Table_right.inTime)>], matchingColumn = `currencyPair, timeColumn = `time)

def appendTime(engine,isLeftTable,mutable data)
{
    data[`inTime] = now(true)
    appendForJoin(engine,isLeftTable,data)
}
subscribeTable(tableName="HKFEMTStreamTable", actionName="joinLeft", offset=0, handler=appendTime{bestPricEejEngine, true,}, msgAsTable=true)
subscribeTable(tableName="CFETSTStreamTable", actionName="joinRight", offset=0, handler=appendTime{bestPricEejEngine, false,}, msgAsTable=true)

• 选取最优价及市场的函数配置在等值连接引擎中
• 等值连接引擎的输出参数配置为响应式状态引擎，表变量 fillPrevSch 表示响应式状态引擎输入的表结构，其字段顺序和类型要和等值连接引擎的输出完全一致
• ffill：此函数用在响应式状态引擎中，表示对参数列中的空值用相同货币对的非空前值做填充
• 自定义函数 appendTime 在数据写入等值连接引擎前增加了写入时间，方便后续做延时的统计，该时间精度为纳秒

通过 Python API 订阅结果表，能够实时获取最优价及对应的价源，订阅的代码如下：

import dolphindb as ddb
import numpy as np
from threading import Event

def resultProcess(lst):
    print(lst)
   
s = ddb.session()
s.enableStreaming()
s.subscribe(host = "192.168.100.3", port = 8848, handler = resultProcess, tableName = "bestPriceStreamTable", actionName="pySub", offset=-1, resub=False)

Event().wait()

• 执行 Python 代码前，必须先在 DolphinDB server 端定义流数据表 bestPriceStreamTable
• s.enableStreaming() 函数表示启用流数据功能
• subscribe 方法
  • host 和 port 参数为 DolphinDB server 的 IP 地址和端口
  • handler 参数为回调函数，示例代码自定义了 resultProcess 回调函数，动作为打印实时接收到的数据
  • tableName 参数为 DolphinDB server 端的流数据表，示例代码订阅了 bestPriceStreamTable
  • offset 参数设置为 -1，表示订阅流数据表最新记录
  • resub 参数为是否需要自动重连

Excel 是常用的办公软件，DolphinDB 的 Excel 插件能够把实时计算的结果推送至 Excel。在 Excel 插件连接到 DolphinDB server 之后，能够获取到共享的流数据表，本案例如下图所示：

右击 bestPriceStreamTable 流数据表，选择 Export by Subscribe，会弹出输入主键的对话框，如下图：

在对话框中输入货币对列 currencyPair，点击确定，即完成订阅。

Grafana 是一个开源的数据可视化 Web 应用程序，擅长动态展示时序数据，支持多种数据源。DolphinDB 开发了 Grafana 数据源插件 (dolphindb-datasource)，让用户在 Grafana 面板 (dashboard) 上通过编写查询脚本、订阅流数据表的方式，与 DolphinDB 进行交互 (基于 WebSocket)，实现 DolphinDB 时序数据的可视化。详细教程参考：Grafana 连接 DolphinDB 数据源

本案例展示货币对人民币对美元的最优价，Grafana 中的 Query 代码如下：

select  gmtime(time) as ts, bestPrice from bestPriceStreamTable where currencyPair='USD/CNY' context by second(time) limit -1

因为 Grafana 默认显示 UTC 时间，和 DolphinDB server 内的数据时间存在 8 个小时时差，所以 Grafana 中的 Query 需要使用 gmtime 函数进行时区的转换。

本章节在前文已搭建的整体框架的基础上，进行行情的回放，模拟实时数据写入。代码如下：

//replay data
submitJob("replayL","replayToLeft",replay,CFETS_Tick,`CFETSStreamTable,`time,,100)
submitJob("replayR","replayToRight",replay,CFETC_Tick,`CFETCTStreamTable,`time,,100)
t = getRecentJobs(2)

执行完后，查看变量 t 内的信息，如下图所示：

如果 endTime 和 errorMsg 字段为空，说明任务正在正常运行中。

通过 GUI，VScode 插件等客户端能够查询结果表，查询语句如下：

select top 29 * from bestPriceStreamTable order by time desc

多次执行此语句，可发现结果在实时发生变化。VScode 的数据查询结果如下：

• delay 列表示计算延时，单位是毫秒