SensorFM 微调健康数据时，睡眠分期或活动识别的原始标签通常是文本事件，模型输出却是每个时间步的概率。要让训练真正跑通，先要把标签时间轴和传感器采样时间轴对齐，再按输出头选择损失函数，而不是上来就调模型。

健康数据微调的标签设计，核心是把文本事件转换为与采样频率一致的类别序列。单标签任务用整数编码并计算 class_weight，多标签任务用多热矩阵并设置 pos_weight。损失函数按输出层选择：多分类用 CrossEntropyLoss，多标签用 BCEWithLogitsLoss，极端不平衡可改用 focal loss。训练数据必须按受试者分组划分，并用按类别 F1 评估。以上配置可先用 100 条样本跑 3 个 epoch 验证流程，再结合环境确认实际效果。

将文本事件标注转换成固定时间戳的类别序列

事件清单通常记录为“开始时间、结束时间、类别”，例如睡眠分期表。传感器以固定采样率采集，比如 10Hz 或 100Hz。转换的思路是：先建立一个长度为总时长*采样率的空数组，然后遍历事件，把每个事件的时间区间填充为对应类别。注意事件区间是左闭右开，否则边界样本会重叠。

import numpy as np

# 事件表：(开始秒, 结束秒, 类别)
events = [
    (0.0, 10.0, 'wake'),
    (10.0, 30.0, 'nrem'),
    (30.0, 40.0, 'rem'),
]
fs = 1  # 1Hz，仅用于演示；实际按传感器频率设置
total_duration = 40

labels = np.empty(int(total_duration * fs), dtype=object)
for start, end, cls in events:
    s = int(start * fs)
    e = int(end * fs)
    labels[s:e] = cls

# 查看时间轴对齐后的标签分布
class_map = {'wake': 0, 'nrem': 1, 'rem': 2}
y = np.array([class_map[x] for x in labels])
print(labels)
print(np.bincount(y))  # 每类样本数，用于后续不平衡处理

如果模型输入是窗口而不是单个采样点，则用窗口起始时间对齐到标签数组，取该窗口的众数或首值作为窗口标签。窗口滑动时，每个窗口对应一个类别，避免把窗口内多个阶段混在一起。

处理多标签与类别不平衡问题

睡眠分期通常互斥，适合单标签整数编码；活动识别可能同时包含“走路+拿手机”，需要多标签多热编码。先确定任务类型，再构造标签矩阵。单标签的类别数等于分类数，多标签的每个位置代表一个类别是否存在。

# 单标签：y 已经是 [0,1,2,...] 形状 (n_samples,)
# 多标签示例：三类活动
Y_multihot = np.zeros((len(events), 3), dtype=np.float32)
# 假设每个样本有 active_classes 列表
# Y_multihot[np.arange(len(events)), active_classes] = 1

计算类别不平衡时，单标签用 compute_class_weight 生成每个类别的权重，然后传给损失函数或采样器。多标签则需要为每个标签计算正负样本不平衡比，用 pos_weight 传入 BCEWithLogitsLoss。

from sklearn.utils.class_weight import compute_class_weight

classes = np.array([0, 1, 2])
weights = compute_class_weight(class_weight='balanced', classes=classes, y=y)
class_weight_dict = dict(zip(classes, weights))
print(class_weight_dict)

在 DataLoader 里可以用 WeightedRandomSampler 按权重过采样少数类，也可以直接把 class_weight 传给 CrossEntropyLoss。先试后者，改动小且稳定。

from torch.utils.data import WeightedRandomSampler

sample_weights = np.array([class_weight_dict[label] for label in y])
sampler = WeightedRandomSampler(sample_weights, num_samples=len(sample_weights), replacement=True)
# train_loader = DataLoader(dataset, batch_size=32, sampler=sampler)

选择与任务匹配的损失函数

输出头决定了损失函数。单标签多分类用 CrossEntropyLoss，它内部会先做 softmax，所以模型输出 logits 即可。多标签分类用 BCEWithLogitsLoss，它内部带 sigmoid，也接收 logits。如果类别极不平衡且单标签，可以用 focal loss 替代 CrossEntropyLoss，让模型更关注难分的少数类。

