4、数据清洗实战：处理缺失值、重复值、异常时间戳

好，咱们进入第四讲。前面几章我们把数据采集和存储搞定了，但说实话，从传感器拿到的原始数据，那叫一个「脏」。我见过太多新手拿到数据直接开搞，结果模型跑出来一塌糊涂，回头一查，原来是数据本身就有问题。

数据清洗，说白了就是给数据「洗澡」。把脏东西洗掉，才能用。今天咱们就聚焦三个最常见的坑：缺失值、重复值、还有那个让人头疼的异常时间戳。

4.1 缺失值处理：别让「空」坑了你

水质监测里，缺失值太常见了。传感器泡在水里，难免出故障。比如电极老化、通信中断、或者干脆就是采样时打了个盹。我当年在太湖做项目时，有一台pH计每到凌晨三点就罢工，连续一周的数据全是NaN，排查了好久才发现是供电电压不稳。

处理缺失值，我的习惯是三步走：

1. 先定位：用 isnull() 或 isna() 看看哪些列、哪些行有问题。
2. 再分析：缺失比例高不高？是随机缺失还是规律性缺失？
3. 最后决策：删、补、还是留？

来看看代码怎么写。我习惯用Pandas，干净利落。

import pandas as pd
import numpy as np

# 模拟一份水质数据
df = pd.DataFrame({
    '时间': pd.date_range('2024-01-01', periods=10, freq='H'),
    'pH': [7.2, 7.1, np.nan, 7.3, 7.0, np.nan, 7.4, 7.2, 7.1, 7.3],
    '浊度': [3.5, 3.6, 3.4, np.nan, 3.7, 3.5, 3.6, np.nan, 3.8, 3.5],
    '溶解氧': [6.2, 6.1, 6.3, 6.2, np.nan, 6.0, 6.4, 6.2, 6.1, 6.3]
})

# 第一步：定位缺失
print(df.isnull().sum())

# 第二步：填充——用前向填充法（适合时间序列）
df_filled = df.fillna(method='ffill')

# 或者用均值填充
df['pH'].fillna(df['pH'].mean(), inplace=True)

4.2 重复值处理：别让数据「胖」了

重复值，说白了就是数据「胖」了。明明该是一条记录，结果来了两条一模一样的。原因很多：传感器重复上报、数据库写入时网络重试、或者就是采集程序写了个bug。

你想想看，如果重复值不处理，统计均值、方差全都会偏。更严重的是，如果你用这些数据训练模型，模型会「记住」重复的模式，导致过拟合。

处理重复值，我一般分两类：

• 完全重复：所有列都一样，直接删。
• 部分重复：比如时间戳一样但数值不同，那就得看哪个是「真」的。

# 检查完全重复
print(df.duplicated().sum())

# 删除完全重复
df_clean = df.drop_duplicates()

# 按时间戳去重，保留第一条
df_clean = df.drop_duplicates(subset=['时间'], keep='first')

# 或者保留最后一条（适合传感器延迟上报）
df_clean = df.drop_duplicates(subset=['时间'], keep='last')

4.3 异常时间戳：时间线乱了，一切白搭

这个坑，我踩过不止一次。异常时间戳，说白了就是时间数据「不正经」。比如：

• 时间戳格式不统一（有的用'2024-01-01'，有的用'01/01/2024'）
• 时间戳超出合理范围（比如2024年的数据里出现了2099年）
• 时间戳不连续（中间跳过了几个小时）
• 时间戳重复（同一个时间点有多条记录）

为什么会这样？传感器时钟漂移、时区设置错误、或者采集程序的时间函数写错了。我记得有一次，某个站点的数据全部比实际时间晚了8小时，排查了一整天才发现是时区没设成UTC+8。

处理异常时间戳，我的标准流程：

1. 统一格式：全部转成 datetime 类型。
2. 范围校验：设定合理的时间窗口，超出范围的标记为异常。
3. 连续性检查：看时间间隔是否一致，有没有跳变。
4. 修正或剔除：能修正的修正，不能修正的剔除。

# 统一时间格式
df['时间'] = pd.to_datetime(df['时间'], errors='coerce')

# 范围校验：只保留2024年的数据
start = pd.Timestamp('2024-01-01')
end = pd.Timestamp('2024-12-31')
df = df[(df['时间'] >= start) & (df['时间'] <= end)]

# 检查时间间隔是否一致（假设是每小时一条）
df['时间差'] = df['时间'].diff()
print(df['时间差'].value_counts())

# 找出时间间隔异常的记录
异常时间戳 = df[df['时间差'] != pd.Timedelta(hours=1)]
print(f"发现 {len(异常时间戳)} 条异常时间戳记录")

4.4 综合实战：一个完整的清洗流程

好了，咱们把上面三个步骤串起来，写一个完整的清洗函数。这个函数我用了好几年，改了好几版，现在分享给你。

def clean_water_quality_data(df):
    """
    水质数据清洗函数
    处理缺失值、重复值、异常时间戳
    """
    # 1. 备份原始数据
    df_original = df.copy()
    
    # 2. 统一时间格式
    df['时间'] = pd.to_datetime(df['时间'], errors='coerce')
    
    # 3. 删除时间戳为NaT的行
    df = df.dropna(subset=['时间'])
    
    # 4. 按时间排序
    df = df.sort_values('时间')
    
    # 5. 删除完全重复的行
    df = df.drop_duplicates()
    
    # 6. 按时间戳去重（保留第一条）
    df = df.drop_duplicates(subset=['时间'], keep='first')
    
    # 7. 填充缺失值（前向填充）
    numeric_cols = ['pH', '浊度', '溶解氧', '温度', '电导率']
    for col in numeric_cols:
        if col in df.columns:
            df[col] = df[col].fillna(method='ffill')
    
    # 8. 检查时间连续性
    df['时间差'] = df['时间'].diff()
    expected_interval = pd.Timedelta(hours=1)
    df['时间正常'] = df['时间差'].apply(
        lambda x: True if pd.isna(x) or x == expected_interval else False
    )
    
    # 9. 标记异常时间戳
    异常记录 = df[df['时间正常'] == False]
    if len(异常记录) > 0:
        print(f"警告：发现 {len(异常记录)} 条时间戳异常记录")
        # 这里可以加修正逻辑，简单起见先剔除
        df = df[df['时间正常'] == True]
    
    # 10. 删除辅助列
    df = df.drop(columns=['时间差', '时间正常'])
    
    print(f"清洗完成：原始 {len(df_original)} 条，清洗后 {len(df)} 条")
    return df

# 使用示例
df_cleaned = clean_water_quality_data(df)

4.5 小结

数据清洗，听起来枯燥，但它是整个数据分析的基石。你想想看，如果数据都是脏的，后面再漂亮的算法也是白搭。

今天咱们讲了三个核心点：

• 缺失值：先定位，再分析，最后决策。时间序列优先用前向填充。
• 重复值：区分完全重复和部分重复，确认原因后再删。
• 异常时间戳：统一格式、范围校验、连续性检查，三步走。

嗯，这一章就到这儿。下一章咱们会讲「异常检测算法实战」，到时候会用上今天清洗好的数据。所以，这一章的代码，建议你好好跑一遍，把清洗函数存起来，后面会反复用到。

记住：数据清洗不是一次性工作，而是贯穿整个项目周期的。每次拿到新数据，都得先过一遍清洗流程。这是习惯，也是专业素养。