Real-Time Crypto Anomaly Detection

Introduction

30 Second Summary

Cryptocurrency markets never sleep. Prices can spike or crash in seconds, and traders, exchanges, and regulators all need to know the instant something unusual happens. Detecting these anomalies in real-time is the exact problem that fintech teams solve with streaming ML pipelines.

In this project, you will build a production-grade real-time anomaly detection system that streams live Bitcoin trades, scores every trade with a trained machine learning model, and alerts you when the market does something unexpected.

What You'll Build

You can open a single Grafana dashboard and watch live BTC-USD trades flowing through your pipeline, with anomaly spikes lighting up the moment they happen, your registered model serving predictions, and drift metrics confirming whether the model's assumptions still hold.

By the end of this project, you'll have:

• A live streaming pipeline where you can watch real Bitcoin trades arrive from Coinbase's WebSocket, flow through Kafka, get scored by an Isolation Forest model, and see flagged anomalies appear in Elasticsearch.
• A full MLOps workflow where you train a model, log experiments in MLflow, register the best version, and load it by name for live inference.
• A drift monitoring system powered by Evidently that detects when live market data shifts away from your training distribution, with results feeding a Grafana dashboard showing throughput, anomaly rates, and drift status.
• Secret Mission: Embed anomaly patterns as vectors in Qdrant and query for historically similar anomalies to enrich each new detection with context.

Are there any prerequisites?

This project assumes you have experience writing Python and basic familiarity with scikit-learn for machine learning. You should also be comfortable with the basics of Docker (pulling images, running containers).

You will need Python 3.14 and Docker installed on your Mac. Every other service is launched from scratch via Docker Compose.

Before We Start

Before diving into containers and streaming data, take a moment to lock in what you're about to build and why it matters to you.

Launch the Full Infrastructure Stack

Your Python environment is ready. Now you need the infrastructure that your ML pipeline will run on.

A real-time anomaly detection system needs more than just Python code. It needs a message broker, a model registry, metrics collection, dashboards, and data stores all running simultaneously. Docker Compose lets you declare all 13 services in a single YAML file and launch them with one command.

In this step, get ready to:

• Create the Docker Compose file defining all infrastructure services.
• Create configuration files for Prometheus and Grafana.
• Launch the full stack and confirm every service is healthy.

Create the Docker Compose file

The docker-compose.yml file declares every service your pipeline depends on. You will build it in sections, starting with the Kafka event streaming core.

• Create a new file called docker-compose.yml in your project folder and add the Kafka services:

version: "3.8"

services:
  zookeeper:
    image: confluentinc/cp-zookeeper:7.9.0
    container_name: zookeeper
    environment:
      ZOOKEEPER_CLIENT_PORT: 2181
      ZOOKEEPER_TICK_TIME: 2000
    ports:
      - "2181:2181"

  kafka:
    image: confluentinc/cp-kafka:7.9.0
    container_name: kafka
    depends_on:
      - zookeeper
    ports:
      - "9092:9092"
    environment:
      KAFKA_BROKER_ID: 1
      KAFKA_ZOOKEEPER_CONNECT: zookeeper:2181
      KAFKA_ADVERTISED_LISTENERS: PLAINTEXT://kafka:29092,PLAINTEXT_HOST://localhost:9092
      KAFKA_LISTENER_SECURITY_PROTOCOL_MAP: PLAINTEXT:PLAINTEXT,PLAINTEXT_HOST:PLAINTEXT
      KAFKA_INTER_BROKER_LISTENER_NAME: PLAINTEXT
      KAFKA_OFFSETS_TOPIC_REPLICATION_FACTOR: 1

What does this code do?

Zookeeper manages Kafka cluster metadata and coordinates broker leadership. Kafka is the event streaming platform that will carry trade messages between your producer and consumer.

The two listener addresses let Kafka communicate with other containers on kafka:29092 and with your local Python scripts on localhost:9092.

• Below the kafka service, add the Schema Registry and Kafka UI services:

  schema-registry:
    image: confluentinc/cp-schema-registry:7.9.0
    container_name: schema-registry
    depends_on:
      - kafka
    ports:
      - "8081:8081"
    environment:
      SCHEMA_REGISTRY_HOST_NAME: schema-registry
      SCHEMA_REGISTRY_KAFKASTORE_BOOTSTRAP_SERVERS: kafka:29092

  kafka-ui:
    image: provectuslabs/kafka-ui:latest
    container_name: kafka-ui
    depends_on:
      - kafka
      - schema-registry
    ports:
      - "9091:8080"
    environment:
      KAFKA_CLUSTERS_0_NAME: local
      KAFKA_CLUSTERS_0_BOOTSTRAPSERVERS: kafka:29092
      KAFKA_CLUSTERS_0_SCHEMAREGISTRY: http://schema-registry:8081

What does this code do?

Schema Registry validates the format of messages flowing through Kafka topics. Kafka UI gives you a browser-based dashboard at port 9091 to inspect topics, messages, and consumer groups visually.

• Save docker-compose.yml.
• Verify the YAML is valid so far by running:

docker compose config --services

You should see four service names printed: kafka, kafka-ui, schema-registry, and zookeeper.

Seeing a YAML error?

YAML is sensitive to indentation. Every service must be indented exactly two spaces under services:. Each property under a service gets four spaces total. Check that you have no tabs mixed with spaces.

Help me fix a YAML indentation error in my docker-compose.yml

• Next, add the data store services below kafka-ui. These store ML metadata and anomaly records:

  postgres:
    image: postgres:15
    container_name: postgres
    environment:
      POSTGRES_USER: dex
      POSTGRES_PASSWORD: dex
      POSTGRES_DB: dex
    ports:
      - "5432:5432"
    volumes:
      - pgdata:/var/lib/postgresql/data
    healthcheck:
      test: ["CMD-SHELL", "pg_isready -U dex"]
      interval: 5s
      timeout: 3s
      retries: 10

  elasticsearch:
    image: docker.elastic.co/elasticsearch/elasticsearch:8.17.0
    container_name: elasticsearch
    environment:
      - discovery.type=single-node
      - xpack.security.enabled=false
      - "ES_JAVA_OPTS=-Xms512m -Xmx512m"
    ports:
      - "9200:9200"
    volumes:
      - esdata:/usr/share/elasticsearch/data

What does this code do?

PostgreSQL stores MLflow experiment metadata (parameters, metrics, model versions). The healthcheck ensures other services that depend on Postgres wait until it is ready to accept connections.

Elasticsearch stores detected anomaly documents so you can search and analyse them later. Security is disabled for local development simplicity.

• Add the observability services (metrics, dashboards, alerting, tracing) below elasticsearch:

  prometheus:
    image: prom/prometheus:v3.4.1
    container_name: prometheus
    ports:
      - "9090:9090"
    volumes:
      - ./config/prometheus.yml:/etc/prometheus/prometheus.yml

  grafana:
    image: grafana/grafana:11.6.0
    container_name: grafana
    ports:
      - "3000:3000"
    environment:
      GF_SECURITY_ADMIN_USER: admin
      GF_SECURITY_ADMIN_PASSWORD: admin
    volumes:
      - ./config/grafana/datasources.yml:/etc/grafana/provisioning/datasources/datasources.yml

  alertmanager:
    image: prom/alertmanager:v0.28.1
    container_name: alertmanager
    ports:
      - "9093:9093"

  jaeger:
    image: jaegertracing/all-in-one:1.68
    container_name: jaeger
    ports:
      - "16686:16686"
      - "6831:6831/udp"