import torch
import torch.nn as nn
import torch.nn.functional as F

class FocalLoss(nn.Module):
    def __init__(self, alpha=None, gamma=2.0):
        super().__init__()
        self.alpha = alpha
        self.gamma = gamma

    def forward(self, logits, targets):
        ce = F.cross_entropy(logits, targets, reduction='none')
        pt = torch.exp(-ce)
        focal = (1 - pt) ** self.gamma * ce
        if self.alpha is not None:
            alpha_t = self.alpha[targets]
            focal = alpha_t * focal
        return focal.mean()

验证损失函数输入输出是否正确，可以用一个 dummy batch：随机生成 logits 和 targets，分别调用 CrossEntropyLoss 和 BCEWithLogitsLoss，确认 loss 是数值而不是报错。注意 CrossEntropyLoss 的 targets 必须是整数且小于类别数，BCEWithLogitsLoss 的 targets 是浮点多热矩阵。

logits = torch.randn(8, 3)
targets = torch.tensor([0, 1, 2, 0, 1, 2, 0, 1])
loss_ce = torch.nn.CrossEntropyLoss()(logits, targets)
loss_focal = FocalLoss()(logits, targets)
print(loss_ce.item(), loss_focal.item())

配置微调训练循环与评估指标

微调过程中最容易被忽略的是数据划分。健康数据通常一个人有多条样本，直接随机划分会把同一个人的样本同时放进训练集和验证集，导致评估虚高。必须按受试者 ID 分组，整个人的数据只能进一侧。

# 假设 df 包含 subject_id 列
subjects = df['subject_id'].unique()
train_subjects, val_subjects = train_test_split(subjects, test_size=0.2, random_state=42)
train_df = df[df['subject_id'].isin(train_subjects)]
val_df = df[df['subject_id'].isin(val_subjects)]

训练循环中加上学习率调度和早停。每轮记录验证集的按类别 F1（macro average），如果连续多个 epoch 不提升就停，并恢复最佳权重。

from sklearn.metrics import f1_score, confusion_matrix
import copy

best_f1 = 0
patience = 3
counter = 0
scheduler = torch.optim.lr_scheduler.StepLR(optimizer, step_size=5, gamma=0.5)

for epoch in range(max_epochs):
    model.train()
    for x, y in train_loader:
        logits = model(x)
        loss = criterion(logits, y)
        optimizer.zero_grad()
        loss.backward()
        optimizer.step()
    scheduler.step()

    model.eval()
    all_preds, all_labels = [], []
    with torch.no_grad():
        for x, y in val_loader:
            logits = model(x)
            preds = torch.argmax(logits, dim=1).cpu().numpy()
            all_preds.extend(preds)
            all_labels.extend(y.cpu().numpy())
    val_f1 = f1_score(all_labels, all_preds, average='macro', zero_division=0)
    if val_f1 > best_f1:
        best_f1 = val_f1
        best_weights = copy.deepcopy(model.state_dict())
        counter = 0
    else:
        counter += 1
        if counter >= patience:
            break
model.load_state_dict(best_weights)

跑通一次小样本训练并检查 loss 下降

正式跑全量数据前，先抽 100 条样本跑 3 个 epoch，验证标签和损失配置没有方向性错误。记录每个 epoch 的 train_loss 和 val_loss，如果 loss 下降且没有 NaN，说明基本通路是通的。最后打印验证集混淆矩阵，看每个类别是否有预测。

# 用前100条样本
sample_df = df.iloc[:100]
# 构造 DataLoader ...
for epoch in range(3):
    train_loss = 0
    for x, y in train_loader:
        logits = model(x)
        loss = criterion(logits, y)
        train_loss += loss.item()
    val_loss = 0
    # 同样计算 val_loss
    print(f'epoch {epoch+1}: train_loss {train_loss:.4f}, val_loss {val_loss:.4f}')

# 混淆矩阵
cm = confusion_matrix(all_labels, all_preds)
print(cm)

如果 train_loss 不降，优先检查三处：标签映射的类别是否从 0 开始连续编码；模型输出的 logits 维度是否等于类别数；损失函数接收的两个参数顺序是否反了。多标签任务则检查 targets 是否为 float 类型且形状与 logits 一致。