Level 2 量化因子开发和模拟交易的 DolphinDB 最佳实践

分钟频级别的价格走势是量化策略识别趋势、判断强弱、识别拐点的基本输入。例如，收盘价决定模型的预测目标构建，开盘价与最高/最低价则影响回撤、突破策略的判断。同时，需要合理处理异常（如开盘价为 0）与缺失数据。因此，为保证策略使用的数据一致性与稳定性，分钟频价格指标遵循如下计算逻辑：

• 开盘价（open）：

  • 对非开盘、收盘分钟，直接取窗口内首条快照最新价字段的值作为开盘价。

  • 对开盘和收盘分钟：

    • 如果窗口内最新的开盘价为 0，则使用本窗口首条快照最新成交价；

    • 否则采用窗口内最新的开盘价。

• 最高价（high）、最低价（low）：

  • 如果窗口内最后一条数据的最高价（最低价）为 0，分别取本窗口内最新成交价的最大（最小）值。

  • 否则为窗口内最新的最高价（最低价）。

• 收盘价（close）：

  • 取当前窗口最后一条数据的最新价即为收盘价。

成交指标能够反映市场活跃度，是判断“有效突破”与“虚假拉升”的核心参考。例如，量能放大通常伴随趋势延续，量能萎缩可能预示回调。特别是在盘中择时策略中，成交量的实时变化往往决定是否执行信号。由于快照中成交量是累加字段，为得到准确的“本分钟新增成交”，采用差分方式提取每分钟真实变动：

• 取窗口内最新的一条数据的值暂时作为快照成交量、成交额、成交笔数。

• 使用deltas()函数分别计算相邻分钟差值，若差值为空则保留窗口内最新的一条数据的值，否则使用计算出的差值。

• 字段包括：

  • volume：新增成交量。

  • amount：新增成交额。

  • numTrades：新增成交笔数。

快照数据的一大重要价值，在于提供十档盘口的挂单信息，反映当前市场的供需结构。量化策略尤其关注委托量变化，例如通过盘口量价不对称，判断是否存在“压单吸筹”或“拉高出货”。因此，分钟频委托统计是捕捉主力行为的重要窗口。为确保策略捕捉的是每分钟结束时的最新盘口状态，按以下指标对挂单信息进行处理：

• 在计算委托买卖量和金额相关指标时，所有指标均取当前计算窗口内最新一条快照行情的数据：

  • totalSellVolume：委卖总量，取窗口中 totalOfferQuant 字段的最新值。

  • totalBuyVolume：委买总量，取窗口中 totalBuyQuant 字段的最新值。

  • entrustSellAmount：委卖总金额，由卖盘挂单量乘以加权平均卖价计算得到，取窗口内最新一条数据。

  • entrustBuyAmount：委买总金额，类似 entrustSellAmount，由买盘挂单量乘以加权平均买价计算得到。

  • sellAvgPrice，buyAvgPrice：加权平均卖价和买价。

快照数据中包含集合竞价、连续竞价、不规则时间戳与跨市场数据，若不清洗将严重干扰分钟频因子的准确性。特别是在分钟线触发临界点，不标准的时间戳会导致聚合误差与分钟频数据缺失。因此，需对原始快照进行时间清洗和格式化处理，确保其落入正确的聚合窗口。我们采用如下预处理逻辑对快照数据进行筛选与时间标准化：

• 筛选交易时段内的数据：保留集合竞价时段（09:25–09:30）和连续竞价时段（09:30–11:30，13:00–14:57）内的快照数据，剔除无效或非交易时段的数据。

• 优化分钟线触发时机：考虑到每分钟快照数据的最新交易时间为 59 秒，针对这一特点进行了优化处理——当接收到 59 秒的快照数据时，会立即触发分钟线的计算。

以下代码基于 DolphinDB3.00.4版本开发，建议用户使用3.00.4及以上版本

