📚 课程目录第4章批次记录数据清洗与预处理实战

4. 批次记录数据清洗与预处理实战

拿到一份从发酵车间导出的批次记录，第一印象往往是“数据挺全”，但真正开始分析就会发现：同一时刻被记录了两次、pH 在相邻两分钟从 5.2 跳到 6.8 又跳回、温度有的写 ℃ 有的写华氏度、DCS 时间戳与手工记录时间相差十几分钟。这些问题不处理，后续的批间对比、曲线拟合和异常判定都会失真。本节围绕四类高频问题——重复记录、异常跳变、单位混用、时间错位——完成一次可复现的清洗与标准化实操。

一、先明确清洗目标与判定口径

清洗不是“把数据改好看”，而是让每条记录在时间、量纲、语义三个维度上可比较。我们以某啤酒厂发酵批次 FB-2024-017 为例，原始记录包含罐号、批次号、采集时间、参数名、数值、单位、记录来源（DCS 自动采集 / 人工抄录）等字段。清洗目标设定为：

每个参数在同一时间戳上只保留一条有效值；
剔除物理上不可能的跳变，并对短时缺失做合理插值；
所有参数统一到 SI 或行业惯用单位（温度 ℃、压力 kPa、糖度 °P、pH 无量纲）；
统一时间基准，使自动采集与人工记录落在同一时间轴上。

判定口径必须提前写死，否则清洗过程会变成“凭感觉删数据”。例如温度合理区间设为 8～14 ℃（主发酵阶段），pH 设为 3.8～5.5，糖度降幅单次采样不超过 1.5 °P，相邻记录时间间隔超过 30 min 视为断点。

二、重复记录：按业务主键去重

重复通常来自两个渠道：DCS 在通信重连时补传了同一时间点的数据；人工抄录时把同一行誊写了两遍。去重不能简单用 drop_duplicates() 全字段比对，因为人工记录与 DCS 记录的数值往往有微小差异，全字段比对会漏掉真正的重复。正确做法是定义业务主键：(批次号, 参数名, 采集时间)。同一主键下若有多条，按来源优先级保留——DCS 自动采集优先于人工抄录，因为前者无转录误差。

import pandas as pd

df["采集时间"] = pd.to_datetime(df["采集时间"])
df["来源优先级"] = df["记录来源"].map({"DCS": 0, "人工": 1})

df = (df.sort_values(["批次号", "参数名", "采集时间", "来源优先级"])
        .drop_duplicates(subset=["批次号", "参数名", "采集时间"], keep="first")
        .reset_index(drop=True))

去重后要核对记录数变化。示例批次原始 4820 条，去重后 4613 条，减少 207 条，其中约 180 条来自 DCS 重连补传，27 条来自人工重复誊写。若减少量远超预期，说明主键定义过粗，需要检查是否把不同罐号误并。

三、异常跳变：阈值与变化率双判据

发酵过程的参数变化是连续的，pH 在 5 min 内变化通常不超过 0.1，温度不超过 0.3 ℃。因此单点超限和相邻点变化率过大都应视为可疑。实操中建议“标记而非删除”：新增 qc_flag 字段，把可疑点标为 2，原始值保留，后续分析可选择排除或人工复核。

# 以 pH 为例：合理区间 3.8~5.5，相邻变化率阈值 0.1/5min
df = df.sort_values(["参数名", "采集时间"])
df["prev"] = df.groupby("参数名")["数值"].shift(1)
df["dt_min"] = df.groupby("参数名")["采集时间"].diff().dt.total_seconds() / 60

mask_range = (df["参数名"] == "pH") & (~df["数值"].between(3.8, 5.5))
mask_rate = ((df["参数名"] == "pH") &
             (df["数值"].sub(df["prev"]).abs() / df["dt_min"] > 0.02))
df.loc[mask_range | mask_rate, "qc_flag"] = 2

示例中 pH 在 14:05 记录为 6.82，前后分别为 5.21 和 5.24，单点超限且变化率极大，标记为跳变。复核发现是操作工把“5.22”误录为“6.82”，属于典型转录错误。注意变化率判据要除以实际时间间隔，否则采样间隔不均匀时会误判。

四、单位混用：换算并留痕

同一参数在不同来源下单位不一致是常态：温度表有人写 ℉，压力表有的用 psi，糖度有的用 °Bx。换算必须显式记录系数，不能直接覆盖原值。常用换算：℉→℃ 为 (F-32)×5/9；psi→kPa 为 ×6.89476；°Bx 与 °P 在麦汁浓度范围内近似相等，可按 1:1 处理但需在备注中说明。

def to_celsius(v, unit):
    return (v - 32) * 5 / 9 if unit == "℉" else v

df["数值_标准"] = df.apply(
    lambda r: to_celsius(r["数值"], r["单位"]) if r["参数名"] == "温度" else r["数值"],
    axis=1)
df.loc[df["单位"].isin(["℉", "psi"]), "qc_flag"] = 3

换算后要抽查边界值：例如 12.5 ℃ 对应 54.5 ℉，若换算结果落在 12.4～12.6 之间即认为正确。单位字段本身也应标准化为统一符号，避免“摄氏度”“℃”“C”三种写法并存。

五、时间错位：以 DCS 时钟为基准对齐

人工记录通常按整点或巡检时刻填写，与 DCS 时间戳存在系统性偏移。示例批次中，人工记录整体比 DCS 晚 12 min，原因是值班室挂钟未校准。处理方法是选取一个双方都记录的稳定参数（如温度），用互相关或简单差值中位数估计偏移量 Δt，再对人工记录整体平移。平移后统一重采样到 5 min 网格，网格点上的值用线性插值获得。

# 估计人工记录相对 DCS 的偏移（示例：中位差 12 min）
delta = pd.Timedelta(minutes=12)
df.loc[df["记录来源"] == "人工", "采集时间"] -= delta
df.loc[df["记录来源"] == "人工", "qc_flag"] = 4

# 重采样到 5 min 网格
pivot = df.pivot_table(index="采集时间", columns="参数名",
                       values="数值_标准", aggfunc="mean")
grid = pivot.resample("5min").mean().interpolate(method="time", limit=6)

插值要设上限：limit=6 表示最多连续补 6 个网格点（30 min），超过则保留 NaN 并标记为长断点。发酵中后期采样间隔拉长时，硬插值会制造虚假的平滑曲线，反而掩盖真实异常。

六、结果核验与交付

清洗完成后不能只看行数，要做三项核验：其一，各参数的时间覆盖是否连续，断点是否都有标记；其二，换算后的数值是否落在物理区间内；其三，随机抽取 10 个时间点，与原始记录逐条对照，确认平移和插值没有引入错位。示例批次最终输出 4613 条标准化记录，其中插值补全 86 条、跳变标记 9 条、单位换算 214 条、时间平移 312 条，全部带 qc_flag 可追溯。

建议把清洗规则写成配置文件（阈值、换算系数、偏移量、插值上限），每次处理新批次时只改配置不改代码，既保证口径一致，也便于复盘时说明“这条数据为什么被标记”。

经过上述处理，批次记录才具备进入批间对比和趋势分析的条件。清洗的价值不在于让数据“看起来干净”，而在于每一条被修改、被标记、被补全的记录都有明确依据，后续任何结论都能回溯到原始记录。