1. 农产品产地环境数据清洗与异常值识别

任务对象与数据概况

某蔬菜基地提供两份原始记录：土壤温湿度记录（每 30 分钟一条，含 3 个大棚 × 2 个深度）与气象站记录（每小时一条，含气温、降水、风速）。两份数据都存在典型脏数据问题：传感器掉线导致的缺失、网关重传导致的重复时间戳、以及极端天气或探头故障造成的离群值。本案例用 Python（pandas + numpy）完成清洗，输出一张可直接用于产量分析的干净宽表。

先明确判定口径，避免后续反复：

• 缺失：时间戳存在但数值为 NaN，或整条记录缺失（需按时间轴补齐）。
• 重复：同一设备同一时间戳出现多条记录，保留最后一条（网关重传通常后到为准）。
• 异常：物理不可能值（如土壤湿度 > 100%、温度 < -30℃）直接置为 NaN；统计离群值用箱线图 IQR 与 3σ 双准则标记，只标记不删除，保留 is_outlier 标志列供后续分析权衡。

关键实现代码

下面代码自包含，可直接运行。示例数据用固定随机种子生成，模拟真实记录中的缺失、重复与离群值，便于验证清洗逻辑是否生效。

import numpy as np
import pandas as pd

rng = np.random.default_rng(42)

# ---------- 1. 构造示例原始数据 ----------
ts = pd.date_range("2024-05-01", periods=48, freq="30min")  # 24 小时
base = 22 + 6 * np.sin(np.linspace(0, 2 * np.pi, 48))       # 日变化温度
soil = pd.DataFrame({
    "device_id": "GH01-D10",
    "ts": ts,
    "soil_temp": base + rng.normal(0, 0.4, 48),
    "soil_moist": 55 + rng.normal(0, 2, 48),
})
# 注入问题：缺失、重复、离群、物理越界
soil.loc[5, "soil_temp"] = np.nan
soil.loc[12, "soil_moist"] = np.nan
soil.loc[20, "soil_temp"] = 68.0          # 探头故障离群
soil.loc[30, "soil_moist"] = 130.0        # 物理越界
dup = soil.iloc[[8]].copy()
dup["soil_temp"] = dup["soil_temp"] + 0.3  # 重传值略有差异
raw = pd.concat([soil, dup], ignore_index=True)

# ---------- 2. 时间轴对齐 + 去重 ----------
raw["ts"] = pd.to_datetime(raw["ts"])
raw = raw.sort_values(["device_id", "ts"])
clean = raw.drop_duplicates(subset=["device_id", "ts"], keep="last").copy()

# ---------- 3. 物理边界过滤 ----------
bounds = {"soil_temp": (-30, 60), "soil_moist": (0, 100)}
for col, (lo, hi) in bounds.items():
    clean.loc[(clean[col] < lo) | (clean[col] > hi), col] = np.nan

# ---------- 4. 缺失填补（限幅线性插值） ----------
clean = clean.set_index("ts").asfreq("30min")   # 补齐缺失时间戳
for col in ["soil_temp", "soil_moist"]:
    filled = clean[col].interpolate(method="time", limit=2)
    clean[col] = filled

# ---------- 5. 异常值双准则标记 ----------
def flag_outlier(s):
    q1, q3 = s.quantile([0.25, 0.75])
    iqr = q3 - q1
    iqr_hit = (s < q1 - 1.5 * iqr) | (s > q3 + 1.5 * iqr)
    mu, sigma = s.mean(), s.std()
    sig_hit = (s < mu - 3 * sigma) | (s > mu + 3 * sigma)
    return (iqr_hit | sig_hit).astype(int)

for col in ["soil_temp", "soil_moist"]:
    clean[f"{col}_outlier"] = flag_outlier(clean[col])

clean["is_outlier"] = clean[["soil_temp_outlier", "soil_moist_outlier"]].max(axis=1)
print(clean[["soil_temp", "soil_moist", "is_outlier"]].head(10))
print("异常标记数:", int(clean["is_outlier"].sum()))

结果判读与验证

运行后应观察到：第 20 条附近 soil_temp 被标记为异常（68℃ 远超日变化范围），第 30 条 soil_moist 因 130% 越界先被置 NaN 再被插值，不会进入异常统计。验证清洗是否达标，可检查以下三项：