📚 课程目录第4章数据预处理

4. 数据预处理：时间戳对齐、缺失值处理、异常值过滤、数据归一化

数据预处理这事儿，说实话，比算法选型更磨人。我早年做第一个IoT项目时，花了两周调模型，结果发现效果差得离谱。排查了三天，最后发现是时间戳没对齐——两个传感器差了200毫秒，模型学了一堆“假规律”。

从那以后，我养成了一个习惯：预处理花多少时间都不算多。你想想看，原始数据就像刚从矿里挖出来的矿石，你不洗选、不破碎、不筛分，直接扔进高炉，能炼出好钢才怪。

核心观点：数据预处理的质量，直接决定了行为基线的可信度。这一步偷懒，后面所有分析都是空中楼阁。

4.1 时间戳对齐——让所有设备“说同一种时间”

IoT设备的时间戳问题，我归纳为三类：

时钟漂移：设备晶振不准，一天能差几秒到几分钟
采样周期不一致：有的设备每秒上报，有的每5秒上报
网络延迟：数据到达服务器的时间，和实际采集时间不同

我个人习惯的做法是：统一使用UTC时间，以服务器接收时间为基准，反向校准设备时间戳。具体流程如下：

将所有时间戳转换为Unix时间戳（毫秒级）
按设备ID分组，对每个设备的时间序列单独处理
设定一个标准采样间隔（比如1秒），作为对齐基准
使用线性插值或最近邻插值，将数据对齐到标准时间点上

import pandas as pd
import numpy as np

def align_timestamps(df, device_id_col='device_id', timestamp_col='timestamp', 
                     value_col='value', freq='1S'):
    """
    时间戳对齐函数
    df: 原始DataFrame
    freq: 目标采样频率，默认1秒
    """
    aligned_list = []
    
    for device_id, group in df.groupby(device_id_col):
        # 设置时间戳为索引
        group = group.set_index(timestamp_col)
        group.index = pd.to_datetime(group.index, unit='ms')
        
        # 生成标准时间序列
        start = group.index.min().floor(freq)
        end = group.index.max().ceil(freq)
        new_index = pd.date_range(start=start, end=end, freq=freq)
        
        # 重采样+线性插值
        aligned = group.reindex(group.index.union(new_index))
        aligned = aligned.interpolate(method='linear')
        aligned = aligned.loc[new_index]
        
        aligned['device_id'] = device_id
        aligned_list.append(aligned.reset_index())
    
    return pd.concat(aligned_list, ignore_index=True)

经验之谈：我曾经处理过一个智能电表项目，设备上报间隔标称是15秒，实际分析发现，有的设备在整点时会“跳报”——连续上报3-4条数据。如果不做对齐，模型会把这种“跳报”当成异常行为，误报率飙升。后来我加了一个去重+中值滤波的步骤，才把这个问题解决。

4.2 缺失值处理——别让“空洞”误导你的模型

IoT数据缺失的原因太多了：设备掉线、网络丢包、传感器故障、电池耗尽……我见过最夸张的一个项目，某设备一天内缺失了40%的数据点。

处理缺失值，我的原则是：先分类，再处理。

缺失类型	特征	推荐处理方法
随机缺失	零星、无规律	线性插值、前向填充
连续缺失（短）	连续缺失3-5个点	样条插值、移动平均
连续缺失（长）	连续缺失超过10个点	标记为“不可用时段”，不填充
周期性缺失	每天固定时段缺失	检查设备配置，可能是休眠策略

嗯，这里要注意：不要盲目填充。我曾经犯过一个错误——对连续缺失30分钟的数据用了线性插值，结果模型学出了一个“平滑的假象”，把真正的设备故障给掩盖了。后来我改了策略：连续缺失超过阈值（比如采样间隔的5倍），直接标记为NaN，在后续建模中跳过这些时段。

def handle_missing_values(df, value_col='value', max_gap=5):
    """
    智能缺失值处理
    max_gap: 允许连续填充的最大缺失点数
    """
    # 标记连续缺失的区间
    df['is_missing'] = df[value_col].isna().astype(int)
    df['gap_group'] = (df['is_missing'] != df['is_missing'].shift()).cumsum()
    
    # 计算每个缺失区间的长度
    gap_lengths = df[df['is_missing']==1].groupby('gap_group').size()
    
    # 短缺失：线性插值
    short_gaps = gap_lengths[gap_lengths <= max_gap].index
    mask_short = df['gap_group'].isin(short_gaps) & df['is_missing']==1
    df.loc[mask_short, value_col] = df[value_col].interpolate(method='linear')
    
    # 长缺失：保留NaN，后续处理
    # 这里可以添加标记列，记录哪些时段被跳过
    
    return df.drop(columns=['is_missing', 'gap_group'])

