Best Practices for Market Data Replay

The solution we build has the following architecture:

• Market Data Ingestion: Both real-time and historical tick data can be imported to the DolphinDB distributed database via the DolphinDB API or a plugin.
• Function Encapsulation: Functions used during the replay process can be encapsulated as function views. Key parameters such as stock list, replay date, replay rate, data source can still be specified by users.
• User Request: Users can replay market data by calling the function view through DolphinDB clients (such as DolphinDB GUI, VS Code Extension, APIs). Users can also subscribe to and consume real-time replay results on the client side. Additionally, multi-user concurrent replay is supported.

This section introduces the steps for building a market data replay solution, as shown in the following figure:

• Step 1 (Chapter 2): Design and implement a suitable distributed database structure in DolphinDB. Create the necessary tables and import historical market data to serve as the replay data source.
• Step 2 (Chapter 3): Develop a DolphinDB function view that encapsulates the entire replay process. This abstraction allows users to initiate replay requests by simply specifying key parameters, without needing to understand the details of DolphinDB's replay functionality.
• Step 3 (Chapter 4 & 5): Utilize the DolphinDB API to call the previously created function view from external applications. This enables the execution of replay operations outside the DolphinDB environment. Additionally, users can set up subscriptions to consume real-time replay results on the client side.

The stkReplay function is used to perform the replay operation.

def stkReplay(stkList, mutable startDate, mutable endDate, replayRate, replayUuid, replayName)
{
    maxCnt = 50
    returnBody = dict(STRING, STRING)
    startDate = datetimeParse(startDate, "yyyyMMdd")
    endDate = datetimeParse(endDate, "yyyyMMdd") + 1
    sortColumn = "ApplSeqNum"
    if(stkList.size() > maxCnt)
    {
        returnBody["errorCode"] = "0"
        returnBody["errorMsg"] = "Exceeds the limits for a single replay. The maximum is: " + string(maxCnt)
        return returnBody
    }
    if(size(replayName) != 0)
    { 
        for(name in replayName)
        {
            if(not name in ["snapshot", "order", "transaction"])
            {
                returnBody["errorCode"] = "0"
                returnBody["errorMsg"] = "Input the correct name of the data source. Cannot recognize: " + name
                return returnBody
            }
        }
    }
    else
    {
        returnBody["errorCode"] = "0"
        returnBody["errorMsg"] = "Missing data source. Input the correct name of the data source."
        return returnBody
    }
    try 
    {
        if(size(replayName) == 1 && replayName[0] == "snapshot")
        {
            colName = ["timestamp", "biz_type", "biz_data"]
            colType = [TIMESTAMP, SYMBOL, BLOB]
            sortColumn = "NULL"
        }
        else
        {
            colName = ["timestamp", "biz_type", "biz_data", sortColumn]
            colType = [TIMESTAMP, SYMBOL, BLOB, LONG]
        }
        msgTmp = streamTable(10000000:0, colName, colType)
        tabName = "replay_" + replayUuid
        enableTableShareAndPersistence(table=msgTmp, tableName=tabName, asynWrite=true, compress=true, cacheSize=10000000, retentionMinutes=60, flushMode=0, preCache=1000000)
        
        timeRS = cutPoints(09:30:00.000..15:00:00.000, 23)
        
        inputDict = dict(replayName, each(dsTb{timeRS, startDate, endDate, stkList}, replayName))
        dateDict = dict(replayName, take(`MDDate, replayName.size()))
        timeDict = dict(replayName, take(`MDTime, replayName.size()))
        
        jobId = "replay_" + replayUuid
        jobDesc = "replay stock data"
        submitJob(jobId, jobDesc, replayJob{inputDict, tabName, dateDict, timeDict, replayRate, sortColumn})
        returnBody["errorCode"] = "1"
        returnBody["errorMsg"] = "Replay successfully"
        return returnBody
    }
    catch(ex)
    {
        returnBody["errorCode"] = "0"
        returnBody["errorMsg"] = "Exception occurred when replaying: " + ex
        return returnBody
    }
}

Note: The function definition is flexible and can be customized to suit specific requirements.

The parameters passed in will first be validated and formatted:

• The stkList parameter is limited to a maximum of 50 stocks (defined by maxCnt) for a single replay operation.
• Parameters startDate and endDate are formatted by startDate and endDate using the datetimeParse function.
• replayName is the list of data sources to be replayed. It must be one of "snapshot", "order", "transaction", in this example.

If any parameters fail validation, the function reports errors (as defined in returnBody). Upon successful parameter validation, the function proceeds to initialize the output table.

The output table "msgTmp” is defined as a heterogeneous stream table where the BLOB column stores the serialized result of each replayed record. To optimize memory usage, it is shared and persisted with the enableTableShareAndPersistence function.

