DolphinDB 回测引擎使用和性能优化攻略

回测引擎以插件的形式提供服务，其逻辑架构如图 1-1 所示。回测的主要工作流程包括：

1. 回测引擎接收按时间先后顺序回放的数据流，引擎内部把数据流分发给模拟撮合引擎和相应的行情回调函数；
2. 行情回调函数处理策略逻辑并发送委托订单；
3. 回测引擎根据策略委托订单进行风控管理；
4. 通过风险控制的委托订单发送给模拟撮合引擎进行订单撮合；
5. 回测引擎根据订单成交情况实时进行持仓和资金统计管理，策略回测结束返回策略的收益、成交明细等信息。

使用回测引擎时，通常包含图 1-2 所示的 5 个步骤。首先，回测引擎提供多个事件函数，包括策略初始化、每日盘前和盘后回调函数、逐笔、快照和 K 线行情的回调函数、委托和成交回报函数等。用户可以在策略初始化中定义指标，在其他相应的回调函数中编写自定义策略。其次，可以对策略的行情源、资金、订单延时和成交比例等进行配置。再次，根据策略和配置，创建相应的回测引擎。接着，回放数据源，执行回测引擎。最后，调用引擎接口获取回测的结果。

下面通过简单的示例来展示回测引擎的使用流程。

编写自定义策略逻辑：

@state
def pctChg(lastPrice, prevClosePrice){
        return lastPrice\prevClosePrice - 1
}
def initialize(mutable context){
        // 初始化回调函数
        print("initialize")
        // 订阅快照行情的指标
        d = dict(STRING,ANY)
        d["pctChg"] = <pctChg(lastPrice, prevClosePrice)>
        Backtest::subscribeIndicator(context["engine"], "snapshot", d) 
        context["maxPos"] = 500
}
def beforeTrading(mutable context){
        // 每日盘前回调函数
        // 通过 context["tradeDate"] 可以获取当日；
        print("beforeTrading: " + context["tradeDate"])
        // 通过 backtest::setUniverse 可以更换当日股票池
        Backtest::setUniverse(context["engine"], ["000001.XSHE"])
        context["open"] = dict(STRING,BOOL)
}
def onSnapshot(mutable context, msg, indicator){        
    // 查询目前该的持仓
    pos = Backtest::getPosition(context["engine"], msg.symbol)
    longPos = pos.longPosition
    if (indicator.pctChg > 0.01 and longPos <= context.maxPos and
    context["open"][istock] != true){
                Backtest::submitOrder(context["engine"], 
                        (msg.symbol, context["tradeTime"], 5, msg.offerPrice[0], 100, 1), "buy")
                context["open"][istock] = true
    }
}
def onOrder(mutable context, msg){}
def onTrade(mutable context, msg){}
def afterTrading(mutable context){}
def finalize(mutable context){}

配置策略的行情源、资金、成交比例参数，其他参数详情可以见回测插件的接口说明。

