DolphinDB历史数据回放教程

What this page covers

• 背景：回测与实盘代码复用的动机与问题描述。
• 流数据处理框架：发布-订阅-消费模式与 API 概览。
• replay：函数用途与关键参数说明。
• replayDS：将 SQL 转为数据源并按时间拆分的机制。
• 案例：回放金融数据并计算 ETF 价值的流程展示。
• 性能测试：服务器配置、脚本与回放耗时描述。

技能认证特训营第二期正式开启（限时报名）

页面顶部包含活动报名与优惠提示，并提供报名链接。

• 该活动提供报名入口链接。
• 活动信息以限时报名提示呈现。

DolphinDB历史数据回放教程（作者与日期）

页面展示文章标题、作者署名与发布日期信息。

• 作者署名为 Junxi。
• 发布日期为 2021-05-18。

背景：用历史数据以“实时数据”方式回测以复用代码

说明量化策略回测与实盘在事件驱动差异导致两套代码问题，并提出通过流表回放历史数据以复用同一套代码。

• 实盘处理实时数据通常是事件驱动。
• 回测使用历史数据时程序通常不是事件驱动。
• 同一策略可能需要两套代码，耗时且易出错。
• 可将历史数据按时间顺序以“实时数据”方式导入流表。
• 目标是用同一套代码进行回测与实盘交易。

DolphinDB 流数据处理框架与 API 概览

介绍发布-订阅-消费模型、订阅侧处理方式（自定义函数或聚合引擎）以及多语言流数据 API，并给出流数据教程链接。

• 流数据处理框架采用发布-订阅-消费模式。
• 订阅者可用自定义函数处理消息。
• 订阅者可用内置聚合引擎处理消息。
• 流数据接口支持 C++、C#、Java、Python 等语言。
• 页面提供流数据教程链接。

本文范围：介绍 replay / replayDS 并展示回放过程与应用场景

声明本文将介绍 replay 与 replayDS 两个函数并用金融数据展示数据回放。

• 本文将介绍 replay 函数。
• 本文将介绍 replayDS 函数。
• 本文包含金融数据回放示例。

函数介绍：replay

给出 replay 函数签名、用途与参数说明（输入/输出表、日期时间列、回放速率、并行度等）。

• 函数签名：replay(inputTables, outputTables, [dateColumn], [timeColumn], [replayRate], [parallelLevel=1])。
• replay 可将若干表或数据源回放到相应输出表。
• inputTables 可为单表或包含若干表/数据源的元组。
• 输入表与输出表数量需一致并一一对应。
• 每对输入/输出表需要相同结构。
• 不指定 dateColumn 时默认第一列为日期列。
• 若时间列同时包含日期与时间，可将 dateColumn 与 timeColumn 设为同一列。
• 系统根据日期/时间列设定决定回放最小时间精度与批次输出。
• replayRate 表示每秒回放条数，但批次机制可能导致实际速率偏离。
• parallelLevel 表示读取数据的并行度。

函数介绍：replayDS

给出 replayDS 函数签名、用途与参数说明（SQL 元代码、日期/时间列、时间再分区 schema 等）。

• 函数签名：replayDS(sqlObj, [dateColumn], [timeColumn], [timeRepartitionSchema])。
• replayDS 可将 SQL 查询转化为数据源，并与 replay 结合使用。
• replayDS 可按时间顺序将原始 SQL 拆分为若干小 SQL 查询。
• sqlObj 为 SQL 元代码，表示要回放的数据。
• 不指定 dateColumn 时默认第一列为日期列。
• 默认假设日期列是数据源的一个分区列，并据此拆分查询。
• timeRepartitionSchema 可为时间类型向量，用于时间维度拆分。

回放用法示例：单个内存表回放

说明单内存表回放所需参数并提供调用示例。

• 示例调用：replay(inputTable, outputTable, `date, `time, 10)。
• 示例中显式指定 dateColumn 与 timeColumn。

回放用法示例：使用 data source 的单表回放（replayDS + 并行度）

说明大表通过 replayDS 生成数据源后回放，并描述 pipeline 实现与并行读取优势，给出示例代码。