What does this code do?

Prometheus scrapes metrics from your pipeline consumer every 15 seconds. Grafana visualises those metrics in dashboards. Alertmanager handles alert routing, and Jaeger provides distributed tracing so you can track requests across services.

• Save docker-compose.yml.
• Confirm the file now defines 8 services by running:

docker compose config --services | wc -l

You should see 8 printed.

• Add container monitoring, the ML platform, and vector database services below jaeger:

  cadvisor:
    image: gcr.io/cadvisor/cadvisor:v0.51.0
    container_name: cadvisor
    ports:
      - "8080:8080"
    volumes:
      - /var/run/docker.sock:/var/run/docker.sock:ro
      - /sys:/sys:ro
      - /var/lib/docker/:/var/lib/docker:ro

  mlflow:
    image: ghcr.io/mlflow/mlflow:v2.21.3
    container_name: mlflow
    depends_on:
      postgres:
        condition: service_healthy
    ports:
      - "5000:5000"
    environment:
      MLFLOW_BACKEND_STORE_URI: postgresql://dex:dex@postgres:5432/dex
    command: >
      bash -c "pip install --no-cache-dir psycopg2-binary &&
      mlflow server --backend-store-uri postgresql://dex:dex@postgres:5432/dex --host 0.0.0.0 --port 5000"

What does this code do?

cAdvisor monitors resource usage (CPU, memory, network) for every running container. MLflow provides experiment tracking and a model registry. It waits for Postgres to be healthy before starting, then launches its server with the PostgreSQL backend store.

• Finally, add the vector database and the named volumes section at the bottom of the file:

  qdrant:
    image: qdrant/qdrant:v1.14.0
    container_name: qdrant
    ports:
      - "6333:6333"
      - "6334:6334"
    volumes:
      - qdrantdata:/qdrant/storage

volumes:
  pgdata:
  esdata:
  qdrantdata:

What does this code do?

Qdrant is a vector database you will use in the Secret Mission for similarity search over anomaly patterns. The volumes section at the bottom creates named Docker volumes so Postgres, Elasticsearch, and Qdrant persist their data across container restarts.

• Save docker-compose.yml.
• Verify all 13 services are declared by running:

docker compose config --services | wc -l

You should see 13 printed. Every service in your stack is now declared.

Count is wrong or seeing an error?

If the count is less than 13, scroll through docker-compose.yml and check that every service block (from zookeeper to qdrant) is present. Make sure the volumes: section is at the root level (no indentation), not nested under services:.

Help me debug my docker-compose.yml service count

✔️ Awesome, I've got everything!

Great. Double check you have saved docker-compose.yml and that the count shows 13.

ⓧ I'd like to double check the full code

Scroll up through the six code blocks you added. Confirm they appear in this order inside your file:

• version + zookeeper + kafka (the file header and Kafka core).
• schema-registry + kafka-ui (Kafka ecosystem).
• postgres + elasticsearch (data stores).
• prometheus + grafana + alertmanager + jaeger (observability).
• cadvisor + mlflow (container metrics and ML platform).
• qdrant + volumes (vector database and named volumes).

Create the configuration files

The Prometheus and Grafana containers mount local config files from a config/ directory. You need to create these before launching the stack.

• Create the config directory structure by running:

mkdir -p config/grafana

• Confirm the directories exist by running:

ls config

You should see grafana listed.

• Create config/prometheus.yml with the scrape configuration:

global:
  scrape_interval: 15s

scrape_configs:
  - job_name: "prometheus"
    static_configs:
      - targets: ["localhost:9090"]

  - job_name: "cadvisor"
    static_configs:
      - targets: ["cadvisor:8080"]

  - job_name: "anomaly-consumer"
    static_configs:
      - targets: ["host.docker.internal:8000"]

What does this code do?

This tells Prometheus what to scrape every 15 seconds. It collects its own metrics, container metrics from cAdvisor, and your Python consumer's custom metrics via host.docker.internal:8000 (which lets containers reach ports on your host machine).

• Create config/grafana/datasources.yml to auto-configure Grafana's data source:

apiVersion: 1

datasources:
  - name: Prometheus
    type: prometheus
    access: proxy
    url: http://prometheus:9090
    isDefault: true

What does this code do?

This provisioning file tells Grafana to automatically connect to Prometheus on startup. Without it, you would need to manually add the data source through the Grafana UI every time the container restarts.

• Save both config files.
• Verify all config files are in place by running:

ls config/prometheus.yml config/grafana/datasources.yml

Both file paths should be printed without errors.

File not found?

Make sure you created the nested config/grafana/ directory with mkdir -p config/grafana before creating the datasources file. Check that the file is named datasources.yml (not datasource.yml).

Help me find my missing config files

Launch and verify all services

Everything is configured. Time to bring the entire stack online with a single command.

• Launch all 13 services in detached mode by running:

docker compose up -d

What does this command do?

The -d flag runs containers in detached mode so they operate in the background and return your terminal. Docker will pull all the required images on the first run, which may take a few minutes depending on your internet speed.

First run takes time

Docker needs to download approximately 4-5 GB of container images on the first launch. Subsequent launches will be near-instant because images are cached locally. If a pull fails, just run docker compose up -d again.

Before you check the service statuses, what do you expect to see? Do you think all 13 containers will show as running immediately, or might some take longer to start?

• Check the status of all containers by running:

docker compose ps

You should see all 13 containers listed. Most will show a status of Up or running. The MLflow container may show as starting briefly while it waits for Postgres and installs psycopg2.

Some containers exited or restarting?

If a container shows Exited or Restarting, check its logs to find the error. Replace SERVICE with the container name (e.g. mlflow, elasticsearch):

docker compose logs SERVICE

Common causes: Elasticsearch may fail if Docker has less than 4 GB of memory allocated. MLflow may restart once while Postgres finishes its healthcheck. Wait 30 seconds and run docker compose ps again.

Help me debug containers that are not starting

• Open Kafka UI in your browser to confirm the streaming layer is live:

open http://localhost:9091

You should see the Kafka UI dashboard showing one cluster named local with one broker online.

• Open Grafana in your browser to confirm the dashboarding layer is live:

open http://localhost:3000

You should see the Grafana login page. Sign in with username admin and password admin. Grafana will ask you to change the password. You can skip this for local development.

Why so many services?

This is the same infrastructure pattern used by fintech companies for real-time fraud detection. Each service handles one responsibility: Kafka moves data, Prometheus collects metrics, Grafana visualises them, MLflow tracks models, and Elasticsearch enables search.

Running everything locally with Docker Compose means zero cloud costs and full control over your learning environment.

Your entire infrastructure stack is now running. Next up, you will connect to the Coinbase WebSocket and start streaming live Bitcoin trades into Kafka.

