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

# -*- coding: utf-8 -*-
"""icing_validation.py — 叶片结冰签名判别: 冷/暖分层 Δdevz 台内对比 + LOTO null-校准 z。

裁决对象 = per 机组 **结冰型功率亏损签名** (亏损集中低温时段) — 出 flag=候选级筛查, 非确诊;
区别于泛气动退化 (全时段亏损, 归相对排名轴 fleet_sector_devz/EXP-B1-01) 与健康。

来源: gym B 叶片赛道 pc_icing_sig (EXP-B1-02 类型分离坐实 → EXP-B1-03 LOTO-z 换届 H1∧H2∧H3 全 PASS
→ 走设计§7四闸并回 src/ 2026-07-15)。gym 分数≠交付; DP 验收 = tests/sop/test_icing_sig_dp.py。

★核心主张 (EXP-B1-02/03, V-019 复现):
  判别量 = fleet_sector_devz (SOP §4.5, **src 生产件 as-is 复用** — 本模块 import 它, 非内联复制)
  分跑 冷层(temp≤0°C) / 暖层(temp≥5°C) → Δdevz = devz_cold − devz_warm (台内差分消台间偏移) →
  **LOTO null-校准 z**: z(t)=(Δ(t)−median_{s≠t}Δ)/(1.4826·MAD_{s≠t}Δ), 自台不进自己 null (防注入/
  真异常拉宽 MAD 自掩蔽); flag = z ≤ −2 (K=2 沿 SOP 惯例, EXP-B1-03 预注册钉死后验证)。
  数字 (阜山真数据): H1 换台冷限定45% recall **6/6** (z −2.11~−5.46) / H2 泛亏损全拒 (z∈[−0.98,0.74])
  / H3 净场 fp 0/16 (null [−1.21,1.70])。B1-02 绝对阈 Δ≤−2 同注入 0 检出 → 标定单位问题非信号问题。

★物理/统计 trace: 结冰改叶面气动→Cp↓ 仅低温时段; 同台选址/尾流/mast 代表性偏差两层同在→差分消
  (B1-02 H2 实证); 但每层 devz 的 MAD 分母仍含台间散布→绝对 Δ 阈尺度天花板 (B1-02 §6.1), LOTO-z
  把阈标定到本场 null 单位解决 (B1-03)。

★纪律 / 诚实边界 (不掩盖):
- **候选级筛查非确诊**; 无结冰 gold (INJ-only 验证) → 真盘 flag 台 = 候选上报不定谳。
- **数据面前提**: 独立自由来流 ws_ref (mast, 非机舱 self_ref) + 风向扇区 + **真低温面** (冷层
  temp≤0°C n≥min_stratum_n) — 阜山 = 唯一 mast 场, **单场验证不外推** (跨场泛化待第二 mast 场)。
- 检出地板依赖目标台 (B1-03 H4: 30% 亏损部分台可检部分不可) — per台 null 语境已含在 z 里。
- LOTO null n 小 (阜山 16 台) → MAD 估计噪声大; fp 代价面已测 (H3 0/16) 但小群外推谨慎。
- 全窗静态 (无 onset); 原始 Δ≤−2 保留为 flag_conservative 档 (B1-02 保守口径)。
- 对均匀/全场结冰盲 (fleet 中位同降, 同 EXP-B1-01 A12 式诚实 miss); CMS 振动残差雷区不碰。

纯核心无 I/O 可测; 冷暖阈/最小层样本可参; per-farm 数据装配在调用方 (mast join 见 blade_aero 底盘)。
"""
from __future__ import annotations

import numpy as np

import src.sop.discriminators as disc

Z_FLAG = -2.0        # LOTO-z 阈 (K=2, EXP-B1-03 预注册钉死)
DELTA_FLAG = -2.0    # conservative 档: 原始 Δdevz 阈 (EXP-B1-02)
COLD_MAX = 0.0       # 冷层: temp ≤ 0°C (物理结冰阈)
WARM_MIN = 5.0       # 暖层: temp ≥ 5°C (0-5 缓冲带防层间渗漏)
MIN_STRATUM_N = 300