// tse: 时序聚合引擎
// msg: 原始快照数据
def handleRawSnapshot(mutable tse, mutable msg){
    msg = select * from msg
        where ((second(tradeTime) between 09:25:00:11:30:59) || (second(tradeTime) between 12:59:00:14:56:59))
    if (size(msg) == 0) return
    update msg set tradeTime =
        case
            when minute(tradeTime) in 09:25m..09:29m then concatDateTime(date(tradeTime), 09:30:00.000)
            else tradeTime
        end

    msg1 = select * from msg where second(tradeTime) % 60 >= 59
    update msg set tradeTime = timestamp(concatDateTime(date(tradeTime), second(minute(tradeTime)))) + 58 * 1000 + 999 where second(tradeTime) % 60 >= 59
    update msg1 set tradeTime = timestamp(concatDateTime(date(tradeTime), second(minute(tradeTime)))) + 59 * 1000 + 1
    append!(msg, msg1)
    msg.sortBy!(`tradeTime)
    append!(tse, msg)
}

快照数据经过清洗后，需在每分钟的时间窗口内进行高效聚合，从而输出标准化的分钟频行情。在量化场景下，不同股票的快照数量差异极大，传统聚合方式容易漏算、错算、重复计算。而 DolphinDB 的时序聚合引擎滑窗聚合，确保每分钟窗口在数据到齐后立即触发计算，具体处理逻辑包括：

• 设置窗口与步长：窗口大小为 60 秒（1 分钟），步长也是 60 秒，确保每条分钟线独立且无重叠。

• 强制触发机制：针对分钟窗口内只有少量快照数据的情况，设置forceTriggerTime参数强制输出分钟线。

tse = createTimeSeriesEngine(
    name= tseName,
    windowSize=60000,
    step=60000,
    metrics=tseMetrics,
    dummyTable=rawStream,
    outputTable=rse,
    timeColumn=`tradeTime,
    useSystemTime=false,
    keyColumn=`securityID,
    useWindowStartTime=false,
    parallelism=parallelTSE,
    lowLatency = ctx["lowLatency"],
    outputElapsedMicroseconds = true
)

分钟频数据虽然为标准行情摘要，但无法直接用于策略决策，必须进一步加工出如主力行为、盘口强弱、成交偏离等“策略可用特征”。这些因子的生成过程依赖状态变量、滑动窗口、历史状态记忆，需要在毫秒级延迟下完成复杂计算逻辑。为此，DolphinDB 提供响应式状态引擎以流式计算分钟频因子。上一步的分钟聚合结果将被输入至响应式状态引擎，实时生成策略所需的分钟频因子数据：

rse = createReactiveStateEngine(
    name = rseName,
    metrics = rseMetrics,
    dummyTable = tseOutputDummy,
    outputTable = objByName(minStreamTbname, true),
    keyColumn = `securityId,
    parallelism = parallelRSE,
    lowLatency = ctx["lowLatency"],
    outputElapsedMicroseconds = true
)

量化策略通常以资金流向作为判断市场强弱、主力方向的重要输入。而逐笔数据提供了最精细的资金流明细，可以精确拆解为主动买入/卖出金额，进而判断主力行为是否在买入护盘或卖出打压。此外，通过分析最后一笔成交相对于大/小单的偏移，还可以构造微观层面的行为偏差因子。基于这一业务需求，以下指标用于衡量当前分钟内成交的资金流动方向和规模，计算逻辑如下：

• sellAmount：主动卖出成交金额。

• buyAmount：主动买入成交金额。

• TradeBias：衡量最后成交是否偏向大单买入或卖出的因子，计算逻辑为：(lastAmount - sellMinAmount) / abs(sellMinAmount) - (lastAmount - buyMinAmount) / abs(buyMinAmount)

  • lastAmount：当前分钟最后一笔成交金额。

  • buyMinAmount：主动买入方向中，所有单笔成交金额的最小值。

  • sellMinAmount：主动卖出方向中，所有单笔成交金额的最小值。

在市场博弈中，大资金往往通过大单执行交易，而小单则多为散户行为。将主动成交划分为大单/小单，有助于识别是否有主力资金正在悄悄建仓或出货。特别是在某些个股活跃时段，大单持续出现常常意味着背后存在机构行为，值得策略重点关注。因此，为了刻画不同规模主动资金行为，定义以下大/小单资金流指标，用于分钟频上量化资金行为的结构：

将单笔成交量是否大于 50,000 作为判断标准：

• buySmallAmount：主动买入且成交量 ≤ 50,000 的资金总额。

• buyBigAmount：主动买入且成交量 > 50,000 的资金总额。

• sellSmallAmount：主动卖出且成交量 ≤ 50,000 的资金总额。

• sellBigAmount：主动卖出且成交量 > 50,000 的资金总额。

在逐笔成交数据中，存在乱序、时间格式不统一、跨市场混合等情况。若直接进行聚合计算，容易导致数据遗漏或窗口划分错误，直接影响分钟频因子的精度与完整性。因此，为确保逐笔成交数据在分钟频聚合时准确划分窗口，确保所有成交数据有序，需对原始成交数据进行基本的清洗和时间标准化处理。具体步骤如下：

1. 排序对齐：首先对逐笔成交数据按 tradeTime 升序排序，确保数据的时间顺序一致，避免因乱序导致被时序引擎丢弃，造成分钟频指标正确性问题。

2. 数据写入：将时间频率转成秒频，写入时序引擎，以供后续分钟频聚合计算使用。

// tse: 时序聚合引擎
// msg: 原始快照数据
def handleRawTrade(mutable tse, mutable msg){
    msg.sortBy!(`tradeTime)
    // 转成秒频数据，避免乱序问题
    replaceColumn!(msg, `tradeTime, timestamp(datetime(msg.tradeTime)))
    tse.append!(msg)
}

经过标准化的成交数据会被注入时序聚合引擎，进行分钟频窗口划分与初步聚合。逐笔成交密度高但分布不均，必须设置合理的窗口配置与强制触发机制，确保每个自然分钟都有输出。DolphinDB 的时序聚合引擎通过设定parallelism参数，支持多股票并行滑窗聚合，并提供交易时段控制与性能输出功能，确保实盘环境下的实时性。具体处理逻辑如下：

• 窗口配置：设置windowSize=60000和step=60000，即每个聚合窗口大小为 1 分钟，滑动步长也是 1 分钟，实现不重叠的自然分钟划分。

• 强制触发控制：设置forceTriggerTime=100，及时触发不活跃股票的分钟频计算。

tse = createTimeSeriesEngine(
        name= tseName,
        windowSize=60000,
        step=60000,
        metrics=tseMetrics,
        dummyTable=rawStream,
        outputTable=rse,
        timeColumn=`tradeTime,
        useSystemTime=false,
        keyColumn=`securityID,
        useWindowStartTime=false,
        parallelism=parallelTSE,
        lowLatency = ctx["lowLatency"],
        outputElapsedMicroseconds = true
)

基础聚合数据仅提供成交总额、笔数等粗粒度信息，无法满足策略对资金结构、行为偏差、大单交易等特征的需求。DolphinDB 通过响应式状态引擎，对分钟频成交数据流进行细粒度加工，构建具有策略解释力的因子特征，并保持实时输出能力。分钟频聚合结果被输入至状态引擎，实现分钟频因子的流式计算与实时输出：

rse = createReactiveStateEngine(
    name = rseName,
    metrics = rseMetrics,
    dummyTable = tseOutputDummy,
    outputTable = objByName(minStreamTbname, true),
    keyColumn = `securityId,
    parallelism = parallelRSE,
    lowLatency = ctx["lowLatency"],
    outputElapsedMicroseconds = true
)

衍生因子需要同时利用快照因子和成交因子进行计算，这要求系统具备跨表合并、并发处理和状态维护能力。同时，生成的衍生因子还要供下游推理模块实时调用，必须保证数据结构统一、处理路径清晰。因此，DolphinDB 设计了如下几类核心组件构建数据流处理链路：

• EquiJoin 流引擎（EquiJoinEngine）：用于将两个来源（快照分钟频数据与成交分钟频数据）在 tradeTime 和 securityId 维度上匹配，形成分钟频的合并输入流。

• 响应式状态引擎（ReactiveStateEngine）：接收 EquiJoinEngine 输出的合并流数据，在此基础上完成衍生因子计算。该引擎按证券代码进行状态维护与指标更新，支持并发处理。

• 衍生因子流数据表：用于记录状态引擎输出的分钟频因子流，支持持久化、订阅和后续推理生成交易信号。

为了使分钟频因子具备更强的表达能力，需将来源不同的分钟频数据进行统一对齐。DolphinDB 的 EquiJoinEngine 提供了毫秒级别的时间对齐能力，确保合并后的数据可直接用于下游状态引擎的衍生因子计算。该引擎对接来自快照与成交的分钟频因子表，对来自两个来源的分钟频数据表进行合并。

• 合并数据源：

  • 左表（LeftTable）：基于快照行情生成的分钟频因子表。

  • 右表（RightTable）：基于逐笔成交生成的分钟频因子表。

• 合并方式：

  • 匹配字段为 securityId。

  • 对齐的时间字段为 tradeTime。

ejEngine = createEquiJoinEngine(
    name = "EquiJoinEngine",
    leftTable = snapMinDummy,
    rightTable = tradeMinDummy,
    outputTable = rse,
    metrics = sqlCol(colNames[2:]),
    matchingColumn = `securityId,
    timeColumn = `tradeTime
)

合并后的分钟频数据拥有更丰富的信息密度，可以构造如“成交额偏离盘口均价”、“主力行为 VS 挂单反应”、“大单主导趋势”等更具策略价值的复合因子。这类因子通常涉及复杂计算逻辑，因此使用 DolphinDB 的响应式状态引擎实现高效处理。

rse = createReactiveStateEngine(
    name = rseName,
    metrics = facMetrics,
    dummyTable = dummy,
    outputTable = objByName(factorTbname, true),
    keyColumn = `securityId,
    parallelism = parallelRSE,
    lowLatency = ctx["lowLatency"],
    outputElapsedMicroseconds = true
)

计算结果最终写入衍生因子流数据表中，供下一步模型实时推理并生成交易信号。

订单管理功能允许用户为每笔订单指定开始时间（effectiveTime）和结束时间（expireTime），只有在这一区间内，订单才能被正常撮合执行，一旦达到结束时间，未成交或部分成交的订单会触发撤单操作。

• 保障交易执行的时效性：防止委托单无限期“挂单”，避免在市场行情已经变化且策略意图改变后订单仍在市场中暴露风险。

• 合理模拟实际交易环境：真实交易中，很多订单都会设置有效期（限时有效），通过订单管理功能，仿真交易更贴合真实约束。

• 提升仿真精度和风控能力：及时清理过期订单，有效控制交易风险，避免因过期订单引发的误判。

在仿真环境中，部分成交的订单可能会剩余未成交数量，订单管理功能负责：

• 按结束时间自动触发剩余订单状态更新，避免残留委托干扰后续策略演化。

• 输出完整的成交和订单状态信息，保证仿真数据的完整性和准确性。

• 量化交易策略中，订单生命周期极短，订单管理插件精准模拟订单有效期至关重要。

• 资金管理与风险控制严苛的机构策略，需对订单执行的时限和成交情况设定严格监控。

• 做市商策略中的订单主动撤单和时效控制，保证撮合过程的流畅性和实时反馈。

• 策略测试与回测阶段，对订单撮合时效与部分成交行为进行详细追踪和分析。

附件提供了某日股票原始快照数据和原始成交数据，用于演示实时生成因子到模拟仿真交易的过程。

//导入模块
clearCachedModules()
use SnapTradeStream::CreateTable
use SnapTradeStream::utils
use SnapTradeStream::OHLCFactors
use SnapTradeStream::StreamFrame

// 导入数据
// 路径需修改
colNames, colTypes = getTradeTickColDefs()
tradeData = select * from loadText("/data/trade.csv", ,table(colNames, colTypes)) order by tradeTime

colNames, colTypes = getSnapshotColDefs()
snapshotData = select * from loadText("/data/snapshot.csv", ,table(colNames, colTypes)) order by tradeTime

导入某日成交逐笔数据及快照行情数据之后，需要建立流表，用于后续数据的实时传入和流计算的订阅。

tradeStreamName = "tradeStream"
unsubAndDropAll(tradeStreamName)
sche = schema(tradeData).colDefs
colNames, colTypes = sche.name, sche.typeString
try{
    share(streamTable(100:0, colNames, colTypes), tradeStreamName)
}catch(ex){print(ex)}

snapStreamName = "snapshotStream"
unsubAndDropAll(snapStreamName)
sche = schema(snapshotData).colDefs
colNames, colTypes = sche.name, sche.typeString
try{
    share(streamTable(100:0, colNames, colTypes), snapStreamName)
}catch(ex){print(ex)}

为适应不同数据源的分钟频特征提取需求，系统设计了统一的处理函数映射字典factorFuncDict与并行度配置字典parallelDict，实现高度模块化和灵活调度。

factorFuncDict = {
    "tradeTSE": getTrade1MTSFactor, // 获取成交数据时序聚合引擎指标的函数
    "tradeRSE": getTrade1MReactiveFactor, // 获取成交数据响应式状态引擎指标的函数
    "snapTSE": getSnap1MTSFactor, // 获取快照数据时序聚合引擎指标的函数
    "snapRSE": getSnap1MReactiveFactor, // 获取快照数据响应式状态引擎指标的函数
    "mergedRSE": minDerivedKlineFactors // 获取衍生数据响应式状态引擎指标的函数
}

// 定义并行度参数字典
parallelDict = {
    "tradeTSE": 5,
    "tradeRSE": 5,
    "snapTSE": 5,
    "snapRSE": 5,
    "mergedRSE": 5
}

// 定义模型路径
extraParams = {
    "torchModelPath": "/data/pytorch_model.pt",
    "lgbmModelPath": "/data/lightgbm_model.txt",
    "lowLatency": false
}

• parallelDict配置了每类因子计算任务的并行线程数。当前每类任务均设为5 个并行任务，可根据实际节点资源和流量进行动态调整。

• factorFuncDict为一个字典，键为字符串，值为函数，用于指定获取每个流引擎对应的指标及字段定义，该函数默认定义在klineFactors模块中，比如getTrade1MTSFactor函数，函数定义为：

// 获取成交数据时序聚合引擎指标及字段定义的函数
def getTrade1MTSFactor(){
    d = dict(STRING, ANY, true)
    d["getAmounts(tradeBSFlag, tradePrice, tradeQty, `B)"] = ["buyAmounts", "DOUBLE"]
    d["getAmounts(tradeBSFlag, tradePrice, tradeQty, `S)"] = ["sellAmounts", "DOUBLE"]
    colNames2 = ["buySmallAmount", "buyBigAmount", "sellSmallAmount", "sellBigAmount"]
    colTypes2 = take("DOUBLE", size(colNames2))
    d["calCapitalFlow(buyApplSeqNum,offerApplSeqNum, tradeBSFlag, tradeQty, tradePrice)"] = [colNames2, colTypes2]
    d["lastTradeBias(tradeTime, tradeBSFlag, tradePrice, tradeQty)"] = ["tradeBias", "DOUBLE"]
    return d
}

该函数返回一个字典，其中键为函数调用字符串，值为包含两个元素的列表：第一个元素为指标名称，第二个元素为字段类型。这种设计使得用户可以根据实际业务需求灵活调整配置。当因子计算逻辑发生变化时，用户只需在此处进行相应修改，无需同步更新流数据表和流计算引擎等底层组件，从而显著降低了因子维护和更新的成本，提升了系统的可扩展性和开发效率。

定义所需参数后，调用主函数buildStreamFrame进行订阅计算实时生成分钟频因子。

factorDbname = "dfs://factor"
factorTbname = "minDerivedFactorsTb"
buildStreamFrame(factorDbname, factorTbname, tradeStreamName, snapStreamName, factorFuncDict, parallelDict, extraParams)

其中，buildStreamFrame需要传入七个参数，分别为：

将历史数据通过回放功能插入流表中，以此模拟实时流数据流入的效果，从而触发流计算全流程。

replay(
    [tradeData, snapshotData],
    [objByName(tradeStreamName, true), objByName(snapStreamName, true)],
    `tradeTime,
    `tradeTime,
    100,
    false
)

buildStreamFrame负责创建分钟频因子计算管道。回放完成后，等待衍生因子表产生数据，再启动模型推理和模拟撮合：

maxWaitMs = 60000
waitedMs = 0
do {
    sleep(500)
    waitedMs += 500
} while (size(objByName(factorTbname, true)) == 0 && waitedMs < maxWaitMs)

use SnapTradeStream::streamUtils
derived2Simulator(
    factorTbname,
    extraParams["lgbmModelPath"],
    extraParams["torchModelPath"],
    snapStreamName,
    false
)

实时模拟仿真成交表

以上为实时模拟仿真成交明细结果中的关键字段，其中表字段含义分别为：

通过执行以下代码清理流计算环境：

clearStreamEnv()
each(undef{, SHARED}, [tradeStreamName, snapStreamName])

选取全天 100、200、500、1000 支股票的真实快照数据和成交数据，在关闭与开启低时延参数两种情况下，测试 TSE（流数据日级时间序列引擎：createDailyTimeSeriesEngine）和 RSE（响应式状态引擎： createReactiveStateEngine）的平均计算耗时（单位：微秒）。括号中指标表示开启低时延参数后的耗时与关闭该参数时耗时的比值。

可以观察到，当开启低时延参数时，所有流计算引擎的计算耗时均明显小于关闭该参数的表现。

以下代码完成 Orca 分钟因子框架的初始化：先导入成交与快照样本数据，再建同结构流表，并配置因子函数、并行度、模型路径与结果落库位置。

use SnapTradeStream::CreateTable
use SnapTradeStream::utils
use SnapTradeStream::KlineFactors
use SnapTradeStream::OrcaFrame

// 导入数据
colNames, colTypes = getTradeTickColDefs()
tradeData = select * from loadText("/data/trade.csv", ,table(colNames, colTypes)) order by tradeTime

colNames, colTypes = getSnapshotColDefs()
snapshotData = select * from loadText("/data/snapshot.csv", ,table(colNames, colTypes)) order by tradeTime


// 创建流表
tradeStreamName = "tradeStreamOrca"
unsubAndDropAll(tradeStreamName)
sche = schema(tradeData).colDefs
colNames, colTypes = sche.name, sche.typeString
try{
    share(streamTable(100:0, colNames, colTypes), tradeStreamName)
}catch(ex){print(ex)}

snapStreamName = "snapshotStreamOrca"
unsubAndDropAll(snapStreamName)
sche = schema(snapshotData).colDefs
colNames, colTypes = sche.name, sche.typeString
try{
    share(streamTable(100:0, colNames, colTypes), snapStreamName)
}catch(ex){print(ex)}

// 定义因子函数字典
factorFuncDict = {
    "tradeTSE": getTrade1MTSFactor, // 获取成交数据时序聚合引擎指标的函数
    "tradeRSE": getTrade1MReactiveFactor, // 获取成交数据响应式状态引擎指标的函数
    "snapTSE": getSnap1MTSFactor, // 获取快照数据时序聚合引擎指标的函数
    "snapRSE": getSnap1MReactiveFactor, // 获取快照数据响应式状态引擎指标的函数
    "mergedRSE": minDerivedKlineFactors // 获取衍生数据响应式状态引擎指标的函数
}

// 定义并行度参数字典
parallelDict = {
    "tradeTSE": 5,
    "tradeRSE": 5,
    "snapTSE": 5,
    "snapRSE": 5,
    "mergedRSE": 5
}

// 定义模型路径
extraParams = {
    "torchModelPath": "/data/pytorch_model.pt",
    "lgbmModelPath": "/data/lightgbm_model.txt"
}

// 定义持久化库表路径
factorDbname = "dfs://factor"
factorTbname = "minDerivedFactorsTbOrca"

factorFuncDict各键含义如下：

parallelDict各键含义如下：

extraParams各键含义如下：

通过调用模块中的buildOrcaFrame函数搭建 Orca 流计算框架，参数介绍参考 6.3.4 小节，该函数主要做以下流程操作：

1. 清理现有环境，包括流图、流表、流引擎及相关订阅，确保无残留数据干扰。

2. 通过 Orca 接口创建流计算框架，实现基于原始快照数据和成交数据的分钟频衍生因子指标生成。

3. 订阅这些分钟频衍生指标数据，利用模型进行实时推理，生成交易信号，随后构造委托订单并插入模拟撮合订单管理引擎，最终实现实时模拟交易结果的生成。

buildOrcaFrame(factorDbname, factorTbname, tradeStreamName, snapStreamName, factorFuncDict, parallelDict, extraParams)

通过回放历史数据，触发 Orca 流计算与模拟仿真交易全流程，实时输出模拟交易结果。

replay(
    [tradeData, snapshotData],
    [objByName(tradeStreamName, true), objByName(snapStreamName, true)],
    `tradeTime,
    `tradeTime,
    100,
    false
)

通过执行以下代码清理 orca 环境

clearOrcaEnv()
each(undef{, SHARED}, [tradeStreamName, snapStreamName])