股票中高频 CTA 策略

策略结合快照与逐笔成交两类行情数据，分别构建趋势、动量及成交量因子与过滤指标：

1. MACD 因子（基于快照行情）：将快照行情中的价格趋势变化转化为可用于策略过滤的趋势因子，用于判断标的是否处于更有利于做多的中短期趋势环境。该因子重点关注 MACD 柱线相对零轴的方向切换：

   • 金叉：当 MACD 柱线由负值转为正值，即从零轴下方上穿到零轴上方时，通常代表上升趋势开始确立，可触发多头候选信号。

   • 死叉：当 MACD 柱线由正值转为负值，即从零轴上方下穿到零轴下方时，通常代表上升趋势结束或行情可能反转，可触发多头平仓信号。

2. CCI 因子（基于逐笔成交行情）：将过去 30 秒内成交价相对均价的偏离程度转化为可用于策略判断的动量特征，用于刻画短期价格强弱与反弹/延续机会：

   • 超买区间突破：过去 30 秒内，CCI 因子值从下向上突破 +100 线，表明多头动能强劲。

   • 超卖反弹突破：过去 30 秒内，CCI 由 -100 线下方回升并向上突破 -100 线，表明短期价格可能从超卖区域开始反弹。

3. 高频成交量指标（基于逐笔成交行情）：统计过去 30 秒内的累计成交量，用于辅助验证突破信号的有效性。

本案例以 A 股市场具有高流动性的股票（如科创板股票等）为交易标的。在回测演示中，采用单只标的连续 20 个交易日的 Level 2 快照与逐笔成交高频行情数据（共计约 24.1 万条记录）；策略支持在盘前根据实际需求动态更换待监控的标的池。

在实际量化投研中，交易标的池通常需根据历史截面数据动态构建，并设置严格的流动性门槛、剔除 ST 与日内停复牌股票，确保标的具备充足的交易可行性。

策略采用普通股票现金账户，仅进行多头开仓与平仓操作，不涉及融券做空或杠杆交易。买入开仓采用固定股数（本例单笔固定买入 500 股），卖出平仓则按全部持仓及在途挂单差额进行清仓委托。订单依赖 DolphinDB 模拟撮合引擎（Matching Engine Simulator），按照交易所“价格优先、时间优先”的机制进行高精度逐笔限价单撮合。

策略按以下流程进行信号生成与订单执行：

1. 买入开仓条件：

   • 处于 MACD 金叉状态，标的已进入待买入池（buyList）；

   • 在逐笔成交行情中，满足以下两个动能条件之一：

     1. 过去 30 秒内 CCI 从下向上突破 +100 线，且过去 30 秒累计成交量大于 10,000 股；

     2. 过去 30 秒内 CCI 从下向上突破 -100 线。

   • 防重入风控：检查当前标的多头持仓量（longPosition <= 0）且无在途买单（openOrders.size() == 0）时，提交 500 股限价买单。

2. 卖出平仓条件：

   • 快照行情检测到 MACD 死叉，或标的位于待平仓池（sellList）；

   • 撤销该标的所有未成交买单；

   • 查询当前持仓与未成交平仓单，对剩余可用持仓提交限价卖单平仓；若单次未完全成交，在后续回调中继续委托直至完全平仓。

本示例使用一个案例数据文件 stockCTAData.csv。该文件采用混合宽表格式，同时包含快照行情与逐笔成交记录，并通过 sourceType 字段区分不同类型的行情数据：

• sourceType = 2：快照行情，包含最新价、盘口价格与数量、涨跌停价、前收盘价等字段，用于驱动 onSnapshot 回调。

• sourceType = 1：逐笔成交，包含成交价、成交量、订单类型、买卖订单号等字段，用于驱动 onTick 回调。

• sourceType = 0：逐笔委托（Order）。本策略示例中会在 onTick 中过滤该类型数据，不参与买入信号计算。

下载附录中的 stockCTAData.csv 文件作为回测案例数据，使用时需将脚本中的数据加载路径修改为本地实际路径。

