1. 科研数据清洗与缺失值处理方案设计

任务对象与清洗流程设计

本案例使用的原始数据集为一份模拟的植物生长实验记录，共 120 条观测，字段包括：sample_id（样本编号）、group（处理组 A/B/C）、height_cm（株高）、leaf_area_cm2（叶面积）、chlorophyll_spad（叶绿素 SPAD 值）、biomass_g（生物量）。数据中存在三类问题：完全重复记录、超出合理范围的异常值、以及不同比例的缺失值。清洗流程需要可复用，因此把每一步封装为独立函数，并让缺失值填充策略成为可切换参数。

可复用的清洗函数实现

下面用 Python + pandas 实现完整流程。核心设计是把每个清洗步骤写成独立函数，主流程按顺序调用，缺失值填充策略通过参数传入，便于后续对比不同策略的效果。

import pandas as pd
import numpy as np

# ---------- 构造示例原始数据（含重复、异常、缺失） ----------
np.random.seed(42)
n = 120
raw = pd.DataFrame({
    "sample_id": [f"S{i:03d}" for i in range(1, n + 1)],
    "group": np.random.choice(["A", "B", "C"], n),
    "height_cm": np.random.normal(25, 5, n),
    "leaf_area_cm2": np.random.normal(40, 8, n),
    "chlorophyll_spad": np.random.normal(35, 4, n),
    "biomass_g": np.random.normal(12, 3, n),
})

# 人为注入问题：重复记录 5 条
raw = pd.concat([raw, raw.sample(5, random_state=1)], ignore_index=True)

# 人为注入异常值：株高 200cm、叶面积 -10
raw.loc[10, "height_cm"] = 200
raw.loc[20, "leaf_area_cm2"] = -10

# 人为注入缺失值：不同字段不同比例
for col, frac in [("height_cm", 0.10), ("leaf_area_cm2", 0.15),
                  ("chlorophyll_spad", 0.08), ("biomass_g", 0.12)]:
    idx = raw.sample(frac=frac, random_state=7).index
    raw.loc[idx, col] = np.nan

# ---------- 步骤 1：去重 ----------
def remove_duplicates(df):
    before = len(df)
    df = df.drop_duplicates(subset="sample_id", keep="first").copy()
    return df, before - len(df)

# ---------- 步骤 2：异常值置为缺失 ----------
def flag_outliers(df, rules):
    """rules: {列名: (下限, 上限)}，超出范围置为 NaN"""
    df = df.copy()
    count = {}
    for col, (lo, hi) in rules.items():
        mask = (df[col] < lo) | (df[col] > hi)
        count[col] = int(mask.sum())
        df.loc[mask, col] = np.nan
    return df, count

# ---------- 步骤 3：缺失值填充（策略可切换） ----------
def impute_missing(df, strategy="mean", group_col="group"):
    df = df.copy()
    num_cols = df.select_dtypes(include=np.number).columns
    if strategy == "mean":
        df[num_cols] = df[num_cols].fillna(df[num_cols].mean())
    elif strategy == "median":
        df[num_cols] = df[num_cols].fillna(df[num_cols].median())
    elif strategy == "group_mean":
        df[num_cols] = df.groupby(group_col)[num_cols].transform(
            lambda s: s.fillna(s.mean()))
    else:
        raise ValueError(f"未知策略: {strategy}")
    return df

# ---------- 步骤 4：质量报告 ----------
def quality_report(raw, cleaned, dup_removed, outlier_counts, strategy):
    num_cols = raw.select_dtypes(include=np.number).columns
    rows = []
    for col in num_cols:
        rows.append({
            "字段": col,
            "原始缺失率": f"{raw[col].isna().mean():.1%}",
            "异常值置缺失数": outlier_counts.get(col, 0),
            "填充后缺失数": int(cleaned[col].isna().sum()),
            "原始均值": round(raw[col].mean(), 2),
            "填充后均值": round(cleaned[col].mean(), 2),
            "原始标准差": round(raw[col].std(), 2),
            "填充后标准差": round(cleaned[col].std(), 2),
        })
    report = pd.DataFrame(rows)
    report.attrs["策略"] = strategy
    report.attrs["删除重复数"] = dup_removed
    return report

# ---------- 主流程 ----------
RULES = {
    "height_cm": (5, 60),
    "leaf_area_cm2": (5, 90),
    "chlorophyll_spad": (15, 60),
    "biomass_g": (2, 30),
}

def run_pipeline(raw_df, strategy):
    df, dup_removed = remove_duplicates(raw_df)
    df, outlier_counts = flag_outliers(df, RULES)
    cleaned = impute_missing(df, strategy=strategy)
    report = quality_report(df, cleaned, dup_removed, outlier_counts, strategy)
    return cleaned, report

# 对比三种策略
for s in ["mean", "median", "group_mean"]:
    cleaned, report = run_pipeline(raw, s)
    print(f"\n===== 策略: {s} | 删除重复: {report.attrs['删除重复数']} 条 =====")
    print(report.to_string(index=False))

结果判读与策略选择依据

运行上述代码后，重点观察质量报告中的三组对比：

• 填充后缺失数应全部为 0，否则说明填充逻辑遗漏了字段。
• 原始均值 vs 填充后均值：均值填充会使填充后均值与原始均值接近，但标准差明显缩小；中位数填充在偏态字段上均值偏移更小。
• 原始标准差 vs 填充后标准差：若标准差缩小超过 10%，需警惕后续 t 检验或方差分析中显著性被高估。

策略选择建议：若各处理组均值差异明显（如本例 A/B/C 三组），优先使用 group_mean，因为它保留了组间结构；若字段分布偏态严重，使用 median；只有在缺失率很低（<5%）且分布近似正态时，才使用全局 mean。