动态网格交易、科创板做市、股票 CTA……DolphinDB 中高频策略回测实例之股票篇

def initialize(mutable contextDict){
    //通过Backtest::setUniverse可以更换当日股票池，
    //如Backtest::setUniverse(contextDict["engine"],["688088.XSHG","688157.XSHG","688208.XSHG"])
    print("initialize")
    // 网格策略参数   
    // 初始价格
    contextDict["initPrice"] = dict(SYMBOL,ANY)
    // 网格间距（百分数）
    contextDict["alpha"]=0.01
    // 回落间距（百分数）
    contextDict["beta"] =0.005
    // 每格交易金额
    contextDict["M"] = 100000
    contextDict["baseBuyPrice"] = dict(SYMBOL,ANY)
    contextDict["baseSellPrice"] = dict(SYMBOL,ANY)
    contextDict["lowPrice"]=dict(SYMBOL,ANY)
    contextDict["highPrice"]=dict(SYMBOL,ANY)
    contextDict["N"]=dict(SYMBOL,ANY)
    // 手续费费率
    contextDict["feeRatio"] = 0.00015
    //记录每日统计量
    contextDict['dailyReport' ]= table(1000:0,[`SecurityID,`tradeDate,`closePrice,`buyVolume,
    `buyAmount,`sellVolume,`sellAmount,`PosExposureVolume,`PosExposureValue,`dailyEarning],
    [SYMBOL,DATE,DOUBLE,DOUBLE,DOUBLE,DOUBLE,DOUBLE,DOUBLE,DOUBLE,DOUBLE])
    //日志
    contextDict["log"]=table(10000:0,[`tradeDate,`time,`info],[DATE,TIMESTAMP,STRING])
}

def updateBaseBuyPrice(istock,lastPrice,basePrice,mutable baseBuyPrice,mutable baseSellPrice,mutable N,mutable highPrice,mutable lowPrice,alpha,n,mode=0){
    //根据最新价和最新的基准价更新网格线和最高或者最低价
    baseBuyPrice[istock]=basePrice*(1-alpha)
    baseSellPrice[istock]=basePrice*(1+alpha)
    N[istock]=n
    if(mode==0){
        //买入、卖出等初始化
        lowPrice[istock]=0.
        highPrice[istock]=10000.
    }
    else if(mode==1){
        //下跌,更新下网格线 
        lowPrice[istock]=lastPrice
        highPrice[istock]=10000.        
    }
    else if(mode==2){
        //上涨，更新上网格线
        lowPrice[istock]=0.
        highPrice[istock]=lastPrice         
    }
}

def onSnapshot(mutable contextDict, msg){
    if(second(contextDict["tradeTime"])<09:30:00 or second(contextDict["tradeTime"])>14:57:00){
        return 
    }
    alpha=contextDict["alpha"]
    // 回落间距（百分数）
    beta=contextDict["beta"]
    M=contextDict["M"] 
    for (istock in msg.keys()){
        lastPrice=msg[istock]["lastPrice"]
        ////设置初始价
        if(not istock in contextDict["initPrice"].keys()){
            contextDict["initPrice"][istock]=lastPrice
            updateBaseBuyPrice(istock,lastPrice,lastPrice, contextDict["baseBuyPrice"],
            contextDict["baseSellPrice"] , contextDict["N"], contextDict["highPrice"], contextDict["lowPrice"],alpha,1,0)
        }
        init_price=contextDict["initPrice"][istock]
        if(lastPrice<=contextDict["baseBuyPrice"][istock]){
            //下跌,更新下网格线
            n=floor(log(lastPrice\init_price)\log(1-alpha))+1   
            if(n>contextDict["N"][istock]){
                newBasePrice=init_price*pow((1-alpha),n)
                updateBaseBuyPrice(istock,lastPrice,newBasePrice, contextDict["baseBuyPrice"], contextDict["baseSellPrice"], 
                contextDict["N"], contextDict["highPrice"], contextDict["lowPrice"],alpha,n,1)
            }
        }
        else if(lastPrice>contextDict["baseSellPrice"][istock]){
            //上涨，更新上网格线
            n=floor(log(lastPrice\init_price)\log(1+alpha))+1
            if(n>contextDict["N"][istock]){
                newBasePrice=init_price*pow((1+alpha),n)
                updateBaseBuyPrice(istock,lastPrice,newBasePrice, contextDict["baseBuyPrice"], 
                contextDict["baseSellPrice"] , contextDict["N"], contextDict["highPrice"], contextDict["lowPrice"],alpha,n,2)
            }
        }

        if(contextDict["lowPrice"][istock]>0. and lastPrice>contextDict["lowPrice"][istock]*(1+beta)){
            //买入
            qty=int(contextDict["N"][istock]*M\lastPrice)/100*100
            Backtest::submitOrder(contextDict["engine"], (istock,contextDict["tradeTime"] , 5, lastPrice, qty, 1),"buy")        
            contextDict["initPrice"][istock]=lastPrice
            updateBaseBuyPrice(istock,lastPrice,lastPrice, contextDict["baseBuyPrice"], 
            contextDict["baseSellPrice"] , contextDict["N"], contextDict["highPrice"], contextDict["lowPrice"],alpha,1,0)
        }
        else if(contextDict["highPrice"][istock]<10000. and lastPrice<contextDict["highPrice"][istock]*(1-beta)){
            //卖出
            qty=Backtest::getPosition(contextDict["engine"],istock).todayBuyVolume[0]
            if(qty<=0){
                continue
            }
            qty=min([int(contextDict["N"][istock]*M\lastPrice)/100*100,qty])
            Backtest::submitOrder(contextDict["engine"], (istock,contextDict["tradeTime"] , 5, lastPrice, qty, 3),"sell")
            contextDict["initPrice"][istock]=lastPrice
            updateBaseBuyPrice(istock,lastPrice,lastPrice, contextDict["baseBuyPrice"], contextDict["baseSellPrice"] , 
            contextDict["N"], contextDict["highPrice"], contextDict["lowPrice"],alpha,1,0)
        }
        //实时更新最低或者最高价
		if(contextDict["lowPrice"][istock]>0){
			contextDict["lowPrice"][istock]=min([contextDict["lowPrice"][istock],lastPrice])
		}
		if(contextDict["highPrice"][istock]<10000.){
			contextDict["highPrice"][istock]=max([contextDict["highPrice"][istock],lastPrice])
		}

startDate=2021.01.01
endDate=2021.01.04
userConfig=dict(STRING,ANY)
userConfig["startDate"]=startDate
userConfig["endDate"]=endDate
userConfig["strategyGroup"]= "stock"
userConfig["frequency"]= 0
userConfig["cash"]= 100000000       
userConfig["commission"]= 0.0
userConfig["tax"]= 0.0
userConfig["dataType"]= 2
userConfig["msgAsTable"]= false
strategyName="gridStrategy" 

engine = Backtest::createBacktestEngine(strategyName, userConfig,,initialize,
 beforeTrading,,onSnapshot,onOrder,onTrade,afterTrading,finalized)

Backtest::appendQuotationMsg(engine,gridStrategyData)

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

 def runBacktest_snapshot(userConfig,initialize, beforeTrading,onTick,onSnapshot,
    onOrder,onTrade,afterTrading,finalize,strategyName,startDate,endDate,codes){
        //快照+成交行情进行策略回测
        engine =  Backtest::createBacktestEngine(strategyName, userConfig,,
        initialize, beforeTrading,,onSnapshot,onOrder,onTrade,afterTrading,finalize)
        go
        tradeDates=getMarketCalendar("CFFEX",startDate, endDate)
        for( idate in tradeDates){  
            messageTable=select iif(substr(SecurityID,6,2)=="SH",
            strReplace(SecurityID,"SH",".XSHG"),
            strReplace(SecurityID,"SZ",".XSHE")) as symbol, 
            iif(substr(SecurityID,6,2)=="SH","XSHG","XSHE") as symbolSource,
            dateTime as timestamp,lastPrice, upperLimitPrice as upLimitPrice,
            lowerLimitPrice as downLimitPrice,long(totalBidQty) as totalBidQty,
            long(totalOfferQty) as totalOfferQty,bidPrice,bidOrderQty as bidQty,
            OfferPrice as offerPrice,offerOrderQty as offeQty,
            fixedLengthArrayVector([lastPrice]) as signal,tradePrice2 as tradePrice,
            tradeQty2 as tradeQty,0.0 as prevClosePrice from  
            loadTable("dfs://Level2_tickTrade",`tickTradeTable)
            where SecurityID in codes and  date(dateTime) between startDate:endDate and
            second(dateTime) between startTime:endTime order by timestamp
            Backtest::appendQuotationMsg(engine,messageTable)
        }
        //增加回测结束标志
        temp=select * from messageTable where timestamp=max(timestamp) limit 1
        update temp set symbol="END"
        Backtest::appendQuotationMsg(engine,temp)
        return engine
    }