Stream Live Trades into Kafka

Your infrastructure is running, but an ML pipeline without data is like a factory with no raw materials. The anomaly detection model needs a continuous stream of real trades to learn from and score against.

In this step, you'll write a Kafka producer that taps into Coinbase's public WebSocket feed and streams live BTC-USD trades into your cluster. This proves the data ingestion layer works end-to-end with real market data.

In this step, get ready to:

• Write a Kafka producer that streams live cryptocurrency trades from Coinbase.
• Verify real-time BTC-USD trade data flowing through Kafka UI.
• Identify the raw features you'll engineer for anomaly detection.

Install the Python dependencies

The producer script uses two libraries: confluent-kafka for publishing to Kafka, and websocket-client for connecting to Coinbase's WebSocket feed. Your requirements.txt from the previous step already lists them.

• Install all project dependencies by running:

pip install -r requirements.txt

What does this install?

This installs confluent-kafka (a high-performance Kafka client built on the librdkafka C library), websocket-client (for persistent WebSocket connections), plus data science libraries you'll use in later steps.

Seeing build errors during install?

The confluent-kafka package includes pre-built wheels for most platforms, but if your system needs to compile from source, you may need librdkafka installed separately. On macOS, run brew install librdkafka first.

Ask for help debugging your install.

Write the Kafka producer script

The producer connects to Coinbase's free public WebSocket, listens for BTC-USD match events (every executed trade), and publishes each one to a Kafka topic. No API key is required.

• Create a new file called producer.py in your crypto-anomaly-pipeline folder and add the imports and Kafka configuration:

import json
import time
import websocket
from confluent_kafka import Producer

# Kafka broker running in Docker, exposed on localhost:9092
KAFKA_BOOTSTRAP = "localhost:9092"
TOPIC = "crypto.trades.raw"
COINBASE_WS = "wss://ws-feed.exchange.coinbase.com"

# Producer configuration with a client ID for identification
producer_config = {
    "bootstrap.servers": KAFKA_BOOTSTRAP,
    "client.id": "coinbase-producer",
}

producer = Producer(producer_config)


def delivery_report(err, msg):
    if err is not None:
        print(f"Delivery failed: {err}")

What does this code do?

• KAFKA_BOOTSTRAP points to your Kafka broker on localhost:9092 (the port exposed in your Docker Compose).
• TOPIC is the Kafka topic name where all raw trades will land. Kafka auto-creates topics by default.
• delivery_report is a callback that fires after each message is acknowledged by the broker, printing an error only if delivery failed.

• Below the delivery_report function, add the WebSocket message handler that processes each incoming trade:

def on_message(ws, message):
    data = json.loads(message)
    # Only process actual trade matches, ignore other message types
    if data.get("type") == "match":
        trade = {
            "price": float(data["price"]),
            "size": float(data["size"]),
            "side": data["side"],
            "time": data["time"],
            "trade_id": data["trade_id"],
        }
        # Publish trade to Kafka with trade_id as the message key
        producer.produce(
            TOPIC,
            key=str(trade["trade_id"]),
            value=json.dumps(trade).encode("utf-8"),
            callback=delivery_report,
        )
        producer.poll(0)
        print(f"Trade: ${trade['price']:.2f} | Size: {trade['size']:.6f} | Side: {trade['side']}")

What does on_message do?

Coinbase sends several message types over the WebSocket (subscriptions, heartbeats, last_match). The if data.get("type") == "match" filter ensures you only process actual executed trades.

Each trade is published to Kafka with the trade_id as the message key. This ensures trades for the same ID always land on the same partition. The producer.poll(0) call triggers delivery callbacks without blocking.

• Verify your file is saving correctly by running:

python -c "from confluent_kafka import Producer; print('confluent-kafka ready')"

You should see confluent-kafka ready printed to your terminal. This confirms the library is installed and importable.

• Now add the WebSocket lifecycle handlers and the main entry point at the bottom of producer.py:

def on_open(ws):
    # Subscribe to the matches channel for BTC-USD trades
    subscribe_msg = {
        "type": "subscribe",
        "channels": [{"name": "matches", "product_ids": ["BTC-USD"]}],
    }
    ws.send(json.dumps(subscribe_msg))
    print("Subscribed to BTC-USD matches")


def on_error(ws, error):
    print(f"WebSocket error: {error}")


def on_close(ws, close_status_code, close_msg):
    print("WebSocket closed")
    # Flush any remaining messages in the producer buffer
    producer.flush()


if __name__ == "__main__":
    print("Starting Coinbase trade producer...")
    ws = websocket.WebSocketApp(
        COINBASE_WS,
        on_open=on_open,
        on_message=on_message,
        on_error=on_error,
        on_close=on_close,
    )
    ws.run_forever()

What do these handlers do?

• on_open sends a subscription message to Coinbase requesting the matches channel for BTC-USD. This tells Coinbase to start streaming every executed trade.
• on_close calls producer.flush() to make sure any buffered messages are delivered to Kafka before the process exits.
• ws.run_forever() keeps the WebSocket connection open indefinitely, receiving trades until you press Ctrl+C.

• Save producer.py.

✔️ Awesome, I've got everything!

Great. Double check you've saved your file with all three sections in order.

ⓧ I'd like to double check the full code

Your producer.py should contain these three sections in order. Scroll up and confirm each is present:

• Imports, constants (KAFKA_BOOTSTRAP, TOPIC, COINBASE_WS), producer config, Producer instantiation, and delivery_report function.
• The on_message function that filters for "match" type events and publishes trades to Kafka.
• The on_open, on_error, on_close handlers and the if __name__ == "__main__" block with ws.run_forever().

Run the producer and verify live trades

Before you run this, what do you expect to see in the terminal? The producer will connect to Coinbase's WebSocket and start printing every BTC-USD trade as it happens.

• Start the producer by running:

python producer.py

Within a few seconds you should see output like this:

Starting Coinbase trade producer...
Subscribed to BTC-USD matches
Trade: $104325.50 | Size: 0.001200 | Side: sell
Trade: $104326.00 | Size: 0.050000 | Side: buy
Trade: $104324.75 | Size: 0.003400 | Side: sell

What am I seeing?

Each line is a real BTC-USD trade happening right now on the Coinbase exchange. The price, size (amount of Bitcoin traded), and side (whether the maker was buying or selling) update in real-time. BTC-USD typically trades 1-5 times per second, so you'll see a steady stream of output.

Not seeing any trades?

• Make sure your Docker containers are still running. Run docker compose ps and confirm the kafka container shows as healthy.
• If you see a WebSocket error, check that your internet connection allows outbound WebSocket connections on port 443.
• If you see Subscribed to BTC-USD matches but no trades appear, wait 10-15 seconds. During low-volume periods, trades may arrive less frequently.

Help me debug why my Kafka producer isn't receiving trades from Coinbase.

Leave the producer running in this terminal. Open a new browser tab and navigate to Kafka UI at http://localhost:9091.

• Click on Topics in the left sidebar.
• Click on crypto.trades.raw in the topics list.
• Click the Messages tab to see the live trade messages arriving.

