1. 回灌试验原始数据清洗与格式统一

多源记录的对齐难点

回灌试验现场通常同时存在三类记录：数据采集仪（水位、压力、温度，多为等间隔采样）、流量计累积记录（脉冲或累积量，需差分）、人工班报（启停泵、阀门调整、异常备注）。它们的差异集中在四点：时间基准不同（采集仪用设备时钟、班报用人工时刻）、时间间隔不同（1 min / 5 min / 事件触发）、单位混用（水位 m 与 cm、压力 MPa 与 kPa 与 bar、流量 m³/h 与 L/s、温度 ℃ 与 ℉）、缺失与重复（停机空档、通信丢包、同一时刻多条记录）。

清洗的目标不是“把数据变好看”，而是产出一张统一时间步长、单位一致、可追溯的结构化表：以固定步长（本例取 5 min）为时间轴，每个物理量一列，缺失显式标记，并保留原始来源与处理痕迹。

清洗脚本的关键实现

下面用一段自洽的 Python 脚本完成上述流程。示例数据在脚本内构造，可直接运行；真实使用时把 load_* 函数替换为读取 CSV/Excel 即可，其余逻辑不变。

import pandas as pd
import numpy as np

STEP = "5min"          # 统一时间步长
TZ = "Asia/Shanghai"   # 统一时区

# ---------- 1. 构造示例原始数据（真实场景改为读文件） ----------
raw_level = pd.DataFrame({
    "time": ["2024-05-10 08:00:00", "2024-05-10 08:01:00", "2024-05-10 08:02:00",
             "2024-05-10 08:03:00", "2024-05-10 08:04:00", "2024-05-10 08:05:00",
             "2024-05-10 08:06:00", "2024-05-10 08:07:00", "2024-05-10 08:08:00"],
    "H_cm": [1248.0, 1249.1, 1250.2, 1251.3, 1252.4, 1253.5, 1254.6, 1255.7, 1256.8],
    "P_bar": [0.21, 0.22, 0.23, 0.24, 0.25, 0.26, 0.27, 0.28, 0.29],
    "T_F": [64.8, 64.9, 65.0, 65.1, 65.2, 65.3, 65.4, 65.5, 65.6],
})

raw_flow = pd.DataFrame({
    "time": ["2024-05-10 08:00", "2024-05-10 08:05", "2024-05-10 08:10",
             "2024-05-10 08:15", "2024-05-10 08:20", "2024-05-10 08:25"],
    "V_m3": [0.0, 3.03, 6.08, 9.12, 12.17, 15.21],   # 累积量
})

raw_log = pd.DataFrame({
    "time": ["2024-05-10 08:00", "2024-05-10 08:50"],
    "event": ["start_pump", "comm_lost"],
})

# ---------- 2. 时间解析与基准统一 ----------
def to_dt(s):
    return pd.to_datetime(s).dt.tz_localize(TZ, nonexistent="shift_forward")

raw_level["time"] = to_dt(raw_level["time"])
raw_flow["time"]  = to_dt(raw_flow["time"])
raw_log["time"]   = to_dt(raw_log["time"])

# ---------- 3. 单位换算到基准单位 ----------
raw_level["H_m"]   = raw_level["H_cm"] / 100.0          # cm -> m
raw_level["P_MPa"] = raw_level["P_bar"] * 0.1           # bar -> MPa
raw_level["T_C"]   = (raw_level["T_F"] - 32) * 5 / 9    # F -> C

# ---------- 4. 去重、排序 ----------
def dedup_sort(df, cols):
    df = df.sort_values("time")
    df = df.groupby("time", as_index=False)[cols].mean()  # 同刻取均值
    return df

level = dedup_sort(raw_level, ["H_m", "P_MPa", "T_C"]).set_index("time")
flow  = dedup_sort(raw_flow, ["V_m3"]).set_index("time")

# ---------- 5. 累积量差分 -> 瞬时流量 ----------
flow["dV"] = flow["V_m3"].diff()
flow["dt_h"] = flow.index.to_series().diff().dt.total_seconds() / 3600.0
flow["Q_m3h"] = flow["dV"] / flow["dt_h"]
flow.loc[flow["dV"] < 0, "Q_m3h"] = np.nan   # 换表/复位
flow.loc[flow["dt_h"] <= 0, "Q_m3h"] = np.nan

# ---------- 6. 重采样到统一网格 ----------
grid = pd.date_range(level.index.min(), level.index.max(), freq=STEP, tz=TZ)
level_r = level.reindex(grid).interpolate(method="time", limit=2)  # 最多补 2 步
flow_r  = flow["Q_m3h"].reindex(grid).interpolate(method="time", limit=2)

# ---------- 7. 缺失标记与质量列 ----------
out = pd.DataFrame(index=grid)
out["Q_m3h"] = flow_r
out["P_MPa"] = level_r["P_MPa"]
out["H_m"]   = level_r["H_m"]
out["T_C"]   = level_r["T_C"]
out["flag"]  = np.where(out[["Q_m3h","P_MPa","H_m","T_C"]].isna().any(axis=1),
                        "missing", "ok")
out["src"]   = "level+flow"

# 事件对齐到最近网格
for _, r in raw_log.iterrows():
    idx = out.index.get_indexer([r["time"]], method="nearest")[0]
    out.loc[out.index[idx], "flag"] = "event:" + r["event"]

out = out.reset_index().rename(columns={"index": "timestamp"})
print(out.head(12).to_string(index=False))
out.to_csv("recharge_clean_5min.csv", index=False)

结果判读与验收

脚本输出应满足以下可检查条件，逐条核对即可判断清洗是否合格：

以示例数据为例：08:00–08:45 共 10 个网格点，08:50 因原始水位记录在 08:08 后中断且超出插值上限，被标记为 missing；08:00 的 start_pump 事件对齐到首个网格点，08:50 的 comm_lost 与缺失行重合，便于后续分析时区分“设备停机”与“通信中断”。