When the data source includes both "transaction” and "order”, an extra sort column can be specified for records with the same timestamp. In this example, records in "transaction” and "order” with the same timestamp are sorted by the record number "ApplSeqNum” (See section 3.3 replayJob function for details). In such case, the output table must contain the sort column. If the data source contains only "snapshots”, the sort column is not required.

Then, submit the replay job using the submitJob function. You will be informed about the status of the replay job, whether it has executed successfully or encountered an exception (as defined in returnBody).

The dsTb function is used to prepare the data sources to be replayed.

def dsTb(timeRS, startDate, endDate, stkList, replayName)
{
    if(replayName == "snapshot"){
        tab = loadTable("dfs://Test_snapshot", "snapshot")
	}
	else if(replayName == "order") {
		tab = loadTable("dfs://Test_order", "order")
	}
	else if(replayName == "transaction") {
		tab = loadTable("dfs://Test_transaction", "transaction")
	}
	else {
		return NULL
	}
    ds = replayDS(sqlObj=<select * from tab where MDDate>=startDate and MDDate<endDate and HTSCSecurityID in stkList>, dateColumn='MDDate', timeColumn='MDTime', timeRepartitionSchema=timeRS)
    return ds
}

The dsTb function serves as a higher-level wrapper around the built-in replayDS function.

• It first evaluates the provided replayName to determine which dataset to use.
• Based on the replayName, it loads the corresponding table object from the database.
• Using replayDS, it segments the data with stkList in the range of [startDate, endDate) into multiple data sources based on timeRS, and returns a list of data sources.

The timeRS parameter, which corresponds to the timeRepartitionSchema parameter in the replayDS function, is a vector of temporal types to create finer-grained data sources, ensuring efficient querying of the DFS table and optimal manage memory usage.

The timeRS in this example is defined as a variable in the stkReplay function, timeRS = cutPoints(09:30:00.000..15:00:00.000, 23), which means to divide data in trading hours (09:30:00.000..15:00:00.000) into 23 equal parts.

timeRS = cutPoints(09:30:00.000..15:00:00.000, 3)
startDate = 2021.12.01
endDate = 2021.12.02
stkList = ['000616.SZ']
replayName = ["order"]
ds = dsTb(timeRS, startDate, endDate, stkList, replayName)

DataSource< select [4] * from tab where time(MDTime) < 09:30:00.000,nanotime(MDTime) >= 00:00:00.000000000,date(MDDate) == 2021.12.01,MDDate >= 2021.12.01 and MDDate < 2021.12.02 and SecurityID in ["000616.SZ"] order by MDDate asc,MDTime asc >
DataSource< select [4] * from tab where time(MDTime) < 11:20:00.001,time(MDTime) >= 09:30:00.000,date(MDDate) == 2021.12.01,MDDate >= 2021.12.01 and MDDate < 2021.12.02 and SecurityID in ["000616.SZ"] order by MDDate asc,MDTime asc >
DataSource< select [4] * from tab where time(MDTime) < 13:10:00.001,time(MDTime) >= 11:20:00.001,date(MDDate) == 2021.12.01,MDDate >= 2021.12.01 and MDDate < 2021.12.02 and SecurityID in ["000616.SZ"] order by MDDate asc,MDTime asc >
DataSource< select [4] * from tab where time(MDTime) < 15:00:00.001,time(MDTime) >= 13:10:00.001,date(MDDate) == 2021.12.01,MDDate >= 2021.12.01 and MDDate < 2021.12.02 and SecurityID in ["000616.SZ"] order by MDDate asc,MDTime asc >
DataSource< select [4] * from tab where nanotime(MDTime) <= 23:59:59.999999999,time(MDTime) >= 15:00:00.001,date(MDDate) == 2021.12.01,MDDate >= 2021.12.01 and MDDate < 2021.12.02 and SecurityID in ["000616.SZ"] order by MDDate asc,MDTime asc >

The replayJob function is used to define the replay job.

def replayJob(inputDict, tabName, dateDict, timeDict, replayRate, sortColumn)
{
    if(sortColumn == "NULL")
    {
        replay(inputTables=inputDict, outputTables=objByName(tabName), dateColumn=dateDict, timeColumn=timeDict, replayRate=int(replayRate), absoluteRate=false, parallelLevel=23)
    }
    else
    {
        replay(inputTables=inputDict, outputTables=objByName(tabName), dateColumn=dateDict, timeColumn=timeDict, replayRate=int(replayRate), absoluteRate=false, parallelLevel=23, sortColumns=sortColumn)    
    }
    createEnd(tabName, sortColumn)
}

The replayJob function encapsulates the built-in replay function to perform an N-to-1 heterogeneous replay. Once all required data has been replayed, the createEnd function is called to construct and write an end signal into the replay result, marking the conclusion of the replay process with a special record.

