1. 食品QC原始检测数据清洗与异常值识别

任务边界：为一批理化检测原始记录建立可追溯的质量基线

本案例的对象是一批食品理化检测原始记录（示例为 40 条，含水分、酸价、过氧化值、菌落总数等字段）。目标不是把数据“洗干净”，而是回答三个QC问题：哪些记录不可用、哪些记录可疑、每条被处理的数据依据什么规则。因此清洗动作必须与规则编号绑定，形成可复核的清洗日志与异常清单。

先明确三条判定口径，后续所有处理都引用它们：

• 缺失值：关键字段（样品编号、检测项目、结果值、检出限）为空，或结果值无法解析为数值。缺失不填补，直接标记为“不可判定”，进入异常清单。
• 重复记录：同一“样品编号 + 检测项目 + 检测日期”出现多条，保留首次录入，其余标记为重复，不参与统计。
• 离群点：按检测项目分组，用 IQR 规则（低于 Q1−1.5×IQR 或高于 Q3+1.5×IQR）识别统计离群；同时叠加业务阈值（如水分含量超出产品标准限值、结果为负值）作为强制异常。两者取并集，但需在日志中区分“统计离群”与“业务越限”。

清洗规则与数据流

下图给出本批数据的处理链：原始记录先做字段级校验，再依次经过重复判定、统计离群判定与业务阈值判定，最终分流为“可用数据集”和“异常清单”，所有分流动作写入清洗日志。

用 Python 落地：规则编号 + 日志 + 异常清单

下面代码可直接运行，输入为示例数据（内嵌，便于复现）。核心设计是每条处理动作都写入日志，日志字段固定为：记录号、规则编号、字段、原值、处理动作、判定依据。这样任何一条被剔除或标记的数据都能回溯到具体规则。

import pandas as pd
import numpy as np

# ---------- 示例原始记录（教学示例数据，非实测） ----------
raw = pd.DataFrame({
    "记录号": [f"R{i:03d}" for i in range(1, 41)],
    "样品编号": ["SP001","SP002","SP003","SP004","SP005","SP006","SP007","SP008","SP009","SP010",
               "SP011","SP012","SP013","SP014","SP015","SP016","SP017","SP018","SP019","SP020",
               "SP021","SP022","SP023","SP024","SP025","SP026","SP027","SP028","SP029","SP030",
               "SP031","SP032","SP033","SP034","SP035","SP036","SP037","SP038","SP039","SP040"],
    "检测项目": ["水分"]*10 + ["酸价"]*10 + ["过氧化值"]*10 + ["菌落总数"]*10,
    "结果值": [12.1,12.3,12.0,12.2,12.4,12.1,12.2,12.3,12.0,12.2,
             1.2,1.3,1.1,1.4,1.2,1.3,1.2,1.1,1.3,1.2,
             0.12,0.13,0.11,0.14,0.12,0.13,0.12,0.11,0.13,0.12,
             120,130,110,140,120,130,120,110,130,120],
    "检出限": [0.1]*40,
    "检测日期": ["2024-03-01"]*40
})

# 人为注入异常，模拟真实原始记录中的问题
raw.loc[3, "结果值"] = np.nan          # 缺失
raw.loc[7, "结果值"] = "未检出"         # 不可解析
raw.loc[12, "结果值"] = -0.5            # 负值（业务越限）
raw.loc[15, "结果值"] = 12.0            # 重复记录（与 R011 同键）
raw.loc[22, "结果值"] = 0.85            # 统计离群（过氧化值）
raw.loc[28, "结果值"] = 18.5            # 业务越限（水分超限值 14.0）

# ---------- 业务阈值配置 ----------
LIMITS = {"水分": 14.0, "酸价": 3.0, "过氧化值": 0.5, "菌落总数": 1000}

log_rows = []
def log(rec, rule, field, old, action, basis):
    log_rows.append({"记录号": rec, "规则编号": rule, "字段": field,
                     "原值": old, "处理动作": action, "判定依据": basis})

df = raw.copy()
df["结果值_数值"] = pd.to_numeric(df["结果值"], errors="coerce")

# ---------- R1 字段校验 ----------
for idx, row in df.iterrows():
    if pd.isna(row["结果值_数值"]):
        log(row["记录号"], "R1", "结果值", row["结果值"], "标记不可判定", "结果值缺失或无法解析为数值")
    if pd.isna(row["检出限"]) or row["检出限"] <= 0:
        log(row["记录号"], "R1", "检出限", row["检出限"], "标记不可判定", "检出限缺失或非正")

# ---------- R2 重复判定 ----------
key = ["样品编号", "检测项目", "检测日期"]
dup_mask = df.duplicated(subset=key, keep="first")
for idx in df[dup_mask].index:
    log(df.loc[idx, "记录号"], "R2", "样品编号+检测项目+检测日期",
        f'{df.loc[idx,"样品编号"]}|{df.loc[idx,"检测项目"]}|{df.loc[idx,"检测日期"]}',
        "标记重复，不参与统计", "同键记录已存在，保留首次录入")

# ---------- R3 统计离群（IQR） ----------
outlier_mask = pd.Series(False, index=df.index)
for proj, grp in df.groupby("检测项目"):
    vals = grp["结果值_数值"].dropna()
    if len(vals) < 4:
        continue
    q1, q3 = vals.quantile(0.25), vals.quantile(0.75)
    iqr = q3 - q1
    lo, hi = q1 - 1.5*iqr, q3 + 1.5*iqr
    for idx in grp.index:
        v = df.loc[idx, "结果值_数值"]
        if pd.notna(v) and (v < lo or v > hi):
            outlier_mask[idx] = True
            log(df.loc[idx, "记录号"], "R3", "结果值", v,
                "标记统计离群", f"{proj} IQR 区间 [{lo:.3f}, {hi:.3f}]")

# ---------- R4 业务阈值 ----------
for idx, row in df.iterrows():
    v = row["结果值_数值"]
    if pd.isna(v):
        continue
    if v < 0:
        log(row["记录号"], "R4", "结果值", v, "标记业务越限", "结果值为负")
    limit = LIMITS.get(row["检测项目"])
    if limit is not None and v > limit:
        log(row["记录号"], "R4", "结果值", v, "标记业务越限", f"超出{row['检测项目']}限值 {limit}")

# ---------- 汇总输出 ----------
log_df = pd.DataFrame(log_rows)
bad_ids = set(log_df["记录号"])
clean_df = df[~df["记录号"].isin(bad_ids)].copy()
clean_df = clean_df.drop(columns=["结果值_数值"])

print("清洗日志条数:", len(log_df))
print(log_df.to_string(index=False))
print("\n可用记录数:", len(clean_df), "/", len(df))
print("异常记录数:", len(bad_ids))

结果判读与基线确认

运行后应得到一份日志表和一份可用数据集。判读时重点看三件事：

• 规则命中分布：若 R1 命中过多，说明原始记录录入质量差，应先回溯采样与录入环节，而不是继续做统计；若 R3 命中集中在某一项目，需检查该项目检测方法或仪器是否存在系统偏差。
• 重复与缺失是否被误判：重复判定只保留首次录入，若首次录入本身是错误值，需人工复核后调整保留策略，并在日志中追加“人工复核”动作。
• 业务越限与统计离群是否重叠：同一记录同时命中 R3 和 R4 时，异常清单中应保留两条依据，便于后续原因分析区分“方法波动”与“真实超标”。