CEP 引擎入门：初级高频量价因子策略的实现

在 CEP 系统中，事件是一段时间内发生的特定事情或状态的数据表示，是描述对象变化的属性值的集合。例如“以170.85美元的价格买入1股 AAPL 股票”和“BRK 在2024年6月12日上涨5%”都是事件。它源源不断地产生，并以流的形式存在。

图1-1

在 CEP 系统中，定义一个事件时需要指定事件类型及其包含的每个属性的名称和类型。在 DolphinDB 中，使用类（Class）来定义事件，类名为事件类型，类的对象实例是一次具体的事件。同时，在 DolphinDB 中可以用流数据表来存储事件流，流数据表中的每一行记录存储了一次具体的事件。

图1-2

通常，用户在事件流中除了查找并处理特定的事件外，还会对事件进行额外的规则限制。复杂事件（Complex Event）不仅仅指某一个事件类型，还往往涉及复杂的事件匹配规则。常见的匹配规则有：

• 事件顺序规则：如“MACD 出现金叉后 30 秒内 CCI 进入超买区间”
• 事件属性值规则：如“某一成交事件中，成交量超过 50000 股”
• 时间规则：如“收到 A 股票成交回报后1分钟内不再对 A 进行下单”

在 DolphinDB 中，主要通过 addEventListener 函数的各项参数来表达丰富的匹配规则。

监视器是 CEP 系统中负责监控并响应事件的组件，它包含了完整的业务逻辑。

在 DolphinDB 中，使用类（Class）来定义监视器。监视器在创建 CEP 引擎（createCEPEngine）时被实例化。一个监视器中主要包含以下内容：

• 一个或多个事件监听器（Event Listener)。事件监听器用于捕捉特定规则的事件。
• 一个或多个回调函数。回调函数是在捕捉到特定事件后执行的操作。
• 一个 onload 函数。onload 函数可以被视为初始化监视器的操作，通常包括启动一个或多个事件监听器。该函数仅在监视器被实例化时被执行一次。

事件监听器包含了需要捕捉的特定事件规则，并能够进行实时地规则匹配，一旦条件满足就会触发预设的操作。这些条件可以是简单的数据阈值判断，也可以是复杂的模式匹配（序列模式、计时器等）。

在 DolphinDB 中，执行监视器中的 addEventListener 函数即可启动特定的事件监听器。通过 addEventListener 函数不同的参数配置可以灵活地描述匹配规则。

假设 CEP 引擎具体的使用情景为量化交易，则监视器（Monitor）可以被视为一个包含一个或多个交易逻辑的容器，而事件监听器（Event Listener）则是容器中用于检测特定市场条件的组件。在实际应用中，一个监视器可能包含多个事件监听器，每个监听器负责监测一种特定类型的市场活动或条件。当所有相关条件都满足时，交易策略（即监视器）将执行预定的交易动作，如买入、卖出或调整仓位。

在这个初级案例中，我们将对代码进行详细的解释，以帮助用户理解代码逻辑。而在后续的案例中，会减少一些不必要的代码解释。

CEP 引擎的处理对象为事件，因此要先定义引擎内需要的事件。此处定义了一个名为 StockTick 的事件类，这个类包含股票的名称（name）和价格（price）两个属性，用来代表市场上股票的价格变动信息。

//定义一个事件类，表示每个tick的股票行情事件
class StockTick{
    name :: STRING 
    price :: FLOAT 
    def StockTick(name_, price_){
        name = name_
        price = price_
    }
}

• class StockTick ：定义一个名为 StockTick 的类，用来表示从市场数据源接收的股票行情。
• name :: STRING ：在类中定义一个名为 name 的字符串属性，用于存储股票的名称。
• price :: FLOAT：在类中定义一个名为 price 的浮点属性，用于存储股票的价格。
• def StockTick(name_, price_)：类的构造函数，用于实例化一个 StockTick 对象。

其次，定义一个名为 SimpleShareSearch 的监视器类，该监视器通过变量 newTick 存储最新接收到的 StockTick 事件。同时，监视器在启动时会调用 onload 函数，随即注册一个事件监听器，它会捕获所有的 StockTick 事件，并在回调函数 processTick 中处理这些事件。处理过程中，监视器将接收到的股票信息记录到日志中。

class SimpleShareSearch {
	//保存最新的 StockTick 事件
	newTick :: StockTick 
	def SimpleShareSearch(){
		newTick = StockTick("init", 0.0)
	}
	def processTick(stockTickEvent)
	
