Market Data Replay in DolphinDB

In the following example, we define asof join engine that returns the as of join result of trade and snapshot (from "messageStream") to calculate the transaction cost. See full scripts in 02. CalculateTxnCost_asofJoin.

• Define the shared stream table "prevailingQuotes" as the output table.

  colName = `TradeTime`SecurityID`Price`TradeQty`BidPX1`OfferPX1`Spread`SnapshotTime
  colType = [TIME, SYMBOL, DOUBLE, INT, DOUBLE, DOUBLE, DOUBLE, TIME]
  prevailingQuotesTemp = streamTable(1000000:0, colName, colType)
  enableTableShareAndPersistence(table=prevailingQuotesTemp, tableName="prevailingQuotes", asynWrite=true, compress=true, cacheSize=1000000, retentionMinutes=1440, flushMode=0, preCache=10000)
  prevailingQuotesTemp = NULL

• Define the asof join engine. The user-defined function createSchemaTable is used to get the table schema and use it as a parameter of createAsofJoinEngine.

  def createSchemaTable(dbName, tableName){
  	schema = loadTable(dbName, tableName).schema().colDefs
  	return table(1:0, schema.name, schema.typeString)
  }
  tradeSchema = createSchemaTable("dfs://trade", "trade")
  snapshotSchema = createSchemaTable("dfs://snapshot", "snapshot")
  
  joinEngine=createAsofJoinEngine(name="tradeJoinSnapshot", leftTable=tradeSchema, rightTable=snapshotSchema, outputTable=prevailingQuotes, metrics=<[Price, TradeQty, BidPX1, OfferPX1, abs(Price-(BidPX1+OfferPX1)/2), snapshotSchema.Time]>, matchingColumn=`SecurityID, timeColumn=`Time, useSystemTime=false, delayedTime=1)

  For each stock, the asof join engine matches each record of table "trade" with the most recent record from table "snapshot", and then you can calculate transaction costs with the price and quote.

  Considering the common practice, we set parameter useSystemTime to false in the above script to perform asof join based on the "timeColumn".

  Alternatively, you can set parameter useSystemTime to true or use lookup join engine (see full script in 03. CalculateTxnCost_lookUpJoin).

