Introduction to CEP Engine: Building a Simple Price-Volume Strategy

In the DolphinDB CEP system, an event is a data representation of a specific occurrence or state within a given time frame, essentially a collection of attribute values describing a change in an object. For example, "buying 1 share of AAPL stock at $170.85" and "BRK rose 5% on June 12, 2024" are both events. Events are continuously generated and exist as streams.

When defining an event, we must specify both the event type and the name and type of each attribute it includes. We use classes to represent events, where the class name corresponds to the event type, and the instances of the class represent specific occurrences of the event. Event streams can be stored in stream tables, with each row in the table representing a distinct event.

Complex event processing extends beyond individual events to incorporate sophisticated matching rules and filters across event streams. Common matching rules include:

• Event Sequence Rules: For example, "CCI enters the overbought zone within 30 seconds after MACD shows a golden cross"
• Event Property Value Rules: For example, "A trading event where the trading volume exceeds 50,000 shares"
• Time-based Rules: For example, "After receiving a trade confirmation for Stock A, no new orders for Stock A will be placed within the next minute"

In DolphinDB, these various matching rules are primarily expressed through different parameters of addEventListener.

A monitor is a component in the CEP system responsible for monitoring and responding to events, containing the complete business logic.

Monitors are defined using classes. A monitor is instantiated when creating a CEP engine using createCEPEngine. It contains the following elements:

• One or more event listeners. Event listeners are used to capture events based on specific rules.
• One or more callback functions. These are operations executed after specific events are captured.
• An onloadmethod. It is executed only once when the monitor is instantiated, typically including the startup of one or more event listeners.

An event listener contains specific event rules that need to be captured. It can perform real-time rule matching, triggering preset operations once conditions are met. These conditions range from simple threshold checks to complex market pattern detection - such as sequence-based rules or time-based constraints.

Event listeners for specific events are started by executing addEventListener within the monitor. Different parameter configurations in addEventListener allow for flexible description of matching rules.

For example, in quantitative trading, the monitor can be viewed as a container for trading logic, while event listeners are components within the container used to detect specific market conditions. A monitor may contain multiple event listeners, with each listener responsible for monitoring a specific type of market activity or condition. When all conditions are met, the monitor (i.e., trading strategy) will execute predefined trading actions such as buying, selling, or adjusting positions.

In this basic case study, we'll provide detailed explanations of the code to help you understand the logic. In the subsequent case, we'll focus on more advanced code concepts with fewer step-by-step explanations.

The CEP engine processes events, so we need to first define the events. Here, we define an event class called StockTick, which contains two attributes: the stock name and price, representing price movement information in the market.

// Define an event class representing stock quote events for each tick
class StockTick{
    name :: STRING 
    price :: FLOAT 
    def StockTick(name_, price_){
        name = name_
        price = price_
    }
}

• class StockTick: Defines the StockTick class, used to represent stock ticks received from market data sources.
• name :: STRING: A STRING attribute within the class, used to store the stock name.
• price :: FLOAT: A FLOAT attribute within the class, used to store the stock price.
• def StockTick(name_, price_): Defines the class constructor, used to instantiate a StockTick object.

Next, define a monitor class called SimpleShareSearch. This monitor stores the latest StockTick event in a variable newTick. When the monitor is initialized, it calls the onload method, which registers an event listener to capture all StockTick events and process them in the callback processTick. processTick updates the class with the latest event data and logs details about the stock event.

