创建引擎并实例化监视器后，将首先调用其内部的 onload 函数。回顾上文提到的 CEP 模块所负责的功能，其工作流的源头是监听策略启动事件，一旦监听到策略启动事件才进行下一步的操作。因此，在 onload 函数中，只需考虑设置相关的事件监听器以便启动策略即可。

def onload() {
    addEventListener(handler=initiatingStrategy, eventType="Initiating", times="all")
}

• 设置了一个事件监听器，监听所有的策略启动事件。目的是当监听到该事件时，下一步将启动整个策略，因此在对应的回调函数 initiatingStrategy 中需要包含对行情事件的监听器和创建所需要的计算引擎等步骤。从 onload 方法开始，函数调用的流程与实现的功能大致如下图所示。

接下来从策略启动事件对应的回调函数 initiatingStrategy 开始来介绍具体的代码实现。

def initiatingStrategy(Initiating){
    instanceId = Initiating.instanceId
    factorParams = Initiating.factorParams
    strategyParams = Initiating.strategyParams
    initCash = Initiating.cashAmount
    availableCash = initCash
    initHoldingQty = Initiating.holdingQty
    securitySelected = Initiating.stockPool
    availablePos[securitySelected] = initHoldingQty                  
    preCloTable = select SecurityID, PreCloPrice from loadTable("dfs://testDB",
    `snapshot) where date(TradeTime) = 2023.02.02 context by SecurityID limit 1
    preCloDict = dict(preCloTable.SecurityID, preCloTable.PreCloPrice)
    preClo = preCloDict[securitySelected]
    initHoldingValue =  sum(preClo * initHoldingQty)
    //创建表，统计实时情况
    share(table(100:0,["time","changeAmount","availableCash"],[TIMESTAMP, INT, INT]),
    "CashTable"+instanceId) 
    cashTb = objByName("CashTable"+instanceId)
    cashTb.tableInsert(timestamp,0,availableCash
    //创建模拟撮合引擎
    createMatchingEngine(instanceId)
    //创建信号表，dashboard展示
    share(streamTable(100:0,`securityid`direction`Time`orderPrice`BS_Signal,
    `STRING`INT`TIMESTAMP`DOUBLE`DOUBLE),"signalTb"+instanceId)
    signalTb = objByName("signalTb"+instanceId)
    //监听快照
    addEventListener(handler=processSnapshot, eventType="Snapshot", times="all")
    addEventListener(handler=updateMatchingEngineQuotes, eventType="Snapshot",
     times="all")
    //监听逐笔成交
    addEventListener(handler=processTrans, eventType="Transaction", times="all")
    //监听内部事件，检查下单状态
    addEventListener(handler=orderStateCheck, eventType="InnerOrder", 
    condition=<InnerOrder.instanceId=instanceId>, times="all")
    createCalFactorEngines(instanceId)
}

• 从监听到的 Initiating 事件中获取各种策略设定值，如指标计算参数字典、策略参数字典、初始资金和初始持仓等。

• 从数据库中取出对应股票的昨日收盘价，与初始持仓数量一起计算出初始的总持仓价值。

• 创建并共享一个资金统计表，用于统计策略实时的资金变化，并将初始的现金情况插入表中；创建并共享一个买卖信号表，用于可视化买卖信号。

• 调用 createMatchingEngine 函数，创建模拟撮合引擎。主要步骤包括配置文件修改、行情表和用户订单表结构定义、列名映射配置、成交明细输出表定义和创建引擎。

  def createMatchingEngine(instanceId){
      //撮合引擎cfg
      config = dict(STRING, DOUBLE);
      config["latency"] = 0;
      config["depth"] = 10; 
      config["outputOrderBook"] = 0            //用户订单时延为0
      config["orderBookMatchingRatio"] = 1;    //与订单薄匹配时的成交百分比
      config["dataType"] = 1;                  //行情类别：1表示股票快照
      config["matchingMode"] = 1;  //撮合模式一：与最新成交价以及对手方盘口按配置的比例撮合
      config["matchingRatio"] = 0.1;           //快照模式下，快照的区间成交百分比
      //根据行情表和用户订单表的表结构来创建相应的列名映射字典
      //snapshot输入表结构
      dummyQuotationTable = table(1:0,`SecurityID`Market`TradeTime`LastPrice
      `UpLimitPx`DownLimitPx`TotalBidQty`TotalOfferQty`BidPrice`BidOrderQty
      `OfferPrice`OfferOrderQty,[STRING,STRING,TIMESTAMP,DOUBLE,DOUBLE,DOUBLE,LONG,
      LONG,DOUBLE[],INT[],DOUBLE[],INT[]])
      quotationColMap = dict(`symbol`symbolSource`timestamp`lastPrice`upLimitPrice
      `downLimitPrice`totalBidQty`totalOfferQty`bidPrice`bidQty`offerPrice`offerQty,
       `SecurityID`Market`TradeTime`LastPrice`UpLimitPx`DownLimitPx`TotalBidQty
       `TotalOfferQty`BidPrice`BidOrderQty`OfferPrice`OfferOrderQty)
      //委托订单结构
      dummyUserOrderTable = table(1:0, `symbol`time`orderType`price`qty`BSFlag
      `orderID, [STRING, TIMESTAMP, INT, DOUBLE, LONG, INT, LONG])
      userOrderColMap = dict( `symbol`timestamp`orderType`price`orderQty`direction
      `orderId, `symbol`time`orderType`price`qty`BSFlag`orderID)
      //自定义输出表的形式
      tradeOutputTable  = table(1:0, `orderId`symbol`direction`sendTime`orderPrice
      `orderQty`tradeTime`tradePrice`tradeQty`orderStatus`sysReceiveTime, 
      [LONG, STRING, INT, TIMESTAMP,DOUBLE,LONG, TIMESTAMP,DOUBLE,LONG, INT,
      NANOTIMESTAMP])
      share(tradeOutputTable,"MatchingResult"+instanceId)
      matResultTb = objByName("MatchingResult"+instanceId)
      //撮合深交所股票
      exchange = "XSHE" 
      //创建引擎
      name = "MatchingEngine"+instanceId 
      matchingEngine = MatchingEngineSimulator::createMatchEngine(name, exchange,
      config, dummyQuotationTable, quotationColMap, dummyUserOrderTable, 
      userOrderColMap, matResultTb)
  }    

  • 在加载模拟撮合插件后，可以通过 MatchingEngineSimulator::createMatchEngine 函数创建模拟撮合引擎。
  • exchange 参数指定交易所代码，用于明确撮合规则。
  • 构造了一个 config 字典，并在字典中配置撮合引擎的撮合设定，如订单时延、输入给撮合引擎的行情类别、撮合模式等。更详细的参数设置和含义见模拟撮合引擎使用教程。
  • 对于输入表 dummyQuotationTable 和 dummyUserOrderTable，模拟撮合引擎要求它们必须包含指定的字段名称。如果自定义的行情数据表的字段名称与引擎要求不一致，可以通过 quotationColMap 字典进行映射；同样地，如果自定义的用户委托订单数据表的字段名称与引擎要求不一致，可以通过 userOrderColMap 字典进行映射。
  • 对于输出表 tradeOutputTable，模拟撮合引擎对它的字段名称没有要求，但其输出的每个字段有特定的含义，所以字段的顺序不能改变。

• 创建 4 个事件监听器，其中：

  • 监听所有的快照事件，一旦监听到快照事件，将立刻调用 processSnapshot 函数进行处理。该函数的功能是将快照数据写入响应式状态引擎进行指标的计算。

  def processSnapshot(SnapshotEvent) {       
      timestamp = SnapshotEvent.snapshotTime
      macdCciCalEngine = getStreamEngine("CalMacdCciEngine"+instanceId) 
      insert into macdCciCalEngine values([SnapshotEvent.snapshotTime, 
      SnapshotEvent.securityid, SnapshotEvent.lastPrice, SnapshotEvent.openPrice, 
      SnapshotEvent.highPrice, SnapshotEvent.lowPrice])
  }   

  • 监听所有的快照事件，一旦监听到快照事件，将立刻调用 updateMatchingEngineQuotes 函数进行处理。该函数的功能是将行情数据写入模拟撮合引擎，使模拟撮合引擎中的行情与指标计算、产生信号的操作同步。在本案例中可以看到，在 CEP 引擎中同一个事件可以同时被多个事件监听器监听，触发不同的操作。

    def updateMatchingEngineQuotes(SnapshotEvent){
        SecurityID = SnapshotEvent.securityid
        Market = SnapshotEvent.market
        TradeTime = SnapshotEvent.snapshotTime
        LastPrice = SnapshotEvent.lastPrice
        UplimitPx = SnapshotEvent.upLimitPx
        DownLimitPx = SnapshotEvent.downLimitPx
        TotalBidQty = SnapshotEvent.totalBidQty
        TotalOfferQty = SnapshotEvent.totalOfferQty
        BidPrice = SnapshotEvent.bidPrice
        BidOrderQty = SnapshotEvent.bidOrderQty
        OfferPrice = SnapshotEvent.offerPrice
        OfferOrderQty = SnapshotEvent.offerOrderQty
        quoteTable = table(100:0,["SecurityID", "Market","TradeTime", "LastPrice",
        "UplimitPx","DownLimitPx","TotalBidQty","TotalOfferQty","BidPrice",
        "BidOrderQty","OfferPrice","OfferOrderQty"], [STRING, STRING, TIMESTAMP,
         DOUBLE, DOUBLE, DOUBLE, LONG, LONG, DOUBLE[], INT[], DOUBLE[], INT[]])
        quoteTable.tableInsert([SecurityID],[Market],[TradeTime],[LastPrice],
        [UplimitPx],[DownLimitPx],[TotalBidQty],[TotalOfferQty],[BidPrice],
        [BidOrderQty],[OfferPrice],[OfferOrderQty])
        MatchingEngineSimulator::insertMsg(getStreamEngine("MatchingEngine"+
        instanceId), quoteTable, 1)
    }

  • 监听所有的成交事件，一旦监听到成交事件，将立刻调用 processTrans 函数进行处理。该函数的功能是将成交数据写入响应式状态引擎进行指标的计算。

    def processTrans(transactionEvent) {  
        if(transactionEvent.tradePrice != 0){
          timestamp = transactionEvent.transTime
          volCalEngine = getStreamEngine("CalVolEngine"+instanceId)
          insert into volCalEngine values([transactionEvent.securityid, 
          transactionEvent.tradePrice, transactionEvent.volume, 
          transactionEvent.transTime])   
        }     
        else{return}
    }

  • 监听所有的内部订单事件，一旦监听到内部订单事件，将立刻调用 orderStateCheck 函数进行处理，该函数的功能是检查下单状态后下单。该函数由内部订单事件触发，将在后文再做详细介绍。

• 调用 createCalFactorEngines 函数，创建对应的因子计算引擎（响应式状态引擎）。这是后续一切流程的开始，下面详细介绍该函数的功能实现。

创建两个响应式状态引擎，分别对快照和逐笔数据计算不同的指标。

def createCalFactorEngines(instanceId){
    //MACD CCI计算
    calDummy = streamTable(1:0, ["snapshotTime","securityid","LastPrice","OpenPrice"
    ,"HighPrice","LowPrice"], [TIMESTAMP, STRING, DOUBLE, DOUBLE , DOUBLE , DOUBLE])
    macdFastPeriod = factorParams[`macd][`fastPeriod]
    macdSlowPeriod = factorParams[`macd][`slowPeriod]
    macdSignalPeriod = factorParams[`macd][`signalPeriod]
    cciTimePeriod = factorParams[`cci][`timePeriod]
    calMetrics = [<snapshotTime>, <ta::macd(LastPrice, macdFastPeriod, macdSlowPeriod,
    macdSignalPeriod) as `diff`dea`macdHistogram>, <ta::cci(HighPrice, LowPrice,
    LastPrice, cciTimePeriod) as `cci>]    
    macdCciOutputTable = streamTable(1:0,  ["securityid", "time", "diff", "dea", 
    "macdHistogram", "cci"],  [STRING, TIMESTAMP, DOUBLE, DOUBLE, DOUBLE, DOUBLE])
    share(macdCciOutputTable,"MacdAndCci"+instanceId) 
    macdCciOutputTable = objByName("MacdAndCci"+instanceId)
    macdCciCalEngine = createReactiveStateEngine(name="CalMacdCciEngine"+instanceId,
    metrics=calMetrics, dummyTable=calDummy, outputTable=macdCciOutputTable, 
    outputHandler=handleMacdCciOutput ,keyColumn=`securityid, keepOrder=true, 
    msgAsTable=true)
    //引擎预热
    warmupTable = select TradeTime,SecurityID,LastPrice,OpenPrice,HighPrice,
    LowPrice from loadTable('dfs://testDB',`snapshot) where (second(TradeTime) 
    >= second(09:30:00.000) and second(TradeTime) <= second(15:00:00.000)) and 
    date(TradeTime) = 2023.02.01 order by TradeTime 
    warmupStreamEngine(macdCciCalEngine, warmupTable)
    //vol计算
    volDummy = streamTable(1:0, ["securityid", "TradePrice", "vol", "transTime"], 
    [STRING, DOUBLE, INT, TIMESTAMP])
    volTimePeriod = factorParams[`volume][`timePeriod]
    volMetrics = [<TradePrice>, <tmsum(transTime, vol, volTimePeriod)>, <transTime>]
    volOutputTable = streamTable(1:0, ["securityid", "TradePrice","vol", "time"], 
    [STRING, DOUBLE,INT, TIMESTAMP])
    share(volOutputTable,"Vol"+instanceId)
    volOutputTable = objByName("Vol"+instanceId)
    volCalEngine = createReactiveStateEngine(name="CalVolEngine"+instanceId, 
    metrics=volMetrics, dummyTable=volDummy, outputTable=volOutputTable, 
    outputHandler=handleVolOutput ,keyColumn=`securityid, keepOrder=true, 
    msgAsTable=true)
}

