企业官网建设流程全解析

如果你做过可穿戴设备上的活动识别，大概率遇到过这样一个矛盾：手表或手环每天产生海量加速度、心率、皮肤电导数据，但真正能拿去训练模型的标注数据，却少得可怜。给传感器数据打标签需要人戴着设备做指定动作、记录时间戳、再逐段核对，成本高、效率低，还容易出错。很多人因此把项目卡在数据标注阶段，而不是模型设计阶段。

自监督学习正是为这个错配而生的。它不依赖人工标签，直接从传感器数据本身构造学习信号，先让模型理解“正常状态下的数据长什么样”，再根据少量标注去适配具体任务。本文要讨论的智能可穿戴系统，就是围绕“自监督预训练 + 多模态融合 + 下游任务微调”这一思路展开的，以活动识别和疲劳预测两个典型任务为例，给出可落地的系统设计、代码实现和工程建议。

这篇文章适合正在做可穿戴应用开发、健康监测算法研究，或者想了解自监督学习如何落地到传感器时序数据的读者。读完你会明白自监督学习在可穿戴场景的价值边界在哪里、多模态数据如何组织与融合、活动识别和疲劳预测代码上如何打通，以及真正上线时会遇到哪些坑。

1. 这篇文章真正要解决的问题

先下一个明确判断：可穿戴智能系统的开发瓶颈，已经从“数据不够”变成了“带标签的数据不够”。

传感器采集几乎没有成本，设备戴在用户手上，数据自然产生。真正昂贵的是让数据“可用”的环节。活动识别需要知道某段加速度波形对应走路、跑步还是上下楼，这要求实验者按脚本执行动作，并对齐时间戳；疲劳预测更需要结合问卷、量表或反应测试来标注疲劳等级。这种标注方式放到真实项目里，往往只能覆盖几十个小时的数据，但下游模型动辄需要几万个样本窗口，矛盾立刻显现。

自监督学习切中的就是这个问题。它让模型先在海量无标注数据上做预训练，学会提取有价值的时序特征，然后只用少量标注数据做微调，即可完成特定任务。说得直白一点：以前你是“先雇人把苹果分类好再训练模型”，现在改成“让模型先自己观察大量苹果，再教它认识几个典型类别”。

本文会围绕两条主线展开：

• 主线一是算法。从自监督学习的基本范式讲起，说明它如何在可穿戴多模态数据上做预训练，再迁移到活动识别和疲劳预测。
• 主线二是工程。给出环境准备、数据组织、核心代码、运行验证和常见问题排查，让读者不只是理解概念，还能跑通一条最小链路。

需要提前说明的是，自监督学习不是银弹。它解决的是标注稀缺下的特征学习问题，但它对数据增强、训练稳定性和下游适配都有要求。文章会把这些边界条件一并讲清楚。

2. 基础概念与核心原理

2.1 自监督学习到底在学什么

自监督学习的核心思想，是从数据自身构造监督信号，而不是依赖人工标注。它和传统监督学习的关键区别在于：监督学习学的是输入到标签的映射，自监督学习学的是输入本身的结构。

用一句话概括：自监督学习先让模型通过一个“代理任务”理解数据的内在规律，再把这个理解迁移到真实任务上。

以传感器数据为例，常见代理任务有：

• 掩码重建：随机遮盖某段时间窗口的传感器读数，让模型根据上下文预测被遮住的部分。模型为了预测准确，必须学会加速度波形中隐含的运动模式。
• 对比学习：构造一组互为正样本的增强数据对，让模型学习把同一来源数据映射到相近特征，把不同来源数据映射到不同特征。
• 时序预测：用前一段数据预测后一段数据，迫使模型学习状态转移规律。

其中对比学习的应用最广。它的思想可以类比为“人脸识别中的同一个人在不同光线下的照片应该是同一身份”，只不过在传感器数据里，同一段动作经过加噪、缩放、时间偏移后，对应的语义标签应该保持一致。

2.2 三种主流自监督范式对比

可穿戴领域目前更常用对比式方法。原因是传感器数据天然适合做增强，比如给加速度信号加噪声、缩放、时间扭曲，语义不会改变；同时活动识别和疲劳预测最终都落脚在分类或回归上，对比学习学到的判别性特征可以直接复用。

