Best Practices for Factor Computation

Tick-by-tick data records every individual trade published by the exchange, detailing both the buyer and seller. It is released every 3 seconds, covering all transactions within that period. Each matched trade involves a specific order from both the buyer and the seller.

In the sample dataset above, the fields BuyNo and SellNo indicate the order numbers of the buyer and seller, respectively. Other key fields include:

• SecurityID (Instrument code)
• TradeTime (Transaction timestamp)
• TradePrice (Transaction price)
• TradeQty (Transaction volume)
• TradeAmount (Transaction value)

The tick-by-tick data volume per trading day is approximately 8 GB. The database and table are created based on the partitioning scheme outlined in the table above. Click to view the corresponding script: appendix_2.1_createTickDbAndTable_main.dos.

The market snapshot is published every 3 seconds, including the following fields (at the end of each 3-second interval):

• Intraday cumulative volume (TotalVolumeTrade) and cumulative value (TotalValueTrade) up to that moment.
• Order book snapshot: covering bid and ask orders (Bid for buyers; sellers may be labeled as Offer or Ask), bid/ask prices (BidPrice, OfferPrice), total order quantity (OrderQty), and the number of orders (Orders).
• Latest transaction price (LastPx), opening price (OpenPx), intraday high (HighPx), and intraday low (LowPx).
• Other fields are consistent with tick-by-tick data; refer to the exchange's data dictionary for details.

DolphinDB 2.0 TSDB storage engine supports array vectors, enabling a single cell to store multiple values as a vector. This article provides a detailed comparison between array vector storage and non-array vector storage. In the example, best 10 bid/ask levels are stored as a vector within a single cell.

Database and table creation scripts for both storage formats are available for reference: appendix_2.2_createSnapshotDbAndTable_main.dos.

The market snapshot volume per trading day is approximately 10 GB. The market snapshot is stored as follows：

Each stock’s per-minute data includes four price fields: open, high, low, and close prices, along with the trading volume and value for that minute. Additionally, derived indicators such as VWAP can be calculated based on these basic fields. 1-minute OHLC data is aggregated from tick-by-tick trade.

Daily OHLC data follows the same storage format and fields as 1-minute OHLC data and can be aggregated from 1-minute OHLC or high-frequency data.

Database and table creation scripts for OHLC data are available in: appendix_2.3_createTableKMinute_main.dos.

Panel data uses time as rows, stock ticker as columns, and indicators as values. Performing calculations on panel data can significantly simplify the complexity of scripts (as the calculation expressions can be directly translated from the original mathematical formulas).

Panel data is typically generated using the panel function or exec combined with pivot by. Each row of the result represents a timestamp, and each column corresponds to a stock.

                         000001     000002  000003  000004  ...
                        --------- ---------- ------- ------ ---
2020.01.02T09:29:00.000|3066.336 3212.982  257.523 2400.042 ...
2020.01.02T09:30:00.000|3070.247  3217.087 258.696 2402.221 ...
2020.01.02T09:31:00.000|3070.381 3217.170  259.066 2402.029 ...

DolphinDB's row functions and various sliding window functions can be easily applied to panel data. Factor calculations on panel data simplify complex logic, allowing execution with just a single line of code. For example:

• During the calculation of the Alpha 1 factor, the rowRank function is applied for cross-sectional calculations, while the iif function can directly filter and compute on vectors. Additionally, sliding window functions like mimax and mstd perform computations on each column.

  //alpha 1 
  //Alpha#001 formula：rank(Ts_ArgMax(SignedPower((returns<0?stddev(returns,20):close), 2), 5))-0.5
  
  @state
  def alpha1TS(close){
  	return mimax(pow(iif(ratios(close) - 1 < 0, mstd(ratios(close) - 1, 20),close), 2.0), 5)
  }
  
  def alpha1Panel(close){
  	return rowRank(X=alpha1TS(close), percent=true) - 0.5
  }
  
  input = exec close from loadTable("dfs://k_minute","k_minute") where date(tradetime) between 2020.01.01 : 2020.01.31 pivot by tradetime, securityid
  res = alpha1Panel(input)

