sse

为对接上海证券交易所行情数据服务，DolphinDB 基于《上海证券交易所行情网关 BINARY 数据接口规范 0.61 版》开发了 sse 插件。通过该插件能够订阅上海证券交易所行情数据网关（MDGW）推送的 Level-1 实时行情数据。

安装插件

版本要求

DolphinDB Server：3.00.6 及更高版本，支持 Linux ABI。

安装步骤

  1. 在 DolphinDB 客户端中使用 listRemotePlugins 函数查看可供安装的插件。

    login("admin", "123456")
listRemotePlugins()

  2. 使用 installPlugin 函数安装插件。

    installPlugin("sse")

  3. 使用 loadPlugin 函数加载插件。

    loadPlugin("sse")

接口说明

createSSEConnection

语法

sse::createSSEConnection(id, host, port, senderCompID, targetCompID, outputTable, [option])

详情

建立与上海证券交易所行情服务器的连接，并返回一个连接句柄。

参数

id STRING 类型标量，指定连接 ID，要求全局唯一。

host STRING 类型标量，指定行情服务器 IP 地址。

port 整型标量，指定行情服务器端口。

senderCompID STRING 类型标量，指定发送方代码。

targetCompID STRING 类型标量，指定接收方代码。

outputTable 字典（Dictionary<STRING, ANY>），指定数据输出表，用于接收被订阅的数据。支持普通共享内存表或者普通共享流表，具体要求见 outputTable 支持的 key。

option（可选参数）字典，支持以下 key：

  • ReceivedTime：key 为字符串，value 为 BOOL 类型标量，默认为 false，指定是否增加一列，记录插件接收到数据的时间，类型为 NANOTIMESTAMP。

  • OutputElapsed：key 为字符串，value 为 BOOL 类型标量，默认为 false，指定是否增加一列，记录从插件接收到数据到写入流表前的时间间隔，类型为 LONG，单位为纳秒。

返回值

返回创建的连接句柄。

start

语法

sse::start(conn)

详情

开始接收数据，将数据存储到数据输出表 outputTable 中。

参数

conn 由 createSSEConnection 接口创建的连接句柄。

stop

语法

sse::stop(conn)

详情

停止接收数据。

参数

conn 由 createSSEConnection 接口创建的连接句柄。

closeSSEConnection

语法

sse::closeSSEConnection(conn)

详情

关闭该连接句柄。

参数

conn 由 createSSEConnection 接口创建的连接句柄。

getStatus

语法

sse::getStatus(conn)

详情

获取该连接的状态。

参数

conn 由 createSSEConnection 接口创建的连接句柄。

返回值

返回一个表格，包含以下字段：

列名 类型 含义
startTime TIMESTAMP 订阅开始的时间
firstMsgTime TIMESTAMP 第一条消息收到的时间
lastMsgTime TIMESTAMP 最后一条消息收到的时间
processedMsgCount LONG 已经处理的消息数
lastErrMsg STRING 最后一条错误信息
failedMsgCount LONG 处理失败的消息数
lastFailedTimestamp TIMESTAMP 最后一条错误发生的时间
sdkStatus STRING SDK 当前状态

getSchema

语法

sse::getSchema(dataType, [option])

详情

获取行情数据的表结构。

参数

dataType STRING 类型标量，表示行情的类型，支持的行情类型见 outputTable 支持的 key 表中的 keyName 列。

option（可选参数）字典，支持以下 key：

  • ReceivedTime：key 为字符串，value 为 BOOL 类型标量，默认为 false，指定是否增加一列，记录插件接收到数据的时间，类型为 NANOTIMESTAMP。

  • OutputElapsed：key 为字符串，value 为 BOOL 类型标量，默认为 false，指定是否增加一列，记录从插件接收到数据到写入流表前的时间间隔，类型为 LONG，单位为纳秒。

返回值

返回一个表，包含两列：name，typeString，分别表示该行情表中的字段名字、字段数据类型。

getHandle

语法

sse::getHandle([id])

详情

获取当前已有的连接句柄。

参数

id（可选参数）STRING 类型标量，表示连接 ID。对应调用 createSSEConnection 接口时指定的 id。

返回值

  • 指定 id 时，返回该 id 标识的连接句柄。

  • 不指定 id 时，返回一个字典，包含当前所有连接句柄。

outputTable 支持的 key

keyName 对应行情 说明
MarketStatus MsgType=M101 市场状态消息
IndexSnapshot MsgType=M102 且 MDStreamID=MD001 指数快照
StockSnapshot MsgType=M102 且 MDStreamID=MD002 股票快照
BondDistributionSnapshot MsgType=M102 且 MDStreamID=MD003 债券分销快照
FundSnapshot MsgType=M102 且 MDStreamID=MD004 基金快照
TreasuryBondPreIssueSnapshot MsgType=M102 且 MDStreamID=MD101 国债预发行快照
BondSnapshot MsgType=M102 且 MDStreamID=MD201 债券快照
OptionSnapshot MsgType=M102 且 MDStreamID=MD301 期权快照
FixedPriceSnapshot MsgType=M102 且 MDStreamID=MD102 盘后固定价格快照
IOPVSnapshot MsgType=M102 且 MDStreamID=MDE01 IOPV 快照