• 为了方便用户使用 TA - Lib 中的技术指标，DolphinDB 提供了以 DolphinScript 实现的 TA-Lib 中包含的指标函数，并封装在 DolphinDB ta 模块 中。 createReactiveStateEngine 函数通过 metrics 参数指定计算指标，本案例用到了 ta 模块（已经在创建监视器前声明 use ta）来方便地表达指标逻辑，省去了用户编写复杂计算公式的步骤。
• 以输入快照数据、计算 MACD 和 CCI 指标的响应式状态引擎为例，引擎计算的结果将交由 handleMacdCciOutput 函数处理。在该函数中将进行下单信号的判断，我们将在后文进行详细介绍。创建计算成交量指标的响应式状态引擎的流程类似，其计算结果交由 handleVolOutput 函数处理。注意，设置 outputHandler 参数后，引擎不再将每次的计算结果写到输出表 macdCciOutputTable，而是会不断地调用 outputHandler 参数指定的函数处理计算结果。

在创建计算引擎的函数中，有一段关于引擎预热的代码：

warmupTable = select TradeTime,SecurityID,LastPrice,OpenPrice,HighPrice,LowPrice 
from loadTable('dfs://testDB',`snapshot) where (second(TradeTime) >= 
second(09:30:00.000) and second(TradeTime) <= second(15:00:00.000)) and 
date(TradeTime) = 2023.02.01 order by TradeTime 
warmupStreamEngine(macdCciCalEngine, warmupTable)

• 这段代码的目的是对计算 MACD 和 CCI 指标的响应式状态引擎进行预热。warmupStreamEngine 会把前一天的数据写入流数据引擎并进行计算，但是不输出结果。当当日的数据正常写入此流数据引擎后，滑动窗口将包含前一日的快照数据。在本例中 MACD 指标中涉及时间最久的参数 slowPeriod = 520，即 520 个滑动窗口，若不进行预热，则会出现在开盘后近 30 分钟内指标计算结果为空的情况。

handleVolOutput

def handleVolOutput(volResult){       
    tableInsert(objByName("Vol"+instanceId), volResult)
    priceJustTraded[volResult.securityid[0]] = volResult.TradePrice[0]
    pre30vol[volResult.securityid[0]] = volResult[`vol][0]
}