• 可先用 replayDS 将 SQL 转为数据源，再交由 replay 回放。
• replay 内部使用 pipeline 框架，将取数与输出分开执行。
• 当输入为 data source 时，多块数据可并行读取以减少输出等待。
• 示例并行度为 2 表示两个线程取数据。
• 示例代码包含 replayDS 构造 inputDS，并调用 replay(inputDS, outputTable, `date, `time, 1000, 2)。

回放用法示例：使用 data source 的多表回放

说明 replay 支持多表同时回放（元组输入/输出、结构一致、列存在要求）并给出示例。

• 多张输入表可用元组形式传给 replay。
• 需要分别指定与输入一一对应的输出表。
• 每对输入/输出表必须具有相同表结构。
• 若指定日期列或时间列，则所有表都应存在相应列。
• 示例代码展示对多个数据源 ds1/ds2/ds3 与多个输出表 out1/out2/out3 的 replay 调用。

取消回放

介绍通过 submitJob 的任务取消方式与直接调用时通过控制台任务取消方式，并给出相关函数调用示例。

• submitJob 提交的回放可用 getRecentJob(s) 获取 jobId。
• 获取 jobId 后可用 cancelJob 取消回放。
• 直接调用回放时，可在另一个 GUI session 用 getConsoleJobs 获取 jobId。
• 获取 jobId 后可用 cancelConsoleJob 取消回放任务。

如何使用回放数据（订阅与消费方式）

列出回放数据作为流数据后的三种订阅消费方式（脚本回调、内置流计算引擎、第三方客户端 API）并引用手册链接。

• 方式 1：在 DolphinDB 中订阅，并用脚本自定义回调函数消费流数据。
• 方式 2：在 DolphinDB 中订阅，并用内置流计算引擎处理数据。
• 方式 3：第三方客户端通过流数据 API 订阅与消费数据。
• 页面引用 createCrossSectionalAggregator 用户手册链接。

金融示例：回放美国股市一天 level1 数据并计算 ETF 价值

以 2007-08-17 美国股市 level1 数据为例，展示 quotes 表结构预览、数据划分回放、输出流表、权重与聚合函数、创建横截面聚合引擎与订阅、启动回放与查看输出。

• 示例回放 2007 年 8 月 17 日美国股市一天的 level1 交易数据。
• 示例数据存放在 dfs://TAQ 的 quotes 表。
• 示例展示 quotes 表的列名与类型（按页面展示顺序）。
• 示例提到一天 quotes 数据量为 336,305,414 条。
• 示例使用 replayDS 将数据按时间戳分为 62 个部分。
• 示例给出 cutPoints 与 replayDS 构造 rds 的代码。
• 示例创建输出流表 outQuotes。
• 示例定义 ETF 价值聚合函数 etfVal。
• 示例给出 ETF 权重字典 weights 的成分与权重。
• 订阅 outQuotes 时设置发布表过滤条件以减少网络开销与数据传输。
• 示例调用 setStreamTableFilterColumn(outQuotes, `symbol)。
• 示例创建横截面聚合引擎 createCrossSectionalAggregator。
• 示例订阅 outQuotes 并指定 symbol 列表。
• 示例通过 submitJob 启动回放并设置回放速率与并行度。
• 示例 outputTable 展示字段包含 time 与 etf。

性能测试

给出测试服务器配置、测试脚本，并报告在最快速率且无订阅时回放特定数据量的耗时范围。

• 测试服务器型号为 DELL PowerEdge R730xd。
• CPU 为 Intel Xeon(R) CPU E5-2650 v4（24核 48线程 2.20GHz）。
• 内存为 512 GB（32GB × 16, 2666 MHz）。
• 硬盘容量与构成为 17T HDD（1.7T × 10）。
• 硬盘读取速度为 222 MB/s。
• 硬盘写入速度为 210 MB/s。
• 网络为万兆以太网。
• 测试脚本中 submitJob 调用的 replayRate 为空，并行度为 4。
• 在最快速率且输出表无订阅时，回放 336,305,414 条数据耗时为 90~110 秒。

Facts Index