Mastering Stress: Real-time HRV Anomaly Detection with Time-Series Transformers and HealthKit 🚀

Stress is often called the "silent killer," but your wearable device is screaming the warning signs long before you feel the burnout. The challenge? Most health apps are reactive, telling you that you were stressed yesterday. We want to be proactive.

In this deep dive, we’re going to build a high-performance pipeline for real-time Heart Rate Variability (HRV) anomaly detection. By leveraging Time-Series Transformers (TST), HealthKit, and TimescaleDB, we can predict stress spikes before they happen and trigger system-level interventions via WebSockets. This tutorial covers the intersection of predictive health analytics, wearable technology, and high-frequency data streaming.

The Architecture 🏗️

To handle the high-frequency sampling required for HRV (measuring the milliseconds between heartbeats), we need a robust data flow. Here is how the system components interact:

graph TD
    A[Apple Watch / HealthKit] -->|SDNN/RMSSD Data| B(iOS App)
    B -->|WebSocket Stream| C[FastAPI Gateway]
    C -->|Write Batch| D[(TimescaleDB)]
    C -->|Inference Request| E[PyTorch TST Model]
    E -->|Anomaly Score| F{Threshold Exceeded?}
    F -->|Yes| G[System Intervention / Push Alert]
    F -->|No| H[Continue Monitoring]
    D -.->|Re-training| E

Prerequisites 🛠️

Before we dive into the code, ensure you have the following:

• iOS Development: Xcode and a device with HealthKit access.
• Backend: Python 3.9+, PyTorch, and FastAPI.
• Database: TimescaleDB (PostgreSQL extension for time-series data).

Step 1: Streaming HealthKit Data via WebSockets

Apple’s HealthKit provides high-resolution heart rate data. For HRV, we focus on the SDNN (Standard Deviation of NN intervals). In your iOS app, you'll want to stream these samples to your backend in real-time.

// Swift snippet for streaming HRV samples
import HealthKit

func startHRVStreaming() {
    let hrvType = HKQuantityType.quantityType(forIdentifier: .heartRateVariabilitySDNN)!
    let query = HKAnchoredObjectQuery(type: hrvType, predicate: nil, anchor: lastAnchor, limit: HKObjectQueryNoLimit) { (query, samples, deleted, newAnchor, error) in
        guard let samples = samples as? [HKQuantitySample] else { return }

        for sample in samples {
            let hrvValue = sample.quantity.doubleValue(for: HKUnit.secondUnit(with: .milli))
            let timestamp = sample.startDate.timeIntervalSince1970

            // Send via WebSocket
            self.webSocketTask.send(.string("{\"value\": \(hrvValue), \"ts\": \(timestamp)}"))
        }
    }
    healthStore.execute(query)
}

Step 2: The Brain — Time-Series Transformer (TST)

While LSTMs were the old kings of time-series, Transformers excel at capturing long-range dependencies in physiological data. We use a TimeSeriesTransformer in PyTorch to predict the next HRV window. An "anomaly" is detected when the actual HRV drops significantly below the predicted baseline (indicating sympathetic nervous system dominance).

import torch
import torch.nn as nn

class HRVTransformer(nn.Module):
    def __init__(self, input_dim=1, model_dim=64, num_heads=4, num_layers=3):
        super().__init__()
        self.embedding = nn.Linear(input_dim, model_dim)
        self.pos_encoder = nn.Parameter(torch.zeros(1, 500, model_dim)) # Max 500 time steps

        encoder_layers = nn.TransformerEncoderLayer(d_model=model_dim, nhead=num_heads)
        self.transformer = nn.TransformerEncoder(encoder_layers, num_layers=num_layers)
        self.decoder = nn.Linear(model_dim, 1)

    def forward(self, x):
        # x shape: (batch, seq_len, input_dim)
        x = self.embedding(x) + self.pos_encoder[:, :x.size(1), :]
        x = x.transpose(0, 1) # Transformer expects (seq_len, batch, dim)
        output = self.transformer(x)
        return self.decoder(output[-1, :, :]) # Predict the next value

Step 3: Real-time Ingestion with TimescaleDB

Generic SQL databases struggle with high-frequency health data. TimescaleDB allows us to perform hyper-fast aggregates (e.g., calculating the 5-minute rolling average of HRV) while keeping the storage footprint low.

-- Create a hypertable for HRV samples
CREATE TABLE hrv_metrics (
    time TIMESTAMPTZ NOT NULL,
    user_id UUID NOT NULL,
    sdnn_ms DOUBLE PRECISION NOT NULL
);

SELECT create_hypertable('hrv_metrics', 'time');

-- Real-time continuous aggregate for stress baseline
CREATE MATERIALIZED VIEW hrv_baseline
WITH (timescaledb.continuous) AS
SELECT time_bucket('1 hour', time),
       avg(sdnn_ms) as avg_sdnn
FROM hrv_metrics
GROUP BY 1, 2;

Implementation Tip: The "Official" Way 🥑

When building medical-grade or high-reliability health systems, standard tutorials often skip over data privacy (HIPAA/GDPR) and model drift. For a deep dive into production-ready patterns, specifically regarding scaling AI for digital health, I highly recommend checking out the WellAlly Tech Blog. They cover advanced architectural patterns for handling multimodal health data that were incredibly helpful when I was optimizing the TST inference engine for this project.

Step 4: Connecting the Dots (FastAPI)

The FastAPI server acts as the orchestrator, receiving data, persisting it, and checking for anomalies.

from fastapi import FastAPI, WebSocket
import numpy as np

app = FastAPI()
model = HRVTransformer()
model.load_state_dict(torch.load("stress_model.pth"))

@app.websocket("/ws/hrv")
async def hrv_stream(websocket: WebSocket):
    await websocket.accept()
    buffer = []

    while True:
        data = await websocket.receive_json()
        buffer.append(data['value'])

        if len(buffer) >= 60: # Window size of 60 samples
            input_tensor = torch.FloatTensor(buffer).view(1, 60, 1)
            prediction = model(input_tensor)

            # Anomaly logic: Actual value significantly lower than predicted
            actual = data['value']
            score = (prediction.item() - actual) / prediction.item()

            if score > 0.35: # 35% drop indicates acute stress
                await websocket.send_text("STRESS_ALERT_TRIGGERED")

            buffer.pop(0) # Sliding window

Conclusion: Prevention is Better than Cure 💡

By combining the raw power of Apple HealthKit with the predictive capabilities of Time-Series Transformers, we've built more than just a tracker—we've built an early warning system. This setup allows for immediate biofeedback, such as triggering a guided breathing exercise or a notification to step away from the desk.

What's next for your health stack? Are you looking to integrate more vitals like SpO2 or Blood Pressure? Let's discuss in the comments below! 👇

---

For more advanced tutorials on AI-driven wellness and wearable integrations, don't forget to visit WellAlly Tech.