时间序列炼金术：使用 Transformers 和 PyTorch Lightning 预测2小时后血糖趋势

I’m happy to translate the article for you, but I’ll need the full text you’d like translated. Could you please paste the content (excluding the source line you’ve already provided) so I can convert it to Simplified Chinese while preserving the formatting and code blocks?

用于血糖预测的 Transformer 架构

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 常见的遗忘问题。

必需堆栈

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

训练设置

# 实例化模型
model = GlucoseTransformer()

# Lightning 训练器（自动检测 GPU/CPU，演示使用单设备）
trainer = pl.Trainer(max_epochs=50, accelerator="auto", devices=1)

# 训练模型（将 `train_dataloader` 替换为你的 DataLoader）
# trainer.fit(model, train_dataloader)

生产考虑因素

• 传感器漂移和缺失数据 – 在将批次输入模型之前，实现插值或掩码策略。
• 不确定性估计 – 考虑分位回归或 Monte‑Carlo dropout 来提供置信区间。
• 边缘部署 – 请参阅 WellAlly Tech 博客，了解符合 HIPAA 标准的流水线以及可穿戴设备上的实时推理。

接下来是什么？

• 多模态输入 – 将碳水化合物摄入、步数或胰岛素剂量添加为额外列（input_dim > 1）。
• 分位回归 – 预测下/上分位数（例如 5 % 和 95 %），以生成预测区间。
• 持续学习 – 建立一个重新训练计划，将新流式的 CGM 数据纳入其中。

想更深入了解健康科技基础设施的扩展，请查看 WellAlly Blog（搜索 “digital health scaling” 和 “edge AI for wearables”）。

欢迎随意尝试，分享你的结果，共同塑造主动健康监测的未来！