You should see JSON messages flowing in with fields like price, size, side, time, and trade_id. Each message represents a real Bitcoin trade that just happened on the exchange.

What features will you use for anomaly detection?

Look at the data flowing through. In the next step, you'll engineer these raw fields into features the ML model can learn from:

• price becomes a rolling deviation from the mean (sudden price jumps stand out).
• size becomes a z-score against recent volume (unusually large trades stand out).
• time becomes inter-trade interval (bursts of rapid trades stand out).
• side becomes a binary encoding (detecting imbalanced buy/sell pressure).

Your data pipeline is live. Real Bitcoin trades are streaming from Coinbase into your Kafka cluster every second. Next up, you'll train an Isolation Forest model on this data so it can learn what "normal" looks like and flag anything unusual.

Train and Register the Anomaly Detection Model

In the previous step, you proved the data ingestion layer works. Live trades are flowing from Coinbase into Kafka in real-time.

Now the real-time inference consumer (which you will build next) needs a trained model to score each trade against. Without a baseline of what "normal" looks like, there is nothing to compare incoming trades to. In this step, you will train an Isolation Forest on a batch of collected trades, then register it in MLflow so the consumer can load it by name.

In this step, get ready to:

• Collect a batch of live trades from Kafka and engineer features for anomaly detection.
• Train an Isolation Forest model and log the experiment to MLflow.
• Register the trained model in MLflow's Model Registry so the inference consumer can load it.

Set up the training script and data collection

The training script connects to Kafka as a consumer, reads a batch of existing trades, engineers features from them, trains the model, and registers the result in MLflow. All of this lives in one file.

• Make sure your producer is still running in its terminal (you should see trade output scrolling). It needs to keep feeding trades into Kafka while you train.
• Open a new terminal tab or window for the training work.
• Create train_model.py in your project folder with the following imports and configuration:

import json
import time
import numpy as np
import pandas as pd
from confluent_kafka import Consumer
from sklearn.ensemble import IsolationForest
import mlflow
import mlflow.sklearn

KAFKA_BOOTSTRAP = "localhost:9092"
TOPIC = "crypto.trades.raw"
MLFLOW_TRACKING_URI = "http://localhost:5000"
COLLECT_SECONDS = 120
MODEL_NAME = "crypto-anomaly-detector"

mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)
mlflow.set_experiment("crypto-anomaly-detection")

What does this code do?

This sets up the script's dependencies and configuration constants.

• COLLECT_SECONDS = 120 tells the script to spend 2 minutes reading trades from Kafka before training.
• mlflow.set_tracking_uri points the MLflow client at your local MLflow server running in Docker.
• mlflow.set_experiment creates a new experiment called crypto-anomaly-detection if one does not already exist.

• Below the configuration, add the function that collects trades from Kafka:

def collect_trades():
    consumer = Consumer({
        "bootstrap.servers": KAFKA_BOOTSTRAP,
        "group.id": "training-collector",
        "auto.offset.reset": "earliest",
    })
    consumer.subscribe([TOPIC])

    trades = []
    start = time.time()
    print(f"Collecting trades for {COLLECT_SECONDS} seconds...")

    while time.time() - start < COLLECT_SECONDS:
        msg = consumer.poll(1.0)
        if msg is None:
            continue
        if msg.error():
            continue
        trade = json.loads(msg.value().decode("utf-8"))
        trades.append(trade)

    consumer.close()
    print(f"Collected {len(trades)} trades")
    return trades

What does this code do?

collect_trades() creates a Kafka consumer that reads from the same topic your producer writes to.

• auto.offset.reset: "earliest" means it starts reading from the very beginning of the topic, so you get all trades your producer has published so far.
• The consumer polls for messages in a loop for 120 seconds, collecting every trade it receives into a list.

• Save train_model.py.
• Confirm the imports resolve correctly by running:

python -c "from train_model import collect_trades; print('Imports OK')"

You should see Imports OK printed to the terminal.

Seeing a ModuleNotFoundError?

• Make sure you are in your project's virtual environment where you installed the packages from requirements.txt.
• If you see an error about confluent_kafka or mlflow, run pip install -r requirements.txt again.

Still stuck? Why can't Python find my installed packages?

Engineer features for anomaly detection

Raw trade data (price, size, side, timestamp) is not directly useful for anomaly detection. You need features that capture relative changes and patterns.

• Add the engineer_features function below collect_trades in train_model.py:

def engineer_features(trades):
    df = pd.DataFrame(trades)
    df["price"] = df["price"].astype(float)
    df["size"] = df["size"].astype(float)
    df["time"] = pd.to_datetime(df["time"])
    df = df.sort_values("time").reset_index(drop=True)

    df["price_rolling_mean"] = df["price"].rolling(window=20, min_periods=1).mean()
    df["price_deviation"] = (df["price"] - df["price_rolling_mean"]) / df["price_rolling_mean"]
    df["volume_rolling_mean"] = df["size"].rolling(window=20, min_periods=1).mean()
    df["volume_zscore"] = (df["size"] - df["volume_rolling_mean"]) / df["size"].rolling(window=20, min_periods=1).std().fillna(1)
    df["time_diff"] = df["time"].diff().dt.total_seconds().fillna(0)
    df["side_encoded"] = (df["side"] == "buy").astype(int)

    feature_cols = ["price_deviation", "volume_zscore", "time_diff", "side_encoded"]
    features = df[feature_cols].fillna(0)
    return features, df

What do these features capture?

• price_deviation: How far the current price is from its 20-trade rolling average. A spike here means a sudden price movement.
• volume_zscore: How unusual the trade size is compared to recent trades. High z-scores signal abnormally large or small trades.
• time_diff: Seconds between consecutive trades. A burst of rapid trades or an unusual gap both show up here.
• side_encoded: Whether the trade was a buy (1) or sell (0).

Together, these four features create a compact representation of each trade's context. The Isolation Forest will learn the normal distribution of these features and flag outliers.

Train, log, and register the model in MLflow

With data collection and feature engineering defined, you can now write the function that trains the Isolation Forest, logs the experiment, and registers the model in one MLflow run.

• Add the first part of the train_and_register function below engineer_features in train_model.py:

def train_and_register(features):
    n_estimators = 100
    contamination = 0.05

    with mlflow.start_run():
        model = IsolationForest(
            n_estimators=n_estimators,
            contamination=contamination,
            random_state=42,
        )
        model.fit(features)

        predictions = model.predict(features)
        anomaly_rate = (predictions == -1).sum() / len(predictions)

        mlflow.log_param("n_estimators", n_estimators)
        mlflow.log_param("contamination", contamination)
        mlflow.log_param("n_training_samples", len(features))
        mlflow.log_metric("training_anomaly_rate", anomaly_rate)

What does this code do?

This is the first half of the training function. It trains the model and logs metrics.

• contamination=0.05 tells the model to expect roughly 5% of trades to be anomalies, which sets the decision threshold for scoring.
• mlflow.start_run() opens a tracked experiment run. Everything inside this block is logged together.
• Parameters and the training anomaly rate are logged so you can track what you trained and how it performed.

