科创板做市策略

盘口价格取自快照中的买一价 bidPrice[0] 和卖一价 offerPrice[0]，作为双边报价的计算输入。maxBidAskSpread 为报价价差参数，示例值为 0.03；quotAmount 为报价金额参数，示例值为 100,000。报价数量由报价金额和报价价格计算，并与 200 股比较取较大值。具体报价计算见 onSnapshot 中的委托提交逻辑。

hedgeAmount 为对冲价格偏移量，单位为元，示例值为 0.02。报价订单成交后，买开方向以成交价加该偏移量提交卖平委托，卖开方向以成交价减该偏移量提交买平委托。

maxVolatility_1m 函数用于计算价格波动指标，按最近 n 条行情的 lastPrice 最高值除以最低值再减 1，n 默认为 20。策略上下文中的参数 volatilityThreshold 为该指标的比较阈值，示例值为 0.05。

持仓控制量由 todayBuyVolume 与 todaySellVolume 的差值计算，风控判断使用其绝对值。maxPos 为持仓上限参数，示例值为 2,000；exMaxPos 保存进入持仓超限处理的股票代码，具体加入和移除条件见交易规则。

当日损益取自 Backtest::getTodayPnl 返回的 todayPnl。maxLossAmount 为当日损益比较阈值，示例值为 200,000。在 onSnapshot 的双边报价分支中，仅当当日亏损未超过最大容忍限额（即 todayPnl > -maxLossAmount）时，才继续检查持仓控制量。

本例使用 2023 年 2 月 1 日至 2 月 28 日的科创板股票行情，覆盖 20 个交易日、10 只股票。主回测使用文件中的全部标的，并行参数寻优示例选取其中一只股票。

股票代码为 688125.XSHG、688133.XSHG、688161.XSHG、688266.XSHG、688291.XSHG、688296.XSHG、688376.XSHG、688486.XSHG、688508.XSHG 和 688677.XSHG。

快照回调先查询标签为 MakingMarketOrder 的未完成报价订单。不存在此类订单且价格波动满足条件时，继续判断当日损益和持仓条件，并使用买卖一档价格计算双边报价。

在上述报价、价格波动和当日损益条件满足时，若持仓控制量的绝对值达到 maxPos，则将股票代码加入 exMaxPos，并撤销标签为 hedgeOrder 的对冲委托；若股票已在 exMaxPos 中且持仓控制量低于 maxPos 的四分之一，则移除相应代码。对处于 exMaxPos 中的股票，若仍有未完成订单，则等待后续行情；无未完成订单时，根据持仓方向和数量提交标签为 closePosition 的平仓委托。

成交回调仅处理标签为 MakingMakertOrder 的成交。对冲数量以对应方向的持仓减去昨仓，再扣除同方向未完成对冲委托数量计算；不足 200 股时不提交对冲委托。对冲委托使用 hedgeOrder 标签。

下载附录中的 marketMakingStrategyData.zip，解压得到 marketMakingStrategyData.csv。数据文件须位于 DolphinDB Server 可读取的位置，代码中的文件路径应与实际位置一致。

行情采用逐笔委托、逐笔成交与快照合并的宽表结构。CSV 共包含 2,876,184 行，其中 2,876,183 行为行情数据，最后一行为 END 结束标记。行情中的委托记录为 1,619,594 行，成交记录为 683,693 行，快照记录为 572,896 行，分别以 sourceType 为 0、1、2 区分。文件中的 timestamp 按非递减顺序排列，各标的行情按时间交错组织。

udfLoadMakingStrategyData 先读取文件结构，再指定列的具体类型，最后加载数据。其中 bidPrice、offerPrice 使用 DOUBLE[]，bidQty、offerQty 使用 LONG[]，以保留各档盘口数据。

