数据预处理：给时序数据“洗个澡”

做安全分析这么多年，我有个很深的体会：数据预处理往往比模型本身更磨人。说句实在话，原始的时间序列就像刚挖出来的矿石，杂质多得吓人。你要是不花点功夫清理，后面再好的算法也白搭。

今天咱们就聊聊这块 - 如何给Secrets的访问时序数据做预处理。说白了，就是从脏数据里提炼出有价值的信息。

缺失值处理 - 别让空值带偏了你

我记得第一次实战分析时，直接跳过了缺失值检查。结果模型训练出来，预测全在凌晨三四点跳变。查了半天才发现，夜间的数据本来就有大量缺失。

缺失值是怎么产生的？我总结了几种常见原因：

• 服务暂停：Secrets管理服务重启或维护，期间没有日志产生
• 网络问题：日志传输丢包，导致部分时间点数据丢失
• 采样策略：某些系统只在状态变化时记录，空闲时段就是空值

处理策略上，我个人习惯这么选：

异常点筛除 - 小心别把宝贝当垃圾扔了

异常点筛除是最容易翻车的地方。你想想看，我们的目标就是检测异常访问行为，结果你先把异常点筛掉了，那还检测什么？

所以这里要分清楚两类“异常”：

• 噪声异常：传感器故障、日志错乱导致的数据毛刺。这类要剔除。
• 事件异常：真实的异常访问行为。这类我们要保留，甚至需要标注出来。

我常用的筛除方法是3σ原则和四分位距法。但光靠统计不行，得结合领域知识。举个例子：

import numpy as np
import pandas as pd

def clean_noise(series, method='iqr', threshold=1.5):
    """
    筛除噪声异常，保留事件异常
    
    个人经验：阈值设1.5倍IQR通常够用了
    如果数据本身波动较大，可以放宽到3倍
    """
    if method == 'iqr':
        Q1 = series.quantile(0.25)
        Q3 = series.quantile(0.75)
        IQR = Q3 - Q1
        lower = Q1 - threshold * IQR
        upper = Q3 + threshold * IQR
        
        # 标记哪些是噪声（不要直接删除，先标记）
        noise_mask = (series < lower) | (series > upper)
        return series.mask(noise_mask)  # 异常点置为NaN，后续插值处理
    else:
        mean = series.mean()
        std = series.std()
        z_scores = np.abs((series - mean) / std)
        noise_mask = z_scores > 3
        return series.mask(noise_mask)

重采样与插值 - 把时间对齐了再说

实际生产环境里，不同服务的时间粒度差别很大。有的每秒记录一次，有的十分钟才记一笔。如果不做重采样，你根本没法做关联分析。

选择重采样的频率，我一般遵循这个原则：

• 以数据最稀疏的那个服务为基准
• 至少保留统计意义上的有效样本数
• 考虑业务峰值时段的细粒度需求

常用方法对比：

插值方法这块，我踩过不少坑。一开始图省事用线性插值，结果在拐点处误差特别大。后来改用三次样条插值，效果好很多，但计算量也上来了。

def smart_resample(df, freq='5min', method='linear'):
    """
    智能重采样，自动处理缺失时段
    
    :param df: DataFrame, index必须是DateTimeIndex
    :param freq: 目标频率, 如'5min', '1H'
    :param method: 插值方法: 'linear','spline','polynomial'
    """
    # 先重采样（聚合）
    resampled = df.resample(freq).agg({
        'access_count': 'sum',
        'unique_users': 'nunique',
        'error_count': 'mean'
    })
    
    # 再插值（填充空值）
    if method == 'linear':
        return resampled.interpolate(method='linear')
    elif method == 'spline':
        return resampled.interpolate(method='spline', order=3)
    elif method == 'polynomial':
        return resampled.interpolate(method='polynomial', order=5)

完整的预处理流程

说了这么多，我画了张流程图把整个过程串起来。这张图是我自己项目中一直用的框架，经过不少次迭代才打磨成现在这样。