先把窗口切法和缺失标记定下来——SensorFM 的上游数据准备边界

文章导读
一段 8 小时记录要切成训练样本,窗口长度、重叠率、缺失段去留这三件事没有通用最优值,但它们必须在上游一次定死,并且能被配置文件和日志验证。把窗口切法和缺失标记先固化下来,后面模型结构、训练策略的调整才有可比较的基线;否则同一份数据每次切出的样本不同,效果变化说不清来自哪里。
📋 目录
  1. A 列出窗口长度与重叠率的最小组合实验
  2. B 按时间戳切窗而不是按行号切
  3. C 生成缺失掩码并与信号一起保存
  4. D 检查切窗后是否出现样本泄漏
  5. E 固定随机种子并留存切窗参数
A A

一段 8 小时记录要切成训练样本,窗口长度、重叠率、缺失段去留这三件事没有通用最优值,但它们必须在上游一次定死,并且能被配置文件和日志验证。把窗口切法和缺失标记先固化下来,后面模型结构、训练策略的调整才有可比较的基线;否则同一份数据每次切出的样本不同,效果变化说不清来自哪里。

窗口长度、重叠率和缺失处理没有通用最优值,但必须在上游一次定死并写进可复现的配置。建议先用 10s/30s/60s 与 0%/50% 六组组合做最小实验,按时间戳而不是按行号切窗,显式生成与信号同维度的缺失掩码随样本落盘,再检查重叠窗口是否造成训练与验证区间相交。参数固定后,随机种子、起始偏移和划分方式一并留存,换机器重跑应得到相同的样本清单。

列出窗口长度与重叠率的最小组合实验

先用少量组合摸清样本量与信息密度的取舍,不要一上来就扫参。步长与样本数的关系先按下面的方式算,T 为记录可用总时长:

  • hop = window_sec × (1 − overlap_ratio)
  • 样本数 = floor((T − window_sec) / hop) + 1,末尾不足一个窗口的部分由配置决定丢弃还是补齐

以 T = 28800s(8 小时连续记录、无缺失)为例,六组组合的样本数与取舍方向如下。实际记录有缺失段时,先扣除缺失时长,再代入同一个公式。

窗口长度重叠率步长 hop8h 样本数算式取舍方向
10s0%10sfloor((28800−10)/10)+1 = 2880样本最多,单窗信息最少,短事件常被切断
10s50%5sfloor((28800−10)/5)+1 = 5759边界事件更可能完整落在某个窗口内,样本量翻倍,训练与验证更易交叉
30s0%30sfloor((28800−30)/30)+1 = 960折中档,适合大多数周期性生理或运动信号
30s50%15sfloor((28800−30)/15)+1 = 1919保留折中窗口,同时缓解状态切换被切碎的问题
60s0%60sfloor((28800−60)/60)+1 = 480样本最少,适合缓慢漂移类信号,短时异常会被稀释
60s50%30sfloor((28800−60)/30)+1 = 959样本量与 30s/0% 接近,但单窗更长,需关注标签在窗口内的归属规则

挑组合时看三件事:单位时间内能产出多少样本、窗口长度是否覆盖目标事件的最短持续时间、以及训练侧一次能装下多少样本。每组跑完把样本总数、被丢弃的尾段时长、含缺失点的样本数写进切窗日志,作为下一轮比较的依据。选定后不要在同一批实验里再改窗口参数,否则对比结果没有意义。

先把窗口切法和缺失标记定下来——SensorFM 的上游数据准备边界

按时间戳切窗而不是按行号切

传感器记录常见采样不齐:丢包、重连、设备降频都会让相邻两行的真实间隔不一致。按行号切片会把不同真实时长的片段当成等长样本,窗口的物理含义就丢了。下面的骨架按时间索引切窗,并在边界处加断言。

import numpy as np

def slice_by_time(ts, sig, win_sec, hop_sec, missing=None):
    # ts: datetime64[ns] 一维数组,严格递增;sig: (N, C)
    assert np.all(np.diff(ts) > np.timedelta64(0, 'ns')), '时间戳必须严格递增'
    win = np.timedelta64(int(round(win_sec * 1e9)), 'ns')
    hop = np.timedelta64(int(round(hop_sec * 1e9)), 'ns')
    start, last = ts[0], ts[-1]
    while start + win <= last:
        stop = start + win
        i0 = np.searchsorted(ts, start, side='left')
        i1 = np.searchsorted(ts, stop, side='left')
        seg_ts, seg_sig = ts[i0:i1], sig[i0:i1]
        assert len(seg_ts) >= 2, '窗口内有效点不足'
        span = (seg_ts[-1] - seg_ts[0]) / np.timedelta64(1, 's')
        assert span <= win_sec + 1e-6, f'窗口跨度 {span}s 超出 {win_sec}s'
        hit = [] if missing is None else [(a, b) for a, b in missing if a < stop and b > start]
        yield seg_ts, seg_sig, hit
        start = start + hop