不过对比学习有一个常见陷阱：模型退化。如果增强方式设计不合理，模型可能找到一个“偷懒”的解法，把所有样本映射到同一个特征向量。后面我们会讲到如何通过具体实现规避这个问题。

2.3 多模态系统为什么是刚需

单一的加速度信号能识别走路、跑步，但在“疲劳预测”这类任务上就力不从心了。疲劳是一个涉及自主神经、肌肉活动和主观感受的综合状态，仅靠运动信号很难捕捉生理层面的变化。

这就需要引入多模态数据。以主流智能可穿戴系统为例，通常包含以下信号：

多模态系统的价值在于信息互补。运动模态告诉你“人正在做什么”，心率模态告诉你“负荷有多大”，皮肤电模态告诉你“应激状态有多强”。当三个模态同时指向“高强度运动 + 心率恢复变慢 + 皮肤电反应增强”时，疲劳预测的置信度会明显高于单一模态。

但多模态也带来代价：采样率不一致、时间戳偏移、缺失值更多、特征对齐更复杂。这些问题在第5章的代码中会有具体处理方案。

2.4 活动识别与疲劳预测：两种不同粒度的任务

文章标题把活动识别和疲劳预测放在一起，容易让人以为它们是同一个任务的两种叫法。实际上它们的任务粒度和特征需求差别很大。

活动识别是短时任务。走路、跑步、上下楼通常在几秒内就能判断，输入窗口一般为2到5秒，使用高频IMU信号即可获得较好效果。

疲劳预测是长时任务。疲劳状态的评估通常需要观察分钟级甚至小时级的生理趋势，比如心率变异性随时间的变化、EDA基线漂移、活动强度累积量。它的输入窗口更长，更依赖低频统计特征和上下文信息。

这种差异直接影响了模型设计：活动识别可以用高分辨率局部特征，疲劳预测则需要序列建模和特征聚合。在同一个系统里同时做两个任务，比较合理的方案是“共享底层编码器、分离任务头”，这也是第5章代码的主干结构。

3. 系统架构与整体流程

这一节先给出系统全貌，方便读者在进入代码前建立整体认知。

整个智能可穿戴系统分为五个模块：

1. 数据采集端：穿戴设备采集IMU、PPG、EDA等原始信号，定时上传或本地存储。
2. 数据预处理层：对多模态信号进行时间对齐、去噪、重采样、滑窗切分。
3. 自监督预训练层：使用无标注数据，通过对比学习训练底层编码器。
4. 下游任务适配层：冻结编码器或低学习率微调，分别训练活动识别头与疲劳预测头。
5. 部署推理层：将模型量化为轻量版本，在端侧或边缘侧完成实时预测。

数据流向如下：

可穿戴设备原始信号 ↓ 时间对齐与重采样（IMU/PPG/EDA 统一时间轴） ↓ 滑窗切分（2秒窗口用于活动识别，30秒窗口用于疲劳预测） ↓ 自监督预训练（对比学习，不需要标签） ↓ ┌──────────────┴──────────────┐ ↓ ↓ 活动识别头（分类） 疲劳预测头（回归/分类）

设计时有几个值得注意的点：

• 自监督预训练建议使用所有采集到的无标注数据，包括平时佩戴的日常数据，而不局限于实验脚本数据。数据量越大，预训练特征越通用。
• 活动识别与疲劳预测虽然任务粒度不同，但底层编码器可以是同一个。预训练阶段学到的运动节律、生理波动表征，对两个任务都有帮助。
• 推理阶段，活动识别窗口短、频率高；疲劳预测窗口长、频率低。系统会维护一个30秒到数分钟的滑动缓冲区，按需要输出两种结果。

4. 环境准备与数据说明

4.1 硬件与运行环境

本文的代码以Python实现，使用PyTorch。如果你想直接跑通完整流程，建议准备以下环境：

• 操作系统：Ubuntu 20.04/22.04，或Windows 10/11，macOS也可以运行但训练速度会慢一些。
• Python版本：3.8及以上，建议3.9或3.10。
• PyTorch：1.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 tqdm

4.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.pth

5. 完整示例代码实现