def icing_signature_verdict(df, *, turbine_col, power_col, ws_ref_col, sector_col, temp_col,
                            cold_max: float = COLD_MAX, warm_min: float = WARM_MIN,
                            min_stratum_n: int = MIN_STRATUM_N) -> dict:
    """per-机组 结冰签名裁决。

    df 须含: turbine_col / power_col (kW) / ws_ref_col (独立 mast 自由来流, 密度归一后传入) /
    sector_col (风向扇区 int) / temp_col (环温°C)。调用方负责清洗+mast join (与 sensor_validation 同分工)。

    返回 {tid: {"flag": LOTO-z≤-2, "z_delta", "delta_devz", "flag_conservative": Δ≤-2,
                "devz_cold", "devz_warm", "n_cold", "n_warm", "tier": None|"INSUFFICIENT"}}。
    tier=INSUFFICIENT 三成因 (RV-1 审补全): ①任一层 n<min_stratum_n ②该层 devz 不可估
    ③LOTO null 台数<2 (有效 Δ 台太少, null 不可标定)。①② → 全字段弃权;
    ③ → LOTO flag 弃权 (flag=False, z_delta=None), 但 delta_devz/flag_conservative **仍报**
    (原始 Δ 档不依赖 null, 保留信息) — 判读: tier=INSUFFICIENT 时 flag_conservative 只作参考。
    """
    df = df.copy()
    df[turbine_col] = df[turbine_col].astype(str)   # 统一 str, 防 groupby/devz 键型漂移
    cold = df[df[temp_col] <= cold_max]
    warm = df[df[temp_col] >= warm_min]

    def _devz(sub):
        try:
            r = disc.fleet_sector_devz(sub, turbine_col=turbine_col, power_col=power_col,
                                       ws_ref_col=ws_ref_col, sector_col=sector_col)
            return {str(t): v.get("sc_devz") for t, v in r["per_turbine"].items()}
        except Exception:
            return {}   # 层级判据失败 → 全台该层 None → 逐台落 INSUFFICIENT (显式 tier, 非静默干净)

    zc, zw = _devz(cold), _devz(warm)
    nc = cold.groupby(turbine_col).size()
    nw = warm.groupby(turbine_col).size()
    out: dict = {}
    for t in sorted(df[turbine_col].unique()):
        n_c = int(nc.get(t, 0))
        n_w = int(nw.get(t, 0))
        c, w = zc.get(t), zw.get(t)
        ok = (c is not None and w is not None and np.isfinite(c) and np.isfinite(w)
              and n_c >= min_stratum_n and n_w >= min_stratum_n)
        if not ok:
            out[t] = {"flag": False, "z_delta": None, "delta_devz": None,
                      "flag_conservative": False, "devz_cold": c, "devz_warm": w,
                      "n_cold": n_c, "n_warm": n_w, "tier": "INSUFFICIENT"}
            continue
        d = float(c - w)
        out[t] = {"flag": False, "z_delta": None, "delta_devz": round(d, 3),
                  "flag_conservative": bool(d <= DELTA_FLAG),
                  "devz_cold": round(float(c), 3), "devz_warm": round(float(w), 3),
                  "n_cold": n_c, "n_warm": n_w, "tier": None}
    # LOTO null-校准 z (EXP-B1-03): 自台不进自己的 null
    ds = {t: v["delta_devz"] for t, v in out.items() if v["delta_devz"] is not None}
    for t, v in out.items():
        if v["delta_devz"] is None:
            continue
        others = np.array([x for s, x in ds.items() if s != t], float)
        if len(others) < 2:
            v["tier"] = "INSUFFICIENT"    # null 台数不足 → 不可标定 (显式, 非静默)
            continue
        med = float(np.median(others))
        mad = 1.4826 * float(np.median(np.abs(others - med)))
        z = (v["delta_devz"] - med) / mad if mad > 0 else np.nan
        v["z_delta"] = round(float(z), 2) if np.isfinite(z) else None
        v["flag"] = bool(np.isfinite(z) and z <= Z_FLAG)
    return out