def udfLoadMakingStrategyData(filePath) {
    columnNames = [
        "symbol", "symbolSource", "timestamp", "sourceType", "orderType", "price", "qty", "buyNo",
        "sellNo", "direction", "channelNo", "seqNum", "lastPrice", "upLimitPrice", "downLimitPrice",
        "totalBidQty", "totalOfferQty", "bidPrice", "bidQty", "offerPrice", "offerQty", "prevClosePrice",
        "reserve1", "reserve2"
    ]
    colTypes = [
        "STRING", "INT", "TIMESTAMP", "INT", "INT", "DOUBLE", "LONG", "LONG", "LONG", "INT", "INT",
        "LONG", "DOUBLE", "DOUBLE", "DOUBLE", "LONG", "LONG", "DOUBLE[]", "LONG[]", "DOUBLE[]",
        "LONG[]", "DOUBLE", "DOUBLE", "DOUBLE"
    ]
    schemaDict = dict(columnNames, colTypes)
    schema = extractTextSchema(filePath)
    for (col in columnNames) {
        update schema set type = schemaDict[col] where name = col
    }
    return loadText(filePath, schema = schema)
}
marketMakingStrategyDataW = udfLoadMakingStrategyData("marketMakingStrategyData.csv")

select count(*) from marketMakingStrategyDataW where date(timestamp) == 2023.02.28

maxVolatility_1m 计算价格波动指标；initialize 通过 Backtest::subscribeIndicator 订阅快照指标；beforeTrading 在每日盘前初始化持仓超限股票列表。

@state
def maxVolatility_1m(lastPrice, n=20) {
    // 计算最近 n 条行情的最高价与最低价之比减 1。
    return mmax(lastPrice, n) / mmin(lastPrice, n) - 1.
}

def initialize(mutable context) {
    print("initialize")
    // 订阅快照价格波动指标。
    d = dict(STRING, ANY)
    d["maxVolatility_1m"] = <maxVolatility_1m(lastPrice, 20)>
    Backtest::subscribeIndicator(context["engine"], "snapshot", d)
}

def beforeTrading(mutable context) {
    // 每日盘前清空持仓超限股票列表。
    context["exMaxPos"] = array(STRING, 0)
}

onSnapshot 根据未完成报价、价格波动、当日损益和持仓条件执行报价及平仓操作。以下保留完整的行情回调实现。

