如果你做过可穿戴设备上的活动识别大概率遇到过这样一个矛盾手表或手环每天产生海量加速度、心率、皮肤电导数据但真正能拿去训练模型的标注数据却少得可怜。给传感器数据打标签需要人戴着设备做指定动作、记录时间戳、再逐段核对成本高、效率低还容易出错。很多人因此把项目卡在数据标注阶段而不是模型设计阶段。自监督学习正是为这个错配而生的。它不依赖人工标签直接从传感器数据本身构造学习信号先让模型理解“正常状态下的数据长什么样”再根据少量标注去适配具体任务。本文要讨论的智能可穿戴系统就是围绕“自监督预训练 多模态融合 下游任务微调”这一思路展开的以活动识别和疲劳预测两个典型任务为例给出可落地的系统设计、代码实现和工程建议。这篇文章适合正在做可穿戴应用开发、健康监测算法研究或者想了解自监督学习如何落地到传感器时序数据的读者。读完你会明白自监督学习在可穿戴场景的价值边界在哪里、多模态数据如何组织与融合、活动识别和疲劳预测代码上如何打通以及真正上线时会遇到哪些坑。1. 这篇文章真正要解决的问题先下一个明确判断可穿戴智能系统的开发瓶颈已经从“数据不够”变成了“带标签的数据不够”。传感器采集几乎没有成本设备戴在用户手上数据自然产生。真正昂贵的是让数据“可用”的环节。活动识别需要知道某段加速度波形对应走路、跑步还是上下楼这要求实验者按脚本执行动作并对齐时间戳疲劳预测更需要结合问卷、量表或反应测试来标注疲劳等级。这种标注方式放到真实项目里往往只能覆盖几十个小时的数据但下游模型动辄需要几万个样本窗口矛盾立刻显现。自监督学习切中的就是这个问题。它让模型先在海量无标注数据上做预训练学会提取有价值的时序特征然后只用少量标注数据做微调即可完成特定任务。说得直白一点以前你是“先雇人把苹果分类好再训练模型”现在改成“让模型先自己观察大量苹果再教它认识几个典型类别”。本文会围绕两条主线展开主线一是算法。从自监督学习的基本范式讲起说明它如何在可穿戴多模态数据上做预训练再迁移到活动识别和疲劳预测。主线二是工程。给出环境准备、数据组织、核心代码、运行验证和常见问题排查让读者不只是理解概念还能跑通一条最小链路。需要提前说明的是自监督学习不是银弹。它解决的是标注稀缺下的特征学习问题但它对数据增强、训练稳定性和下游适配都有要求。文章会把这些边界条件一并讲清楚。2. 基础概念与核心原理2.1 自监督学习到底在学什么自监督学习的核心思想是从数据自身构造监督信号而不是依赖人工标注。它和传统监督学习的关键区别在于监督学习学的是输入到标签的映射自监督学习学的是输入本身的结构。用一句话概括自监督学习先让模型通过一个“代理任务”理解数据的内在规律再把这个理解迁移到真实任务上。以传感器数据为例常见代理任务有掩码重建随机遮盖某段时间窗口的传感器读数让模型根据上下文预测被遮住的部分。模型为了预测准确必须学会加速度波形中隐含的运动模式。对比学习构造一组互为正样本的增强数据对让模型学习把同一来源数据映射到相近特征把不同来源数据映射到不同特征。时序预测用前一段数据预测后一段数据迫使模型学习状态转移规律。其中对比学习的应用最广。它的思想可以类比为“人脸识别中的同一个人在不同光线下的照片应该是同一身份”只不过在传感器数据里同一段动作经过加噪、缩放、时间偏移后对应的语义标签应该保持一致。2.2 三种主流自监督范式对比范式代理任务优点缺点适用场景生成式自监督掩码重建、未来预测适合细粒度特征学习训练开销较大长序列建模对比式自监督正负样本判别特征判别力强依赖数据增强设计下游分类任务联合嵌入自监督多视角特征对齐可以融合不同模态容易收敛到平凡解多模态系统可穿戴领域目前更常用对比式方法。原因是传感器数据天然适合做增强比如给加速度信号加噪声、缩放、时间扭曲语义不会改变同时活动识别和疲劳预测最终都落脚在分类或回归上对比学习学到的判别性特征可以直接复用。不过对比学习有一个常见陷阱模型退化。如果增强方式设计不合理模型可能找到一个“偷懒”的解法把所有样本映射到同一个特征向量。后面我们会讲到如何通过具体实现规避这个问题。2.3 多模态系统为什么是刚需单一的加速度信号能识别走路、跑步但在“疲劳预测”这类任务上就力不从心了。疲劳是一个涉及自主神经、肌肉活动和主观感受的综合状态仅靠运动信号很难捕捉生理层面的变化。这就需要引入多模态数据。以主流智能可穿戴系统为例通常包含以下信号模态传感器采样率信息含义运动模态IMU加速度计陀螺仪常用50Hz~128Hz肢体运动模式、步态特征心率模态PPG光学传感器常用25Hz~64Hz心率、心率变异性皮肤电模态EDA传感器常用4Hz~32Hz交感神经活跃程度温度模态皮肤温度传感器低频体温调节变化多模态系统的价值在于信息互补。运动模态告诉你“人正在做什么”心率模态告诉你“负荷有多大”皮肤电模态告诉你“应激状态有多强”。当三个模态同时指向“高强度运动 心率恢复变慢 皮肤电反应增强”时疲劳预测的置信度会明显高于单一模态。但多模态也带来代价采样率不一致、时间戳偏移、缺失值更多、特征对齐更复杂。这些问题在第5章的代码中会有具体处理方案。2.4 活动识别与疲劳预测两种不同粒度的任务文章标题把活动识别和疲劳预测放在一起容易让人以为它们是同一个任务的两种叫法。实际上它们的任务粒度和特征需求差别很大。活动识别是短时任务。走路、跑步、上下楼通常在几秒内就能判断输入窗口一般为2到5秒使用高频IMU信号即可获得较好效果。疲劳预测是长时任务。疲劳状态的评估通常需要观察分钟级甚至小时级的生理趋势比如心率变异性随时间的变化、EDA基线漂移、活动强度累积量。它的输入窗口更长更依赖低频统计特征和上下文信息。这种差异直接影响了模型设计活动识别可以用高分辨率局部特征疲劳预测则需要序列建模和特征聚合。在同一个系统里同时做两个任务比较合理的方案是“共享底层编码器、分离任务头”这也是第5章代码的主干结构。3. 系统架构与整体流程这一节先给出系统全貌方便读者在进入代码前建立整体认知。整个智能可穿戴系统分为五个模块数据采集端穿戴设备采集IMU、PPG、EDA等原始信号定时上传或本地存储。数据预处理层对多模态信号进行时间对齐、去噪、重采样、滑窗切分。自监督预训练层使用无标注数据通过对比学习训练底层编码器。下游任务适配层冻结编码器或低学习率微调分别训练活动识别头与疲劳预测头。部署推理层将模型量化为轻量版本在端侧或边缘侧完成实时预测。数据流向如下可穿戴设备原始信号 ↓ 时间对齐与重采样IMU/PPG/EDA 统一时间轴 ↓ 滑窗切分2秒窗口用于活动识别30秒窗口用于疲劳预测 ↓ 自监督预训练对比学习不需要标签 ↓ ┌──────────────┴──────────────┐ ↓ ↓ 活动识别头分类 疲劳预测头回归/分类设计时有几个值得注意的点自监督预训练建议使用所有采集到的无标注数据包括平时佩戴的日常数据而不局限于实验脚本数据。数据量越大预训练特征越通用。活动识别与疲劳预测虽然任务粒度不同但底层编码器可以是同一个。预训练阶段学到的运动节律、生理波动表征对两个任务都有帮助。推理阶段活动识别窗口短、频率高疲劳预测窗口长、频率低。系统会维护一个30秒到数分钟的滑动缓冲区按需要输出两种结果。4. 环境准备与数据说明4.1 硬件与运行环境本文的代码以Python实现使用PyTorch。如果你想直接跑通完整流程建议准备以下环境操作系统Ubuntu 20.04/22.04或Windows 10/11macOS也可以运行但训练速度会慢一些。Python版本3.8及以上建议3.9或3.10。PyTorch1.13及以上建议最新稳定版版本以实际安装环境为准。第三方库numpy、pandas、scikit-learn、tqdm。硬件如果没有GPU用小规模数据也能跑通如果数据量大建议使用单张CUDA显卡显存8GB以上即可。建议使用虚拟环境管理依赖python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install torch numpy pandas scikit-learn tqdm4.2 数据格式说明为了方便演示本文将多模态数据抽象为一个统一格式CSV文件每一行是一个采样点的多模态读数列包含时间戳和各通道数值。示例格式如下timestamp,acc_x,acc_y,acc_z,gyro_x,gyro_y,gyro_z,ppg,eda 0.000,-0.42,8.11,1.02,-0.03,0.11,0.08,0.324,1.05 0.020,-0.45,8.02,1.05,-0.02,0.12,0.07,0.326,1.06 ...其中acc表示加速度gyro表示陀螺仪ppg用于计算心率特征eda为皮肤电导。真实项目中不同传感器的采样率往往不同这里先统一为50Hz方便构建训练窗口。4.3 数据目录组织推荐按如下方式组织数据目录data/ ├── raw/ # 原始采集数据 │ ├── subject_01.csv │ ├── subject_02.csv │ └── ... ├── processed/ # 预处理后的滑窗数据 │ ├── windows_2s.npy │ └── windows_30s.npy ├── pretrain/ # 自监督预训练数据集无标注 ├── train/ # 微调用有标注数据集 │ ├── activity_labels.csv │ └── fatigue_labels.csv └── models/ # 模型权重输出 ├── encoder_pretrained.pth └── head_activity.pth5. 完整示例代码实现从这一节开始进入核心实操。代码会划分为四个文件数据加载与预处理、自监督预训练、活动识别微调、疲劳预测微调。每个文件都尽量保持可直接运行的完整结构。5.1 数据加载与滑动窗口预处理文件路径data_preprocess.py这一部分负责把原始CSV读取为滑窗样本并做多模态时间对齐。import numpy as np import pandas as pd from sklearn.preprocessing import StandardScaler SAMPLE_RATE 50 # 统一采样率单位 Hz def load_multimodal_data(csv_path): 读取多模态传感器CSV数据返回时间戳与特征矩阵。 df pd.read_csv(csv_path) timestamps df[timestamp].values # IMU PPG EDA 多模态通道 feature_cols [ acc_x, acc_y, acc_z, gyro_x, gyro_y, gyro_z, ppg, eda ] features df[feature_cols].values.astype(np.float32) return timestamps, features def standardize_channels(features, scalerNone): 对每个通道独立做标准化避免量纲差异影响训练。 if scaler is None: scaler StandardScaler() features_norm scaler.fit_transform(features) else: features_norm scaler.transform(features) return features_norm, scaler def sliding_window(features, window_size, step_size): 滑窗切分。 window_size: 窗口内采样点数如2秒窗口为100点 step_size: 滑窗步长如1秒窗口为50点 n_samples features.shape[0] windows [] start 0 while start window_size n_samples: win features[start:start window_size, :] windows.append(win) start step_size return np.stack(windows, axis0) def preprocess_pipeline(csv_path, window_seconds2.0, step_seconds1.0): 完整预处理流程读取、标准化、滑窗。 timestamps, features load_multimodal_data(csv_path) features_norm, scaler standardize_channels(features) window_size int(window_seconds * SAMPLE_RATE) step_size int(step_seconds * SAMPLE_RATE) windows sliding_window(features_norm, window_size, step_size) return windows, scaler if __name__ __main__: # 一个简单自测将单个文件处理为滑窗样本 windows, scaler preprocess_pipeline(data/raw/subject_01.csv) print(窗口形状:, windows.shape) print(窗口维度: (窗口数, 采样点数, 通道数))这段代码的关键逻辑有三处特征列的选择直接决定模型看到的模态。示例中选择了加速度、陀螺仪、PPG和EDA四个模态因此输入矩阵的通道数为8。标准化必须在所有数据上统一进行不能每个文件单独fit否则会引入不同的归一化尺度。滑窗的window_size和step_size直接决定样本数和语义粒度。活动识别用2秒窗口即可疲劳预测建议使用更长的30秒窗口。5.2 对比学习自监督预训练文件路径pretrain.py这一部分是整个系统的核心也是代码量最大的部分。下面实现一个面向时序多模态数据的对比学习框架。网络结构是“编码器 投影头”。编码器提取时序特征投影头负责把特征映射到对比学习空间。import torch import torch.nn as nn import torch.nn.functional as F import numpy as np from torch.utils.data import DataLoader, Dataset首先定义一个用于预训练的数据增强模块。时序数据的增强与图像不同需要保证物理合理性class SensorAugmentation: 多模态传感器数据的增强策略 1. 高斯噪声模拟传感器噪声 2. 幅度缩放模拟个体差异 3. 时间掩码模拟部分通道的短时缺失 def __init__(self, noise_std0.05, scale_range(0.8, 1.2), mask_ratio0.15): self.noise_std noise_std self.scale_range scale_range self.mask_ratio mask_ratio def __call__(self, x): # x: (window_size, num_channels) x x.clone() # 1. 高斯噪声 noise torch.randn_like(x) * self.noise_std x x noise # 2. 幅度缩放每个通道独立缩放 scale torch.empty(x.shape[1]).uniform_(*self.scale_range) x x * scale # 3. 时间掩码随机遮盖部分时间步的所有通道 mask_len int(self.mask_ratio * x.shape[0]) mask_start np.random.randint(0, x.shape[0] - mask_len) x[mask_start:mask_start mask_len, :] 0.0 return x然后是编码器和对比损失函数class Encoder(nn.Module): 轻量级1D-CNN编码器。 input_channels: 多模态通道数示例中为8 def __init__(self, input_channels8, hidden_dim128, output_dim64): super().__init__() self.conv1 nn.Sequential( nn.Conv1d(input_channels, 64, kernel_size7, padding3), nn.BatchNorm1d(64), nn.ReLU(), nn.MaxPool1d(2) ) self.conv2 nn.Sequential( nn.Conv1d(64, 128, kernel_size5, padding2), nn.BatchNorm1d(128), nn.ReLU(), nn.MaxPool1d(2) ) self.conv3 nn.Sequential( nn.Conv1d(128, hidden_dim, kernel_size3, padding1), nn.BatchNorm1d(hidden_dim), nn.ReLU(), ) self.gap nn.AdaptiveAvgPool1d(1) self.fc nn.Linear(hidden_dim, output_dim) def forward(self, x): # x: (batch, channels, window_size) x self.conv1(x) x self.conv2(x) x self.conv3(x) x self.gap(x).squeeze(-1) return self.fc(x) class ProjectionHead(nn.Module): 投影头将编码器特征映射到对比学习空间。 def __init__(self, input_dim64, hidden_dim32, output_dim16): super().__init__() self.net nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, output_dim) ) def forward(self, x): return F.normalize(self.net(x), dim-1)对比学习最核心的损失函数是NT-Xent。它的目的是让正样本对的相似度尽量高负样本对的相似度尽量低def nt_xent_loss(z1, z2, temperature0.1): z1, z2: (batch_size, feat_dim) 同一批样本经过两种增强得到z1和z2对角线位置是正样本对。 batch_size z1.shape[0] z torch.cat([z1, z2], dim0) # (2*batch, feat_dim) sim torch.mm(z, z.T) / temperature # 相似度矩阵 # 构造正负样本掩码 mask torch.eye(2 * batch_size, devicez.device).bool() positives torch.cat([ torch.arange(batch_size, 2 * batch_size), torch.arange(0, batch_size) ]).to(z.device) # 对角线错位位置为正样本 pos_mask torch.zeros_like(sim, dtypetorch.bool) for i in range(2 * batch_size): pos_mask[i, positives[i]] True # 计算损失 exp_sim torch.exp(sim) exp_sim.masked_fill_(mask, 0) log_prob sim - torch.log(exp_sim.sum(dim1, keepdimTrue) 1e-8) loss -log_prob[pos_mask].mean() return loss数据集和训练循环如下class UnlabeledSensorDataset(Dataset): 无标注传感器窗口数据集。 def __init__(self, windows): self.windows torch.tensor(windows, dtypetorch.float32) self.augment SensorAugmentation() def __len__(self): return len(self.windows) def __getitem__(self, idx): x self.windows[idx] x1 self.augment(x) x2 self.augment(x) return x1.permute(1, 0), x2.permute(1, 0) def train_pretrain(windows, epochs50, batch_size256, lr1e-3, devicecuda): 自监督预训练入口。 windows: (N, window_size, num_channels) dataset UnlabeledSensorDataset(windows) loader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) encoder Encoder().to(device) projector ProjectionHead().to(device) optimizer torch.optim.Adam( list(encoder.parameters()) list(projector.parameters()), lrlr ) scheduler torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_maxepochs) encoder.train() projector.train() for epoch in range(epochs): total_loss 0.0 for x1, x2 in loader: x1, x2 x1.to(device), x2.to(device) h1, h2 encoder(x1), encoder(x2) z1, z2 projector(h1), projector(h2) loss nt_xent_loss(z1, z2) optimizer.zero_grad() loss.backward() optimizer.step() total_loss loss.item() scheduler.step() if (epoch 1) % 10 0: print(fEpoch [{epoch1}/{epochs}] Loss: {total_loss/len(loader):.4f}) torch.save(encoder.state_dict(), data/models/encoder_pretrained.pth) print(预训练编码器已保存: data/models/encoder_pretrained.pth) return encoder对比学习训练时需要注意的几点批次大小会影响负样本数量。较小的batch会导致负样本不足对比学习效果下降。如果显存有限可以用梯度累积模拟大batch。学习率不宜过大否则容易导致训练不稳定和特征退化。建议使用余弦退火调度器。温度系数temperature默认取0.1这个值会影响特征的松弛程度值越小模型越关注难样本。5.3 活动识别微调活动识别是一个分类任务。我们冻结编码器主干只在顶部增加一个线性分类头这样既能验证预训练特征的质量又能避免小样本下的过拟合。文件路径finetune_activity.pyimport torch import torch.nn as nn from torch.utils.data import DataLoader, Dataset from sklearn.metrics import accuracy_score, f1_score class ActivityDataset(Dataset): 有标注活动识别数据集。 windows: (N, window_size, num_channels) labels: (N,) def __init__(self, windows, labels): self.windows torch.tensor(windows, dtypetorch.float32) self.labels torch.tensor(labels, dtypetorch.long) def __len__(self): return len(self.labels) def __getitem__(self, idx): return self.windows[idx].permute(1, 0), self.labels[idx] def linear_eval_activity(encoder, train_windows, train_labels, val_windows, val_labels, epochs30, batch_size128, lr1e-3, devicecuda): 线性评估冻结编码器只训练分类头。 这是验证自监督特征质量最常用的实验协议。 train_ds ActivityDataset(train_windows, train_labels) val_ds ActivityDataset(val_windows, val_labels) train_loader DataLoader(train_ds, batch_sizebatch_size, shuffleTrue) val_loader DataLoader(val_ds, batch_sizebatch_size, shuffleFalse) encoder.eval() encoder.to(device) for param in encoder.parameters(): param.requires_grad False num_classes len(set(train_labels.tolist())) head nn.Linear(64, num_classes).to(device) criterion nn.CrossEntropyLoss() optimizer torch.optim.Adam(head.parameters(), lrlr) for epoch in range(epochs): head.train() total_loss 0.0 for x, y in train_loader: x, y x.to(device), y.to(device) with torch.no_grad(): feat encoder(x) # 只取编码器特征 out head(feat) loss criterion(out, y) optimizer.zero_grad() loss.backward() optimizer.step() total_loss loss.item() if (epoch 1) % 10 0: print(fActivity Finetune Epoch [{epoch1}/{epochs}] Loss: {total_loss/len(train_loader):.4f}) # 验证 head.eval() all_preds, all_labels [], [] with torch.no_grad(): for x, y in val_loader: x, y x.to(device), y.to(device) feat encoder(x) out head(feat) preds torch.argmax(out, dim1) all_preds.extend(preds.cpu().numpy()) all_labels.extend(y.cpu().numpy()) acc accuracy_score(all_labels, all_preds) f1 f1_score(all_labels, all_preds, averageweighted) print(f活动识别验证准确率: {acc:.4f}, F1: {f1:.4f}) torch.save(head.state_dict(), data/models/head_activity.pth) return acc, f15.4 疲劳预测微调疲劳预测是回归任务目标是输出疲劳分数。这里采用“编码器特征 生理统计特征 回归头”的结构因为疲劳判断不只依赖短时波形还依赖心率变异性、EDA均值等统计指标。文件路径finetune_fatigue.pyimport torch import torch.nn as nn import numpy as np from torch.utils.data import DataLoader, Dataset from sklearn.metrics import mean_squared_error, mean_absolute_error首先定义从窗口计算生理统计特征的函数。因为PPG和EDA的帧率较低直接从原始窗口采样点计算统计特征是常见做法def compute_physio_features(windows): 从多模态滑窗中提取疲劳相关统计特征。 这里以ppg均值、ppg标准差、eda均值、eda标准差、加速度能量为例。 windows: (N, window_size, num_channels) 返回: (N, feat_dim) # 通道索引示例: 0-2 加速度, 3-5 陀螺仪, 6 ppg, 7 eda ppg windows[:, :, 6] eda windows[:, :, 7] acc windows[:, :, 0:3] features np.column_stack([ ppg.mean(axis1), ppg.std(axis1), eda.mean(axis1), eda.std(axis1), np.sqrt((acc ** 2).sum(axis2)).mean(axis1) # 加速度合成模长均值 ]) return features.astype(np.float32) class FatigueDataset(Dataset): 疲劳预测数据集输入多模态窗口标签为疲劳分数。 def __init__(self, windows, labels): self.windows torch.tensor(windows, dtypetorch.float32) self.labels torch.tensor(labels, dtypetorch.float32) self.physio torch.tensor(compute_physio_features(windows)) def __len__(self): return len(self.labels) def __getitem__(self, idx): return self.windows[idx], self.physio[idx], self.labels[idx] class FatigueModel(nn.Module): 编码器 统计特征 回归头的疲劳预测模型。 def __init__(self, encoder, physio_dim5, hidden_dim32): super().__init__() self.encoder encoder self.fusion nn.Sequential( nn.Linear(64 physio_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, 1) ) def forward(self, x, physio): feat self.encoder(x) # (batch, 64) combined torch.cat([feat, physio], dim1) out self.fusion(combined) return out.squeeze(-1) def train_fatigue(encoder, train_windows, train_labels, val_windows, val_labels, epochs30, batch_size128, lr1e-3, devicecuda): 训练疲劳预测模型。 与活动识别不同这里允许编码器低学习率微调 因为疲劳预测依赖更高级的生理语义特征。 train_ds FatigueDataset(train_windows, train_labels) val_ds FatigueDataset(val_windows, val_labels) train_loader DataLoader(train_ds, batch_sizebatch_size, shuffleTrue) val_loader DataLoader(val_ds, batch_sizebatch_size, shuffleFalse) model FatigueModel(encoder).to(device) criterion nn.MSELoss() optimizer torch.optim.Adam(model.parameters(), lrlr) for epoch in range(epochs): model.train() total_loss 0.0 for x, physio, y in train_loader: x, physio, y x.to(device), physio.to(device), y.to(device) pred model(x, physio) loss criterion(pred, y) optimizer.zero_grad() loss.backward() optimizer.step() total_loss loss.item() if (epoch 1) % 10 0: print(fFatigue Epoch [{epoch1}/{epochs}] Loss: {total_loss/len(train_loader):.4f}) # 验证 model.eval() all_preds, all_labels [], [] with torch.no_grad(): for x, physio, y in val_loader: x, physio, y x.to(device), physio.to(device), y.to(device) pred model(x, physio) all_preds.extend(pred.cpu().numpy()) all_labels.extend(y.cpu().numpy()) mse mean_squared_error(all_labels, all_preds) mae mean_absolute_error(all_labels, all_preds) print(f疲劳预测验证 MSE: {mse:.4f}, MAE: {mae:.4f}) torch.save(model.state_dict(), data/models/fatigue_model.pth) return mse, mae5.5 训练总入口文件路径main.py为了便于组合上面的步骤提供一个训练入口脚本把预训练和微调串联起来import os import numpy as np from data_preprocess import preprocess_pipeline from pretrain import train_pretrain from finetune_activity import linear_eval_activity from finetune_fatigue import train_fatigue RAW_DIR data/raw PROCESSED_DIR data/processed MODEL_DIR data/models def build_unlabeled_windows(): 用所有原始数据文件构建无标注滑动窗口用于自监督预训练。 all_windows [] for fname in os.listdir(RAW_DIR): if not fname.endswith(.csv): continue csv_path os.path.join(RAW_DIR, fname) windows, _ preprocess_pipeline(csv_path, window_seconds2.0) all_windows.append(windows) return np.concatenate(all_windows, axis0) def main(): os.makedirs(PROCESSED_DIR, exist_okTrue) os.makedirs(MODEL_DIR, exist_okTrue) # Step 1: 构建无标注数据并做自监督预训练 print( 构建无标注数据 ) unlabeled_windows build_unlabeled_windows() print(无标注窗口数量:, unlabeled_windows.shape[0]) print( 自监督预训练 ) encoder train_pretrain(unlabeled_windows, epochs50) # Step 2: 活动识别微调 # 实际项目中会从标注数据中读取活动标签 # 这里以随机生成的小数据演示接口调用方式 print( 活动识别微调 ) labeled_windows unlabeled_windows[:2000] fake_activity_labels np.random.randint(0, 4, sizelabeled_windows.shape[0]) linear_eval_activity(encoder, labeled_windows, fake_activity_labels, labeled_windows, fake_activity_labels) # Step 3: 疲劳预测微调 print( 疲劳预测微调 ) fatigue_labels np.random.rand(labeled_windows.shape[0]).astype(np.float32) train_fatigue(encoder, labeled_windows, fatigue_labels, labeled_windows, fatigue_labels) if __name__ __main__: main()注意main.py中的活动标签和疲劳标签是随机生成的仅用于演示接口调用方式。真实项目中你需要从activity_labels.csv和fatigue_labels.csv读取标签并按窗口的时间戳对齐。6. 运行结果与效果验证6.1 自监督预训练阶段的验证运行预训练后最直观的验证指标是对比损失contrastive loss是否持续下降。理想的训练曲线应该是前期快速下降中后期缓慢收敛。如果发现loss几乎不变或直接变为0通常说明模型出现了退化即编码器把所有输入映射到了同一特征点。以下是通过命令行运行预训练脚本的示例python main.py预期输出大致如下 构建无标注数据 无标注窗口数量: 125000 自监督预训练 Epoch [10/50] Loss: 3.2451 Epoch [20/50] Loss: 2.7612 Epoch [30/50] Loss: 2.4835 Epoch [40/50] Loss: 2.2972 Epoch [50/50] Loss: 2.18466.2 用线性评估验证特征质量活动识别的结果验证有一个行业通用协议线性评估linear probing。做法是把编码器冻结只训练一个线性分类头看分类准确率。这种方法的好处是能直接反映预训练特征的线性可分性。如果自监督预训练真的学到了好特征那么即使只用一个线性分类头也应该取得接近“端到端全监督训练”的效果。如果线性评估准确率很低则说明预训练特征质量不足不应该直接进入后续微调。在真实数据集上线性评估的准确率通常能接近有监督训练的80%到95%具体取决于数据集复杂度和样本量。如果明显偏低优先检查数据增强策略和预训练epoch数量。6.3 疲劳预测效果验证疲劳预测属于回归任务常用指标是MSE和MAE。需要注意疲劳是一个连续变量个体之间基线差异极大直接比较绝对误差意义有限。更好的方式是按照受试者划分数据集观察模型是否能在未知用户上保持稳定误差。如果MAE普遍偏高可以考虑把疲劳问题转化为“疲劳/不疲劳”的二分类问题降低任务难度先验证系统整体链路是否通畅。6.4 运行失败的优先排查顺序如果代码运行失败建议按以下顺序排查查看堆栈信息确认是否import错误重点检查PyTorch版本。检查数据路径是否存在数据文件是否可以正常读取为DataFrame。查看窗口形状确认(window_size, num_channels)维度是否与预训练代码期望一致。如果显存不足调小batch_size或window_size。7. 常见问题与排查思路问题现象可能原因排查方式解决方案预训练loss几乎不变数据增强过强或过弱负样本不足打印增强后的数据分布检查batch_size调整噪声标准差增大batch或使用梯度累积预训练loss很快变为接近0模型退化将所有样本映射到同一点检查编码器输出的特征标准差是否接近0降低学习率调整温度系数增加难负样本预训练loss下降但线性评估准确率低代理任务与下游任务语义不一致换一个预训练范式对比尝试掩码重建或加入时间预测辅助任务活动识别验证准确率低于预期滑窗窗口过短特征不足尝试不同窗口长度做对比实验将窗口从2秒扩展为3秒或5秒多模态时间轴对不齐传感器采样率不一致检查原始数据时间戳使用插值算法重采样到统一采样率疲劳预测MAE偏高标签定义不统一个体差异大按受试者分组统计误差做用户独立划分进行个体归一化微调阶段灾难性遗忘编码器参与微调后丢失通用特征对比冻结编码器和微调编码器的差异使用低学习率微调或只微调最后一两层端侧推理内存不足模型参数量过大查看模型参数总量使用深度可分离卷积替代普通卷积量化到int88. 最佳实践与工程建议8.1 数据增强是自监督学习的生命线图像领域有成熟的增强策略但传感器时序数据的增强需要结合物理含义设计。推荐的组合是高斯噪声、幅度缩放、时间掩码和时序裁剪。其中幅度缩放要考虑传感器量程时间掩码的比例不宜过高否则模型会学会直接忽略缺失通道。更强的增强不总是带来更好的效果。经验是增强强度应该与数据量匹配。数据量大时可以用更强的增强数据量小时过于激进的增强反而会让代理任务变得过于困难。8.2 按受试者划分数据避免数据泄漏可穿戴数据有一个容易被忽视的问题同一受试者的数据高度自相关。第1分钟的走路数据和第50分钟的走路数据很相似如果训练集和验证集都包含同一受试者的窗口验证集指标会虚高。正确做法是按受试者划分比如第1到第5号受试者作为训练集第6号作为验证集第7号作为测试集。这样才能评估模型在新用户上的泛化能力这也是可穿戴场景真正关心的问题。8.3 警惕时序泄漏滑窗切分时相邻窗口之间存在大量重叠。如果直接随机划分训练集和验证集同一个原始时间段的数据会出现在两边导致信息泄漏。推荐做法是先按“连续时间段”切分再分配窗口。例如先把每个受试者的数据切成不重叠的段再把段分配到不同集合。这样能更真实地模拟在线使用时的场景。8.4 系统要做多任务共享而不是多任务堆叠活动识别和疲劳预测同时存在时比较高效的做法是共享底层编码器分别训练任务头。这样有几个好处预训练只需要做一次后续两个任务都复用同一套特征。活动识别的监督信号可以通过多任务学习帮助疲劳预测反之亦然。端侧部署时只需要保留一个主干网络任务头参数很少资源开销可控。如果两个任务输入窗口长度差异过大可以改成“编码器 时间池化层 任务头”的结构让活动识别使用短窗口疲劳预测使用长窗口并做时序池化。8.5 端侧部署建议自监督预训练通常在服务器上完成但推理需要落到手表、手环或手机上。部署时有几个实用建议将训练好的PyTorch模型转换为ONNX再用推理框架部署。优先量化到int8传感器数据本身精度有限量化带来的精度损失通常可以接受。在端侧运行滑动窗口计算时使用环形缓冲区避免频繁申请内存。如果设备算力有限可以把自监督编码器蒸馏成更小的学生模型。部署前必须在测试机或真实设备上验证推理延迟和功耗不能只看服务器上的指标。8.6 数据合规与隐私边界可穿戴数据属于个人健康信息系统上线前需要明确数据采集范围、存储位置和用户的知情同意机制。具体注意点包括敏感生物信号如PPG、EDA不做必要性说明的模块不要采集。原始数据尽量做本地处理降低传输与泄露风险。如果数据需要上传应做匿名化处理并对传输链路加密。模型训练只使用获得授权的数据不能使用来源不明的公开数据集替换真实用户数据。算法评估应在获得授权的测试环境中完成更改采集逻辑后要重新走评审流程。这些不是可选的加分项而是健康类应用的基本边界。8.7 实验管理自监督学习的实验变量很多增强策略、温度系数、网络结构、预训练epoch、微调方案。如果不做实验管理很容易陷入改一组参数跑一次代码的循环。建议至少记录如下信息数据版本和划分方式。增强策略的参数配置。预训练损失和线性评估指标的曲线。微调时是否冻结编码器、学习率多少。模型文件与训练日志的对应关系。一套简单的做法是每次实验新建一个带时间戳的文件夹把配置文件、日志和模型权重都放到同一个目录。9. 总结与后续学习方向自监督学习在智能可穿戴系统里的价值不是取代监督学习而是解决“有数据、少标签”条件下如何学习的问题。本文给出的对比学习框架先用无标注的传感器窗口训练底层编码器再通过活动识别和疲劳预测两个下游任务验证特征的可复用性整体链路已经覆盖从数据处理到模型训练再到效果验证的完整流程。如果你要在这个方向继续深入建议按以下顺序学习先把本文代码在自采的小数据集上跑通理解每个模块的输入输出。再尝试替换不同的自监督范式比如掩码重建或时序预测对比特征质量。接着引入公开的可穿戴数据集做基准实验按受试者划分评估泛化性能。最后考虑多任务联合训练和端侧部署把实验室模型变成可用系统。可穿戴数据本质上是一种长尾分布的数据用户的动作模式、设备佩戴位置、传感器型号都会改变数据分布。自监督预训练在这类场景里之所以值得投入正是因为它让模型先从大量无标注的“常规状态”中学会表征再去适应少量标注的“特定任务”。这个思路不仅适用于活动识别和疲劳预测也适用于跌倒检测、压力估计、睡眠分期等更多方向。把这套框架掌握好你在可穿戴算法方向上的大多数项目都可以直接复用。 自监督学习在可穿戴多模态数据中的应用：活动识别与疲劳预测
如果你做过可穿戴设备上的活动识别大概率遇到过这样一个矛盾手表或手环每天产生海量加速度、心率、皮肤电导数据但真正能拿去训练模型的标注数据却少得可怜。给传感器数据打标签需要人戴着设备做指定动作、记录时间戳、再逐段核对成本高、效率低还容易出错。很多人因此把项目卡在数据标注阶段而不是模型设计阶段。自监督学习正是为这个错配而生的。它不依赖人工标签直接从传感器数据本身构造学习信号先让模型理解“正常状态下的数据长什么样”再根据少量标注去适配具体任务。本文要讨论的智能可穿戴系统就是围绕“自监督预训练 多模态融合 下游任务微调”这一思路展开的以活动识别和疲劳预测两个典型任务为例给出可落地的系统设计、代码实现和工程建议。这篇文章适合正在做可穿戴应用开发、健康监测算法研究或者想了解自监督学习如何落地到传感器时序数据的读者。读完你会明白自监督学习在可穿戴场景的价值边界在哪里、多模态数据如何组织与融合、活动识别和疲劳预测代码上如何打通以及真正上线时会遇到哪些坑。1. 这篇文章真正要解决的问题先下一个明确判断可穿戴智能系统的开发瓶颈已经从“数据不够”变成了“带标签的数据不够”。传感器采集几乎没有成本设备戴在用户手上数据自然产生。真正昂贵的是让数据“可用”的环节。活动识别需要知道某段加速度波形对应走路、跑步还是上下楼这要求实验者按脚本执行动作并对齐时间戳疲劳预测更需要结合问卷、量表或反应测试来标注疲劳等级。这种标注方式放到真实项目里往往只能覆盖几十个小时的数据但下游模型动辄需要几万个样本窗口矛盾立刻显现。自监督学习切中的就是这个问题。它让模型先在海量无标注数据上做预训练学会提取有价值的时序特征然后只用少量标注数据做微调即可完成特定任务。说得直白一点以前你是“先雇人把苹果分类好再训练模型”现在改成“让模型先自己观察大量苹果再教它认识几个典型类别”。本文会围绕两条主线展开主线一是算法。从自监督学习的基本范式讲起说明它如何在可穿戴多模态数据上做预训练再迁移到活动识别和疲劳预测。主线二是工程。给出环境准备、数据组织、核心代码、运行验证和常见问题排查让读者不只是理解概念还能跑通一条最小链路。需要提前说明的是自监督学习不是银弹。它解决的是标注稀缺下的特征学习问题但它对数据增强、训练稳定性和下游适配都有要求。文章会把这些边界条件一并讲清楚。2. 基础概念与核心原理2.1 自监督学习到底在学什么自监督学习的核心思想是从数据自身构造监督信号而不是依赖人工标注。它和传统监督学习的关键区别在于监督学习学的是输入到标签的映射自监督学习学的是输入本身的结构。用一句话概括自监督学习先让模型通过一个“代理任务”理解数据的内在规律再把这个理解迁移到真实任务上。以传感器数据为例常见代理任务有掩码重建随机遮盖某段时间窗口的传感器读数让模型根据上下文预测被遮住的部分。模型为了预测准确必须学会加速度波形中隐含的运动模式。对比学习构造一组互为正样本的增强数据对让模型学习把同一来源数据映射到相近特征把不同来源数据映射到不同特征。时序预测用前一段数据预测后一段数据迫使模型学习状态转移规律。其中对比学习的应用最广。它的思想可以类比为“人脸识别中的同一个人在不同光线下的照片应该是同一身份”只不过在传感器数据里同一段动作经过加噪、缩放、时间偏移后对应的语义标签应该保持一致。2.2 三种主流自监督范式对比范式代理任务优点缺点适用场景生成式自监督掩码重建、未来预测适合细粒度特征学习训练开销较大长序列建模对比式自监督正负样本判别特征判别力强依赖数据增强设计下游分类任务联合嵌入自监督多视角特征对齐可以融合不同模态容易收敛到平凡解多模态系统可穿戴领域目前更常用对比式方法。原因是传感器数据天然适合做增强比如给加速度信号加噪声、缩放、时间扭曲语义不会改变同时活动识别和疲劳预测最终都落脚在分类或回归上对比学习学到的判别性特征可以直接复用。不过对比学习有一个常见陷阱模型退化。如果增强方式设计不合理模型可能找到一个“偷懒”的解法把所有样本映射到同一个特征向量。后面我们会讲到如何通过具体实现规避这个问题。2.3 多模态系统为什么是刚需单一的加速度信号能识别走路、跑步但在“疲劳预测”这类任务上就力不从心了。疲劳是一个涉及自主神经、肌肉活动和主观感受的综合状态仅靠运动信号很难捕捉生理层面的变化。这就需要引入多模态数据。以主流智能可穿戴系统为例通常包含以下信号模态传感器采样率信息含义运动模态IMU加速度计陀螺仪常用50Hz~128Hz肢体运动模式、步态特征心率模态PPG光学传感器常用25Hz~64Hz心率、心率变异性皮肤电模态EDA传感器常用4Hz~32Hz交感神经活跃程度温度模态皮肤温度传感器低频体温调节变化多模态系统的价值在于信息互补。运动模态告诉你“人正在做什么”心率模态告诉你“负荷有多大”皮肤电模态告诉你“应激状态有多强”。当三个模态同时指向“高强度运动 心率恢复变慢 皮肤电反应增强”时疲劳预测的置信度会明显高于单一模态。但多模态也带来代价采样率不一致、时间戳偏移、缺失值更多、特征对齐更复杂。这些问题在第5章的代码中会有具体处理方案。2.4 活动识别与疲劳预测两种不同粒度的任务文章标题把活动识别和疲劳预测放在一起容易让人以为它们是同一个任务的两种叫法。实际上它们的任务粒度和特征需求差别很大。活动识别是短时任务。走路、跑步、上下楼通常在几秒内就能判断输入窗口一般为2到5秒使用高频IMU信号即可获得较好效果。疲劳预测是长时任务。疲劳状态的评估通常需要观察分钟级甚至小时级的生理趋势比如心率变异性随时间的变化、EDA基线漂移、活动强度累积量。它的输入窗口更长更依赖低频统计特征和上下文信息。这种差异直接影响了模型设计活动识别可以用高分辨率局部特征疲劳预测则需要序列建模和特征聚合。在同一个系统里同时做两个任务比较合理的方案是“共享底层编码器、分离任务头”这也是第5章代码的主干结构。3. 系统架构与整体流程这一节先给出系统全貌方便读者在进入代码前建立整体认知。整个智能可穿戴系统分为五个模块数据采集端穿戴设备采集IMU、PPG、EDA等原始信号定时上传或本地存储。数据预处理层对多模态信号进行时间对齐、去噪、重采样、滑窗切分。自监督预训练层使用无标注数据通过对比学习训练底层编码器。下游任务适配层冻结编码器或低学习率微调分别训练活动识别头与疲劳预测头。部署推理层将模型量化为轻量版本在端侧或边缘侧完成实时预测。数据流向如下可穿戴设备原始信号 ↓ 时间对齐与重采样IMU/PPG/EDA 统一时间轴 ↓ 滑窗切分2秒窗口用于活动识别30秒窗口用于疲劳预测 ↓ 自监督预训练对比学习不需要标签 ↓ ┌──────────────┴──────────────┐ ↓ ↓ 活动识别头分类 疲劳预测头回归/分类设计时有几个值得注意的点自监督预训练建议使用所有采集到的无标注数据包括平时佩戴的日常数据而不局限于实验脚本数据。数据量越大预训练特征越通用。活动识别与疲劳预测虽然任务粒度不同但底层编码器可以是同一个。预训练阶段学到的运动节律、生理波动表征对两个任务都有帮助。推理阶段活动识别窗口短、频率高疲劳预测窗口长、频率低。系统会维护一个30秒到数分钟的滑动缓冲区按需要输出两种结果。4. 环境准备与数据说明4.1 硬件与运行环境本文的代码以Python实现使用PyTorch。如果你想直接跑通完整流程建议准备以下环境操作系统Ubuntu 20.04/22.04或Windows 10/11macOS也可以运行但训练速度会慢一些。Python版本3.8及以上建议3.9或3.10。PyTorch1.13及以上建议最新稳定版版本以实际安装环境为准。第三方库numpy、pandas、scikit-learn、tqdm。硬件如果没有GPU用小规模数据也能跑通如果数据量大建议使用单张CUDA显卡显存8GB以上即可。建议使用虚拟环境管理依赖python -m venv .venv source .venv/bin/activate # Windows 下使用 .venv\Scripts\activate pip install torch numpy pandas scikit-learn tqdm4.2 数据格式说明为了方便演示本文将多模态数据抽象为一个统一格式CSV文件每一行是一个采样点的多模态读数列包含时间戳和各通道数值。示例格式如下timestamp,acc_x,acc_y,acc_z,gyro_x,gyro_y,gyro_z,ppg,eda 0.000,-0.42,8.11,1.02,-0.03,0.11,0.08,0.324,1.05 0.020,-0.45,8.02,1.05,-0.02,0.12,0.07,0.326,1.06 ...其中acc表示加速度gyro表示陀螺仪ppg用于计算心率特征eda为皮肤电导。真实项目中不同传感器的采样率往往不同这里先统一为50Hz方便构建训练窗口。4.3 数据目录组织推荐按如下方式组织数据目录data/ ├── raw/ # 原始采集数据 │ ├── subject_01.csv │ ├── subject_02.csv │ └── ... ├── processed/ # 预处理后的滑窗数据 │ ├── windows_2s.npy │ └── windows_30s.npy ├── pretrain/ # 自监督预训练数据集无标注 ├── train/ # 微调用有标注数据集 │ ├── activity_labels.csv │ └── fatigue_labels.csv └── models/ # 模型权重输出 ├── encoder_pretrained.pth └── head_activity.pth5. 完整示例代码实现从这一节开始进入核心实操。代码会划分为四个文件数据加载与预处理、自监督预训练、活动识别微调、疲劳预测微调。每个文件都尽量保持可直接运行的完整结构。5.1 数据加载与滑动窗口预处理文件路径data_preprocess.py这一部分负责把原始CSV读取为滑窗样本并做多模态时间对齐。import numpy as np import pandas as pd from sklearn.preprocessing import StandardScaler SAMPLE_RATE 50 # 统一采样率单位 Hz def load_multimodal_data(csv_path): 读取多模态传感器CSV数据返回时间戳与特征矩阵。 df pd.read_csv(csv_path) timestamps df[timestamp].values # IMU PPG EDA 多模态通道 feature_cols [ acc_x, acc_y, acc_z, gyro_x, gyro_y, gyro_z, ppg, eda ] features df[feature_cols].values.astype(np.float32) return timestamps, features def standardize_channels(features, scalerNone): 对每个通道独立做标准化避免量纲差异影响训练。 if scaler is None: scaler StandardScaler() features_norm scaler.fit_transform(features) else: features_norm scaler.transform(features) return features_norm, scaler def sliding_window(features, window_size, step_size): 滑窗切分。 window_size: 窗口内采样点数如2秒窗口为100点 step_size: 滑窗步长如1秒窗口为50点 n_samples features.shape[0] windows [] start 0 while start window_size n_samples: win features[start:start window_size, :] windows.append(win) start step_size return np.stack(windows, axis0) def preprocess_pipeline(csv_path, window_seconds2.0, step_seconds1.0): 完整预处理流程读取、标准化、滑窗。 timestamps, features load_multimodal_data(csv_path) features_norm, scaler standardize_channels(features) window_size int(window_seconds * SAMPLE_RATE) step_size int(step_seconds * SAMPLE_RATE) windows sliding_window(features_norm, window_size, step_size) return windows, scaler if __name__ __main__: # 一个简单自测将单个文件处理为滑窗样本 windows, scaler preprocess_pipeline(data/raw/subject_01.csv) print(窗口形状:, windows.shape) print(窗口维度: (窗口数, 采样点数, 通道数))这段代码的关键逻辑有三处特征列的选择直接决定模型看到的模态。示例中选择了加速度、陀螺仪、PPG和EDA四个模态因此输入矩阵的通道数为8。标准化必须在所有数据上统一进行不能每个文件单独fit否则会引入不同的归一化尺度。滑窗的window_size和step_size直接决定样本数和语义粒度。活动识别用2秒窗口即可疲劳预测建议使用更长的30秒窗口。5.2 对比学习自监督预训练文件路径pretrain.py这一部分是整个系统的核心也是代码量最大的部分。下面实现一个面向时序多模态数据的对比学习框架。网络结构是“编码器 投影头”。编码器提取时序特征投影头负责把特征映射到对比学习空间。import torch import torch.nn as nn import torch.nn.functional as F import numpy as np from torch.utils.data import DataLoader, Dataset首先定义一个用于预训练的数据增强模块。时序数据的增强与图像不同需要保证物理合理性class SensorAugmentation: 多模态传感器数据的增强策略 1. 高斯噪声模拟传感器噪声 2. 幅度缩放模拟个体差异 3. 时间掩码模拟部分通道的短时缺失 def __init__(self, noise_std0.05, scale_range(0.8, 1.2), mask_ratio0.15): self.noise_std noise_std self.scale_range scale_range self.mask_ratio mask_ratio def __call__(self, x): # x: (window_size, num_channels) x x.clone() # 1. 高斯噪声 noise torch.randn_like(x) * self.noise_std x x noise # 2. 幅度缩放每个通道独立缩放 scale torch.empty(x.shape[1]).uniform_(*self.scale_range) x x * scale # 3. 时间掩码随机遮盖部分时间步的所有通道 mask_len int(self.mask_ratio * x.shape[0]) mask_start np.random.randint(0, x.shape[0] - mask_len) x[mask_start:mask_start mask_len, :] 0.0 return x然后是编码器和对比损失函数class Encoder(nn.Module): 轻量级1D-CNN编码器。 input_channels: 多模态通道数示例中为8 def __init__(self, input_channels8, hidden_dim128, output_dim64): super().__init__() self.conv1 nn.Sequential( nn.Conv1d(input_channels, 64, kernel_size7, padding3), nn.BatchNorm1d(64), nn.ReLU(), nn.MaxPool1d(2) ) self.conv2 nn.Sequential( nn.Conv1d(64, 128, kernel_size5, padding2), nn.BatchNorm1d(128), nn.ReLU(), nn.MaxPool1d(2) ) self.conv3 nn.Sequential( nn.Conv1d(128, hidden_dim, kernel_size3, padding1), nn.BatchNorm1d(hidden_dim), nn.ReLU(), ) self.gap nn.AdaptiveAvgPool1d(1) self.fc nn.Linear(hidden_dim, output_dim) def forward(self, x): # x: (batch, channels, window_size) x self.conv1(x) x self.conv2(x) x self.conv3(x) x self.gap(x).squeeze(-1) return self.fc(x) class ProjectionHead(nn.Module): 投影头将编码器特征映射到对比学习空间。 def __init__(self, input_dim64, hidden_dim32, output_dim16): super().__init__() self.net nn.Sequential( nn.Linear(input_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, output_dim) ) def forward(self, x): return F.normalize(self.net(x), dim-1)对比学习最核心的损失函数是NT-Xent。它的目的是让正样本对的相似度尽量高负样本对的相似度尽量低def nt_xent_loss(z1, z2, temperature0.1): z1, z2: (batch_size, feat_dim) 同一批样本经过两种增强得到z1和z2对角线位置是正样本对。 batch_size z1.shape[0] z torch.cat([z1, z2], dim0) # (2*batch, feat_dim) sim torch.mm(z, z.T) / temperature # 相似度矩阵 # 构造正负样本掩码 mask torch.eye(2 * batch_size, devicez.device).bool() positives torch.cat([ torch.arange(batch_size, 2 * batch_size), torch.arange(0, batch_size) ]).to(z.device) # 对角线错位位置为正样本 pos_mask torch.zeros_like(sim, dtypetorch.bool) for i in range(2 * batch_size): pos_mask[i, positives[i]] True # 计算损失 exp_sim torch.exp(sim) exp_sim.masked_fill_(mask, 0) log_prob sim - torch.log(exp_sim.sum(dim1, keepdimTrue) 1e-8) loss -log_prob[pos_mask].mean() return loss数据集和训练循环如下class UnlabeledSensorDataset(Dataset): 无标注传感器窗口数据集。 def __init__(self, windows): self.windows torch.tensor(windows, dtypetorch.float32) self.augment SensorAugmentation() def __len__(self): return len(self.windows) def __getitem__(self, idx): x self.windows[idx] x1 self.augment(x) x2 self.augment(x) return x1.permute(1, 0), x2.permute(1, 0) def train_pretrain(windows, epochs50, batch_size256, lr1e-3, devicecuda): 自监督预训练入口。 windows: (N, window_size, num_channels) dataset UnlabeledSensorDataset(windows) loader DataLoader(dataset, batch_sizebatch_size, shuffleTrue) encoder Encoder().to(device) projector ProjectionHead().to(device) optimizer torch.optim.Adam( list(encoder.parameters()) list(projector.parameters()), lrlr ) scheduler torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_maxepochs) encoder.train() projector.train() for epoch in range(epochs): total_loss 0.0 for x1, x2 in loader: x1, x2 x1.to(device), x2.to(device) h1, h2 encoder(x1), encoder(x2) z1, z2 projector(h1), projector(h2) loss nt_xent_loss(z1, z2) optimizer.zero_grad() loss.backward() optimizer.step() total_loss loss.item() scheduler.step() if (epoch 1) % 10 0: print(fEpoch [{epoch1}/{epochs}] Loss: {total_loss/len(loader):.4f}) torch.save(encoder.state_dict(), data/models/encoder_pretrained.pth) print(预训练编码器已保存: data/models/encoder_pretrained.pth) return encoder对比学习训练时需要注意的几点批次大小会影响负样本数量。较小的batch会导致负样本不足对比学习效果下降。如果显存有限可以用梯度累积模拟大batch。学习率不宜过大否则容易导致训练不稳定和特征退化。建议使用余弦退火调度器。温度系数temperature默认取0.1这个值会影响特征的松弛程度值越小模型越关注难样本。5.3 活动识别微调活动识别是一个分类任务。我们冻结编码器主干只在顶部增加一个线性分类头这样既能验证预训练特征的质量又能避免小样本下的过拟合。文件路径finetune_activity.pyimport torch import torch.nn as nn from torch.utils.data import DataLoader, Dataset from sklearn.metrics import accuracy_score, f1_score class ActivityDataset(Dataset): 有标注活动识别数据集。 windows: (N, window_size, num_channels) labels: (N,) def __init__(self, windows, labels): self.windows torch.tensor(windows, dtypetorch.float32) self.labels torch.tensor(labels, dtypetorch.long) def __len__(self): return len(self.labels) def __getitem__(self, idx): return self.windows[idx].permute(1, 0), self.labels[idx] def linear_eval_activity(encoder, train_windows, train_labels, val_windows, val_labels, epochs30, batch_size128, lr1e-3, devicecuda): 线性评估冻结编码器只训练分类头。 这是验证自监督特征质量最常用的实验协议。 train_ds ActivityDataset(train_windows, train_labels) val_ds ActivityDataset(val_windows, val_labels) train_loader DataLoader(train_ds, batch_sizebatch_size, shuffleTrue) val_loader DataLoader(val_ds, batch_sizebatch_size, shuffleFalse) encoder.eval() encoder.to(device) for param in encoder.parameters(): param.requires_grad False num_classes len(set(train_labels.tolist())) head nn.Linear(64, num_classes).to(device) criterion nn.CrossEntropyLoss() optimizer torch.optim.Adam(head.parameters(), lrlr) for epoch in range(epochs): head.train() total_loss 0.0 for x, y in train_loader: x, y x.to(device), y.to(device) with torch.no_grad(): feat encoder(x) # 只取编码器特征 out head(feat) loss criterion(out, y) optimizer.zero_grad() loss.backward() optimizer.step() total_loss loss.item() if (epoch 1) % 10 0: print(fActivity Finetune Epoch [{epoch1}/{epochs}] Loss: {total_loss/len(train_loader):.4f}) # 验证 head.eval() all_preds, all_labels [], [] with torch.no_grad(): for x, y in val_loader: x, y x.to(device), y.to(device) feat encoder(x) out head(feat) preds torch.argmax(out, dim1) all_preds.extend(preds.cpu().numpy()) all_labels.extend(y.cpu().numpy()) acc accuracy_score(all_labels, all_preds) f1 f1_score(all_labels, all_preds, averageweighted) print(f活动识别验证准确率: {acc:.4f}, F1: {f1:.4f}) torch.save(head.state_dict(), data/models/head_activity.pth) return acc, f15.4 疲劳预测微调疲劳预测是回归任务目标是输出疲劳分数。这里采用“编码器特征 生理统计特征 回归头”的结构因为疲劳判断不只依赖短时波形还依赖心率变异性、EDA均值等统计指标。文件路径finetune_fatigue.pyimport torch import torch.nn as nn import numpy as np from torch.utils.data import DataLoader, Dataset from sklearn.metrics import mean_squared_error, mean_absolute_error首先定义从窗口计算生理统计特征的函数。因为PPG和EDA的帧率较低直接从原始窗口采样点计算统计特征是常见做法def compute_physio_features(windows): 从多模态滑窗中提取疲劳相关统计特征。 这里以ppg均值、ppg标准差、eda均值、eda标准差、加速度能量为例。 windows: (N, window_size, num_channels) 返回: (N, feat_dim) # 通道索引示例: 0-2 加速度, 3-5 陀螺仪, 6 ppg, 7 eda ppg windows[:, :, 6] eda windows[:, :, 7] acc windows[:, :, 0:3] features np.column_stack([ ppg.mean(axis1), ppg.std(axis1), eda.mean(axis1), eda.std(axis1), np.sqrt((acc ** 2).sum(axis2)).mean(axis1) # 加速度合成模长均值 ]) return features.astype(np.float32) class FatigueDataset(Dataset): 疲劳预测数据集输入多模态窗口标签为疲劳分数。 def __init__(self, windows, labels): self.windows torch.tensor(windows, dtypetorch.float32) self.labels torch.tensor(labels, dtypetorch.float32) self.physio torch.tensor(compute_physio_features(windows)) def __len__(self): return len(self.labels) def __getitem__(self, idx): return self.windows[idx], self.physio[idx], self.labels[idx] class FatigueModel(nn.Module): 编码器 统计特征 回归头的疲劳预测模型。 def __init__(self, encoder, physio_dim5, hidden_dim32): super().__init__() self.encoder encoder self.fusion nn.Sequential( nn.Linear(64 physio_dim, hidden_dim), nn.ReLU(), nn.Linear(hidden_dim, 1) ) def forward(self, x, physio): feat self.encoder(x) # (batch, 64) combined torch.cat([feat, physio], dim1) out self.fusion(combined) return out.squeeze(-1) def train_fatigue(encoder, train_windows, train_labels, val_windows, val_labels, epochs30, batch_size128, lr1e-3, devicecuda): 训练疲劳预测模型。 与活动识别不同这里允许编码器低学习率微调 因为疲劳预测依赖更高级的生理语义特征。 train_ds FatigueDataset(train_windows, train_labels) val_ds FatigueDataset(val_windows, val_labels) train_loader DataLoader(train_ds, batch_sizebatch_size, shuffleTrue) val_loader DataLoader(val_ds, batch_sizebatch_size, shuffleFalse) model FatigueModel(encoder).to(device) criterion nn.MSELoss() optimizer torch.optim.Adam(model.parameters(), lrlr) for epoch in range(epochs): model.train() total_loss 0.0 for x, physio, y in train_loader: x, physio, y x.to(device), physio.to(device), y.to(device) pred model(x, physio) loss criterion(pred, y) optimizer.zero_grad() loss.backward() optimizer.step() total_loss loss.item() if (epoch 1) % 10 0: print(fFatigue Epoch [{epoch1}/{epochs}] Loss: {total_loss/len(train_loader):.4f}) # 验证 model.eval() all_preds, all_labels [], [] with torch.no_grad(): for x, physio, y in val_loader: x, physio, y x.to(device), physio.to(device), y.to(device) pred model(x, physio) all_preds.extend(pred.cpu().numpy()) all_labels.extend(y.cpu().numpy()) mse mean_squared_error(all_labels, all_preds) mae mean_absolute_error(all_labels, all_preds) print(f疲劳预测验证 MSE: {mse:.4f}, MAE: {mae:.4f}) torch.save(model.state_dict(), data/models/fatigue_model.pth) return mse, mae5.5 训练总入口文件路径main.py为了便于组合上面的步骤提供一个训练入口脚本把预训练和微调串联起来import os import numpy as np from data_preprocess import preprocess_pipeline from pretrain import train_pretrain from finetune_activity import linear_eval_activity from finetune_fatigue import train_fatigue RAW_DIR data/raw PROCESSED_DIR data/processed MODEL_DIR data/models def build_unlabeled_windows(): 用所有原始数据文件构建无标注滑动窗口用于自监督预训练。 all_windows [] for fname in os.listdir(RAW_DIR): if not fname.endswith(.csv): continue csv_path os.path.join(RAW_DIR, fname) windows, _ preprocess_pipeline(csv_path, window_seconds2.0) all_windows.append(windows) return np.concatenate(all_windows, axis0) def main(): os.makedirs(PROCESSED_DIR, exist_okTrue) os.makedirs(MODEL_DIR, exist_okTrue) # Step 1: 构建无标注数据并做自监督预训练 print( 构建无标注数据 ) unlabeled_windows build_unlabeled_windows() print(无标注窗口数量:, unlabeled_windows.shape[0]) print( 自监督预训练 ) encoder train_pretrain(unlabeled_windows, epochs50) # Step 2: 活动识别微调 # 实际项目中会从标注数据中读取活动标签 # 这里以随机生成的小数据演示接口调用方式 print( 活动识别微调 ) labeled_windows unlabeled_windows[:2000] fake_activity_labels np.random.randint(0, 4, sizelabeled_windows.shape[0]) linear_eval_activity(encoder, labeled_windows, fake_activity_labels, labeled_windows, fake_activity_labels) # Step 3: 疲劳预测微调 print( 疲劳预测微调 ) fatigue_labels np.random.rand(labeled_windows.shape[0]).astype(np.float32) train_fatigue(encoder, labeled_windows, fatigue_labels, labeled_windows, fatigue_labels) if __name__ __main__: main()注意main.py中的活动标签和疲劳标签是随机生成的仅用于演示接口调用方式。真实项目中你需要从activity_labels.csv和fatigue_labels.csv读取标签并按窗口的时间戳对齐。6. 运行结果与效果验证6.1 自监督预训练阶段的验证运行预训练后最直观的验证指标是对比损失contrastive loss是否持续下降。理想的训练曲线应该是前期快速下降中后期缓慢收敛。如果发现loss几乎不变或直接变为0通常说明模型出现了退化即编码器把所有输入映射到了同一特征点。以下是通过命令行运行预训练脚本的示例python main.py预期输出大致如下 构建无标注数据 无标注窗口数量: 125000 自监督预训练 Epoch [10/50] Loss: 3.2451 Epoch [20/50] Loss: 2.7612 Epoch [30/50] Loss: 2.4835 Epoch [40/50] Loss: 2.2972 Epoch [50/50] Loss: 2.18466.2 用线性评估验证特征质量活动识别的结果验证有一个行业通用协议线性评估linear probing。做法是把编码器冻结只训练一个线性分类头看分类准确率。这种方法的好处是能直接反映预训练特征的线性可分性。如果自监督预训练真的学到了好特征那么即使只用一个线性分类头也应该取得接近“端到端全监督训练”的效果。如果线性评估准确率很低则说明预训练特征质量不足不应该直接进入后续微调。在真实数据集上线性评估的准确率通常能接近有监督训练的80%到95%具体取决于数据集复杂度和样本量。如果明显偏低优先检查数据增强策略和预训练epoch数量。6.3 疲劳预测效果验证疲劳预测属于回归任务常用指标是MSE和MAE。需要注意疲劳是一个连续变量个体之间基线差异极大直接比较绝对误差意义有限。更好的方式是按照受试者划分数据集观察模型是否能在未知用户上保持稳定误差。如果MAE普遍偏高可以考虑把疲劳问题转化为“疲劳/不疲劳”的二分类问题降低任务难度先验证系统整体链路是否通畅。6.4 运行失败的优先排查顺序如果代码运行失败建议按以下顺序排查查看堆栈信息确认是否import错误重点检查PyTorch版本。检查数据路径是否存在数据文件是否可以正常读取为DataFrame。查看窗口形状确认(window_size, num_channels)维度是否与预训练代码期望一致。如果显存不足调小batch_size或window_size。7. 常见问题与排查思路问题现象可能原因排查方式解决方案预训练loss几乎不变数据增强过强或过弱负样本不足打印增强后的数据分布检查batch_size调整噪声标准差增大batch或使用梯度累积预训练loss很快变为接近0模型退化将所有样本映射到同一点检查编码器输出的特征标准差是否接近0降低学习率调整温度系数增加难负样本预训练loss下降但线性评估准确率低代理任务与下游任务语义不一致换一个预训练范式对比尝试掩码重建或加入时间预测辅助任务活动识别验证准确率低于预期滑窗窗口过短特征不足尝试不同窗口长度做对比实验将窗口从2秒扩展为3秒或5秒多模态时间轴对不齐传感器采样率不一致检查原始数据时间戳使用插值算法重采样到统一采样率疲劳预测MAE偏高标签定义不统一个体差异大按受试者分组统计误差做用户独立划分进行个体归一化微调阶段灾难性遗忘编码器参与微调后丢失通用特征对比冻结编码器和微调编码器的差异使用低学习率微调或只微调最后一两层端侧推理内存不足模型参数量过大查看模型参数总量使用深度可分离卷积替代普通卷积量化到int88. 最佳实践与工程建议8.1 数据增强是自监督学习的生命线图像领域有成熟的增强策略但传感器时序数据的增强需要结合物理含义设计。推荐的组合是高斯噪声、幅度缩放、时间掩码和时序裁剪。其中幅度缩放要考虑传感器量程时间掩码的比例不宜过高否则模型会学会直接忽略缺失通道。更强的增强不总是带来更好的效果。经验是增强强度应该与数据量匹配。数据量大时可以用更强的增强数据量小时过于激进的增强反而会让代理任务变得过于困难。8.2 按受试者划分数据避免数据泄漏可穿戴数据有一个容易被忽视的问题同一受试者的数据高度自相关。第1分钟的走路数据和第50分钟的走路数据很相似如果训练集和验证集都包含同一受试者的窗口验证集指标会虚高。正确做法是按受试者划分比如第1到第5号受试者作为训练集第6号作为验证集第7号作为测试集。这样才能评估模型在新用户上的泛化能力这也是可穿戴场景真正关心的问题。8.3 警惕时序泄漏滑窗切分时相邻窗口之间存在大量重叠。如果直接随机划分训练集和验证集同一个原始时间段的数据会出现在两边导致信息泄漏。推荐做法是先按“连续时间段”切分再分配窗口。例如先把每个受试者的数据切成不重叠的段再把段分配到不同集合。这样能更真实地模拟在线使用时的场景。8.4 系统要做多任务共享而不是多任务堆叠活动识别和疲劳预测同时存在时比较高效的做法是共享底层编码器分别训练任务头。这样有几个好处预训练只需要做一次后续两个任务都复用同一套特征。活动识别的监督信号可以通过多任务学习帮助疲劳预测反之亦然。端侧部署时只需要保留一个主干网络任务头参数很少资源开销可控。如果两个任务输入窗口长度差异过大可以改成“编码器 时间池化层 任务头”的结构让活动识别使用短窗口疲劳预测使用长窗口并做时序池化。8.5 端侧部署建议自监督预训练通常在服务器上完成但推理需要落到手表、手环或手机上。部署时有几个实用建议将训练好的PyTorch模型转换为ONNX再用推理框架部署。优先量化到int8传感器数据本身精度有限量化带来的精度损失通常可以接受。在端侧运行滑动窗口计算时使用环形缓冲区避免频繁申请内存。如果设备算力有限可以把自监督编码器蒸馏成更小的学生模型。部署前必须在测试机或真实设备上验证推理延迟和功耗不能只看服务器上的指标。8.6 数据合规与隐私边界可穿戴数据属于个人健康信息系统上线前需要明确数据采集范围、存储位置和用户的知情同意机制。具体注意点包括敏感生物信号如PPG、EDA不做必要性说明的模块不要采集。原始数据尽量做本地处理降低传输与泄露风险。如果数据需要上传应做匿名化处理并对传输链路加密。模型训练只使用获得授权的数据不能使用来源不明的公开数据集替换真实用户数据。算法评估应在获得授权的测试环境中完成更改采集逻辑后要重新走评审流程。这些不是可选的加分项而是健康类应用的基本边界。8.7 实验管理自监督学习的实验变量很多增强策略、温度系数、网络结构、预训练epoch、微调方案。如果不做实验管理很容易陷入改一组参数跑一次代码的循环。建议至少记录如下信息数据版本和划分方式。增强策略的参数配置。预训练损失和线性评估指标的曲线。微调时是否冻结编码器、学习率多少。模型文件与训练日志的对应关系。一套简单的做法是每次实验新建一个带时间戳的文件夹把配置文件、日志和模型权重都放到同一个目录。9. 总结与后续学习方向自监督学习在智能可穿戴系统里的价值不是取代监督学习而是解决“有数据、少标签”条件下如何学习的问题。本文给出的对比学习框架先用无标注的传感器窗口训练底层编码器再通过活动识别和疲劳预测两个下游任务验证特征的可复用性整体链路已经覆盖从数据处理到模型训练再到效果验证的完整流程。如果你要在这个方向继续深入建议按以下顺序学习先把本文代码在自采的小数据集上跑通理解每个模块的输入输出。再尝试替换不同的自监督范式比如掩码重建或时序预测对比特征质量。接着引入公开的可穿戴数据集做基准实验按受试者划分评估泛化性能。最后考虑多任务联合训练和端侧部署把实验室模型变成可用系统。可穿戴数据本质上是一种长尾分布的数据用户的动作模式、设备佩戴位置、传感器型号都会改变数据分布。自监督预训练在这类场景里之所以值得投入正是因为它让模型先从大量无标注的“常规状态”中学会表征再去适应少量标注的“特定任务”。这个思路不仅适用于活动识别和疲劳预测也适用于跌倒检测、压力估计、睡眠分期等更多方向。把这套框架掌握好你在可穿戴算法方向上的大多数项目都可以直接复用。