Popular Libraries

Import the aesera, and sklearn libraries.

import aesara
import aesara.tensor as at
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import MinMaxScaler
import joblib

You need the joblib library to store models.

Get some historical market data to train and test the model. For example, to get data for the SPY ETF during 2020 and 2021, run:

qb = QuantBook()
symbol = qb.add_equity("SPY", Resolution.DAILY).symbol
history = qb.history(symbol, datetime(2020, 1, 1), datetime(2022, 1, 1)).loc[symbol]

You need some historical data to prepare the data for the model. If you have historical data, manipulate it to train and test the model. In this example, use the following features and labels:

The following image shows the time difference between the features and labels:

Follow these steps to prepare the data:

1. Obtain the close price and return direction series.

close = history['close']
returns = data['close'].pct_change().shift(-1)[lookback*2-1:-1].reset_index(drop=True)
labels = pd.Series([1 if y > 0 else 0 for y in returns])   # binary class

2. Loop through the close Series and collect the features.

lookback = 5
lookback_series = []
for i in range(1, lookback + 1):
    df = data['close'].shift(i)[lookback:-1]
    df.name = f"close-{i}"
    lookback_series.append(df)
X = pd.concat(lookback_series, axis=1)
# Normalize using the 5 day interval
X = MinMaxScaler().fit_transform(X.T).T[4:]

3. Convert the lists of features and labels into numpy arrays.

X = np.array(features)
y = np.array(labels)

4. Split the data into training and testing periods.

X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2)

You need to prepare the historical data for training before you train the model. If you have prepared the data, build and train the model. In this example, build a Logistic Regression model with log loss cross entropy and square error as cost function. Follow these steps to create the model:

1. Generate a dataset.

# D = (input_values, target_class)
D = (np.array(X_train), np.array(y_train))

2. Initialize variables.

# Declare Aesara symbolic variables
x = at.dmatrix("x")
y = at.dvector("y")

# initialize the weight vector w randomly using share so model coefficients keep their values
# between training iterations (updates)
rng = np.random.default_rng(100)
w = aesara.shared(rng.standard_normal(X.shape[1]), name="w")

# initialize the bias term
b = aesara.shared(0., name="b")

3. Construct the model graph.

# Construct Aesara expression graph
p_1 = 1 / (1 + at.exp(-at.dot(x, w) - b))       # Logistic transformation
prediction = p_1 > 0.5                          # The prediction thresholded
xent = y * at.log(p_1) - (1 - y) * at.log(1 - p_1)  # Cross-entropy log-loss function
cost = xent.mean() + 0.01 * (w ** 2).sum()      # The cost to minimize (MSE)
gw, gb = at.grad(cost, [w, b])                  # Compute the gradient of the cost

4. Compile the model.

train = aesara.function(
          inputs=[x, y],
          outputs=[prediction, xent],
          updates=((w, w - 0.1 * gw), (b, b - 0.1 * gb)))
predict = aesara.function(inputs=[x], outputs=prediction)

5. Train the model with training dataset.

pred, err = train(D[0], D[1])

# We can also inspect the final outcome
print("Final model:")
print(w.get_value())
print(b.get_value())
print("target values for D:")
print(D[1])
print("prediction on D:")
print(predict(D[0]))    # whether > 0.5 or not

You need to build and train the model before you test its performance. If you have trained the model, test it on the out-of-sample data. Follow these steps to test the model:

1. Call the predict method with the features of the testing period.

y_hat = predict(np.array(X_test))

2. Plot the actual and predicted labels of the testing period.

df = pd.DataFrame({'y': y_test, 'y_hat': y_hat}).astype(int)
df.plot(title='Model Performance: predicted vs actual return direction in closing price', figsize=(12, 5))

3. Calculate the prediction accuracy.

correct = sum([1 if x==y else 0 for x, y in zip(y_test, y_hat)])
print(f"Accuracy: {correct}/{y_test.shape[0]} ({correct/y_test.shape[0]}%)")

You can save and load aesera models using the Object Store.

Save Models

Follow these steps to save models in the Object Store:

1. Set the key name of the model to be stored in the Object Store.

model_key = "model"

2. Call the GetFilePathget_file_path method with the key.

file_name = qb.object_store.get_file_path(model_key)

This method returns the file path where the model will be stored.

3. Call the dump method with the model and file path.

joblib.dump(predict, file_name)

If you dump the model using the joblib module before you save the model, you don't need to retrain the model.

Load Models

You must save a model into the Object Store before you can load it from the Object Store. If you saved a model, follow these steps to load it:

1. Call the ContainsKeycontains_key method with the model key.

qb.object_store.contains_key(model_key)

This method returns a boolean that represents if the model_key is in the Object Store. If the Object Store does not contain the model_key, save the model using the model_key before you proceed.

2. Call GetFilePath with the key.

file_name = qb.object_store.get_file_path(model_key)

This method returns the path where the model is stored.

3. Call load with the file path.

loaded_model = joblib.load(file_name)

This method returns the saved model.