Bitcoin Price Prediction in Real Time using MLOps

To see how I used the @step decorator, check out the GitHub link below (steps folder) to walk through the code for other steps in the pipeline i.e. data cleaning, feature engineering, data segmentation, model training, and model testing.

Step 5: Data cleaning

In this step, we will create different techniques to clean the imported data. We will discard unwanted columns and missing values ​​from the data.

class DataPreprocessor:
    def __init__(self, data: pd.DataFrame):
        
        self.data = data
        logging.info("DataPreprocessor initialized with data of shape: %s", data.shape)

    def clean_data(self) -> pd.DataFrame:
        """
        Performs data cleaning by removing unnecessary columns, dropping columns with missing values,
        and returning the cleaned DataFrame.

        Returns:
            pd.DataFrame: The cleaned DataFrame with unnecessary and missing-value columns removed.
        """
        logging.info("Starting data cleaning process.")

        # Drop unnecessary columns, including '_id' if it exists
        columns_to_drop = [
            'UNIT', 'TYPE', 'MARKET', 'INSTRUMENT', 
            'FIRST_MESSAGE_TIMESTAMP', 'LAST_MESSAGE_TIMESTAMP', 
            'FIRST_MESSAGE_VALUE', 'HIGH_MESSAGE_VALUE', 'HIGH_MESSAGE_TIMESTAMP', 
            'LOW_MESSAGE_VALUE', 'LOW_MESSAGE_TIMESTAMP', 'LAST_MESSAGE_VALUE', 
            'TOTAL_INDEX_UPDATES', 'VOLUME_TOP_TIER', 'QUOTE_VOLUME_TOP_TIER', 
            'VOLUME_DIRECT', 'QUOTE_VOLUME_DIRECT', 'VOLUME_TOP_TIER_DIRECT', 
            'QUOTE_VOLUME_TOP_TIER_DIRECT', '_id'  # Adding '_id' to the list
        ]
        logging.info("Dropping columns: %s")
        self.data = self.drop_columns(self.data, columns_to_drop)

        # Drop columns where the number of missing values is greater than 0
        logging.info("Dropping columns with missing values.")
        self.data = self.drop_columns_with_missing_values(self.data)

        logging.info("Data cleaning completed. Data shape after cleaning: %s", self.data.shape)
        return self.data

    def drop_columns(self, data: pd.DataFrame, columns: list) -> pd.DataFrame:
        """
        Drops specified columns from the DataFrame.

        Returns:
            pd.DataFrame: The DataFrame with the specified columns removed.
        """
        logging.info("Dropping columns: %s", columns)
        return data.drop(columns=columns, errors="ignore")

    def drop_columns_with_missing_values(self, data: pd.DataFrame) -> pd.DataFrame:
        """
        Drops columns with any missing values from the DataFrame.

        Parameters:
            data: pd.DataFrame
                The DataFrame from which columns with missing values will be removed.
        
        Returns:
            pd.DataFrame: The DataFrame with columns containing missing values removed.
        """
        missing_columns = data.columns[data.isnull().sum() > 0]
        if not missing_columns.empty:
            logging.info("Columns with missing values: %s", missing_columns.tolist())
        else:
            logging.info("No columns with missing values found.")
        return data.loc[:, data.isnull().sum() == 0]

Step 6: Feature engineering

This step takes the cleaned data from the previous_data_cleaning step. We developed new features such as Simple Moving Average (SMA), Exponential Moving Average (EMA), and reactive and convolutional statistics to capture trends, reduce noise, and make more reliable predictions from time series data. Additionally, we scale features and target variables using Minmax scaling.

import joblib
import pandas as pd
from abc import ABC, abstractmethod
from sklearn.preprocessing import MinMaxScaler


# Abstract class for Feature Engineering strategy
class FeatureEngineeringStrategy(ABC):
    @abstractmethod
    def generate_features(self, df: pd.DataFrame) -> pd.DataFrame:
        pass