The inputDict, dateDict and timeDict are defined as variables in the stkReplay function:

• inputDict is a dictionary indicating data sources to be replayed. The higher-order function each is used to define the data source for each replayName.
• dateDict and timeDict are dictionaries indicating the date and time columns of the data sources, used for data sorting.

The replayJob function includes a sortColumn parameter for additional data ordering. If the data source consists only of "snapshots” (where ordering is not a concern), sortColumn should be set to NULL. In this case, the built-in replay function is called without its sortColumns parameter. For other data sources ("order" and "transaction"), sortColumn can be used to specify an extra column for sorting the data with the same timestamp.

The createEnd function is used to define the end signal. It can be optionally defined.

def createEnd(tabName, sortColumn)
{
    dbName = "dfs://End"
    tbName = "endline"
    if(not existsDatabase(dbName))
    {
        db = database(directory=dbName, partitionType=VALUE, partitionScheme=2023.04.03..2023.04.04)
        endTb = table(2200.01.01T23:59:59.000 as DateTime, `END as point, long(0) as ApplSeqNum)
        endLine = db.createPartitionedTable(table=endTb, tableName=tbName, partitionColumns=`DateTime)
        endLine.append!(endTb)
    }
     
    ds = replayDS(sqlObj=<select * from loadTable(dbName, tbName)>, dateColumn='DateTime', timeColumn='DateTime')
    
    inputEnd = dict(["end"], [ds])
    dateEnd = dict(["end"], [`DateTime])
    timeEnd = dict(["end"], [`DateTime])
    if(sortColumn == "NULL")
    {
        replay(inputTables=inputEnd, outputTables=objByName(tabName), dateColumn=dateEnd, timeColumn=timeEnd, replayRate=-1, absoluteRate=false, parallelLevel=1)
    }
    else
    {
        replay(inputTables=inputEnd, outputTables=objByName(tabName), dateColumn=dateEnd, timeColumn=timeEnd, replayRate=-1, absoluteRate=false, parallelLevel=1, sortColumns=sortColumn)
    }
}

The createEnd function signals the end of the replay process, writing a record labeled "end” to the output table (specified by tabName). inputEnd, dateEnd, timeEnd are dictionaries with the string "end” as the key, which corresponds to the second column (biz_type) in the output table.

To streamline parsing and consumption of the output table, a separate database is created specifically for the end signal. Within this database, a partitioned table is established. This table must include a time column, with other fields being optional. A simulated record is written to this table. This record is not strictly defined and can be customized as needed. For instructions on how to create database and table, refer to DolphinDB Tutorial: Distributed Database.

The createEnd function also includes a sortColumn parameter to determine whether to call the built-in replay function with or without its sortColumns parameter. The following figure demonstrates an end signal in the output table when sortColumn is set to NULL.

This example runs on a Linux OS, with DolphinDB C++ API environment set up. With executable files compiled, you can run the code from the command line interface. The full script is available in the Appendices: C++ code.

The C++ application interacts with the DolphinDB server using a DBConnection object (conn). Through DBConnection, you can execute scripts and functions on the DolphinDB server and transfer data in both directions.

The application invokes the stkReplay function on the DolphinDB server using the DBConnection::run function. The args holds all the necessary arguments for the stkReplay function. The function's result is captured in a dictionary named 'result'. The application checks the "errorCode” key of the result using the get method. If the errorCode is not 1, it indicates an execution error. In this case, the application returns an error message and terminates the program.

DictionarySP result = conn.run("stkReplay", args);
string errorCode = result->get(Util::createString("errorCode"))->getString();
if (errorCode != "1") 
{
    std::cout << result->getString() << endl;
    return -1;
}

string uuid(int len)
{
    char* str = (char*)malloc(len + 1);
    srand(getpid());
    for (int i = 0; i < len; ++i)
    {
        switch (i % 3)
        {
        case 0:
            str[i] = 'A' + std::rand() % 26;
            break;
        case 1:
            str[i] = 'a' + std::rand() % 26;
            break;
        default:
            str[i] = '0' + std::rand() % 10;
            break;
        }
    }
    str[len] = '\0';
    std::string rst = str;
    free(str);
    return rst;
}

This example runs on a Windows OS, with DolphinDB Python API environment set up. The full script is available in the Appendices: Python Code.

The Python application interacts with the DolphinDB server using a session object (s). Through sessions, you can execute scripts and functions on the DolphinDB server and transfer data in both directions.

Once the necessary variables (stk_list, start_date, end_date, replay_rate, replay_uuid, and replay_name) have been defined, the upload method is used to upload these objects to DolphinDB server and the run method is invoked to execute the stkReplay function.