def onSnapshot(mutable context, msg, indicator) {
    quotAmount = context["quotAmount"]
    maxBidAskSpread = context["maxBidAskSpread"]
    hedgeAmount = context["hedgeAmount"]
    maxPos = context["maxPos"]
    maxLossAmount = context["maxLossAmount"]
    volatilityThreshold = context["volatilityThreshold"]
    bjFlag = "MakingMakertOrder"
    exMaxPos = context["exMaxPos"]
    istock = msg.symbol
    opens = Backtest::getOpenOrders(context["engine"], istock, , bjFlag)

    // 无未完成报价且价格波动低于阈值时，继续检查损益和持仓条件。
    if (opens.size() == 0 and indicator.maxVolatility_1m < volatilityThreshold) {
        pnl = Backtest::getTodayPnl(context["engine"], istock).todayPnl
        if (pnl < maxLossAmount) {
            pos = Backtest::getPosition(context["engine"], istock)
            pos = abs(double(pos.todayBuyVolume) - double(pos.todaySellVolume))
            // 持仓控制量达到上限时，撤销对冲委托并记录股票。
            if (pos >= maxPos) {
                Backtest::cancelOrder(context["engine"], istock, , "hedgeOrder")
                if (not istock in exMaxPos) {
                    exMaxPos = exMaxPos.append!(istock)
                }
            }
            // 持仓控制量低于恢复阈值时，移除超限标记。
            if (istock in exMaxPos and pos < maxPos \ 4) {
                exMaxPos = exMaxPos[exMaxPos != istock]
            }
            if (pos < maxPos \ 4 or not (istock in exMaxPos)) {
                askPrice0 = msg.offerPrice[0]
                bidPrice0 = msg.bidPrice[0]
                // 买卖一档价格均为正时，计算并提交双边报价。
                if (all([askPrice0, bidPrice0] > 0)) {
                    midPrice = askPrice0 + bidPrice0
                    buyPrice = min(midPrice - maxBidAskSpread \ (2 + maxBidAskSpread), bidPrice0)
                    sellPrice = max(midPrice + maxBidAskSpread \ (2 + maxBidAskSpread), askPrice0)
                    qty = max(ceil(quotAmount \ [buyPrice, sellPrice]), 200)
                    Backtest::submitOrder(context.engine, (
                        istock, context.tradeTime, 5, buyPrice, qty[0], 1), "MakingMakertOrder")
                    Backtest::submitOrder(context.engine, (
                        istock, context.tradeTime, 5, sellPrice, qty[1], 2), "MakingMakertOrder")
                }
            }
        }
    }

    context["exMaxPos"] = exMaxPos
    // 对超限股票执行平仓；仍有未完成订单时等待后续行情。
    if (istock in exMaxPos) {
        if (size(Backtest::getOpenOrders(context.engine, istock, , )) > 0) {
            return
        }
        pos = Backtest::getPosition(context.engine, istock)
        pos = double(pos.todayBuyVolume) - double(pos.todaySellVolume)
        if (pos >= 200) {
            askPrice0 = msg.offerPrice[0]
            // 排除非正价格和空值。
            if (any([askPrice0] <= 0) or any(isNull([askPrice0]))) {
                return
            }
            Backtest::submitOrder(context.engine, (
                istock, context.tradeTime, 5, askPrice0, pos, 3), "closePosition")
        } else if (pos <= -200) {
            bidPrice0 = msg.bidPrice[0]
            // 排除非正价格和空值。
            if (any([bidPrice0] <= 0) or any(isNull([bidPrice0]))) {
                return
            }
            Backtest::submitOrder(context.engine, (
                istock, context.tradeTime, 3, bidPrice0, abs(pos), 4), "closePosition")
        }
    }
}

onOrder 保留委托回报回调，函数体中未编写处理逻辑。getOpenQty 汇总指定方向的未完成委托数量，onTrade 使用该数量计算待提交的对冲数量，并按 hedgeAmount 设置对冲价格。

def onOrder(mutable context, orders) {
    // 保留委托回报回调。
}

def getOpenQty(openOrders, bsflag) {
    // 汇总指定方向的未完成委托数量。
    qty = 0
    for (i in openOrders) {
        if (i.direction == bsflag and i.openQty > 0) {
            qty = qty + i.openQty
        }
    }
    return qty
}

def onTrade(mutable context, trades) {
    hedgeOffsetAmount = context["hedgeAmount"]
    // 仅处理报价订单的成交回报。
    if (trades.label != "MakingMakertOrder") {
        return
    }
    bsFlag = 3
    stock = trades.symbol
    pos = Backtest::getPosition(context.engine, stock)
    vol = pos.longPosition - pos.lastDayLongPosition
    if (trades.direction == 2) {
        bsFlag = 4
        vol = pos.shortPosition - pos.lastDayShortPosition
    }
    openQty = getOpenQty(Backtest::getOpenOrders(context.engine, stock, , "hedgeOrder"), bsFlag)
    if (isNull(openQty)) {
        openQty = 0
    }
    // 扣除同方向未完成的对冲数量；不足 200 股时不提交委托。
    vol = vol - openQty
    if (vol < 200) {
        return
    }
    if (bsFlag == 3) {
        price = trades.tradePrice + hedgeOffsetAmount
    } else {
        price = trades.tradePrice - hedgeOffsetAmount
    }
    Backtest::submitOrder(context.engine, (
        stock, trades.tradeTime, 5, price, vol, bsFlag), "hedgeOrder")
}

afterTrading 为盘后回调，finalize 为回测结束回调。

def afterTrading(mutable context) {
    // 读取当前交易日期。
    tradeDate = context["tradeDate"]
}