# Concrete class for calculating SMA, EMA, RSI, and other features
class TechnicalIndicators(FeatureEngineeringStrategy):
    def generate_features(self, df: pd.DataFrame) -> pd.DataFrame:
        
        # Calculate SMA, EMA, and RSI
        df['SMA_20'] = df['CLOSE'].rolling(window=20).mean()
        df['SMA_50'] = df['CLOSE'].rolling(window=50).mean()
        df['EMA_20'] = df['CLOSE'].ewm(span=20, adjust=False).mean()
        
        # Price difference features
        df['OPEN_CLOSE_diff'] = df['OPEN'] - df['CLOSE']
        df['HIGH_LOW_diff'] = df['HIGH'] - df['LOW']
        df['HIGH_OPEN_diff'] = df['HIGH'] - df['OPEN']
        df['CLOSE_LOW_diff'] = df['CLOSE'] - df['LOW']

        # Lagged features
        df['OPEN_lag1'] = df['OPEN'].shift(1)
        df['CLOSE_lag1'] = df['CLOSE'].shift(1)
        df['HIGH_lag1'] = df['HIGH'].shift(1)
        df['LOW_lag1'] = df['LOW'].shift(1)

        # Rolling statistics
        df['CLOSE_roll_mean_14'] = df['CLOSE'].rolling(window=14).mean()
        df['CLOSE_roll_std_14'] = df['CLOSE'].rolling(window=14).std()

        # Drop rows with missing values (due to rolling windows, shifts)
        df.dropna(inplace=True)

        return df
        
# Abstract class for Scaling strategy
class ScalingStrategy(ABC):
    @abstractmethod
    def scale(self, df: pd.DataFrame, features: list, target: str):
        pass

# Concrete class for MinMax Scaling
class MinMaxScaling(ScalingStrategy):
    def scale(self, df: pd.DataFrame, features: list, target: str):
        """
        Scales the features and target using MinMaxScaler.

        Parameters:
            df: pd.DataFrame
                The DataFrame containing the features and target.
            features: list
                List of feature column names.
            target: str
                The target column name.

        Returns:
            pd.DataFrame, pd.DataFrame: Scaled features and target
        """
        scaler_X = MinMaxScaler(feature_range=(0, 1))
        scaler_y = MinMaxScaler(feature_range=(0, 1))

        X_scaled = scaler_X.fit_transform(df[features].values)
        y_scaled = scaler_y.fit_transform(df[[target]].values)

        joblib.dump(scaler_X, 'saved_scalers/scaler_X.pkl')
        joblib.dump(scaler_y, 'saved_scalers/scaler_y.pkl')

        return X_scaled, y_scaled, scaler_y


# FeatureEngineeringContext: This will use the Strategy Pattern
class FeatureEngineering:
    def __init__(self, feature_strategy: FeatureEngineeringStrategy, scaling_strategy: ScalingStrategy):
        self.feature_strategy = feature_strategy
        self.scaling_strategy = scaling_strategy

    def process_features(self, df: pd.DataFrame, features: list, target: str):
        # Generate features using the provided strategy
        df_with_features = self.feature_strategy.generate_features(df)

        # Scale features and target using the provided strategy
        X_scaled, y_scaled, scaler_y = self.scaling_strategy.scale(df_with_features, features, target)

        return df_with_features, X_scaled, y_scaled, scaler_y

Step 7: Data Classification

Now, we divide the processed data into training and testing datasets with a ratio of 80:20.

import logging
from abc import ABC, abstractmethod
import numpy as np
from sklearn.model_selection import train_test_split

# Set up logging configuration
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

# Abstract Base Class for Data Splitting Strategy
class DataSplittingStrategy(ABC):
    @abstractmethod
    def split_data(self, X: np.ndarray, y: np.ndarray):
        pass

