手环、手表导出的加速度和心率，通常不能直接丢给 SensorFM 这类可穿戴时序模型。模型吃的是一条条等间隔、通道对齐的多路序列，而导出文件里往往是加速度 50 Hz、心率 1 Hz、时间戳各记各的，中间还断了几段。要先做的三件事是：把字段和时间戳核对清楚、统一到同一时间栅格、把缺失段标成掩码而不是补零。这三步基本都得自己做，没有哪个包能替你判断某一路通道到底是什么含义。

可穿戴原始导出数据通常不能直接进入时序模型，需要先做字段核对、采样率统一、时间戳对齐、缺失掩码四步最小预处理。适用场景是同一受试者的加速度与心率等多通道数据。先确认时间戳单位、时区与各通道采样率，再重采样到统一栅格并记录插值与抗混叠方式，最后用 0/1 掩码标记缺口。验证方式是打印对齐后的首尾时间、总长度与掩码占比，与原始文件人工比对。整段缺失不宜靠插值修复。

把手环导出文件里的字段和时间戳摊开看

导出文件常见两类：CSV（一行一条记录）和按通道存的二进制或 JSON。无论哪种，先确认下面这些字段到底存不存在、单位是什么，不要凭印象认定 acc_x 就是某个固定方向。这一步的产物应该是一张手写或脚本打印出来的字段表，而不是脑子里的一句“大概是这样”。

• 时间戳：是秒还是毫秒，是整数 epoch 还是带时区的 ISO 字符串；同一文件内是否单调递增、有没有重复值。
• 三轴加速度：字段名、单位（g 还是 m/s²）、量程与符号；三轴是否共用同一个时间戳。
• 心率：是 BPM 还是 RR 间期；是否带质量标记，比如 confidence、contact 之类的列。
• 步数：多数是累计值或按分钟聚合，采样率远低于加速度，不要当等间隔序列用。
• 各通道采样频率：对每个通道单独取相邻时间戳差值的中位数，再换算成 Hz。

如果某个通道 diff(t) 的中位数波动明显，说明存在丢帧或不规则采样，后面就必须按时间戳重采样，而不能假设固定步长。同时要选定哪一路时间戳作为主时钟，通常选采样率最高、缺失最少的那一路，其余通道向它靠。

统一到同一时间栅格并记录插值方式

统一采样率的目的是让不同频率的数据进同一条管线，而不是“把数据变好看”。降采样（比如 100 Hz 变 50 Hz）之前先做低通，否则高频成分会折叠回低频，混进信号里；升采样（比如心率 1 Hz 拉到 50 Hz）不会增加任何信息，只是让长度对齐，所以插值方式要如实记录，别默认它是真实观测。

插值方式上，线性插值简单、不会过冲，适合心率这类本身变化慢的通道；样条更平滑但可能在缺口边缘产生超出原始范围的取值，只有在缺口很短且通道本身平滑时才考虑。下面这段不依赖特定框架，只用 numpy，替换掉通道名就能跑。

import numpy as np

def lowpass(x, k=5):
    # 极简滑动平均，仅作抗混叠的低通近似；装了 scipy 可用 butter + filtfilt 替换
    kernel = np.ones(k) / k
    return np.convolve(x, kernel, mode='same')

def resample_to_grid(t, x, grid, method='linear', anti_alias=True):
    # t：原始时间戳，单位秒，严格递增；grid：目标时间栅格，单位秒
    dt_src = np.median(np.diff(t))
    dt_dst = np.median(np.diff(grid))
    if anti_alias and dt_dst > dt_src:   # 降采样：先低通再抽点
        x = lowpass(x)
    if method == 'linear':
        y = np.interp(grid, t, x)
    else:
        from scipy.interpolate import CubicSpline
        y = CubicSpline(t, x)(grid)
    assert y.shape == grid.shape, (y.shape, grid.shape)
    print('src_pts=%d dst_pts=%d dt=%.6f->%.6f' % (len(t), len(grid), dt_src, dt_dst))
    return y

