模拟撮合引擎使用教程

以 Level-2 3s行情快照数据为模拟撮合的行情输入，成交比例设置为0.5，区间成交比例设置0.1，每3s每只股票下单1000股，测试统计股票数量分别为100只，500只，2000只时耗时（耗时包括从原始分布式表查询数据和计算的耗时）。完整的测试脚本将在附件中提供。

def strategy(mutable engine,msg,mutable contextdict,mutable logTB){ 
	/*
	 * 每只股票每3秒下单1000股
	 */
	try{
		temp=select last(DateTime) as time ,last(lastPx) as price from msg group by SecurityId
		codes=distinct(msg.SecurityId)
		prices=dict(temp.SecurityId,temp.price)
		for (icode in codes){
			appendMsg(engine,(icode,contextdict["now"],5,prices[icode],1000,1,1),2)
			}
		//log
		text="open_signal: "+string(contextdict["now"])+"  code : "+ 
		concat(temp.SecurityId,",")+" price:  "+string(concat(string(temp.price),","))
		logTB.append!(table(text as info))		
		}
	catch(ex){
		logTB.append!(table("error:"+string(contextdict["now"])+ex[1] as info))
	}
}



timer{
	for (idate in 2022.04.15..2022.04.15){
		contextdict["now"]=idate
		tb = select "XSHE" as symbolSource,SecurityId,DateTime,lastPrice,highestPrice,lowestPrice,deltas(TotalVolumeTrade)  as newTradeQty ,BidPrice,
		BidOrderQty,OfferPrice,OfferOrderQty from loadTable("dfs://level2_tickTrade",`tickTradeTable) where  SecurityId in universe and 
		date(DateTime)=idate and  DateTime.second() between 09:30:00:15:00:00
		context by SecurityId csort DateTime order by DateTime map  
		times=tb.DateTime
		nTimes=cumsum(groupby(count,times,times).values()[1])
		i=1
		for (i in 0..(size(nTimes)-1)){
			if(i==0){
				contextdict["now"]=times[nTimes[0]-1]
				msg=tb[0:nTimes[0]]
			}
			else{
				contextdict["now"]=times[nTimes[i]-1]
				msg=tb[nTimes[i-1]:nTimes[i]]
				}		
			if(size(msg)>0){
				savehqdataToMatchEngine(engine, msg,logTB)	
				strategy(engine,msg, contextdict, logTB)
				}
			}

		MatchingEngineSimulator::resetMatchEngine(engine)
	}
}

成交明细结果表如下：

• 加载插件

  loadPlugin("<path>/MatchEngineTest.txt")

  本算法交易插件命名为MatchEngineTest，txt文件在该插件源码的bin目录下。

• 插件接口

  MatchEngineTest::runTestTickData(messageTable, stockList, engineConfig)

  messageTable 读入的是通过 replay 回放行情数据（逐笔委托和逐笔成交）后的数据。

  stockList 是用户下单的股票列表。

  engineConfig 是一个字典，包含三个配置参数：

  • engineConfig[“engineName“] 为本次测试的名称，它同时也是创建模拟撮合引擎时的名称。在创建多个模拟撮合引擎时，引擎名称不能重复。
  • engineConfig[“market“] 为使用的交易所，”XSHG”（上交所）或者”XSHE”。
  • engineConfig[“hasSnapshotOutput“]和模拟撮合引擎中的“outputOrderBook”参数配置相同，表示是否输出订单簿。

• 输入输出说明

  输入数据表 messageTable 的列名和对应的类型如下：

  colName = `msgTime`msgType`msgBody`sourceType`seqNum
  colType =  [TIMESTAMP, SYMBOL, BLOB, INT, INT]

  msgType 的值有 order, trade 和 snapshot，分别表示逐笔委托、逐笔成交和快照数据。插件内部会将 messageTable 表中的 BLOB 格式数据进行解析。该插件中算法交易的数据源只使用了逐笔数据，逐笔委托和逐笔成交数据的结构为 tickSchema，tickSchema 在 C++ 代码中的表结构定义如下：

  TableSP tickSchema = Util::createTable({"symbol", "symbolSource", "TradeTime", "sourceType", "orderType", "price", "qty", "buyNo", "sellNo", "direction", "seqNum"}, \
              {DT_SYMBOL, DT_SYMBOL, DT_TIMESTAMP, DT_INT, DT_INT, DT_DOUBLE, DT_INT, DT_INT, DT_INT, DT_INT, DT_INT}, 0, 0);

  如果算法交易需要利用到快照数据，那么数据源的输入需要包含快照数据，快照数据的表结构在 C++ 代码中的定义如下：

  TableSP snapshotSchema = Util::createTable({"SecurityId", "TradeTime", "LastPrice", "BidPrice", "OfferPrice", "BidOrderQty", "OfferOrderQty", "TotalValueTrade", "TotalVolumeTrade", "seqNum"}, \
              {DT_SYMBOL, DT_TIMESTAMP, DT_DOUBLE, DATA_TYPE(ARRAY_TYPE_BASE + DT_DOUBLE), DATA_TYPE(ARRAY_TYPE_BASE + DT_DOUBLE), DATA_TYPE(ARRAY_TYPE_BASE + DT_INT), \
              DATA_TYPE(ARRAY_TYPE_BASE + DT_INT), DT_DOUBLE, DT_INT, DT_INT}, 0, 0);

  在插件代码中，将模拟撮合引擎输出的 tradeOutputTable 和 snapshotOutput 的两个结果表保存为了共享表，用户可以查找这两个表来查看模拟撮合的结果。

  // 输出结果
  vector<string> arg0 = {"tmp_tradeOutputTable"};
  vector<ConstantSP> arg1 = {tradeOutputTable};
  heap_->currentSession()->run(arg0, arg1);
  runScript("share tmp_tradeOutputTable as "+ testName_ + "_tradeOutputTable");
  
  arg0 = {"tmp_snapshotOutput"};
  arg1 = {snapshotOutput};
  heap_->currentSession()->run(arg0, arg1);
  runScript("share tmp_snapshotOutput as "+ testName_ + "_snapshotOutput");

  在分股票多线程执行算法交易时，每个线程有单独的 tradeOutputTable 和 snapshotOutput，用户查看股票成交情况时，需要将各个线程的结果表进行汇总后输出。

  tradeResult = table(100:0, MatchEngineTest1_tradeOutputTable.schema().colDefs.name, MatchEngineTest1_tradeOutputTable.schema().colDefs.typeString)
  for (i in 1..thread_num) {
      tradeResult.append!(objByName("MatchEngineTest"+i+"_tradeOutputTable"))
  }
  // 查看发送的用户订单的总成交量
  select Symbol, sum(VolumeTraded) from tradeResult where OrderStatus != 3 and OrderStatus != 2 group by Symbol order by Symbol

本节将从插件入口开始讲述代码执行的整体流程，完整的 C++ 代码将在附件中提供。

• 插件接口函数

  每次调用 runTestTickData 时会创建一个 TickDataMatchEngineTest 类，并调用该类的 runTestTickData 方法。

  ConstantSP runTestTickData(Heap *heap, std::vector<ConstantSP> &arguments) {
      TableSP messageStream = arguments[0];
      ConstantSP stockList = arguments[1];
      DictionarySP engineConfig = arguments[2];
  
      TickDataMatchEngineTest test(heap, engineConfig);
      test.testTickData(messageStream, stockList);
  
      return new String("done");
  }

  TickDataMatchEngineTest 类的构造函数定义如下：

  TickDataMatchEngineTest::TickDataMatchEngineTest(Heap * heap, ConstantSP engineConfig)
      : heap_(heap) {
      string name = engine_config->get(new String("engineName"))->getString();
      market_ = engine_config->get(new String("market"))->getString();
      if (market_ == "XSHE") {
          marketOrderType_ = static_cast<int>(UserOrderType::XSHEOppositeBestPrice);
      }
      else {
          marketOrderType_ = static_cast<int>(UserOrderType::XSHGBestFivePricesWithLimit);
      }
      bool hasSnapshotOutput = engine_config->get(new String("hasSnapshotOutput"))->getBool();
      test_name_ = name;
      // 创建模拟撮合引擎
      engine_ = createMatchEngine(name, market_, hasSnapshotOutput);
      ...
  }

  testTickData 函数的定义如下：

  bool TickDataMatchEngineTest::testTickData(TableSP messageStream, ConstantSP stockList) {
      ...
      // MsgDeserializer是对BLOB格式数据的一个解析类，tickDeserializer用于解析逐笔数据
      MsgDeserializer tickDeserializer(tickSchema);
      ...
      ConstantSP nowTimestamp = messageCols[0]->get(0);
      ConstantSP lastTimestamp = nowTimestamp;
      for (int i = 0; i < messageStream->size(); ++i) {
          nowTimestamp = messageCols[0]->get(i);
          string data = messageCols[2]->getString(i);
          DataInputStream stream(data.c_str(), data.size());
          if (nowTimestamp->getLong() != lastTimestamp->getLong()) {
              context.timestamp = lastTimestamp;
              TableSP msgTable = tickDeserializer.getTable(); // 同一时间戳为一批行情数据
              saveQuatationToEngine(msgTable); // 发送行情数据到引擎
              handleStrategy(context, msgTable); // 执行算法策略生成用户订单并发送用户订单到引擎
              tickDeserializer.clear();
              lastTimestamp = nowTimestamp;
          }
          tickDeserializer.deserialize(&stream); // 解析该行数据并存储到该解析类内部的表中
      }
      ...
  }

• 为引擎插入行情数据和用户订单

  保存行情数据和发送用户订单的方式都和 DolphinDB 脚本中的方式相同。调用 appendMsg 方法向模拟撮合引擎传入数据。

  appendMsgFunc = heap_->currentSession()->getFunctionDef("appendMsg");
  ...
  void TickDataMatchEngineTest::saveQuatationToEngine(ConstantSP data) {
      vector<ConstantSP> args = {engine_, data, new Int(1)};
      appendMsgFunc->call(heap_, args);
  }
  
  void TickDataMatchEngineTest::saveOrdersToEngine(ConstantSP data) {
      vector<ConstantSP> args = {engine_, data, new Int(2)};
      appendMsgFunc->call(heap_, args);
  }

• 策略执行

  下单时，需要构建一个用户订单表，内存表的存储方式也是列存，因此需要先构建 VectorSP 类型的列，再通过数据列来创建表。对于深交所数据，下单时的市价单设置为0，对于上交所数据，下单时的市价单设置为1，对于市价单类型的定义可以参考插件接口手册。表的构建完成后，调用 saveOrdersToEngine 函数将用户订单发送到引擎。

  void TickDataMatchEngineTest::handleStrategy(ContextDict& context, ConstantSP msgTable) {
      // 判断策略触发时间
      ConstantSP timestampToTrigger = new Timestamp(tradeDate_->getInt() * (long long)86400000 + timeToTrigger_->getInt());
      if (!strategyFlag_ || timestampToTrigger->getLong() > context.timestamp->getLong()) {
          return;
      }
      // 获取未完成订单
      vector<ConstantSP> args = {engine_};
      TableSP openOrders = getOpenOrdersFunc->call(heap_, args);
      int openOrderSize = openOrders->size();
  
      if (openOrderSize > 0) {
          // 撤单
          ConstantSP symbols = openOrders->getColumn(2);
          ConstantSP times = createVectorByValue(context.timestamp, openOrderSize);
          ConstantSP ordertypes = createVectorByValue(new Int(static_cast<int>(UserOrderType::CancelOrder)), openOrderSize);
          ConstantSP prices = createVectorByValue(new Double(0.0), openOrderSize);
          ConstantSP qtys = openOrders->getColumn(5);
          ConstantSP directions = createVectorByValue(new Int(1), openOrderSize);
          ConstantSP orderIds = openOrders->getColumn(0);
          vector<ConstantSP> openOrderList = {symbols, times, ordertypes, prices, qtys, directions, orderIds};
          TableSP ordertable = Util::createTable({"SecurityId", "time", "orderType", "price", "qty", "direction", "orderId"}, openOrderList);
          saveOrdersToEngine(ordertable);
      }
      // 每只股票下市价单（卖一价）
      int stockSize = context.stockList->size();
      ConstantSP symbols = Util::createVector(DT_SYMBOL, stockSize);
      ConstantSP times = createVectorByValue(context.timestamp, stockSize);
      ConstantSP ordertypes = createVectorByValue(new Int(marketOrderType_), stockSize);
      ConstantSP prices = createVectorByValue(new Double(100.0), stockSize);
      ConstantSP qtys = createVectorByValue(new Int(2000), stockSize);
      ConstantSP directions = createVectorByValue(new Int(1), stockSize);
      ConstantSP orderIds = createVectorByValue(new Int(1), stockSize);
      for (int i = 0; i < stockSize; ++i) {
          symbols->set(i, context.stockList->get(i));
      }
      vector<ConstantSP> limitOrderList = {symbols, times, ordertypes, prices, qtys, directions, orderIds};
      TableSP ordertable = Util::createTable({"SecurityId", "time", "orderType", "price", "qty", "direction", "orderId"}, limitOrderList);
      saveOrdersToEngine(ordertable);
  
      // 更新策略触发时间
      timeToTrigger_ = new Time(timeToTrigger_->getInt() + 60 * 1000);
      if (timeToTrigger_->getInt() > timeToEnd_->getInt()) {
          if (intervalIdx_ + 1 < int(timeIntervals_.size())) {
              ++intervalIdx_;
              timeToTrigger_ = timeIntervals_[intervalIdx_][0];
              timeToEnd_ = timeIntervals_[intervalIdx_][1];
          }
          else {
              // 策略结束
              strategyFlag_ = false;
          }
      }
  }

本次测试使用多线程的方式，将行情数据按照对股票代码取模划分给不同线程。整个测试过程分为行情数据回放和模拟撮合两个部分。我们先使用 replayDS 和 replay 方法从分布式表中读取和回放行情数据，再将回放后的数据传入算法交易插件。

1. 行情数据回放

   • 首先需要设置交易所并获取需要下单的股票代码。当 market 设置为 sz 时，symbols 存储了深交所成交表中的所有股票的代码。当 market 设置为 sh 时，symbols 存储了上交所成交表中的所有股票的代码。

     trade_dfs = loadTable("dfs://TSDB_tradeAndorder", "trade")
     market = "sz"  // market = "sz" or "sh"
     if (market == "sz") {
         market_name = "XSHE"
         symbols = exec SecurityId from trade_dfs where SecurityId like "0%" or SecurityId like "3%" group by SecurityId
     }
     else {
         market_name = "XSHG"
         symbols = exec SecurityId from trade_dfs where SecurityId like "6%" group by SecurityId
     }

   • 对股票代码进行取模，将数据划分给不同的线程。

     def genSymbolListWithHash(symbolTotal, thread_num) {
         symbol_list = []
         for (i in 0:thread_num) {
             symbol_list.append!([]$STRING)
         }
         for (symb in symbolTotal) {
             idx = int(symb) % thread_num
             tmp_list = symbol_list[idx]
             tmp_list.append!(symb)
             symbol_list[idx] = tmp_list
         }
         return symbol_list
     }
     
     thread_num = 20 // 线程数
     symbol_list = genSymbolListWithHash(symbols, thread_num)

   • order 和 trade 表分别代表逐笔委托和逐笔成交数据，通过replay回放的方式将这两张表回放到同一张表内，回放后的数据是按照时间列排序的。此外，当同一时间有多条记录时，可以按照指定的字段排序。由于这两张表是分布式表，我们需要将replayDS方法和replay方法一起搭配使用。

     order_dfs = loadTable("dfs://TSDB_tradeAndorder", "order")
     trade_dfs = loadTable("dfs://TSDB_tradeAndorder", "trade")
     
     def replayBySymbol(market, marketName, symbolList, order, trade, i) {
         if (market == "sz") {
             ds1 = replayDS(sqlObj=<select SecurityId as symbol, marketName as symbolSource, TradeTime, 0 as sourceType, 
                 iif(OrderType in ["50"], 2, iif(OrderType in ["49"], 1, 3)) as orderType, Price as price, OrderQty as qty, int(ApplSeqNum) as buyNo, int(ApplSeqNum) as sellNo,
                 int(string(char(string(direction)))) as direction, int(SeqNo) as seqNum from order
                 where Market = market and date(TradeTime)==2022.04.14 and SecurityId in symbolList>, dateColumn = "TradeTime", timeColumn = "TradeTime")
             ds2 = replayDS(sqlObj=<select SecurityId as symbol, marketName as symbolSource, TradeTime, 1 as sourceType, 
                 iif(BidApplSeqNum==0|| OfferApplSeqNum==0,1,0) as orderType, TradePrice as price, int(tradeQty as qty), int(BidApplSeqNum) as buyNo, int(OfferApplSeqNum) as sellNo,
                 0 as direction, int(ApplSeqNum) as seqNum from trade
                 where Market = market and date(TradeTime)==2022.04.14 and SecurityId in symbolList>, dateColumn = "TradeTime", timeColumn = "TradeTime")
         }
         else {
             ds1 = replayDS(sqlObj=<select SecurityId as symbol, marketName as symbolSource, TradeTime, 0 as sourceType, 
                 iif(OrderType == "A", 2, 10) as orderType, Price as price, OrderQty as qty, int(ApplSeqNum) as buyNo, int(ApplSeqNum) as sellNo,
                 iif(Side == "B", 1, 2) as direction, int(SeqNo) as seqNum from order
                 where Market = market and date(TradeTime)==2022.04.14 and SecurityId in symbolList>, dateColumn = "TradeTime", timeColumn = "TradeTime")
             ds2 = replayDS(sqlObj=<select SecurityId as symbol, marketName as symbolSource, TradeTime, 1 as sourceType, 
                 0 as orderType, TradePrice as price, int(tradeQty as qty), int(BidApplSeqNum) as buyNo, int(OfferApplSeqNum) as sellNo,
                 0 as direction, int(TradeIndex) as seqNum from trade
                 where Market = market and date(TradeTime)==2022.04.14 and SecurityId in symbolList>, dateColumn = "TradeTime", timeColumn = "TradeTime")
         }
     
         inputDict  = dict(["order", "trade"], [ds1, ds2])
     
         colName = `msgTime`msgType`msgBody`sourceType`seqNum
         colType =  [TIMESTAMP, SYMBOL, BLOB, INT, INT]
     
         messageTemp = table(100:0, colName, colType)
         share(messageTemp, "MatchEngineTest" + i)
     
         // 当市场为深交所时，相同时间的数据需要按照先逐笔委托再逐笔成交的顺序排序
         // 逐笔委托单的sourceType为0，逐笔成交单的sourceType为1，因此可以按照sourceType字段排序
         if (market == "sz") {
             replay(inputDict, "MatchEngineTest" + i,`TradeTime,`TradeTime,,,1,`sourceType`seqNum)
         }
         // 上交所需要按照先逐笔成交再逐笔委托的顺序排序
         // 这里上交所的数据中seqNum是严格排序的，因此可以直接按照seqNum排序
         else {
             replay(inputDict, "MatchEngineTest" + i,`TradeTime,`TradeTime,,,1,`seqNum)
         }
     }
     
     // 计算行情回放的总耗时
     timer {
         job_list = [] // job_list存储了提交的所有作业id列表
         for (i in 1..thread_num) {
             job_list.append!(submitJob("TestJob" + i, "", replayBySymbol, market, market_name, symbol_list[i-1], order_dfs, trade_dfs, i))
         }
         // getJobReturn的参数true表示阻塞当前线程直到作业完成后返回
         for (i in 0 : thread_num) {
             getJobReturn(job_list[i], true)
         }
     }
     
     // 回放结果可以以磁盘表的形式存储，方便后续测试
     db = database("<path>/" + market + "_messages_" + thread_num + "_part")
     for (i in 1..thread_num) {
          saveTable(db, objByName("MatchEngineTest" + i), "MatchEngineTest" + i)
     }

   • 行情数据根据股票代码进行回放后，我们得到了thread_num个共享内存表，MatchEngineTest+数字是表的名称。为了方便后续测试，用户可以选择保存回放结果。由于csv文件不支持BLOB格式数据，用户可以以磁盘表的形式存储这些数据表。

2. 模拟撮合

   • 首先用户需要加载模拟撮合引擎插件和算法交易插件。

     loadPlugin("<path>/PluginMatchingEngineSimulator.txt")
     loadPlugin("<path>/MatchEngineTest.txt")

   • 数据准备和参数设置，这里的线程数量需要和行情回放时的相同。注意该算法测试插件接收的行情数据字段为msgTimemsgTypemsgBodysourceType`seqNum，这是逐笔委托和逐笔成交两张行情表经过replay回放到一张表后的结果。

     thread_num = 20
     market = "sz"
     market_name = "XSHE"
     
     messagesList = [] // 存储每个线程的行情数据表
     symbolList = []  // 存储每个线程的股票代码
     message_num = 0
     for (i in 1..thread_num) {
         // 从磁盘表中读取回放后的行情数据表，并获取表中的股票代码
         messages = select * from loadTable("<path>/" + market + "_messages_" + thread_num + "_part", "MatchEngineTest"+i)
         symbols = exec distinct left(msgBody, 6) from messages
         message_num += messages.size()
         messagesList.append!(messages)
         symbolList.append!(symbols)
     }

   • doMatchEngineTest表示单个线程需要执行的任务，任务主要包含设置需要传入的engine_config参数和调用算法测试插件接口，算法测试插件内部具体的执行过程在上一小节中已经介绍。注意这里的engine_config字典需要在函数内部创建，这是因为engine_config被传入插件接口的方式是一个引用，而非engine_config的一个拷贝副本。如果engine_config在函数外创建，多线程执行时一个线程对engine_config的修改会影响另一个线程。

     def doMatchEngineTest(messageData, symbolList, engineName, marketName, hasSnapshotOutput) {
         engine_config = dict(string, any)
         engine_config["engineName"] = engineName
         engine_config["market"] = marketName
         engine_config["hasSnapshotOutput"] = hasSnapshotOutput
         MatchEngineTest::runTestTickData(messageData, symbolList, engine_config)
     }
     
     ClearAllMatchEngine()
     
     // 计算模拟撮合的总耗时
     timer {
         test_job_list = [] // 存储了所有测试作业的id
         hasSnapshotOutput = false
     
         for (i in 1..thread_num) {
             messages = objByName("MatchEngineTest" + i)
             test_job_list.append!(submitJob("EngineTestJob" + i, "", doMatchEngineTest, 
                                 messages, symbol_list[i-1], "MatchEngineTest" + i, 
                                 market_name, hasSnapshotOutput))
         }
         // 等待所有作业结束后返回
         for (i in 0 : thread_num) {
             getJobReturn(test_job_list[i], true)
         }
     }

   • 当所有多线程任务结束后，需要将每个线程的算法交易结果（tradeOutputTable表和snapshotOutput表）进行合并，合并结果分别存储在tradeResult表和snapshotResult表。

     tradeResult = table(100:0, MatchEngineTest1_tradeOutputTable.schema().colDefs.name, MatchEngineTest1_tradeOutputTable.schema().colDefs.typeString)
     snapshotResult = table(100:0, MatchEngineTest1_snapshotOutput.schema().colDefs.name, MatchEngineTest1_snapshotOutput.schema().colDefs.typeString)
     for (i in 1..thread_num) {
         tradeResult.append!(objByName("MatchEngineTest"+i+"_tradeOutputTable"))
         snapshotResult.append!(objByName("MatchEngineTest"+i+"_snapshotOutput"))
     }
     
     // 结果合并到tradeResult表后，删除每个线程的tradeOutputTable表和snapshotOutput表
     for (i in 1..thread_num) {
         try {
             undef("MatchEngineTest" + i + "_tradeOutputTable", SHARED)
             undef("MatchEngineTest" + i + "_snapshotOutput", SHARED)
         } catch(ex) {print(ex)}
     }

   • 查询tradeResult表查看每只股票的总成交数量。

     select Symbol, sum(VolumeTraded) from tradeResult where OrderStatus !=2 and OrderStatus != 3 group by Symbol order by Symbol

行情回放耗时包含了行情数据读入和回放的时间，模拟撮合耗时包含了解析 BLOB 数据的时间和模拟撮合引擎的计算时间，这两个耗时都由timer计算得到。

以下分别显示了深交所逐笔数据和上交所逐笔数据在不同测试配置下的撮合结果。测试脚本在划分股票代码时采用了哈希方式，将股票代码取模并分配给不同线程。由于每只股票对应的行情数据量不同，可能导致不同线程执行时间的差异，尽量保证所有线程的行情数据量相近，以减少测试的总体耗时。

从测试结果可以看出，模拟撮合的耗时和线程数并非是成比例的。这是因为模拟撮合是计算密集型的任务，当线程数过多时，增加线程数会引入更多的资源竞争和调度开销，这些额外的开销影响了总体运行的耗时。