# Concrete Strategy for Simple Train-Test Split
class SimpleTrainTestSplitStrategy(DataSplittingStrategy):
    def __init__(self, test_size=0.2, random_state=42):
    
        self.test_size = test_size
        self.random_state = random_state

    def split_data(self, X: np.ndarray, y: np.ndarray):
    
        logging.info("Performing simple train-test split.")
        X_train, X_test, y_train, y_test = train_test_split(
            X, y, test_size=self.test_size, random_state=self.random_state
        )
        logging.info("Train-test split completed.")
        return X_train, X_test, y_train, y_test

# Context Class for Data Splitting
class DataSplitter:
    def __init__(self, strategy: DataSplittingStrategy):
        
        self._strategy = strategy

    def set_strategy(self, strategy: DataSplittingStrategy):
        
        logging.info("Switching data splitting strategy.")
        self._strategy = strategy

    def split(self, X: np.ndarray, y: np.ndarray):
        
        logging.info("Splitting data using the selected strategy.")
        return self._strategy.split_data(X, y)

Step 8: Model Training

In this step, we train the LSTM model by stopping early to prevent overfitting, and by using MLflow's automatic logging to track our model and test and save the trained model as lstm_model.keras.

import numpy as np
import logging
import mlflow
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Input, LSTM, Dropout, Dense
from tensorflow.keras.regularizers import l2
from tensorflow.keras.callbacks import EarlyStopping
from typing import Any

# Abstract Base Class for Model Building Strategy
class ModelBuildingStrategy:
    
    def build_and_train_model(self, X_train: np.ndarray, y_train: np.ndarray, fine_tuning: bool = False) -> Any:
        pass

# Concrete Strategy for LSTM Model
class LSTMModelStrategy(ModelBuildingStrategy):
    def build_and_train_model(self, X_train: np.ndarray, y_train: np.ndarray, fine_tuning: bool = False) -> Any:
        """
        Trains an LSTM model on the provided training data.

        Parameters:
            X_train (np.ndarray): The training data features.
            y_train (np.ndarray): The training data labels/target.
            fine_tuning (bool): Not applicable for LSTM, defaults to False.

        Returns:
            tf.keras.Model: A trained LSTM model.
        """
        logging.info("Building and training the LSTM model.")

        # MLflow autologging
        mlflow.tensorflow.autolog()

        logging.info(f"shape of X_train:{X_train.shape}")

        # LSTM Model Definition
        model = Sequential()
        model.add(Input(shape=(X_train.shape[1], X_train.shape[2])))

        model.add(LSTM(units=50, return_sequences=True, kernel_regularizer=l2(0.01)))
        model.add(Dropout(0.3))
        model.add(LSTM(units=50, return_sequences=False))
        model.add(Dropout(0.2))
        model.add(Dense(units=1))  # Adjust the number of units based on your output (e.g., regression or classification)

        # Compiling the model
        model.compile(optimizer="adam", loss="mean_squared_error")

        # Early stopping to avoid overfitting
        early_stopping = EarlyStopping(monitor="val_loss", patience=10, restore_best_weights=True)

        # Fit the model
        history = model.fit(
            X_train,
            y_train,
            epochs=50,
            batch_size=32,
            validation_split=0.1,
            callbacks=[early_stopping],
            verbose=1
        )

        mlflow.log_metric("final_loss", history.history["loss"][-1])

        # Saving the trained model
        model.save("saved_models/lstm_model.keras")
        logging.info("LSTM model trained and saved.")

        return model

# Context Class for Model Building Strategy
class ModelBuilder:
    def __init__(self, strategy: ModelBuildingStrategy):
        self._strategy = strategy

    def set_strategy(self, strategy: ModelBuildingStrategy):
        self._strategy = strategy

    def train(self, X_train: np.ndarray, y_train: np.ndarray, fine_tuning: bool = False) -> Any:
        return self._strategy.build_and_train_model(X_train, y_train, fine_tuning)

Step 9: Testing the Model

Since this is a regression problem, we use test metrics such as Mean Squared Error (MSE), Root Mean Squared Error (MSE), Mean Absolute Error (MAE), and R-squared.

