主买是指以卖方的报价成交，主卖是指以买方的报价成交。实时统计主买/主卖交易量能够实时监控资金的流入流出情况。本案例基于中证 1000 指数，介绍如何利用 DolphinDB 流数据处理框架，实时高效计算主买/主卖交易量。完整的代码附件已发布在知乎。
在实际生产环境中，数据往往是以“流”的形式实时注入到数据表中，如何根据数据流实时计算并响应结果是业务面临的重要问题。针对此类问题，DolphinDB 开发了一套完善的流数据的订阅发布机制和多种流计算引擎，为多样化的实时场景提供了灵活的解决方案。
本案例通过行情回放模拟实时数据流，并通过流数据引擎的级联构建计算模型，以实现每分钟输出中证1000实时主买/主卖的交易量的需求。
实现思路
- 计算公式：
- 计算流程：
1. 使用行情回放功能模拟注入数据到流表；
2. 用时间序列引擎订阅流表数据，并计算每只股票每分钟的主买/主卖交易量；
3. 用响应式状态引擎进一步计算每只股票的累计主买/主卖交易量；
4. 用横截面引擎计算 1000 只股票的累计主买/主卖交易量并加权，输出结果。
实时计算主买/主卖交易量
Step 1: 模拟实时行情数据注入
- replay 函数
该函数可以将一个或多个历史数据表（源）的记录以一定的速率写入到目标表中，以模拟流数据实时注入。
根据指定数据表（源）的不同，DolphinDB 提供一对一、多对多和多对一异构回放三种不同的回放形式；根据不同的场景需求，用户还可以指定匀速、倍速和极速三种不同回放速度。
本案例中，我们采用极速回放模式，将中证1000成分股的逐笔成交数据回放到流数据表中。
Step 2: 计算每只股票每分钟的累计交易量
我们首先使用 DolphinDB 内置的时间序列引擎计算每只股票每分钟的累计交易量。
- 时间序列引擎
DolphinDB 提供了三种时间序列引擎，分别为时间序列引擎、 日级时间序列引擎和会话窗口引擎，它们都以时间度量窗口。
时序引擎可以按照指定频率对时序数据进行滑动聚合计算，如计算 K 线等。
日级时序引擎在时序引擎基础上进一步做了扩展。除了可以实现时序引擎的全部功能外，它还可以指定交易时间段，将一个自然日之内各个交易时段开始之前的所有未参与计算的数据，并入该交易时段的第一个窗口进行计算。
会话窗口引擎与时间序列引擎的计算规则和触发计算的方式相同，不同之处在于时间序列引擎具有固定的窗口长度和滑动步长，但会话窗口引擎的窗口不是按照固定的频率产生的，其窗口长度也不是固定的。
本案例中我们使用 createTimeSeriesEngine 创建时间序列引擎，并通过特定的参数设置实现引擎级联；使用 subscribeTable 订阅流数据表，以将订阅数据注入引擎。
Step 3: 累加每只股票每分钟的累计交易量
接下来我们使用响应式状态引擎计算每只股票累计的交易量。
- 响应式状态引擎
在计算有状态因子时，除了需要最新的数据，还需要历史数据或之前计算得到的中间结果，因此有状态因子计算需要存储状态，以供后续因子计算使用，且每次计算都会更新状态。
响应式状态引擎专为这种场景设计，通过在引擎中输入已经验证的因子代码（表达式或函数）以及实时行情数据，输出实时因子值。由于在静态的历史数据集上开发和验证高频因子远比在流数据上开发更为简单，响应式状态引擎显著降低了流式高频因子的开发成本和难度。
响应式状态引擎应用广泛，例如计算有状态的高频因子、主买成交量占比，以及大小单资金流等。
- 优化的状态函数
同时，由于状态算子计算时需要用到历史状态，如果每一次计算都使用全量数据，会导致性能不佳。状态函数的优化，也就是增量方式的流式实现非常关键。目前有约50个状态函数在 DolphinDB 响应式状态引擎中的实现得到了优化。
本案例中，我们使用 createReactiveStateEngine 创建响应式状态引擎，并使用函数 cumsum 来计算每只股票每分钟的主买/主卖累计成交金额。cumsum 采用了增量计算方法，性能优于全量计算的方式
Step 4: 计算1000只股票的累加交易量
最后，我们使用横截面引擎计算加权平均累计交易量。
- 横截面引擎
横截面引擎适用于对截面数据进行实时计算。如金融场景下，使用某个指数的所有成分股的最新价格计算该指数的内在价值。
本例中横截面引擎的触发机制为每插入一条触发一次计算，因此同一时间戳下，每只股票的输入都会产生一条计算结果。
结果展示
计算的最终结果保存在键值内存表中，可以通过 DolphinDB 所有 API 随时查询。
我们还可以用 Grafana 实时监控结果。
本案例为大家展示了如何基于 DolphinDB 流数据处理框架，实时计算中证1000主买/主卖的交易量。用户可以通过时间序列引擎、响应式状态引擎和横截面引擎级联的方式，实现按分钟输出中证1000实时主买/主卖的交易量。整个实现过程简易便捷，是一套适用于业务的高效的解决方案。