// 导入 csv 样例数据
col = [`symbol, `symbolSource, `timestamp, `sourceType, `orderType, `price, `qty, `buyNo, `sellNo, `direction, `channelNo, `seqNum, `lastPrice, `upLimitPrice, `downLimitPrice, `totalBidQty, `totalOfferQty, `bidPrice, `bidQty, `offerPrice, `offerQty, `prevClosePrice, `reserve1, `reserve2]
type = ["STRING", "INT", "TIMESTAMP", "INT", "INT", "DOUBLE", "LONG", "LONG", "LONG", "INT", "INT", "LONG", "DOUBLE", "DOUBLE", "DOUBLE", "LONG", "LONG", "DOUBLE[]", "LONG[]", "DOUBLE[]", "LONG[]", "DOUBLE", "DOUBLE", "DOUBLE"]
sch = table(col as name, type as type)
stockCTAData = loadText("/path/to/stockCTAData.csv", schema=sch)

在 DolphinDB 中，对于需要跨行或滑动窗口保留中间计算状态的因子计算，需使用 @state 装饰器定义状态函数。Backtest 插件在初始化阶段通过 Backtest::subscribeIndicator 订阅这些状态函数，并在行情回放过程中自动进行流式增量计算，将指标结果注入到对应的事件回调中。

本策略定义了 3 个技术指标函数：

• macd：基于快照行情中的最新价 lastPrice 计算 MACD 柱线值，并返回当前值与上一周期值，用于在 onSnapshot 中判断 MACD 金叉和死叉。

• myCCI：基于逐笔成交价格计算过去 30 秒滑动窗口内的 CCI 指标，并返回当前值与上一时刻值，用于在 onTick 中判断短期动量突破。

• tradeVol30s：统计过去 30 秒内的逐笔成交量，用于辅助验证 CCI 突破信号的有效性。

@state
def macd(lastPrice, short_=240, long_=520, m=180) {
    dif = ewmMean(lastPrice, span=short_, adjust=false) - ewmMean(lastPrice, span=long_, adjust=false)
    dea = ewmMean(dif, span=m, adjust=false)
    macd = (dif - dea) * 2
    return round(macd, 4), round(prev(macd), 4)
}

@state
def myCCI(close, t, orderType) {
    high = tmmax(t, iif(orderType==0, close, NULL), 30s)
    low = tmmin(t, iif(orderType==0, close, NULL), 30s)
    tp = iif(orderType==0, (high + low + close) \ 3, NULL)
    avg_ = tmavg(t, iif(orderType==0, close, NULL), 30s)
    tmmad = tmavg(t, abs(tp - avg_), 30s)
    cci = ffill((tp - avg_) \ (0.015 * tmmad))
    return round(cci, 4), round(prev(cci), 4)
}

@state
def tradeVol30s(qty, time, orderType){
    return tmsum(time, iif(orderType==0, qty, NULL), 30s)
}

本策略由 5 个核心函数协同组成：

• getOpenQty 负责统计在途挂单未成交量；

• initialize 负责在回测初始化阶段通过流式引擎订阅快照与逐笔指标，并预置策略上下文；

• beforeTrading 负责在每日开盘前重置当日待买入池（buyList）与待平仓池（sellList），支持动态更换当日股票池；

• onSnapshot 作为快照行情回调，负责监控中期趋势（MACD 金叉/死叉）并管理多轮清仓与撤单；

• onTick 作为逐笔成交行情回调，负责捕捉微观动能（CCI 突破与成交量放量）并在严格的防重入风控下执行买入开仓。

在中高频多源行情回测中，快照与逐笔行情以统一的时间轴向前推进。策略通过 context 字典在两个回调之间跨流传递标的状态：快照负责趋势把关（维护 buyList 候选池），逐笔负责精准开仓（从 buyList 触发开仓并移出），从而实现多频率特征的无缝协同。

未成交量查询函数：getOpenQty

getOpenQty(openOrders) 用于遍历当前标的在途的所有未成交委托，累加计算剩余待成交的总股数。在中高频限价平仓过程中，该函数用于精确计算剩余可用持仓，防止因上一笔委托尚未完全撮合成交而引发超额重复报单。