import logging
import numpy as np
from abc import ABC, abstractmethod
from sklearn.preprocessing import MinMaxScaler
from sklearn.metrics import mean_squared_error, mean_absolute_error, r2_score
from typing import Dict

# Setup logging configuration
logging.basicConfig(level=logging.INFO, format="%(asctime)s - %(levelname)s - %(message)s")

# Abstract Base Class for Model Evaluation Strategy
class ModelEvaluationStrategy(ABC):
    @abstractmethod
    def evaluate_model(self, model, X_test, y_test, scaler_y) -> Dict[str, float]:
        pass

# Concrete Strategy for Regression Model Evaluation
class RegressionModelEvaluationStrategy(ModelEvaluationStrategy):
    def evaluate_model(self, model, X_test, y_test, scaler_y) -> Dict[str, float]:
        # Predict the data
        y_pred = model.predict(X_test)

        # Ensure y_test and y_pred are reshaped into 2D arrays for inverse transformation
        y_test_reshaped = y_test.reshape(-1, 1)
        y_pred_reshaped = y_pred.reshape(-1, 1)

        # Inverse transform the scaled predictions and true values
        y_pred_rescaled = scaler_y.inverse_transform(y_pred_reshaped)
        y_test_rescaled = scaler_y.inverse_transform(y_test_reshaped)

        # Flatten the arrays to ensure they are 1D
        y_pred_rescaled = y_pred_rescaled.flatten()
        y_test_rescaled = y_test_rescaled.flatten()

        # Calculate evaluation metrics
        mse = mean_squared_error(y_test_rescaled, y_pred_rescaled)
        rmse = np.sqrt(mse)
        mae = mean_absolute_error(y_test_rescaled, y_pred_rescaled)
        r2 = r2_score(y_test_rescaled, y_pred_rescaled)

        # Logging the metrics
        logging.info("Calculating evaluation metrics.")
        metrics = {
            "Mean Squared Error - MSE": mse,
            "Root Mean Squared Error - RMSE": rmse,
            "Mean Absolute Error - MAE": mae,
            "R-squared - R²": r2
        }

        logging.info(f"Model Evaluation Metrics: {metrics}")
        return metrics

# Context Class for Model Evaluation
class ModelEvaluator:
    def __init__(self, strategy: ModelEvaluationStrategy):
        
        self._strategy = strategy

    def set_strategy(self, strategy: ModelEvaluationStrategy):
        
        logging.info("Switching model evaluation strategy.")
        self._strategy = strategy

    def evaluate(self, model, X_test, y_test, scaler_y) -> Dict[str, float]:
        
        logging.info("Evaluating the model using the selected strategy.")
        return self._strategy.evaluate_model(model, X_test, y_test, scaler_y)  

Now we will organize all the above steps into a pipeline. Let's create a new file training_pipeline.py.

from zenml import Model, pipeline

@pipeline(
    model=Model(
        # The name uniquely identifies this model
        name="bitcoin_price_predictor"
    ),
)
def ml_pipeline():
    # Data Ingestion Step
    raw_data = ingest_data()  
    # Data Cleaning Step
    cleaned_data = clean_data(raw_data)
    # Feature Engineering Step
    transformed_data, X_scaled, y_scaled, scaler_y = feature_engineering_step(
        cleaned_data
    )
    # Data Splitting 
    X_train, X_test, y_train, y_test = data_splitter_step(X_scaled=X_scaled, y_scaled=y_scaled)
    # Model Training
    model = model_training_step(X_train, y_train)
    # Model Evaluation
    evaluator = model_evaluation_step(model, X_test=X_test, y_test=y_test, scaler_y= scaler_y)

    return evaluator

Here, @pipeline decorator is used to define the ml_pipeline() function as a pipeline in ZenML.

To view the training pipeline dashboard, simply run the run_pipeline.py script. Let's create a run_pipeline.py file.

import click
from pipelines.training_pipeline import ml_pipeline