def runBacktestParallelMode(userConfig,initialize, beforeTrading,onTick,
    onSnapshot,onOrder,onTrade,afterTrading,finalize,startDate,endDate,codes,
    parallelMode=4,nSize=1){
        dates=getMarketCalendar("CFFEX",startDate, endDate)
        destroyAllBacktestEngine()
        jobs= array(STRING, 0, 10)
        //把股票分成 n 份，并行
        cuts=cut(codes,nSize)
        for( icodes in cuts){
            strategyName=strReplace("backtest"+concat(icodes,""),".","")        
            jobId=submitJob(strategyName,strategyName+"job",runBacktest_snapshot,
            userConfig,initialize, beforeTrading,onTick,onSnapshot,onOrder,onTrade,
            afterTrading,finalize,strategyName,min(dates),max(dates),icodes)
            jobs=jobs.append!(jobId)    
        }
        //获取每个引擎的回测结果，并返回，具体代码请见附件
    }

dailyReport,tradeOutputTable,engines,removejobs=runBacktestParallelMode(userConfig,
        initialize, beforeTrading,,onSnapshot,onOrder,onTrade,afterTrading,finalize,
        startDate,endDate,codes,4,n)

def initialize(mutable contextDict,userParam){
	print("initialize")
	// 报价模块参数
	contextDict["maxBidAskSpread"]=userParam["maxBidAskSpread"]//双边报价最大的买卖价差
	...
}

