심박 해킹: Apple Watch 데이터에서 Dynamic Time Warping(DTW)으로 심장 부정맥 감지

Source: Dev.to

위에 제공된 Source 링크에 있는 전체 텍스트를 번역하려면 해당 기사 내용을 복사해서 여기 붙여 주세요.
링크 자체와 코드 블록, URL 등은 그대로 유지하고, 본문만 한국어로 번역해 드리겠습니다.

소개

Apple Watch에서 Health 데이터를 내보낸 적이 있다면, 그것이 원시 잠재력의 금광이라는 것을 알 것입니다. 수천 개의 전압 샘플을 실시간 ECG 분석이나 심장 부정맥 식별과 같은 실행 가능한 의료 인사이트로 전환하는 것은 단순히 그래프를 그리는 것이 아니라, 잡음이 많고 비선형적인 세계에서 패턴 매칭을 하는 것입니다.

이 튜토리얼에서는 **Dynamic Time Warping (DTW)**과 신호 처리에 깊이 들어갑니다. 원시 Apple Watch health data를 조기 심방세동(AFib) 마커를 위한 견고한 탐지 파이프라인으로 변환합니다. 여러분이 바이오‑인포매틱스 매니아이든 복잡한 시계열 데이터를 다루는 개발자이든, 이 가이드는 SciPy 신호 처리와 FastDTW를 활용해 혼돈 속에서 “리듬”을 찾는 방법을 보여줍니다. 💓

Pro‑Tip: 의료 애플리케이션을 구축하려면 높은 정밀도가 필요합니다. 보다 프로덕션에 적합한 예제와 의료 AI의 고급 패턴을 보려면 엔지니어링 심층 탐구를 제공하는 WellAlly Blog를 확인하세요.

도전 과제: 왜 단순히 유클리드 거리만 사용하지 않을까?

ECG 신호는 까다롭습니다. 두 심장 박동은 동일한 근본적인 병리를 나타낼 수 있지만 속도나 위상이 약간 다를 수 있습니다. 표준 유클리드 거리는 고정된 시간 간격에서 점들을 비교하기 때문에 실패합니다; 한 박동이 약간 “늘어났”다면 전체 비교가 깨집니다.

Dynamic Time Warping (DTW) 은 시간 축을 “워핑”하여 두 시퀀스 사이의 최적 정렬을 찾음으로써 이를 해결합니다.

시스템 아키텍처

우리 파이프라인이 원시 전압을 진단 유사도 점수로 변환하는 방식은 다음과 같습니다:

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]

사전 요구 사항

따라 하려면 다음 스택이 포함된 Python 환경이 필요합니다:

• SciPy – 신호 필터링
• FastDTW – 효율적인 비선형 시퀀스 정렬
• PyHealth – 의료 데이터 표준 처리
• Plotly – 인터랙티브 ECG 차트

pip install scipy fastdtw pyhealth plotly numpy

Step 1: 잡음 전처리

웨어러블에서 수집한 원시 ECG 데이터는 “근육 아티팩트”(예: 팔을 움직이는 경우) 때문에 매우 노이즈가 많습니다. 우리는 Butterworth 밴드‑pass 필터를 사용하여 인간 심장에 관련된 주파수(0.5 Hz – 45 Hz)만을 유지합니다.

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):
    """Apply a Butterworth band‑pass filter."""
    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 usage:
# raw_ecg = get_apple_watch_data()
# cleaned_ecg = butter_bandpass_filter(raw_ecg)

단계 2: PyHealth로 비트 구간 나누기

우리는 30초짜리 데이터를 한 번에 비교하고 싶지 않으며, 신호를 개별 심박(R‑R 간격)으로 구분해야 합니다.

import scipy.signal
from pyhealth.datasets import AppleHealthDataset

def segment_heartbeats(signal, sampling_rate=512):
    """
    Detect R‑peaks and extract segments around each peak.
    """
    # Simplified R‑peak detection; replace with a robust method for production.
    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: DTW 패턴 매처

이것이 핵심입니다. 들어오는 심박 세그먼트를 정상적인 동성 리듬의 “템플릿”과 비교합니다. 높은 DTW 거리는 잠재적인 부정맥을 나타냅니다.

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

# Example:
# distance = calculate_similarity(normal_segment, suspicious_segment)
# print(f"Anomaly Score: {distance}")

단계 4: Plotly로 “Warp” 시각화

정적인 차트는 지루합니다. Plotly를 사용하여 신호가 정상에서 벗어나는 부분을 시각화해봅시다. 📈

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()

한계를 넘어: 프로덕션 수준의 헬스 AI

DTW는 강력하지만, 프로덕션 수준의 의료 앱은 종종 딥러닝(예: LSTM 또는 Transformer)과 결합하여 사용자 집단 간의 큰 변동성을 처리합니다.

Jupyter 노트북에서 HIPAA 준수 클라우드 아키텍처로 확장하려면, 보다 고급 신호 처리 패턴을 탐색해 보세요.

The team at WellAlly Tech provides incredible resources on bridging 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

우리는 성공적으로:

• Filtered raw wearable noise using SciPy’s Butterworth band‑pass filter.
• Segmented the ECG into individual heartbeats with PyHealth.
• Compared each segment to a healthy template using FastDTW.
• Visualized raw vs. cleaned signals and DTW alignment with Plotly.

From here you can extend the pipeline with more sophisticated arrhythmia detection models, integrate real‑time streaming data, or deploy the solution in a secure, production environment. Happy coding—and may your heart stay in perfect rhythm!

• 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!