@click.command()
def main():
    run = ml_pipeline()

if __name__=="__main__":
    main()

Now we have finished building the pipeline. Run the command below to view the pipelines dashboard.

python run_pipeline.py

After running the above command it will return the tracking dashboard URL, which looks like this.

The training pipeline looks like this on the dashboard, given below:

Step 10: Model Deployment

So far we have built the model and pipes. Now let's push the pipeline to production where users can make predictions.

Continuous Delivery Pipeline

from zenml.integrations.mlflow.steps import mlflow_model_deployer_step

@pipeline
def continuous_deployment_pipeline():
    trained_model = ml_pipeline()
    mlflow_model_deployer_step(workers=3,deploy_decision=True,model=trained_model,)

This pipeline is responsible for continuously releasing trained models. It starts running the ml_pipeline() from the train_pipeline.py file to train the model, then use the Mlflow Model Deployer feeding the trained model using the continuous_pipeline().

The Inference Pipeline

We use an inference pipeline to make predictions on new data, using an applied model. Let's see how to use this pipeline in our project.

@pipeline
def inference_pipeline(enable_cache=True):
    """Run a batch inference job with data loaded from an API."""
    batch_data = dynamic_importer()
    model_deployment_service = prediction_service_loader(
        pipeline_name="continuous_deployment_pipeline",
        step_name="mlflow_model_deployer_step",
    )
    predictor(service=model_deployment_service, input_data=batch_data)

Let's see about each of the functions called indexing pipeline below:

dynamic_importer()

This function loads new data, performs data processing, and returns data.

@step
def dynamic_importer() -> str:
    """Dynamically imports data for testing the model with expected columns."""

    try:
        data = {
            'OPEN': [0.98712925, 1.],'HIGH': [0.57191823, 0.55107652],'LOW': [1., 0.94728144],'VOLUME': [0.18186191, 0.],'SMA_20': [0.90819243, 1.],'SMA_50': [0.90214911, 1.],'EMA_20': [0.89735654, 1.],'OPEN_CLOSE_diff': [0.61751032, 0.57706902],'HIGH_LOW_diff': [0.01406254, 0.02980481],
            'HIGH_OPEN_diff': [0.13382262, 0.09172282],
            'CLOSE_LOW_diff': [0.14140073, 0.28523136],'OPEN_lag1': [0.64467168, 1.],
            'CLOSE_lag1': [0.98712925, 1.],
            'HIGH_lag1': [0.77019885, 0.57191823],
            'LOW_lag1': [0.64465093, 1.],
            'CLOSE_roll_mean_14': [0.94042809, 1.],'CLOSE_roll_std_14': [0.22060724, 0.35396897],
        }

        df = pd.DataFrame(data)
        data_array = df.iloc[0].values
        reshaped_data = data_array.reshape((1, 1, data_array.shape[0]))  # Single sample, 1 time step, 17 features

        logging.info(f"Reshaped Data: {reshaped_data.shape}")

        json_data = pd.DataFrame(reshaped_data.reshape((reshaped_data.shape[0], reshaped_data.shape[2]))).to_json(orient="split")

        return json_data

    except Exception as e:
        logging.error(f"Error during importing data from dynamic importer: {e}")
        raise e

predict_service_loader()

This work is decorated with it @step. We load the application service along with the applied model based on pipeline_name, and step_name, when our applied model is ready to process new data prediction queries.

The row where_services=mlflow_model_deployer_component.find_model_server() searches for an available service to use based on the given parameters such as pipe name and pipe step name. If no services are available, it indicates that the deployment pipeline was not created or encountered a problem with the deployment pipeline, so it throws a RuntimeError.

@step(enable_cache=False)
def prediction_service_loader(pipeline_name: str, step_name: str) -> MLFlowDeploymentService:
    model_deployer = MLFlowModelDeployer.get_active_model_deployer()

    existing_services = model_deployer.find_model_server(
        pipeline_name=pipeline_name,
        pipeline_step_name=step_name,
    )

    if not existing_services:
        raise RuntimeError(
            f"No MLflow prediction service deployed by the "
            f"{step_name} step in the {pipeline_name} "
            f"pipeline is currently "
            f"running."
        )
    return existing_services[0]

