//use backtestUtils /**Implement the following CTA strategy logic based on level 2 snapshot data and tick-by-tick trades: 1. Calculate the MACD indicator from snapshot data. After the MACD indicator shows a golden cross, execute a buy when either of the following conditions is met: ------------1.1. Based on tick-by-tick trades, buy 500 shares when the CCI indicator calculated from trade prices over the past 30 seconds crosses above the +100 line into the overbought range and the trading volume over the past 30 seconds exceeds 50,000 shares. ------------1.2. Buy 500 shares when the CCI indicator calculated from trade prices over the past 30 seconds crosses above the -100 line. 2. Sell when the MACD indicator shows a death cross*/ @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) } def initialize(mutable context){ // Use Backtest::setUniverse to change the stock universe for the day // For example, Backtest::setUniverse(context["engine"],["688088.XSHG","688157.XSHG","688208.XSHG"]) print("initialize") // Subscribe to indicators calculated based on snapshot data d = dict(STRING, ANY) d["macd"] = d["prevMacd"] = Backtest::subscribeIndicator(context["engine"], "snapshot", d) d = dict(STRING, ANY) d["cci"] = d["prevcci"] = d["tradeVol30s"]= Backtest::subscribeIndicator(context["engine"], "trade", d) // Record daily statistics context["buyVol"] = 500 } def beforeTrading(mutable context){ // Daily pre-market callback // 1. Get the current trading date from context["tradeDate"]; // 2. Use Backtest::setUniverse to change the stock universe for the day // For example, Backtest::setUniverse(context["engine"],["688088.XSHG","688157.XSHG","688208.XSHG"]) // print ("beforeTrading: "+context["tradeDate"]) context["buyList"] = array(SYMBOL, 0) context["sellList"] = array(SYMBOL, 0) } def getOpenQty(openOrders){ qty = 0 for( i in openOrders){ qty = i.openQty + qty } return qty } def onSnapshot(mutable context, msg, indicator){ //msg is a dictionary containing the latest tick data if(second(context.tradeTime) < 09:40:00 or second(context.tradeTime) > 14:57:00){ return } //Record buy and sell signals. if(indicator.prevMacd < 0 and indicator.macd > 0){//The MACD indicator shows a golden cross pos=Backtest::getPosition(context.engine,msg.symbol).longPosition if((pos <= 0) and (not msg.symbol in context.buyList)){ context["buyList"] = context["buyList"].append!(msg.symbol) } } else if((indicator.prevMacd > 0 and indicator.macd < 0) or (msg.symbol in context.sellList)){//Close the position when the MACD indicator shows a death cross //Cancel unfilled orders. Backtest::cancelOrder(context.engine, msg.symbol, , "buy") pos = Backtest::getPosition(context.engine, msg.symbol).longPosition openQty = getOpenQty(Backtest::getOpenOrders(context.engine, msg.symbol,, "close")) if(pos-openQty>0){//Sell the position. Backtest::submitOrder(context.engine,( msg.symbol, context.tradeTime, 5, round(msg.lastPrice-0.02,3), pos-openQty, 3), "close") } if(not msg.symbol in context.sellList){ context["sellList"]=context["sellList"].append!(msg.symbol) } } } def onTick(mutable context, msg, indicator){ //print msg if(second(context.tradeTime) < 09:40:00 or second(context.tradeTime) > 14:57:00 or msg.sourceType==0){ return } if(msg.symbol in context.buyList){ buyFlag = false // Buy when the indicator breaks above the +100 line from below and enters the overbought zone, provided that trading volume over the past 30 seconds exceeds 10,000 shares. if(indicator.prevcci < 100. and indicator.cci >= 100. and indicator.tradeVol30s > 10000){ buyFlag =true } // Buy when the indicator breaks above the -100 line from below. if( indicator.prevcci < -100. and indicator.cci >= -100. ){ buyFlag = true } if(buyFlag == false){ return } // Has a position. pos=Backtest::getPosition(context.engine, msg.symbol).longPosition if(pos > 0){ return } //Has an open order. opens=Backtest::getOpenOrders(context.engine, msg.symbol, , "buy") if(opens.size() > 0){ return } // Backtest::submitOrder(context.engine, ( msg.symbol, context.tradeTime, 5, round(msg.price,2), context.buyVol, 1), "buy") context["buyList"] = context["buyList"][context.buyList != msg.symbol] context["sellList"] = context["sellList"][context.sellList != msg.symbol] } } def onOrder( mutable context,orders){ /*orders is a list of dictionaries orderId->11 symbol->688981.XSHG timestamp->2023.02.01T09:33:33.000 qty->5956 price->0 status->4;-1 (order rejected);0 (partially filled); 1 (fully filled); 2 (canceled); 4 (submitted) direction->3 tradeQty->0 tradeValue->0 label-> updateTime->2023.02.01T09:33:33.660*/ } def onTrade(mutable context,trades){ /** trades is a class orderId->11 symbol->688981.XSHG tradePrice->41.96 tradeQty->5956 tradeValue->249913.760000000009313 totalFee->0 totalVolume->5956 totalValue->249913.760000000009313 direction->3 tradeTime->2023.02.01T09:33:33.660 orderPrice->41.96 label->*/ } def afterTrading(mutable context){ /** Called daily after market close*/ //Daily post-market callback function for calculating post-market indicators } def finalized (mutable context){ // print('finalized') } // Step 2: Configure the strategy and create the engine startDate=2023.02.01 endDate=2023.02.28 userConfig=dict(STRING,ANY) userConfig["startDate"]=startDate userConfig["endDate"]=endDate //Strategy type, currently fixed to stock userConfig["strategyGroup"]= "stock" //Generate snapshots from tick-by-tick data at the specified frequency userConfig["frequency"]= 0 //Order latency userConfig["latency"]= 50 userConfig["cash"]= 100000000 //Commission userConfig["commission"]= 0.0 //Stamp duty userConfig["tax"]= 0.0 //Use tick-by-tick market data userConfig["dataType"]= 6 //Whether to subscribe to tick-by-tick market data userConfig["enableSubscriptionToTickQuotes"]= true //Whether to obtain the order's position in the market-data order queue userConfig["outputQueuePosition"]= 1 //Tick-by-tick data format: table or dict userConfig["msgAsTable"]= true 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"]=dict(STRING,ANY) context["sellList"]=dict(STRING,ANY) userConfig["context"] = context //userConfig["context"]= Context() //codes=(exec distinct(SecurityID) from loadTable("dfs://level2_tl","snapshot") where tradetime.date()=startDate and market=`sh and SecurityID like "6%")[:1] //// //prevClosePrice=select first(PreCloPrice) as prevClose from loadTable("dfs://level2_tl","snapshot") where //tradetime.date() between startDate:endDate and SecurityID in codes group by SecurityID as symbol,tradetime.date() as tradeDate //update prevClosePrice set symbol=symbol+".XSHE" //prevClosePrice.replaceColumn!(`symbol,string(prevClosePrice.symbol)) //stockCTAPrevClosePriceData=prevClosePrice //userConfig["prevClosePrice"]=stockCTAPrevClosePriceData try{Backtest::dropBacktestEngine(strategyName)}catch(ex){print ex} engine = Backtest::createBacktester(strategyName, userConfig, callbacks,true) // Step 3: Retrieve data and run the strategy backtest // //timer stockCTADataW=getTickHqDataWideTB(startDate,endDate,codes) // Start backtesting //Import sample CSV data 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 ) stockCTADataW=loadText("/ssd/ssd7/mhxiang/Data/stockCTADataW.csv",schema=sch) timer Backtest::appendQuotationMsg(engine, stockCTADataW) //exec distinct(symbol) from stockCTADataW // Step 4: Get backtest results //Trade details tradeDetails=Backtest::getTradeDetails(long(engine)) //Query the current list of unfilled (incomplete) orders openOrders=Backtest::getOpenOrders(long(engine)) //Daily positions dailyPosition=Backtest::getDailyPosition(long(engine)) // Available funds enableCash=Backtest::getAvailableCash(long(engine)) //Unfilled order details openOrders=tradeDetails[tradeDetails.orderStatus==-3] //Daily portfolio metrics totalPortfolios=Backtest::getDailyTotalPortfolios(long(engine)) //Comprehensive backtest results returnSummary=Backtest::getReturnSummary(long(engine)) lastprice= select last(lastPrice) as closePrice from stockCTADataW 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 /**********************The following is the parallel backtesting code****************************************************/ /*startDate=2023.02.01 endDate=2023.02.28 userConfig=dict(STRING,ANY) userConfig["startDate"]= startDate userConfig["endDate"]=endDate // Strategy type, currently fixed to stock userConfig["strategyGroup"]= "stock" // Generate snapshots from tick-by-tick data at the specified frequency userConfig["frequency"]= 0 userConfig["cash"]= 100000000 // Commission userConfig["commission"]= 0.0 // Stamp duty userConfig["tax"]= 0.0 // Order matching mode, 1 or 2 userConfig["dataType"]= 0 // Whether to subscribe to tick-by-tick quotes userConfig["enableSubscriptionToTickQuotes"]= true // Whether to obtain the order's position in the market userConfig["outputQueuePosition"]= 2 // Tick-by-tick data format: table or dict userConfig["msgAsTable"]= false strategyName="stockCTA" // Step 3: Run parallel strategy backtests codes=(exec distinct(SecurityID) from loadTable("dfs://level2_tl","snapshot") where tradetime.date() between startDate:endDate and market=`sh and SecurityID like "6%")[:50] // 1: Parallelize by stock // 2: Parallelize by trading day // 3: Parallelize each trading day separately // 4: Divide the stocks into n groups for parallel processing parallelMode=2 n=5 dailyReport,tradeOutputTable,engines,removeJods,ts=runBacktestParallelMode(strategyName,userConfig,initialize, beforeTrading,onTick, onSnapshot,onOrder,onTrade,afterTrading,finalized,startDate,endDate,codes,parallelMode,n) // Step 4: Retrieve parallel computation results // Elapsed time statistics t=select *from getRecentJobs(10000) where jobDesc in ts.keys() t=select * ,ts[string(jobDesc)] as runBacktest from t where rowNo(endTime)>t.size()-size(ts.keys() )-1 t=select (max(endTime) -min(startTime))\1000 as "Total elapsed time (s)",mean(endTime-startTime)\1000 as "Average total elapsed time (s)",mean(runBacktest)\1000 as "Average backtest execution time (s)",mean(endTime-startTime-runBacktest)\1000 as "Average data replay time (s)",mean(endTime-startTime)\1000\60 as "Average total elapsed time (min)",mean(runBacktest)\1000\60 as "Average backtest execution time (min)",mean(endTime-startTime-runBacktest)\1000\60 as "Average data replay time (min)" from t print(t)*/ //