Have you ever looked at your Heart Rate Variability (HRV) on your Apple Watch or Oura Ring and wondered what it actually means for your health? HRV is more than just a metric; it's a window into your Autonomic Nervous System. However, raw biometric data is incredibly noisy. To turn these data points into actionable insights, we need time-series anomaly detection to identify when your body is deviating from its baseline—signaling potential overtraining, high stress, or an oncoming flu.
In this tutorial, we will explore how to build a real-time physiological stress forecasting system using LSTM Neural Networks, TensorFlow, and InfluxDB. By leveraging deep learning for time-series, we can move past simple threshold alerts and create a personalized "Early Warning System" for your body. If you're looking for more production-ready patterns on biometric signal processing, be sure to check out the deep-dive articles at WellAlly Tech Blog.
🏗 The Architecture
To handle high-frequency wearable data, we need a pipeline that can ingest, store, and analyze sequences effectively. Here is how the data flows from your finger to the AI model:
graph TD
A[Oura Ring / Apple Watch] -->|REST API/HealthKit| B(Data Ingestion Script)
B -->|Cleaned Data| C[(InfluxDB Time-Series)]
C -->|Windowed Sequences| D[Pandas Preprocessing]
D -->|Feature Scaling| E[LSTM Autoencoder Model]
E -->|Reconstruction Error| F{Anomaly Detected?}
F -->|Yes| G[Push Notification: 'Take a Rest Day']
F -->|No| H[Dashboard Update]
🛠 Prerequisites
Before we dive into the code, ensure you have the following stack ready:
- Python 3.9+
- TensorFlow/Keras: For building our LSTM Autoencoder.
- InfluxDB: Optimized for time-series storage.
- Pandas: For data manipulation.
- Oura API Access: Or any CSV export of your HRV data.
🚀 Step 1: Ingesting Data from Oura API
First, we need to pull the HRV (specifically rmssd - Root Mean Square of Successive Differences) data.
import requests
import pandas as pd
def fetch_oura_hrv(api_token, start_date, end_date):
url = f'https://api.ouraring.com/v2/usercollection/daily_readiness'
headers = {'Authorization': f'Bearer {api_token}'}
params = {'start_date': start_date, 'end_date': end_date}
response = requests.get(url, headers=headers, params=params)
data = response.json()['data']
# Extracting HRV (rmssd) and timestamps
df = pd.DataFrame([{
'time': d['day'],
'hrv': d['readiness_score_delta'] # Simplified example
} for d in data])
df['time'] = pd.to_datetime(df['time'])
return df.set_index('time')
🗄 Step 2: Storing in InfluxDB
Why InfluxDB? Because physiological data is inherently time-bound. InfluxDB allows us to perform windowed queries (e.g., "Give me the last 30 days of HRV") with lightning speed.
from influxdb_client import InfluxDBClient, Point, WritePrecision
from influxdb_client.client.write_api import SYNCHRONOUS
def store_hrv_data(df, client_config):
client = InfluxDBClient(url=client_config['url'], token=client_config['token'], org=client_config['org'])
write_api = client.write_api(write_options=SYNCHRONOUS)
for timestamp, row in df.iterrows():
point = Point("biometrics") \
.tag("user", "dev_advocate") \
.field("hrv_rmssd", float(row['hrv'])) \
.time(timestamp, WritePrecision.NS)
write_api.write(bucket=client_config['bucket'], record=point)
🧠 Step 3: The LSTM Autoencoder Logic
The secret sauce for anomaly detection is the LSTM Autoencoder. Instead of predicting the next value, the model tries to reconstruct the input sequence.
- Normal State: The model learns to reconstruct your "healthy" HRV patterns with low error.
- Anomaly State: When you are sick or stressed, your HRV pattern changes. The model fails to reconstruct it accurately, resulting in a high Reconstruction Error.
import numpy as np
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dropout, RepeatVector, TimeDistributed, Dense
def build_lstm_autoencoder(sequence_length, n_features):
model = Sequential([
# Encoder
LSTM(64, activation='relu', input_shape=(sequence_length, n_features), return_sequences=True),
LSTM(32, activation='relu', return_sequences=False),
RepeatVector(sequence_length),
# Decoder
LSTM(32, activation='relu', return_sequences=True),
LSTM(64, activation='relu', return_sequences=True),
TimeDistributed(Dense(n_features))
])
model.compile(optimizer='adam', loss='mae')
return model
# Example: 7 days of data as a sequence
model = build_lstm_autoencoder(sequence_length=7, n_features=1)
model.summary()
💡 The "Official" Way to Scale
While the code above works for a personal project, deploying this at scale for thousands of users requires robust data engineering and model monitoring. If you're interested in how to handle massive time-series streams or integrating with more complex multimodal vision data (like analyzing workout form alongside heart rate), check out the advanced implementation guides at wellally.tech/blog. It’s a goldmine for developers looking to push the boundaries of HealthTech. 🥑
📈 Step 4: Detecting Deviations
Once the model is trained on your "healthy" months, we calculate the Mean Absolute Error (MAE). If the MAE on new data exceeds a certain threshold (e.g., 2 standard deviations), we trigger an alert.
def detect_anomalies(model, data, threshold):
reconstructions = model.predict(data)
loss = np.mean(np.abs(reconstructions - data), axis=1)
anomalies = loss > threshold
return anomalies, loss
# Let's say threshold is calculated from training distribution
# anomalies, loss = detect_anomalies(model, x_test, threshold=0.15)
🎯 Conclusion
By moving from static thresholds to LSTM-based anomaly detection, we can account for the natural seasonality of our bodies. Your "low" HRV on a Monday might be normal after a marathon, but that same "low" on a sedentary Thursday could be the first sign of a viral infection. 🤒
Ready to try it?
- Pull your data via the Oura API.
- Train the Autoencoder on your "best" health month.
- Set up a Cron job to run this every morning.
What are you using to track your health data? Let me know in the comments! 👇
Top comments (0)