• Still inside the with mlflow.start_run(): block (indented at the same level as the lines above), add the model registration and artifact logging:

        mlflow.sklearn.log_model(
            model,
            artifact_path="model",
            registered_model_name=MODEL_NAME,
        )

        features.to_csv("/tmp/training_reference.csv", index=False)
        mlflow.log_artifact("/tmp/training_reference.csv")

        print(f"Model registered: {MODEL_NAME}")
        print(f"Training anomaly rate: {anomaly_rate:.4f}")
        print(f"Training samples: {len(features)}")

What does this code do?

This second half registers the model and saves a reference dataset.

• registered_model_name=MODEL_NAME adds the model to MLflow's Model Registry so the inference consumer can load it by name later.
• The training reference CSV is saved as an artifact. The consumer will use this later to compare live data against the training distribution for drift detection.

• Now add the main entry point at the bottom of train_model.py:

if __name__ == "__main__":
    trades = collect_trades()
    if len(trades) < 50:
        print("Not enough trades collected. Let the producer run longer.")
        exit(1)
    features, df = engineer_features(trades)
    train_and_register(features)
    print("Training complete. Check MLflow at http://localhost:5000")

What does this code do?

The main block orchestrates the full pipeline: collect trades, check you have enough data (at least 50 trades), engineer features, then train and register the model.

If fewer than 50 trades were collected, the script exits early with a helpful message. This avoids training on too little data, which would produce a model that cannot reliably distinguish normal from anomalous.

• Save train_model.py and confirm all functions are defined by running:

python -c "from train_model import train_and_register; print('All functions OK')"

You should see All functions OK with no errors.

Seeing a SyntaxError?

• Check for missing colons at the end of def lines and with statements.
• Make sure the indentation inside with mlflow.start_run(): is consistent (8 spaces for the body of the with block inside the function).

Need help? Debug my train_model.py syntax error

Before you run the training script, what do you expect to happen? Will it finish instantly, or will it take time?

• Run the training script:

python train_model.py

The script prints Collecting trades for 120 seconds... and then sits quietly for 2 minutes while it gathers data. After the collection window closes, you will see output confirming the model was registered.

What should I see when it finishes?

• Collected 150 trades (your number will vary depending on market activity).
• Model registered: crypto-anomaly-detector.
• Training anomaly rate: 0.0500 (approximately, matching the 5% contamination).
• Training complete. Check MLflow at http://localhost:5000.

• Open your browser and go to http://localhost:5000.
• Click on the crypto-anomaly-detection experiment in the left sidebar.
• You should see one run listed with your logged parameters (n_estimators, contamination, n_training_samples) and the training_anomaly_rate metric.
• Click Models in the top navigation to confirm crypto-anomaly-detector appears in the Model Registry with version 1.

Seeing 'Not enough trades collected'?

• Your producer may have stopped. Switch to the producer terminal and check it is still printing trades. Restart it with python producer.py if needed.
• If BTC-USD trading is quiet (weekends or low-volume periods), let the producer run for 3-5 minutes before rerunning python train_model.py.
• Confirm your Kafka container is running with docker compose ps.

Still not collecting trades? Help me debug why my training script is not receiving Kafka messages

✔️ Awesome, I've got everything!

Great. Double check you have saved train_model.py and confirmed your model appears in MLflow.

ⓧ I'd like to double check the full code

Your train_model.py should contain these sections in order. Scroll up and confirm each one is present:

• Imports and configuration (json, time, numpy, pandas, confluent_kafka, sklearn, mlflow) plus constants (KAFKA_BOOTSTRAP, TOPIC, MLFLOW_TRACKING_URI, COLLECT_SECONDS, MODEL_NAME) and the two mlflow.set_* calls.
• collect_trades() function that creates a Kafka consumer and polls for 120 seconds.
• engineer_features() function that computes price_deviation, volume_zscore, time_diff, and side_encoded.
• train_and_register() function that trains the Isolation Forest, logs params/metrics, registers the model, and saves the reference CSV.
• if __name__ == "__main__": block that orchestrates the pipeline with a 50-trade minimum check.

Your model is trained and registered. Next up, you will build the real-time inference consumer that loads this model from MLflow and scores every incoming trade as it arrives.

Build the Real-Time Inference Consumer

Your trained Isolation Forest model is registered in MLflow, ready to score trades. The producer is pumping live data into Kafka. Now you need the piece that connects them: a consumer that reads every trade, runs it through the model, and acts on the result.

In this step you will build that consumer. It loads the model from MLflow, engineers features in real-time (exactly the same features used during training), scores each trade, and writes anomalies into Elasticsearch for later search. It also exposes Prometheus metrics so the observability stack can scrape inference performance.

In this step, get ready to:

• Write a Kafka consumer that loads the registered model and scores every trade in real-time.
• Index detected anomalies into Elasticsearch with full trade context.
• Expose Prometheus metrics and verify anomalies appear in the database.

Set up the consumer's imports and configuration

The consumer connects to multiple services at once: Kafka for messages, MLflow for the model, Elasticsearch for storage, and Prometheus for metrics. All of these need to be configured up front.

• Create a new file called consumer.py in your project folder. Add the imports, configuration constants, and Prometheus metric definitions by pasting this code:

import json
import time
import numpy as np
import pandas as pd
from collections import deque
from confluent_kafka import Consumer
from elasticsearch import Elasticsearch
from prometheus_client import start_http_server, Counter, Histogram, Gauge
import mlflow
import mlflow.sklearn
from evidently.report import Report
from evidently.metric_preset import DataDriftPreset

KAFKA_BOOTSTRAP = "localhost:9092"
TOPIC = "crypto.trades.raw"
MLFLOW_TRACKING_URI = "http://localhost:5000"
MODEL_NAME = "crypto-anomaly-detector"
ES_HOST = "http://localhost:9200"
DRIFT_WINDOW = 100

mlflow.set_tracking_uri(MLFLOW_TRACKING_URI)

trades_processed = Counter("trades_processed_total", "Total trades processed")
anomalies_detected = Counter("anomalies_detected_total", "Total anomalies detected")
inference_latency = Histogram("inference_latency_seconds", "Model inference latency")
drift_detected = Gauge("drift_detected", "Whether data drift is detected (1=yes, 0=no)")

es = Elasticsearch(ES_HOST)
trade_buffer = deque(maxlen=500)
reference_data = None

What does this code do?

• The four Prometheus objects (Counter, Histogram, Gauge) define the metrics that Prometheus will scrape from port 8000.
• trade_buffer is a rolling window of the last 500 trades, used to compute rolling statistics for feature engineering.
• DRIFT_WINDOW = 100 means drift is checked every 100 trades to avoid overhead on every single inference call.
• reference_data will hold the training distribution so Evidently can compare live data against it.

• Save consumer.py.
• Confirm the file exists by running:

ls consumer.py

You should see consumer.py listed in the output.

• Add the functions that load your trained model and its reference dataset from MLflow. Paste this below the existing code in consumer.py:

def load_model():
    client = mlflow.MlflowClient()
    versions = client.get_latest_versions(MODEL_NAME)
    if not versions:
        raise RuntimeError(f"No versions found for model {MODEL_NAME}")
    model_uri = f"models:/{MODEL_NAME}/{versions[0].version}"
    model = mlflow.sklearn.load_model(model_uri)
    print(f"Loaded model: {MODEL_NAME} v{versions[0].version}")
    return model


def load_reference():
    global reference_data
    client = mlflow.MlflowClient()
    versions = client.get_latest_versions(MODEL_NAME)
    run_id = versions[0].run_id
    artifact_path = mlflow.artifacts.download_artifacts(
        run_id=run_id, artifact_path="training_reference.csv"
    )
    reference_data = pd.read_csv(artifact_path)
    print(f"Loaded reference data: {len(reference_data)} samples")

What does this code do?

• load_model() fetches the latest version of crypto-anomaly-detector from MLflow's Model Registry and loads it into memory as a scikit-learn object.
• load_reference() downloads the training_reference.csv artifact you logged during training. This CSV becomes the baseline distribution for drift detection.

• Save consumer.py and verify the file imports cleanly by running:

python -c "import consumer; print('Imports OK')"

You should see Imports OK printed with no errors.

Seeing an ImportError?

• Make sure your virtual environment is activated and you installed the dependencies from requirements.txt in Step 1.
• Confirm all Docker services are still running with docker compose ps. MLflow needs to be reachable for the import to succeed.

Why won't my consumer.py import?

Add feature engineering and drift detection functions

The consumer needs to compute the same four features used during training, but for a single trade at a time using a rolling buffer. It also needs to detect when incoming data drifts from the training distribution.

• Add the single-trade feature engineering function below the existing code in consumer.py:

def engineer_features_single(trade, buffer):
    prices = [t["price"] for t in buffer]
    sizes = [t["size"] for t in buffer]

    price_mean = np.mean(prices[-20:]) if len(prices) >= 20 else np.mean(prices)
    price_deviation = (trade["price"] - price_mean) / price_mean if price_mean > 0 else 0

    vol_mean = np.mean(sizes[-20:]) if len(sizes) >= 20 else np.mean(sizes)
    vol_std = np.std(sizes[-20:]) if len(sizes) >= 20 else 1
    volume_zscore = (trade["size"] - vol_mean) / vol_std if vol_std > 0 else 0

    time_diff = 0
    if len(buffer) >= 2:
        prev_time = pd.to_datetime(buffer[-1]["time"])
        curr_time = pd.to_datetime(trade["time"])
        time_diff = (curr_time - prev_time).total_seconds()

    side_encoded = 1 if trade["side"] == "buy" else 0

    return np.array([[price_deviation, volume_zscore, time_diff, side_encoded]])

What does this code do?

This mirrors the batch feature engineering from train_model.py but operates on one trade at a time. It computes price_deviation (how far the price is from its 20-trade rolling mean), volume_zscore (trade size relative to recent volume), time_diff (seconds since the last trade), and side_encoded (buy=1, sell=0).

The feature names and order must match training exactly. The Isolation Forest expects the same 4-dimensional input it was fitted on.

• Now add the drift detection function below engineer_features_single:

def check_drift(current_window):
    global reference_data
    if reference_data is None or len(current_window) < DRIFT_WINDOW:
        return

    current_df = pd.DataFrame(
        current_window,
        columns=["price_deviation", "volume_zscore", "time_diff", "side_encoded"],
    )

    report = Report(metrics=[DataDriftPreset()])
    report.run(reference_data=reference_data, current_data=current_df)
    result = report.as_dict()

    is_drift = result["metrics"][0]["result"]["dataset_drift"]
    drift_detected.set(1 if is_drift else 0)

    if is_drift:
        print("DRIFT DETECTED: Current data distribution differs from training!")

What does this code do?

Every 100 trades, the consumer compares the current feature distributions against the training reference using Evidently's DataDriftPreset. If statistical tests detect a shift (e.g., price volatility has changed), it sets the drift_detected Prometheus gauge to 1. Prometheus scrapes this value so Grafana can visualize it.

• Save consumer.py and confirm it still imports cleanly:

python -c "import consumer; print('All functions defined')"

You should see All functions defined with no errors.

Build the main consumer loop and verify anomalies in Elasticsearch

Now for the core loop. The run_consumer function ties everything together: it loads the model, subscribes to Kafka, and enters an infinite loop where it reads, scores, and acts on each trade.

• Add the run_consumer function setup below your existing functions in consumer.py. This loads the model, creates the Kafka consumer, and starts the Prometheus metrics server:

def run_consumer():
    model = load_model()
    load_reference()

    consumer = Consumer({
        "bootstrap.servers": KAFKA_BOOTSTRAP,
        "group.id": "anomaly-detector",
        "auto.offset.reset": "latest",
    })
    consumer.subscribe([TOPIC])

    start_http_server(8000)
    print("Consumer running. Prometheus metrics at :8000")

    feature_window = []
    trade_count = 0

What does this code do?

• start_http_server(8000) launches a Prometheus metrics endpoint. The prometheus.yml you created in Step 1 already scrapes host.docker.internal:8000.
• auto.offset.reset: latest means the consumer only reads NEW messages arriving after it starts, not historical ones.
• feature_window accumulates feature vectors so drift detection can compare a batch against the training reference.

• Directly below, add the main message loop with feature engineering, scoring, and Prometheus metric recording:

    try:
        while True:
            msg = consumer.poll(1.0)
            if msg is None:
                continue
            if msg.error():
                continue

            trade = json.loads(msg.value().decode("utf-8"))
            trade["price"] = float(trade["price"])
            trade["size"] = float(trade["size"])
            trade_buffer.append(trade)

            if len(trade_buffer) < 5:
                continue

            features = engineer_features_single(trade, list(trade_buffer))

            start_time = time.time()
            prediction = model.predict(features)[0]
            score = model.score_samples(features)[0]
            latency = time.time() - start_time

            trades_processed.inc()
            inference_latency.observe(latency)

            feature_window.append(features[0].tolist())
            trade_count += 1

What does the scoring loop do?

• The buffer guard (if len(trade_buffer) < 5) waits for at least 5 trades before computing features, ensuring rolling statistics are meaningful.
• model.predict(features) returns 1 for normal or -1 for anomaly. model.score_samples(features) returns how anomalous the point is (more negative = more unusual).
• trades_processed.inc() and inference_latency.observe(latency) push live metrics to the Prometheus endpoint for Grafana to display.

• Finally, add the anomaly handling, drift check, exception handler, and the main entry point below the code you just pasted:

            if prediction == -1:
                anomalies_detected.inc()
                anomaly_doc = {
                    "timestamp": trade["time"],
                    "price": trade["price"],
                    "size": trade["size"],
                    "side": trade["side"],
                    "anomaly_score": float(score),
                    "features": {
                        "price_deviation": features[0][0],
                        "volume_zscore": features[0][1],
                        "time_diff": features[0][2],
                    },
                }
                es.index(index="anomalies", document=anomaly_doc)
                print(f"ANOMALY: ${trade['price']:.2f} | Score: {score:.4f}")

            if trade_count % DRIFT_WINDOW == 0 and len(feature_window) >= DRIFT_WINDOW:
                check_drift(feature_window[-DRIFT_WINDOW:])

    except KeyboardInterrupt:
        print("Shutting down consumer...")
    finally:
        consumer.close()