• 把引擎的计算结果写入对应的输出表中，这与策略逻辑无关，仅供 Dashboard 可视化展示。
• 更新存储最新成交价的字典，字典的键值为股票，之后将用最新的成交价格作为委托买入或卖出的价格。
• 更新存储30秒成交量之和的字典 pre30vol，字典的键值为股票。

handleMacdCciOutput

函数调用流程与大致功能如下图所示。

def handleMacdCciOutput(macdCciResult) {
    tableInsert(macdCciOutputTable, macdCciResult)
    if(second(macdCciResult.time)<09:30:00 or second(macdCciResult.time) >15:00:00 ){
      return
    }    
    self.checkGoldenCross(macdCciResult.securityid[0], macdCciResult.diff[0], 
    macdCciResult.dea[0])
    preDiff[macdCciResult.securityid[0]] = macdCciResult.diff[0]
    preDea[macdCciResult.securityid[0]] = macdCciResult.dea [0]   
    if (pre30cci[macdCciResult.securityid[0]] == NULL) {
      pre30cci[macdCciResult.securityid[0]] = macdCciResult.cci[0]
    }
    else if(pre30cci[macdCciResult.securityid[0]] != NULL and 
    newCci[macdCciResult.securityid[0]] == NULL){
      newCci[macdCciResult.securityid[0]] = macdCciResult.cci[0]
    }
    else{pre30cci[macdCciResult.securityid[0]] = newCci[macdCciResult.securityid[0]]
    newCci[macdCciResult.securityid[0]] = macdCciResult.cci[0]}
}

