Calculating High-Frequency Factors in Real-Time

This example calculates the ratio between the current best ask price and the best ask price from 30 quotes earlier, requiring retention of at least 30 historical quotes per stock. An in-memory table 'history' is created to store all stock states.

quotesData = loadText(yourDir + "sampleQuotes.csv")

defg factorAskPriceRatio(x){
	cnt = x.size()
	if(cnt < 31) return double()
	else return x[cnt - 1]/x[cnt - 31]
}

def factorHandler(mutable history, mutable factors, msg){
	history.append!(select symbol, askPrice1 from msg)
	syms = msg.symbol
	t = select factorAskPriceRatio(askPrice1) as factor from history where symbol in syms group by symbol
	factors.append!(t.update!("timestamp", now()).reorderColumns!("timestamp"))
}

x=quotesData.schema().colDefs
share streamTable(100:0, x.name, x.typeString) as quotes1
history = table(1000000:0, `symbol`askPrice1, [SYMBOL,DOUBLE])
share streamTable(100000:0, `timestamp`symbol`factor, [TIMESTAMP,SYMBOL,DOUBLE]) as factors
subscribeTable(tableName = "quotes1", offset=0, handler=factorHandler{history, factors}, msgAsTable=true, batchSize = 3000, throttle=0.005)

replay(inputTables=quotesData, outputTables=quotes1, dateColumn=`date, timeColumn=`time)

Key code explanation:

• The user-defined aggregate function factorAskPriceRatio is used to calculate the factor for each stock.
• The function factorHandler is defined as the message handler:
  • Line 10 saves the current batch of messages to the in-memory table history.
  • Line 11 extracts the stock symbols from the current messages.
  • Line 12 calculates the factor for each stock.
  • Line 13 outputs the factors to the factors table.
• Two parameters history and factors that maintain the historical states and generated factors of factorHandler are partially fixed.

Query results:

select top 10 * from factors where isValid(factor)

Partitioned in-memory tables enable parallel factor computation, with SQL queries generating tasks per partition executed by streaming and global worker threads. Maximum parallelism is workerNum + 1 when partitions exceed workerNum.

The SQL syntax for in-memory and partitioned in-memory tables is mostly the same, with the primary difference being the table creation.

history_model = table(1000000:0, `symbol`askPrice1, [SYMBOL,DOUBLE])
syms = format(600000..601000, "000000")
db = database(partitionType=VALUE, partitionScheme=syms)
history = db.createPartitionedTable(table=history_model, tableName=`history, partitionColumns=`symbol)

For value-partitioned tables, when calculating factors with simple logic, you can directly compute the factor in each partition instead of using SQL. The following code redefines the function factorHandler which uses the built-in function getTablet to get all tablets corresponding to all stock symbols, then loops through each stock to calculate the factor. This looped approach, although single-threaded, can outperform SQL when working with a large number of small partitions.

def factorHandler(mutable history, mutable factors, msg){
	history.append!(select symbol, askPrice1 from msg)
	syms = msg.symbol
	tables = getTablet(history, syms)
	cnt = syms.size()
	v = array(DOUBLE, cnt)
	for(i in 0:cnt){
		v[i] = factorAskPriceRatio(tables[i].askPrice1)
	}
	factors.tableInsert([take(now(), cnt), syms, v])
}

The following example uses a dictionary with STRING keys and tuple values, storing quote data per stock. The dictUpdate! function is used to update this dictionary, and then the factor for each stock is calculated in a loop. Since each stock's historical data is stored separately, no grouping is required, making this method highly efficient for large-scale or per-stock processing.

defg factorAskPriceRatio(x){
	cnt = x.size()
	if(cnt < 31) return double()
	else return x[cnt - 1]/x[cnt - 31]
}
def factorHandler(mutable historyDict, mutable factors, msg){
	historyDict.dictUpdate!(function=append!, keys=msg.symbol, parameters=msg.askPrice1, initFunc=x->array(x.type(), 0, 512).append!(x))
	syms = msg.symbol.distinct()
	cnt = syms.size()
	v = array(DOUBLE, cnt)
	for(i in 0:cnt){
	    v[i] = factorAskPriceRatio(historyDict[syms[i]])
	}
	factors.tableInsert([take(now(), cnt), syms, v])
}

x=quotesData.schema().colDefs
share streamTable(100:0, x.name, x.typeString) as quotes1
history = dict(STRING, ANY)
share streamTable(100000:0, `timestamp`symbol`factor, [TIMESTAMP,SYMBOL,DOUBLE]) as factors
subscribeTable(tableName = "quotes1", offset=0, handler=factorHandler{history, factors}, msgAsTable=true, batchSize=3000, throttle=0.005)