避坑指南：我曾经在一个温湿度监测项目中，发现缺失值处理后的数据出现了“阶梯状”突变。排查后发现，是因为不同设备的时间戳精度不同——有的精确到毫秒，有的只精确到秒。对齐时，毫秒级数据被“截断”到了秒级，造成了数据重复。解决方案：在时间戳对齐前，先统一时间精度。

4.3 异常值过滤——把“脏数据”挡在门外

异常值，说白了就是那些“明显不对劲”的数据点。比如温度传感器突然报出-200°C，或者电流值瞬间飙到1000A。这些数据如果不处理，会严重扭曲行为基线的统计特征。

我常用的异常值检测方法有三种：

3σ原则：适用于正态分布的数据，超出均值±3倍标准差的值视为异常
IQR方法：适用于非正态分布，超出Q1-1.5*IQR或Q3+1.5*IQR的值视为异常
物理阈值：根据设备规格书设定硬边界，比如温度不超过85°C

我个人更推荐组合使用。先过物理阈值，再过统计方法。为什么？因为物理阈值能快速干掉那些“离谱”的异常，而统计方法能发现那些“看起来合理但实际异常”的数据点。

def filter_outliers(df, value_col='value', method='iqr', 
                    lower_bound=None, upper_bound=None):
    """
    异常值过滤
    method: 'iqr' 或 '3sigma'
    lower_bound/upper_bound: 物理阈值
    """
    filtered = df.copy()
    
    # 第一步：物理阈值过滤
    if lower_bound is not None:
        filtered = filtered[filtered[value_col] >= lower_bound]
    if upper_bound is not None:
        filtered = filtered[filtered[value_col] <= upper_bound]
    
    # 第二步：统计方法过滤
    if method == 'iqr':
        Q1 = filtered[value_col].quantile(0.25)
        Q3 = filtered[value_col].quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - 1.5 * IQR
        upper = Q3 + 1.5 * IQR
    elif method == '3sigma':
        mean = filtered[value_col].mean()
        std = filtered[value_col].std()
        lower = mean - 3 * std
        upper = mean + 3 * std
    
    filtered = filtered[(filtered[value_col] >= lower) & 
                        (filtered[value_col] <= upper)]
    
    return filtered

一个小技巧：不要直接删除异常值。我习惯的做法是，先把异常值标记出来，单独分析。有时候，异常值本身就是一个重要的信号——比如设备即将故障的前兆。把这些“异常”单独建一个数据集，用于故障预测模型的训练，效果往往不错。

4.4 数据归一化——让不同量纲的数据“平起平坐”

归一化，说白了就是把不同尺度的数据映射到同一个范围。比如温度（0-100°C）和湿度（0-100%RH）虽然数值范围相似，但物理意义不同。更极端的是，电流（0-10A）和电压（0-220V）直接没法比。

常用的归一化方法：

方法	公式	输出范围	适用场景
Min-Max归一化	(x - min) / (max - min)	[0, 1]	数据分布已知，无极端值
Z-score标准化	(x - mean) / std	均值0，标准差1	数据近似正态分布
Robust标准化	(x - median) / IQR	中位数0，IQR为1	数据含异常值

我建议：对于IoT行为基线，优先使用Robust标准化。为什么？因为IoT数据太“脏”了，异常值防不胜防。Min-Max归一化对异常值极其敏感——一个离群点就能把整个缩放比例拉偏。Z-score也好不到哪去，均值和标准差都会被异常值影响。Robust标准化用中位数和四分位距，对异常值鲁棒得多。

from sklearn.preprocessing import RobustScaler

def normalize_features(df, feature_cols, scaler=None):
    """
    数据归一化，使用RobustScaler
    """
    if scaler is None:
        scaler = RobustScaler()
        df[feature_cols] = scaler.fit_transform(df[feature_cols])
    else:
        df[feature_cols] = scaler.transform(df[feature_cols])
    
    return df, scaler

注意：归一化参数（均值、标准差、中位数等）必须只从训练数据中计算，然后用同样的参数去变换测试数据。我见过有人把全部数据一起归一化，然后做训练测试切分——这属于数据泄露，会导致模型性能评估虚高。实际部署时，你会发现模型表现远不如预期。

4.5 本章知识体系

下面这张图，把数据预处理的四个核心步骤串起来了。你可以把它当作一个检查清单，每次处理数据时对照着走一遍。

这四个步骤，我建议你按顺序执行，不要跳步。时间戳对齐是基础，基础不牢，后面全是错的。缺失值处理要“看人下菜碟”，别一刀切。异常值过滤要“手下留情”，别把有用的信号当噪声删了。归一化要“选对工具”，RobustScaler是IoT场景的首选。

数据预处理做好了，后面的行为基线建模就是水到渠成的事。嗯，这一章的内容就到这里，代码片段你可以直接拿去用，但记得根据你自己的数据特点做调整。