	def onload() {
		addEventListener(handler=processTick, eventType="StockTick", times="all")
	} 
	def processTick(stockTickEvent) { 
		newTick = stockTickEvent
		str = "StockTick event received" + 
			" name = " + newTick.name + 
			" Price = " + newTick.price.string()
		writeLog(str)
	}
}

• class SimpleShareSearch：定义一个名为 SimpleShareSearch 的监视器类。
• newTick :: StockTick：定义一个属性 newTick，类型为 StockTick，用于存储最新接收到的 StockTick 事件。
• def SimpleShareSearch()：类的构造函数。
• def onload()：监视器必须包含的方法，当 CEP 引擎被创建时将会实例化监视器，这将调用 onload 方法对监视器进行初始化。在本例中启动了一个事件监听器 。
• addEventListener(handler=processTick, eventType="StockTick", times="all")：该事件监听器在 onload中注册，在创建引擎时立刻开始监听所有类型为 StockTick 的事件，并对所有 StockTick 事件使用processTick方法处理。times 参数的默认设置为 "all"，表示监听并响应每一次匹配的事件。如果 times=1，事件监听器将在找到一个对应事件后终止 。更为详细的参数设置规则可以参考相应文档。
• def processTick(stockTickEvent)：定义事件处理方法，接收一个 StockTick 类型的事件作为参数。类的成员方法需要至少先声明或者定义之后才能被调用，因此需要先声明processTick方法以防止onload中的addEventListener函数调用报错。在收到 StockTick 事件之后执行processTick，将在日志中记录 StockTick 相关信息。
• newTick = stockTickEvent：更新监视器中的属性 newTick 为最新接收到的事件。在本例中，这一操作并不是必选的，仅仅为了展示监视器中可以存储运行过程中的变量这一功能。例如，基于此功能可以扩展本例为同时打印出上一条行情与最新的行情。

创建一个 CEP 引擎，配置好相关的表结构和监视器，以便引擎能够接收和处理 StockTick 类型的事件。

dummyTable = table(array(STRING, 0) as eventType, array(BLOB, 0) as eventBody)
try {dropStreamEngine(`simpleMonitor)} catch(ex) {}
createCEPEngine(name="simpleMonitor", monitors=<SimpleShareSearch()>, 
dummyTable=dummyTable, eventSchema=[StockTick])

• dummyTable：包含 eventType 和 eventBody 两列的表作为引擎的输入表，这个表将被用作 CEP 引擎创建时的 dummyTable 参数。在 2.4 小节将进一步介绍该参数的含义。
• try {dropStreamEngine(`simpleMonitor)} catch(ex) {}：这是 CEP 引擎脚本中常见的清理环境方法。若已存在同名引擎，则引擎无法被成功创建，因此在创建引擎前尝试删除叫 simpleMonitor 的流引擎，并捕获可能因为引擎并不存在而返回的异常。
• createCEPEngine：创建 CEP 引擎，命名为 simpleMonitor。同时，指定引擎将使用 SimpleShareSearch 类的实例作为其监视器，dummyTable 为输入表，最后指定引擎将输入表中的数据解析为 StockTick 事件类。

在 DolphinDB 中可以通过两种不同的方法，向 CEP 引擎注入事件：

• appendEvent(engine, events)：直接将事件实例写入 CEP 引擎的方法。
• append! / tableInsert / insert into：通过向 CEP 引擎写入一张表的方式来向引擎输入数据。这种方式在事件注入引擎之前，需要先将其序列化为 BLOB 格式写入异构流数据表中，再通过上述几个函数之一向引擎输入数据。因此，在上文中创建的 CEP 引擎输入表 dummyTable 中进行了相应的设置，eventType 列储存事件类型，eventBody 列储存对应事件的序列化结果。该方法的好处是可以将多个不同的事件类型写入到一个异构流数据表中。然后通过订阅该异构流数据表，向引擎输入事件数据。

向 CEP 引擎输入数据，查看日志并检查引擎运行结果是否符合预期。

stockTick1 = StockTick('600001',6.66)
getStreamEngine(`simpleMonitor).appendEvent(stockTick1)
stockTick2 = StockTick('300001',1666.66)
getStreamEngine(`simpleMonitor).appendEvent(stockTick2)

