시계열 연금술: Transformers와 PyTorch Lightning을 사용한 2시간 후 혈당 추세 예측

Source: Dev.to

포도당 예측을 위한 트랜스포머 아키텍처

graph TD
    A[InfluxDB: Raw CGM Data] --> B[Pandas: Feature Engineering]
    B --> C[Scikit-learn: Scalers & Windowing]
    C --> D[Transformer Encoder]
    D --> E[Self-Attention Layers]
    E --> F[Linear Projection Head]
    F --> G[2‑Hour Forecast: 24 Intervals]
    G --> H{Insight: Hypo/Hyper Alert}

이 모델은 마지막 6 시간 동안의 CGM 데이터(5 분 샘플링 속도로 72 포인트)를 입력으로 받아 다음 2 시간(24 포인트)을 예측합니다. 셀프‑어텐션을 통해 네트워크는 30 분 전의 조깅이든 두 시간 전의 탄수화물 풍부한 식사이든 과거 관측치를 가중치로 반영할 수 있으며, RNN/LSTM에서 흔히 발생하는 망각 문제를 피할 수 있습니다.

Required Stack

InfluxDB에서 CGM 데이터 가져오기

import pandas as pd
from influxdb_client import InfluxDBClient

def fetch_cgm_data(bucket: str, org: str, token: str, url: str) -> pd.DataFrame:
    client = InfluxDBClient(url=url, token=token, org=org)
    query = f'''
    from(bucket: "{bucket}")
      |> range(start: -7d)
      |> filter(fn: (r) => r["_measurement"] == "glucose_level")
      |> pivot(rowKey:["_time"], columnKey: ["_field"], valueColumn: "_value")
    '''
    df = client.query_api().query_data_frame(query)
    df['_time'] = pd.to_datetime(df['_time'])
    return df.set_index('_time')

# Example usage:
# df = fetch_cgm_data("health_metrics", "my_org", "SECRET_TOKEN", "http://localhost:8086")

데이터 준비

from sklearn.preprocessing import StandardScaler

# Assume the raw glucose column is named 'glucose'
scaler = StandardScaler()
df['glucose_scaled'] = scaler.fit_transform(df[['glucose']])

신호를 표준화(≈ 0 ± 1)하는 것이 필수적입니다; 원시 CGM 값은 40 ~ 400 mg/dL 범위이며 학습을 불안정하게 만들 수 있습니다.

모델 정의

import torch
import torch.nn as nn
import pytorch_lightning as pl

class GlucoseTransformer(pl.LightningModule):
    def __init__(
        self,
        input_dim: int = 1,
        model_dim: int = 64,
        n_heads: int = 4,
        n_layers: int = 3,
        output_dim: int = 24,
    ):
        super().__init__()
        self.save_hyperparameters()

        # Project raw input to model dimension
        self.input_fc = nn.Linear(input_dim, model_dim)

        # Learnable positional encoding (max seq length = 500)
        self.pos_encoder = nn.Parameter(torch.zeros(1, 500, model_dim))

        encoder_layer = nn.TransformerEncoderLayer(
            d_model=model_dim, nhead=n_heads, batch_first=True
        )
        self.transformer_encoder = nn.TransformerEncoder(
            encoder_layer, num_layers=n_layers
        )

        # Map the final hidden state to the 24‑step forecast
        self.output_fc = nn.Linear(model_dim, output_dim)
        self.loss_fn = nn.MSELoss()

    def forward(self, x):
        # x: [batch, seq_len, features]
        x = self.input_fc(x) + self.pos_encoder[:, : x.size(1), :]
        x = self.transformer_encoder(x)
        # Use the last time step's representation for prediction
        return self.output_fc(x[:, -1, :])

    def training_step(self, batch, batch_idx):
        x, y = batch
        y_hat = self(x)
        loss = self.loss_fn(y_hat, y)
        self.log("train_loss", loss, prog_bar=True)
        return loss

    def configure_optimizers(self):
        return torch.optim.Adam(self.parameters(), lr=1e-3)

훈련 설정

# Instantiate the model
model = GlucoseTransformer()

# Lightning trainer (auto‑detect GPU/CPU, single device for demo)
trainer = pl.Trainer(max_epochs=50, accelerator="auto", devices=1)

# Fit the model (replace `train_dataloader` with your DataLoader)
# trainer.fit(model, train_dataloader)

생산 고려 사항

• 센서 드리프트 및 누락 데이터 – 모델에 배치를 입력하기 전에 보간 또는 마스킹 전략을 구현합니다.
• 불확실성 추정 – 신뢰 구간을 도출하기 위해 분위수 회귀 또는 몬테카를로 드롭아웃을 고려합니다.
• 엣지 배포 – 웰얼리 테크 블로그에서 HIPAA‑준수 파이프라인 및 웨어러블 디바이스에 대한 실시간 추론을 확인하세요.

다음은?

• 멀티모달 입력 – 탄수화물 섭취량, 걸음 수, 또는 인슐린 투여량을 추가 열(input_dim > 1)로 추가합니다.
• 분위 회귀 – 하위/상위 백분위수(예: 5 % 및 95 %)를 예측하여 예측 구간을 생성합니다.
• 지속 학습 – 새로 스트리밍되는 CGM 데이터를 포함하는 재학습 일정을 설정합니다.

건강 기술 인프라 확장에 대해 더 깊이 알아보려면 WellAlly Blog (“digital health scaling” 및 “edge AI for wearables” 검색)를 확인하세요.

실험을 자유롭게 해보고, 결과를 공유하며, 능동적인 건강 모니터링의 미래를 함께 만들어가세요!