class SimpleShareSearch {
	// Stores the latest StockTick event
	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: Defines a monitor class named SimpleShareSearch.
• newTick :: StockTick: An attribute newTick of type StockTick, used to store the latest StockTick event.
• def SimpleShareSearch(): Defines the constructor of the class.
• def onload(): A required method for the monitor. When the CEP engine is created, it instantiates the monitor and calls onload to initialize the monitor.. In this example, an event listener is started.
• addEventListener(handler=processTick, eventType="StockTick", times="all"): Registers an event listener within the onload method. When the engine is created, it immediately starts listening for events of type StockTick and processes each matching event with processTick. The default setting for the times parameter is "all", indicating that the listener will respond to every matching event. If times=1, the event listener will stop after processing the first matching event. For detailed parameter settings, refer to the function documentation.
• def processTick(stockTickEvent): This method handles incoming StockTick events, logging their details when received. As with any class member method, processTick must be declared before it's used - in this case, before the addEventListener call in onload.
• newTick = stockTickEvent: Updates the monitor's newTick attribute with the latest event. In this case, this step is not strictly necessary but is included to demonstrate that a monitor can store variables during runtime. This capability could be used to compare previous and current stock quotes

Create a CEP engine, configure the table schema and monitor, so that the engine can receive and process StockTick events.

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: A table with two columns, eventType and eventBody. It is specified as the dummyTable parameter in createCEPEngine, serving as the input table for the CEP engine. The dummyTable parameter will be further explained in section 2.4.
• try {dropStreamEngine(simpleMonitor) } catch(ex) {}: This is a common environment cleanup approach in CEP engine applications. If an engine with the same name already exists, the engine creation will fail. Therefore, before creating the engine, an attempt is made to drop the existing stream engine named simpleMonitor, and any exceptions that arise (e.g., if the engine does not exist) are caught.
• createCEPEngine: Creates a CEP engine named simpleMonitor. The engine will use an instance of the SimpleShareSearch class as its monitor, dummyTable as the input table. The engine will interpret the data in the input table as StockTick event objects.

There are two ways for ingesting events into the CEP engine:

• Directly ingesting events into the CEP engine using the appendEvent(engine, events) function, avoiding serialization and deserialization.
• Serialize events into BLOB format and store the data in a heterogeneous stream table. Then append the table to the CEP engine. As shown in the example above, the engine input table, dummyTable, is configured with the eventType column to store event types, and the eventBody column to store the serialized event data. The benefit of this approach is that it allows different event types to be written into a single stream table. Events can then be ingested into the engine through subscription.

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

• In the script above, two StockTick events are simulated. The first event has a stock code of 600001 and a price of 6.66, while the second event has a stock code of 300001 and a price of 1666.66.
• The getStreamEngine function is used to return the handle of the CEP engine simpleMonitor. The appendEvent function is used to directly ingest the two events into the CEP engine.

After ingesting data into the CEP engine, check the logs using the DolphinDB web interface. Refresh page by clicking the Refresh button. We can see that the CEP engine has logged the two events.

Let's revisit the implementation process of this case.

1. Define the event class that the CEP engine will listen to and process. In this example, a StockTick class with two attributes, name and price, is defined.
2. Define a monitor class, which must include the onload method. This method will be called when the CEP engine is created. In this example, onload invokes the addEventListener function to start listening for events immediately after the engine is created. Each time an event is detected, it is handled by a specified callback function.
3. Define the callback function, which processes the events. In this example, the callback logs the stock price changes.
4. Create the CEP engine. In this example, the engine is created with a name, monitor (monitors), input table schema (dummyTable), and the event types to be processed (eventSchema).
5. Simulate event data and ingest it into the engine, then check the logs to verify the results.

This example demonstrates a simple use case of DolphinDB's CEP engine in real-time event monitoring and response. We've covered the basic functionality of the CEP engine, its core concepts, and its execution process.

While DolphinDB's other stream computing frameworks can also process single-type event streams, the CEP engine excels when dealing with more diverse and complex event types, as well as intricate patterns that require monitoring and response. The following case demonstrates how the CEP engine enhances development efficiency in such scenarios.

The strategy determines whether to place or cancel an order based on the latest price change and cumulative trading volume of each stock from tick data. The strategy logic is presented in the diagram below:

Specifically:

• For each tick, two real-time factors are calculated: the price change (denoted as “ROC”) relative to the lowest price within the last 15 seconds, and the cumulative trading volume over the past minute (demoted as “volume“).
• When the strategy initializes, the thresholds (ROC₀ and volume₀) for the two factors per stock are established . As these factors update in real time, the strategy checks if ROC > ROC₀ and volume > volume₀:
  • If both conditions are met, an order is triggered.
  • If the order remains unfilled after one minute, a cancellation is triggered.

Compared with Case I, this case has the following additional complexities:

• The CEP engine incorporates a reactive state engine and state functions to compute factor values. The reactive state engine supports high-performance incremental calculations on real-time data.
• Sending Events to External Systems. During the execution of the strategy, when specific conditions are met, the CEP engine needs to send certain events (in this case, order and cancel events) to external systems through the emitEvent function.
• Using a Timeout Timer. The addEventListener function includes optional parameters for configuring event listener triggers and trigger counts. These optional parameters allow for different types of trigger behaviors, such as event matching, timers, and timeouts. In this example, a new timer is started after each order is placed. If the timer expires, the corresponding operation is triggered.

This section provides a detailed explanation of the core code for the case. The complete script can be found in the appendices.

The onload method calls initDataView to initialize the data view.

def onload() {
    // Initialize data view
    initDataView()
    ...  
}

The definition of initDataView is implemented within the monitor.

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 first creates and shares a stream table called "strategyDV", which serves as the output table for the data view engine. strategyDV defines the schema of the output and specifies the metrics to be monitored.
• The data view engine is created through createDataViewEngine, which sets the engine's key. For each key value, the engine only retains the latest record.
• The each higher-order function applies find to each stock's strategy parameter dictionary, extracting the factor threshold values (“ROCThreshold” and “volumeThreshold”) for each stock.
• Finally, the threshold values and initial values are inserted into the data view.

updateDataViewItems is used to update values of a data view through keys. In this case example, it is used in multiple methods.

• Updating timeout order count: When an execution report event times out, updateDataViewItems increases the number of timeout orders and updates the data view.

  def execReportExceedTimeHandler(orderid, exceedTimeSecurityid){
  	emitEvent(CancelOrder(orderid)) // Send cancel order event externally
  	timeoutOrderNum = (exec timeoutOrderNum from self.dataview 
  	where securityid=exceedTimeSecurityid)[0] + 1
  	updateDataViewItems(engine=self.dataview, keys=exceedTimeSecurityid,
  	 valueNames=`timeoutOrderNum, newValues=timeoutOrderNum) // Update data view
  }

• Updating execution amount and order count: After calculating the execution amount and number of orders for a given securityid, updateDataViewItems updates them in the data view.

  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)) // Update data view   
  }      

• Updating ROC and volume data: When the factor calculation engine outputs results, updateDataViewItems updates the related ROC and volume data.

  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))
      ......
  }

• Updating new order count and amount: When creating new orders, updateDataViewItems updates the number of orders and order amount.

  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)) // Update data view
      ...
  } 

The state of the data view at a certain point during strategy execution is shown in the figure below.