• openOrders：ANY VECTOR（字典数组），由 Backtest::getOpenOrders 查询返回的在途未成交订单列表。每个元素为一个包含订单详情的字典，其中 openQty 字段表示该笔委托当前尚未成交的剩余股数。

函数返回一个整型标量 qty，表示所有在途挂单的未成交股数总和。若当前无挂单，则返回 0。

def getOpenQty(openOrders){
    qty = 0
    for(i in openOrders){
        qty = i.openQty + qty
    }
    return qty
}

初始化函数：initialize

initialize(mutable context) 是回测引擎的初始化回调函数，在回测引擎启动前仅执行一次。

• context：可变 DICTIONARY，由回测引擎统一注入并维护的全局上下文对象。包含引擎实例句柄（context["engine"]）以及策略运行所需的各类动态参数。

initialize主要负责两项任务：

1. 指标订阅：调用 Backtest::subscribeIndicator 分别为快照数据和逐笔成交数据订阅所需计算的指标：

   • 快照数据：使用 macd(lastPrice, 240, 520, 180)[0] 和 macd(lastPrice, 240, 520, 180)[1] 分别订阅当前与上一周期的 MACD 柱线值，用于判断金叉与死叉；

   • 逐笔成交数据：使用 myCCI(price, timestamp, orderType)[0] 和 myCCI(price, timestamp, orderType)[1] 订阅过去 30 秒滑动窗口的 CCI 动量因子，使用 tradeVol30s(qty, timestamp, orderType) 订阅过去 30 秒的累计成交量。

2. 初始化策略参数：在 context 字典中设置单笔买入数量 buyVol（默认 500 股）。待买入列表 buyList 和待卖出列表 sellList 不在 initialize 中初始化，而是在 beforeTrading 盘前回调以及创建引擎前的 context 配置中进行初始化与每日重置。

def initialize(mutable context){
	// 通过Backtest::setUniverse可以更换当日股票池，
	// 如Backtest::setUniverse(context["engine"],["688088.XSHG","688157.XSHG","688208.XSHG"])
	print("initialize")
	// 订阅快照行情的指标

	d=dict(STRING,ANY)
	d["macd"]=<macd(lastPrice,240,520,180)[0]>
	d["prevMacd"]=<macd(lastPrice,240,520,180)[1]>
	Backtest::subscribeIndicator(context["engine"], "snapshot", d)

	d=dict(STRING,ANY)
	d["cci"]=<myCCI(price,timestamp,orderType)[0]>
	d["prevcci"]=<myCCI(price,timestamp,orderType)[1]>
	d["tradeVol30s"]=<tradeVol30s(qty,timestamp,orderType)>
	Backtest::subscribeIndicator(context["engine"], "trade", d)

	// 记录每日统计量
	context["buyVol"] =500	
}

盘前回调函数：beforeTrading

beforeTrading(mutable context) 是每日开盘前触发的生命周期回调，适合放在 initialize 之后、onSnapshot 与 onTick 之前定义。其主要作用是在每个交易日开始前重置跨日状态，避免上一交易日遗留的待买入/待平仓标的影响当日信号判断；如需按日动态调整交易标的池，也可以在该回调中调用 Backtest::setUniverse。

• context：可变 DICTIONARY，策略全局上下文对象。这里重置 buyList 与 sellList，并在需要时维护每日可交易标的池。

def beforeTrading(mutable context){
	//每日盘前回调函数
	//1、通过context["tradeDate"]可以获取当日；
	//2、通过Backtest::setUniverse可以更换当日股票池，
	//如Backtest::setUniverse(context["engine"],["688088.XSHG","688157.XSHG","688208.XSHG"])
	//print ("beforeTrading: "+context["tradeDate"])
	context["buyList"] =array(SYMBOL,0)
	context["sellList"] =array(SYMBOL,0)

}

快照行情回调函数：onSnapshot

onSnapshot(mutable context, msg, indicator) 是快照行情（Level 2 Snapshot）的事件驱动回调函数。每当引擎回放到一条快照行情时触发，主要负责监控 MACD 趋势转向、筛选多头候选标的并执行平仓及撤单管理。

