Why Stream Processing Matters — and How DolphinDB Does It

8 min read Original article ↗

DolphinDB

Press enter or click to view image in full size

In modern financial systems, when you compute is often as important as what you compute.

Finding the highest price over the past three months is a classic analytical task. But detecting abnormal spreads, aggressive trading behavior, or intraday price patterns as they happen is what truly differentiates a real-time trading or risk system from a traditional analytics platform.

This is where stream processing becomes indispensable.

DolphinDB is designed to excel in both historical batch analytics and real-time stream processing, enabling users to move seamlessly between backtesting and live production without changing their data model or programming paradigm. In this article, we explore DolphinDB’s streaming framework through practical, real-world examples — from stateless calculations to stateful windowed aggregations — and explain how it unifies stream and batch processing in a single system.

Why Stream Processing Matters

Batch processing works well when data is finite and analysis is retrospective. However, modern market data is:

  • Continuous — tick-by-tick quotes and trades never stop
  • High-frequency — updates arrive every millisecond or faster
  • Latency-sensitive — delays of milliseconds can invalidate decisions

Stream processing addresses these challenges by analyzing data incrementally as it arrives, instead of waiting for a complete dataset.

Batch vs. Stream Processing at a Glance

Press enter or click to view image in full size

DolphinDB treats stream processing not as an add-on, but as a first-class computing model, sharing the same language, data structures, and optimization engine as batch analytics.

What is Stream Processing?

Streaming data refers to a continuous flow of time-series data generated by ongoing events. Unlike static, bounded historical data, streaming data has several defining characteristics:

  • Dynamic: Continuous data flow with no predetermined end or fixed schema.
  • Ordered: Records often contain timestamps or sequence numbers for ordering.
  • Large-scale: Streaming data is often generated at high velocity, thus requiring high-performance parallel processing and scalable infrastructure.
  • Real-time: Real-time nature requires ultra-low latency processing to deliver immediate insightsts.

Stream processing analyzes incoming data incrementally in real-time rather than waiting to process entire datasets. This approach optimizes resource utilization and enables rapid decision-making.

Example 1: Real-Time Bid–Ask Spread Calculation

Let’s start with a simple but common market microstructure metric: bid–ask spread.

Calculation Logic

The bid-ask spread is defined as the difference between the best offer price (offerPrice0) and best bid price (bidPrice0) divided by the average price, calculated as:

An example of the input data is as shown below.

Batch Processing

The calculation logic can be easily scripted with SQL queries based on the formula.

// Mock input data
tick = table(1:0, `securityID`dateTime`bidPrice0`bidOrderQty0`offerPrice0`offerOrderQty0, 
[SYMBOL, TIMESTAMP, DOUBLE, LONG, DOUBLE, LONG])
insert into tick values(`000001, 2023.01.01T09:30:00.000, 19.98, 100, 19.99, 120)
insert into tick values(`000001, 2023.01.01T09:30:03.000, 19.96, 130, 19.99, 120)
insert into tick values(`000001, 2023.01.01T09:30:06.000, 19.90, 120, 20.00, 130)
// Calculate bid-ask spread
select securityID, dateTime, (offerPrice0 - bidPrice0) * 2 / (offerPrice0 + bidPrice0) as factor from tick

Output:

Stream Processing

Unlike batch processing, stream processing operates on continuously growing data streams. Calculation for the bid-ask spread is triggered each time when new record arrives, instead of all at once. Therefore, stream processing introduces two key concepts:

  • Stream Table: Stores and publishes data streams.
  • Subscription: Enables immediate data processing upon receiving new records.
// Create input stream table
share(table=streamTable(1:0, `securityID`dateTime`bidPrice0`bidOrderQty0`offerPrice0`offerOrderQty0, [SYMBOL,TIMESTAMP,DOUBLE,LONG,DOUBLE,LONG]), sharedName=`tick)
// Create output stream table
share(table=streamTable(1:0, ["securityID", "dateTime", "factor"], [SYMBOL, TIMESTAMP, DOUBLE]), sharedName=`resultTable)
go
// Define processing function
def factorCalFunc(msg){
    tmp = select securityID, dateTime, (offerPrice0 - bidPrice0) * 2 / (offerPrice0 + bidPrice0) as factor from msg
    objByName("resultTable").append!(tmp)    
}
// Subscribe to the table tick
subscribeTable(tableName="tick", actionName="factorCal", offset=-1, handler=factorCalFunc, msgAsTable=true)

This script first creates a table tick as the publishing table, and then subscribes to tick using subscribeTable. The subscription task processes incoming data with UDF factorCalFunc (where bid-ask spread is defined) and outputs results to resultTable each time new record is inserted into the tick table.

Example 2: Active Buy Volume Ratio

Not all streaming metrics are stateless. The active buy volume ratio depends on historical context within a rolling time window, making it a stateful computation.

Calculation Logic

The active buy volume ratio is defined as the ratio of active buy volume to total trade volume, calculated as follows:

Press enter or click to view image in full size

where actVolumet represents the total volume of trades executed by aggressive buyers within the interval from t-window to t, and totalVolumet represents the total volume of the trades executed within the same interval.

The signal function I is defined as follows:

Press enter or click to view image in full size

An example of the input data is as shown below.

Streaming Implementation

The active buy volume ratio is calculated with a 5-minute sliding window, where each new record triggers an aggregation of data within that window. Different from the bid-ask spread calculation, this case requires both the latest input record and historical data within the specified timeframe. The following concepts are introduced:

  • Incremental Computation: Instead of recalculating the data for each 5-minute window, the total trade volume can be updated following the incremental computation logic: Total Volume(t) = Total Volume(t-1) + Current Volume — Expired Volume.
  • Stateful Computation: Previous calculation results and data are cached for subsequent calculations, and they are collectively known as states. Since the active buy volume ratio depends on both incoming records and past data, its calculation is stateful. In contrast, computing the bid-ask spread based solely on each new record is stateless.
  • Streaming Engines: DolphinDB provides streaming engines that internally deal with incremental computing optimization and state caching.
// Create input stream table
share(table=streamTable(1:0, `securityID`tradeTime`tradePrice`tradeQty`tradeAmount`buyNo`sellNo, [SYMBOL, TIMESTAMP, DOUBLE, INT, DOUBLE, LONG, LONG]), sharedName=`trade)
// Create output stream table
share(table=streamTable(1:0, ["securityID", "tradeTime", "factor"], [SYMBOL, TIMESTAMP, DOUBLE]), sharedName=`resultTable)
go
// Create a reactive state engine
createReactiveStateEngine(name="reactiveDemo", 
metrics=<[tradeTime, tmsum(tradeTime, iif(buyNo>sellNo, tradeQty, 0), 5m)\tmsum(tradeTime, tradeQty, 5m)]>, dummyTable=trade, outputTable=resultTable, keyColumn="securityID")
// Subscribe to table trade
subscribeTable(tableName="trade", actionName="factorCal", offset=-1, handler=getStreamEngine("reactiveDemo"), msgAsTable=true)