stk_list = ['000616.SZ','000681.SZ']
start_date = '20211201'
end_date = '20211201'
replay_rate = -1
replay_name = ['snapshot']
s.upload({'stk_list':stk_list, 'start_date':start_date, 'end_date':end_date, 'replay_rate':replay_rate, 'replay_uuid':uuidStr, 'replay_name':replay_name})
s.run("stkReplay(stk_list, start_date, end_date, replay_rate, replay_uuid, replay_name)")

def uuid(length):
    str=""
    for i in range(length):
        if(i % 3 == 0):
            str += chr(ord('A') + random.randint(0, os.getpid() + 1) % 26)
        elif(i % 3 == 1):
            str += chr(ord('a') + random.randint(0, os.getpid() + 1) % 26)
        else:
            str += chr(ord('0') + random.randint(0, os.getpid() + 1) % 10)
    return str

uuidStr = uuid(16)

• Processor family: Intel(R) Xeon(R) Silver 4216 CPU @ 2.10GHz
• CPU(s): 64
• Memory: 503 GB
• Disk: SSD
• OS: CentOS Linux release 7.9.2009 (Core)
• DolphinDB Server: version 2.00.9.3 (released on 2023.03.29)

• Server Version: 2.00.9.3 (released on 2023.03.29)
• Deployment: standalone mode
• Configuration: cluster.cfg

maxMemSize=128
maxConnections=5000
workerNum=24
webWorkerNum=2
chunkCacheEngineMemSize=16
newValuePartitionPolicy=add
logLevel=INFO
maxLogSize=512
node1.volumes=/ssd/ssd3/pocTSDB/volumes/node1,/ssd/ssd4/pocTSDB/volumes/node1
node2.volumes=/ssd/ssd5/pocTSDB/volumes/node2,/ssd/ssd6/pocTSDB/volumes/node2
node3.volumes=/ssd/ssd7/pocTSDB/volumes/node3,/ssd/ssd8/pocTSDB/volumes/node3
diskIOConcurrencyLevel=0
node1.redoLogDir=/ssd/ssd3/pocTSDB/redoLog/node1
node2.redoLogDir=/ssd/ssd4/pocTSDB/redoLog/node2
node3.redoLogDir=/ssd/ssd5/pocTSDB/redoLog/node3
node1.chunkMetaDir=/ssd/ssd3/pocTSDB/metaDir/chunkMeta/node1
node2.chunkMetaDir=/ssd/ssd4/pocTSDB/metaDir/chunkMeta/node2
node3.chunkMetaDir=/ssd/ssd5/pocTSDB/metaDir/chunkMeta/node3
node1.persistenceDir=/ssd/ssd6/pocTSDB/persistenceDir/node1
node2.persistenceDir=/ssd/ssd7/pocTSDB/persistenceDir/node2
node3.persistenceDir=/ssd/ssd8/pocTSDB/persistenceDir/node3
maxPubConnections=128
subExecutors=24
subThrottle=1
persistenceWorkerNum=1
node1.subPort=8825
node2.subPort=8826
node3.subPort=8827
maxPartitionNumPerQuery=200000
streamingHAMode=raft
streamingRaftGroups=2:node1:node2:node3
node1.streamingHADir=/ssd/ssd6/pocTSDB/streamingHADir/node1
node2.streamingHADir=/ssd/ssd7/pocTSDB/streamingHADir/node2
node3.streamingHADir=/ssd/ssd8/pocTSDB/streamingHADir/node3
TSDBCacheEngineSize=16
TSDBCacheEngineCompression=false
node1.TSDBRedoLogDir=/ssd/ssd3/pocTSDB/TSDBRedoLogDir/node1
node2.TSDBRedoLogDir=/ssd/ssd4/pocTSDB/TSDBRedoLogDir/node2
node3.TSDBRedoLogDir=/ssd/ssd5/pocTSDB/TSDBRedoLogDir/node3
TSDBLevelFileIndexCacheSize=20
TSDBLevelFileIndexCacheInvalidPercent=0.6
lanCluster=0
enableChunkGranularityConfig=true

Note: Modify configuration parameter persistenceDir with your own path.

• Processor family: Intel(R) Core(TM) i5-11500 @ 2.70GHz 2.71 GHz
• CPU(s): 12
• Memory: 16 GB
• OS: Windows 11 Pro
• DolphinDB GUI Version: 1.30.20.1

See GUI to install DolphinDB GUI.

• C++ API Version: release200.9
  Note: It is recommended installing C++API with the version corresponding to DolphinDB server. For example, install the API of release 200 for the DolphinDB server V2.00.9.
• For instruction on installing and using C++ API, see C++ API Reference Guide.

• Python API Version: 1.30.21.1
• For instruction on installing and using Python API, see Python API Reference Guide.