• Define a stream filter engine.

  def appendLeftStream(msg){
  	tempMsg = select * from msg where Price > 0 and Time>=09:30:00.000
  	getLeftStream(getStreamEngine(`tradeJoinSnapshot)).tableInsert(tempMsg)
  }
  filter1 = dict(STRING,ANY)
  filter1["condition"] = "trade"
  filter1["handler"] = appendLeftStream
  filter2 = dict(STRING,ANY)
  filter2["condition"] = "snapshot"
  filter2["handler"] = getRightStream(getStreamEngine(`tradeJoinSnapshot))
  schema = dict(["trade", "snapshot"], [tradeSchema, snapshotSchema])
  engine = streamFilter(name="streamFilter", dummyTable=messageStream, filter=[filter1, filter2], msgSchema=schema)

  The ingested data is deserialized according to parameter msgSchema and processed based on the handler of parameter filter. The serial execution of multiple handlers guarantees the strict chronological order of data processing.

  The "messageStream" table is splitted into two streams. Table "trade" is processed with user-defined function appendLeftStream before it is ingested into asof join engine as the left table. Table "snapshot" is directly ingested into asof join engine as the right table.

• Subscribe to the replayed data (heterogeneous stream table)

  subscribeTable(tableName="messageStream", actionName="tradeJoinSnapshot", offset=-1, handler=engine, msgAsTable=true, reconnect=true)

  With parameter offset set to -1, the subscription starts with the next new message. To consume data from the first record, it is recommended to submit the subscription first and replay as background job afterwards.

• Check the result

  
  
  

  Within each group of the matchingColumn ("SecurityID"), the order of the output is the same as the order of the input.

In the following example, we send subscribed table ("messageStream") to Kafka consumer.

You need to start Kafka server and install DolphinDB Kafka Plugin first. For details about test environment configuration, see Development Environment. For full scripts about this example, see 04. SendMsgToKafka.

• Load Kafka Plugin and define a Kafka producer

  loadPlugin("/DolphinDB/server/plugins/kafka/PluginKafka.txt")
  go
  producerCfg = dict(STRING, ANY)
  producerCfg["metadata.broker.list"] = "localhost"
  producer = kafka::producer(producerCfg)

  Replace the loadPlugin path and you can modify producerCfg as required. Kafka and DolphinDB server are on the same server, thus, the metadata.broker.list parameter is set as localhost.

• Define the function to send messages to Kafka topic

  def sendMsgToKafkaFunc(dataType, producer, msg){
  	startTime = now()
  	try {
  		kafka::produce(producer, "topic-message", 1, msg, true)
  		cost = now() - startTime
  		writeLog("[Kafka Plugin] Successed to send" + dataType + ":" + msg.size() + "rows," + cost + "ms.")
  	}
  	catch(ex) {writeLog("[Kafka Plugin] Failed to send msg to kafka with error:" +ex)}
  }

  Function kafka::produce sends messages of any table schema in json format to specified Kafka topic. You can check the status of a batch of messages sent previously with the writeLog function.

  You can consume subscribed data after sending messages to Kafka.

• Define a stream filter engine

  filter1 = dict(STRING,ANY)
  filter1["condition"] =  "order"
  filter1["handler"] = sendMsgToKafkaFunc{"order", producer}
  filter2 = dict(STRING,ANY)
  filter2["condition"] = "trade"
  filter2["handler"] = sendMsgToKafkaFunc{"trade", producer}
  filter3 = dict(STRING,ANY)
  filter3["condition"] = "snapshot"
  filter3["handler"] = sendMsgToKafkaFunc{"snapshot", producer}
  
  schema = dict(["order","trade", "snapshot"], [loadTable("dfs://order", "order"), loadTable("dfs://trade", "trade"), loadTable("dfs://snapshot", "snapshot")])
  
  engine = streamFilter(name="streamFilter", dummyTable=messageStream, filter=[filter1, filter2, filter3], msgSchema=schema)

  The ingested data is deserialized according to parameter msgSchema and processed based on the handler of parameter filter. The serial execution of multiple handlers guarantees the strict chronological order of data processing.

  sendMsgToKafka{"order", producer} is a partial application of functional programming, which fixes part of the parameters of the sendMsgToKafka function to generate a new function with fewer parameters.

• Subscribe to the replayed data (heterogeneous stream table)

  subscribeTable(tableName="messageStream", actionName="sendMsgToKafka", offset=-1, handler=engine, msgAsTable=true, reconnect=true)

  With parameter offset set to -1, the subscription starts with the next new message. To consume data from the first record, it is recommended to submit the subscription and then to submit the replay as the background job.

• Check the result in the terminal

  Start consumption from the first topic named "topic-message".

  ./bin/kafka-console-consumer.sh --bootstrap-server localhost:9092 --from-beginning --topic topic-message

  Return:

  
  
  

The following example uses function ThreadedClient::subscribe to subscribe to the replayed data ("messageStream") and outputs results in real-time.

To run the script below, you have to install C++ API first (refer to Development Environment). For full scripts about this example, see 05. SubscribeInCppApi.

int main(int argc, char *argv[]){
    DBConnection conn;
    string hostName = "127.0.0.1";
    int port = 8848;
    bool ret = conn.connect(hostName, port);

    conn.run("login(\"admin\", \"123456\")");
    DictionarySP t1schema = conn.run("loadTable(\"dfs://snapshotL2\", \"snapshotL2\").schema()");
    DictionarySP t2schema = conn.run("loadTable(\"dfs://trade\", \"trade\").schema()");
    DictionarySP t3schema = conn.run("loadTable(\"dfs://order\", \"order\").schema()");

    unordered_map<string, DictionarySP> sym2schema;
    sym2schema["snapshot"] = t1schema;
    sym2schema["trade"] = t2schema;
    sym2schema["order"] = t3schema;
    StreamDeserializerSP sdsp = new StreamDeserializer(sym2schema);
    auto myHandler = [&](Message msg) {
            const string &symbol = msg.getSymbol();
            cout << symbol << ":";
            size_t len = msg->size();
            for (int i = 0; i < len; i++) {
                    cout <<msg->get(i)->getString() << ",";
            }
            cout << endl;
    };

    int listenport = 10260;
    ThreadedClient threadedClient(listenport);
    auto thread = threadedClient.subscribe(hostName, port, myHandler, "messageStream", "printMessageStream", -1, true, nullptr, false, false, sdsp);
    cout<<"Successed to subscribe messageStream"<<endl;
    thread->join();

    return 0;
}

Instance StreamDeserializerSP is specified as a parameter (filter) of function ThreadedClient::subscribe. Therefore, when subscribing, the ingested data is deserialized and sent to the user-defined callback function myHandler.

The listenport parameter is the subscription port for the single-threaded client, you can set any free port of the server where the C++ program is hosted.

• Check the result in the terminal

  
  
  

  Note:

  If the script (in above sections) is executed repeatedly, the overwrite error may be thrown. Therefore, you have to remove all objects that are stored in your environment ( including operations such as unsubscribing tables, droping stream tables and streaming engines, etc.) To clean your environment, see 06. RemoveObjectsFromEnvironment.

A performance testing is conducted for heterogeneous replay.

For details about test environment configuration, seeDevelopment Environment.

Test data: see Test Data for more information.

Test script:

timeRS = cutPoints(09:15:00.000..15:00:00.000, 100)
orderDS = replayDS(sqlObj=<select * from loadTable("dfs://order", "order") where Date = 2020.12.31>, dateColumn=`Date, timeColumn=`Time, timeRepartitionSchema=timeRS)
tradeDS = replayDS(sqlObj=<select * from loadTable("dfs://trade", "trade") where Date = 2020.12.31>, dateColumn=`Date, timeColumn=`Time, timeRepartitionSchema=timeRS)
snapshotDS = replayDS(sqlObj=<select * from loadTable("dfs://snapshot", "snapshot") where Date =2020.12.31>, dateColumn=`Date, timeColumn=`Time, timeRepartitionSchema=timeRS)
inputDict = dict(["order", "trade", "snapshot"], [orderDS, tradeDS, snapshotDS])

submitJob("replay", "replay stock market", replay{inputDict, messageStream, `Date, `Time, , , 3})

With the maximum speed (replayRate is unspecified) and no subscriptions to the output table, it takes 4m18s to replay 101,081,629 records. Around 390,000 records are replayed per second and the maximum memory usage reaches 4.7GB.