span 断言确认窗口内首尾时间戳的跨度不超过设定长度;len(seg_ts) 断言排除空窗;返回的 hit 列出与该窗口相交的缺失区间,直接交给下一步生成掩码。searchsorted 要求时间戳严格递增,如果原始数据里有重复时间戳,先在读入阶段去重或按同秒聚合,不要带着重复值进切窗函数。

生成缺失掩码并与信号一起保存

补零后的数值和真实测量到的零在数组里无法区分,训练时会被当成有效信号。做法是生成与信号同形状的掩码,随样本一起落盘,训练侧按掩码决定是否计入损失或先做插补。

import numpy as np

def build_mask(ts, missing):
    mask = np.ones(ts.shape, dtype=np.uint8)
    for a, b in missing:
        idx = np.where(np.logical_and(ts >= a, ts < b))[0]
        mask[idx] = 0
    return mask

def make_sample(i0, i1, sig, mask, label):
    s, m, y = sig[i0:i1], mask[i0:i1], label[i0:i1]
    assert s.shape == m.shape, 'mask 必须与 signal 同形状 (T, C)'
    assert m.ndim == s.ndim
    return s, m, y

sig = load_signal()    # (N, C),按实际通道数替换
label = load_label()   # (N, K)
mask = build_mask(ts, missing)
s, m, y = make_sample(i0, i1, sig, mask, label)
np.savez_compressed('win_000123.npz', signal=s, mask=m, label=y)
print(s.shape, m.shape, y.shape, int((m == 0).sum()))

掩码维度必须与信号一致:信号是 (T, C),掩码也是 (T, C)。只存一个 (T,) 的全局缺失标记表达不了多通道各自缺失的情况,比如某一路电极松动而其他路正常。落盘时建议给每个样本带一份元数据(起始时间戳、受试者 ID、命中缺失区间数),标签单独一个数组,不要和掩码拼在同一维度里。

先把窗口切法和缺失标记定下来——SensorFM 的上游数据准备边界

检查切窗后是否出现样本泄漏

50% 重叠意味着相邻样本共享一半数据。如果划分时按样本随机打散,同一段原始记录的窗口会同时进训练集和验证集。最直接的检查方式是比较样本的时间区间,把高度重叠的跨集合样本对打印出来。

def overlap_sec(x, y):
    lo, hi = max(x[0], y[0]), min(x[1], y[1])
    return max(0.0, (hi - lo).total_seconds())

# spans: [(start_ts, end_ts, split_name), ...],split_name 取 train 或 val
def print_leaks(spans, ratio=0.9):
    for i in range(len(spans)):
        for j in range(i + 1, len(spans)):
            x, y = spans[i], spans[j]
            if x[2] == y[2]:
                continue
            dur = min((x[1] - x[0]).total_seconds(), (y[1] - y[0]).total_seconds())
            ov = overlap_sec(x, y)
            if ov > ratio * dur:
                print('区间高度重叠', x[:2], y[:2], '重叠', round(ov, 3), '秒')

按人划分(受试者 ID 作为划分键)时,同一受试者的全部窗口只进一个集合,通常能直接避开这类泄漏,代价是验证集覆盖的人少一些。如果确实要按时间划分,在训练段和验证段之间留一段不小于一个窗口的隔离带,隔离带内的样本两边都不进。多人多天数据建议先按人分组,再在组间做划分,最后才考虑组内的时间顺序。

固定随机种子并留存切窗参数

切窗参数写成配置文件,随代码一起进版本管理,不要散落在脚本里。

windowing:
  window_sec: 30
  overlap_ratio: 0.5
  hop_sec: 15            # 由 window_sec * (1 - overlap_ratio) 推得,不手写
  start_offset_sec: 0    # 相对记录起始的偏移,用于避开设备预热段
  drop_last_partial: true
time:
  field: timestamp_utc   # 时间列名
  timezone: UTC
missing:
  mask_value: 0          # 缺失位置置 0,不做插值,只标记
split:
  by: subject_id         # subject_id 或 time
  val_ratio: 0.2
  guard_sec: 30          # 按 time 划分时的隔离带长度
repro:
  seed: 42
  save_manifest: true

复现验证按下面的顺序做,每步都能通过日志或产物比对确认:

  1. 同一份原始记录连续跑两次切窗,把 manifest 里样本起始时间戳排序后取哈希对比,两次应一致。
  2. 换机器重跑,比对样本总数和每个窗口的首尾时间戳;对不上时优先查时区解析和浮点时间戳精度,而不是先怀疑切窗逻辑。
  3. 启动时断言 hop_sec 等于 window_sec × (1 − overlap_ratio),防止手写步长和重叠率互相矛盾。
  4. 把 seed、window_sec、overlap_ratio、start_offset_sec、划分键写进 manifest,样本产出后不再改动参数;要改就重新生成样本清单,不要复用旧样本混着训。