• 同理，先将计算结果插入到对应的共享表中以供 Dashboard 可视化展示。

• 调用金叉检查函数 checkGoldenCross 检查是否产生金叉：

  def checkGoldenCross(securityid, diff, dea) {
      if (self.preDiff[securityid] != NULL and self.preDea[securityid] != NULL) {
          if (self.preDiff[securityid] < self.preDea[securityid] and diff > dea) {  
              newDiff[securityid] = diff
              checkCciAndVol(securityid) 
          }
          if (self.preDiff[securityid] > self.preDea[securityid] and diff < dea){ 
              newDiff[securityid] = diff
              emptyPosition(securityid)
          }
      }
  }    

  • 若产生死叉，调用函数 emptyPosition 进行清仓：函数将从持仓状态字典中获取死叉股票的现有持仓量，同时获取死叉出现时的时间戳作为委托卖单的时间。若死叉股票有持仓，则一方面构造 SellOrder 事件，通过 emitEvent 接口将其向外发送给事件输出队列；一方面按照模拟撮合引擎订单结构，构造相应的委托卖单订单消息，通过对应的插件接口发送给模拟撮合引擎。发出委托消息后，记录卖出信号供 Dashboard 展示使用，并更新死叉股票的持仓数量为0、将卖出的金额加入到可用现金中。

    def emptyPosition(securityid){ 
        availableQty = availablePos[securityid]
        sellTime = timestamp
        if (availableQty > 0){
            sellPrice = priceJustTraded[securityid]
            direction = 2 
            orderType = 5
            orderID = long(orderCount)
            sellAmount = sellPrice * availableQty
            sellOrder = SellOrder(securityid, sellTime, orderType, sellPrice, availableQty, direction, orderID)  
            emitEvent(sellOrder)            
            t = objByName("signalTb"+instanceId)
            insert into t values (securityid, direction, sellTime, sellPrice, newDiff[securityid])
            msgBody = (securityid, sellTime, orderType, sellPrice, availableQty, direction, orderID)
            MatchingEngineSimulator::insertMsg(getStreamEngine("MatchingEngine"+instanceId), msgBody, 2)
            orderCount = orderCount + 1
            availablePos[securityid] = 0 
            availableCash += sellAmount
            cashTb.tableInsert(sellTime,sellAmount,availableCash)
        }
    }

  • 若产生金叉，则调用函数 checkCciAndVol，进一步检查 CCI 和成交量条件是否达成。函数将判断 CCI 指标在最近30秒内是否发生了上穿的情况：

    • 若 CCI 上穿 -100 的阈值，则构造内部订单事件（InnerOrder），并通过 routeEvent 将该事件插入到当前 CEP 子引擎的事件队列的队首；

    • 若 CCI 上穿 100 的阈值，并且过去 30 秒的成交量之和大于 50000，则同样将内部订单事件插入到当前 CEP 子引擎的事件队列的队首。

      def checkCciAndVol(securityid){
          if(pre30cci[securityid] < strategyParams[`cci][`lowerBoundary] and 
          newCci[securityid] > strategyParams[`cci][`lowerBoundary]){
              innerOrderTime = timestamp
              innerOrder = InnerOrder(securityid, innerOrderTime, 
              priceJustTraded[securityid], strategyParams[`order][`orderQty], 
              instanceId)
              routeEvent(innerOrder)
          }
          else if(pre30cci[securityid] < strategyParams[`cci][`upperBoundary] and
           newCci[securityid] > strategyParams[`cci][`upperBoundary] and 
           pre30vol[securityid] > strategyParams[`volume][`triggerAmount]){ 
              innerOrderTime = timestamp
              innerOrder = InnerOrder(securityid, innerOrderTime, 
              priceJustTraded[securityid], strategyParams[`order][`orderQty], 
              instanceId)
              routeEvent(innerOrder)
          }    
      }

  • 在金叉死叉判断结束后，更新对应的 MACD、CCI 指标字典，作为下一次信号判断的依据，字典的键值均为股票。