• 在上面的代码中，模拟了2个 StockTick 事件。事件1的股票代码为600001，股价为6.66；事件2的股票代码为300001，股价为1666.66元。
• 通过 getStreamEngine 函数返回名为"simpleMonitor"的 CEP 引擎的句柄，然后通过 appendEvent 方法直接向 CEP 引擎写入两个事件实例。
• 在 DolphinDB 的 Web 端选择日志查看并在页面右上角刷新，即可看到 CEP 引擎已经识别了输入的事件，并且执行了写入日志的操作。

图2-1

1. 定义 CEP 引擎将要监听或处理的事件类。在本例中，定义了一个有名称和价格两个属性的 StockTick 类作为监控的对象，用来代表市场上股票的价格变动信息。
2. 定义一个监视器类，其中必须包含 onload方法，该方法将在创建 CEP 引擎时被首先调用。本例中 onload 方法中调用了 addEventListener函数，即在引擎创建时就启动监听，并在每一次监听到对应的事件后交由指定回调函数处理。
3. 定义回调函数，即处理事件的方法。本例中该函数将股票价格变动写入了日志。
4. 创建 CEP 引擎。在本例中，创建引擎时对命名了引擎，指定了引擎的监视器（monitors）、输入表结构（dummyTable）和要处理的事件类型（eventSchema）。
5. 模拟事件数据并输入引擎，查看日志检查结果是否符合设计预期。

至此，便实现了对市场行情的实时监控，并且以一定的格式记录到了日志中。通过这种方式，DolphinDB 的 CEP 引擎提供了一种高效的机制来实时监测和响应事件流中的事件，使得用户可以根据实时数据变动快速地进行应对，而上文仅仅是一个简单的金融领域的例子。

在上面的例子中，我们对 CEP 引擎和它的关键概念、基本的代码运行逻辑有了初步的了解，并且可以实现单一事件流监控并记录日志的操作。但这样的单一类型流数据计算和监控用 DolphinDB 已有的流计算框架也能够实现，那么还为什么需要额外的 CEP 引擎呢？

因为上文中的案例不涉及复杂的事件匹配规则。当所需要处理的事件类型越多、越复杂，所需监控的事件模式和特定模式下需执行的响应操作越多、越复杂，则在 CEP 引擎框架下，通过 DolphinDB 脚本直接进行事件描述和匹配等操作进行的程序开发效率就越高。CEP 引擎既保留了流计算引擎高吞吐、低时延、流批一体等强大优势，又在灵活性与事件描述语言等方面进行了改进。

基于股票逐笔成交数据，策略将根据每支股票最新的成交价涨幅和累计成交量判断是否执行下单或撤单等操作，具体流程如下所示。

图3-1

具体的策略判断逻辑细节如下：

• 根据每一笔成交数据触发计算两个实时因子：最新成交价相对于15秒内最低成交价的涨幅 （变量 ROC）和过去1分钟的累计成交量 （变量 volume ）；
• 在策略启动时设定每支股票的两个因子阈值（ ROC0、volume0 ），每当实时因子值更新后判断是否 ROC > ROC0 且 volume > volume0。
  • 若是，则触发下单；
  • 若下单后1分钟内仍未成交，则触发对应的撤单。

比案例 I 更进一步之处

• 在 CEP 引擎内使用了响应式状态引擎（Reactive State Engine, RSE）计算因子。 响应式状态引擎支持实时数据的高性能增量计算。在本例中的 CEP 引擎内部，使用了响应式状态引擎及其内置的状态函数对两个因子值进行计算。
• 向外部系统发送事件。 在策略执行的过程中，当满足某些具体的条件时，CEP 引擎内部需要将特定事件（本例中为下单和撤单事件）通过 emitEvent 接口发送到 CEP 引擎外部。
• 使用了超时计时器。 在 addEventListener 函数中有关于事件监听器触发方式和触发次数的可选参数设置。通过这些可选参数可以实现不同的监听触发效果，如事件匹配、计时器、定时器等。本例在每次下单后开启一个新的计时器，若计时超时则触发对应操作。

在本小节中将详细介绍案例的代码实现，完整代码附件见文末附录。

def onload() {
    // 初始化 data view
    initDataView()
    ......  
}

• 在 onload 中调用初始化 DataView 的函数，然后在监视器内完善 initDataView 方法。

