4. 基于固定阈值的钻孔水压突升预警规则实现

规则设计：把“突升”定义成可计算的判据

固定阈值预警的核心不是“水压变大就报警”，而是把突升定义为一个可复算的判据。本例采用相邻采样点差分作为单位时间上升量：设清洗后水压序列为 P[i]，采样间隔为 dt（分钟），则

dP[i] = (P[i] - P[i-1]) / dt，单位 kPa/min。当 dP[i] ≥ TH 时触发预警。为抑制单点毛刺，增加两个工程约束：连续确认数 N（连续 N 个点满足才报警）与冷却时间（同一事件不重复刷屏）。触发时记录触发时刻、触发点前后各 K 个采样点作为证据片段。

规则判断逻辑实现

下面是一段可直接运行的 Python 实现，输入为清洗后的 (timestamp, pressure) 序列，输出预警事件列表。关键点：差分按真实时间间隔计算，避免采样间隔不均导致误判；连续确认用计数器实现；冷却时间用“上次触发时刻”控制。

from datetime import datetime, timedelta

def detect_pressure_surge(records, th=0.8, n_confirm=2, cooldown_min=30, k_ctx=3):
    """
    records: [(datetime, pressure_kpa), ...] 已按时间升序、已清洗
    th: 单位时间上升量阈值 kPa/min
    n_confirm: 连续满足阈值的点数
    cooldown_min: 同一事件冷却时间（分钟）
    k_ctx: 证据片段前后各取点数
    返回: 预警事件列表
    """
    events = []
    hit_streak = 0
    last_trigger = None

    for i in range(1, len(records)):
        t_prev, p_prev = records[i - 1]
        t_cur, p_cur = records[i]
        dt_min = (t_cur - t_prev).total_seconds() / 60.0
        if dt_min <= 0:
            continue  # 时间异常，跳过
        dp = (p_cur - p_prev) / dt_min

        if dp >= th:
            hit_streak += 1
        else:
            hit_streak = 0

        if hit_streak >= n_confirm:
            if last_trigger is None or (t_cur - last_trigger) >= timedelta(minutes=cooldown_min):
                lo = max(0, i - k_ctx)
                hi = min(len(records), i + k_ctx + 1)
                events.append({
                    "trigger_time": t_cur,
                    "dp_value": round(dp, 3),
                    "th": th,
                    "n_confirm": n_confirm,
                    "context": records[lo:hi],
                    "status": "active",
                })
                last_trigger = t_cur
            hit_streak = 0  # 触发后重置，避免同一爬升段反复计数

    return events

用示例数据验证判据

构造一段 5 分钟采样的示例序列：前段平稳，中段出现一次快速爬升，后段回落。用上面的函数运行，预期在爬升段连续两点超阈值后产生 1 条事件，且证据片段覆盖触发点前后各 3 点。

base = datetime(2024, 6, 1, 8, 0)
pressures = [120.0, 120.3, 120.1, 120.4, 120.2, 120.5,
             121.0, 122.2, 123.6, 124.1, 123.8, 123.5, 123.2]
records = [(base + timedelta(minutes=5 * i), p) for i, p in enumerate(pressures)]

events = detect_pressure_surge(records, th=0.8, n_confirm=2, cooldown_min=30, k_ctx=3)
for e in events:
    print(e["trigger_time"], e["dp_value"], len(e["context"]))

按示例数据手算：第 6→7 点 dP=(122.2-121.0)/5=0.24；第 7→8 点 dP=(123.6-122.2)/5=0.28；第 8→9 点 dP=(124.1-123.6)/5=0.10。若 TH=0.8，则本例不会触发——这正说明阈值必须与真实波动尺度匹配。把 TH 调整为 0.2、N=2 时，第 7、8 点连续满足，触发 1 条事件，证据片段为 7 个点。实际使用时先用正常段统计 dP 分布再定 TH，不要照搬示例数值。

结果判读与记录要点

记录时保留 trigger_time、dp_value、阈值与连续确认参数，以及前后 K 点原始压力值，便于事后复算与阈值调优。事件状态先置 active，待压力回落到基线并稳定后再置 closed，为后续统计突升频次和持续时间提供依据。