从这一节开始进入核心实操。代码会划分为四个文件：数据加载与预处理、自监督预训练、活动识别微调、疲劳预测微调。每个文件都尽量保持可直接运行的完整结构。

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, scaler=None): """对每个通道独立做标准化，避免量纲差异影响训练。""" 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, axis=0) def preprocess_pipeline(csv_path, window_seconds=2.0, step_seconds=1.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_std=0.05, scale_range=(0.8, 1.2), mask_ratio=0.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_channels=8, hidden_dim=128, output_dim=64): super().__init__() self.conv1 = nn.Sequential( nn.Conv1d(input_channels, 64, kernel_size=7, padding=3), nn.BatchNorm1d(64), nn.ReLU(), nn.MaxPool1d(2) ) self.conv2 = nn.Sequential( nn.Conv1d(64, 128, kernel_size=5, padding=2), nn.BatchNorm1d(128), nn.ReLU(), nn.MaxPool1d(2) ) self.conv3 = nn.Sequential( nn.Conv1d(128, hidden_dim, kernel_size=3, padding=1), 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_dim=64, hidden_dim=32, output_dim=16): 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, temperature=0.1): """ z1, z2: (batch_size, feat_dim) 同一批样本经过两种增强得到z1和z2，对角线位置是正样本对。 """ batch_size = z1.shape[0] z = torch.cat([z1, z2], dim=0) # (2*batch, feat_dim) sim = torch.mm(z, z.T) / temperature # 相似度矩阵 # 构造正负样本掩码 mask = torch.eye(2 * batch_size, device=z.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, dtype=torch.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(dim=1, keepdim=True) + 1e-8) loss = -log_prob[pos_mask].mean() return loss

数据集和训练循环如下：

class UnlabeledSensorDataset(Dataset): """无标注传感器窗口数据集。""" def __init__(self, windows): self.windows = torch.tensor(windows, dtype=torch.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, epochs=50, batch_size=256, lr=1e-3, device="cuda"): """ 自监督预训练入口。 windows: (N, window_size, num_channels) """ dataset = UnlabeledSensorDataset(windows) loader = DataLoader(dataset, batch_size=batch_size, shuffle=True) encoder = Encoder().to(device) projector = ProjectionHead().to(device) optimizer = torch.optim.Adam( list(encoder.parameters()) + list(projector.parameters()), lr=lr ) scheduler = torch.optim.lr_scheduler.CosineAnnealingLR(optimizer, T_max=epochs) 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(f"Epoch [{epoch+1}/{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.py

import 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, dtype=torch.float32) self.labels = torch.tensor(labels, dtype=torch.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, epochs=30, batch_size=128, lr=1e-3, device="cuda"): """ 线性评估：冻结编码器，只训练分类头。 这是验证自监督特征质量最常用的实验协议。 """ train_ds = ActivityDataset(train_windows, train_labels) val_ds = ActivityDataset(val_windows, val_labels) train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True) val_loader = DataLoader(val_ds, batch_size=batch_size, shuffle=False) 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(), lr=lr) 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(f"Activity Finetune Epoch [{epoch+1}/{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, dim=1) 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, average="weighted") print(f"活动识别验证准确率: {acc:.4f}, F1: {f1:.4f}") torch.save(head.state_dict(), "data/models/head_activity.pth") return acc, f1

5.4 疲劳预测微调

疲劳预测是回归任务，目标是输出疲劳分数。这里采用“编码器特征 + 生理统计特征 + 回归头”的结构，因为疲劳判断不只依赖短时波形，还依赖心率变异性、EDA均值等统计指标。

文件路径：finetune_fatigue.py

import 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(axis=1), ppg.std(axis=1), eda.mean(axis=1), eda.std(axis=1), np.sqrt((acc ** 2).sum(axis=2)).mean(axis=1) # 加速度合成模长均值 ]) return features.astype(np.float32) class FatigueDataset(Dataset): """疲劳预测数据集，输入多模态窗口，标签为疲劳分数。""" def __init__(self, windows, labels): self.windows = torch.tensor(windows, dtype=torch.float32) self.labels = torch.tensor(labels, dtype=torch.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_dim=5, hidden_dim=32): 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], dim=1) out = self.fusion(combined) return out.squeeze(-1) def train_fatigue(encoder, train_windows, train_labels, val_windows, val_labels, epochs=30, batch_size=128, lr=1e-3, device="cuda"): """ 训练疲劳预测模型。 与活动识别不同，这里允许编码器低学习率微调， 因为疲劳预测依赖更高级的生理语义特征。 """ train_ds = FatigueDataset(train_windows, train_labels) val_ds = FatigueDataset(val_windows, val_labels) train_loader = DataLoader(train_ds, batch_size=batch_size, shuffle=True) val_loader = DataLoader(val_ds, batch_size=batch_size, shuffle=False) model = FatigueModel(encoder).to(device) criterion = nn.MSELoss() optimizer = torch.optim.Adam(model.parameters(), lr=lr) 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(f"Fatigue Epoch [{epoch+1}/{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, mae

5.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_seconds=2.0) all_windows.append(windows) return np.concatenate(all_windows, axis=0) def main(): os.makedirs(PROCESSED_DIR, exist_ok=True) os.makedirs(MODEL_DIR, exist_ok=True) # Step 1: 构建无标注数据并做自监督预训练 print("===== 构建无标注数据 =====") unlabeled_windows = build_unlabeled_windows() print("无标注窗口数量:", unlabeled_windows.shape[0]) print("===== 自监督预训练 =====") encoder = train_pretrain(unlabeled_windows, epochs=50) # Step 2: 活动识别微调 # 实际项目中会从标注数据中读取活动标签 # 这里以随机生成的小数据演示接口调用方式 print("===== 活动识别微调 =====") labeled_windows = unlabeled_windows[:2000] fake_activity_labels = np.random.randint(0, 4, size=labeled_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.1846