• context：可变 DICTIONARY，包含引擎句柄、全局变量及当前交易时间等；

• msg：DICTIONARY，当前标的的最新快照行情记录，包含 symbol（股票代码）、receivedTime（接收时间戳）、lastPrice（最新价）、买卖十档盘口等字段；

• indicator：DICTIONARY，底层流式引擎实时计算并注入的快照指标结果，直接提供 indicator.macd（当前 MACD 柱线）与 indicator.prevMacd（上一时点 MACD 柱线）。

执行逻辑：

1. MACD 金叉识别：当 prevMacd < 0 且 macd > 0 时，表明上升趋势确立。若当前无多头持仓且标的尚未在待买池中，将其加入 context["buyList"]；

2. MACD 死叉与平仓处理：当 prevMacd > 0 且 macd < 0（出现死叉）或标的仍处于待平仓列表 sellList 中时：

   • 先调用 Backtest::cancelOrder 撤销该标的所有尚未成交的买单，阻断逆势加仓；

   • 查询当前多头持仓 pos 与在途未成交平仓单总量 openQty，计算净待平仓量 pos - openQty；

   • 若 pos-openQty > 0，以微低于最新价的价格（如 lastPrice - 0.02）提交限价卖出平仓单；

   • 将标的维持在 sellList 中，确保在后续快照中持续追踪补发委托，直至持仓完全清空。

def onSnapshot(mutable context, msg,indicator){
	////msg 可以为字典，最新时刻的tick数据
	if(second(context["tradeTime"])<09:40:00 or second(context["tradeTime"])>14:57:00){
		return 
	}
	//记录买入卖出信号
	buyList=context["buyList"]
	sellList=context["sellList"]
	istock=msg.symbol
	macd=indicator.macd
	prevMacd=indicator.prevMacd
	if(prevMacd<0 and macd>0){//MACD 指标出现金叉
		pos=Backtest::getPosition(context.engine,istock).longPosition
		if((pos<=0) and (not istock in buyList)){	
			buyList=buyList.append!(istock)
		}
	}
	else if((prevMacd>0 and macd<0) or (istock in sellList)){//MACD 指标死叉,平仓
		//未成交的订单进行撤单
		Backtest::cancelOrder(context.engine,istock,,"buy")
		pos=Backtest::getPosition(context.engine,istock).longPosition
		openQty=getOpenQty(Backtest::getOpenOrders(context.engine,istock,,"close"))
		if(pos-openQty>0){//卖出持仓
			Backtest::submitOrder(context.engine,(
			istock,context["tradeTime"], 5, round(msg.lastPrice-0.02,3), pos-openQty, 3),"close")
		}
		if(not istock in sellList){
			sellList=sellList.append!(istock)
		}
	}
	context["buyList"]=buyList	
	context["sellList"]=sellList			
}

逐笔成交回调函数：onTick

onTick(mutable context, msg, indicator) 是逐笔成交行情（Level 2 Trade）的高频事件回调函数。每当引擎接收到一笔逐笔成交数据时触发，负责捕捉微观动量突破并在严格的防重入机制下执行买入开仓。

• context：可变 DICTIONARY，策略全局上下文对象；

• msg：DICTIONARY，当前逐笔成交记录，包含 symbol（股票代码）、timestamp（成交时间戳）、price（成交价）、qty（成交量）及 orderType（订单类型）等；

• indicator：DICTIONARY，底层流式引擎实时计算并注入的逐笔高频指标，包含 indicator.cci（当前 CCI 值）、indicator.prevcci（上一时点 CCI 值）及 indicator.tradeVol30s（过去 30 秒累计成交量）。

执行逻辑：

1. 候选池过滤：仅对处于 context["buyList"]（已通过快照 MACD 金叉确认趋势）中的标的进行逐笔计算；

2. 微观动量突破验证：满足以下任一条件时生成买入信号：

   • 超买顺势突破：过去 30 秒内 CCI 从下向上突破 +100 线，且过去 30 秒累计成交量大于 10,000 股（tradeVol30s > 10000）；

   • 超卖反弹突破：过去 30 秒内 CCI 从下向上突破 -100 线；