• When calculating the Alpha 98 factor, three panel datasets—vwap, open, and vol—are used. Within each matrix, the rowRank function is applied for cross-sectional calculations, while m-functions are used for columnar computations. Additionally, binary operations are performed between matrices. In DolphinDB, all these operations can be accomplished with a single line of code.

  //alpha 98
  //Alpha #98 formula:
  (rank(decay_linear(correlation(vwap, sum(adv5, 26.4719), 4.58418), 7.18088)) -
  rank(decay_linear(Ts_Rank(Ts_ArgMin(correlation(rank(open), rank(adv15), 20.8187), 8.62571), 6.95668), 8.07206)))
  
  def prepareDataForDDBPanel(raw_data, start_time, end_time){
  	t = select tradetime,securityid, vwap,vol,open from raw_data where date(tradetime) between start_time : end_time
  	return dict(`vwap`open`vol, panel(t.tradetime, t.securityid, [t.vwap, t.open, t.vol]))
  }
  
  @state
  def alpha98Panel(vwap, open, vol){
  	return rowRank(X = mavg(mcorr(vwap, msum(mavg(vol, 5), 26), 5), 1..7),percent=true) - rowRank(X=mavg(mrank(9 - mimin(mcorr(rowRank(X=open,percent=true), rowRank(X=mavg(vol, 15),percent=true), 21), 9), true, 7), 1..8),percent=true)
  }
  
  raw_data = loadTable("dfs://k_minute","k_day")
  start_time = 2020.01.01
  end_time = 2020.12.31
  input = prepareDataForDDBPanel(raw_data, start_time, end_time)
  timer alpha98DDBPanel = alpha98Panel(input.vwap, input.open, input.vol)

  Factor calculation based on panel data typically involves two stages: panel data preparation and factor computation. The panel data preparation stage can be time-consuming, which impacts overall computational efficiency. To address this, DolphinDB's TSDB engine offers wide format storage, where panel data is directly stored in the database (each column in the panel corresponds to a column in the table). This allows panel data to be accessed directly via SQL queries, eliminating the need for row-column transposition and significantly reducing preparation time. Wide format storage is more efficient than narrow format storage, especially when handling large datasets, and can greatly improve query and computation performance. In Chapter 5, we will provide a detailed comparison of the performance between wide and narrow format storage.

DolphinDB adopts a columnar structure for both storage and computation. If a factor's calculation only involves a single stock’s time-series data, without cross-sectional analysis across multiple stocks, it can be efficiently performed using SQL queries. This is achieved by grouping data by stock and applying the factor function over a specified time period. When data is partitioned by stock in the database, this method enables highly efficient in-database parallel computation.

def sum_diff(x, y){
    return (x-y)\(x+y)
}

@state
def factorDoubleEMA(price){
    ema_2 = ema(price, 2)
    ema_4 = ema(price, 4)
    sum_diff_1000 = 1000 * sum_diff(ema_2, ema_4)
    return ema(sum_diff_1000, 2) - ema(sum_diff_1000, 3)
}

res = select tradetime, securityid, `doubleEMA as factorname, factorDoubleEMA(close) as val from loadTable("dfs://k_minute","k_minute") where  tradetime between 2020.01.01 : 2020.01.31 context by securityid

In the example above, we define a factor function, factorDoubleEMA, which operates solely on a stock’s price series. Using SQL, we apply the context by clause to group data by stock code and then call the factorDoubleEMA function to compute the factor series for each stock. Note that context by is an extension of group by in DolphinDB SQL, making it a unique feature of DolphinDB. context by is designed for vectorized computation, where the input and output have the same length.

All technical analysis indicators in TA-Lib are time-series factor functions. These functions can be computed using either SQL queries or the panel data approach. However, factors like Alpha 1 and Alpha 98 (mentioned in Section 2.1) involve both time-series and cross-sectional computations, making them cross-sectional factors. These cannot be encapsulated into a single function and called within a single SQL statement.

For cross-sectional factors, it is recommended to pass the entire table as an input to a UDF factor function, perform SQL operations within the function, and return a table as the output.

//alpha1
def alpha1SQL(t){
	res = select tradetime, securityid, mimax(pow(iif(ratios(close) - 1 < 0, mstd(ratios(close) - 1, 20), close), 2.0), 5) as val from t context by securityid
	return select tradetime, securityid, rank(val, percent=true) - 0.5 as val from res context by tradetime
}
input = select tradetime,securityid, close from loadTable("dfs://k_day_level","k_day") where tradetime between 2010.01.01 : 2010.12.31
alpha1DDBSql = alpha1SQL(input)

//alpha98
def alpha98SQL(mutable t){
	update t set adv5 = mavg(vol, 5), adv15 = mavg(vol, 15) context by securityid
	update t set rank_open = rank(X = open,percent=true), rank_adv15 =rank(X=adv15,percent=true) context by date(tradetime)
	update t set decay7 = mavg(mcorr(vwap, msum(adv5, 26), 5), 1..7), decay8 = mavg(mrank(9 - mimin(mcorr(rank_open, rank_adv15, 21), 9), true, 7), 1..8) context by securityid
	return select tradetime,securityid, `alpha98 as factorname, rank(X =decay7,percent=true)-rank(X =decay8,percent=true) as val from t context by date(tradetime)
}
input = select tradetime,securityid, vwap,vol,open from  loadTable("dfs://k_day_level","k_day") where tradetime between 2010.01.01 : 2010.12.31
alpha98DDBSql = alpha98SQL(input)

Factors derived from data at different frequencies exhibit distinct characteristics. This section illustrates best practices for factor computation based on minute-level, daily, snapshot, and tick-by-tick data.

defg dayReturnSkew(close){
	return skew(ratios(close))	
}

minReturn = select `dayReturnSkew as factorname, dayReturnSkew(close) as val from loadTable("dfs://k_minute_level", "k_minute") where date(tradetime) between 2020.01.02 : 2020.01.31 group by date(tradetime) as tradetime, securityid

/* output
tradetime  securityid factorname    val               
---------- ---------- ------------- -------
2020.01.02 000019     dayReturnSkew 11.8328
2020.01.02 000048     dayReturnSkew 11.0544
2020.01.02 000050     dayReturnSkew 10.6186
*/

@state
def flow(buy_vol, sell_vol, askPrice1, bidPrice1){
        buy_vol_ma = round(mavg(buy_vol, 5*60), 5)
        sell_vol_ma = round(mavg(sell_vol, 5*60), 5)
        buy_prop = iif(abs(buy_vol_ma+sell_vol_ma) < 0, 0.5 , buy_vol_ma/ (buy_vol_ma+sell_vol_ma))
        spd = askPrice1 - bidPrice1
        spd = iif(spd < 0, 0, spd)
        spd_ma = round(mavg(spd, 5*60), 5)
        return iif(spd_ma == 0, 0, buy_prop / spd_ma)
}

res_flow = select TradeTime, SecurityID, `flow as factorname, flow(BidOrderQty[1],OfferOrderQty[1], OfferPrice[1], BidPrice[1]) as val from loadTable("dfs://LEVEL2_Snapshot_ArrayVector","Snap") where date(TradeTime) <= 2020.01.30 and date(TradeTime) >= 2020.01.01 context by SecurityID

/* output sample
TradeTime               SecurityID factorname val              
----------------------- ---------- ---------- -----------------
2020.01.22T14:46:27.000 110065     flow       3.7587
2020.01.22T14:46:30.000 110065     flow       3.7515
2020.01.22T14:46:33.000 110065     flow       3.7443
...
*/

def mathWghtCovar(x, y, w){
 	v = (x - rowWavg(x, w)) * (y - rowWavg(y, w))

	return rowWavg(v, w)
 }

@state
 def mathWghtSkew(x, w){
 	x_var = mathWghtCovar(x, x, w)
 	x_std = sqrt(x_var)
 	x_1 = x - rowWavg(x, w)
	x_2 = x_1*x_1
 	len = size(w)
 	adj = sqrt((len - 1) * len) \ (len - 2)
	skew = rowWsum(x_2, x_1) \ (x_var * x_std) * adj \ len
	return iif(x_std==0, 0, skew)
 }

//weights:
 w = 10 9 8 7 6 5 4 3 2 1

//Weighted skewness factorfactor
resWeight =  select TradeTime, SecurityID, `mathWghtSkew as factorname, mathWghtSkew(BidPrice, w)  as val from loadTable("dfs://LEVEL2_Snapshot_ArrayVector","Snap")  where date(TradeTime) = 2020.01.02 map
resWeight1 =  select TradeTime, SecurityID, `mathWghtSkew as factorname, mathWghtSkew(matrix(BidPrice0,BidPrice1,BidPrice2,BidPrice3,BidPrice4,BidPrice5,BidPrice6,BidPrice7,BidPrice8,BidPrice9), w)  as val from loadTable("dfs://snapshot_SH_L2_TSDB", "snapshot_SH_L2_TSDB")  where date(TradeTime) = 2020.01.02 map


/* output
 TradeTime               SecurityID factorname val               
----------------------- ---------- ---------- ------
...
2020.01.02T09:30:09.000 113537     array_1    -0.8828 
2020.01.02T09:30:12.000 113537     array_1    0.7371 
2020.01.02T09:30:15.000 113537     array_1    0.6041 
...
*/

//Calculate OHLC based on market snapshot, vwap
tick_aggr = select first(LastPx) as open, max(LastPx) as high, min(LastPx) as low, last(LastPx) as close, sum(totalvolumetrade) as vol,sum(lastpx*totalvolumetrade) as val,wavg(lastpx, totalvolumetrade) as vwap from loadTable("dfs://LEVEL2_Snapshot_ArrayVector","Snap") where date(TradeTime) <= 2020.01.30 and date(TradeTime) >= 2020.01.01 group by SecurityID, bar(TradeTime,1m) 

@state
def buyTradeRatio(buyNo, sellNo, tradeQty){
    return cumsum(iif(buyNo>sellNo, tradeQty, 0))\cumsum(tradeQty)
}

factor = select TradeTime, SecurityID, `buyTradeRatio as factorname, buyTradeRatio(BuyNo, SellNo, TradeQty) as val from loadTable("dfs://tick_SH_L2_TSDB","tick_SH_L2_TSDB") where date(TradeTime)<2020.01.31 and time(TradeTime)>=09:30:00.000 context by SecurityID, date(TradeTime) csort TradeTime

/* output
TradeTime           SecurityID factorname val              
------------------- ---------- ---------- ------
2020.01.08T09:30:07 511850     buyTradeRatio    0.0086
2020.01.08T09:30:31 511850     buyTradeRatio    0.0574
2020.01.08T09:30:36 511850     buyTradeRatio    0.0569
...
*/

In financial applications, both raw data and computed indicators typically exhibit temporal continuity. For a simple five-period moving average, mavg(close, 5), each update can be performed incrementally. When a new data point is received, subtract the value from the fifth prior period and add the new value to the existing sum. This avoids recalculating the sum of each update, improving computational efficiency. This approach, known as incremental computation, is a core concept in stream processing that significantly reduces real-time computation latency.

DolphinDB provides a wide range of built-in operators optimized for quantitative finance, many of which leverage incremental computation. Additionally, DolphinDB implements incremental optimization for user-defined functions. In the previous section, some UDF factor functions were marked with the @state modifier, indicating that they support incremental computation.

@state
def buyTradeRatio(buyNo, sellNo, tradeQty){
    return cumsum(iif(buyNo>sellNo, tradeQty, 0))\cumsum(tradeQty)
}

tickStream = table(1:0, `SecurityID`TradeTime`TradePrice`TradeQty`TradeAmount`BuyNo`SellNo, [SYMBOL,DATETIME,DOUBLE,INT,DOUBLE,LONG,LONG])
result = table(1:0, `SecurityID`TradeTime`Factor, [SYMBOL,DATETIME,DOUBLE])
factors = <[TradeTime, buyTradeRatio(BuyNo, SellNo, TradeQty)]>
demoEngine = createReactiveStateEngine(name="demo", metrics=factors, dummyTable=tickStream, outputTable=result, keyColumn="SecurityID")

insert into demoEngine values(`000155, 2020.01.01T09:30:00, 30.85, 100, 3085, 4951, 0)
insert into demoEngine values(`000155, 2020.01.01T09:30:01, 30.86, 100, 3086, 4951, 1)
insert into demoEngine values(`000155, 2020.01.01T09:30:02, 30.80, 200, 6160, 5501, 5600)

select * from result
/*
SecurityID TradeTime           Factor
---------- ------------------- ------
000155     2020.01.01T09:30:00 1
000155     2020.01.01T09:30:01 1
000155     2020.01.01T09:30:02 0.5
*/

@state
 def factorSmallOrderNetAmountRatio(tradeAmount, sellCumAmount, sellOrderFlag, prevSellCumAmount, prevSellOrderFlag, buyCumAmount, buyOrderFlag, prevBuyCumAmount, prevBuyOrderFlag){

	cumsumTradeAmount = cumsum(tradeAmount)
 	smallSellCumAmount, bigSellCumAmount = dynamicGroupCumsum(sellCumAmount, prevSellCumAmount, sellOrderFlag, prevSellOrderFlag, 2)

	smallBuyCumAmount, bigBuyCumAmount = dynamicGroupCumsum(buyCumAmount, prevBuyCumAmount, buyOrderFlag, prevBuyOrderFlag, 2) 

	f = (smallBuyCumAmount - smallSellCumAmount) \ cumsumTradeAmount
	return smallBuyCumAmount, smallSellCumAmount, cumsumTradeAmount, f
}

def createStreamEngine(result){
	tradeSchema = createTradeSchema()
	result1Schema = createResult1Schema()
	result2Schema = createResult2Schema()
	engineNames = ["rse1", "rse2", "res3"]
	cleanStreamEngines(engineNames)
	
	metrics3 = <[TradeTime, factorSmallOrderNetAmountRatio(tradeAmount, sellCumAmount, sellOrderFlag, prevSellCumAmount, prevSellOrderFlag, buyCumAmount, buyOrderFlag, prevBuyCumAmount, prevBuyOrderFlag)]>
	rse3 = createReactiveStateEngine(name=engineNames[2], metrics=metrics3, dummyTable=result2Schema, outputTable=result, keyColumn="SecurityID")
	
	metrics2 = <[BuyNo, SecurityID, TradeTime, TradeAmount, BuyCumAmount, PrevBuyCumAmount, BuyOrderFlag, PrevBuyOrderFlag, factorOrderCumAmount(TradeAmount)]>
	rse2 = createReactiveStateEngine(name=engineNames[1], metrics=metrics2, dummyTable=result1Schema, outputTable=rse3, keyColumn="SellNo")
	
	metrics1 = <[SecurityID, SellNo, TradeTime, TradeAmount, factorOrderCumAmount(TradeAmount)]>
	return createReactiveStateEngine(name=engineNames[0], metrics=metrics1, dummyTable=tradeSchema, outputTable=rse2, keyColumn="BuyNo")
}

result = createResultTable()
rse = createStreamEngine(result)
insert into rse values(`000155, 1000, 1001, 2020.01.01T09:30:00, 20000)
insert into rse values(`000155, 1000, 1002, 2020.01.01T09:30:01, 40000)
insert into rse values(`000155, 1000, 1003, 2020.01.01T09:30:02, 60000)
insert into rse values(`000155, 1004, 1003, 2020.01.01T09:30:03, 30000)

select * from result
/*
SecurityID TradeTime smallBuyOrderAmount smallSellOrderAmount totalOrderAmount factor
---------- ------------------- ------------------- -------------------- ---------------- ------
000155     2020.01.01T09:30:00 20000               20000                20000            0
000155     2020.01.01T09:30:01 60000               60000                60000            0
000155     2020.01.01T09:30:02 0                   120000               120000           -1
000155     2020.01.01T09:30:03 30000               150000               150000           -0.8
*/

@state
def alpha1TS(close){
	return mimax(pow(iif(ratios(close) - 1 < 0, mstd(ratios(close) - 1, 20),close), 2.0), 5)
}

def alpha1Panel(close){
	return rowRank(X=alpha1TS(close), percent=true) - 0.5
}

inputSchema = table(1:0, ["SecurityID","TradeTime","close"], [SYMBOL,TIMESTAMP,DOUBLE])
result = table(10000:0, ["TradeTime","SecurityID", "factor"], [TIMESTAMP,SYMBOL,DOUBLE])
metrics = <[SecurityID, alpha1Panel(close)]>
streamEngine = streamEngineParser(name="alpha1Parser", metrics=metrics, dummyTable=inputSchema, outputTable=result, keyColumn="SecurityID", timeColumn=`tradetime, triggeringPattern='keyCount', triggeringInterval=4000)

The previous section introduced DolphinDB’s unified batch-stream processing, where the same factor function is used for both batch and stream calculation, ensuring code reuse.

DolphinDB also offers a more streamlined approach: during historical data modeling, data replay can be used to compute factors directly with the streaming engine, eliminating the need for a separate batch processing pipeline. This ensures consistency in computation logic while improving development efficiency.

While the previous section demonstrated the computation of factorDoubleEMA using SQL-based batch processing, this section illustrates how to calculate factorDoubleEMA via data replay in a streaming context. For complete implementation details, refer to appendix_4.2_streamComputationOfDoubleEmaFactor_main.dos.

//Create a streaming engine and pass the factor algorithm factorDoubleEMA

factors = <[TradeTime, factorDoubleEMA(close)]>
 demoEngine = createReactiveStateEngine(name=engineName, metrics=factors, dummyTable=inputDummyTable, outputTable=resultTable, keyColumn="SecurityID")

	
//demo_engine subscribes to the snapshotStreamTable table

subscribeTable(tableName=snapshotSharedTableName, actionName=actionName, handler=append!{demoEngine}, msgAsTable=true)
 
//Create a data source for replay; during market hours, switch to writing real-time market data directly to snapshotStreamTable
inputDS = replayDS(<select SecurityID, TradeTime, LastPx from tableHandle where date(TradeTime)<2020.07.01>, `TradeTime, `TradeTime)

DolphinDB provides APIs for a variety of popular programming languages, including C++, Java, JavaScript, C#, Python, and Go. Programs written in these languages can leverage the corresponding DolphinDB APIs to subscribe to streaming tables from a DolphinDB server. Below is a simple example demonstrating how to subscribe to a streaming table using the Python API. For the details, refer to appendix_4.3.1_python_callback_handler_subscribing_stream_main.py.

current_ddb_session.subscribe(host=DDB_datanode_host,tableName=stream_table_shared_name,actionName=action_name,offset=0,resub=False,filter=None,port=DDB_server_port,
handler=python_callback_handler,# Pass the callback function to receive messages.
)

In live trading, streaming data is typically ingested into message queues in real time for downstream consumption. DolphinDB supports pushing results to messaging middleware for integration with trading systems. The example demonstrates how to use DolphinDB’s open-source ZMQ plugin to stream real-time results into a ZMQ message queue, enabling consumption by subscribers such as trading systems or market display modules. Beyond ZMQ, DolphinDB provides support for other mainstream messaging systems. All DolphinDB plugins are open-source, and users can develop custom plugins based on the open-source framework as needed.

An example of pushing streaming data from DolphinDB to a ZMQ message queue:

1. Start the ZMQ Message Consumer

   This program acts as the ZeroMQ message queue server, listening for incoming data. For the full implementation, refer to appendix_4.3.2_zmq_consuming_ddb_stream_main.py.

   zmq_context = Context()
    zmq_bingding_socket = zmq_context.socket(SUB) # Set socket options
   
   zmq_bingding_socket.bind("tcp://*:55556")		
    async def loop_runner():    
        while True:
            msg=await zmq_bingding_socket.recv() # Blocking loop until stream data is received
   
           print(msg) # Implement downstream message processing logic here	
   asyncio.run(loop_runner()) 

2. Initialize Factor Streaming Computation and Publishing result

   Once the ZMQ consumer is running, DolphinDB needs to start stream processing and publish the results. The following code demonstrates the DolphinDB implementation. For details of DolphinDB, refer to appendix_4.3.3_streamComputationOfDoubleEmaFactorPublishingOnZMQ_main.dos.

   // Table scheme for message queue output
   resultSchema=table(1:0,["SecurityID","TradeTime","factor"], [SYMBOL,TIMESTAMP,DOUBLE])
   
   def zmqPusherTable(zmqSubscriberAddress,schemaTable){
   	SignalSender=def (x) {return x}
   	pushingSocket = zmq::socket("ZMQ_PUB", SignalSender)
   	zmq::connect(pushingSocket, zmqSubscriberAddress)
   	pusher = zmq::createPusher(pushingSocket, schemaTable)
   	return pusher
   }
   
   // demoEngine pushes data to ZMQ queue - modify this string for different ZMQ addresses
   zmqSubscriberAddress="tcp://192.168.1.195:55556"
   
   // Create a table to send ZMQ packets to the specified address, with scheme matching resultSchema
   pusherTable=zmqPusherTable(zmqSubscriberAddress,resultSchema)
   
   // Create a streaming engine outputting to pusher
   demoEngine = createReactiveStateEngine(name="reactiveDemo", metrics=<[TradeTime,doubleEma(LastPx)]>, dummyTable=snapshotSchema, outputTable=pusherTable, keyColumn="SecurityID",keepOrder=true)

We evaluate three storage models by storing one year of minute-level data for five factors, comparing their efficiency in terms of data volume, storage usage, and write performance.

Analysis of Results

• Write Performance: TSDB wide format writes 4× faster than OLAP narrow format and 5× faster than TSDB narrow format.
• Storage Efficiency: OLAP and TSDB narrow format have similar storage sizes, while TSDB wide format is slightly smaller.
• Compression Ratio: TSDB narrow format achieves the highest compression, followed by OLAP narrow format, with TSDB wide format performing the worst.

Key Influencing Factors

• Data Volume & Efficiency: narrow format stores 3× more data per row, making TSDB wide format faster and more space-efficient, though its compression suffers.
• Data Randomness: High randomness in the dataset reduces compression efficiency in TSDB wide format.
• Storage Engine Optimization: TSDB narrow format applies sorting and indexing, slightly impacting its storage, write speed, and compression compared to OLAP.

For detailed implementation, refer to appendix_5.1_factorDataSimulation.zip.

To evaluate query performance under large-scale data conditions, we simulate a dataset of 4,000 stocks, 200 factors, and one year of minute-level data. The table below summarizes the dataset characteristics and partitioning strategies:

We evaluate query performance across OLAP narrow format, TSDB narrow format, and TSDB wide format by testing multiple query scenarios. For detailed script, refer to appendix_5.2_factorQueryTest.dos.

• TSDB significantly outperforms OLAP for point querieswithin a given time range.
• Due to fewer rows, TSDB wide format achieves the fastest response time.

Query 2: Single Factor, Single Stock, One Year (Minute-Level Data)

• TSDB wide format is the fastest, as it requires fewer file reads due to larger partitions and pre-sorted data.
• OLAP narrow format is the slowest because it scans more partitions.

Query 3: Single Factor, All Stocks, One Year (Minute-Level Data)

Query 4: Three Factors, All Stocks, One Year (Minute-Level Data)

• Query time scales linearly with data volume.
• TSDB wide format remains the most efficient due to its compact storage.

Query 5: Single Stock, All Factors, One Year (Minute-Level Data)

For wide format storage, queries should only select the necessary stock code column. The SQL query are as follows:

// Query a narrow table, retrieve all fields via wildcard (*)
tsdb_symbol_all=select  * from tsdb_min_factor where symbol=`sz000056 

// Query a wide table, selectively retrieve specific fields
select mtime,factorname,sz000001 from tsdb_wide_min_factor

The results show that both TSDB wide format and TSDB narrow format achieve fast query performance, whereas OLAP narrow format takes over 100 times longer. This discrepancy arises because OLAP narrow format partitions data by time and factor, requiring a full table scan to retrieve all factors for a single stock. In contrast, TSDB wide format only needs to read three columns, significantly improving query efficiency.

Furthermore, TSDB narrow format also demonstrates relatively good query performance compared to OLAP narrow format. This is due to TSDB’s built-in indexing on stock codes, enabling fast data retrieval.

In summary, factor storage strategies should be planned based on expected query patterns. Given the performance results, we recommend using TSDB wide format for factor storage to optimize query efficiency.

Different storage models allow real-time panel data generation using the following DolphinDB scripts.

Single-Factor, All Stocks, One Year (Minute-Level Panel Data)

// Narrow format storage
olap_factor_year_pivot_1=select val from olap_min_factor where factorcode=`f0002 pivot by tradetime,symbol 
// Wide format storage
wide_tsdb_factor_year=select * from tsdb_wide_min_factor where factorname =`f0001

TSDB wide format achieves over 10× faster query performance than both OLAP and TSDB narrow format. This advantage stems from its inherent data structure, which is already stored in a panel-like format, eliminating the need for transformation. In contrast, both OLAP and TSDB narrow format require a pivot by operation to reshape the data, introducing additional deduplication and sorting overhead. As dataset size increases, this overhead becomes the primary performance bottleneck, further widening the gap between wide and narrow formats.

Three-Factor, All Stocks, One Year (Minute-Level Panel Data)

// Narrow format storage
olap_factor_year_pivot=select val from olap_min_factor where factorcode in ('f0001','f0002','f0003') pivot by tradetime,symbol ,factorcode
// Wide format storage
wide_tsdb_factor_year=select * from tsdb_wide_min_factor where factorname in ('f0001','f0002','f0003')

As dataset size increases, TSDB wide format continues to demonstrate superior performance. narrow format storage suffer from the increased computational overhead of additional pivot by operations, leading to excessive deduplication and sorting costs. Eventually, both OLAP and TSDB narrow formats fail to complete the query within a reasonable time.

Key Advantages of TSDB Wide Format Storage

• Storage Efficiency: Despite having a lower compression ratio, TSDB wide format consumes the least disk space due to its significantly lower data volume per row (one-third of the narrow format).
• Write Performance: TSDB wide format writes data 4× faster than OLAP narrow format and 5× faster than TSDB narrow format.
• Direct Data Retrieval: Across different query scenarios, TSDB wide format achieves at least 1.5× speed improvement over both OLAP and TSDB narrow formats, with some cases reaching over 100× speed gains.
• Panel Data Queries: TSDB wide format retrieves panel data at least 10× faster than OLAP and TSDB narrow formats.
• Non-Partitioned Queries: For queries that do not align with the partition key (e.g., retrieving data by stock in a factor-partitioned table), TSDB wide format achieves 300× and 500× speed gains over OLAP and TSDB narrow formats, respectively.

In summary, if the number of stocks and factors remains stable over time, TSDB wide format is the optimal choice for factor storage. Users can structure the wide table with either stocks or factors as columns based on their specific query needs.

Once a factor exhibits stable directional signals, an investment strategy can be constructed and validated through historical data backtesting.

This section focuses on vectorized factor backtesting, following these steps:

1. Compute factor scores based on cumulative multi-day stock returns using daily OHLC data.
2. Rank factors and allocate position weights accordingly.
3. Multiply position weights with stock daily returns, then aggregate results to construct the portfolio return curve.

Complete code reference: appendix_6.1_vectorisedFactorBacktest_main.dos.

DolphinDB supports both narrow format storage and wide format storage. For efficient correlation calculations, array vector is recommended to store multiple factors of the same stock at the same timestamp.

Factor Autocorrelation Matrix Calculation (Narrow Format Storage):

1. Convert daily factor data into array vector based on time and stock.
2. Compute the correlation matrix within an in-memory table.
3. Aggregate correlations across stocks and compute the mean.

day_data = select toArray(val) as factor_value from loadTable("dfs://MIN_FACTOR_OLAP_VERTICAL","min_factor") where date(tradetime) = 2020.01.03 group by tradetime, securityid
result = select toArray(matrix(factor_value).corrMatrix()) as corr from day_data group by securityid
corrMatrix = result.corr.matrix().avg().reshape(size(distinct(day_data.factorname)):size(distinct(day_data.factorname)))

In quantitative finance, multi-factor models are typically constructed using one of the following approaches:

• Weighted Aggregation: Compute a weighted average of expected returns for each stock based on different factor weights, and select the stocks with the highest expected returns.
• Regression Models: These models utilize historical data to quantify the relationship between factors and returns, optimizing portfolio allocation accordingly.

DolphinDB offers a range of built-in regression models, including Ordinary Least Squares (OLS) Regression, Ridge Regression, and Generalized Linear Models (GLM). Notably, olsEx(OLS regression), the'cholesky'method in ridge(Ridge regression), andglm(GLM) all support distributed and parallel computation.

Additionally, DolphinDB supports Lasso Regression, Elastic Net Regression, Random Forest Regression, and AdaBoost Regression. Among them, adaBoostRegressorandrandomForestRegressorsupport distributed and parallel computation.

Factor development requires collaboration between the Quant and IT teams. The Quant team is responsible for developing and maintaining factor logic, while the IT team manages the factor computation framework. To ensure modularity and flexibility, factor logic should be decoupled from the computational framework, allowing factor developers to submit updates independently while the framework automatically updates and recalculates factors. This section focuses on managing factor logic, while computation framework and operations are covered in Sections 7.3 and 7.6.

It is recommended to encapsulate each factor as a UDF function. DolphinDB provides two methods for managing UDF functions:

• Function View:
  • Advantages: Centralized management—once deployed, all nodes in the cluster can access it; supports permission control; supports modular organization.
  • Disadvantages: Not Easily Modifiable—cannot be edited directly; must be dropped and recreated.
• Module:
  • Advantages: Supports modular management, organizing functions into hierarchical structures; can be preloaded during system initialization or dynamically loaded using the use statement.
  • Disadvantages: Must be manually deployed to all relevant nodes; does not support function-level permission control.

Future DolphinDB releases aim to integrate the strengths of both Function View and Module for a unified management approach.

When refactoring factor code, adjusting the computational framework, or upgrading the database, rigorous correctness testing is essential to ensure reliable factor calculations. DolphinDB provides a built-in unit testing framework for automated testing, ensuring factor logic integrity.

Key components of DolphinDB's unit testing framework include:

• test function: Executes individual test files or all test files within a directory.
• @testing macro: Defines test cases.
• assert statement: Verifies whether computation results meet expectations.
• eqObj function: Compares actual results with expected values.

The following example demonstrates unit testing for the factorDoubleEMA function, covering two batch-processing test cases and one streaming test case. Refer to the provided scripts for the complete implementation: appendix_7.2_doubleEMATest.dos.

@testing: case = "factorDoubleEMA_without_null"
re = factorDoubleEMA(0.1 0.1 0.2 0.2 0.15 0.3 0.2 0.5 0.1 0.2)
assert 1, eqObj(re, NULL NULL NULL NULL NULL 5.788743 -7.291889 7.031123 -24.039933 -16.766359, 6)

@testing: case = "factorDoubleEMA_with_null"
re = factorDoubleEMA(NULL 0.1 0.2 0.2 0.15 NULL 0.2 0.5 0.1 0.2)
assert 1, eqObj(re, NULL NULL NULL NULL NULL NULL 63.641310 60.256608  8.156385 -0.134531, 6)

@testing: case = "factorDoubleEMA_streaming"
try{dropStreamEngine("factorDoubleEMA")}catch(ex){}
input = table(take(1, 10) as id, 0.1 0.1 0.2 0.2 0.15 0.3 0.2 0.5 0.1 0.2 as price)
out = table(10:0, `id`price, [INT,DOUBLE])
rse = createReactiveStateEngine(name="factorDoubleEMA", metrics=<factorDoubleEMA(price)>, dummyTable=input, outputTable=out, keyColumn='id')
rse.append!(input)
assert 1, eqObj(out.price, NULL NULL NULL NULL NULL 5.788743 -7.291889 7.031123 -24.039933 -16.766359, 6)

The previous sections focused on the core logic of factor computation, without addressing parallel or distributed computing techniques for performance optimization. In practical engineering applications, parallelism can be achieved along the following levels: security (by stock), factor and time.

DolphinDB provides four primary methods for parallel (or distributed) computation:

1. Implicit SQL parallelism: When executing SQL queries on a distributed table, the engine automatically pushes down computations to partitions, enabling parallel execution.
2. Map-Reduce computation: Utilizing multiple data sources and executing parallel computation via the mr (map-reduce) function.
3. Job submission: Executing multiple tasks in parallel using submitJob or submitJobEx.
4. peach/ploop parallelism: Running parallel computations locally using peach or ploop (not recommended for factor computation).

Why peach/ploop is not recommended in factor computation

DolphinDB's computation framework distinguishes between two types of threads:

• Workers: Accept jobs, decompose them into multiple tasks, and distribute them to local executors or remote workers.
• Executors: Execute local, short-duration tasks, typically without further decomposition.

Using peach or ploop forces execution on local executors. If a subtask requires further decomposition (e.g., a distributed SQL query), it may significantly degrade system throughput.

This section will focus on the first three approaches to parallelizing factor computation.

resWeight =  select TradeTime, SecurityID, `mathWghtSkew as factorname, mathWghtSkew(BidPrice, w)  as val from loadTable("dfs://LEVEL2_Snapshot_ArrayVector","Snap")  where date(TradeTime) = 2020.01.02

minReturn = select `dayReturnSkew as factorname, dayReturnSkew(close) as val from loadTable("dfs://k_minute_level", "k_minute") where date(tradetime) between 2020.01.02 : 2020.01.31 group by date(tradetime) as tradetime, securityid map

//Repartition the data by stock into 10 HASH partitions
ds = repartitionDS(<select * from loadTable("dfs://k_minute_level", "k_minute") where date(tradetime) between 2020.01.02 : 2020.03.31>, `securityid, HASH,10)

def factorDoubleEMAMap(table){
	return select tradetime, securityid, `doubleEMA as factorname, factorDoubleEMA(close) as val from table context by securityid map
}

res = mr(ds,factorDoubleEMAMap,,unionAll)

def writeDayReturnSkew(dBegin,dEnd){
	dReturn = select `dayReturnSkew as factorname, dayReturnSkew(close) as val from loadTable("dfs://k_minute_level", "k_minute") where date(tradetime) between dBegin : dEnd group by date(tradetime) as tradetime, securityid
	//Load into the factor database
	loadTable("dfs://K_FACTOR_VERTICAL","factor_k").append!(dReturn)
	}

for (i in 0..11){
	dBegin = monthBegin(temporalAdd(2020.01.01,i,"M"))
	dEnd = monthEnd(temporalAdd(2020.01.01,i,"M"))
	submitJob("writeDayReturnSkew","writeDayReturnSkew_"+dBegin+"_"+dEnd, writeDayReturnSkew,dBegin,dEnd)
}

Memory management has always been a major concern for both operations and research personnel. This section briefly introduces how to efficiently manage memory in DolphinDB from both batch and stream processing perspectives. For more detailed information on memory management, please refer to Memory Management.

When setting up DolphinDB, the maximum available memory per node can be set through the maxMemSize parameter in the dolphindb.cfg file for a standalone or in the cluster.cfg file for a cluster, to control memory usage for computation and transactions.

Memory Management for Batch Processing

For example, in SQL queries, when processing half a year’s worth of market snapshot, intermediate variables in the batch process can consume up to 21 GB of memory. If the available memory is less than 21 GB, an "Out of Memory" error will occur. In this case, the job can be split into smaller tasks and submitted separately to avoid memory overflow.

After completing large-scale computations, the undef function can be used to assign variables to NULL, or the session can be closed to immediately release memory.

Memory Management for Stream Processing

In the case of data replay, for instance, the function enableTableShareAndPersistence is used to persist the stream table, with a cache size set to 800,000 rows. When the stream table exceeds this limit, older data is persisted to disk, freeing up memory space for new data to be written. This allows the stream table to continuously process data well beyond the 800,000 rows threshold.

Strict access control is essential for factor data management. DolphinDB offers a powerful, flexible, and secure permission system, supporting table-level and function/view-level access control. For more details on access control, please refer to User Access Control.

In practice, three common management models are used:

• Development team as administrators with full control over the database

  In this case, the development team can be granted the DB_OWNER privilege, enabling them to create and manage databases and tables, as well as manage permissions for their own data and tables.

  login("admin", "123456");
  createUser("user1", "passwd1")
  grant("user1", DB_OWNER)

• Operations team manages the database, while the development team has only read/write permissions

  In this case, the database administrator can grant data table access to the factor development team, or create a user group with the appropriate permissions and add relevant users to this group for centralized management.

  //Grant TABLE_READ to user1
  createUser("user1", "passwd1")
  grant("user1", TABLE_READ, "dfs://db1/pt1")
  
  //Grant TABLE_READ to group1name
  createGroup("group1name", "user1")
  grant("group1name", TABLE_READ, "dfs://db1/pt1")

• Development team has read-only access to specific data, not the entire database or table

  While DolphinDB does not natively support data-level access control within tables, its flexible permission system allows for other methods of implementing such control. For instance, by granting the function view privilege (e.g., VIEW_EXEC), data access control can be applied to specific factors within a table.

  A complete code example for factor table access control can be found in appendix_7.5.3_factorTableControll.dos. In this example, although user “u1” does not have read access to the table, they are still able to access the factor1 data within the table.

  // Grant u1 read-only access specifically to factor1
  createUser("u1", "111111")
  // Define the factor access function
  def getFactor1Table(){
      t=select * from loadTable("dfs://db1","factor") where factor_name="factor1";
      return t;
  }
  //Save the function
  addFunctionView(getFactor1Table)
  // Grant u1 execution privileges on the getFactor1Table function
  grant("u1", VIEW_EXEC, "getFactor1Table");
  
  // Note: The user must log in again to load the newly granted permissions
  factor1_tab=getFactor1Table()

Factor computation jobs fall into three main categories: full computation, interactive recalculation, and incremental computation.

Jobs can be executed in the following ways:

• Interactive Execution – Triggered directly via GUI or API.
• Job Submission (submitJob) – Sends tasks to the server job queue for execution, independent of the client.
• Scheduled Execution (scheduleJob) – Automates periodic recalculations, such as end-of-day updates.

Daily frequency data is typically aggregated from tick data or other high-frequency data, resulting in a relatively small data volume. Daily factors are often used to compute momentum factors or more complex factors that require long-term observation. It is recommended to partition data by year. Factor calculations at the daily frequency often span both time and stock dimensions, making panel-mode computation a natural choice for efficient processing. However, the optimal computation mode may vary depending on the storage model—different models support different execution strategies to best accommodate performance and flexibility needs.

In appendix_8.1_case1_daily .dos, data for 10 years and 4000 stocks is simulated, with a total data volume of approximately 1 GB before compression. The code demonstrates best practices for daily frequency factors, including Alpha 1, Alpha 98, as well as different calculation methods (panel or SQL mode) for writing to narrow format and wide format storage.

Minute frequency data is typically derived from tick or market snapshot and has a significantly larger volume than daily frequency factors. For partition design, it is recommended to use VALUE partitioning by month in combination with HASH partitioning on stock ID. Minute frequency factors are primarily used for intraday return calculations, where SQL-based computation is preferred, utilizing specific columns as inputs. Many factors require incremental computation after the market closes, necessitating systematic engineering management. To streamline this process, it is advisable to maintain these factors in dimension tables and execute batch computations through scheduled tasks.

In appendix_8.2_case2_minute.dos, data for one year and 4000 stocks is simulated, with a total data volume of approximately 20 GB before compression. The code demonstrates best practices for minute frequency factors, including intraday return skew factors, factorDoubleEMA, and the entire process of storing data in both narrow and wide formats, as well as best practices for engineering the daily incremental computation of factors.

Market snapshot typically refers to data captured at 3-second intervals. In practice, real-time factors are derived from this data, based on quotes, trading volume, and other relevant fields. It is recommended to use field names as parameters in UDF functions. Due to the market depth nature of market snapshot, conventional storage methods can result in significant space usage. To address this, it is advisable to store this data in array vectors, which optimizes disk space and simplifies the code by avoiding the need to select multiple duplicate fields.

In appendix_8.3_case3_snapshot.dos, 20 days of market snapshot are simulated and stored both as standard market snapshot and in format of array vector. The code also illustrates best practices for both stateful factors (e.g., flow) and stateless factors (e.g., weight skew) in a unified stream-batch processing framework.

Tick data represents the most granular transaction records provided by exchanges, published at 3-second intervals, capturing all executed trades within each window. Tick-based factors are classified as high-frequency factors. During the prototyping and model development phase, batch processing can be applied to small data samples for validation. Once the model is finalized, the same computation logic can be deployed to a real-time stream processing framework (i.e., unified batch and stream processing), optimizing memory usage, enhancing real-time performance, and accelerating model iteration.

In appendix_8.4_case4_streamTick.dos, the calculation and stream processing of tick data factors are demonstrated.