def finalize(mutable context) {
    print('finalize')
}

userConfig 保存引擎配置，context 保存策略参数。主回测采用逐笔宽表输入，dataType 为 6，frequency 为 0，订单延迟 latency 为 50 毫秒；初始资金为 1 亿元，手续费率和印花税率均为 0。

startDate = 2023.02.01
endDate = 2023.02.28
userConfig = dict(STRING, ANY)
userConfig["startDate"] = startDate
userConfig["endDate"] = endDate
userConfig["strategyGroup"] = "stock"
userConfig["frequency"] = 0
userConfig["cash"] = 100000000
userConfig["dataType"] = 6
userConfig["latency"] = 50
userConfig["commission"] = 0.0
userConfig["tax"] = 0.0
userConfig["msgAsTable"] = false

// 设置报价、对冲和风险控制参数。
context = dict(STRING, ANY)
context["maxBidAskSpread"] = 0.03 // 报价价差参数。
context["quotAmount"] = 100000 // 报价金额。
context["hedgeAmount"] = 0.02 // 对冲价格偏移量。
context["maxPos"] = 2000 // 持仓数量阈值。
context["maxLossAmount"] = 200000 // 当日盈亏判断阈值。
context["volatilityThreshold"] = 0.05 // 最近 20 条快照的价格波幅阈值。
context["exMaxPos"] = array(STRING, 0)
userConfig["context"] = context

将初始化、盘前、快照、成交、委托、盘后和结束回调注册到 callbacks，调用 Backtest::createBacktester 创建引擎。第四个参数为 true，表示启用 JIT。创建前的清理语句会删除名为 marketMakingStrategy 的已有回测引擎。

strategyName = "marketMakingStrategy"
callbacks = dict(STRING, ANY)
callbacks["initialize"] = initialize
callbacks["beforeTrading"] = beforeTrading
callbacks["onSnapshot"] = onSnapshot
callbacks["onTrade"] = onTrade
callbacks["onOrder"] = onOrder
callbacks["afterTrading"] = afterTrading
callbacks["finalize"] = finalize

try {
    Backtest::dropBacktestEngine(strategyName)
} catch (ex) {
    print ex
}
engine = Backtest::createBacktester(strategyName, userConfig, callbacks, true)
go

将行情宽表 marketMakingStrategyDataW 传入 Backtest::appendQuotationMsg，由快照触发策略决策，逐笔行情推进订单撮合。timer 用于统计该语句的执行耗时。

附录 CSV 的最后一行已经包含 symbol 为 END、timestamp 为 2023.02.28T16:00:00.000 的结束标记。主回测直接回放完整文件即可完成结束处理；并行回测函数按日筛选指定股票行情后，会单独构造并追加 END 记录。

timer Backtest::appendQuotationMsg(engine, marketMakingStrategyDataW)

回放后获取成交明细、未完成订单、每日持仓、可用资金、每日组合指标及收益概述。结果查询代码如下。

tradeDetails = Backtest::getTradeDetails(engine)
// 当前未完成订单。
currentOpenOrders = Backtest::getOpenOrders(engine)
// 每日持仓。
dailyPosition = Backtest::getDailyPosition(engine)
// 可用资金。
enableCash = Backtest::getAvailableCash(engine)
// 从明细中筛选订单状态为 -3 的记录。
unfilledOrderRecords = tradeDetails[tradeDetails.orderStatus == -3]
// 每日组合指标。
totalPortfolios = Backtest::getDailyTotalPortfolios(engine)
// 回测绩效汇总。
returnSummary = Backtest::getReturnSummary(engine)

tradeDetails 保存订单交易明细，dailyPosition、enableCash、totalPortfolios 和 returnSummary 分别用于查看每日持仓、可用资金、每日权益指标和收益概述。currentOpenOrders 保存 Backtest::getOpenOrders 返回的当前未完成订单；unfilledOrderRecords 保存 tradeDetails 中 orderStatus 等于 −3 的历史未成交订单记录。