6.2 用线性评估验证特征质量

活动识别的结果验证有一个行业通用协议：线性评估（linear probing）。做法是把编码器冻结，只训练一个线性分类头，看分类准确率。

这种方法的好处是能直接反映预训练特征的线性可分性。如果自监督预训练真的学到了好特征，那么即使只用一个线性分类头，也应该取得接近“端到端全监督训练”的效果。如果线性评估准确率很低，则说明预训练特征质量不足，不应该直接进入后续微调。

在真实数据集上，线性评估的准确率通常能接近有监督训练的80%到95%，具体取决于数据集复杂度和样本量。如果明显偏低，优先检查数据增强策略和预训练epoch数量。

6.3 疲劳预测效果验证

疲劳预测属于回归任务，常用指标是MSE和MAE。需要注意：

• 疲劳是一个连续变量，个体之间基线差异极大，直接比较绝对误差意义有限。
• 更好的方式是按照受试者划分数据集，观察模型是否能在未知用户上保持稳定误差。
• 如果MAE普遍偏高，可以考虑把疲劳问题转化为“疲劳/不疲劳”的二分类问题，降低任务难度，先验证系统整体链路是否通畅。

6.4 运行失败的优先排查顺序

如果代码运行失败，建议按以下顺序排查：

1. 查看堆栈信息，确认是否import错误，重点检查PyTorch版本。
2. 检查数据路径是否存在，数据文件是否可以正常读取为DataFrame。
3. 查看窗口形状，确认(window_size, num_channels)维度是否与预训练代码期望一致。
4. 如果显存不足，调小batch_size或window_size。

7. 常见问题与排查思路

8. 最佳实践与工程建议

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. 总结与后续学习方向

自监督学习在智能可穿戴系统里的价值，不是取代监督学习，而是解决“有数据、少标签”条件下如何学习的问题。本文给出的对比学习框架，先用无标注的传感器窗口训练底层编码器，再通过活动识别和疲劳预测两个下游任务验证特征的可复用性，整体链路已经覆盖从数据处理到模型训练再到效果验证的完整流程。

如果你要在这个方向继续深入，建议按以下顺序学习：

• 先把本文代码在自采的小数据集上跑通，理解每个模块的输入输出。
• 再尝试替换不同的自监督范式，比如掩码重建或时序预测，对比特征质量。
• 接着引入公开的可穿戴数据集做基准实验，按受试者划分评估泛化性能。
• 最后考虑多任务联合训练和端侧部署，把实验室模型变成可用系统。

可穿戴数据本质上是一种长尾分布的数据：用户的动作模式、设备佩戴位置、传感器型号都会改变数据分布。自监督预训练在这类场景里之所以值得投入，正是因为它让模型先从大量无标注的“常规状态”中学会表征，再去适应少量标注的“特定任务”。这个思路不仅适用于活动识别和疲劳预测，也适用于跌倒检测、压力估计、睡眠分期等更多方向。把这套框架掌握好，你在可穿戴算法方向上的大多数项目都可以直接复用。