4. 时间序列数据对齐与重采样

任务说明

在工业PHM平台中，不同传感器与PLC系统的采样周期往往不一致（如振动信号1kHz、温度1Hz、压力0.5Hz），且存在时间戳抖动与缺失。本案例目标：实现一个时间序列对齐与重采样工具，将多源不规则采样数据统一到目标频率网格上，并输出对齐后的规整数据表，供后续特征提取与模型训练直接使用。

实现方案

采用先对齐、后重采样的两步策略。对齐阶段：以目标时间网格为基准，对每个数据源执行最近邻前向填充（不引入未来数据，适用于实时PHM管道）。重采样阶段：对对齐后的等间隔序列，按需执行降采样聚合（均值）或升采样插值（线性），最终输出统一频率的DataFrame。代码使用Pandas实现，核心逻辑约40行，不依赖额外库。

代码实现

import pandas as pd
import numpy as np

def align_resample(data_dict, start, end, target_freq, method='mean'):
    """
    多源时间序列对齐与重采样
    参数:
        data_dict: dict, {列名: DataFrame(index=DatetimeIndex, 单列数值)}
        start/end: str, 目标区间
        target_freq: str, 目标频率, 如 '1s', '100ms'
        method: 'mean'(降采样) 或 'linear'(升采样)
    返回:
        DataFrame, 索引为目标时间网格, 列为各数据源
    """
    # 1. 生成统一目标时间网格
    grid = pd.date_range(start=start, end=end, freq=target_freq)
    result = pd.DataFrame(index=grid)

    for col, df in data_dict.items():
        # 2. 对齐: 最近邻前向填充(不引入未来数据)
        aligned = df.reindex(grid, method='ffill')
        # 3. 边界处理: 开头无前值时用最近的后值回填(仅限起始段)
        if aligned.isna().any():
            aligned = aligned.fillna(method='bfill')
        # 4. 重采样: 若目标频率低于原始频率则聚合, 否则插值
        if method == 'mean':
            # 先按目标网格对齐后, 对重复网格点取均值(处理原始频率高于目标的情况)
            aligned = aligned.groupby(level=0).mean()
        elif method == 'linear':
            # 线性插值(处理原始频率低于目标的情况)
            aligned = aligned.interpolate(method='time')
        result[col] = aligned

    return result

# ========== 样例数据(模拟3个不同采样率的传感器) ==========
# 振动: 100Hz, 采集0~2秒
vib_idx = pd.date_range('2024-01-01 00:00:00', periods=200, freq='10ms')
vib = pd.DataFrame({'vib': np.sin(np.linspace(0, 4*np.pi, 200)) + 0.1*np.random.randn(200)}, index=vib_idx)

# 温度: 1Hz, 采集0~2秒(存在抖动)
temp_idx = pd.to_datetime(['2024-01-01 00:00:00.000', '2024-01-01 00:00:01.050',
                           '2024-01-01 00:00:02.100'])
temp = pd.DataFrame({'temp': [22.5, 22.7, 23.0]}, index=temp_idx)

# 压力: 0.5Hz, 采集0~2秒
pres_idx = pd.date_range('2024-01-01 00:00:00', periods=2, freq='2s')
pres = pd.DataFrame({'pres': [101.3, 101.5]}, index=pres_idx)

# ========== 执行对齐: 统一到1Hz ==========
data_sources = {'vib': vib, 'temp': temp, 'pres': pres}
aligned_1hz = align_resample(data_sources,
                             start='2024-01-01 00:00:00',
                             end='2024-01-01 00:00:02',
                             target_freq='1s',
                             method='mean')
print("=== 对齐到1Hz(振动降采样均值) ===")
print(aligned_1hz)

# ========== 执行对齐: 统一到100ms(升采样) ==========
aligned_100ms = align_resample(data_sources,
                               start='2024-01-01 00:00:00',
                               end='2024-01-01 00:00:02',
                               target_freq='100ms',
                               method='linear')
print("\n=== 对齐到100ms(温度/压力线性插值) ===")
print(aligned_100ms.head(5))

落地与验证

运行方式：保存为 align_resample.py，执行 python align_resample.py。需安装 pandas 与 numpy。

验证方法：观察两个输出DataFrame：

• aligned_1hz 应包含3行（00:00:00、00:00:01、00:00:02），每行均有vib/temp/pres三列非空值。vib列是原始100个点按秒聚合的均值，temp列在00:00:01处取最近前值22.5（因原始1.05s晚于网格点），pres列在00:00:01处前向填充101.3。
• aligned_100ms 应包含21行（0~2秒每100ms），vib列因原始频率更高且method='linear'，实际按网格取最近前值（ffill已生效）；temp与pres列通过线性插值填补中间点。

判定标准：两个输出均无NaN值，索引严格等间隔，且各列数值处于原始数据合理范围内（vib∈[-1.2,1.2]，temp∈[22.4,23.1]，pres∈[101.2,101.6]），即视为通过。