Testing machine learning models on simulated real-time data feeds | Deephaven

To maintain a competitive edge in the crypto space, you should be using AI to predict prices and manage investments. Doing so isn’t as difficult as you might think.

This is the fourth of a six-part blog series on real-time crypto price predictions with AI. In this blog, I’ll test the AI models I created in the previous two installments on a simulated real-time feed. Keep up with the rest of the blog series:

1. Acquire up-to-date crypto data with Apache Airflow
2. Implement an LSTM model with TensorFlow
3. Implement a linear regression model with Nvidia RAPIDS
4. Test the models on simulated real-time data
5. Implement the models on real-time crypto data from Coinbase
6. Share AI predictions with URIs

Testing code on simulated real-time data has applications that reach far and wide beyond machine learning. Whether you want to calculate financial indicators, find outliers, smooth data, or even just watch data tick, simulation is key to maximize efficacy.

Deephaven’s TableReplayer allows you to simulate true real-time data by using a historical data set with timestamps. By supplying start and end times in the Deephaven format, you can replay the historical data you’re familiar with and test your algorithms on it.

In the previous two blogs, I built and trained models in both TensorFlow and Nvidia RAPIDS. Both models used a training set called train_dh. I additionally created a test set, test_dh, but never used it. The test set contains BTC price data from two different date ranges. I finally get to use it in conjunction with TableReplayer to test my models on real-time data. This workflow is powerful for real-time AI applications, since you get to see if your models work on an actual real-time feed.

In the code below, I aptly call my table replayer replayer, and supply it a start and end time in Deephaven format. The add_table method tells it which table to replay, followed by the name of the column that contains date-times. From there, call start and the table ticks with the simulated real-time feed. The dates correspond to the testing set for the TensorFlow model.

from deephaven.replay import TableReplayer
import deephaven.time as dhtu

start_time = dhtu.to_datetime("2022-07-27T21:10:00 NY")
end_time = dhtu.to_datetime("2022-07-27T22:10:00 NY")

replayer = TableReplayer(start_time, end_time)
crypto_replayed = replayer.add_table(test_dh, "Date")
replayer.start()

TensorFlow LSTM model​

The code for building the TensorFlow model can be found here. I’ll be using some values found in the code in that blog, including n_input and n_features. I’ll define how to apply the model to the data feed in a function.

from deephaven.replay import TableReplayer
from deephaven import numpy as dhnp
from deephaven.learn import gather
import deephaven.time as dhtu
from deephaven import learn

import numpy as np

start_time = dhtu.to_datetime("2022-07-27T21:10:00 NY")
end_time = dhtu.to_datetime("2022-07-27T22:10:00 NY")

replayer = TableReplayer(start_time, end_time)
crypto_replayed = replayer.add_table(test_dh, "Date")
replayer.start()

def table_to_numpy_double(rows, cols):
    return gather.table_to_numpy_2d(rows, cols, np_type=np.double)

def get_predicted_price(data, idx):
    return data

new_data = dhnp.to_numpy(train_dh.select(["Price"])).reshape(-1, 1)
first_eval_batch = new_data[-n_input:]
last_three = first_eval_batch.reshape((1, n_input, n_features))

def predict_with_model(data):
    global last_three
    
    current_pred = model.predict(last_three)
    current_pred=scaler.inverse_transform(current_pred)
    current_pred=current_pred.reshape(1,-1)[0]
    add_data = data[0]
    value = ((add_data[0]- x_min)/(x_max - x_min))* (1 - (-1)) + (-1)
    value=np.array([value])
    
    last_three = np.append(last_three[:,1:,:],[[value]],axis=1)

    return current_pred

The final step is to use these functions and code to apply the model. I use deephaven.learn to apply the model to data in the replayed table.

lstm_in_realtime = learn.learn(
    table=crypto_replayed,
    model_func=predict_with_model,
    inputs=[learn.Input("Price", table_to_numpy_double)],
    outputs=[learn.Output("Predicted_price", get_predicted_class, "double")],
    batch_size=1
)

Nvidia RAPIDS linear regression model​

The same principles apply here. We need some code to define how our model gets applied to the real-time feed. That code will look very similar to what’s in the previous section. Except this time, we’ll apply it to a different data set of crypto prices on August 9, 2022.

from deephaven.replay import TableReplayer
from deephaven import time as dhtu
from deephaven.learn import gather
from deephaven import learn

import numpy as np

start_time = to_datetime("2022-08-09T19:18:00 NY")
end_time = to_datetime("2022-08-09T20:18:00 NY")

replayer = TableReplayer(start_time, end_time)
crypto_replayed = replayer.add_table(test_dh, "Date")
replayer.start()

def use_fitted_model(features):
    return linear_regression_gpu.predict(features)

def table_to_numpy_double(rows, cols):
    return gather.table_to_numpy_2d(rows, cols, np_type=np.double)

def get_predicted_price(data, idx):
    return data

linreg_in_realtime = learn.learn(
    table = crypto_replayed,
    model_func = use_fitted_model,
    inputs = [learn.Input(["Price_1","Price_2","Price_3"], table_to_numpy)],
    outputs = [learn.Output("Predicted_Price", get_predicted_price, "double")],
    batch_size = 1
)

I’m happy with the outputs of the models. Now that I’ve tested them on simulated real-time feeds, I’m ready for the big leagues. In the next blog, I’ll use these models on true real-time data.