使用示例

// 加载 sse 插件
login("admin", "123456")
installPlugin("sse")
loadPlugin("sse")
// 获取 schema 并建表
// 市场状态消息 (MsgType=M101)
M101Schema = sse::getSchema("MarketStatus")
M101Table = table(1:0, M101Schema.name, M101Schema.typeString)
share M101Table as M101
// 股票快照 (MsgType=M102, MDStreamID=MD002)
StockSnapshotSchema = sse::getSchema("StockSnapshot")
StockSnapshotSchemaTable = table(1:0, StockSnapshotSchema.name, StockSnapshotSchema.typeString)
share StockSnapshotSchemaTable as StockSnapshot
// 指数快照 (MsgType=M102, MDStreamID=MD001)
IndexSnapshotSchema = sse::getSchema("IndexSnapshot")
IndexSnapshotTable = table(1:0, IndexSnapshotSchema.name, IndexSnapshotSchema.typeString)
share IndexSnapshotTable as IndexSnapshot
// 债券分销快照 (MsgType=M102, MDStreamID=MD003)
BondDistributionSnapshotSchema = sse::getSchema("BondDistributionSnapshot")
BondDistributionSnapshotTable = table(1:0, BondDistributionSnapshotSchema.name, BondDistributionSnapshotSchema.typeString)
share BondDistributionSnapshotTable as BondDistributionSnapshot
// 基金快照 (MsgType=M102, MDStreamID=MD004)
FundSnapshotSchema = sse::getSchema("FundSnapshot")
FundSnapshotTable = table(1:0, FundSnapshotSchema.name, FundSnapshotSchema.typeString)
share FundSnapshotTable as FundSnapshot
// 国债预发行快照 (MsgType=M102, MDStreamID=MD101)
TreasuryBondPreIssueSnapshotSchema = sse::getSchema("TreasuryBondPreIssueSnapshot")
TreasuryBondPreIssueSnapshotTable = table(1:0, TreasuryBondPreIssueSnapshotSchema.name, TreasuryBondPreIssueSnapshotSchema.typeString)
share TreasuryBondPreIssueSnapshotTable as TreasuryBondPreIssueSnapshot
// 债券快照 (MsgType=M102, MDStreamID=MD201)
BondSnapshotSchema = sse::getSchema("BondSnapshot")
BondSnapshotTable = table(1:0, BondSnapshotSchema.name, BondSnapshotSchema.typeString)
share BondSnapshotTable as BondSnapshot
// 期权 (MsgType=M102, MDStreamID=MD301)
OptionSnapshotSchema = sse::getSchema("OptionSnapshot")
OptionSnapshotTable = table(1:0, OptionSnapshotSchema.name, OptionSnapshotSchema.typeString)
share OptionSnapshotTable as OptionSnapshot
// 盘后固定价格快照 (MsgType=M102, MDStreamID=MD102)
FixedPriceSnapshotSchema = sse::getSchema("FixedPriceSnapshot")
FixedPriceSnapshotSchemaTable = table(1:0, FixedPriceSnapshotSchema.name, FixedPriceSnapshotSchema.typeString)
share FixedPriceSnapshotSchemaTable as FixedPriceSnapshot
// IOPV快照 (MsgType=M102, MDStreamID=MDE01)
IOPVSnapshotSchema = sse::getSchema("IOPVSnapshot")
IOPVSnapshotTable = table(1:0, IOPVSnapshotSchema.name, IOPVSnapshotSchema.typeString)
share IOPVSnapshotTable as IOPVSnapshot
// 构建 outputTable 字典
tableDic = dict(STRING, ANY);
tableDic["MarketStatus"] = M101
tableDic["StockSnapshot"] = StockSnapshot
tableDic["IndexSnapshot"] = IndexSnapshot
tableDic["BondDistributionSnapshot"] = BondDistributionSnapshot
tableDic["FundSnapshot"] = FundSnapshot
tableDic["TreasuryBondPreIssueSnapshot"] = TreasuryBondPreIssueSnapshot
tableDic["BondSnapshot"] = BondSnapshot
tableDic["OptionSnapshot"] = OptionSnapshot
tableDic["FixedPriceSnapshot"] = FixedPriceSnapshot
tableDic["IOPVSnapshot"] = IOPVSnapshot
conn = sse::createSSEConnection("con1", "127.0.0.1", 8946, "senderID", "targetID", tableDic) // 建立与上证行情服务器的连接
sse::start(conn)  // 开始接收数据
sse::stop(conn)   // 停止接收数据
sse::getStatus(conn)  // 获取连接状态
sse::closeSSEConnection(conn) // 关闭连接句柄