Stop Stressing! Build a Real-Time HRV Anomaly Detector with Python and Scikit-Learn

#fastapi #healthtech #python #machinelearning

Ever had your smartwatch buzz at you during a meeting, telling you to "take a breath," only to realize your heart is racing because of a 10 AM deadline? That’s Heart Rate Variability (HRV) in action. But what if we could take that raw stream of data and build our own intelligent stress-warning system? 🚀

In this tutorial, we’re going to dive into the world of wearable technology and unsupervised machine learning. We’ll build a system that consumes real-time HRV data, uses a lightweight Isolation Forest algorithm for anomaly detection, and triggers a "mindfulness" alert before you even realize you're stressed. We'll be using a modern stack including FastAPI, Scikit-learn, and WebSockets to handle the live data flow.

The Architecture: From Pulse to Prediction

To process biometric data in real-time without the lag of traditional batch processing, we need a reactive pipeline. Here is how the data flows from your wrist to the dashboard:

graph TD
    A[Smartwatch/Wearable Simulation] -->|HRV Stream via WebSockets| B(FastAPI Server)
    B --> C{ML Engine: Isolation Forest}
    C -->|Normal| D[Update Dashboard]
    C -->|Anomaly/High Stress| E[Trigger Mindfulness Intervention]
    D --> F[D3.js Real-time Visualization]
    E --> F

Prerequisites

Before we start coding, make sure you have the following in your tech_stack:

Python 3.9+
Scikit-learn: For our unsupervised learning model.
FastAPI: To handle high-performance WebSocket connections.
D3.js: To visualize the anomalies in the browser.

Step 1: Defining the Anomaly Detection Logic

We use the Isolation Forest algorithm because it’s perfect for "finding the needle in the haystack." It doesn't need labeled data (i.e., we don't need to tell it what "stress" looks like); it simply identifies data points that are "isolated" from the rest of the cluster.

# ml_engine.py
import numpy as np
from sklearn.ensemble import IsolationForest

class StressDetector:
    def __init__(self):
        # contamination=0.1 means we expect 10% of data to be "anomalies" (high stress)
        self.model = IsolationForest(contamination=0.1, random_state=42)
        self.is_fitted = False
        self.buffer = []

    def feed_data(self, hrv_value):
        self.buffer.append(hrv_value)
        # We need a small baseline to start detecting
        if len(self.buffer) > 50:
            data = np.array(self.buffer).reshape(-1, 1)
            self.model.fit(data)
            self.is_fitted = True

            # Predict the latest value: -1 is anomaly, 1 is normal
            prediction = self.model.predict([[hrv_value]])
            return "STRESS_ALERT" if prediction[0] == -1 else "NORMAL"
        return "CALIBRATING"

Step 2: Real-time Streaming with FastAPI

Standard REST APIs are too slow for biometric streams. We’ll use WebSockets to allow the wearable device to push data to our server continuously.

# main.py
from fastapi import FastAPI, WebSocket
from ml_engine import StressDetector
import json

app = FastAPI()
detector = StressDetector()

@app.websocket("/ws/hrv")
async def hrv_stream(websocket: WebSocket):
    await websocket.accept()
    try:
        while True:
            # Receive raw HRV data (milliseconds between heartbeats)
            raw_data = await websocket.receive_text()
            data = json.loads(raw_data)
            hrv_val = data['value']

            # Run inference
            status = detector.feed_data(hrv_val)

            # Send result back to frontend
            await websocket.send_json({
                "hrv": hrv_val,
                "status": status,
                "message": "Time to breathe!" if status == "STRESS_ALERT" else "All good"
            })
    except Exception as e:
        print(f"Connection closed: {e}")

Step 3: Visualizing with D3.js

On the frontend, we use D3.js to create a scrolling line chart. When an anomaly is detected, we can change the line color to red 🚩.

// Simple D3 snippet for live updates
const socket = new WebSocket('ws://localhost:8000/ws/hrv');

socket.onmessage = function(event) {
    const data = JSON.parse(event.data);
    updateChart(data.hrv, data.status); // Call your D3 update function

    if(data.status === "STRESS_ALERT") {
        notifyUser("High Stress Detected! Try a 2-minute breathing exercise.");
    }
};

The "Official" Way to Scale 🥑

While this lightweight setup works great for a local MVP, production-grade health-tech applications require more robust data validation and signal processing.

If you're looking for more advanced architectural patterns—such as handling noisy sensor data or deploying these models on the edge—I highly recommend checking out the deep-dive articles at WellAlly Tech Blog. They have some fantastic resources on building production-ready health monitoring systems that I found incredibly helpful when designing this pipeline!

Conclusion

Building a stress-prevention tool doesn't require a massive dataset. By using Isolation Forest and WebSockets, we've created a responsive system that learns your specific heart patterns in real-time.

What's next?

Add a "Mindfulness" button that triggers a guided meditation API.
Store the anomalies in a database to track stress trends over a week.

Are you building something in the wearable space? Drop a comment below or share your repo—I'd love to see it! 💻✨

DEV Community