Price Prediction with Current Data and a Pre-Trained Model
Using the LSTM model for price prediction. This final and most rewarding step in our data driven workflow is easilly set. Armed with a well trained model and the latest data from Binance or Yahoo Finance at our hands (gathered as described in this post), we are finally ready to reap the reward for our hard work: use the model for a price prediction of this asset for the coming hours or days. Compared to the original training of the model, forecasting with a limited dataset can now be done very quickly, a matter of a few minutes instead of hours!
Get Data, Load the Model and Go!
LSTM model usage for price prediction is the purpose of the UseLSTM class. Note: We recently updated the class to now use our generic timeseries sequencer instead of the ‘old’ Keras’ TimeseriesGenerator.
This Python class loads a pre-save model, loads the datasets, than compiles the model and fits it on the current data to finally make a price prediction for the next 12 hours and plots the data with the projected forecast.
Preliminaries
Things start with loading the necessary libraries such as keras, tensorflow, plotly and several utilities. Also things needed for mult-threading and communicating with the Main Window are loaded. The class is shown here for use with the PyQt6 GUI Framework but it can be used just as well in a CL terminal application or a Jupyther Notebook.
# Copyright (c) 2025 Hans De Weme
# Licensed under the MIT License (https://opensource.org/licenses/M
# Class: UseLSTM
# Purpose: load a pre-save model, load the dataset, compile the model and fit it on the current data
# then finally make a price prediction for the next 12 hours and plot the data with the forecast attached
import numpy as np
import pandas as pd
import json
import os
os.environ['TF_CPP_MIN_LOG_LEVEL'] = '1'
import tensorflow as tf
# from keras.models import TimeseriesGenerator
from custom_keras import CustomSequenceGenerator
from keras.callbacks import EarlyStopping
from keras.models import load_model
from sklearn.model_selection import train_test_split
from keras.callbacks import LearningRateScheduler
from pandas.tseries.offsets import DateOffset
from sklearn.preprocessing import MinMaxScaler
from pathlib import Path
from get_current_data import getCurrent
from PyQt6.QtCore import QThread, pyqtSignal, pyqtSlot
import plotly.graph_objs as go
import plotly.io as pio
import warnings
warnings.filterwarnings("ignore")
Initialization
When initializing the class connects to the calling parent and loads the general settings. The Main Window then starts this worker thread by calling the run() method, that orchestrates the processing: loading the model, getting the dataset, using the model on the dataset.
class UseLSTM(QThread):
progress_signal = pyqtSignal(str) # Signal to communicate progress (string message) back to the main thread
failure_signal = pyqtSignal(str) # Signal to request saving the model
file_requested = pyqtSignal(str)
file_selected = pyqtSignal(str)
request_input_signal = pyqtSignal()
received_input_signal = pyqtSignal(float)
def __init__(self, asset, settings, kind, parent=None):
super().__init__() # necessary for QObject, needed for pyqtSignal
self.user_input = None
self.received_input_signal.connect(self.on_input_received)
self.parent = parent
self.selected_file = None
self.file_selected.connect(self.on_file_selected) # Connect to update internal state
self.progress_signal.connect(parent.set_status_message)
self.settings = settings
self.asset = asset.upper()
self.FREQ = '1h'
self.kind = kind
if self.kind == 'S':
self.MARKT = self.asset
else:
self.MARKT = self.asset+'USDT'
self.hps = False
self.loss = 'mean_squared_error'
self.opti = 'SGD'
self.progress_signal.emit("Libraries loaded!")
def run(self):
if self.get_model() == False:
self.failure_signal.emit("No Valid Model Found!")
return
else:
if self.get_data() == False:
self.failure_signal.emit("No Data Loaded!")
return
self.use_model()
Loading Paramaters, Model and Dataset
Before the LSTM model can be used for actual price prediction, the class must assure it can find and read the pre-saved hyperparameters, it can find and load the correct model -if needed with the user chosing the one intended- and it receives the intended dataset
def load_settings(self, pad):
if not Path(pad).exists():
print(f"File '{pad}' does not exist.")
return False
try:
with open(pad, 'r') as file:
self.hp = json.load(file)
print("\nSettings File with Hyperparameters loaded successfully.")
return True
except json.JSONDecodeError:
print("\nError: Settings File exists but contains invalid JSON.")
return False
except Exception as e:
print(f"\nAn unexpected error occurred trying to read Settings File: {e}")
return None
def get_model(self): # Load LSTM model from Disk when present
self.suc = False
substring = self.MARKT+'_LSTM_'
pad = self.settings['models']
pad = Path(pad)
dir = pad.resolve()
print('looking for saved LSTM model in :'+str(dir))
matching_dirs = [subdir for subdir in dir.iterdir() if subdir.is_dir() and substring in subdir.name]
if len(matching_dirs) == 0:
print(f"No directories containing '{substring}' found.")
return False
elif len(matching_dirs) > 1:
print(f"Multiple directories containing '{substring}' found:")
for directory in matching_dirs:
print(f"- {directory}")
self.file_requested.emit('LSTM') # Open a file dialog and get the selected file path in the Main Window
while self.selected_file is None: # Wait until the file is selected
self.msleep(100) # Sleep briefly to allow main thread to update
self.lstm_naam = self.selected_file
if self.lstm_naam == "No Directory selected":
return False
else:
self.suc = True
else:
print(f"Found exactly one directory: {matching_dirs[0]}")
self.suc = True
self.lstm_naam = matching_dirs[0]
if self.suc == True:
hp_file = 'LSTM_hp_'+self.MARKT+'.json'
hp_path = dir / hp_file
if self.load_settings(hp_path) == True:
self.hps =True
self.loss = self.hp['loss']
self.opti = self.hp['optimizer']
print('Hyperparameters loaded: Optimizer, Loss')
return self.suc
@pyqtSlot(str) # Slot to update selected file path
def on_file_selected(self, file_path):
self.selected_file = file_path
def get_data(self):
self.data = getCurrent(self.kind, self.asset, trim = True, input_callback=self.request_input_from_main_thread) # trim dataset down to close price only
if self.data.get_data(self.kind) == True:
self.df = pd.DataFrame(self.data.df)
return True
else:
return False
def request_input_from_main_thread(self): # This method is used as a callback for get_current_data module
self.request_input_signal.emit()
while self.user_input is None: # Wait until the input is received
self.msleep(100)
return self.user_input
@pyqtSlot(float)
def on_input_received(self, input_value):
self.user_input = input_value
Using the Model
We finally load the model and plot the dataset. After normalizing the data it is split into training and testing data.
We then compile the model and fit it on the training data and evaluate it against the test data, plotting prediction and actual data. At last, with the method make_future_forecast a forecast for the next 12 hours beyond the dataset is made and plotted as well.
def use_model(self):
lstm_model = load_model(self.lstm_naam, custom_objects=None, compile=False, options=None)
# when on Keras 3, use added safe_mode = False
# lstm_model = load_model(self.lstm_naam, custom_objects=None, compile=False, safe_mode=False)
print('* * * L S T M model geladen: '+str(self.lstm_naam))
self.progress_signal.emit("Model loaded!"+str(self.lstm_naam))
N_INPUT = 12 # number of new datapoints to predict
N_TARGETS = 1 # number of targets to predict: price (df['close'])
BATCH_SIZE = 32 # 128 number of sequences in a training batch (must be a power of 2)
SEQUENCE_SIZE = 36 # number of datapoints in a training sequence (we predict 12 hours, 0.5 day, so let's use a size of 1.5 days)
TRAIN_SPLIT = 0.2 # size of test data set apart from train data
train_size = int(len(self.df) * (1-TRAIN_SPLIT)) # save original test data for later comparison
testo = self.df.iloc[train_size:]
plot_data = [go.Scatter(x=testo.index, y=testo['close'], name='price' )]
plot_layout = go.Layout(title=self.MARKT+' prijsinfo testset')
fig = go.Figure(data=plot_data, layout=plot_layout)
fig.show()
total = self.df # normalize price values: total = df scaled
scaler = MinMaxScaler()
scaler.fit(total)
total = scaler.transform(total)
train, test = train_test_split(total, test_size=TRAIN_SPLIT, shuffle=False)
train, vali = train_test_split(train, test_size=TRAIN_SPLIT, shuffle=False)
train_generator = CustomSequenceGenerator(train, SEQUENCE_SIZE, BATCH_SIZE, shuffle=False)
vali_generator = CustomSequenceGenerator(vali, SEQUENCE_SIZE, BATCH_SIZE, shuffle=False, drop_last=False)
test_generator = CustomSequenceGenerator(test, SEQUENCE_SIZE, BATCH_SIZE, shuffle=False)
lstm_model.compile(self.opti, self.loss)
def scheduler(epoch, lr):
if epoch < 10:
return lr
else:
return lr * tf.math.exp(-0.1)
lr_schedule = LearningRateScheduler(scheduler)
earlystop = EarlyStopping(monitor='val_loss', patience=3, restore_best_weights=True)
try:
history = lstm_model.fit(train_generator, epochs=50, validation_data=vali_generator, callbacks=[earlystop, lr_schedule], verbose=1)
except Exception as e:
print(f"Error during model-fit: {e}")
print("\n* * * Model history: {}".format(history.history.keys())) # plot the result loss during training epochs
hist = pd.DataFrame(history.history)
hist['epoch'] = history.epoch
plot_data = [go.Scatter(x=hist['epoch'], y=hist['loss'], name='loss' ), go.Scatter(x=hist['epoch'], y=hist['val_loss'], name='value_loss')]
plot_layout = go.Layout(title='Training loss')
fig = go.Figure(data=plot_data, layout=plot_layout)
pio.show(fig)
# evaluate against the test set data
result=lstm_model.evaluate(test_generator)
print("\n* * * Evaluate training against the testset, results: {}".format(result))
y_hat = lstm_model.predict(test_generator) # test model against the test set
y_hat_inverse = scaler.inverse_transform(y_hat)
aligned_testo = testo.iloc[SEQUENCE_SIZE:] # Skip the initial points in testo to align with y_hat_inverse
if len(y_hat_inverse) < len(aligned_testo):
aligned_testo = aligned_testo.iloc[:len(y_hat_inverse)]
plot_data = [
go.Scatter(x=aligned_testo.index, y=aligned_testo['close'], mode='lines', name='Actual', line=dict(color='green')),
go.Scatter(x=aligned_testo.index, y=y_hat_inverse.flatten(), mode='lines', name='Predicted', line=dict(color='red'))
]
plot_layout = go.Layout(title=self.MARKT+' Testset - Prediction', xaxis_title='Time', yaxis_title='Price')
fig = go.Figure(data=plot_data, layout=plot_layout) # take the predictions and plot the result
pio.show(fig)
future_df = self.make_future_forecast(model=lstm_model, df=self.df, scaler=scaler, n_input=SEQUENCE_SIZE, n_steps=N_INPUT, n_targets=N_TARGETS)
plot_data = [go.Scatter(x=future_df.index, y=future_df['Prediction'], name='Forecast', line=dict(color='green'))]
layout = go.Layout(title=self.MARKT + ' Price Projection', width=1200, height=800)
fig = go.Figure(data=plot_data, layout=layout)
fig.show()
print(future_df)
def make_future_forecast(self, model, df, scaler, n_input, n_steps, n_targets=1):
"""
Predict n future steps from the end of df using the trained model.
Args:
model: Trained LSTM model
df: Original unscaled dataframe (e.g. self.df)
scaler: Previously fit MinMaxScaler
n_input: Number of timesteps per input sequence
n_steps: Number of steps to forecast
n_targets: Number of output features (default=1)
Returns:
DataFrame with datetime index and predicted values
"""
last_sequence = df[-n_input:] # Extract the last known input window (unscaled)
last_scaled = scaler.transform(last_sequence)
sequence = last_scaled.reshape((1, n_input, n_targets))
pred_list = []
for _ in range(n_steps):
prediction = model.predict(sequence, verbose=0)
pred_list.append(prediction[0]) # Store raw prediction
sequence = np.concatenate([sequence[:, 1:, :], prediction[:, np.newaxis, :]], axis=1) # Update input sequence
pred_array = np.array(pred_list) # Inverse transform the predictions
pred_unscaled = scaler.inverse_transform(pred_array)
last_timestamp = df.index[-1] # Build a future datetime index
future_index = [last_timestamp + pd.DateOffset(hours=i + 1) for i in range(n_steps)]
forecast_df = pd.DataFrame(pred_unscaled, index=future_index, columns=['Prediction'])
return forecast_dfA not so optimistic price forecast beyond the test set.
Functional and Technical Documentation for UseLSTM Class
The UseLSTM class is a PyQt6-based implementation that loads a pre-trained Long Short-Term Memory (LSTM) model to make price predictions for a financial asset over the next 12 hours. It includes functionality for:
Forecasting & Plotting
- Loading a previously trained LSTM model.
- Fetching and preprocessing the latest market data.
- Fine-tuning the model using the most recent data.
- Making price predictions and plotting the forecast.
- Emitting progress updates via PyQt signals.
2. Functional Documentation
2.1 Key Features
- Model Loading: Searches for and loads a pre-trained LSTM model.
- Data Acquisition: Retrieves the latest market data using the
getCurrentmodule. - Preprocessing: Normalizes and structures data for LSTM inference.
- Fine-Tuning: Fits the model with new data before making predictions.
- Prediction and Visualization: Generates forecasts and plots them using Plotly.
- Progress Updates: Uses PyQt signals to communicate status updates to the main application.
2.2 Inputs and Outputs
| Input | Description |
|---|---|
asset | The name of the financial asset (e.g., “BTCUSDT”). |
settings | Dictionary containing paths and hyperparameters. |
kind | Type of asset (e.g., cryptocurrency or stock). |
| Output | Description |
|---|---|
| Progress Messages | Status updates via PyQt signals. |
| Model Checkpoints | Updated model after fine-tuning. |
| Forecasted Prices | Graphical plots of predicted vs. actual prices. |
| Log Messages | Console logs for debugging and analysis. |
2.3 User Interaction
- Users are prompted to select a model if multiple are found.
- Progress updates are displayed during execution.
- Users may need to provide input for dataset trimming.
3. Technical Documentation
3.1 Class Definition
class UseLSTM(QThread)
Inherits from QThread to allow execution in a separate thread, preventing UI blocking.
3.2 Dependencies
import os
import json
import numpy as np
import pandas as pd
import tensorflow as tf
from keras.models import load_model
from keras.preprocessing.sequence import TimeseriesGenerator
from keras.callbacks import EarlyStopping, LearningRateScheduler
from sklearn.preprocessing import MinMaxScaler
from sklearn.model_selection import train_test_split
from pandas.tseries.offsets import DateOffset
from PyQt6.QtCore import QThread, pyqtSignal, pyqtSlot
import plotly.graph_objs as go
import plotly.io as pio
from get_current_data import getCurrent
3.3 Attributes
| Attribute | Description |
|---|---|
progress_signal | Signal for progress updates. |
failure_signal | Signal for error notifications. |
file_requested | Signal to request file selection. |
file_selected | Signal to capture selected file. |
request_input_signal | Signal to request user input. |
received_input_signal | Signal to receive user input. |
df | Pandas DataFrame containing price data. |
scaler | MinMaxScaler object for data normalization. |
lstm_model | The loaded LSTM model. |
3.4 Key Methods
run()
Executes the full pipeline:
- Calls
get_model()to load the trained LSTM model. - Calls
get_data()to fetch the latest market data. - Calls
use_model()to fine-tune and make predictions.
get_model()
- Searches for a saved LSTM model in the user-defined directory.
- Loads the corresponding hyperparameter settings if available.
- Prompts the user if multiple models are found.
get_data()
- Retrieves the latest price data using
getCurrent(). - Trims the dataset to retain only the relevant columns.
use_model()
- Normalizes and splits the dataset into train, validation, and test sets.
- Fine-tunes the LSTM model using recent data.
- Generates price forecasts for the next 12 hours.
- Plots actual vs. predicted prices and projected future prices.
load_settings(pad: str) -> bool
- Loads hyperparameter settings from a JSON file.
request_input_from_main_thread()
- Sends a signal to request user input and waits for the response.
on_file_selected(file_path: str)
- Stores the selected file path when a user chooses a model.
on_input_received(input_value: float)
- Captures and stores the received user input.
3.5 Model Fine-Tuning
| Parameter | Default Value |
|---|---|
BATCH_SIZE | 128 |
SEQUENCE_SIZE | 36 |
N_INPUT | 12 |
LOSS | ‘mean_squared_error’ |
OPTIMIZER | ‘SGD’ |
EPOCHS | 10 |
TRAIN_SPLIT | 0.2 |
3.6 Prediction Workflow
- Load the pre-trained LSTM model.
- Prepare the dataset for inference.
- Fine-tune the model using the latest market data.
- Generate predictions for the next 12 hours.
- Plot actual vs. predicted prices.
- Store predictions in a DataFrame for further analysis.
4. Conclusion
The UseLSTM class provides a streamlined way to apply a trained LSTM model to real-time financial data, fine-tune it, and make short-term price forecasts. It is designed to integrate with PyQt applications, allowing interactive selection and feedback mechanisms.