Best Practices for Level 2 Quantitative Factor Development and Simulated Trading

Minute-level price movements provide a foundation for quantitative strategies to identify trends, assess strength, and detect turning points. For example, the closing price determines the model's prediction target, while the opening, high, and low prices affect decisions in drawdown and breakout strategies. You must also handle anomalies, such as an opening price of 0, and missing data appropriately. To ensure that strategies use consistent and reliable data, calculate minute-level price indicators as follows:

• Opening price (open):

  • For minutes other than the opening and closing minutes, use the latest price from the first snapshot in the window as the opening price.

  • For the opening and closing minutes:

    • If the most recent opening price in the window is 0, use the last traded price from the first snapshot in the window.

    • Otherwise, use the most recent opening price in the window.

• High price (high) and low price (low):

  • If the high price (or low price) in the last record of the window is 0, use the maximum (or minimum) last traded price in the window.

  • Otherwise, use the most recent high price (or low price) in the window.

• Closing price (close):

  • Use the latest price from the last record of the current window as the closing price.

Trading indicators reflect market activity and are key indicators for distinguishing a valid breakout from a false rally. For example, rising trading volume often accompanies a continuing trend, while declining volume may signal a pullback. For intraday timing strategies in particular, real-time changes in trading volume often determine whether to execute a signal. Because snapshot trading volume is cumulative, calculate the actual change for each minute using differences between successive values:

• Initially, use the values in the most recent record in the window for snapshot trading volume, trading value, and number of trades.

• Use the deltas() function to calculate the differences between adjacent minutes for each field. If a difference is null, retain the value from the most recent record in the window; otherwise, use the calculated difference.

• The fields include:

  • volume: the increase in trading volume.

  • amount: the increase in trading value.

  • numTrades: the increase in the number of trades.

One key benefit of snapshot data is that it provides order book information for the top 10 price levels, revealing the market's current supply and demand. Quantitative strategies focus on changes in order volume. For example, they may use order book volume-price imbalances to identify potential accumulation through sell pressure or distribution after driving prices up. Minute-level order statistics therefore provide an important window into the behavior of major market participants. To ensure that strategies capture the latest order book state at the end of each minute, process order book data using the following indicators:

• When calculating indicators related to buy and sell order volume and value, use the most recent snapshot record in the current calculation window for every indicator:

  • totalSellVolume: total sell order volume, taken from the latest value of the totalOfferQuant field in the window.

  • totalBuyVolume: total buy order volume, taken from the latest value of the totalBuyQuant field in the window.

  • entrustSellAmount: total value of sell orders, calculated by multiplying the sell order volume by the weighted average sell price. Use the latest record in the window.

  • entrustBuyAmount: total value of buy orders, calculated in the same way as entrustSellAmount by multiplying the buy order volume by the weighted average buy price.

  • sellAvgPrice and buyAvgPrice: the weighted average sell and buy prices.

Snapshot data includes data from opening auction sessions, continuous trading sessions, irregular timestamps, and multiple markets. Without cleaning, these can significantly compromise the accuracy of minute-level factors. In particular, nonstandard timestamps near minute-bar trigger points can cause aggregation errors and missing minute-level data. Therefore, you need to clean and format the timestamps in raw snapshots to ensure that each record falls into the correct aggregation window. We use the following preprocessing logic to filter snapshot data and standardize its timestamps:

• Filter data to trading sessions: Retain snapshots from the opening auction session (09:25–09:30) and the continuous trading sessions (09:30–11:30 and 13:00–14:57). Discard invalid data and data from outside trading hours.

• Optimize minute-bar trigger timing: The latest snapshot for each minute arrives at second 59. To account for this, the calculation is triggered as soon as the snapshot for second 59 arrives.

The following code was developed for DolphinDB 3.00.4. We recommend using version 3.00.4 or later.