replay(inputTables=quotesData, outputTables=quotes1, dateColumn=`date, timeColumn=`time)

Query results:

select top 10 * from factors where isValid(factor)

Each method has distinct advantages suited to different scenarios based on factor complexity, data volume, and performance needs:

• In-Memory Table: Simple to use and allows factor calculation via SQL. However, it has lower performance, especially for per-stock calculations.
• Partitioned In-Memory Table: Enables parallel computation, offering better efficiency than regular in-memory tables, especially for larger datasets. It can handle both simple and complex factors, offering a balance between SQL-based calculations and direct partition-based calculations.
• Dictionary: Most efficient for simple factors, excelling at both bulk and individual stock processing, though less suitable for complex factor calculations.

Real-time factor calculation frequently requires processing grouped by stock. Pre-grouping data by stock can substantially reduce computation time, achievable through dictionaries and partitioned in-memory tables (as detailed in Section 5).

We tested three state storage methods with 4,000 stocks, each maintaining 300 historical records. The batch processing time (including data insertion, factor calculation, and result insertion) for 4,000 messages (one per stock) was measured as:

• Regular in-memory tables: 180 ms
• Partitioned in-memory tables: 32 ms
• Dictionaries: 20 ms

This shows dictionaries and partitioned in-memory tables significantly enhance computation efficiency. However, two considerations arise when using these methods:

• Handling New Stock Symbols: Intraday streaming may introduce new stocks not accounted for during initialization.
  • For dictionaries, we can use the initFunc parameter of the dictUpdate! function to specify an initialization function. For example, in the factorHandler definition, use the following lambda function to create a pre-allocated vector for 512 elements.

    x->array(x.type(), 0, 512).append!(x)

  • For value-partitioned in-memory tables, you can configure newValuePartitionPolicy=add to automatically create new value partitions.
• Grouping by Multiple Fields: Currently, neither dictionary nor partitioned in-memory tables support composite fields. A workaround is to concatenate multiple fields into a single string field.

High-frequency calculations demand low latency, but processing messages individually reduces throughput. The subscribeTable function provides two optional parameters to adjust this balance:

• throttle: Maximum wait time (in seconds) before processing a batch of messages (minimum: 0.001s).
• batchSize: Maximum number of messages accumulated before processing.

When either condition is met, unprocessed messages are sent to the task queue for handling by the specified message handler.

Parallel computing increases system throughput, enabling more messages to be processed per unit of time. In streaming computations, parallelism can be implemented in two ways.

For message processing functions containing complex logic with loops and conditional statements that cannot be vectorized, DolphinDB's Just-In-Time (JIT) compilation version provides substantial performance improvements. For details, refer to DolphinDB Tutorial: JIT Compilation.

The following example demonstrates JIT compilation for factor calculations requiring analysis of the top 5 bid/ask levels and corresponding order book data. Since this processing cannot be vectorized, JIT compilation offers optimal performance enhancement.

@jit
 def sum_diff(x, y){
     return 1.0 * (x-y)/(x+y) 
 }
 
 @jit
 def wbvol(bp, bv, price, jump) {
     return exp(0.6*(bp-price)/jump) * bv
 }
 
 @jit
 def wavol(ap, av, price, jump) {
     return exp(0.6 * (price - ap)/jump) * av
 }
 
 @jit
 def factor1(ap1, ap2, ap3, ap4, ap5, av1, av2, av3, av4, av5, bp1, bp2, bp3, bp4, bp5, bv1, bv2, bv3, bv4, bv5, mp, initMP, delta){
     n = ap1.size()
     re = array(DOUBLE, n)
     for(i in 0:n){
         jump = ceil(initMP[i] * 15.0 / 100.0) / 100.0
         w_av1 = 0.0  
         w_bv1 = 0.0  
         w_av1 = wavol(ap1[i], av1[i], mp[i], jump) + wavol(ap2[i], av2[i], mp[i], jump) + wavol(ap3[i], av3[i], mp[i], jump) + wavol(ap4[i], av4[i], mp[i], jump) + wavol(ap5[i], av5[i], mp[i], jump)
         w_bv1 = wbvol(bp1[i], bv1[i], mp[i], jump) + wbvol(bp2[i], bv2[i], mp[i], jump) + wbvol(bp3[i], bv3[i], mp[i], jump) + wbvol(bp4[i], bv4[i], mp[i],jump) + wbvol(bp5[i], bv5[i], mp[i], jump)
         if(delta[i]>0){
             re[i] = sum_diff(w_bv1*1.1, w_av1)
         }else{
             re[i] = sum_diff(w_bv1, w_av1 * 1.1)
         }		
     }
     return re
 }

 // message handler
