2. 钻孔水压水量原始数据质量检查与清洗

为什么原始数据不能直接进入预警

钻孔水压、水量监测数据在采集、传输、入库环节都可能出现三类质量问题：缺失（传感器掉线、通信中断导致时间戳空缺）、异常（毛刺、跳变、量程越界、长时间恒定不变）、重复（重传或补传造成同一时刻多条记录）。若不清洗直接做阈值判断，一个尖峰毛刺就可能触发误报，一段缺失又会让变化率计算失真。因此清洗的目标不是"让数据好看"，而是产出时间戳唯一、等间隔、量纲一致、异常可追溯的分析集。

检查规则与清洗策略

本例采用如下判定口径（示例参数，可按现场采样周期调整）：

• 重复记录：同一 timestamp 出现多条，保留采集时间最早的一条，其余标记为重复并剔除。
• 缺失值：以 10 min 为标称采样间隔重建时间轴，空缺位置记为 NaN；连续缺失 ≤ 3 个点用线性插值补齐，连续缺失 > 3 个点（即 >30 min）不插值，保留 NaN 并在预警分析中跳过该段。
• 异常值：采用中位数绝对偏差（MAD）做稳健判别，|x − median| > 3 × 1.4826 × MAD 判为离群；同时叠加物理量程约束（水压 0–6 MPa、水量 0–50 m³/h），越界直接判异常。异常点先置 NaN，再按缺失规则处理，避免用异常值参与插值。

用 Python 完成检查与清洗

下面代码自包含，直接构造一段含缺失、异常、重复的示例数据，然后按上述规则清洗并输出质量报告。示例数据为教学构造，非现场实测。

import numpy as np
import pandas as pd

# 1) 构造示例原始数据：10 min 间隔，共 60 个标称时刻
rng = pd.date_range("2024-05-01 00:00", periods=60, freq="10min")
base_p = 2.10 + 0.02 * np.sin(np.arange(60) / 6)      # 水压基线 MPa
base_q = 18.0 + 0.5 * np.cos(np.arange(60) / 8)       # 水量基线 m³/h
df = pd.DataFrame({"timestamp": rng,
                   "pressure": base_p,
                   "flow": base_q})

# 人为制造问题
df.loc[10, "pressure"] = np.nan          # 单点缺失
df.loc[20:22, "pressure"] = np.nan       # 连续 3 点缺失（可插值）
df.loc[35:40, "flow"] = np.nan           # 连续 6 点缺失（不插值）
df.loc[15, "pressure"] = 5.80            # 毛刺异常（接近量程上限）
df.loc[45, "flow"] = 62.0                # 越界异常
df = pd.concat([df, df.iloc[[5]]], ignore_index=True)  # 重复记录
df = df.sample(frac=1, random_state=1).reset_index(drop=True)  # 打乱顺序

# 2) 重复记录检查
dup_mask = df.duplicated(subset="timestamp", keep="first")
n_dup = int(dup_mask.sum())
df = df[~dup_mask].sort_values("timestamp").reset_index(drop=True)

# 3) 重建等间隔时间轴，暴露缺失
full_idx = pd.date_range(df["timestamp"].min(), df["timestamp"].max(), freq="10min")
df = df.set_index("timestamp").reindex(full_idx)
df.index.name = "timestamp"
n_missing_raw = int(df["pressure"].isna().sum() + df["flow"].isna().sum())

# 4) 异常值检查：量程 + MAD
LIMITS = {"pressure": (0.0, 6.0), "flow": (0.0, 50.0)}
anomaly_log = []
for col, (lo, hi) in LIMITS.items():
    s = df[col]
    med = s.median()
    mad = (s - med).abs().median()
    thr = 3 * 1.4826 * mad
    out_range = (s < lo) | (s > hi)
    out_mad = (s - med).abs() > thr
    mask = (out_range | out_mad) & s.notna()
    for ts in df.index[mask]:
        anomaly_log.append((ts, col, float(s.loc[ts])))
    df.loc[mask, col] = np.nan

# 5) 缺失处理：连续缺失 ≤3 点线性插值，>3 点保留 NaN
def fill_short_gaps(s, max_gap=3):
    is_na = s.isna()
    grp = (is_na != is_na.shift()).cumsum()
    for _, seg in s.groupby(grp):
        if seg.isna().all() and len(seg) <= max_gap:
            s.loc[seg.index] = np.nan  # 占位，稍后统一插值
    return s.interpolate(method="linear", limit=max_gap, limit_area="inside")

df["pressure"] = fill_short_gaps(df["pressure"])
df["flow"] = fill_short_gaps(df["flow"])

# 6) 质量报告
report = {
    "重复记录剔除数": n_dup,
    "原始缺失单元格": n_missing_raw,
    "异常点判定数": len(anomaly_log),
    "清洗后剩余NaN": int(df.isna().sum().sum()),
    "总记录数": len(df),
}
print(report)
print(pd.DataFrame(anomaly_log, columns=["timestamp", "param", "value"]))

结果判读与通过条件

运行后应看到：重复记录剔除数为 1；异常点判定包含 2024-05-01 02:30 pressure 5.80 与 2024-05-01 07:30 flow 62.0 两条（具体时刻取决于构造顺序，以打印结果为准）；清洗后剩余 NaN 主要来自 flow 的 6 点长缺失段，pressure 的 1 点与 3 点缺失应已被线性插值补齐。判读要点：

• 若"清洗后剩余 NaN"为 0，说明长缺失段被误插值，需检查 limit 与 limit_area 设置。
• 若异常点判定数远大于预期，多为 MAD 阈值过严或数据本身波动大，应结合量程约束复核，而不是直接放宽阈值。
• 插值点必须可追溯：建议在清洗时额外生成 is_interpolated 布尔列，供后续突变特征计算时区分真实值与补值。