guanlan-src/app_ontology/app_ontology_guanlan/sop/hydraulic_validation.py

# -*- coding: utf-8 -*-
"""hydraulic_validation.py — 液压校核: 齿轮箱双压通道 Δp 共模/差模机制分型 (候选级筛查).

裁决对象 = per 机组 **液压回路机制分型候选** (滤芯堵 vs 泵退化) — 区别于 contract_gate (per点清洗) /
temp NBM (部件发热)。本模块 = 压力轴: 单通道 fleet-NBM 能检出压力异常但**分不开机制** (EXP-M3-17 尸检),
双压通道共模/差模分解才是压力的对的工具。

来源: gym M 赛道 a22_dp_discriminator (EXP-M3-21, 2026-07-12 ESTABLISHED → 忠实内联并回 src/ 2026-07-15,
逐字段对拍相等, 见 EXP 文档 §8)。阈值 = EXP-M3-21 预注册值 (Z_FLAG=2.0 / DIFF_HI=0.60 / DIFF_LO=0.40 /
基线窗前20% / 当前窗后40% / 每窗每台 ≥200 gen 样本), **未重调**。
DP 验收 = tests/sop/test_hydraulic_dp_dp.py。

★物理 (First Principles): 液压回路 油泵(yidukou 中位≈3.49bar) → 滤芯 → 齿轮箱入口(≈1.94bar),
滤芯两侧压降 Δp = pump − inlet ≈ +1.5bar (稳态)。
  - **滤芯堵塞** = 流阻升 → Δp 增大 → 入口(下游)与泵(上游)**分离** = **差模** (两通道残差方向/幅度不同)。
  - **泵退化** = 建压能力降 → 整回路压力**同向整体移** (Δp≈不变) = **共模**。
  分型键 = |D| vs |C| 物理量纲(bar)占比, 对符号约定不敏感。

★机制 (no black box): per台 (inlet, pump) 各做**自参考时间差分** (当前窗中位 − 基线窗中位, 消台间良性
静态标定偏置 — 泵压台间良性 spread MAD≈0.33bar, 静态 fleet 残差会把它误判差模) → **fleet 去共模**
(减 fleet 中位 delta, 消季节整场同漂) → 分解 共模 C=(r_in+r_pump)/2 / 差模 D=r_in−r_pump →
检出 score = max(|z_C|,|z_D|) (fleet-MAD z), flag = score ≥ z_flag; 分型 (仅 flagged 台有意义)
diff_ratio = |D|/(|D|+|C|): ≥ diff_hi → 'filter' | ≤ diff_lo → 'pump' | 之间 → 'ambiguous'。

★验证状态 (诚实, 必读):
  - **候选级筛查, 非确诊** (SOP §0.2 候选): 无液压 gold (无确诊故障标签场) → **INJ-only 验证**。
    gym 台架 (yidukou 25台×整年, 2场景×3seed): recall 1.0 / target_pctl 94 / 分型 11/12 显式正确
    + 1 保守 ambiguous (diff_ratio 0.594 差 0.006 触 HI), **零 filter↔pump 混淆** (簇 0.59-0.78 vs
    0.06-0.14, 间隙 ≥0.45)。真实故障形态未经 gold 验证。
  - **A6 真盘候选未裁**: yidukou 未注入数据上 A6 泵侧 +1.76bar / score 16.9 / diff_ratio 0.637 →
    filter 型签名 (滤芯背压)。无 gold / 无独立量 → 候选, 不定论。
  - **需双压通道场, 覆盖面窄**: cleaned 正本实测**仅 yidukou** (契约 dec_g2000_127) 同时有
    齿轮箱入口压力+齿轮箱油泵压力; dongqi/dongfang 点表配置含同名通道 (未清洗落地)。用前必核场契约。
  - INJ 台架 fp_rate 0.04–0.22, 主体 = 真实未标注漂移 (即 A6), 非算法噪声 — 无 gold 分母虚高。

★诚实边界 (算法结构性):
  - **只检观察窗内发生的变化**: t0 起已存在的恒定故障 (基线窗前) 被自参考差分消掉 → 不可见
    (同 S 赛道静态恒偏教义: 无外部绝对参照 INSUFFICIENT)。
  - onset 落基线窗内或跨窗 → delta 被稀释 → 漏检/低估; 窗设计假设早期(前 base_frac)近似健康平稳。
  - **fleet 去共模前提 = 机群多数健康**: 批次共性液压故障 (>~50% 台同机制同期) 会被 fleet 中位
    减掉 → 盲 (相对判据批次盲区, 场型识别先于判据)。
  - 泵退化的 INJ 建模 = 两通道**等幅**同降 (一阶均匀压损); 真实泵退化 inlet/pump 耦合比 ≠1 时
    diff_ratio 偏离 pump 簇 → 落 ambiguous (保守方向, 不误报 filter)。
  - diff_hi=0.60 对 filter 侧偏保守 (INJ 1/6 target 落 ambiguous), 但从不错分为 pump。
  - **score = fleet-相对 z (序统计门)**: 健康/极安静 fleet 上 top-order 台可机会性过 z=2 (物理量级
    仅 ~0.01bar; INJ 台架 fp 0.04-0.22, EXP H3 本就 PARTIAL) → "干净零 flag" 不是本算法性质;
    判读必须结合物理量级 (C_bar/D_bar: 真机制 ~0.5-1.8bar vs 噪声 <0.05bar) 与 fleet 分布。
  - mechanism = 机制**分型候选**, 非部件确诊; 收口需独立量 (滤芯压差表读数/油样/泵电流/现场检查)。
  - 停机态压力语义不同 → 只用 gen 态样本; fleet 双窗过门 <min_fleet 台 → **全场显式 abstain**
    (note, 非静默"干净"); 缺列同理 abstain。

纯核心 (hydraulic_dp_verdict) 无 I/O 可测; per-farm 加载在调用方 (skill/report 层)。
"""
from __future__ import annotations

