1. 环境监测数据异常初筛与分类

为什么“初筛”不能只看一个阈值

环境实验室的原始监测数据往往同时存在三类问题：离群点（数值偏离总体分布）、缺失值（未采样、仪器掉线、记录漏填）、逻辑矛盾（数值本身合法但违反业务约束，如“检出浓度 > 采样体积对应上限”“pH=9.6 却标注为中性样品”“平行样相对偏差超限”）。如果只用一个固定阈值筛，会把正常的浓度波动误判为离群，也会漏掉“数值正常但逻辑错误”的记录。因此初筛的核心不是“找最大值”，而是建立一条可复用的分类判断链：先判缺失，再判逻辑，最后用分布方法判离群，并对每条异常给出类型标签与处置建议。

判断链的四个判定层

下面这张图给出本节要落地的分类流程。注意判定顺序很关键：缺失值必须先处理，否则后续统计量（均值、标准差、四分位距）会被空值污染；逻辑矛盾优先于统计离群，因为一条“浓度 500 mg/L 但采样体积 0 mL”的记录在统计上可能并不极端，却一定是错的。

用一组示例数据走完判断链

假设某批次 8 条 COD 监测记录（单位 mg/L），并附带采样体积与平行样信息。为便于演示，数据为教学示例，非实测结果。

第1层（缺失）：S3 的 COD 为空，直接标“缺失值”，不参与后续统计。第2层（逻辑）：S4 采样体积为 0，浓度却存在，属于逻辑矛盾；S5 平行样 RD=28% 超过常用 20% 控制限，也归为逻辑矛盾（精密度不达标）。第3层（离群）：对剩余有效值 {42,45,46,43,210} 排序为 42,43,45,46,210。Q1=43，Q3=46，IQR=3，上界=46+1.5×3=50.5。S8 的 210 远超上界，标“统计离群”。

可复用的初筛脚本（Python）

把上面的判断链固化成函数，输入一张 DataFrame，输出带 异常类型 与 处置建议 两列的结果表。关键点：缺失判定要覆盖空串、NA、占位符；逻辑校验用业务阈值参数化；离群按指标分组计算 IQR。

import pandas as pd
import numpy as np

def screen_anomalies(df, value_col, volume_col=None,
                     rd_col=None, rd_limit=20.0,
                     method_upper=None, group_col=None):
    """
    df: 原始数据表
    value_col: 浓度列名
    volume_col: 采样体积列名（可选）
    rd_col: 平行样相对偏差列名（可选）
    rd_limit: 平行样RD控制限(%)
    method_upper: 方法检出上限（可选）
    group_col: 按指标分组计算IQR的列名（可选）
    返回：新增 异常类型 / 处置建议 两列
    """
    out = df.copy()
    out["异常类型"] = "正常"
    out["处置建议"] = "纳入统计"

    # 第1层：缺失
    raw = out[value_col]
    missing_mask = raw.isna() | raw.astype(str).str.strip().isin(["", "-", "NA", "N/A", "null"])
    out.loc[missing_mask, "异常类型"] = "缺失值"
    out.loc[missing_mask, "处置建议"] = "补采或标记剔除，不参与统计"

    # 第2层：逻辑矛盾（仅对非缺失行）
    valid = ~missing_mask
    if volume_col is not None:
        vol_bad = valid & (pd.to_numeric(out[volume_col], errors="coerce") <= 0)
        out.loc[vol_bad, "异常类型"] = "逻辑矛盾"
        out.loc[vol_bad, "处置建议"] = "核对采样体积原始记录"
    if rd_col is not None:
        rd_bad = valid & (pd.to_numeric(out[rd_col], errors="coerce") > rd_limit)
        out.loc[rd_bad, "异常类型"] = "逻辑矛盾"
        out.loc[rd_bad, "处置建议"] = f"平行样RD超{rd_limit}%，重测"
    if method_upper is not None:
        up_bad = valid & (pd.to_numeric(out[value_col], errors="coerce") > method_upper)
        out.loc[up_bad, "异常类型"] = "逻辑矛盾"
        out.loc[up_bad, "处置建议"] = "超出方法上限，稀释复测"

    # 第3层：分布离群（仅对仍为“正常”的行）
    still_ok = out["异常类型"] == "正常"
    num = pd.to_numeric(out[value_col], errors="coerce")
    if group_col is not None:
        groups = out[group_col]
    else:
        groups = pd.Series(["all"] * len(out), index=out.index)

    for g, idx in groups.groupby(groups).groups.items():
        sub = num.loc[idx][still_ok.loc[idx]].dropna()
        if len(sub) < 4:
            continue
        q1, q3 = sub.quantile(0.25), sub.quantile(0.75)
        iqr = q3 - q1
        lo, hi = q1 - 1.5 * iqr, q3 + 1.5 * iqr
        flag = (num.loc[sub.index] < lo) | (num.loc[sub.index] > hi)
        out.loc[sub.index[flag], "异常类型"] = "统计离群"
        out.loc[sub.index[flag], "处置建议"] = "查仪器/消解/空白，保留待复核"

    return out

# 教学示例数据
df = pd.DataFrame({
    "编号": ["S1","S2","S3","S4","S5","S6","S7","S8"],
    "指标": ["COD"]*8,
    "COD": [42,45,np.nan,48,44,46,43,210],
    "采样体积": [50,50,50,0,50,50,50,50],
    "平行样RD": [3,5,np.nan,4,28,6,2,7],
})
res = screen_anomalies(df, "COD", volume_col="采样体积",
                       rd_col="平行样RD", rd_limit=20,
                       group_col="指标")
print(res[["编号","COD","异常类型","处置建议"]])

运行后预期输出（教学示例，非实测）：S3 为缺失值，S4、S5 为逻辑矛盾，S8 为统计离群，其余为正常。若把 rd_limit 改为 30，则 S5 会回到正常——这说明阈值是业务约定，必须在报告中写明所用控制限，否则初筛结论不可复现。