一段 8 小时记录要切成训练样本,窗口长度、重叠率、缺失段去留这三件事没有通用最优值,但它们必须在上游一次定死,并且能被配置文件和日志验证。把窗口切法和缺失标记先固化下来,后面模型结构、训练策略的调整才有可比较的基线;否则同一份数据每次切出的样本不同,效果变化说不清来自哪里。
窗口长度、重叠率和缺失处理没有通用最优值,但必须在上游一次定死并写进可复现的配置。建议先用 10s/30s/60s 与 0%/50% 六组组合做最小实验,按时间戳而不是按行号切窗,显式生成与信号同维度的缺失掩码随样本落盘,再检查重叠窗口是否造成训练与验证区间相交。参数固定后,随机种子、起始偏移和划分方式一并留存,换机器重跑应得到相同的样本清单。
列出窗口长度与重叠率的最小组合实验
先用少量组合摸清样本量与信息密度的取舍,不要一上来就扫参。步长与样本数的关系先按下面的方式算,T 为记录可用总时长:
- hop = window_sec × (1 − overlap_ratio)
- 样本数 = floor((T − window_sec) / hop) + 1,末尾不足一个窗口的部分由配置决定丢弃还是补齐
以 T = 28800s(8 小时连续记录、无缺失)为例,六组组合的样本数与取舍方向如下。实际记录有缺失段时,先扣除缺失时长,再代入同一个公式。
| 窗口长度 | 重叠率 | 步长 hop | 8h 样本数算式 | 取舍方向 |
|---|---|---|---|---|
| 10s | 0% | 10s | floor((28800−10)/10)+1 = 2880 | 样本最多,单窗信息最少,短事件常被切断 |
| 10s | 50% | 5s | floor((28800−10)/5)+1 = 5759 | 边界事件更可能完整落在某个窗口内,样本量翻倍,训练与验证更易交叉 |
| 30s | 0% | 30s | floor((28800−30)/30)+1 = 960 | 折中档,适合大多数周期性生理或运动信号 |
| 30s | 50% | 15s | floor((28800−30)/15)+1 = 1919 | 保留折中窗口,同时缓解状态切换被切碎的问题 |
| 60s | 0% | 60s | floor((28800−60)/60)+1 = 480 | 样本最少,适合缓慢漂移类信号,短时异常会被稀释 |
| 60s | 50% | 30s | floor((28800−60)/30)+1 = 959 | 样本量与 30s/0% 接近,但单窗更长,需关注标签在窗口内的归属规则 |
挑组合时看三件事:单位时间内能产出多少样本、窗口长度是否覆盖目标事件的最短持续时间、以及训练侧一次能装下多少样本。每组跑完把样本总数、被丢弃的尾段时长、含缺失点的样本数写进切窗日志,作为下一轮比较的依据。选定后不要在同一批实验里再改窗口参数,否则对比结果没有意义。
按时间戳切窗而不是按行号切
传感器记录常见采样不齐:丢包、重连、设备降频都会让相邻两行的真实间隔不一致。按行号切片会把不同真实时长的片段当成等长样本,窗口的物理含义就丢了。下面的骨架按时间索引切窗,并在边界处加断言。
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、命中缺失区间数),标签单独一个数组,不要和掩码拼在同一维度里。
检查切窗后是否出现样本泄漏
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
复现验证按下面的顺序做,每步都能通过日志或产物比对确认:
- 同一份原始记录连续跑两次切窗,把 manifest 里样本起始时间戳排序后取哈希对比,两次应一致。
- 换机器重跑,比对样本总数和每个窗口的首尾时间戳;对不上时优先查时区解析和浮点时间戳精度,而不是先怀疑切窗逻辑。
- 启动时断言 hop_sec 等于 window_sec × (1 − overlap_ratio),防止手写步长和重叠率互相矛盾。
- 把 seed、window_sec、overlap_ratio、start_offset_sec、划分键写进 manifest,样本产出后不再改动参数;要改就重新生成样本清单,不要复用旧样本混着训。