自检点：打印出来的 src_pts 和 dst_pts 数量级要能对上——目标栅格点数应该约等于时间跨度除以步长，差得离谱就是时间戳单位判错了（秒当毫秒是常见错误）。另外把重采样前后各通道的首尾取值各打印一行，确认没有整体平移或整体缩放。

对齐多条时间序列的起点与漂移

同一只手表的不同传感器，各自记时的情况很常见，加上设备时钟本身会有缓慢漂移，直接按数组下标对齐会整体错位。稳妥做法是先用时间戳求公共区间，再在公共区间上按固定步长生成索引，最后让所有通道都用这同一个索引去取值。

fs = 50.0                                  # 统一目标采样率，需与下游模型输入一致
step = 1.0 / fs
t0 = max(acc_t[0], hr_t[0])                # 公共起点
t1 = min(acc_t[-1], hr_t[-1])              # 公共终点
grid = np.arange(t0, t1 + 1e-9, step)      # 固定栅格，浮点误差留个小尾巴

acc = resample_to_grid(acc_t, acc_z, grid)
hr  = resample_to_grid(hr_t,  hr_v,  grid)

assert len(acc) == len(hr) == len(grid), (len(acc), len(hr), len(grid))
print('grid head=%s tail=%s n=%d' % (grid[0], grid[-1], len(grid)))

对齐后要核对的就三项：所有通道首尾时间戳是否一致、总长度是否相等、公共区间是否被明显截短。如果交集比原始数据短很多，说明两路数据的时间范围差得大，要么是某一侧时钟设置错了，要么是其中一路只覆盖了半天，这时候应该回头查导出设置，而不是默默用截短后的数据继续跑。

把缺失段标成掩码而不是补零

补零是最容易埋坑的一步：模型分不清“这里真的是 0 g”和“这里根本没采到”，尤其在心率上，0 BPM 会被当成真实观测。正确做法是在重采样之前先记录每个时刻“原本有没有值”，再把这层信息一起重采样成 0/1 掩码。

present = np.isfinite(hr_raw).astype(float)          # 重采样前先记录是否存在
mask = resample_to_grid(hr_t, present, grid)
mask = (mask > 0.999).astype(np.uint8)               # 边界处按阈值取整为 0/1

def max_gap(m):
    gaps, cur = 0, 0
    for v in m:
        cur = cur + 1 if v == 0 else 0
        gaps = max(gaps, cur)
    return gaps

print('mask_mean=%.3f max_gap=%d' % (mask.mean(), max_gap(mask)))

短缺口（比如几秒的丢帧）可以在掩码里保留、信号按插值填上，让模型自己从掩码判断可信度；整段缺失（比如用户摘下手表几十分钟）不建议插值，插出来的曲线看着连续，实际上是编的，应把这一段单独标记，训练时整段跳过或截断。保存时掩码必须和信号同形状、同长度，按通道各存一份，例如 acc_mask 对应 acc、hr_mask 对应 hr，维度对不上就说明前面某一步长度校验被绕过了。

固定预处理参数并存成可复用格式

同一个人、同一批数据，今天和下周跑出来的结果应当一致。做法是把所有可能影响输出的参数写进一个旁挂的配置文件，和信号文件放在一起，而不是散落在脚本里。

{
  'fs': 50.0,
  'grid_unit': 's',
  'timezone': 'UTC',
  'interp': {'acc': 'linear', 'hr': 'linear'},
  'anti_alias': {'enabled': true, 'method': 'moving_average', 'k': 5},
  'gap': {'short_gap_max_s': 2, 'min_seg_s': 30},
  'mask_dtype': 'uint8',
  'version': 'prep_v1'
}

命名建议带上受试者、设备、起始时间戳、目标采样率和预处理版本，比如 <subject>_<device>_<起始时间戳>_50hz_v1.npz，这样一眼能看出这份数据是用哪套参数生成的。重跑验证的方式也很直接：同一份原始文件跑两遍，对输出的 npz 做一次校验和比对；如果两次不一致，通常是 np.arange 的浮点累加、插值库版本或时区处理有差异，把这几处固定住再重跑。