if __name__ == "__main__":
    run_consumer()

What does the anomaly block do?

• When prediction == -1, the Isolation Forest is saying "this trade is an outlier". The consumer packages the full trade context into a document and indexes it into Elasticsearch's anomalies index.
• es.index(index="anomalies", document=anomaly_doc) creates the index automatically on first write. Elasticsearch auto-creates indexes when action.auto_create_index is enabled (the default).
• Every 100 trades (trade_count % DRIFT_WINDOW == 0), the consumer runs a drift check comparing recent features against the training baseline.

• Save consumer.py.

Make sure the producer is still running

The consumer reads from the same crypto.trades.raw topic your producer writes to. Switch to the terminal where producer.py is running and confirm you still see trade output. If it stopped, restart it with python producer.py in a separate terminal before continuing.

Before you run the consumer, what do you think will happen? The model was trained with a 5% contamination rate, so roughly 1 in 20 trades should be flagged as anomalous.

• Open a new terminal tab and run the consumer:

python consumer.py

You should see output confirming the model and reference data loaded, followed by ANOMALY alerts interspersed with normal trade processing:

Loaded model: crypto-anomaly-detector v1
Loaded reference data: 247 samples
Consumer running. Prometheus metrics at :8000
ANOMALY: $104523.12 | Score: -0.1832
ANOMALY: $104498.77 | Score: -0.2104

Let it run for 30 seconds to accumulate some anomalies. Now verify they landed in Elasticsearch.

• Open another terminal tab and query the anomaly count:

curl http://localhost:9200/anomalies/_count

You should see a JSON response with a count greater than 0, confirming anomaly documents are being indexed.

Seeing zero anomalies or a connection error?

• If the count is 0, let the consumer run longer. The buffer needs at least 5 trades before scoring begins.
• If you see a connection refused error, confirm Elasticsearch is running with docker compose ps elasticsearch. The container should show status Up.
• If you see a model-not-found error, confirm you completed Step 3 and the model is registered in MLflow at http://localhost:5000.

Why aren't anomalies appearing in Elasticsearch?

✔️ Awesome, I've got everything!

Great. Double check you have saved consumer.py and it is running without import errors.

ⓧ I'd like to double check the full code

Your consumer.py should contain these sections in order. Scroll through your file and confirm each is present:

• Imports (json, time, numpy, pandas, deque, confluent_kafka, elasticsearch, prometheus_client, mlflow, evidently).
• Configuration constants (KAFKA_BOOTSTRAP, TOPIC, MLFLOW_TRACKING_URI, MODEL_NAME, ES_HOST, DRIFT_WINDOW).
• Prometheus metric definitions (trades_processed, anomalies_detected, inference_latency, drift_detected).
• Elasticsearch client and global state (es, trade_buffer, reference_data).
• load_model() and load_reference() functions.
• engineer_features_single() function.
• check_drift() function.
• run_consumer() function with the full while loop, anomaly handling, drift check, and KeyboardInterrupt handler.
• if __name__ == "__main__": run_consumer() entry point.

Your real-time inference pipeline is live. Trades flow in from Coinbase, get scored by your Isolation Forest, and anomalies land in Elasticsearch for search. The model flags unusual trades, but you have no visibility into whether the live data has shifted from what it was trained on. That drift gap is exactly what the next step closes with a Grafana dashboard.

Add Drift Monitoring and Grafana Dashboard

Your consumer is scoring every trade and flagging anomalies. But here is the gap: the model was trained on data from a few minutes ago, and live crypto markets shift constantly.

Without visibility into whether your model's assumptions still hold, you are flying blind. In this step, you will confirm that the Evidently drift monitoring already built into your consumer is reporting to Prometheus, then build a Grafana dashboard that gives you a single pane of glass across the entire pipeline.

In this step, get ready to:

• Confirm Prometheus is scraping your consumer's drift and inference metrics.
• Build a Grafana dashboard with panels for throughput, anomaly rate, inference latency, and drift status.
• Observe the live pipeline and watch drift metrics respond to changing market conditions.

Verify Prometheus is scraping your consumer metrics

Your consumer already exposes four Prometheus metrics on port 8000, including the drift_detected gauge that Evidently updates every 100 trades. Prometheus needs to be actively scraping these metrics for Grafana to display them.

• Open the Prometheus targets page in your browser:

open http://localhost:9090/targets

• Find the anomaly-consumer job in the list.
• Confirm its State shows UP with a green indicator.

What does this confirm?

The config/prometheus.yml you created earlier includes a scrape job targeting host.docker.internal:8000. That is your consumer's Prometheus metrics endpoint. If the target shows UP, Prometheus is successfully pulling trades_processed_total, anomalies_detected_total, inference_latency_seconds, and drift_detected every 15 seconds.

• Verify the drift metric exists by querying it directly. Open the Prometheus expression browser:

open http://localhost:9090/graph

• Type drift_detected into the expression field and click Execute.

You should see a result with value 0 (no drift yet) or 1 (drift detected). Either value confirms the metric is flowing.

Target showing DOWN or missing?

• Make sure your consumer is still running (python consumer.py in a separate terminal). The Prometheus metrics server starts when the consumer starts.
• If you are on macOS, confirm host.docker.internal resolves correctly. Docker Desktop for Mac supports this by default.
• Check that no other process is using port 8000 by running lsof -i :8000.

Need more help? Why is my Prometheus target showing as DOWN for the anomaly-consumer job?

Create the Grafana dashboard

Grafana is already running and pre-configured with Prometheus as a datasource (via the provisioning file you created earlier). Now you will build a dashboard with four panels that visualize the entire pipeline at a glance.

• Open Grafana in your browser:

open http://localhost:3000

• Log in with username admin and password admin.
• If prompted to change the password, click Skip.
• Click Dashboards in the left sidebar menu.
• Click New and select New Dashboard.
• On the empty dashboard, click + Add visualization.
• Select Prometheus as the data source.

The panel editor opens. You will create your first panel: trade throughput.

• In the query editor at the bottom, enter this PromQL query:

rate(trades_processed_total[1m])

• Click Refresh (or Run queries) to see the graph populate.
• Under Panel options on the right, set the title to Trade Throughput (trades/sec).
• Click Back to dashboard in the top-left to see your panel placed on the dashboard.

You should see a time-series line showing how many trades per second your consumer is processing.

What does this query do?

The rate() function calculates the per-second increase of the trades_processed_total counter over a 1-minute window. This turns a monotonically increasing counter into a meaningful "trades per second" rate that shows real-time throughput.

• Add a second panel by clicking Add in the dashboard header and selecting Visualization.
• Select Prometheus as the data source.
• Enter this PromQL query for anomaly detection rate:

rate(anomalies_detected_total[1m]) * 60

• Click Refresh to see the graph.
• Set the panel title to Anomaly Rate (anomalies/min).
• Click Back to dashboard.

This panel shows spikes whenever your Isolation Forest flags unusual trades.