import numpy as np
import pandas as pd


def _mad_scale(x: np.ndarray) -> float:
    """1.4826×MAD (稳健标度); <3 有限样本 → 0.0 (调用侧 or 1e-9 守零)。忠实自 gym _mad_scale。"""
    x = x[np.isfinite(x)]
    if x.size < 3:
        return 0.0
    return float(1.4826 * np.median(np.abs(x - np.median(x))))


def hydraulic_dp_verdict(
    df: pd.DataFrame, *, inlet_col: str, pump_col: str, time_col: str, turbine_col: str,
    power_col: str | None = None, gen_flag_col: str | None = None, gen_power_min: float = 50.0,
    base_frac: float = 0.2, cur_frac: float = 0.4, min_per_win: int = 200,
    z_flag: float = 2.0, diff_hi: float = 0.60, diff_lo: float = 0.40, min_fleet: int = 5,
) -> dict:
    """per-机组 液压 Δp 共模/差模机制分型 (候选级)。

    返回 {tid: {"flag": bool, "score": max(|z_C|,|z_D|), "mechanism": 'filter'|'pump'|'ambiguous'|None,
                "diff_ratio": |D|/(|D|+|C|) 或 None(弃权), "r_inlet_bar"/"r_pump_bar": 去共模后自参考
                残差(bar), "C_bar"/"D_bar": 共模/差模分量(bar), "z_C"/"z_D": fleet-MAD z, "note"?: 弃权原因}}。
    mechanism 仅 flagged 台非 None (未 flag → None, 分型无意义); diff_ratio 对所有过双窗门的台报
    (含未 flag, 供排查), 弃权台为 None。
    gen 态: gen_flag_col (≥0.5 为运行) 优先, 否则 power_col > gen_power_min, 都缺 → 全行参与
    (调用方须自行保证已滤停机; 停机压力语义不同)。time_col 须 datetime64。
    阈值为 EXP-M3-21 预注册值, 勿因单场重调 (变更走 gym 重验)。
    """
    # NaN-safe: Arrow str backend 下 astype(str) 留 float NaN → sorted 混型崩; dropna + 显式 str()
    tids = sorted({str(x) for x in df[turbine_col].dropna().unique()})
    NULL = {"flag": False, "score": 0.0, "mechanism": None, "diff_ratio": None,
            "r_inlet_bar": None, "r_pump_bar": None, "C_bar": None, "D_bar": None,
            "z_C": None, "z_D": None}
    if inlet_col not in df.columns or pump_col not in df.columns:
        return {t: {**NULL, "note": "双压通道列缺; abstain"} for t in tids}

    # gen 态过滤 (压力回路随负载; 停机压力语义不同) — 忠实自 gym run()
    if gen_flag_col and gen_flag_col in df.columns:
        gen = df[gen_flag_col].astype(float) >= 0.5
    elif power_col and power_col in df.columns:
        gen = df[power_col] > float(gen_power_min)
    else:
        gen = pd.Series(True, index=df.index)
    g = df[gen]
    if len(g) == 0:
        return {t: {**NULL, "note": "gen 态样本为 0; abstain"} for t in tids}

    # 双窗: 基线 = 时间跨度前 base_frac, 当前 = 后 cur_frac (跳过中段过渡, 早晚清晰对比)
    t0 = g[time_col].min()
    t1 = g[time_col].max()
    b_end = t0 + (t1 - t0) * base_frac
    c_start = t1 - (t1 - t0) * cur_frac
    base_mask = g[time_col] < b_end
    cur_mask = g[time_col] >= c_start

    # 自参考 delta (per 台, 每通道): 当前窗中位 − 基线窗中位 (消台间恒偏置)
    def _delta(col):
        # count = 非 NaN 计数 (RV-1 must-fix: size 含 NaN 行 → 全 NaN 死列可过门并静默输出
        # flag=False/score=nan 无 note, 违弃权契约; 列 liveness 失效是已知 SCADA 故障模式)
        b = g[base_mask].groupby(turbine_col)[col].agg(["median", "count"])
        c = g[cur_mask].groupby(turbine_col)[col].agg(["median", "count"])
        j = b.join(c, lsuffix="_b", rsuffix="_c")
        ok = (j["count_b"] >= min_per_win) & (j["count_c"] >= min_per_win)
        return (j["median_c"] - j["median_b"])[ok]

    di = _delta(inlet_col)
    dp = _delta(pump_col)
    common_tids = di.index.intersection(dp.index)
    if len(common_tids) < min_fleet:
        return {t: {**NULL, "note": f"仅 {len(common_tids)} 台满足双窗样本门 (<{min_fleet}); "
                                    "fleet 去共模不可靠, abstain"} for t in tids}
    di = di.loc[common_tids]
    dp = dp.loc[common_tids]

    # fleet 去共模: 减 fleet 中位 delta (消季节整场同漂; 留台特异变化) → 共模/差模分解
    ri = di - di.median()
    rp = dp - dp.median()
    C = (ri + rp) / 2.0       # 共模残差 (bar)
    D = ri - rp               # 差模残差 (bar) = Δp 残差
    sC = _mad_scale(C.to_numpy()) or 1e-9
    sD = _mad_scale(D.to_numpy()) or 1e-9

    out = {t: {**NULL, "note": "双窗有效(非NaN)样本不足 (<min_per_win, 含死列/全NaN通道); abstain"}
           for t in tids}
    for tid in common_tids:
        cbar = float(C.loc[tid])
        dbar = float(D.loc[tid])
        zc = cbar / sC
        zd = dbar / sD
        score = max(abs(zc), abs(zd))
        denom = abs(dbar) + abs(cbar)
        diff_ratio = (abs(dbar) / denom) if denom > 1e-9 else 0.0
        flag = score >= z_flag
        if not flag:
            mech = None                       # gym 'none' → None (未 flag 分型无意义)
        elif diff_ratio >= diff_hi:
            mech = "filter"
        elif diff_ratio <= diff_lo:
            mech = "pump"
        else:
            mech = "ambiguous"
        out[str(tid)] = {
            "flag": bool(flag), "score": round(float(score), 3), "mechanism": mech,
            "diff_ratio": round(diff_ratio, 3),
            "r_inlet_bar": round(float(ri.loc[tid]), 3), "r_pump_bar": round(float(rp.loc[tid]), 3),
            "C_bar": round(cbar, 3), "D_bar": round(dbar, 3),
            "z_C": round(zc, 2), "z_D": round(zd, 2),
        }
    return out