// tse: time-series engine
// msg: raw snapshot data
def handleRawSnapshot(mutable tse, mutable msg){
    msg = select * from msg
        where ((second(tradeTime) between 09:25:00:11:30:59) || (second(tradeTime) between 12:59:00:14:56:59))
    if (size(msg) == 0) return
    update msg set tradeTime =
        case
            when minute(tradeTime) in 09:25m..09:29m then concatDateTime(date(tradeTime), 09:30:00.000)
            else tradeTime
        end

    msg1 = select * from msg where second(tradeTime) % 60 >= 59
    update msg set tradeTime = timestamp(concatDateTime(date(tradeTime), second(minute(tradeTime)))) + 58 * 1000 + 999 where second(tradeTime) % 60 >= 59
    update msg1 set tradeTime = timestamp(concatDateTime(date(tradeTime), second(minute(tradeTime)))) + 59 * 1000 + 1
    append!(msg, msg1)
    msg.sortBy!(`tradeTime)
    append!(tse, msg)
}

After cleaning, snapshot data must be efficiently aggregated in one-minute windows to produce standardized minute-level market data. In quantitative trading, snapshot counts vary widely across stocks, so traditional aggregation methods can easily omit, miscount, or double-count data. DolphinDB's time-series engine instead ensures that each one-minute window is processed as soon as all its data arrives. The processing steps are as follows:

• Set the window and step: Set both the window size and step to 60 seconds (1 minute) so that each minute bar is independent and non-overlapping.

• Force-trigger mechanism: For one-minute windows containing only a small number of snapshots, set the forceTriggerTime parameter to force output of the minute bar.

tse = createTimeSeriesEngine(
    name= tseName,
    windowSize=60000,
    step=60000,
    metrics=tseMetrics,
    dummyTable=rawStream,
    outputTable=rse,
    timeColumn=`tradeTime,
    useSystemTime=false,
    keyColumn=`securityID,
    useWindowStartTime=false,
    parallelism=parallelTSE,
    lowLatency = ctx["lowLatency"],
    forceTriggerTime=100,
    outputElapsedMicroseconds = true
)

Although minute-level data provides a standardized market summary, it cannot be used directly to make strategy decisions. It must first be transformed into features that strategies can use, such as behavior of major market participants, order book strength, and trading deviations. Generating these factors requires stateful variables, sliding windows, and historical state, and involves complex calculations that must be completed with millisecond-level latency. To support this, DolphinDB provides a reactive stateful engine for processing minute-level factors in real time. The minute-level aggregation results from the previous step are fed into the reactive stateful engine, which generates the minute-level factors required by the strategy in real time:

rse = createReactiveStateEngine(
    name = rseName,
    metrics = rseMetrics,
    dummyTable = tseOutputDummy,
    outputTable = objByName(minStreamTbname, true),
    keyColumn = `securityId,
    parallelism = parallelRSE,
    lowLatency = ctx["lowLatency"],
    outputElapsedMicroseconds = true
)

Quantitative strategies typically use capital flows to assess market strength and the trading direction of major market participants. Tick-by-tick data provides the most granular view of capital flows. It can be precisely decomposed into active buy and sell trading values, making it possible to determine whether major market participants are buying to support prices or selling to suppress them. In addition, analyzing how the last trade deviates from large- and small- order trades makes it possible to construct a microstructure-level behavioral bias factor. To address this requirement, the following factors measure the direction and scale of capital flows from trades executed during the current minute. The calculation logic is as follows:

• sellAmount: trading value of active sell trades.

• buyAmount: trading value of active buy trades.

• TradeBias: a factor that measures whether the last trade is biased toward a large buy or sell order. It is calculated as: (lastAmount-sellMinAmount)/abs(sellMinAmount)-(lastAmount-buyMinAmount)/abs(buyMinAmount)

  • lastAmount: trading value of the last trade in the current minute.

  • buyMinAmount: minimum single-trade value among all active buy trades.

  • sellMinAmount: minimum single-trade value among all active sell trades.

In market trading, major market participants typically execute trades through large orders, while small orders are more commonly associated with retail investors. Classifying active trades as large or small orders helps identify whether major market participants are quietly accumulating or distributing positions. In particular, when certain stocks are active, the continued appearance of large orders often indicates institutional activity and warrants close attention from the strategy. Accordingly, the following large- and small-order capital-flow factors are defined to quantify the structure of active capital flows at the minute frequency:

A single trade volume greater than 50,000 is used as the classification threshold:

• buySmallAmount: total capital involved in active buy trades with a trading volume ≤50,000.

• buyBigAmount: total capital involved in active buy trades with a trading volume >50,000.

• sellSmallAmount: total capital involved in active sell trades with a trading volume ≤50,000.

• sellBigAmount: total capital involved in active sell trades with a trading volume >50,000.

Tick-by-tick trade data may contain out-of-order records, inconsistent time formats, and data from multiple markets. Aggregating the data directly can result in missing records or incorrect window assignments, affecting the accuracy and completeness of minute-frequency factors. Therefore, to assign tick-by-tick trade data to the correct minute-frequency windows and ensure that all trades are ordered correctly, the raw trade data must undergo cleansing and time standardization. The specific steps are as follows:

1. Sort and align: First, sort the tick-by-tick trade data by tradeTime in ascending order to ensure a consistent chronological sequence. This prevents the time-series engine from discarding out-of-order records, which could compromise the correctness of minute-frequency factors.

2. Write data: Convert the data to second-level frequency and write the data to the time-series engine for subsequent minute-frequency aggregation.

// tse: time-series engine
// msg: raw snapshot data
def handleRawTrade(mutable tse, mutable msg){
    msg.sortBy!(`tradeTime)
    // Convert the data to second-level frequency to prevent out-of-order records
    replaceColumn!(msg, `tradeTime, timestamp(datetime(msg.tradeTime)))
    tse.append!(msg)
}

The standardized trade data is ingested into the time-series engine, which divides it into minute-frequency windows and performs initial aggregation. Tick-by-tick trades are dense but unevenly distributed. Appropriate window settings and a forced-trigger mechanism are therefore required to ensure that every calendar minute produces an output. The time-series engine supports parallel sliding-window aggregation for multiple stocks through the parallelism parameter. It also provides trading-session controls and performance output to ensure real-time processing in live trading environments. The processing logic is as follows:

• Window configuration: Set windowSize=60000 and step=60000. Each aggregation window is therefore 1 minute long, with a sliding step of 1 minute, producing non-overlapping calendar-minute windows.

• Forced-trigger control: Set forceTriggerTime=100 to trigger minute-frequency calculations promptly for inactive stocks.

tse = createTimeSeriesEngine(
        name= tseName,
        windowSize=60000,
        step=60000,
        metrics=tseMetrics,
        dummyTable=rawStream,
        outputTable=rse,
        timeColumn=`tradeTime,
        useSystemTime=false,
        keyColumn=`securityID,
        useWindowStartTime=false,
        parallelism=parallelTSE,
        lowLatency = ctx["lowLatency"],
        forceTriggerTime=100,
        outputElapsedMicroseconds = true
)

Basic aggregated data provides only coarse-grained information such as total trading value and trade count, which cannot meet strategy requirements for features such as capital-flow structure, behavioral bias, and large-order trading. DolphinDB uses a reactive stateful engine to process minute-frequency trade data streams at a fine-grained level, construct interpretable factor features for strategies, and maintain real-time output. The minute-frequency aggregation results are fed into the stateful engine to calculate and output minute-frequency factors in real time:

rse = createReactiveStateEngine(
    name = rseName,
    metrics = rseMetrics,
    dummyTable = tseOutputDummy,
    outputTable = objByName(minStreamTbname, true),
    keyColumn = `securityId,
    parallelism = parallelRSE,
    lowLatency = ctx["lowLatency"],
    outputElapsedMicroseconds = true
)

Calculating derived factors requires both snapshot factors and trade factors, so the system must support cross-table merging, concurrent processing, and state management. The resulting derived factors must also be available to downstream inference modules in real time, which requires a consistent data structure and a clear processing path. DolphinDB therefore provides the following core components to build the stream processing pipeline:

• Equi join engine: Matches two sources, snapshot-minute data and trade-minute data, by tradeTime and securityId to create a merged minute-level input stream.

• Reactive stateful engine: Receives the merged streaming data output by the equi join engine and calculates derived factors from it. The engine maintains state and updates indicators by security symbol, while supporting concurrent processing.

• Derived-factor stream table: Stores the minute-level factor stream output by the stateful engine and supports persistence, subscription, and subsequent inference for generating trading signals.

To enhance the expressive power of minute-level factors, minute-level data from different sources must be aligned to a common time frame. The equi join engine provides millisecond-level time alignment, ensuring that the merged data can be used directly to calculate derived factors in the downstream stateful engine. The engine connects to the minute-level factor tables generated from snapshots and trades, then merges the minute-level data tables from the two sources.

• Data sources to be joined:

  • Left table: the minute-level factor table generated from snapshots.

  • Right table: the minute-level factor table generated from tick-by-tick trades.

• Join conditions:

  • The matching field is securityId.

  • The time field used for alignment is tradeTime.

ejEngine = createEquiJoinEngine(
    name = "EquiJoinEngine",
    leftTable = snapMinDummy,
    rightTable = tradeMinDummy,
    outputTable = rse,
    metrics = sqlCol(colNames[2:]),
    matchingColumn = `securityId,
    timeColumn = `tradeTime
)

The merged minute-level data contains richer information and can be used to construct composite factors with greater strategic value, such as "trading value deviation from the average bid-ask price," "major-participant behavior versus order placement response," and "large-order-driven trends." These factors typically involve complex calculation logic, so the reactive stateful engine is used for efficient processing.

rse = createReactiveStateEngine(
    name = rseName,
    metrics = facMetrics,
    dummyTable = dummy,
    outputTable = objByName(factorTbname, true),
    keyColumn = `securityId,
    parallelism = parallelRSE,
    lowLatency = ctx["lowLatency"],
    outputElapsedMicroseconds = true
)

The results are ultimately written to the derived-factor stream table for real-time model inference and trading signal generation.

The order management feature allows you to specify a start time (effectiveTime) and end time (expireTime) for each order. The order can be matched and executed only during this interval. When the end time is reached, any unfilled or partially filled order triggers a cancellation.

• Timely trade execution: Prevents orders from remaining open indefinitely and avoids exposing orders to risk after market conditions and the strategy's intent have changed.

• Realistic simulation of the trading environment: In live trading, many orders have a specified validity period. Order management enables simulated trading to reflect these real-world constraints more accurately.

• Improved simulation accuracy and risk control: Removes expired orders promptly, effectively controls trading risks, and prevents expired orders from causing erroneous judgments.

In a simulated environment, a partially filled order may have an unfilled remainder. The order management feature is responsible for the following:

• Automatically updates the status of the remaining order at the specified end time, preventing residual orders from interfering with subsequent strategy evolution.

• Outputs complete execution and order status information to ensure the completeness and accuracy of the simulated data.

• Orders in quantitative trading strategies often have extremely short lifecycles, making precise simulation of order validity periods essential.

• Institutional strategies with strict capital management and risk control requirements need to closely monitor order execution time limits and fill status.

• In market-making strategies, proactive order cancellation and validity-period control help ensure smooth order matching and real-time feedback.

• Strategy testing and backtesting require detailed tracking and analysis of order-matching timeliness and partial-fill behavior.

The attached files provide raw stock snapshot and trade data for a single day to demonstrate real-time factor generation and simulated trading.

//Import the module
clearCachedModules()
use SnapTradeStream::CreateTable
use SnapTradeStream::utils
use SnapTradeStream::OHLCFactors
use SnapTradeStream::StreamFrame

// Import the data
// Update the path
colNames, colTypes = getTradeTickColDefs()
tradeData = select * from loadText("/data/trade.csv", ,table(colNames, colTypes)) order by tradeTime

colNames, colTypes = getSnapshotColDefs()
snapshotData = select * from loadText("/data/snapshot.csv", ,table(colNames, colTypes)) order by tradeTime

After importing the day's tick-by-tick trade and snapshot data, create stream tables to receive data in real time and enable downstream stream processing.

tradeStreamName = "tradeStream"
unsubAndDropAll(tradeStreamName)
sche = schema(tradeData).colDefs
colNames, colTypes = sche.name, sche.typeString
try{
    share(streamTable(100:0, colNames, colTypes), tradeStreamName)
}catch(ex){print(ex)}

snapStreamName = "snapshotStream"
unsubAndDropAll(snapStreamName)
sche = schema(snapshotData).colDefs
colNames, colTypes = sche.name, sche.typeString
try{
    share(streamTable(100:0, colNames, colTypes), snapStreamName)
}catch(ex){print(ex)}

To support minute-frequency feature extraction across different data sources, the system uses a unified function mapping dictionary, factorFuncDict, and a parallelism configuration dictionary, parallelDict. This design enables modularity and flexible scheduling.

factorFuncDict = {
    "tradeTSE": getTrade1MTSFactor, // Function for retrieving time-series engine metrics from trade data
    "tradeRSE": getTrade1MReactiveFactor, // Function for retrieving reactive stateful engine metrics from trade data
    "snapTSE": getSnap1MTSFactor, // Function for retrieving time-series engine metrics from snapshot data
    "snapRSE": getSnap1MReactiveFactor, // Function for retrieving reactive stateful engine metrics from snapshot data
    "mergedRSE": minDerivedKlineFactors // Function for retrieving reactive stateful engine metrics from derived data
}

// Define the parallelism parameter dictionary
parallelDict = {
    "tradeTSE": 5,
    "tradeRSE": 5,
    "snapTSE": 5,
    "snapRSE": 5,
    "mergedRSE": 5
}

// Define the model path
extraParams = {
    "torchModelPath": "/data/pytorch_model.pt",
    "lgbmModelPath": "/data/lightgbm_model.txt",
    "lowLatency": false
}

• parallelDict specifies the number of parallel threads for each type of factor calculation task. Each task type is currently configured with 5 parallel tasks. You can adjust this setting based on the available node resources and data volume.

• factorFuncDict is a dictionary whose keys are strings and whose values are functions. It specifies the metrics and field definitions for each streaming engine. These functions are defined in the OHLCFactors module by default. For example, the getTrade1MTSFactor function is defined as follows:

// Function for retrieving time-series engine metrics and field definitions from trade data
def getTrade1MTSFactor(){
    d = dict(STRING, ANY, true)
    d["getAmounts(tradeBSFlag, tradePrice, tradeQty, `B)"] = ["buyAmounts", "DOUBLE"]
    d["getAmounts(tradeBSFlag, tradePrice, tradeQty, `S)"] = ["sellAmounts", "DOUBLE"]
    colNames2 = ["buySmallAmount", "buyBigAmount", "sellSmallAmount", "sellBigAmount"]
    colTypes2 = take("DOUBLE", size(colNames2))
    d["calCapitalFlow(buyApplSeqNum,offerApplSeqNum, tradeBSFlag, tradeQty, tradePrice)"] = [colNames2, colTypes2]
    d["lastTradeBias(tradeTime, tradeBSFlag, tradePrice, tradeQty)"] = ["tradeBias", "DOUBLE"]
    return d
}

The function returns a dictionary whose keys are function-call strings and whose values are two-element lists. The first element is the metric name, and the second is the field type. This design lets you adjust the configuration to meet your business needs. When factor calculation logic changes, you only need to update it here. You do not need to update underlying components such as stream tables and streaming engines, which significantly reduces factor maintenance and update costs while improving system extensibility and development efficiency.

After defining the required parameters, call the main function buildStreamFrame to subscribe to the data and calculate minute-frequency factors in real time.

factorDbname = "dfs://factor"
factorTbname = "minDerivedFactorsTb"
buildStreamFrame(factorDbname, factorTbname, tradeStreamName, snapStreamName, factorFuncDict, parallelDict, extraParams)

buildStreamFrame takes seven parameters:

Replay historical data into the stream tables to simulate the arrival of real-time streaming data and trigger the end-to-end streaming calculation workflow.

replay(
    [tradeData, snapshotData],
    [objByName(tradeStreamName, true), objByName(snapStreamName, true)],
    `tradeTime,
    `tradeTime,
    100,
    false
)

buildStreamFrame creates the pipeline for calculating minute-level factors. After the replay finishes, wait for data to appear in the derived factor table, then start model inference and simulated order matching:

maxWaitMs = 60000
waitedMs = 0
do {
    sleep(500)
    waitedMs += 500
} while (size(objByName(factorTbname, true)) == 0 && waitedMs < maxWaitMs)

use SnapTradeStream::StreamUtils
derived2Simulator(
    factorTbname,
    extraParams["lgbmModelPath"],
    extraParams["torchModelPath"],
    snapStreamName,
    false
)

Real-time simulated trade table

The following table lists key fields in the real-time simulated trade details:

Run the following code to clean up the streaming environment:

clearStreamEnv()
each(undef{, SHARED}, [tradeStreamName, snapStreamName])

Using a full day of real snapshot and trade data for 100, 200, 500, and 1,000 stocks, we measured the average calculation time, in microseconds, for the daily time-series engine (TSE) and the reactive stateful engine (RSE), with the lowLatency parameter disabled and enabled. The value in parentheses is the ratio of the calculation time with the lowLatency parameter enabled to the time with it disabled.

When the lowLatency parameter is enabled, all streaming engines have significantly lower computation times than when it is disabled.

The following code initializes the Orca minute-level factor framework. It imports sample trade and snapshot data, creates stream tables with the same schemas, and configures the factor functions, parallelism, model path, and result storage location.

use SnapTradeStream::CreateTable
use SnapTradeStream::utils
use SnapTradeStream::OHLCFactors
use SnapTradeStream::OrcaFrame

// Import the data
colNames, colTypes = getTradeTickColDefs()
tradeData = select * from loadText("/data/trade.csv", ,table(colNames, colTypes)) order by tradeTime

colNames, colTypes = getSnapshotColDefs()
snapshotData = select * from loadText("/data/snapshot.csv", ,table(colNames, colTypes)) order by tradeTime


// Create stream tables
tradeStreamName = "tradeStreamOrca"
unsubAndDropAll(tradeStreamName)
sche = schema(tradeData).colDefs
colNames, colTypes = sche.name, sche.typeString
try{
    share(streamTable(100:0, colNames, colTypes), tradeStreamName)
}catch(ex){print(ex)}

snapStreamName = "snapshotStreamOrca"
unsubAndDropAll(snapStreamName)
sche = schema(snapshotData).colDefs
colNames, colTypes = sche.name, sche.typeString
try{
    share(streamTable(100:0, colNames, colTypes), snapStreamName)
}catch(ex){print(ex)}

// Define the factor function dictionary
factorFuncDict = {
    "tradeTSE": getTrade1MTSFactor, // Function for retrieving time-series engine metrics from trade data
    "tradeRSE": getTrade1MReactiveFactor, // Function for retrieving reactive stateful engine metrics from trade data
    "snapTSE": getSnap1MTSFactor, // Function for retrieving time-series engine metrics from snapshot data
    "snapRSE": getSnap1MReactiveFactor, // Function for retrieving reactive stateful engine metrics from snapshot data
    "mergedRSE": minDerivedKlineFactors // Function for retrieving reactive stateful engine metrics from derived data
}

// Define the parallelism parameter dictionary
parallelDict = {
    "tradeTSE": 5,
    "tradeRSE": 5,
    "snapTSE": 5,
    "snapRSE": 5,
    "mergedRSE": 5
}

// Define the model path
extraParams = {
    "torchModelPath": "/data/pytorch_model.pt",
    "lgbmModelPath": "/data/lightgbm_model.txt"
}

// Define the persistent database and table path
factorDbname = "dfs://factor"
factorTbname = "minDerivedFactorsTbOrca"

The keys in factorFuncDict are defined as follows:

The keys in parallelDict are defined as follows:

The keys in extraParams are described below:

Call the buildOrcaFrame function in the module to set up the Orca stream processing framework. For parameter details, see Section 6.3.4. The function performs the following steps:

1. Clean up the existing environment, including stream graphs, stream tables, streaming engines, and related subscriptions, to prevent leftover data from interfering with processing.

2. Create a stream processing framework through the Orca API to generate minute-level derived factors from raw snapshot and trade data.

3. Subscribe to the minute-level derived factor data and use a model to perform real-time inference and generate trading signals. Then construct orders and insert them into the order matching simulator to produce real-time simulated trading results.

buildOrcaFrame(factorDbname, factorTbname, tradeStreamName, snapStreamName, factorFuncDict, parallelDict, extraParams)

Replay historical data to run the full Orca stream processing and simulated trading workflow and output simulated trading results in real time.

replay(
    [tradeData, snapshotData],
    [objByName(tradeStreamName, true), objByName(snapStreamName, true)],
    `tradeTime,
    `tradeTime,
    100,
    false
)

Run the following code to clean up the Orca environment.

clearOrcaEnv()
each(undef{, SHARED}, [tradeStreamName, snapStreamName])