predict()

The function takes the model supplied by MLFlow via the MLFlowDeploymentService and the new data. The data is continuously processed to match the model's expected format for real-time guidance.

@step(enable_cache=False)
def predictor(
    service: MLFlowDeploymentService,
    input_data: str,
) -> np.ndarray:

    service.start(timeout=10)

    try:
        data = json.loads(input_data)
        data.pop("columns", None)
        data.pop("index", None)

        if isinstance(data["data"], list):
            data_array = np.array(data["data"])
        else:
            raise ValueError("The data format is incorrect, expected a list under 'data'.")

        if data_array.shape != (1, 1, 17):
            data_array = data_array.reshape((1, 1, 17))  # Adjust the shape as needed

        try:
            prediction = service.predict(data_array)
        except Exception as e:
            raise ValueError(f"Prediction failed: {e}")
        return prediction
    
    except json.JSONDecodeError:
        raise ValueError("Invalid JSON format in the input data.")
    except KeyError as e:
        raise ValueError(f"Missing expected key in input data: {e}")
    except Exception as e:
        raise ValueError(f"An error occurred during data processing: {e}")

To visualize the continuous deployment and targeting pipeline, we need to use the run_deployment.py script, where configuration and prediction will be defined. (Please check the run_deployment.py code on GitHub given below).

@click.option(
    "--config",
    type=click.Choice([DEPLOY, PREDICT, DEPLOY_AND_PREDICT]),
    default=DEPLOY_AND_PREDICT,
    help="Optionally you can choose to only run the deployment "
    "pipeline to train and deploy a model (`deploy`), or to "
    "only run a prediction against the deployed model "
    "(`predict`). By default both will be run "
    "(`deploy_and_predict`).",
)

Now let's use the run_deployment.py file to see the dashboard of the continuous deployment pipeline and the reference pipeline.

python run_deployment.py

Continuous Delivery Pipeline – Output

Inference Pipeline – Output

After running the run_deployment.py file you can see a link to the MLflow dashboard that looks like this.

mlflow ui --backend-store-uri file:/root/.config/zenml/local_stores/cd1eb06a-179a-4f83-9bae-9b9a5b1bd27f/mlruns

Now you need to copy and paste the above MLflow UI link into your command line and run it.

Here is the MLflow dashboard, where you can see the test metrics and model parameters:

Step 11: Creating a Streaming App

Streamlit is an amazing open source, Python-based framework, used to create interactive UIs, we can use Streamlit to build web applications quickly, without knowing backend or frontend development. First, we need to install Streamlit on our system.

#Install streamlit in our local PC
pip install streamlit

#To run the streamlit local web server
streamlit run app.py

Also, you can find the code on GitHub for the Streamlit app.

Here is GitHub Code and Project Video Description for better understanding.

The conclusion

In this article, we have successfully built an end-to-end, production-ready Bitcoin Price Prediction MLOps project. From getting data through an API and preprocessing it to model training, testing, and deployment, our project highlights the important role of MLOps in connecting development and production. We are close to shaping the future of real-time Bitcoin price prediction. APIs provide seamless access to external data, such as Bitcoin price data from the CCData API, eliminating the need for pre-existing datasets.

Key Takeaways

• APIs enable seamless access to external data, such as Bitcoin price data from the CCata API, eliminating the need for existing datasets.
• ZenML and MLflow are powerful tools that facilitate the development, tracking, and implementation of machine learning models in real-world applications.
• We followed best practices for optimizing data import, cleaning, feature engineering, model training, and testing.
• Continuous distribution and targeting pipelines are essential to ensure that models are always functional and available in production environments.

The media shown in this article does not belong to Analytics Vidhya and is used at the discretion of the Author.

Frequently Asked Questions