def initDataView(){
    share(streamTable(1:0, `securityid`strategyid`ROCThreshold
`volumeThreshold`ROC`volume`newOrderNum`newOrderAmount`executionOrderNum
`executionAmount`timeoutOrderNum`updateTime, 
`STRING`INT`INT`INT`INT`INT`INT`DOUBLE`INT`DOUBLE`INT`TIMESTAMP), "strategyDV")
    dataview = createDataViewEngine(name="Strategy_"+strategyid, 
outputTable=objByName(`strategyDV), keyColumns=`securityId, timeColumn=`updateTime) 
    num = strategyParams.size()
    securityids = strategyParams.keys()
    ROCThresholds = each(find{,"ROCThreshold"}, strategyParams.values())
    volumeThresholds = each(find{,"volumeThreshold"}, strategyParams.values()) 
    dataview.tableInsert(table(securityids, take(self.strategyid, num) as 
strategyid, ROCThresholds, volumeThresholds, take(int(NULL), num) as ROC, 
take(int(NULL), num) as volume, take(0, num) as newOrderNum, 
take(0, num) as newOrderAmount, take(0, num) as executionOrderNum, 
take(0, num) as executionAmount, take(0, num) as timeoutOrderNum))
}

• 函数 initDataView 首先创建并共享了一个叫”strategyDV”的流数据表，用作 DataViewEngine 的输出表。该表规定了输出表的结构，指定了希望监控的相应数据。
• 通过 createDataViewEngine 创建了引擎，设定了引擎键值，对于每个键值，引擎都只保留最新的 1 条数据。
• find 函数从每个股票的策略参数字典中查找 ROCThreshold 和 volumeThreshold 键的值。
• each 函数把 find 函数应用到每个股票的策略参数字典上，为每个股票提取出策略因子的阈值。
• 最后将提取的阈值和初始状态值插入到 DataView 的表中。

updateDataViewItems 可以更新 DataView 引擎中指定键值对应的指定属性的值。它被多次用在不同的方法中。

• 更新超时订单数量：当一个成交回报事件超时未收到时，函数会增加超时订单的数量并更新相应的 DataView 。

def execReportExceedTimeHandler(orderid, exceedTimeSecurityid){
	emitEvent(CancelOrder(orderid)) // 发送撤单事件到外部
	timeoutOrderNum = (exec timeoutOrderNum from self.dataview 
	where securityid=exceedTimeSecurityid)[0] + 1
	updateDataViewItems(engine=self.dataview, keys=exceedTimeSecurityid,
	 valueNames=`timeoutOrderNum, newValues=timeoutOrderNum) // 更新data view
}

• 更新成交金额和订单数量：先计算给定 securityid 的成交金额和订单数量，再用updateDataViewItems方法更新它们。

def execReportHandler(execReportEvent) {
	executionAmount = (exec executionAmount from self.dataview 
	where securityid=execReportEvent.securityid)[0] + 
	execReportEvent.price*execReportEvent.volume
	executionOrderNum = (exec executionOrderNum from self.dataview 
	where securityid=execReportEvent.securityid)[0] + 1
	updateDataViewItems(engine=self.dataview, keys=execReportEvent.securityid, 
	valueNames=["executionAmount","executionOrderNum"], 
	newValues=(executionAmount,executionOrderNum)) // 更新data view		
}

• 更新 ROC 和成交量数据：当因子计算引擎输出结果时，更新相关的 ROC 和成交量数据。

def handleFactorCalOutput(factorResult){
    factorSecurityid = factorResult.securityid[0]
    ROC = factorResult.ROC[0]
    volume = factorResult.volume[0]
    lastPrice = factorResult.lastPrice[0] 
    updateDataViewItems(engine=self.dataview, keys=factorSecurityid, 
    valueNames=["ROC","volume"], newValues=(ROC,volume))
    ......
}

• 更新新订单数量和金额：在创建新订单时，更新订单数量和金额。

if (ROC>strategyParams[factorSecurityid][`ROCThreshold] && volume
>strategyParams[factorSecurityid][`volumeThreshold]) {
    ...
    newOrderNum = (exec newOrderNum from self.dataview 
where securityid=factorSecurityid)[0] + 1
    newOrderAmount = (exec newOrderAmount from self.dataview 
where securityid=factorSecurityid)[0] + lastPrice*0.98*10
    updateDataViewItems(engine=self.dataview, keys=factorSecurityid, 
valueNames= ["newOrderNum", "newOrderAmount"], 
newValues=(newOrderNum, newOrderAmount)) // 更新data view
    ...
}

在策略运行时，某一时刻的 DataView 情况如下图所示。

图4-1