def onTrade(mutable contextDict,trades){
	 // trades 为字典列表
	///成交主推回调
	///在这里处理对冲模块
	hedgeOffsetAmount=contextDict["hedgeAmount"] // 对冲偏移金额
	for (itrade in trades){
		...
		///下达对冲单
		Backtest::submitOrder(contextDict["engine"], (stock, itrade.tradeTime, 5, price, vol, bsFlag),"hedgeOrder")
	}
}

userConfig=dict(STRING,ANY)
//逐笔行情
userConfig["dataType"]=0
//策略参数设置
userParam=dict(STRING,FLOAT)
userParam["maxBidAskSpread"]=0.03//双边报价最大的买卖价差
userParam["quotAmount"]=100000  ///报价金额
...
engine = Backtest::createBacktestEngine(strategyName, userConfig,,
initialize{,userParam}, beforeTrading,,onSnapshot,onOrder,onTrade,afterTrading,finalize)
///开始执行回测
Backtest::appendQuotationMsg(engine,select * from tickHqTable)

// step 3:并行策略回测
jobs=array(STRING, 0, 10)
codes=(exec distinct(SecurityID) from loadTable("dfs://Level 2_tl","snapshot") where SecurityID like "68%")[:10]
for(hedgeAmount in [0.01,0.02,0.03,0.04]){
	for(maxVolatility_1m in [0.03,0.04,0.05]){
	//策略参数设置
		userParam=dict(STRING,FLOAT)
		userParam["maxBidAskSpread"]=0.03//双边报价最大的买卖价差
		userParam["quotAmount"]=100000  ///报价金额
		// 对冲模块参数
		userParam["hedgeAmount"]=hedgeAmount//对冲偏移金额
		//风控参数
		userParam["maxPos"]=2000//最大单边持仓上限
		userParam["maxLossAmount"]=200000 //当日亏损上限金额
		userParam["maxVolatility_1m"]=maxVolatility_1m //最新一分钟
		strategyName="marketMakingStrategy"
		strategyName=strategyName+strReplace(concat(each(def(a,b):a+"_"+string(b),userParam.keys(),userParam.values())),".","")
		jobId=submitJob(strategyName,strategyName+"job",runBacktest_tick,userConfig,initialize{,userParam}, beforeTrading,,
		onSnapshot,onOrder,onTrade,afterTrading,finalize,strategyName,startDate,endDate,codes)
		jobs=jobs.append!(jobId)
	}
}

def initialize(mutable contextDict){
	print("initialize")
	//订阅快照行情的指标
	d=dict(STRING,ANY)
	d["macd"]=<macd(lastPrice,240,520,180)[0]>
	d["prevMacd"]=<macd(lastPrice,240,520,180)[1]>
	Backtest::subscribeIndicator(contextDict["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(contextDict["engine"], "trade", d)
}

def onSnapshot(mutable contextDict, msg){
	////msg可以为字典或表，最新时刻的tick数据
	if(second(contextDict["tradeTime"])<09:40:00 or second(contextDict["tradeTime"])>14:57:00){
		return 
	}
	buyList=contextDict["buyList"]
	///记录买入卖出信号处理
	for (istock in msg.keys()){
		macd=msg[istock]["macd"]
		prevMacd=msg[istock]["prevMacd"]
		if(prevMacd<0 and macd>0){
			pos=Backtest::getPosition(contextDict["engine"],istock).longPosition[0]
			if((pos<0) and (not istock in buyList)){	
				buyList=buyList.append!(istock)
			}
		}
		else if(prevMacd>0 and macd<0){
			pos=Backtest::getPosition(contextDict["engine"],istock).longPosition[0]
			openQty=sum(nullFill(Backtest::getOpenOrders(contextDict["engine"],istock,,"close").openQty,0))
			if(pos-openQty>0){
				//卖出
				pos=pos-openQty
				price=round(msg[istock]["lastPrice"]-0.02,3)
				Backtest::submitOrder(contextDict["engine"], (istock,contextDict["tradeTime"], 5, price, pos, 3),"close")
			}
		}
	}
	contextDict["buyList"]=buyList		
}