// 配置参数
config = {
        startDate: 2022.04.11,// 回测开始日期
        endDate: 2022.04.11, // 回测结束日期
        strategyGroup: `stock,
        frequency: 0,
        cash: 100000000,  // 策略初始资金
        commission: 0.00015,
        tax: 0.001,
        dataType: 1,// 回测行情，dataType = 1 表示快照行情
        msgAsTable: false,
        "context": {
                maxPos:0,
                open: {
                        A: true
                } 
        }
}

在指定策略回调函数后，即可通过 createBacktester 接口创建回测引擎：

// 创建回测引擎
callbacks = {
        initialize: initialize,
        beforeTrading: beforeTrading,
    onSnapshot: onSnapshot,
    onOrder: onOrder,
    onTrade: onTrade,
        afterTrading: afterTrading,
    finalize: finalize
}
name = "BackTester1"
engine = Backtest::createBacktester(name, config, callbacks, false)

按照回测引擎接口说明文档获取对应的行情数据后，即可通过 appendQuotationMsg 接口执行策略回测：

// 数据写入，执行策略回测。其中 messageTable 为符合引擎要求的数据，具体要求参见回测引擎的接口说明文档
Backtest::appendQuotationMsg(engine, messageTable)

回测结束后，可通过不同的接口获取回测的结果：

// 获取回测结果
Backtest::getDailyPosition(engine)// 获取每日持仓数据
Backtest::getDailyTotalPortfolios(engine)// 获取策略每日权益指标
Backtest::getReturnSummary(engine)// 获取策略的收益概述
Backtest::getContextDict(engine)// 获取策略的逻辑上下文
Backtest::getTradeDetails(engine)// 获取交易明细表

DolphinDB 回测引擎基于 DolphinDB 高性能的分布式存储和计算架构开发，支持通过 DolphinScript、Python 或 C++ 完成中高频策略的研发和测试。由于回测引擎中 JIT 优化的实现依赖 DolphinDB 原生的 JIT 支持，因此目前只有 DolphinScript 支持 JIT 优化。在开启 JIT 优化时，策略回调函数的实现不能用默认参数和部分应用；策略中的全局变量的定义需要在 config 的 context 配置项中声明。

DolphinDB 回测引擎在 3.00.2.1 JIT 版本支持 JIT 优化，只需在创建回测引擎时把 createBacktester 函数的 jit 参数设置为 true 即可。

Backtest::createBacktester(name, config, eventCallbacks, [jit=false], [securityReference])

在非 JIT 版本的 server 中，回测引擎不支持 JIT 优化，jit 设置为 true 时会报错：

[PLUGIN::BACKTEST] This version of server doesn't support JIT function..

回测引擎的策略事件函数提供了策略全局变量 context，行情消息 msg 和行情指标 indicator。

• context 为字典，若要使用 JIT 优化，context 中用户定义的全局变量不支持表，其他所有 DolphinDB 的数据类型均支持。

• msg 为字典。高频策略回测行情事件回调函数 onSnapShot 和 onTick 中，通过 msg.lastPrice 和 msg.price 可以获取最新价。onBar 行情回调函数中，msg 字典的 key 是标的代码，value 值为该标的对应的行情，如通过 msg[istock].close 和 msg[istock].low 获取最新收盘价和最新低价。

• 行情指标 indicator 的数据类型同相应行情的 msg 的数据类型一致。

策略实现过程需要实时保存某些行情，可以通过如下方式。本例中，在回测行情事件回调函数 onSnapShot 中实时保存最高价和最低价指标。

def onSnapshot(mutable context, msg, indicator){
        context["highPrice"] = max(msg.highPrice, context["highPrice"])
        context["lowPrice"] = min(msg.lowPrice, context["lowPrice"])
} 
// context 的生命周期不会随着每一次回调函数运行完成而结束；
// context 存储的变量会始终存在整个回调过程中，除非在其他回调函数中进行修改

为了提高性能，每一次事件函数调用时，msg 用的是同一个对象实例，所以在策略函数中使用时，需要注意，不能保存起来留作后用。因此我们不建议按照下述例子，在事件函数中保存 msg。

def onBar(mutable context, msg, indicator){
        context["msg"] = msg
}

DolphinDB 的中高频回测引擎默认依赖 DolphinDB 的模拟撮合引擎，模拟撮合引擎是仿照交易所按照价格优先、时间优先顺序撮合的高精度订单撮合系统。如果用户的策略实现逻辑不想依赖高精度的订单撮合系统，比如部分中低频策略，用户希望订单以策略委托的订单价格进行成交，可以把创建引擎时的配置项中的 matchingMode 设置为 3。

config["matchingMode"] = 3

该设置表示策略的委托订单按照用户的委托价格成交，同时也有涨跌停的限制。即买入时以涨停价买入会失败，卖出时以跌停价卖出会失败。

DolphinDB 回测引擎不仅支持复杂的实时指标计算，还支持用户将预先计算完成的指标与行情数据一并回放，实现策略逻辑。

Backtest::subscribeIndicator(engine, marketDataType, metrics)

indicatorDict = dict(STRING, ANY) 
indicatorDict["mavg"] = <mavg(lastPrice, 20)>

congfig["enableIndicatorOptimize"] = true // 开启指标计算优化
config["dataRetentionwindow"] = "ALL" // 保存全部行情数据

@state
def myAtr(high, low, close, m = 14, n = 10) {
        prevClosePrice = prev(close)
        tr = rowMax([high - low,abs(high - prevClosePrice),abs(low - prevClosePrice)])
        atr = ema(tr, m)
        mAtr = mavg(atr,n)
        return atr,mAtr
}
def initialize(mutable contextDict){
        d = dict(STRING, ANY)
        d["ATR"] = <myAtr(signal[0], signal[1], signal[2], 14, 10)[0]>
        d["mATR"] = <myAtr(signal[0], signal[1], signal[2], 14, 10)[1]>
        d["RSI"] = <ta::rsi(signal[2], 14)>
        // 其中， signal 来自行情数据，是三列行情数据构成的数组向量：
        // fixedLengthArrayVector([backwardFactorHighPrice,backwardFactorLowPrice,backwardFactorClosePrice])
        Backtest::subscribeIndicator(contextDict["engine"], "kline", d)
}

update quoteTB set signal = fixedLengthArrayVector([ATR, prev(ATR), RSI]) context by symbol

def onBar(mutable contextDict, msg, indicator){
        keys = msg.keys()
        for(i in keys){ 
                price = msg[i].close
                signal = msg[i].signal
                atr = signal[0]
                matr = signal[1]
                rsi = signal[2]
        }
}

def onBar(mutable context, msg, indicator){
        closeList = context["closeList"] // 实时记录历史的收盘价
        lowPriceList = context["lowPriceList"] // 实时记录历史的最低价
        highPriceList = context["highPriceList"] // 实时记录历史的最高价
        keys = msg.keys()
        for(i in keys){
                istock = msg[i].symbol
                price = msg[i].close
                if( type(closeList[istock]) == VOID){
                        closeList[istock] = array(DOUBLE, 0, 14)
                        lowPriceList[istock] = array(DOUBLE, 0, 14)
                        highPriceList[istock] = array(DOUBLE, 0, 14)
                }
                closeList[istock] = closeList[istock].append!(price)
                lowPriceList[istock] = lowPriceList[istock].append!(msg[i].low)
                highPriceList[istock] = highPriceList[istock].append!(msg[i].high)
                if(closeList[istock].size() >= 14){
                        n = size(closeList[istock])
                        closeList[istock] = closeList[istock][n - 14:]
                        lowPriceList[istock] = lowPriceList[istock][n - 14:]
                        highPriceList[istock] = highPriceList[istock][n - 14:]
                }
                atr_ = myatr(highPriceList[istock], lowPriceList[istock], 
            closeList[istock], 14, 10)
                atr = atr_[0]
                matr = atr_[1]
                rsi = RSI(closeList[istock], 14) 
        }
        context["closeList"] = closeList
        context["lowPriceList"] = lowPriceList
        context["highPriceList"] = highPriceList
}

当进行策略回测时，DolphinDB 的回测引擎支持行情以流式的形式进行回放，其中批量行情回放能显著提升性能。本例子中，使用快照数据作为行情，实现当日最新的价格相比昨日收盘价上涨 1%时，买入100 股的操作，最大持仓不超过 500，每日开仓最多开仓一次。

def onSnapshot(mutable context, msg, indicator){        
        // 查询目前的持仓
        pos = Backtest::getPosition(context["engine"], istock)
        if (indicator.pctChg > 0.01 and pos.longPosition <= context["maxPos"] and
        context["open"][istock] != true){
                    // Backtest::submitOrder 是回测引擎提供的下单接口
                    Backtest::submitOrder(context["engine"], 
                    (msg.symbol, context["tradeTime"], 5, msg.offerPrice[0], 100, 1), "buy")
                    context["open"][istock] = true
        }
}

基于上述策略，模拟回测数据插入，如下分别比对 10 万条快照行情下，逐条写入和批量写入的性能。其中批量写入相比逐条写入，性能提升了至少 5 倍左右。因此在模拟回测阶段，建议使用批量写入。行情批量写入时可以批量读取缓存数据，提高行情解析的速度。

n = 100000
t = table(take("000001.XSHE",n) as symbol, take("XSHE",n) as symbolSource, 
take(concatDateTime(2022.04.11, transFreq(10:00:00..11:00:00, "3S").distinct()), n) as timestamp, 
randUniform(7.0, 10.0, n) as lastPrice, randUniform(9.0, 10.0, n) as upLimitPrice, 
randUniform(7.0, 8.0, n) as downLimitPrice, take(10000, n) as totalBidQty, take(10000, n) as totalOfferQty,
take(arrayVector([10], [6.9, 6.8, 6.7, 6.6, 6.5, 6.4, 6.3, 6.2, 6.1,6.0]), n) as bidPrice,
take(arrayVector([10], [800, 900, 1000, 1100, 1200, 1000, 1000, 1000, 1000, 1000]), n) as bidQty, take(arrayVector([10],
[7.1, 7.2, 7.3, 7.4, 7.5, 7.6, 7.7, 7.8, 7.9, 8.0]), n) as offerPrice,
take(arrayVector([10], [1000, 1000, 1000, 1000, 1000, 1000, 1000, 1000, 1000, 1000]), n) as offerQty,
take(arrayVector([1], [double()]), n) as signal, randUniform(6.0, 7.0, n) as prevClosePrice)

messageTable.append!(t)
timer{Backtest::appendQuotationMsg(engine, messageTable)}
for (i in 0:messageTable.size()){
        tmp = select * from messageTable where rowNo(symbol) = i
        t = evalTimer(Backtest::appendQuotationMsg{engine2, tmp})
        tsRun = tsRun.append!(t)
}
try{Backtest::dropBacktestEngine(strategyName)}catch(ex){print ex}
tsRun = array(DOUBLE, 0, 10)
engine2 = Backtest::createBacktester(strategyName, userConfig, callbacks,, basicInfo)
// 每10条写入
a = 0
for (i in 1..messageTable.size()){
        if(a >= size(messageTable)){continue}
        tmp = select * from messageTable where rowNo(symbol) >= a and rowNo(symbol) < (a + 10*i)
        t = evalTimer( Backtest::appendQuotationMsg{engine2,tmp})
        tsRun = tsRun.append!(t)
        a = a + 10*i
}
print(sum(tsRun))

在等号赋值逻辑上， DolphinDB 与 Python 并不相同。 Python 的 '=' 表示一份引用，简单地将一个变量指向现有的变量，这个操作无需额外的内存分配，也不需要复制数据；对于 DolphinDB， '=' 的赋值逻辑取决于对象和标志位。在创建字典时，默认使用浅拷贝，因此在利用 '=' 做赋值操作时，会复制一份对象；如果设置 transient 标志，则变成深拷贝，递归地复制对象下的所有子对象；在回测回调函数中，行情数据 msg 默认使用深拷贝方式，等号赋值会极大地降低回测速度，因此需要尽量避免使用等号直接引用整个 msg 或 msg 字段（ indicator 除外）。

在大部分的 CTA 策略中，高精度的订单模拟撮合和实时风险控制是确保策略成功和有效执行的关键因素。策略的主要逻辑可能是基于分钟等聚合的 K 线行情，订单的成交依赖高精度的订单撮合引擎或者策略止盈止损依赖实时的 tick 级高频行情。DolphinDB 可以实现回测引擎以实时的 tick 行情作为回测引擎行情输入，同时按配置的指定频率触发 onBar 行情回调，策略可以基于高频行情回调函数 onSnapshot 进行实时止盈止损等操作。具体涉及的回测引擎的配置项为 callbackForSnapshot 和 frequency。

另外，回测引擎内置了实时止盈止损的算法订单，配置项 enableAlgoOrder 设置为true 时，可以开启算法订单：

Backtest::submitOrder(engine, msg, [label=""], [orderType = 0])

msg 为元组，包含的成员变量分别是标的代码、交易所代码、时间、订单类型、委托订单价格、止损价、止盈价、委托订单数量、买卖方向、滑点、委托订单有效性、委托订单到期时间。算法订单的类型 orderType 的取值包括：

• 0 ：默认值，表示一般订单

• 5：限价止盈止损订单

• 6：市价止盈止损订单

• 8：双边报价订单（仅期货、期权支持）

• 9：自动下单，此时买卖方向只支持设置为 1 或 2

• 10：TWAP (Time-Weighted Average Price) 算法单，将大额订单按照时间间隔均匀拆分成多个小订单执行

• 11：VWAP (Volume-Weighted Average Price) 算法单，将大额订单按照成交量分布比例拆分成多个小订单执行

• 12：移动止盈止损订单

具体实现案例：

下面以实现布林带突破和 RSI 作为开仓信号的外汇 CTA 趋势策略为例，介绍高频行情实时合成行情的实现。该策略以每小时进行布林带和 RSI 的指标计算，以 tick 行情实时进行风险监控进行止盈止损，下面为具体的策略逻辑：

• 做多开仓：当前 RSI 大于 70 且布林带正在上移并打破上轨

• 做空开仓：当前 RSI 小于 30 且布林带正在下移并打破下轨

如果存在多头或空头委托订单，需根据 tick 行情判断是否撤单、止盈止损。

代码实现：

首先订阅指标：

use ta
def initialize(mutable context){
        print("initialize")
        d = dict(STRING,ANY)
        d["rsi"] = <ta::rsi(lastPrice, 11)>
        d["bhigh"] = <ta::bBands(lastPrice, 20, 2, 2, 0)[0]>
        d["bmid"] = <ta::bBands(lastPrice, 20, 2, 2, 0)[1]>
        d["blow"] = <ta::bBands(lastPrice, 20, 2, 2, 0)[2]>
        Backtest::subscribeIndicator(context["engine"], "snapshot_kline", d)
        
}

策略逻辑实现：当 RSI 大于 70 且布林带正在上移并打破上轨买入，当 RSI 小于 30 且布林带正在下移并打破下轨卖出，并通过提交算法订单进行实时止盈止损操作。

def onBar(mutable context, msg,indicator){
        istock=msg.keys()[0]
        if(indicator[istock].rsi <= 0){ return }
        position=Backtest::getPosition(context["engine"],istock)
        longpos = position.longPosition
        shortpos = position.shortPosition
        if(indicator[istock].rsi >70. and msg[istock].offerPrice[0]>indicator.bhigh and msg[istock].close>msg[istock].open){
                if(longpos <1){
                        orderId=Backtest::submitOrder(context.engine, (istock,msg[istock].symbolSource ,
                        context.barTime,5, round(msg[istock].offerPrice[0],5), 
                        msg[istock].offerPrice[0] -context.sl+context.Slippage , msg[istock].offerPrice[0]+ context.tp+context.Slippage, 
                        2, 1,context.Slippage, 0, context.barTime+36000000),"openBuy", 5)
                        return
                }
        }
        if(indicator[istock].rsi<30. and msg[istock].bidPrice[0]<indicator.blow and msg[istock].close<msg[istock].open){
                if(shortpos <1){
                        orderId=Backtest::submitOrder(context.engine, (istock,msg[istock].symbolSource,
                        context.barTime,5, round(msg[istock].bidPrice[0],5), 
                        msg[istock].bidPrice[0]+context.sl-context.Slippage, 
                        msg[istock].bidPrice[0] - context.tp-context.Slippage, 2, 2, context.Slippage , 0, 
                        context.barTime+36000000),"openSell", 5)
                        return
                }
        }
}

性能测试：以上示例实现，共 29,317,091 条行情数据，168 笔订单，耗时 11 秒。

在开启 JIT 优化时，策略回调函数的实现不能用默认参数和部分应用；策略中的全局变量的定义需要在 config 的 context 配置项中声明；JIT 优化不支持直接对字典嵌套赋值。

def onSnapshot(mutable context, msg, indicator = NULL){}// 该函数JIT会编译不通过

def onSnapshot(mutable context, msg, indicator, myParam){}// 为策略回调函数增加参数
eventCallbacks["onSnapshot"] = onSnapshot{,,,myParam}

context = dict(STRING, ANY)
context["myParam"] = myParam
config["context"] = context

config = dict(STRING, ANY)
config["startDate"] = 2023.01.01 // 回测开始日期
config["endDate"] = 2024.10.30 // 回测结束日期
config["strategyGroup"] = "stock" // 策略类型
// 配置策略全局变量
context = dict(STRING, ANY)
context["initPrice"] = 0.
context["feeRatio"] = 0.00002
context["alpha"] = 0.01
context["M"] = 0.01
context["highPrice"] = dict(STRING, ANY)
config["context"] = context

JIT: Data type does not match when assign value to attribute of ...

def onBar(context, msg, indcator){
    context["test"][istock] = msg.close
}

 Attribute [("test","000001.XSHE")] doesn't exist

def onBar(context, msg, indcator){
      temp = context["test"]
      temp[istock] = msg.close
      temp["test"]=temp
}

def fun() {
    return 1., 2.
}
def onSnapshot(context,msg,indicator){
    // 不支持的方法：
    a,b = fun()
    // 报错 codegen for statement type MULTIASSIGN not supported: a, b fun()
    
    // 支持的方法：
    tmp = fun()
    a = tmp[0]
    a = tmp[1]
}