在通过 initiatingStrategy 函数初始化策略时，已经设置了一个事件监听器对内部订单事件进行监听。当该事件被插入到当前 CEP 子引擎的事件处理队列的队首，将被该监听器捕获，触发对应方法 orderStateCheck 进行下单状态检查。

def orderStateCheck(innerOrder){
    if(orderState[innerOrder.securityid] == 1){return}
    else {buyOrder(innerOrder)}
}

• 状态检查函数将检查下单状态字典，若状态为 "1" 表示“1s内刚下过单”，则不允许下单，反之允许下单。若允许下单，则调用下单函数 buyOrder：

  def buyOrder(innerOrder){
      orderPrice = innerOrder.priceJustTraded
      orderQty = innerOrder.quantity
      buyAmount = orderPrice * orderQty
      if (availableCash > buyAmount){           
          orderSecurity = innerOrder.securityid
          orderTime = innerOrder.timestamp
          direction = 1
          orderType = 5
          orderCount = orderCount + 1
          orderID = long(orderCount)
          buyOrder = BuyOrder(orderSecurity, orderTime, orderType, orderPrice, 
          orderQty, direction, orderID)
          emitEvent(buyOrder)
          msgBody = (orderSecurity, orderTime, orderType, orderPrice, orderQty, 
          direction, orderID)
          t = objByName("signalTb"+instanceId)
          insert into t values (orderSecurity, direction, orderTime, orderPrice, 
          newDiff[orderSecurity])
          MatchingEngineSimulator::insertMsg(
          getStreamEngine("MatchingEngine"+instanceId), msgBody, 2)
          //下单后把下单状态改为1，设置一个定时器1秒后更新下单状态
          orderState[orderSecurity] = 1
          addEventListener(handler=refreshOrderState{orderSecurity}, times=1, wait=1s)
          availableCash -= buyAmount
          cashTb.tableInsert(orderTime, -buyAmount, availableCash)
      }
  }

  • 首先判断是否有足够的可用现金，随后构造了委托订单消息传递给模拟撮合引擎，并将下单状态修改为 "1"。在可用现金中扣去下单所需的金额。

  • 将买入信号写入表中供 Dashboard 展示使用，构造了 BuyOrder 事件并发送到事件输出队列。

  • 更新持仓状态字典值，并通过 addEventListener 设置了一个定时器，令 times=1 且 wait=1s ，即1秒后执行1次回调函数 refreshOrderState，刷新对应股票的下单状态。这是为了避免出现交易中信号被反复触发，短时间内大量下单的情况。

    def refreshOrderState(orderSecurity){orderState[orderSecurity] = 0}