Orca 声明式 DStream API 应用：账户持仓损益实时监控

具体数据结构如下所示：

• 持仓信息表：

  列名	类型	备注
  AccountID	SYMBOL	账户代码
  SecurityID	SYMBOL	证券代码
  Date	DATE	日期
  SecurityName	STRING	证券名称
  Threshold	INT	阈值数量
  OpenVolume	INT	期初数量
  PreVolume	INT	盘前持仓
  PreClose	DOUBLE	昨收价

• 委托和成交数据表：

• 行情快照数据表：

本文在持仓监控指标的计算中自定义了一系列指标，其具体含义和代码实现如下所示。需要注意的是，在流计算引擎中使用自定义函数计算指标时，需在定义前添加声明 @state。

CanceledVolume：撤单的委托数量

• 公式：

其中 TolVolume_i 表示委托 i 的初始委托股数，CumVolume_i 表示委托 i 的成交数量。

• 代码实现：

@state
def calCanceledVolume(Mark, Type, Volume){
    return iif(Mark != 0, NULL, cumfirstNot(iif(Type==1, Volume, NULL)).nullFill!(0)-cumsum(iif(Mark in [-1, 1], Volume, 0)))
}

PositionVolume：实时持仓数量

• 公式：

• 代码实现：

@state
def calPositionVolume(Type, PreVolume, CumBuyVol, CumSellVol){
    positionVolume = iif(Type==2, PreVolume+CumBuyVol-CumSellVol, ffill(PreVolume+CumBuyVol-CumSellVol))
    return iif(isNull(positionVolume), PreVolume, positionVolume)
}

ThresholdDeviation：阈值偏离度

• 公式：

• 代码实现：

@state
def calThresholdDeviation(Type, PreVolume, CumBuyVol, CumSellVol, Threshold){
    thresholdDeviation = iif(Threshold>0 and calPositionVolume(Type, PreVolume, CumBuyVol, CumSellVol)>0, (calPositionVolume(Type, PreVolume, CumBuyVol, CumSellVol)-Threshold)\Threshold, 0)
    return round(thresholdDeviation, 6)
}

PositionDeviation：持仓偏离度

• 公式：

• 代码实现：

@state
def calPositionDeviation(Type, OpenVolume, PreVolume, CumBuyVol, CumSellVol){
	positionDeviation = iif(OpenVolume>0 and calPositionVolume(Type, PreVolume, CumBuyVol, CumSellVol)>0, calPositionVolume(Type, PreVolume, CumBuyVol, CumSellVol)\OpenVolume-1, 1)
	return round(positionDeviation, 6)
}

BuyVolume：当日买入数量

• 公式：

• 代码实现：

@state
def calBuyVolume(Type, CumBuyVol){
    buyVolume = iif(Type==2, CumBuyVol, ffill(CumBuyVol))
    return iif(isNull(buyVolume), 0, buyVolume)
}

BuyPrice：当日买入均价

• 公式：

其中，n 表示当天到当前时刻为止的成交买单数量，Price_i 和 Volume_i 分别表示买单 i 的成交价格和成交数量。

• 代码实现：

@state
def calBuyPrice(Type, BSFlag, CumBuyVol, BuyPrice){
    buyPrice = cumsum(iif(isNull(prev(deltas(ffill(CumBuyVol)))) and CumBuyVol!=NULL, BuyPrice*CumBuyVol, ffill(BuyPrice)*deltas(ffill(CumBuyVol))))\ffill(CumBuyVol)
    return round(iif(isNull(buyPrice), 0, buyPrice), 6)
}

SellVolume：当日卖出数量

• 公式：

• 代码实现：

@state
def calSellVolume(Type, CumSellVol){
    sellVolume = iif(Type==2, CumSellVol, ffill(CumSellVol))
    return iif(isNull(sellVolume), 0, sellVolume)
}

SellPrice：当日卖出均价

• 公式：

其中，m 表示当天到当前时刻为止的成交卖单数量，Price_j 和 Volume_j 分别表示卖单 j 的成交价格和成交数量。

• 代码实现：

@state
def calSellPrice(Type, BSFlag, CumSellVol, SellPrice){
    sellPrice = cumsum(iif(isNull(prev(deltas(ffill(CumSellVol)))) and CumSellVol!=NULL, SellPrice*CumSellVol, ffill(SellPrice)*deltas(ffill(CumSellVol))))\ffill(CumSellVol)
    return round(iif(isNull(sellPrice), 0, sellPrice), 6)
}

