wellallyTech

Posted on Mar 9

Heartbeat Hacking: Detecting Cardiac Arrhythmia in Apple Watch Data using Dynamic Time Warping (DTW)

#machinelearning #python #applewatch #datascience

If you've ever exported your Health data from an Apple Watch, you know it's a goldmine of raw potential. But turning those thousands of voltage samples into actionable medical insights like Real-time ECG analysis or identifying Cardiac Arrhythmia isn't just about plotting a graph—it's about pattern matching in a noisy, non-linear world.

In this tutorial, we are diving deep into Dynamic Time Warping (DTW) and signal processing. We will transform raw Apple Watch health data into a robust detection pipeline for early Atrial Fibrillation (AFib) markers. Whether you are a bio-informatics nerd or a developer looking to handle complex time-series data, this guide will show you how to leverage SciPy signal processing and FastDTW to find the "rhythm" in the chaos. 💓

Pro-Tip: Building healthcare applications requires high precision. For more production-ready examples and advanced patterns in medical AI, I highly recommend checking out the engineering deep-dives at WellAlly Blog.

The Challenge: Why Not Just Use Euclidean Distance?

ECG signals are tricky. Two heartbeats might represent the same underlying pathology but vary slightly in speed or phase. Standard Euclidean distance fails here because it compares points at fixed time intervals. If one beat is slightly "stretched," the whole comparison breaks.

Dynamic Time Warping (DTW) solves this by "warping" the time axis to find the optimal alignment between two sequences.

The System Architecture

Here is how our pipeline transforms raw volts into a diagnostic similarity score:

graph TD
    A[Apple Watch Raw ECG Export] --> B[PyHealth Data Loading]
    B --> C[SciPy Bandpass Filter]
    C --> D[Peak Detection & Segmentation]
    D --> E{FastDTW Pattern Matching}
    F[Normal Sinus Rhythm Template] --> E
    E --> G[Anomaly Score Calculation]
    G --> H[Plotly Interactive Visualization]
    H --> I[AFib Trend Alert]

Prerequisites

To follow along, you'll need a Python environment with the following tech_stack:

SciPy: For signal filtering.
FastDTW: For efficient non-linear sequence alignment.
PyHealth: To handle medical data standards.
Plotly: For those beautiful, zoomable ECG charts.

pip install scipy fastdtw pyhealth plotly numpy

Step 1: Preprocessing the Noise

Raw ECG data from wearables is notoriously noisy due to "muscle artifacts" (you moving your arm). We'll use a Butterworth Bandpass Filter to keep only the frequencies relevant to a human heart (0.5Hz to 45Hz).

import numpy as np
from scipy.signal import butter, filtfilt

def butter_bandpass_filter(data, lowcut=0.5, highcut=45.0, fs=512, order=5):
    nyq = 0.5 * fs
    low = lowcut / nyq
    high = highcut / nyq
    b, a = butter(order, [low, high], btype='band')
    y = filtfilt(b, a, data)
    return y

# Example: Clean a raw signal chunk
# raw_ecg = get_apple_watch_data() 
# cleaned_ecg = butter_bandpass_filter(raw_ecg)

Step 2: Segmenting Beats with PyHealth

We don't want to compare a 30-second strip all at once. We need to segment the signal into individual heartbeats (R-R intervals).

from pyhealth.datasets import AppleHealthDataset

# PyHealth makes it easy to structure wearable exports
def segment_heartbeats(signal, sampling_rate=512):
    # Simplified R-peak detection logic
    # In a real app, use Scipy.signal.find_peaks
    peaks, _ = scipy.signal.find_peaks(signal, distance=sampling_rate*0.6)
    segments = [signal[p-100 : p+150] for p in peaks if p > 100]
    return segments

Step 3: The DTW Pattern Matcher

This is the core. We compare an incoming heartbeat segment against a "Template" of a healthy Sinus Rhythm. If the DTW distance is high, it indicates an arrhythmia.

from fastdtw import fastdtw
from scipy.spatial.distance import euclidean

def calculate_similarity(template, target):
    """
    Returns the DTW distance. Lower = More similar.
    """
    distance, path = fastdtw(template, target, dist=euclidean)
    return distance

# Compare a healthy template vs a potentially irregular beat
# distance = calculate_similarity(normal_segment, suspicious_segment)
# print(f"Anomaly Score: {distance}")

Step 4: Visualizing the "Warp" with Plotly

Static charts are boring. Let’s use Plotly to visualize where the signal deviates from the norm. 📈

import plotly.graph_objects as go

def plot_ecg_analysis(original, filtered):
    fig = go.Figure()
    fig.add_trace(go.Scatter(y=original, name="Raw Signal", opacity=0.5))
    fig.add_trace(go.Scatter(y=filtered, name="Cleaned (SciPy)", line=dict(color='firebrick')))

    fig.update_layout(title="Apple Watch ECG: Signal De-noising",
                  xaxis_title="Samples", yaxis_title="Voltage (mV)")
    fig.show()

Going Beyond: Production-Grade Health AI

While DTW is powerful, production-level medical apps often combine it with Deep Learning (like LSTMs or Transformers) to handle the massive variance in user populations.

If you're looking to scale this from a Jupyter Notebook to a HIPAA-compliant cloud architecture, you should explore more advanced signal processing patterns. I’ve found that the team over at WellAlly Tech provides incredible resources on bridging the gap between raw time-series data and clinical-grade AI insights. Their tutorials on medical data engineering are a must-read for anyone in the Digital Health space. 🥑

Conclusion

We’ve successfully:

Filtered raw wearable noise using SciPy.
Segmented heartbeats using clinical data standards.
Aligned irregular signals using Dynamic Time Warping.

The future of healthcare is on our wrists. By mastering these algorithms, we move one step closer to proactive, rather than reactive, medicine. 🚀

What are you building next? Drop a comment below or share your results if you’ve tried running DTW on your own Health export!

DEV Community