3. 防重入与持仓风控拦截：

   • 检查持仓量：若 pos > 0（已有持仓），直接退出，避免超仓；

   • 检查在途委托：若 opens.size() > 0（上一笔买单仍在撮合队列排队），直接退出，防止高频 Tick 密集到达时连续重复发单；

4. 提交开仓订单：

   • 调用 Backtest::submitOrder 按照当前成交价提交 500 股限价买单；

   • 下单完成后立即将标的从 buyList 移除，确保单次金叉周期内仅触发一次开仓。

def onTick(mutable context, msg,indicator){
	if(second(context["tradeTime"])<09:40:00 or second(context["tradeTime"])>14:57:00){
		return 
	}
	buyList=context["buyList"]
	sellList=context["sellList"]

	if(msg.sourceType==0){
		return 
	}
	istock=msg.symbol
	if(istock in buyList){
		cci=indicator.cci
		prevcci=indicator.prevcci
		tradeVol30s=indicator.tradeVol30s
		buyFlag=false
		//指标从下向上突破+100线进入超买区间时，过去30s内的成交量大于10000股是买入
		if(prevcci<100. and cci>=100. and tradeVol30s>10000){
			buyFlag=true
		}
		//指标从下向上突破-100线，买入
		if(prevcci<-100. and cci>=-100. ){
			buyFlag=true
		}
		if(buyFlag==false){
			return 
		}
		//有持仓
		pos=Backtest::getPosition(context["engine"],istock).longPosition
		if(pos>0){
			return
		}
		//有在途单
		opens=Backtest::getOpenOrders(context["engine"],istock,,"buy")
		if(opens.size()>0){
			return
		}
		Backtest::submitOrder(context["engine"], (
		istock,msg.timestamp, 5, round(msg.price,2), context["buyVol"], 1),"buy")
		buyList=buyList[buyList!=istock]
		sellList=sellList[sellList!=istock]
	}
	context["buyList"]=buyList
	context["sellList"]=sellList
}

在创建回测引擎前，需要配置回测基础信息字典 userConfig、策略全局上下文字典 context 以及回测事件驱动回调字典 callbacks。

策略上下文配置（context）：context 字典用于在各个生命周期回调函数（beforeTrading、onSnapshot、onTick 等）之间维护和传递策略全局状态与控制变量。本示例中主要维护三个变量：buyVol 表示单笔买入开仓股数，固定为 500 股；buyList 表示待买入候选池，快照出现 MACD 金叉时加入标的，逐笔触发开仓后移出；sellList 表示待平仓监控池，快照出现 MACD 死叉时加入标的，持仓完全清空后移出。

回测引擎配置（userConfig）：userConfig 字典包含基础资金、交易周期、逐笔撮合延时及高频行情格式等核心参数：

回调函数字典（callbacks）：将前文定义的核心回调函数及生命周期钩子注册到 callbacks 字典中，并在创建回测引擎时传入。示例代码保留了 onOrder、onTrade、afterTrading 和 finalize 等空回调，主要用于展示完整生命周期；实际策略中如不需要处理订单回报、成交回报或盘后逻辑，也可以省略对应注册项。

// step 2：策略配置与引擎创建
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["latency"]= 50
userConfig["cash"]= 100000000
///手续费,
userConfig["commission"]= 0.0
//印花税,
userConfig["tax"]= 0.0
//行情选择逐笔
userConfig["dataType"]= 6
//是否订阅逐笔行情
userConfig["enableSubscriptionToTickQuotes"]= true
//是否需要获取委托订单在行情中的位置
userConfig["outputQueuePosition"]= 1


strategyName="stockCTA"
callbacks=dict(STRING,ANY)
callbacks["initialize"] = initialize
callbacks["beforeTrading"] = beforeTrading
callbacks["onSnapshot"]=onSnapshot 
callbacks["onTick"]=onTick 
callbacks["onOrder"]=onOrder
callbacks["onTrade"]=onTrade
callbacks["afterTrading"]=afterTrading
callbacks["finalize"]=finalized
context=dict(STRING,ANY)
context["buyVol"]=500
context["buyList"] =array(SYMBOL,0)
context["sellList"] =array(SYMBOL,0)
userConfig["context"] = context