• Add a third panel (Add > Visualization, select Prometheus).
• Enter this PromQL query for inference latency p95:

histogram_quantile(0.95, rate(inference_latency_seconds_bucket[5m]))

• Click Refresh.
• Set the panel title to Inference Latency p95 (seconds).
• Click Back to dashboard.

Why p95 instead of average?

The 95th percentile shows you the worst-case latency that 95% of inferences complete within. An average can hide occasional spikes. In production ML pipelines, p95 latency is the standard SLO metric because it reveals tail latency issues that would otherwise be invisible.

• Add the final panel (Add > Visualization, select Prometheus).
• Enter this PromQL query for the Evidently drift status:

drift_detected

• Click Refresh.
• Set the panel title to Data Drift Status.
• Click Back to dashboard.

What does the drift panel show?

This gauge shows 0 when the live data distribution matches the training baseline, and 1 when Evidently detects a statistically significant shift. Your consumer runs a DataDriftPreset report every 100 trades, comparing the current feature window against training_reference.csv from your MLflow experiment.

• Now save the dashboard. Click Save dashboard in the top-right corner.
• Enter Crypto Anomaly Pipeline as the dashboard title.
• Click Save.

Your dashboard now has four panels showing trade throughput, anomaly rate, inference latency, and drift status.

Panels showing No data?

• Make sure your producer and consumer are both still running. Grafana can only show data that Prometheus has scraped from the consumer.
• Check that the time range in Grafana (top-right dropdown) is set to Last 15 minutes or Last 5 minutes. If it is set to a longer range, the recent data may not be visible.
• Verify that Prometheus is collecting data by going back to http://localhost:9090/graph and querying trades_processed_total. If this returns no data, the issue is between Prometheus and your consumer.

Need more help? My Grafana panels are showing No data even though my consumer is running. Help me debug the connection between Grafana, Prometheus, and my consumer metrics.

Observe the live pipeline with drift monitoring

Your full pipeline is now instrumented end-to-end. The producer streams live trades, the consumer scores them and reports drift, and Grafana visualizes everything in real-time.

• Return to your Grafana dashboard and set the auto-refresh interval to 10 seconds (click the refresh dropdown in the top-right and select 10s).
• Watch the dashboard for 1-2 minutes. You should see the trade throughput line moving steadily, anomaly spikes appearing intermittently, and the drift status toggling if market conditions shift enough.

Before you check the final verification, take a moment to predict: do you think the drift panel will show 0 or 1 right now? Your model was trained just minutes ago on the same market. Think about whether enough time has passed for conditions to measurably shift.

• Check the Data Drift Status panel value.

Most likely the panel shows 0 (no drift). Your model was trained on very recent data, so the live distribution still matches. Over hours or days, as the market shifts, this value will flip to 1 and you will know it is time to retrain.

Why does drift matter for production ML?

In production, models degrade silently. The accuracy drops but no error fires. Drift monitoring is your early warning system. When Evidently detects that price distributions or volume patterns have shifted away from your training data, you know the model's predictions are becoming less reliable before users notice.

This is the same pattern fintech companies use to monitor fraud detection models. The model itself does not know it is stale. You need external monitoring to tell you.

You now have a fully observable ML pipeline. Live trades flow through Kafka, the Isolation Forest scores every trade, Evidently watches for distribution shifts, and Grafana shows you the complete picture in one dashboard. This is the production MLOps stack that separates a notebook experiment from a deployed system.

Secret mission

Vector-Based Anomaly Pattern Retrieval with Qdrant

When your pipeline detects an anomaly, the natural next question is: have we seen something like this before? In this mission, you'll embed anomaly patterns as vectors in Qdrant and query for historically similar anomalies, enriching every alert with context from the past.

Clean Up Your Resources

Clean Up Your Resources

Decide whether to keep your resources running, pause them to come back later, or delete them entirely. This project runs entirely locally using Docker, so there are no ongoing financial costs. However, the 13 containers consume significant RAM and disk space while running.

Resources you used:

• Docker containers: 13 services (Kafka, Zookeeper, Schema Registry, Kafka UI, PostgreSQL, Elasticsearch, Prometheus, Grafana, Alertmanager, Jaeger, cAdvisor, MLflow, Qdrant).
• Docker volumes: trade data in Kafka, anomaly documents in Elasticsearch, anomaly vectors in Qdrant, MLflow metadata in PostgreSQL.
• Local project files: docker-compose.yml, producer.py, consumer.py, train_model.py, requirements.txt, and the config/ directory.

Keep everything running

No action needed. Choose this if you're still actively building or want to keep collecting live trade data and detecting anomalies in real-time.

• All 13 containers continue running and processing live BTC-USD trades.
• Your Grafana dashboard stays live at http://localhost:3000 with ongoing throughput, anomaly rate, and drift metrics.
• Monitor container resource usage with cAdvisor at http://localhost:8080.
• Your Elasticsearch anomaly index and Qdrant vector collection continue growing as new anomalies are detected.

Pause - I'll come back to this later

Shut down all running containers to free up RAM while preserving your data volumes. You can resume the full pipeline later without retraining your model or losing anomaly history.

• Stop the producer and consumer scripts in their terminal windows by pressing Ctrl+C.
• Stop all Docker containers by running this command from your project folder:

docker compose stop

• When you're ready to resume, start everything back up by running:

docker compose start

Your Kafka topics, Elasticsearch index, Qdrant vectors, MLflow experiments, and PostgreSQL data all persist in Docker volumes.

Delete - I don't want to use this again

Remove all containers, volumes, and local project files. This permanently deletes your trade data, trained models, anomaly history, and vector embeddings.

• Stop the producer and consumer scripts in their terminal windows by pressing Ctrl+C.
• Remove all Docker containers and their data volumes by running this command from your project folder:

docker compose down -v

• Confirm the containers are gone by running:

docker compose ps

You should see an empty list with no running containers.

• Delete your local project files by navigating to your Desktop and removing the project folder:

cd ~/Desktop
rm -rf crypto-anomaly-pipeline

Confirm with ls

Run ls ~/Desktop to verify the project folder is gone. If your folder had a different name, substitute it in the rm -rf command above.

That's a Wrap!

That's a Wrap!

Nice work. You just built a production-grade real-time anomaly detection pipeline that streams live Bitcoin trades, runs ML inference on every single one, and alerts you when something unusual hits the market.

You've learned how to:

• Stream live cryptocurrency trades from Coinbase's WebSocket into Kafka and score every trade in real-time with an Isolation Forest anomaly detection model, indexing flagged anomalies into Elasticsearch.
• Build a complete MLOps stack with experiment tracking and model registry in MLflow, data drift monitoring in Evidently, and a Grafana dashboard showing pipeline throughput, anomaly rates, and drift status.
• Orchestrate 13+ services from scratch using Docker Compose, including Kafka, Schema Registry, PostgreSQL, Elasticsearch, Prometheus, Grafana, Alertmanager, Jaeger, cAdvisor, MLflow, and Qdrant.
• Secret Mission: Implemented vector-based anomaly pattern retrieval with Qdrant to find historically similar anomalies using cosine similarity search.

Ready to quiz yourself?