NetBuyVolume：当日净买入数量

• 公式：

• 代码实现：

@state
def calNetBuyVolume(NetVolume){
    netBuyVolum = ffill(NetVolume)
    return iif(isNull(netBuyVolum), 0, netBuyVolum)
}

FreezeVolume：冻结持仓

• 公式：

其中 CanceledVolume_si 表示卖单 i 中撤单的委托数量。

• 代码实现：

@state
def calFreezeVolume(VOLUME, BSFlag, Mark, Type){
	return cumsum(iif(BSFlag=="B", 0, iif(Mark==0, -calCanceledVolume(Mark, Type, VOLUME), iif(Mark in [1, -1], -VOLUME, VOLUME))))
}

AvailableVolume：可用持仓

• 公式：

• 代码实现：

@state
def calAvailableVolume(Type, PreVolume, CumBuyVol, CumSellVol, VOLUME, BSFlag, Mark){
	availableVolume = PreVolume-calSellVolume(Type, CumSellVol)-calFreezeVolume(VOLUME, BSFlag, Mark, Type)
	return  availableVolume
}

AvailableVolumeRatio：可用持仓比例

• 公式：

• 代码实现：

@state
def calAvailableVolumeRatio(Type, PreVolume, CumBuyVol, CumSellVol, VOLUME, BSFlag, Mark){
	availableVolumeRatio = calAvailableVolume(Type, PreVolume, CumBuyVol, CumSellVol, VOLUME, BSFlag, Mark)\PreVolume
	return round(iif(isNull(availableVolumeRatio), 0, availableVolumeRatio), 6)
}

Profit：当日盈亏

• 公式：

• 代码实现：

@state
def calProfit(Type, PreVolume, BSFlag, SellPrice, BuyPrice, CumSellVol, CumBuyVol, LastPx, PreClose){
    profit = (PreVolume-calSellVolume(Type, CumSellVol))*(LastPx-PreClose)+calSellVolume(Type, CumSellVol)*(calSellPrice(Type, BSFlag, CumSellVol, SellPrice)-PreClose)+calBuyVolume(Type, CumBuyVol)*(LastPx-calBuyPrice(Type, BSFlag, CumBuyVol, BuyPrice))
    return round(profit, 6)
}

本小节将分步构建流图实现持仓监控指标的实时计算。

// 创建数据目录
if (!existsCatalog("positionMonitorDemo")) {
	createCatalog("positionMonitorDemo")
}
go

// 创建流图
use catalog positionMonitorDemo
try { dropStreamGraph("positionMonitor") } catch (ex) {}
positionMonitorGraph = createStreamGraph("positionMonitor") 

// 委托和成交原始数据表 
colNameMarketData = `AccountID`Type`OrderNo`SecurityID`Date`Time`BSFlag`Price`Volume`TradeNo`State`Mark`NetVolume`CumSellVol`CumBuyVol`SellPrice`BuyPrice`ReceivedTime
colTypeMarketData = `SYMBOL`INT`INT`SYMBOL`DATE`TIME`SYMBOL`DOUBLE`INT`INT`SYMBOL`INT`INT`INT`INT`DOUBLE`DOUBLE`NANOTIMESTAMP
MarketDataStream = positionMonitorGraph.source("MarketDataStream", colNameMarketData, colTypeMarketData).parallelize("AccountID", 2)
// 持仓监控基础信息表
colNamePositionInfo = `AccountID`SecurityID`Date`SecurityName`Threshold`OpenVolume`PreVolume`PreClose
colTypePositionInfo = `SYMBOL`STRING`DATE`STRING`INT`INT`INT`DOUBLE
PositionInfo = positionMonitorGraph.source("PositionInfo", colNamePositionInfo, colTypePositionInfo)
// 行情快照数据表
colNameSnapshot = `SecurityID`Date`Time`LastPx
colTypeSnapshot = `SYMBOL`DATE`TIME`DOUBLE
SnapshotStream = positionMonitorGraph.source("SnapshotStream", colNameSnapshot, colTypeSnapshot)

