Real-time Optimal Price Selection from Multiple Forex Sources

The environment must first be cleaned, and the necessary stream tables must be created to receive and process the forex market data.

def cleanEnvironment(){
    try{ unsubscribeTable(tableName = `HSBCStreamTable, actionName = "joinLeft") }catch(ex){ print(ex) }
    try{ dropStreamTable(`HSBCStreamTable)}catch(ex){ print(ex) }
    try{ unsubscribeTable(tableName = "CFETSTStreamTable", actionName = "joinRight")}catch(ex){ print(ex) }
    try{ dropStreamTable(`CFETSTStreamTable)}catch(ex){ print(ex) }
    try{ dropStreamTable(`bestPriceStreamTable)}catch(ex){ print(ex) }
    try{ dropStreamEngine(`bestPricEngine)}catch(ex){ print(ex) }
    try{ dropStreamEngine(`fillPrevEngine)}catch(ex){ print(ex) }
}
cleanEnvironment()
def createStreamTableFunc()
{
    // Define streamTables that receive original market data: HSBCStreamTable and CFETSTStreamTable
    colName = `market`currencyPair`time`bid`bidVolume
	colType = [`SYMBOL,`SYMBOL,`TIMESTAMP,"DOUBLE","INT"]
    HSBCTemp = streamTable(20000000:0,colName,colType)
    try{ enableTableShareAndPersistence(table = HSBCTemp, tableName = "HSBCStreamTable", asynWrite = true, compress = true, cacheSize = 20000000, retentionMinutes = 1440, flushMode = 0, preCache = 10000) }
	catch(ex){ print(ex) }
    CFETSTemp = streamTable(20000000:0,colName,colType)
    try{ enableTableShareAndPersistence(table = CFETSTemp, tableName = "CFETSTStreamTable", asynWrite = true, compress = true, cacheSize = 20000000, retentionMinutes = 1440, flushMode = 0, preCache = 10000) }
	catch(ex){ print(ex) }
    // Define streamTables that save results: bestPriceTempStreamTable
    colName = `currencyPair`time`market`bestPrice`leftBid`rightBid`delay
	colType = [`SYMBOL,`TIMESTAMP,`INT,`DOUBLE,`DOUBLE,`DOUBLE,`DOUBLE]
    bestPriceTemp = streamTable(20000000:0,colName,colType)
    try{ enableTableShareAndPersistence(table = bestPriceTemp, tableName = "bestPriceStreamTable", asynWrite = true, compress = true, cacheSize = 20000000, retentionMinutes = 1440, flushMode = 0, preCache = 10000) }
	catch(ex){ print(ex) }
}
createStreamTableFunc()
go

The above script calls enableTableShareAndPersistence to create three shared persisted tream tables. The “HSBCStreamTable” and “CFETSTStreamTable” are used to receive and publish real-time market data from HSBC and CFETST, respectively. The “bestPriceStreamTable” is used to receive and publish the output results table after processing by the engine.

Note: To ensure the successful execution of the enableTableShareAndPersistence command, it is necessary to first configure the configuration parameter persistenceDir in the configuration file (dolohindb.cfg in standalone mode and cluster.cfg in cluster mode). For details of this configuration parameter, see Reference.

Define a function to implement the selection of the optimal priceand market. All input parameters of this function, except for the threshold, are vectors corresponding to the columns in the “tick” table. The threshold represents the anomaly thresholds for each currency pair generated from the simulated data. This function will later be specified as a parameter in the equi join engine.

def bestPriceAndMarket(leftBid, rightBid, leftMarket, rightMarket, currencyPair, threshold)
{
    price = iif(leftBid > threshold[currencyPair] and rightBid > threshold[currencyPair],NULL,iif((leftBid > threshold[currencyPair]),rightBid,iif(rightBid > threshold[currencyPair],leftBid,max(leftBid,rightBid))))
    market = iif(leftBid > threshold[currencyPair] and rightBid > threshold[currencyPair],NULL,iif((leftBid > threshold[currencyPair]),2,iif(rightBid > threshold[currencyPair],1,iif(leftBid > rightBid,1,2))))
    return market, price
}