上述代码首先构造回测配置字典 userConfig，设置回测区间、股票策略类型、逐笔事件驱动模式、初始资金、费率、逐笔撮合延迟及行情格式等基础参数；随后定义回调函数字典 callbacks 注册策略的完整生命周期回调；最后构造策略上下文字典 context，用于保存单笔委托股数及在跨流回调中共享的待买入/待卖出标的池，并将其挂载到userConfig["context"] 中。

在回测过程中，模拟撮合引擎主要负责逐笔行情下的订单簿撮合，并模拟订单在各档位中的排队等待与成交优先级；Backtest 插件则负责按照时间序列回放多源行情流、调度触发各生命周期回调、流式增量计算技术指标，并全程维护账户资金、持仓与交易税费。创建回测引擎前，需确保服务器端已安装并加载与当前 DolphinDB Server 版本匹配的两个插件。

创建回测引擎时可通过 jit 参数控制是否启用 JIT 模式：

try { Backtest::dropBacktestEngine(strategyName) } catch(ex) { print ex }
// 非 JIT 模式
engine = Backtest::createBacktester(strategyName, userConfig, callbacks, false)
// 开启 JIT 优化模式（需要 3.00.2.1 及以上 JIT Server 版本）
// engine = Backtest::createBacktester(strategyName, userConfig, callbacks, true)

数据加载完成后，调用 Backtest::appendQuotationMsg 将排序好的行情表数据一次性写入回测引擎中。引擎会根据数据中的 timestamp（时间戳）严格按时序推进模拟时钟，并依据 sourceType（数据源类型）分发数据并触发策略：sourceType = 1（逐笔成交）触发 onTick 回调与动量开仓逻辑；sourceType = 2（快照行情）触发 onSnapshot 回调与趋势平仓逻辑；同时驱动模拟撮合引擎更新订单簿状态与撮合成交。

本案例使用 CSV 静态数据表执行离线回测，Backtest::appendQuotationMsg(engine, stockCTAData) 同步返回后即表示行情数据已全部注入并完成回测；无需额外调用 appendEndMarker。当前引擎实例会保留在变量 engine 中，供后续提取回测指标与交易明细。

Backtest::appendQuotationMsg(engine,  stockCTAData)

回测执行完成后，可通过 Backtest 插件提供的一系列分析接口提取回测指标概览、订单与成交明细、持仓演变和每日组合权益等数据；

tradeDetails=Backtest::getTradeDetails(engine)
//查询当前的未成交（未完成）订单列表
openOrders=Backtest::getOpenOrders(engine)
//每日持仓
dailyPosition=Backtest::getDailyPosition(engine)
// 可用资金
enableCash=Backtest::getAvailableCash(engine)
//未成交订单明细
openOrders=tradeDetails[tradeDetails.orderStatus==-3]
//日组合指标展示
totalPortfolios=Backtest::getDailyTotalPortfolios(engine)
//回测结果综合展示
returnSummary=Backtest::getReturnSummary(engine)

也可以利用 DolphinDB 强大的向量化 SQL 对回测明细进行自定义聚合分析。

lastprice= select last(lastPrice)  as closePrice from  stockCTAData where sourceType==2 group by timestamp.date() as tradeDate,symbol as SecurityID
dailyReport=select sum(tradeQty) as volume,sum(tradePrice*tradeQty) as amount  from Backtest::getTradeDetails(engine)
where OrderStatus in [0,1]  group by symbol as SecurityID, direction as BSFlag,  sendTime.date() as tradeDate
dailyReport=select tradeDate as tradeDate,sum(iif(BSFlag==1 ,volume,0)) as buyVolume, 
sum(iif(BSFlag==1,amount,0)) as buyAmount, 
sum(iif(BSFlag==3 ,volume,0))as closeVolume,
sum(iif(BSFlag==3 ,amount,0)) as closeAmount
from dailyReport group by SecurityID,tradeDate