// 关联持仓监控基础信息表
marketJoinPositionInfo = MarketDataStream.lookupJoinEngine(
    rightStream = PositionInfo, 
    metrics = [
        <Type>, <OrderNo>, <Date>, <Time>, <BSFlag>, <Price>,
        <Volume>, <TradeNo>, <State>, <Mark>, <NetVolume>, <CumSellVol>,
        <CumBuyVol>, <SellPrice>, <BuyPrice>, <ReceivedTime>, <SecurityName>, 
        <Threshold>, <OpenVolume>, <PreVolume>, <PreClose>
    	], 
    matchingColumn = `AccountID`SecurityID, 
    rightTimeColumn = `Date
    )

// 每一笔成交相应关联 snapshot 获取市场最新的价格
marketJoinSnapshot = marketJoinPositionInfo.lookupJoinEngine(
	rightStream = SnapshotStream, 
	metrics = [
		<AccountID>, <Type>, <OrderNo>, <Date>, <Time>, <BSFlag>, <Price>, 
        <Volume>, <TradeNo>, <State>, <Mark>, <NetVolume>, <CumSellVol>, 
        <CumBuyVol>, <SellPrice>, <BuyPrice>, <ReceivedTime>, <SecurityName>, 
        <Threshold>, <OpenVolume>, <PreVolume>, <PreClose>, <LastPx>
		],
	matchingColumn = `SecurityID, 
	rightTimeColumn = `Date
	)

// 实时计算监控指标
metrics = [<ReceivedTime>, <Date>, <Time>, <SecurityName>
	, <calPositionVolume(Type, PreVolume, CumBuyVol, CumSellVol) as `PositionVolume>
	, <Threshold>
	, <calThresholdDeviation(Type, PreVolume, CumBuyVol, CumSellVol, Threshold) as `ThresholdDeviation>
	, <OpenVolume>
	, <calPositionDeviation(Type, OpenVolume, PreVolume, CumBuyVol, CumSellVol) as `PositionDeviation>
	, <PreClose>, < PreVolume>
	, <calBuyVolume(Type, CumBuyVol) as `BuyVolume>
	, <calBuyPrice(Type, BSFlag, CumBuyVol, BuyPrice) as `BuyPrice>
	, <calSellVolume(Type, CumSellVol) as `SellVolume>
	, <calSellPrice(Type, BSFlag, CumSellVol, SellPrice) as `SellPrice>
	, <calNetBuyVolume(NetVolume) as `NetBuyVolume>
	, <calAvailableVolume(Type, PreVolume, CumBuyVol, CumSellVol, Volume, BSFlag, Mark) as `AvailableVolume>
	, <calAvailableVolumeRatio(Type, PreVolume, CumBuyVol, CumSellVol, Volume, BSFlag, Mark) as `AvailableVolumeRatio>
	, <calFreezeVolume(Volume, BSFlag, Mark, Type) as `FreezeVolume>
	, <LastPx>
	, <calProfit(Type, PreVolume, BSFlag, SellPrice, BuyPrice, CumSellVol, CumBuyVol, LastPx, PreClose) as `Profit>
	, <now(true) as UpdateTime>
    ]
positionMonitor = marketJoinSnapshot.reactiveStateEngine(
	metrics=metrics, 
	keyColumn=`AccountID`SecurityID, 
	keepOrder=true)
	.sync()
	.sink("PositionMonitorStream")

// 提交流图
positionMonitorGraph.submit()

由于持仓基础信息表是静态数据，因此可以在每日开盘前调用 appendOrcaStreamTable 方法来批量插入数据。

为模拟委托成交数据和行情快照数据的实时传入，本教程使用 useOrcaStreamTable 函数将模拟数据传入 Orca 流表。useOrcaStreamTable 函数通过 Orca 流表名称定位其所在节点，并在该节点获取流表对象。随后将该流表作为首个参数传递给用户指定的函数并执行。具体代码如下：

// 开盘前写入静态持仓基础信息
positionInfo = select * from loadTable("dfs://positionMonitorData", "positionInfo")
appendOrcaStreamTable("PositionInfo", positionInfo)
// 数据回放模拟实时数据写入
// 回放委托成交数据
useOrcaStreamTable("MarketDataStream", def (table) {
    submitJob("replayOrderTrade", "replayOrderTrade", def (table) {
        ds1 = replayDS(sqlObj=<select * from loadTable("dfs://positionMonitorData", "marketData") where Date=2023.02.01>, dateColumn=`Date, timeColumn=`Time, timeRepartitionSchema=cutPoints(09:30:00.000..15:00:00.000, 50))
        replay(inputTables=ds1, outputTables=table, dateColumn=`Date, timeColumn=`Time, replayRate=1, absoluteRate=false, preciseRate=true)
    }, table)
})
// 回放行情快照数据
useOrcaStreamTable("SnapshotStream", def (table) {
    submitJob("replaySnapshot", "replaySnapshot", def (table) {
        ds2 = replayDS(sqlObj=<select * from loadTable("dfs://positionMonitorData", "snapshot") where Date=2023.02.01>, dateColumn=`Date, timeColumn=`Time, timeRepartitionSchema=cutPoints(09:30:00.000..15:00:00.000, 50))
        replay(inputTables=ds2, outputTables=table, dateColumn=`Date, timeColumn=`Time, replayRate=1, absoluteRate=false, preciseRate=true)
    }, table)
})

在本文中，我们通过自定义的数据回放函数来模拟行情快照的实时传入。在自定义的函数中，replayDS 函数可以将输入的 SQL 表达式，根据时间维度划分为多个数据源，作为 replay 函数的输入，逐一进行回放。其中，timeRepartitionSchema 参数用于划分数据源，通过 cutPoints 函数可以将一个交易日均匀划分为50个桶。replayRate 参数决定了数据回放的速率，在 replayRate=1、absoluteRate=false 的情况下，数据将根据时间列的时间跨度原速完成回放。最后，通过 submitJob 函数将数据回放任务提交到 Orca 流表所在的数据节点。

创建流图后，我们可以通过 Web 以及 Orca 提供的 API 接口来查看流图的运行状态。

getStreamGraphInfo("positionMonitorDemo.orca_graph.positionMonitor")

getOrcaStreamTableMeta('positionMonitorDemo.orca_table.MarketDataStream')

getOrcaStreamEngineMeta("positionMonitorDemo.orca_graph.positionMonitor")

通过以下 SQL 语句，可以查询持仓监控指标的计算结果：

result = select * from positionMonitorDemo.orca_table.PositionMonitorStream

其中，PositionMonitorStream 是之前定义的用于存放计算结果的流数据表。在查询流图中的对象时，需要输入查询对象完整的全限定名。查询结果如下所示：

通过编辑 Web 中的数据面板，可以对计算结果进行可视化展示，如下图所示。该面板的 json 配置文件可以在附录中获取。

• 模拟委托成交流水

  • 每秒 1 万条

  • 3,000 支标的

  • 800个账户（每个账户均有2,300+标的）

  • 总共 72,026,955 条记录（09:30-11:30 半个交易日）

• 模拟静态持仓信息

  • 模拟数据

  • 当日800个账户所持的每支标的为一条记录

• 真实行情快照

  • 3 秒频 Level 2 股票行情快照

时延单位：微秒

并发数	回放速度	委托成交流水TPS	全链路总时延（平均/99分位）
8	1X	1w	743 / 1622
16	1X	1w	697 / 1391
8	5X	5w	1045 / 8928
16	5X	5w	877 / 4155

• 并发度为 8 时，面对 1 万 TPS 的流量，全链路平均时延为 743 微秒。

• 同样，并发度为 8 时，即使系统的流量增加至 5 万 TPS，全链路平均时延依然保持在约 1 毫秒。

• 通过提升并发数，可以进一步降低计算时延。尤其在大压力下，并发度由 8 并发增加至 16，99 分位时延能够降低一倍多，这得益于更多的算力保证了系统在流量峰值时刻的表现。

本测试使用了 DolphinDB server 的单节点模式，具体的软硬件环境配置如下表所示：

环境	型号/配置/软件名称/软件版本号
DolphinDB	3.00.4 2025.09.09 LINUX_JIT x86_64
物理服务器1台	系统：CentOS Linux 7 (Core)
	内核：3.10.0-1160.el7.x86_64
	CPU：Intel(R) Xeon(R) Gold 5220R CPU @ 2.20GHz
	license 限制： 8 核 256 GB
	配置文件配置测试节点最大可用内存：256 GB