In the code above:

• The statement iif(cond, trueResult, falseResult) indicates that if the condition cond is met, it returns trueResult; otherwise, it returns falseResult. This is equivalent to running if(cond[i]) truResult(cond[i]) else falseResult(cond[i]) statement.
• To facilitate subsequent data filling, the market type is converted to an integer, with the left table returning 1 and the right table returning 2.

We implement this solution by combining the equi join engine and the reactive state engine into a pipeline, where the output of the equi join engine serves as the input for the reactive state engine. The code is as follows:

fillPrevSch = table(1:0,`time`currencyPair`market`bestPrice`leftBid`rightBid`inTime,[`TIMESTAMP,`SYMBOL,`INT,`DOUBLE,`DOUBLE,`DOUBLE,`NANOTIMESTAMP])
// create reactiveStateEngine
fillPrevEngine = createReactiveStateEngine(name = "fillPrevEngine", metrics = [<Time>,<ffill(Market)>,<ffill(BestPrice)>,<leftBid>,<rightBid>,<(now(true) - inTime) \ 1000000>], dummyTable = fillPrevSch, outputTable = bestPriceStreamTable, keyColumn="CurrencyPair")
HSBC_Table_left = table(1:0,`market`currencyPair`time`bid`bidVolume`inTime,[`SYMBOL,`SYMBOL,`TIMESTAMP,`DOUBLE,`INT,`NANOTIMESTAMP])
CFETS_Table_right = table(1:0,`market`currencyPair`time`bid`bidVolume`inTime,[`SYMBOL,`SYMBOL,`TIMESTAMP,`DOUBLE,`INT,`NANOTIMESTAMP])
// create equiJoinEngine
bestPricEejEngine = createEquiJoinEngine(name = "bestPricEngine", leftTable = HSBC_Table_left, rightTable = CFETS_Table_right, outputTable = fillPrevEngine, metrics = [<bestPriceAndMarket(HSBC_Table_left.bid,CFETS_Table_right.bid,HSBC_Table_left.market,CFETS_Table_right.market,currencyPair,threshold) as `market`bestPrice>,<HSBC_Table_left.bid>,<CFETS_Table_right.bid>,<max(HSBC_Table_left.inTime,CFETS_Table_right.inTime)>], matchingColumn = `currencyPair, timeColumn = `time)
def appendTime(engine,isLeftTable,mutable data)
{
    data[`inTime] = now(true)
    appendForJoin(engine,isLeftTable,data)
}
subscribeTable(tableName="HSBCStreamTable", actionName="joinLeft", offset=0, handler=appendTime{bestPricEejEngine, true,}, msgAsTable=true)
subscribeTable(tableName="CFETSTStreamTable", actionName="joinRight", offset=0, handler=appendTime{bestPricEejEngine, false,}, msgAsTable=true)

In the code above：

• The table variable "fillPrevSch" specifies the input table schema of the reactive state engine, and its column types and order must match exactly with those of the output table (outputTable) of the equi join engine.
• In the metrics of the reactive state engine, the ffill function is specified to forward fill null values. Non-empty previous values for the same currency pair are used for filling.
• The parameter outputTable of the equi join engine is specified as the reactive state engine ("fillPrevEngine").
• The function bestPriceAndMarket, which selects the optimal price and market, is specified in the parameter metrics of the equi join engine.
• The user-defined function appendTime reserves a time column (in nanoseconds) to record the timestamp of data ingested into the equi join engine, facilitating subsequent delay statistics.

We can subscribe to the output table of the reactive state engine (optimal price results) through APIs, plugins, or other methods for operations such as visualization and monitoring.

In this section, we implement the replay of market data and simulate real-time data writing. The code is as follows:

//replay data
submitJob("replayL","replayToLeft",replay,CFETS_Tick,`CFETSTStreamTable,`Time,,100)
submitJob("replayR","replayToRight",replay,HSBC_Tick,`HSBCStreamTable,`Time,,100)
t = getRecentJobs(2)

After execution, check the information in the variable "t":

If the "endTime" and "errorMsg" fields are empty, it indicates that the task is running as expected.