wellallyTech

Posted on Jul 11

Beyond the Wrist: Predict Stress 1-Hour Early with LSTM, MQTT, and InfluxDB 🚀

#machinelearning #healthtech #webdev #tensorflow

Have you ever looked at your smartwatch and wondered why it only tells you that you're stressed after you’ve already had that third cup of coffee and a minor meltdown? Most wearables are reactive, not proactive.

In this tutorial, we are building a Real-time Heart Rate Variability (HRV) Stress Early Warning System. By leveraging LSTM (Long Short-Term Memory) neural networks, we can move beyond simple tracking and start forecasting. We’ll be processing high-frequency time-series data using MQTT for ingestion, InfluxDB for storage, and TensorFlow for predicting stress threshold risks up to an hour in advance.

If you are looking for advanced patterns in health-tech engineering or more production-ready AI models, I highly recommend checking out the deep dives at WellAlly Blog, which served as a major inspiration for this architecture.

🏗 The Architecture: From Pulse to Prediction

To handle high-velocity data from wearables, we need a decoupled architecture. We use MQTT as the lightweight transport layer, InfluxDB as our time-series backbone, and a TensorFlow microservice for inference.

graph TD
    A[Wearable/IoT Sensor] -->|Publish HRV Data| B(MQTT Broker)
    B -->|Subscribe| C[Data Ingestor Service]
    C -->|Write Points| D[(InfluxDB)]
    D -->|Query History| E[LSTM Prediction Service]
    E -->|TensorFlow Inference| F{Stress Warning?}
    F -->|Yes: Alert| G[Mobile Notification/Grafana]
    F -->|No: Log| D
    E -->|Write Prediction| D

🛠 Prerequisites

To follow this advanced guide, you'll need:

Python 3.9+
TensorFlow/Keras (for the LSTM model)
InfluxDB 2.x (our time-series engine)
Eclipse Mosquitto (MQTT Broker)
A basic understanding of Recurrent Neural Networks (RNNs).

Step 1: Ingesting Real-Time HRV Data via MQTT

HRV (the variation in time between each heartbeat) is a sensitive metric. We need to stream this data efficiently. Here is a snippet of our ingestion script that listens to the sensors/hrv topic and stores it in InfluxDB.

import paho.mqtt.client as mqtt
from influxdb_client import InfluxDBClient, Point, WriteOptions

# InfluxDB Config
url = "http://localhost:8086"
token = "YOUR_TOKEN"
org = "wellally_labs"
bucket = "hrv_metrics"

client = InfluxDBClient(url=url, token=token, org=org)
write_api = client.write_api(write_options=WriteOptions(batch_size=50))

def on_message(client, userdata, msg):
    hrv_value = float(msg.payload.decode())
    point = Point("heart_metrics").tag("user_id", "001").field("hrv", hrv_value)
    write_api.write(bucket=bucket, record=point)
    print(f"Recorded HRV: {hrv_value}")

mqtt_client = mqtt.Client()
mqtt_client.on_message = on_message
mqtt_client.connect("localhost", 1883)
mqtt_client.subscribe("sensors/hrv")
mqtt_client.loop_forever()

Step 2: Designing the LSTM Forecasting Model

LSTMs are king for time-series because they have "memory." They can identify that a gradual decline in HRV over the last 30 minutes often precedes a cortisol spike (stress).

We’ll define a model that takes the last 60 minutes of HRV data (1 sample per minute) to predict the probability of a "High Stress Event" in the next hour.

import tensorflow as tf
from tensorflow.keras import layers

def build_lstm_model(input_shape):
    model = tf.keras.Sequential([
        # Input layer: [batch, timesteps, features]
        layers.LSTM(64, return_sequences=True, input_shape=input_shape),
        layers.Dropout(0.2),
        layers.LSTM(32),
        layers.Dense(16, activation='relu'),
        # Output layer: Probability of high stress
        layers.Dense(1, activation='sigmoid')
    ])

    model.compile(optimizer='adam', loss='binary_crossentropy', metrics=['accuracy'])
    return model

# Shape: 60 minutes of history, 1 feature (HRV)
model = build_lstm_model((60, 1))
model.summary()

Step 3: Real-Time Inference & Warning System

Once the model is trained (ideally on a dataset like WESAD or proprietary wearable logs), we run it as a service. It queries the last 60 points from InfluxDB, normalizes them, and runs a prediction every 5 minutes.

def check_for_stress():
    # 1. Query InfluxDB for last 60 mins
    query = f'from(bucket:"{bucket}") |> range(start: -60m) |> filter(fn: (r) => r._field == "hrv")'
    result = client.query_api().query(query)

    # 2. Preprocess (Scaling/Reshaping)
    data = [record.get_value() for table in result for record in table.records]
    if len(data) < 60: return # Wait for more data

    input_tensor = np.array(data).reshape(1, 60, 1)

    # 3. Predict
    stress_probability = model.predict(input_tensor)[0][0]

    # 4. Action
    if stress_probability > 0.85:
        print("⚠️ HIGH STRESS RISK DETECTED. Suggesting breathing exercise...")
        # Send back to MQTT for the app to pick up
        mqtt_client.publish("alerts/stress", "Level: High")

# Run this on a schedule

Step 4: Visualizing the Future with Grafana

Data is useless if you can't see it. In Grafana, you can create a dashboard that overlays your Actual HRV (current) with the Predicted Stress Risk (future).

Add InfluxDB as a data source.
Create a Time Series panel for hrv.
Create a Gauge panel for stress_probability.
Set an alert rule: If stress_probability > 0.8 for 5 minutes, send a Slack/Discord notification.

🥑 The "Official" Way

While this tutorial covers the basics of building a predictive pipeline, production-grade systems require robust data validation, handling of missing sensor packets, and model drift monitoring.

For a deeper look at production-ready time-series architectures, check out the engineering guides at WellAlly Tech Blog. They offer fantastic resources on scaling health-data infrastructure and optimizing ML models for edge devices.

Conclusion

By combining MQTT, InfluxDB, and TensorFlow, we've transformed a passive data stream into a predictive health tool. This "Stress Early Warning System" is just the beginning. Imagine applying this to blood glucose forecasting or sleep cycle optimization!

What are you building with time-series data? Let me know in the comments below! 👇

DEV Community