def factor1_handler(mutable d, mutable factor, msg){
     start = now(true)
     n = msg.size()
     mp = (msg.bidPrice1+msg.askPrice1)/2
     dictUpdate!(d,append!, msg.symbol, mp)
     delta = array(DOUBLE, n)
     initMP = array(DOUBLE, n)
     sym = msg.symbol
     for(i in 0:n){
         &his_mp = d[sym[i]]
         initMP[i] = his_mp[0]
         delta[i] = nullFill(his_mp.tail()-his_mp.tail(5).avg(), 0)
     }
     factorValue = factor1(msg.askPrice1, msg.askPrice2, msg.askPrice3, msg.askPrice4, msg.askPrice5, msg.askVolume1, msg.askVolume2, msg.askVolume3, msg.askVolume4, msg.askVolume5, msg.bidPrice1, msg.bidPrice2, msg.bidPrice3, msg.bidPrice4, msg.bidPrice5, msg.bidVolume1, msg.bidVolume1, msg.bidVolume1, msg.bidVolume1, msg.bidVolume1, mp, initMP, delta)
     factor.tableInsert(take(start,n), take(now(true), n), sym, take("factor1", n), factorValue)
}

def clear(){
	try{
	unsubscribeTable(, `Trade, `act_factor)
	undef(`Trade, SHARED)
	undef(`factor_result, SHARED)
	}catch(ex){}
}

login("admin","123456")
clear()
go

quotesData = loadText(yourDir + "sampleQuotes.csv")

x=quotesData.schema().colDefs
share(streamTable(1000000:0, x.name, x.typeString), "quotes")

d = dict(STRING, ANY)
for(id in quotesData[`symbol].distinct())
     d[id]= array(DOUBLE,0,0)
dictUpdate!(d,append!, quotesData.symbol, (quotesData.bidPrice1+quotesData.askPrice1)/2)
share streamTable(100:0,`starttime`endtime`symbol`factorName`orderbook_factor_15, [LONG,LONG,SYMBOL,SYMBOL,DOUBLE]) as factor_result
subscribeTable(tableName="quotes", actionName="act_factor", offset=-1, handler=factor1_handler{d, factor_result}, msgAsTable=true);

replay(inputTables=quotesData, outputTables=quotes, dateColumn=`date, timeColumn=`time)

Message handlers are the core of stream processing, with two primary debugging approaches available:

(1) Debug the message handling function separately;

(2) Print logs within the function.

The main input to a message handler is the message itself, which can be either a table or a tuple. Debugging can be initiated by constructing appropriate test data structures matching these formats. For complex functions requiring step-by-step analysis, a recommended practice involves creating variables mirroring the function parameters and executing the function body sequentially. Using the factorHandler function as an example, we construct three test variables historyDict, factors, and msg:

k=4000
syms = string(1..k)
n = 1200000
min_history = table(take(syms, n) as symbol, rand(10.0, n) as ap)
msg = table(string(1..k) as symbol, rand(10.0, k) as ap)
historyDict = dict(STRING, ANY)
historyDict.dictUpdate!(append!, min_history.symbol, min_history.ap, x->array(x.type(), 0, 512).append!(x))
factors = streamTable(100000:0, `timestamp`symbol`factor, [TIMESTAMP,SYMBOL,DOUBLE])

An alternative debugging method involves inserting diagnostic logging using the writeLog function, with the output visible in DolphinDB's system logs.

While this tutorial utilizes limited sample datasets, production deployments should incorporate comprehensive backtesting using substantial historical data volumes. This validation process helps identify potential issues in program logic, business logic, and system performance. DolphinDB provides the functions replay and replayDS for replaying tick data from historical databases, which can be referred to the user manual for more information.