This script first creates trade as the publishing table and subscribes to it, allowing the data to be processed in real time by the reactive state engine. The metrics of the engine define how to calculate active buy volume ratio with tmsum function (which is incrementally optimized within the engine).

The following script inserts records into the trade table for demonstrating how the data is consumed in the engine.

insert into trade values(`000155, 2020.01.01T09:30:00.000, 30.85, 100, 3085, 4951, 0)
insert into trade values(`000155, 2020.01.01T09:31:00.000, 30.86, 100, 3086, 4952, 1)
insert into trade values(`000155, 2020.01.01T09:32:00.000, 30.85, 200, 6170, 5001, 5100)
insert into trade values(`000155, 2020.01.01T09:33:00.000, 30.83, 100, 3083, 5202, 5204)
insert into trade values(`000155, 2020.01.01T09:34:00.000, 30.82, 300, 9246, 5506, 5300)
insert into trade values(`000155, 2020.01.01T09:35:00.000, 30.82, 500, 15410, 5510, 5600)
insert into trade values(`000155, 2020.01.01T09:36:00.000, 30.87, 800, 24696, 5700, 5600)

Output in the resultTable:

This section provides a preliminary understanding of stateful factor stream computing in DolphinDB through the streaming engines. However, the calculations so far have been event-driven, meaning each incoming record triggers an immediate computation. In the next sections, we will explore how to implement computations at fixed time intervals instead of responding to every event.

Time in Streaming: Event Time vs. System Time

Time is one of the most subtle — and critical — concepts in stream processing.

  • System Time: When the engine processes the record
  • Event Time: When the event actually occurred

DolphinDB allows users to choose between them via useSystemTime.

  • System time → lower latency
  • Event time → consistent results between live trading and historical replay

Example 3: Real-Time 1-Minute OHLC Aggregation

OHLC bars are a cornerstone of trading systems, yet tricky in streaming scenarios.

The 1-minute OHLC is calculated with sliding windows. Unlike calculations introduced in the previous sections, data within a timeframe is aggregated and output one record.

Implementation

// Create input stream table
share(table=streamTable(1:0, `securityID`tradeTime`price`qty, [SYMBOL,TIMESTAMP,DOUBLE,INT]), sharedName=`trade)
// Create output stream table
share(table=streamTable(1:0, `tradeTime`securityID`open`high`low`close, [TIMESTAMP,SYMBOL,DOUBLE,DOUBLE,DOUBLE,DOUBLE]), sharedName=`OHLC)
go
// Create time-series engine
createTimeSeriesEngine(name="timeSeriesDemo", windowSize=60000, step=60000, metrics=<[first(price),max(price),min(price),last(price)]>, dummyTable=trade, 
outputTable=OHLC, timeColumn=`tradeTime, useSystemTime=false, keyColumn=`securityID)
// Subscribe to table trade
subscribeTable(tableName="trade", actionName="OHLCCal", offset=-1, handler=getStreamEngine("timeSeriesDemo"), msgAsTable=true)

This script first creates a table trade as the publishing table, and subscribes to the table to ingest the data to a time-series engine. The metrics implement 1-minute OHLC with four built-in functions, first, max, min, and last, which are optimized with incremental algorithm. Additionally, windows are determined based on event time (from timeColumn) in this case.

The following script inserts some records into the trade table for demonstrating how the data is consumed in the engine.

insert into trade values(`000155, 2020.01.01T09:30:10.000, 9.76, 100)
insert into trade values(`000155, 2020.01.01T09:30:40.000, 9.73, 100)
insert into trade values(`000155, 2020.01.01T09:31:00.000, 9.74, 100)
insert into trade values(`000155, 2020.01.01T09:31:10.000, 9.80, 200)
insert into trade values(`000155, 2020.01.01T09:31:20.000, 9.83, 100)
insert into trade values(`000155, 2020.01.01T09:32:10.000, 10.02, 500)

Output in the resultTable:

Conclusion

Stream processing is no longer optional in modern financial systems — it is foundational.

DolphinDB’s streaming framework provides:

  • Stateless and stateful stream computation
  • Built-in incremental optimization
  • Flexible time semantics
  • Unified stream–batch processing

By abstracting away the complexity of state management and windowing, DolphinDB allows users to focus on what to compute, not how to manage the stream.

Learn more about us: https://dolphindb.com/

Thanks for your reading! To keep up with our latest news, please follow our Twitter @DolphinDB_Inc and Linkedin. You can also join our Slack to chat with the author! 😉