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

# -*- coding: utf-8 -*-
"""sensor_validation.py — 传感器校核: 温度传感器方差退化检测 (二阶矩轴).

裁决对象 = per (机组×通道) **仪表可信度** — 区别于 contract_gate (per点清洗) / health (per机组部件)。
本模块 = 方差轴: 抓传感器**离差变化** (膨胀=接触噪声/退化, 坍缩=量化/失分辨), 一切均值/level 判据天然盲。

来源: gym S 赛道 sensor_varshift (EXP-S2-01 净胜 → EXP-S3-05 RV-1 gate CLOSED=HOLD 独立审4check全过 →
并回 src/ 2026-07-12)。gym 分数≠交付, 本件是走完设计§7四闸 (冻结suite/3原则审/RV-1独立审/labeled DP test)
后的生产实现; DP 验收 = tests/sop/test_sensor_variance_dp.py。

★核心主张 (可分性, 载荷解耦 = 仪表签名):
  仪表方差退化 = **载荷无关** (各载荷箱同向离差变化); 真过程故障 (间歇/载荷共动) 只高载箱动。
  故 score = 载荷分箱早late log-MAD 变化的 **min-across-bins** (两向), 去共模 → 真载荷共动故障低载箱≈0
  → 不误判为传感器坏 (RV-1 独立审实测: 真6℃故障 misattr 0.0, 仅~100℃非物理才破)。

★纪律 (RV-1 审 + gym 教训):
- **窗内去趋势 load-bearing** (减 detrend_win 滚动中位): 否则慢均值移/真故障斜坡撑大长窗 std → 假阳。
- **弃权区**: 近 LSB 极静通道 (σ≲0.2℃, 原生量化≈注入坍缩) → 低置信, 调用方须知。
- **红线 (可分性)**: 载荷共动真故障 misattr ≤ cap; 单臂用于方差轴, 非通用仪表判据 (drift/swap 另有臂)。

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

import numpy as np
import pandas as pd


def _fleet_residual(df: pd.DataFrame, channel: str, power_col: str,
                    rpm_col: str | None, ambient_col: str | None,
                    gen_mask: np.ndarray):
    """fleet OLS: T ~ 1 + P + rpm + T_amb (全机群 gen-only 池化) → 逐行残差。缺列以 0 占位 (退化为可用子集)。"""
    y = df[channel].to_numpy(float)
    feats = [np.ones(len(df))]
    for c in (power_col, rpm_col, ambient_col):
        feats.append(df[c].to_numpy(float) if c and c in df.columns else np.zeros(len(df)))
    X = np.column_stack(feats)
    finite = np.isfinite(y) & np.all(np.isfinite(X), axis=1)
    fitmask = finite & gen_mask
    if fitmask.sum() < 50:
        return None, finite & gen_mask
    beta, *_ = np.linalg.lstsq(X[fitmask], y[fitmask], rcond=None)
    return y - X @ beta, finite & gen_mask


def _load_bins(power: np.ndarray, ok: np.ndarray, n_bins: int):
    edges = np.nanquantile(power[ok], np.linspace(0, 1, n_bins + 1))
    edges[0], edges[-1] = -np.inf, np.inf
    return np.clip(np.digitize(power, edges) - 1, 0, n_bins - 1)


def _windows(t: pd.Series, early_frac: float, late_frac: float):
    days = (t - t.min()).dt.total_seconds().to_numpy() / 86400.0
    dmax = max(days.max(), 1.0)
    return days <= early_frac * dmax, days >= (1.0 - late_frac) * dmax


def _mad(s: np.ndarray) -> float:
    return float(np.median(np.abs(s - np.median(s)))) if len(s) else np.nan


def sensor_variance_verdict(
    df: pd.DataFrame, *, channel: str, time_col: str, turbine_col: str, power_col: str,
    rpm_col: str | None = None, ambient_col: str | None = None, gen_flag_col: str | None = None,
    n_bins: int = 4, early_frac: float = 0.35, late_frac: float = 0.35,
    log_thresh: float = 0.30, detrend_win: int = 432, min_cell_n: int = 30,
) -> dict:
    """per-机组 传感器方差退化裁决 (方差轴)。

    返回 {tid: {"score": 载荷无关方差变化幅度(log10, 任一向), "flag": bool, "direction": inflate|collapse,
                "n_bins": 参与分箱数}}。
    机制: fleet-NBM 残差 → 窗内去趋势 (减 detrend_win 滚动中位, 防慢均值染离差) → per(台,载荷箱) 稳健离差 MAD
    → 早late log10(MAD) 变化 → 去共模 (减 per箱 fleet 中位变化) → **min-across-bins 两向** = 载荷无关分量。
    flag = score > log_thresh (0.30 ≈ 离差 ×2)。df 须按 (turbine, time) 排序 (调用方保证或本函数排)。
    """
    df = df.sort_values([turbine_col, time_col], kind="mergesort").reset_index(drop=True)
    # NaN-safe: Arrow str backend 下 astype(str) 留 float NaN → sorted 混型崩; dropna + 显式 str()
    tids = sorted({str(x) for x in df[turbine_col].dropna().unique()})
    gen = ((df[gen_flag_col] == 1).to_numpy() if gen_flag_col and gen_flag_col in df.columns
           else np.ones(len(df), dtype=bool))
    resid, ok = _fleet_residual(df, channel, power_col, rpm_col, ambient_col, gen)
    if resid is None:
        return {t: {"score": 0.0, "flag": False, "direction": None, "n_bins": 0} for t in tids}

    tid_s = df[turbine_col].astype(str)
    # 窗内去趋势: per台 detrend_win 滚动中位 (防慢均值移伪装成离差)
    rs = pd.Series(np.where(ok, resid, np.nan))
    detr = rs.groupby(tid_s.values).transform(
        lambda x: x - x.rolling(detrend_win, min_periods=max(20, detrend_win // 5),
                                center=True).median()).to_numpy()
    binid = _load_bins(df[power_col].to_numpy(float), ok, n_bins)
    early, late = _windows(df[time_col], early_frac, late_frac)
    sub = pd.DataFrame({"tid": tid_s.values, "bin": binid, "dt": detr, "early": early, "late": late})
    sub = sub[np.isfinite(sub["dt"])]

    eg = sub[sub["early"]].groupby(["tid", "bin"])["dt"].agg([_mad, "size"])
    lg = sub[sub["late"]].groupby(["tid", "bin"])["dt"].agg([_mad, "size"])
    raw = {}
    for t in tids:
        for b in range(n_bins):
            if (t, b) in eg.index and (t, b) in lg.index:
                em, es = eg.loc[(t, b), "_mad"], eg.loc[(t, b), "size"]
                lm, ls = lg.loc[(t, b), "_mad"], lg.loc[(t, b), "size"]
                if es >= min_cell_n and ls >= min_cell_n and em > 1e-6 and lm > 1e-6:
                    raw[(t, b)] = float(np.log10(lm) - np.log10(em))
    from collections import defaultdict
    bag = defaultdict(list)
    for (t, b), v in raw.items():
        bag[b].append(v)
    binmed = {b: float(np.median(vs)) for b, vs in bag.items()}   # per箱 fleet 共模

    out = {}
    for t in tids:
        shifts = [raw[(t, b)] - binmed.get(b, 0.0) for b in range(n_bins) if (t, b) in raw]
        if not shifts:
            out[t] = {"score": 0.0, "flag": False, "direction": None, "n_bins": 0}
            continue
        infl = min(shifts)                    # 全箱膨胀 (载荷无关)
        coll = min(-s for s in shifts)        # 全箱坍缩
        score = max(infl, coll)
        out[t] = {"score": round(float(score), 4), "flag": bool(score > log_thresh),
                  "direction": ("inflate" if infl >= coll else "collapse"), "n_bins": len(shifts)}
    return out


def _member_resid_shift(df, channel, power_col, rpm_col, ambient_col, gen_mask,
                        turbine_col, time_col, early_frac, late_frac):
    """单通道 fleet 残差的 per台 早late 均值移 (含载荷相关分量; swap 判别用原始移非载荷无关截距)。"""
    resid, ok = _fleet_residual(df, channel, power_col, rpm_col, ambient_col, gen_mask)
    if resid is None:
        return {}
    early, late = _windows(df[time_col], early_frac, late_frac)
    tid_s = df[turbine_col].astype(str)
    sub = pd.DataFrame({"tid": tid_s.values, "r": resid, "early": early, "late": late, "ok": ok})
    sub = sub[sub["ok"]]
    e = sub[sub["early"]].groupby("tid")["r"].mean()
    l = sub[sub["late"]].groupby("tid")["r"].mean()
    return {t: float(l.get(t, 0.0) - e.get(t, 0.0)) for t in sorted(tid_s.unique())}


def sensor_swap_verdict(
    df: pd.DataFrame, *, channel_a: str, channel_b: str, time_col: str, turbine_col: str,
    power_col: str, rpm_col: str | None = None, ambient_col: str | None = None,
    gen_flag_col: str | None = None, early_frac: float = 0.35, late_frac: float = 0.35,
    member_shift_thresh: float = 1.0,
) -> dict:
    """per-机组 **通道互换 (接线/标签错) 检测** — 限定于**冗余对** (channel_a, channel_b)。

    返回 {tid: {"swap_flag": bool, "shift_a": 通道a fleet残差早late移, "shift_b": 同b,
                "both_opposite": bool, "score": min(|a|,|b|) 当反号否则0}}。
    ★scope (3原则审定): 本件**非通用仪表判据**, 只判"这对冗余通道是否被对调"。签名 = 两成员 fleet 残差
      **反向移** (一升一降, ∵互换) — 区别于单成员真故障 (只一个动)。gym EXP-S3-06/08: redundancy 检出 +
      discrim2 判别; 本函数 = 两者合一的 both-opposite 判据。
    ★去共模 (RV-1 gate 逮 → 修): sa/sb **减 per通道 fleet 中位移** 再判反号 — 否则**全场共模早late漂移**
      (未建模季节/环温/油脂老化/设定值改) 与单成员真故障叠加会假造"反向", 令真故障被误判 swap。
      实测 (RV-1 独立审复现): 未去共模时共模漂移 ≥1.5℃ 反向 → 单成员真故障 misattr **100%**; 去共模后 → 0。
    ★诚实界: (a) 可检度 ∝ 两通道差异 (近同通道趋不可检, score→0); (b) 去共模后可分性稳 (审者原型 60/60→0);
      残余 misattr 仅噪声级; (c) 需 fleet≥~6 台 (中位去共模需多台) + 两通道都在 df; (d) fit 不足→note 显式弃权。
    """
    # 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 = {"swap_flag": False, "shift_a": 0.0, "shift_b": 0.0, "both_opposite": False, "score": 0.0}
    if channel_a not in df.columns or channel_b not in df.columns:
        return {t: {**NULL, "note": "channel missing"} for t in tids}
    df = df.sort_values([turbine_col, time_col], kind="mergesort").reset_index(drop=True)
    gen = ((df[gen_flag_col] == 1).to_numpy() if gen_flag_col and gen_flag_col in df.columns
           else np.ones(len(df), dtype=bool))
    sa = _member_resid_shift(df, channel_a, power_col, rpm_col, ambient_col, gen,
                             turbine_col, time_col, early_frac, late_frac)
    sb = _member_resid_shift(df, channel_b, power_col, rpm_col, ambient_col, gen,
                             turbine_col, time_col, early_frac, late_frac)
    if not sa or not sb:                                    # fleet OLS 不足 → 显式弃权 (非静默全零"干净")
        return {t: {**NULL, "note": "insufficient fit"} for t in tids}
    med_a = float(np.median(list(sa.values())))            # per通道 fleet 中位 = 共模早late漂移
    med_b = float(np.median(list(sb.values())))
    out = {}
    for t in tids:
        ia, ib = sa.get(t, 0.0) - med_a, sb.get(t, 0.0) - med_b   # 去共模后再判反向
        opp = (ia > 0) != (ib > 0)
        score = min(abs(ia), abs(ib)) if opp else 0.0
        out[t] = {"swap_flag": bool(opp and score > member_shift_thresh),
                  "shift_a": round(ia, 3), "shift_b": round(ib, 3),
                  "both_opposite": bool(opp), "score": round(float(score), 4)}
    return out


def _load_indep_intercept(shifts_by_bin: dict, bin_center: dict) -> float:
    """载荷分箱 shift 对 bin 中心载荷回归 → 截距 = 载荷无关分量 (可分性统计)。"""
    xy = [(bin_center[b], v) for b, v in shifts_by_bin.items() if b in bin_center]
    if len(xy) >= 2:
        xs = np.array([p[0] for p in xy]); ys = np.array([p[1] for p in xy])
        return float(np.polyfit(xs, ys, 1)[1])
    if len(xy) == 1:
        return float(xy[0][1])
    return 0.0


def sensor_drift_verdict(
    df: pd.DataFrame, *, channel: str, time_col: str, turbine_col: str, power_col: str,
    rpm_col: str | None = None, ambient_col: str | None = None, gen_flag_col: str | None = None,
    n_bins: int = 4, win_days: int = 14, studentize_k: float = 3.0, min_cell_n: int = 20,
) -> dict:
    """per-机组 传感器**慢漂移/零位偏置**检测 (load-independent level-shift 轴).

    返回 {tid: {"score": studentized z, "flag": bool, "level_shift": 载荷无关level移(℃), "direction": up|down}}。
    机制 (no black box): fleet-NBM 残差 → 滚动窗(win_days) per载荷箱去共模 → 载荷无关截距 a_w 时序 →
      **median(late半)−median(early半)** 鲁棒 level-shift (中位对噪声窗鲁棒, LESSONS#28) →
      **fleet-MAD studentize** z=(shift−fleet中位)/fleet-MAD, flag **|z|>k 两侧** (**先验 3-MAD, 非拟合** → 自归一干净离散度+跨场迁移)。
    ★两侧 (RV-1 gate 逮 → 修): flag=|z|>k 而非 z>k — 零位偏置**双向** (向上/向下), 单侧会漏向下漂移 (−10℃→z=−71 仍不报); direction 报升/降。
    ★可分性: 用载荷无关截距 → 真载荷共动过程故障截距≈0 → 不触发 (misattr 0.0, gym EXP-S3-09 实测双场)。
      **界 (RV-1 逮, 诚实)**: 只对 **≈线性/凸** 载荷共动成立; **凹/饱和**载荷共动真故障 (载荷指数 ≲0.85) 线性外推到 load=0
      得伪正截距 → 被误判为向上漂移 (实测 exp0.8→z 8.6 误报)。凹故障需下游部件诊断复核, 不可单凭本臂定"仪表漂移"。
    ★caveat (诚实, 根本限制): **安静主轴承类通道** (per台短时 σ≲0.2℃) 的 per-seed 检出 floor ~78%ile =
      fleet-relative 根本地板 (idiosyncratic 季节残差 ≈ 慢漂移, 无外部绝对参照不可解, 同 t0/gain 教义);
      有环温建模 + 适度噪声通道最强, 跨部件/无环温转弱 (gym gjd 齿箱 HS 轴承 64.9 vs fushan 88.9)。
    """
    df = df.sort_values([turbine_col, time_col], kind="mergesort").reset_index(drop=True)
    # 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 = {"score": 0.0, "flag": False, "level_shift": 0.0, "direction": None}
    gen = ((df[gen_flag_col] == 1).to_numpy() if gen_flag_col and gen_flag_col in df.columns
           else np.ones(len(df), dtype=bool))
    resid, ok = _fleet_residual(df, channel, power_col, rpm_col, ambient_col, gen)
    if resid is None:
        return {t: {**NULL, "note": "insufficient fit"} for t in tids}
    P = df[power_col].to_numpy(float)
    binid = _load_bins(P, ok, n_bins)
    bin_center = {b: float(np.nanmedian(P[ok & (binid == b)])) for b in range(n_bins)
                  if (ok & (binid == b)).any()}
    t0 = df[time_col].min()
    day = (df[time_col] - t0).dt.total_seconds().to_numpy() / 86400.0
    winid = (day // win_days).astype(int)
    n_win = int(winid.max()) + 1
    tid_s = df[turbine_col].astype(str)
    sub = pd.DataFrame({"tid": tid_s.values, "win": winid, "bin": binid, "resid": resid, "ok": ok})
    sub = sub[sub["ok"]]
    g = sub.groupby(["tid", "win", "bin"])["resid"].agg(["mean", "size"])
    cell = {k: row["mean"] for k, row in g.iterrows() if row["size"] >= min_cell_n}
    from collections import defaultdict
    bag = defaultdict(list)
    for (t, w, b), val in cell.items():
        bag[(w, b)].append(val)
    fleet_med = {k: float(np.median(vs)) for k, vs in bag.items()}   # per(win,bin) 共模
    half = n_win / 2.0
    level = {}
    for t in tids:
        e_aw, l_aw = [], []
        for w in range(n_win):
            shifts = {b: cell[(t, w, b)] - fleet_med[(w, b)] for b in range(n_bins)
                      if (t, w, b) in cell and (w, b) in fleet_med}
            if shifts:
                aw = _load_indep_intercept(shifts, bin_center)
                (e_aw if w < half else l_aw).append(aw)
        level[t] = (float(np.median(l_aw) - np.median(e_aw)) if len(e_aw) >= 2 and len(l_aw) >= 2 else 0.0)
    vals = np.array(list(level.values()), dtype=float)
    med = float(np.median(vals))
    scale = max(float(1.4826 * np.median(np.abs(vals - med))), 0.05)
    out = {}
    for t in tids:
        s = level[t]
        z = (s - med) / scale
        out[t] = {"score": round(float(z), 4), "flag": bool(abs(z) > studentize_k),   # 两侧 (RV-1 修): 向下漂移不漏
                  "level_shift": round(float(s), 4), "direction": ("up" if s > 0 else "down")}
    return out


def sensor_gear_parity_verdict(
    df: pd.DataFrame, *, genrpm_col: str, rotor_col: str, time_col: str, turbine_col: str,
    gen_flag_col: str | None = None, load_col: str | None = None,
    min_genrpm: float = 50.0, min_rotor: float = 2.0,
    early_frac: float = 0.35, late_frac: float = 0.35,
    rel_shift_thresh: float = 0.004, disp_thresh: float = 0.7,
    min_cell_n: int = 50, n_shift_bins: int = 5, min_bin_n: int = 30,
    min_overlap_frac: float = 0.35,
) -> dict:
    """per-机组 **转速/编码器校核** — 齿比恒等式 genrpm ≡ k·rotor (齿轮箱机械定比, 硬物理奇偶).

    返回 {tid: {"score": 双轴归一 max(≥1=坏), "flag": bool, "ratio_shift_rel": 共同支撑齿比**相对**移(箱间中位),
                "disp_log": log10(MAD_late/MAD_early) on 共同支撑, "axis": shift|disp, "direction": up|down,
                "fleet_ratio": 该台中位齿比, "fleet_dev_rel": 相对fleet偏(**非判据**), "n_e"/"n_l": 窗内样本}}。

    ★机制 (no black box): per台 r = genrpm/rotor (gen ∧ rotor>min ∧ genrpm>min); **两轴均在 early∩late 载荷
    共同支撑上算** (RV-1 round-4 结构性修, 见下), flag 当 max(两轴归一)>1:
      - **共同支撑 = [max(early_lo,late_lo), min(early_hi,late_hi)]** (各窗 2-98 分位); 覆盖 < min_overlap_frac×窄窗跨度
        → 窗间载荷分布不可比 → **两轴弃权** (flag=False + note, INSUFFICIENT), 不硬判。
      - **主轴 shift** = |共同支撑上**等宽分箱** median(r|late)−median(r|early) 的箱间中位| / fleet中位齿比 / rel_shift_thresh。
        真**均匀** offset/gain 各载荷箱同向移 → 箱间中位保留 (需 ≥2 箱一致符号, 否则判 0); 载荷分布 confound → 箱内
        载荷匹配 → ≈0 归零。
      - **弥散轴 disp** = max(0, log10 MAD_late/MAD_early) on 共同支撑 / disp_thresh。freeze 某轴 → r=const/另一轴
        随另一轴弥散爆 → MAD 比升; **共同支撑上算 = 消掉"窗间载荷分布跨度差致 r 散布差"的 disp confound** (log-ratio 标度无关)。
    ★confound 根因与修 (RV-1 3 轮独立审逼出, 核心教训): 齿比 r 若载荷无关 (纯机械编码器) 则恒闭; 但 genrpm 若为**电气
    推算转速** (滑差载荷耦合, 区域2 滑差∝load²) 则 r 载荷相关 → "窗间载荷分布移 (季节/尾流/限电) × 载荷相关 r" =
    confound。**早期分位分箱有无界端箱** (edges=±inf) 窗间载荷不匹配 → 近不重叠窗 (early5→late14m/s) 泄漏整场误报;
    disp 池化全幅同理被窗间散布差污染。**共同支撑等宽分箱 = 每箱两窗都有覆盖 → 结构性根除两泄漏** (实测: 5→14 窗
    quad滑差±5% 从 59/60 误报 → 0; narrow→wide disp 泄漏 10/10 → 0; freeze/offset 检出全保)。
    ★可分性 (王牌, 精确代数 vs 温度统计式): 真转速共变两轴同乘 g → r=(g·genrpm)/(g·rotor)=r **恒等不变**
      → 各箱 shift≈0 ∧ disp≈0 → 天然免疫 (gym EXP-R1-01 INV-sep misattr 0.0; RV-1 EXP-C 复现: 真15%共变 misattr 0)。
    ★诚实界 (3原则审 + RV-1 独立审 4 轮定):
      - **load-bearing 前提 (RV-1 逮, 必声明)**: genrpm 宜为**高速轴机械编码器** (齿轮箱刚性耦合 → r 严格载荷无关, 恒闭);
        若为**电气推算转速** (定子频率+滑差反推), 滑差载荷耦合使 r 载荷相关 → 靠**共同支撑分箱**兜 (箱内载荷匹配)。
      - **confound 免疫 (RV-1 round-4 结构修后实测)**: 共同支撑分箱对 DFIG 电气推算转速滑差 (线性∧区域2二次形) **至 ±5%
        全窗 (含近不重叠 5→14) misattr=0** (物理 DFIG 滑差仅 ±1-2%, ≥2.5× 余量); 窗间重叠不足则弃权非误报。
      - **局部故障检出地板 (RV-1 FINDING2, 已声明安全方向)**: **只破 <3/5 载荷箱的单侧故障** (如高速编码器高转速丢数)
        会被 broad 轴箱间 median 洗掉 → 漏检 (方向安全=不误报, 非红线; candidate-screening, 靠下游部件诊断补)。
        **曾试** max|单箱| 副轴兜之, RV-1 round-3 逮: 二次滑差 confound 无界箱单箱破 (≤7×阈) 与真局部故障单箱破 (14×阈)
        幅值重叠, 无单一阈可分 → 会重开 FINDING1 红线 → **撤回**。局部检出留 backlog (窗内残差回归/时序持续)。
      - **pinned-load 检出缺口 (RV-1 round-5, 安全方向)**: late 窗载荷全 pin 在 rated (单一载荷水平, per台方差极小) →
        共同支撑内 <2 箱 → shift 轴弃权 (note 标注, **非静默漏**); 此时均匀 offset 可能漏检 (无第二载荷水平校齿比) →
        属"单载荷水平不可校验"的诚实 INSUFFICIENT, 方差回升即恢复。
      - **disp 免疫 scope (RV-1 round-5)**: confound 免疫声明限 **光滑滑差 (线性∧区域2二次形)**; 病态**非光滑 (阶跃不连续)
        滑差**在超包线幅 (≥±7%, 远超物理 ±1-2%) 可偶触 disp 轴 (实测 step±7% 1/480) → 真 DFIG 滑差光滑, 属超范围病态。
      - go/no-go **不定位哪一路错** (gen 还是 rotor); 需功率/频率第三参照。
      - **fleet_dev_rel = 副字段非判据**: 静态恒偏可能真硬件异质 (齿比铭牌/编码器线数不同), 无铭牌不可辨
        (同温度 levelparity REJECT + power-curve NTF 无绝对参照 → INSUFFICIENT 教义)。
      - **适用范围 = 齿轮箱机型 (k≫1)**; 直驱 (k≈1, genrpm≈rotor 常同一传感器反推) **超范围**, 调用方须自行判。
      - 需 rotor 列在 df ∧ fleet≥~6 台 (相对化用 fleet 中位齿比); 缺则 note 显式弃权 (非静默全零"干净")。

    来源: gym R 赛道 sensor_gear_parity (EXP-R1-01, pctl 96 零 seed 方差 → §7 四闸 → 并回 src/ 2026-07-12)。
    产化改动 (对 gym champion, 均经 RV-1 独立审 4 轮): (a) shift 轴绝对→**相对** (÷fleet中位齿比) 跨机 k 迁移, 同 gjd
    净场地板定标 (0.5/117.64≈0.42%); (b) 两轴改**共同支撑等宽分箱** (FINDING1 + 窗间不匹配泄漏结构性修, gym 单场无);
    (c) 重叠不足 → 双轴弃权非发假 flag。DP 验收 = tests/sop/test_sensor_gear_parity_dp.py (含 slip×窗 misattr=0 回归闸
    + FINDING2 局部漏检 known-limitation 锁定)。
    """
    # 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 = {"score": 0.0, "flag": False, "ratio_shift_rel": 0.0, "disp_log": 0.0,
            "axis": None, "direction": None, "fleet_ratio": None, "fleet_dev_rel": 0.0, "n_e": 0, "n_l": 0}
    if genrpm_col not in df.columns or rotor_col not in df.columns:
        return {t: {**NULL, "note": "genrpm/rotor 列缺; abstain"} for t in tids}
    df = df.sort_values([turbine_col, time_col], kind="mergesort").reset_index(drop=True)
    gen = ((df[gen_flag_col] == 1).to_numpy() if gen_flag_col and gen_flag_col in df.columns
           else np.ones(len(df), dtype=bool))
    a = df[genrpm_col].to_numpy(float)
    b = df[rotor_col].to_numpy(float)
    ok = gen & np.isfinite(a) & np.isfinite(b) & (b > min_rotor) & (a > min_genrpm)
    with np.errstate(divide="ignore", invalid="ignore"):
        r = np.where(ok, a / b, np.nan)                        # b>min_rotor 已守; errstate 静默被掩位
    fleet_med = float(np.nanmedian(r[ok])) if ok.any() else np.nan
    if not np.isfinite(fleet_med) or fleet_med <= 0:
        return {t: {**NULL, "note": "fleet 齿比不可估; abstain"} for t in tids}
    early, late = _windows(df[time_col], early_frac, late_frac)
    tv = df[turbine_col].astype(str).to_numpy()
    # 载荷代理: 显式 load_col (>rated 更佳) 否则 rotor (region2 单调于风/载荷, 自足无需额外列)
    load_arr = (df[load_col].to_numpy(float) if (load_col and load_col in df.columns) else b)

    def _cs_axes(re_r, re_l, rl_r, rl_l):
        """early∩late **载荷共同支撑**上算 shift + disp (RV-1 round-4 结构修: 消无界箱/池化的窗间不匹配 confound)。
        返回 (shift, disp, note): 重叠不足 → (None, None, 弃权); 共同支撑内箱<2 → (None, disp, shift弃权) disp 仍算。
        shift=箱间 median (均匀故障各箱同向→保留; confound→箱内匹配≈0); disp=共同支撑上 MAD 比 (freeze 抓, 消散布confound)。"""
        if len(re_l) < 2 * min_bin_n or len(rl_l) < 2 * min_bin_n:
            return None, None, "弃权(载荷样本不足)"
        e_lo, e_hi = np.nanpercentile(re_l, [2, 98])
        l_lo, l_hi = np.nanpercentile(rl_l, [2, 98])
        lo, hi = max(e_lo, l_lo), min(e_hi, l_hi)
        narrow = min(e_hi - e_lo, l_hi - l_lo)
        if not (hi > lo) or narrow <= 0 or (hi - lo) < min_overlap_frac * narrow:
            return None, None, "弃权(窗间载荷重叠不足→INSUFFICIENT)"
        em = (re_l >= lo) & (re_l <= hi); lm = (rl_l >= lo) & (rl_l <= hi)
        rer, rlr = re_r[em], rl_r[lm]                          # 共同支撑内样本 (两窗同载荷范围)
        mad_e, mad_l = _mad(rer), _mad(rlr)                    # disp: 共同支撑上 MAD 比 (消窗间散布 confound)
        disp = (float(np.log10(mad_l / mad_e)) if (mad_e and mad_l and mad_e > 1e-9 and mad_l > 1e-9) else 0.0)
        edges = np.linspace(lo, hi, n_shift_bins + 1)          # 共同支撑上**等宽箱** (无无界端箱 → 每箱两窗都覆盖)
        eb = np.clip(np.digitize(re_l[em], edges) - 1, 0, n_shift_bins - 1)
        lb = np.clip(np.digitize(rl_l[lm], edges) - 1, 0, n_shift_bins - 1)
        sb = []
        for bn in range(n_shift_bins):
            reb = rer[eb == bn]; rlb = rlr[lb == bn]
            if len(reb) >= min_bin_n and len(rlb) >= min_bin_n:
                sb.append(float(np.median(rlb) - np.median(reb)))
        if len(sb) < 2:
            return None, disp, "shift弃权(共同支撑<2箱); disp仍算"
        med = float(np.median(sb))
        if med == 0.0:
            return 0.0, disp, None
        frac_same = float(np.mean(np.sign(sb) == np.sign(med)))   # 各载荷箱同向 = 真单侧故障签名
        return (med if frac_same >= 0.6 else 0.0), disp, None      # 箱间符号不一致 (载荷confound/freeze) → 0

    out = {}
    for t in tids:
        sel = (tv == t) & ok
        e_m, l_m = sel & early, sel & late
        re = r[e_m]; rl = r[l_m]; re_l = load_arr[e_m]; rl_l = load_arr[l_m]
        fe = np.isfinite(re) & np.isfinite(re_l); fl = np.isfinite(rl) & np.isfinite(rl_l)
        re, re_l = re[fe], re_l[fe]; rl, rl_l = rl[fl], rl_l[fl]
        per_med = float(np.median(r[sel][np.isfinite(r[sel])])) if sel.any() else np.nan
        if len(re) < min_cell_n or len(rl) < min_cell_n:
            out[t] = {**NULL, "fleet_ratio": (round(per_med, 4) if np.isfinite(per_med) else None),
                      "n_e": int(len(re)), "n_l": int(len(rl)), "note": "窗内样本不足"}
            continue
        shift, disp, snote = _cs_axes(re, re_l, rl, rl_l)     # 两轴均在共同支撑上算
        rel_shift = (shift / fleet_med) if shift is not None else 0.0   # 相对化: 跨机 k 迁移
        shift_norm = (abs(rel_shift) / rel_shift_thresh) if shift is not None else 0.0
        disp_norm = (max(0.0, disp) / disp_thresh) if disp is not None else 0.0
        score = max(shift_norm, disp_norm)
        fleet_dev_rel = ((per_med - fleet_med) / fleet_med if np.isfinite(per_med) else 0.0)
        rec = {"score": round(float(score), 4), "flag": bool(score > 1.0),
               "ratio_shift_rel": round(rel_shift, 6), "disp_log": round(float(disp), 4) if disp is not None else None,
               "axis": ("disp" if disp_norm >= shift_norm else "shift"),
               "direction": ("up" if (shift or 0.0) > 0 else "down"),
               "fleet_ratio": round(per_med, 4), "fleet_dev_rel": round(float(fleet_dev_rel), 6),
               "n_e": int(len(re)), "n_l": int(len(rl))}
        if snote:
            rec["note"] = snote
        out[t] = rec
    return out


def _phase_derived_rel(L, bar, sel, min_n: int = 100) -> float:
    """per-台/段 三相**最小相对 pair-diff** = min_{i<j} median|l_i−l_j| / median(bar)。
    派生/复制相 (同一测量写多列, 仅舍入差) → →0; 独立测量 → ≥~1e-3 (gucheng 真盘双峰隙 24×)。
    样本 <min_n 或 bar 非正 → **inf** (不可判; 令调用方守卫 `rel<tol` 退化为 False = 既有"非派生"语义)。"""
    if sel.sum() < min_n:
        return np.inf
    b = float(np.median(bar[sel]))
    if not np.isfinite(b) or b <= 0:
        return np.inf
    k = L.shape[1]
    best = np.inf
    for i in range(k):
        for j in range(i + 1, k):
            r = float(np.median(np.abs(L[sel, i] - L[sel, j]))) / b
            if r < best:
                best = r
    return best


def _phase_derived_guard(L, bar, sel, rel_tol: float = 1e-4) -> bool:
    """per-台**派生/复制相守卫** (RV-1 E2 审逮): 任一相对 median|l_i−l_j|/median(bar) < rel_tol →
    疑似同一测量写多列 (仅舍入差) → **parity/level/swap 全不可证** (m≡0 是伪"干净")。
    gucheng 真盘 11/24 台如此 (rel ~3e-6..5e-5), 与独立台 (≥3.7e-4) 双峰隙 7× → 1e-4 不误伤。
    教训: 前提核查必 **per-台** — 行级 fleet 池化 ("23%行相等→独立") 掩盖 per-台双峰 = 聚合层错误
    (3 First Principles #2)。核心判据抽入 _phase_derived_rel (契约层 channel_semantics 按月复用同判据)。"""
    return _phase_derived_rel(L, bar, sel) < rel_tol


def sensor_phase_parity_verdict(
    df: pd.DataFrame, *, phase_cols: list, power_col: str, time_col: str, turbine_col: str,
    gen_flag_col: str | None = None, gen_power_min: float = 50.0, min_ibar: float = 10.0,
    early_frac: float = 0.35, late_frac: float = 0.35, thresh: float = 0.02,
    n_bins: int = 4, min_bin_n: int = 30, min_overlap_frac: float = 0.35,
) -> dict:
    """per-机组 **功率/电气传感器校核** — 三相电流平衡奇偶 (l1≈l2≈l3, 天然内建冗余三元组).

    返回 {tid: {"score": 去共模载荷匹配后 imbalance 早late移, "flag": bool, "suspect_phase": 破平衡的相列名,
                "phase_dev_shift": 该相 signed 相对偏移, "n_load_bins": 参与箱数, "note"?: 弃权原因}}。

    ★机制 (no black box): imbalance m = max_i|l_i/Ibar − 1| (Ibar=三相均值); **标度不变** (三相同缩 → l_i/Ibar
      逐相不变 → m 逐位不变)。单相 CT 故障 (gain/bias/drift/freeze) 破单相幅值平衡 → m 抬升; 三相共缩 (真负载变化)
      → m 恒不变。score = **共同支撑载荷分箱 early-late median(m) 移**的箱间中位 (**不去共模**, 见下)。
      定位: per台 late-early 各相 signed dev 均值移, 最大绝对者 = suspect_phase (**仅 ≥3 相报**, 见诚实界)。
    ★可分性 (王牌, 物理零点锚定 vs 温度经验 fleet 均值): 三相共缩×g → m 逐位不变 → misattr 精确 0
      (gym EXP-E1-01: 共缩×1.15 misattr 0.0 全seed; 物理预言, 不靠 fleet 建模质量 → 比温度 fleetdrift 无环温跨场
      REJECT 的软肋更硬)。奇偶地板锚在**物理零点 (平衡)** 非经验 fleet 均值 → **无需 fleet 相对化** — 这正是
      **不去共模**的依据 (RV-1 逮): 去共模对 confound 免疫**零贡献** (箱内载荷匹配已消, A/B 实测 0/60 不变), 却把
      ≥~50% 机群**批次共性 CT 故障**中位减掉 → 归零遮蔽 (5/10→0, 8/10→0 检出; 删后全恢复且零新 FP) — 复现 memory
      relative-criterion-batch-blindness (批次退化场 fleet 基线污染致相对判据盲), 同齿比 round-3 local 轴 = 净负增项。
    ★confound 防护 (承齿比 §7 LESSONS #42, 主动套用): 真 CT 噪声是**加性/精度等级** (低电流相对误差大) → m 对健康机
      **载荷相关** (低载 m 地板抬, 实测低/高载比 ~7×); "窗间载荷分布移 (季节/限电) × 载荷相关 m" = confound。双重防护:
      (a) **共同支撑等宽分箱** (early∩late 功率重叠区, 无无界端箱 → 每箱两窗都覆盖; 重叠 < min_overlap_frac×窄窗 →
      弃权 INSUFFICIENT); (b) **箱内载荷匹配守卫** (箱内 early/late 功率中位差 >半箱宽 → 该箱剔除)。RV-1 实测: 真载荷
      相关 m (加性/CT-class) × 近不重叠窗 × 6seed → 0/60 misattr, 双重防护足, 无需第三重。
    ★诚实界 (3原则审 + RV-1 独立审定):
      - **只检 onset 起故障, 盲于 t0 起恒定单相偏** (early-late 差分, 同 S 赛道 levelparity/齿比静态恒偏 =
        无外部绝对参照 INSUFFICIENT; E2 level 轴待做)。
      - **suspect_phase 仅 ≥3 相报** (RV-1 逮): 2 相时 Ibar=mean(l1,l2) → 单相故障 δ 致 dev1=+δ/2, dev2=−δ/2
        **等幅反号** → argmax 由噪声定 = 抛硬币 (实测 40% 正确) → len(ph)==2 时 suspect_phase=None (检出 score 仍
        有效, 只是不定位)。双相同时故障时 suspect 只报最大者, 会误导"只一相坏" — 候选非定论。
      - **检出地板 (RV-1 定标, 加性噪声)**: gain+5% 在真加性噪声下 score≈0.0185 vs 阈 0.02 → margin 仅 ~1.5× →
        低幅 gain (≲5%) 可能欠阈漏检 (安全方向); 乘性噪声环境 (干净 CT) 下同幅检出。
      - **数据前提 (RV-1 E2 审更正)**: 须**实测三相非派生/复制** — 早先"gucheng l1==l2 仅23%行→独立成立"是
        **聚合层错误** (行级 fleet 池化掩盖 per-台双峰; 3 First Principles #2): per-台实测 gucheng **11/24 台复制相**
        (rel pair-diff ~3e-6, 仅舍入差) vs 13/24 独立 (≥3.7e-4), 双峰隙 7× → 本函数内建 _phase_derived_guard
        (rel<1e-4 → INSUFFICIENT note 非静默"干净")。独立台的极平衡 (自然地板≤~1.6%) → 低幅故障可能欠阈。
      - **近不重叠窗 → 弃权非误报** (共同支撑修, 承齿比 round-4); go/no-go 只判"三相不平衡", suspect_phase 是**候选**
        (需现场核 CT vs grid 电压不平衡/绕组); 需 ≥2 相在 df ∧ power 列 (不需 fleet 相对化 → 单台也可判)。
      - **suspect_phase 在"两相同偏"模式下恰指唯一健康相** (gucheng 真盘 GC11/GC15 电压 episode 实证, 2026-07-15
        场级 review): dev 相对 bar 归一 → 两相同低时最大|dev|是**没动的第三相** → 定位反指。破解 = **fleet 同窗
        绝对量对照** (per相 vs 其余台中位) 定哪相动, 属调用方 review 层步骤, 本函数 suspect 仅"哪相与其余不一致"候选。
      - **相接线 swap (l1↔l2 双相反向)** 本臂不专门判 (E2 队列); 单相故障族 (gain/bias/drift/freeze) 是本臂 scope。

    来源: gym E 赛道 sensor_phase_parity (EXP-E1-01, pctl95.8/recall1.0/fp0.0 + suspect_phase l1全命中 → §7四闸 →
    并回 src/ 2026-07-15)。产化改动 (对 gym champion, 主动套齿比 §7 教训 + RV-1 审): (a) 载荷分箱改**共同支撑等宽**
    (gym 无界分位箱 → 近不重叠窗泄漏, 齿比 RV-1 逮同族); (b) 加箱内载荷匹配守卫 + 重叠不足弃权; (c) **删 gym 的
    去共模** (RV-1 A/B 证冗余且遮蔽批次故障); (d) 2 相不报 suspect_phase。DP 验收 =
    tests/sop/test_sensor_phase_parity_dp.py (confound 闸用真加性噪声 + 批次检出回归闸)。
    """
    # 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 = {"score": 0.0, "flag": False, "suspect_phase": None, "phase_dev_shift": 0.0, "n_load_bins": 0}
    ph = [c for c in phase_cols if c in df.columns]
    if len(ph) < 2 or power_col not in df.columns:
        return {t: {**NULL, "note": "三相通道<2 或缺功率列; abstain"} for t in tids}
    df = df.sort_values([turbine_col, time_col], kind="mergesort").reset_index(drop=True)
    L = np.column_stack([df[c].to_numpy(float) for c in ph])
    with np.errstate(invalid="ignore", divide="ignore"):
        Ibar = np.nanmean(L, axis=1)
        dev = L / Ibar[:, None] - 1.0                          # 逐相 signed 相对偏 (定位用)
        m = np.nanmax(np.abs(dev), axis=1)                     # imbalance (标度不变)
    gen = ((df[gen_flag_col] == 1).to_numpy() if gen_flag_col and gen_flag_col in df.columns
           else (df[power_col].to_numpy(float) > gen_power_min))
    ok = gen & np.isfinite(m) & (Ibar > min_ibar)
    P = df[power_col].to_numpy(float)
    early, late = _windows(df[time_col], early_frac, late_frac)
    tv = df[turbine_col].astype(str).to_numpy()
    e_ok, l_ok = ok & early, ok & late
    if e_ok.sum() < 2 * min_bin_n or l_ok.sum() < 2 * min_bin_n:
        return {t: {**NULL, "note": "窗内 gen 样本不足; abstain"} for t in tids}
    # fleet-wide 共同支撑 (承齿比 round-4: 无无界端箱, 每箱两窗都覆盖)
    e_lo, e_hi = np.nanpercentile(P[e_ok], [2, 98])
    l_lo, l_hi = np.nanpercentile(P[l_ok], [2, 98])
    lo, hi = max(e_lo, l_lo), min(e_hi, l_hi)
    narrow = min(e_hi - e_lo, l_hi - l_lo)
    if not (hi > lo) or narrow <= 0 or (hi - lo) < min_overlap_frac * narrow:
        return {t: {**NULL, "note": "窗间功率重叠不足→INSUFFICIENT"} for t in tids}
    edges = np.linspace(lo, hi, n_bins + 1)
    binid = np.clip(np.digitize(P, edges) - 1, 0, n_bins - 1)
    in_cs = (P >= lo) & (P <= hi)
    match_tol = 0.5 * (hi - lo) / n_bins                       # 箱内载荷匹配容差 = 半箱宽
    # per台 per箱 早late median(m) 移 (共同支撑内 + 箱内载荷匹配守卫)
    raw_shift = {t: {} for t in tids}
    for t in tids:
        base = (tv == t) & ok & in_cs
        for b in range(n_bins):
            e_in = base & early & (binid == b); l_in = base & late & (binid == b)
            if e_in.sum() >= min_bin_n and l_in.sum() >= min_bin_n:
                if abs(float(np.median(P[e_in])) - float(np.median(P[l_in]))) <= match_tol:   # 箱内载荷真匹配
                    raw_shift[t][b] = float(np.median(m[l_in]) - np.median(m[e_in]))
    # 不去共模 (RV-1 逮): m 锚物理零点无需 fleet 相对化; 去共模对 confound 零贡献 (箱内匹配已消) 且遮蔽批次共性 CT 故障
    out = {}
    for t in tids:
        shifts = list(raw_shift[t].values())
        base = (tv == t) & ok & in_cs
        if _phase_derived_guard(L, Ibar, (tv == t) & ok):      # RV-1 E2 逮: 复制相 m≡0 = 伪"干净"
            out[t] = {**NULL, "note": "疑似派生/复制相 (rel pair-diff <1e-4): parity 不可证 → INSUFFICIENT"}
            continue
        if not shifts:
            out[t] = {**NULL, "note": "共同支撑内无载荷匹配箱; abstain"}
            continue
        score = float(np.median(shifts))
        rec = {"score": round(score, 5), "flag": bool(score > thresh), "n_load_bins": len(shifts),
               "suspect_phase": None, "phase_dev_shift": 0.0}
        if len(ph) >= 3:                                       # 定位仅 ≥3 相 (RV-1 逮: 2相 dev 等幅反号=抛硬币)
            eb, lb = base & early, base & late                 # late-early 各相 signed dev 均值移
            if eb.sum() >= min_bin_n and lb.sum() >= min_bin_n:
                with np.errstate(invalid="ignore"):
                    psh = np.nanmean(dev[lb], axis=0) - np.nanmean(dev[eb], axis=0)
                if not np.all(np.isnan(psh)):
                    j = int(np.nanargmax(np.abs(psh)))
                    rec["suspect_phase"] = ph[j]; rec["phase_dev_shift"] = round(float(psh[j]), 5)
        out[t] = rec
    return out


def sensor_phase_level_verdict(
    df: pd.DataFrame, *, phase_cols: list, power_col: str, time_col: str, turbine_col: str,
    gen_flag_col: str | None = None, gen_power_min: float = 50.0, min_ibar: float = 10.0,
    level_thresh: float = 0.03, min_n: int = 200,
) -> dict:
    """per-机组 **三相电流/电压 level 校核** — 全窗绝对 imbalance vs **物理零点** (t0 起恒定单相偏可检).

    返回 {tid: {"score": 高载半区 median(m)/level_thresh (≥1=坏), "flag": bool, "level_m": 绝对不平衡水平,
                "suspect_phase": 恒偏相 (仅≥3相), "phase_dev_level": 该相 signed dev, "n": 样本, "note"?: 弃权/低置信}}。

    ★机制 (no black box): m = max_i|l_i/Ibar−1|; score = per台**高载半区** (P ≥ fleet gen 中位) 的 median(m) /
      level_thresh。全窗量无窗差分 → **onset 差分法盲的 t0 恒偏可检** (gym EXP-E2-01 交叉印证: 差分法对 t0 场景
      pctl 30-45 盲, level 轴 95.8)。
    ★教义突破依据 (EXP-E2-01 TRUE): 温度 levelparity REJECT 的根因 = 无外部绝对参照 (per台基线异质不可解);
      电气 m 锚**物理零点 (三相平衡)** + 场健康地板窄 → 绝对阈 a-priori 可定 (0.03 ≈ 2×全窗最坏健康台,
      gucheng 定标非 tune)。可分性代数继承: 真负载共缩 → m 逐位不变 → level 不变 (misattr 0 实测)。
    ★高载半区 (产化硬化, 送审前 probe 逮): 加性 CT 噪声 → m 地板**载荷相关** (gucheng 最低载箱健康地板 max
      0.0274 = 距阈仅 1.1×; 高载半区 0.0086 = 3.5× margin) → 纯限电/低载分布台全窗 median 会被噪声地板顶到贴阈
      → **主判据取高载半区** (仪表恒偏载荷无关, 高载 = 噪声地板最低的干净估计); 高载样本 < min_n →
      **INSUFFICIENT (flag 强制 False, score+note 照报)** — 低载数据物理上分不清噪声地板与恒偏, 不 flag
      (DP 逮回退发 flag 的 bug → 修为弃权语义)。
    ★诚实界:
      - **t0 恒偏与"真·自 t0 存在的电气不平衡" (绕组不对称/持续 grid 不平衡) 不可分** — go/no-go 只判"持续
        超地板不平衡", suspect_phase 是候选 (需现场核 CT/VT 校验 vs 绕组/电网侧; 同 E1 grid 裁决语义)。
      - **阈 gucheng 定标 (明阳 MY2.0 geared)**: 跨场先核该场健康地板 (per台 median m 分布) 再用; 地板 >0.015
        的场须重定标。
      - suspect_phase 仅 ≥3 相报 (2 相 dev 等幅反号, 同 sensor_phase_parity_verdict RV-1 教训); **全相噪声型
        退化 (CT 整体噪声↑而非单相偏) 也会 flag (正当: 仪表真退化), 但此时 suspect_phase 随机无意义** (RV-1 逮);
        **"两相同偏"模式下 suspect 恰指唯一健康相** (dev 相对归一的结构性反指; gucheng GC11/GC15 电压 episode
        实证 — 破解 = fleet 同窗绝对量对照, review 层步骤); 电压通道同函数可用 (phase_cols 换电压列 + min_ibar 调 ~100)。
      - **复制相守卫**: _phase_derived_guard (rel pair-diff <1e-4) → INSUFFICIENT note (gucheng 11/24 台复制相,
        m≡0 是伪"干净"; RV-1 E2 逮, 前提核查必 per-台)。
      - 需 ≥2 相 ∧ power 列; 缺/样本不足 → note 显式弃权。

    来源: gym E2 赛道 sensor_phase_level (EXP-E2-01 TRUE: t0 gain×1.10/×1.05 均 pctl 95.8×3seed, clean fp 0,
    共缩 misattr 0 → §7 四闸 → 并回 src/ 2026-07-15)。产化改动: 全窗 median → **高载半区 median** (低载噪声地板
    confound, probe 实测 0.0274 贴阈)。DP 验收 = tests/sop/test_sensor_phase_level_dp.py。
    """
    # 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 = {"score": 0.0, "flag": False, "level_m": 0.0, "suspect_phase": None,
            "phase_dev_level": 0.0, "n": 0}
    ph = [c for c in phase_cols if c in df.columns]
    if len(ph) < 2 or power_col not in df.columns:
        return {t: {**NULL, "note": "相通道<2 或缺功率列; abstain"} for t in tids}
    L = np.column_stack([df[c].to_numpy(float) for c in ph])
    with np.errstate(invalid="ignore", divide="ignore"):
        Ibar = np.nanmean(L, axis=1)
        dev = L / Ibar[:, None] - 1.0
        m = np.nanmax(np.abs(dev), axis=1)
    gen = ((df[gen_flag_col] == 1).to_numpy() if gen_flag_col and gen_flag_col in df.columns
           else (df[power_col].to_numpy(float) > gen_power_min))
    ok = gen & np.isfinite(m) & (Ibar > min_ibar)
    if ok.sum() < min_n:
        return {t: {**NULL, "note": "gen 样本不足; abstain"} for t in tids}
    P = df[power_col].to_numpy(float)
    p_med = float(np.nanmedian(P[ok]))                        # fleet gen 中位 = 高载半区界
    hi = ok & (P >= p_med)
    tv = df[turbine_col].astype(str).to_numpy()
    out = {}
    for t in tids:
        sel_hi = (tv == t) & hi
        if _phase_derived_guard(L, Ibar, (tv == t) & ok):      # RV-1 E2 逮: 复制相 m≡0 = 伪"干净"
            out[t] = {**NULL, "n": int(((tv == t) & ok).sum()),
                      "note": "疑似派生/复制相 (rel pair-diff <1e-4): level 不可证 → INSUFFICIENT"}
            continue
        note = None
        insufficient = False
        if sel_hi.sum() >= min_n:
            use = sel_hi
        else:                                                  # 限电/纯低载台: 低载噪声地板与恒偏物理不可分
            use = (tv == t) & ok
            note = "高载样本不足→INSUFFICIENT (低载噪声地板与恒偏不可分, score仅供参考)"
            insufficient = True
            if use.sum() < min_n:
                out[t] = {**NULL, "n": int(use.sum()), "note": "样本不足; abstain"}
                continue
        lvl = float(np.median(m[use]))
        score = lvl / level_thresh
        rec = {"score": round(score, 4), "flag": bool(score > 1.0 and not insufficient),
               "level_m": round(lvl, 5),
               "suspect_phase": None, "phase_dev_level": 0.0, "n": int(use.sum())}
        if len(ph) >= 3:
            med_dev = np.median(dev[use], axis=0)
            j = int(np.nanargmax(np.abs(med_dev)))
            rec["suspect_phase"] = ph[j]
            rec["phase_dev_level"] = round(float(med_dev[j]), 5)
        if note:
            rec["note"] = note
        out[t] = rec
    return out


def sensor_phase_swap_verdict(
    df: pd.DataFrame, *, phase_cols: list, power_col: str, time_col: str, turbine_col: str,
    gen_flag_col: str | None = None, gen_power_min: float = 50.0, min_ibar: float = 10.0,
    early_frac: float = 0.35, late_frac: float = 0.35, swap_thresh: float = 0.005,
    null_ratio: float = 0.5, detectable_min_d12: float = 0.005,
    n_bins: int = 4, min_bin_n: int = 30, min_overlap_frac: float = 0.35,
) -> dict:
    """per-机组 **相接线 swap 检测** (l1↔l2 接线/标签对调 @onset) — 三相中前两相的互换判别.

    返回 {tid: {"swap_flag": bool, "score": min(|Δdev1|,|Δdev2|)/swap_thresh 当签名成立, "ddev": [Δdev1,Δdev2,Δdev3],
                "detectable": 该台是否在可检域, "d12_persistent": persistent 相间差, "note"?: 弃权原因}}。

    ★机制 (no black box): swap 签名 = **Δdev1 ≈ −Δdev2 反号互换** (两相 persistent 差对调) 且
      **Δdev3 ≈ 0 (第三相 = null 控制)**。区别单相故障 (δ@l1 → Δdev=(+2δ/3,−δ/3,−δ/3): dev3 动且与 dev2 同号
      → null 控制拒)。判别 = opp(1,2) ∧ |Δdev3| < null_ratio×min(|Δdev1|,|Δdev2|) ∧ min > swap_thresh。
      **Δdev 在共同支撑载荷分箱内算** (见下), 箱间 median。
    ★共同支撑载荷分箱 (产化硬化, 送审前 probe 逮): 相间差 dev1−dev2 **强载荷相关** (gucheng 实测 GC1
      -0.035@低载→-0.015@高载, GC5 6.5× 变化) + 三相 dev 和≈0 约束 → **窗间载荷分布大移可造反号 Δdev1/Δdev2 +
      Δdev3 安静 = 完美伪 swap 签名** (gym INV 未测此角) → Δdev 必在 early∩late 载荷共同支撑等宽箱内算
      (箱内载荷匹配 → 载荷驱动的相间差移归零); 重叠不足 → 弃权 INSUFFICIENT (承齿比/电气 §7 四角图谱)。
    ★诚实可检域 (EXP-E2-01 §0 物理): **平衡三相的幅值互换 = 无观测量变化** → persistent |median(dev1−dev2)|
      < detectable_min_d12 的台**结构性不可检** (幅值 SCADA 教义; gucheng 可检域 7/24) → detectable=False +
      note, **不 flag 不误导"已核干净"**。互换幅 = 2×persistent 差, 检出∝该台相间异质度。**可检域判定用
      early 窗本底** (onset 前参照期) — 全窗会被互换本身污染 (early +2a / late −2a 抵消 → 伪"平衡", DP 逮 →
      修); 语义 = early 窗假定 clean (同一切 early-late 差分法前提)。
    ★诚实界: 只判 phase_cols 前两相 (l1↔l2); 其他对 (l1↔l3/l2↔l3) 调用方换列序再跑; 需 ≥3 相 (dev3 null 控制
      by design, 2 相无 null 轴 → abstain); 只检 onset 起互换 (t0 起互换 = 标签错从头, 幅值不可辨, 同 t0 教义);
      需 power 列 ∧ 窗间载荷重叠。**swap_flag = "互换 ∨ 双相反向漂移"候选** (RV-1 逮: l1−2%∧l2+2% 真双相 CT
      反向故障与互换幅值结构不可辨 → 现场辨接线 vs 双 CT; 工单方向仍对)。复制相台由 _phase_derived_guard
      区分 note (非"平衡")。

    来源: gym E2 赛道 sensor_phase_swap (EXP-E2-01 TRUE 可检域内: 域内 pctl 95.8×3seed, INV-sep 单相 gain 误判
    0 → §7 四闸 → 并回 src/ 2026-07-15)。产化改动: Δdev 裸窗差 → **共同支撑载荷分箱** (载荷相关相间差×窗移
    confound, probe 实测机制在)。DP 验收 = tests/sop/test_sensor_phase_swap_dp.py。
    """
    # 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 = {"swap_flag": False, "score": 0.0, "ddev": None, "detectable": None, "d12_persistent": 0.0}
    ph = [c for c in phase_cols if c in df.columns]
    if len(ph) < 3 or power_col not in df.columns:
        return {t: {**NULL, "note": "需≥3相 (dev3 null控制) ∧ 功率列; abstain"} for t in tids}
    df = df.sort_values([turbine_col, time_col], kind="mergesort").reset_index(drop=True)
    L = np.column_stack([df[c].to_numpy(float) for c in ph])
    with np.errstate(invalid="ignore", divide="ignore"):
        Ibar = np.nanmean(L, axis=1)
        dev = L / Ibar[:, None] - 1.0
    gen = ((df[gen_flag_col] == 1).to_numpy() if gen_flag_col and gen_flag_col in df.columns
           else (df[power_col].to_numpy(float) > gen_power_min))
    ok = gen & np.isfinite(dev).all(axis=1) & (Ibar > min_ibar)
    P = df[power_col].to_numpy(float)
    early, late = _windows(df[time_col], early_frac, late_frac)
    tv = df[turbine_col].astype(str).to_numpy()
    e_ok, l_ok = ok & early, ok & late
    if e_ok.sum() < 2 * min_bin_n or l_ok.sum() < 2 * min_bin_n:
        return {t: {**NULL, "note": "窗内 gen 样本不足; abstain"} for t in tids}
    # 共同支撑等宽载荷箱 (承四角图谱: 消载荷相关相间差 × 窗分布移伪签名)
    e_lo, e_hi = np.nanpercentile(P[e_ok], [2, 98])
    l_lo, l_hi = np.nanpercentile(P[l_ok], [2, 98])
    lo, hi = max(e_lo, l_lo), min(e_hi, l_hi)
    narrow = min(e_hi - e_lo, l_hi - l_lo)
    if not (hi > lo) or narrow <= 0 or (hi - lo) < min_overlap_frac * narrow:
        return {t: {**NULL, "note": "窗间载荷重叠不足→INSUFFICIENT"} for t in tids}
    edges = np.linspace(lo, hi, n_bins + 1)
    binid = np.clip(np.digitize(P, edges) - 1, 0, n_bins - 1)
    in_cs = (P >= lo) & (P <= hi)
    out = {}
    for t in tids:
        base = (tv == t) & ok
        d12p = 0.0
        e_base = base & early                                  # 可检域判定用 early 本底 (onset 前参照期):
        if e_base.sum() >= 2 * min_bin_n:                      # 全窗会被互换本身污染 (early+2a/late−2a 抵消→0)
            md = np.median(dev[e_base], axis=0)
            d12p = float(md[0] - md[1])
        detectable = abs(d12p) >= detectable_min_d12
        rec = {"swap_flag": False, "score": 0.0, "ddev": None,
               "detectable": bool(detectable), "d12_persistent": round(d12p, 5)}
        if not detectable:
            rec["note"] = ("疑似派生/复制相 (rel pair-diff <1e-4): swap 不可证 → INSUFFICIENT"
                           if _phase_derived_guard(L, Ibar, base)
                           else "平衡两相: 幅值互换无观测量变化, 结构性不可检 (非'已核干净')")
            out[t] = rec
            continue
        cs = base & in_cs
        dds = []
        for bn in range(n_bins):
            e_in = cs & early & (binid == bn); l_in = cs & late & (binid == bn)
            if e_in.sum() >= min_bin_n and l_in.sum() >= min_bin_n:
                dds.append(np.median(dev[l_in], axis=0) - np.median(dev[e_in], axis=0))
        if len(dds) < 2:
            rec["note"] = "共同支撑内<2载荷箱; abstain"
            out[t] = rec
            continue
        dd = np.median(np.vstack(dds), axis=0)                 # 箱间 median (载荷匹配后)
        d1, d2, d3 = float(dd[0]), float(dd[1]), float(dd[2])
        opp = (d1 > 0) != (d2 > 0)
        pair_min = min(abs(d1), abs(d2))
        null_ok = (abs(d3) < null_ratio * pair_min) if pair_min > 0 else False
        score = (pair_min / swap_thresh) if (opp and null_ok) else 0.0
        rec.update({"swap_flag": bool(score > 1.0), "score": round(float(score), 4),
                    "ddev": [round(d1, 5), round(d2, 5), round(d3, 5)]})
        out[t] = rec
    return out


def channel_semantics_shift_verdict(
    df: pd.DataFrame, *, phase_cols: list, power_col: str, time_col: str, turbine_col: str,
    gen_flag_col: str | None = None, gen_power_min: float = 50.0, min_bar: float = 10.0,
    quantity: str = "current", freq: str = "M", min_month_n: int = 200,
    level_jump: float | None = None, rel_tol: float = 1e-4,
) -> dict:
    """per-机组 **通道语义中途变更 (契约级数据异常)** — 三相电气通道身份指纹按月漂移检测.

    裁决对象 = 通道**语义/量纲是否在记录期内换制** (≠ 仪表故障, ≠ 机组部件): 若某台三相通道中途改测量口径
    (派生↔独立 ∨ 量纲/标度跳变), 则**跨该点的任何依赖该通道分析前后不可比** → 报 WARN (非阻断; 交付语义 =
    该台该通道"分段化", 调用方须按段切窗)。本函数 = parity/level/swap 三臂的**前置分段器** (跨换制点混算 =
    _phase_derived_guard 之外的第二类伪信号)。**电流/电压双支持** (quantity 参数, 见下 — 量纲判据的归一方式按物理分流)。

    返回 {tid: {"flag": bool, "regime_switch": bool, "level_jump": bool,
                "switch_month": 派生↔独立切换月(或None), "regimes": "DDDIIII"(每有效月 D/I 串),
                "level_jump_ratio": 量纲指纹相邻月最大比(≥1), "level_jump_month": 跳变月(或None),
                "n_months": 有效月数, "note"?: 弃权/WARN 摘要}}。

    ★两个正交签名 (任一触发 → flag; ∧/∨ 语义 = OR, 两者同点触发 = 强互证, GC9/GC17 真盘即如此):
      (a) **派生↔独立切换** (载荷无关, 二值, 电流/电压通用): per(台,月) _phase_derived_rel <rel_tol=派生(同一测量
          复制写三列), ≥rel_tol=独立测量; 有效月间 regime 变 = 中途换制。**复用 _phase_derived_guard 同判据**。
      (b) **量纲级 level 跳变** (连续; **归一方式按 quantity 物理分流, Physics First CLAUDE.md #1**):
          - `quantity="current"` (默认, level_jump 默认 3.0): 发电时 I∝P → 指纹 = **功率归一** median(bar/P) 相邻月比。
            **必功率归一**: 真单位/标度 artifact 破坏 I/P, 而季节载荷保持 I/P 不变 → 消掉"raw median 月间健康台
            ×1.5 季节摆动"混淆 (gucheng 实测: 健康 22 台 raw 月比最高 ~1.5×→功率归一后 <1.03×; GC9/GC17 换制点 ~8×)。
          - `quantity="voltage"` (level_jump 默认 1.5): **V≈恒定 (与 P 弱相关, gucheng corr 0.44 非 ∝P)** → 指纹 =
            **raw-level** median(bar) 相邻月比。**禁功率归一**: V/P 会随季节载荷反向摆动 = 假阳源 (与电流相反)。
            阈 1.5× 依据: gucheng 健康电压 raw 月比 max 1.072× (24 台×11 月, p99 1.022) → 1.5 给 1.40× 下margin,
            且抓 √3(L-L↔L-N 换算)=1.732× / 690↔400V=1.725× / V↔kV=1000×。

    ★可分性 (为何不误报健康波动): (电流) 三相共缩/季节/限电 → bar 与 P 同缩 → bar/P 不变 → (b) 静; (电压) V 本恒定
      → raw-level 月比 ≈1 → (b) 静。独立测量 pair-diff 不随载荷变 → (a) 静。故健康台 flag=False (gucheng 真盘: 电流
      24 台 0 FP + 电压 24 台 0 FP 实证)。

    ★诚实界:
      - **只判"换制发生"go/no-go, 不判换制后哪段可信** — 前后两段的仪表校核 (parity/level/swap) 须**按段独立跑**。
      - **level 跳变 ≠ 一定是 artifact**: 真 CT/VT 变比改档/电气重构也触发 (那也确是"语义换制", 跨点不可比 → WARN
        正当); 定性靠现场核 (铭牌/CT-VT 变比台账)。故 verdict = WARN 非"确诊", 属候选级 (SOP §0.2)。
      - **电压 sub-阈换算漏检 (已声明, 安全方向)**: <level_jump 的量纲变 (如某些 VT 变比 ~1.3×, 贴近健康地板)
        本门 raw-level 轴可能漏; 靠信号 (a) 切换 + 下游诊断补。**瞬态 excursion (非永久换制, 如 GC11/GC15 电压
        episode 单相 +3.8% 后复位) 不属本门 scope** — bar 平均后 ~1.3% << 阈, 正当不触发; 由 phase_level/parity 臂管。
      - 需 ≥2 相 ∧ power/time 列 ∧ ≥2 有效月; 缺 → note 显式弃权 (非静默"干净")。全程同一 regime 的台 (全复制或全独立)
        → (a) 不触发 (该台三相 parity 由各函数内 _phase_derived_guard 各自守 INSUFFICIENT, 与本函数正交)。

    来源: RV-1 E2 审 2026-07-15 真盘发现 (gucheng GC9/GC17 电流: 2025-03~07 复制相 rel~3e-6 ∧ 2025-08 起独立 +
    同月量纲 1049A→33A 跳变 = 契约级换制)。电压支持 = 2026-07-15 扩展 (V 物理异于 I → raw-level 归一)。gym
    LESSONS#44 契约层 backlog 落地。DP 验收 = tests/sop/test_channel_semantics_shift_dp.py (电流+电压双模式合成
    regime切换/量纲跳变/归一鲁棒 + gucheng 真盘电流 GC9/GC17 + 电压 0-flag 回归闸)。
    """
    if quantity not in ("current", "voltage"):
        raise ValueError(f"quantity 必为 'current'|'voltage', 收到 {quantity!r}")
    power_norm = quantity == "current"                         # 电流=功率归一; 电压=raw-level (物理分流)
    if level_jump is None:
        level_jump = 3.0 if power_norm else 1.5               # 电压健康地板窄→阈更严 (抓 √3 换算)
    # 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, "regime_switch": False, "level_jump": False, "switch_month": None,
            "regimes": "", "level_jump_ratio": 1.0, "level_jump_month": None, "n_months": 0}
    ph = [c for c in phase_cols if c in df.columns]
    if len(ph) < 2 or power_col not in df.columns or time_col not in df.columns:
        return {t: {**NULL, "note": "相通道<2 或缺功率/时间列; abstain"} for t in tids}
    cols = [turbine_col, time_col, power_col] + ph
    if gen_flag_col and gen_flag_col in df.columns:
        cols.append(gen_flag_col)
    d = df.loc[:, cols].copy()
    tt = pd.to_datetime(d[time_col], errors="coerce")
    keep = tt.notna()
    d, tt = d.loc[keep], tt.loc[keep]
    if len(d) == 0:
        return {t: {**NULL, "note": "时间列全不可解析; abstain"} for t in tids}
    ym = tt.dt.to_period(freq).astype(str).to_numpy()
    L = np.column_stack([pd.to_numeric(d[c], errors="coerce").to_numpy(float) for c in ph])
    import warnings
    with warnings.catch_warnings():
        warnings.simplefilter("ignore", RuntimeWarning)        # 全-NaN 行 nanmean 空切片 (无害)
        bar = np.nanmean(L, axis=1)                             # 三相均值 (电流→Ibar / 电压→Vbar)
    P = pd.to_numeric(d[power_col], errors="coerce").to_numpy(float)
    gen = ((pd.to_numeric(d[gen_flag_col], errors="coerce").to_numpy(float) == 1)
           if gen_flag_col and gen_flag_col in df.columns else (P > gen_power_min))
    ok = gen & np.isfinite(bar) & np.isfinite(P) & (bar > min_bar) & (P > gen_power_min)
    tv = d[turbine_col].astype(str).to_numpy()

    out = {}
    for t in tids:
        sel_t = (tv == t) & ok
        recs = []                                              # (ym, regime 'D'/'I', mag) per 有效月
        for mo in sorted(set(ym[sel_t])):
            m_sel = sel_t & (ym == mo)
            if m_sel.sum() < min_month_n:
                continue
            regime = "D" if _phase_derived_rel(L, bar, m_sel) < rel_tol else "I"
            with np.errstate(invalid="ignore", divide="ignore"):
                mag = (float(np.median(bar[m_sel] / P[m_sel])) if power_norm   # 电流: 功率归一 (I∝P)
                       else float(np.median(bar[m_sel])))                       # 电压: raw-level (V≈const)
            recs.append((mo, regime, mag))
        if len(recs) < 2:
            out[t] = {**NULL, "n_months": len(recs), "note": "有效月<2 (gen 样本不足); abstain"}
            continue
        regimes = "".join(r[1] for r in recs)
        switch_month = next((recs[i][0] for i in range(1, len(recs))
                             if recs[i][1] != recs[i - 1][1]), None)     # (a) 首个 regime 切换月
        max_ratio, lj_month = 1.0, None                                 # (b) 量纲指纹相邻月最大比
        for i in range(1, len(recs)):
            a, b = recs[i - 1][2], recs[i][2]
            if a > 0 and b > 0:
                ratio = max(a / b, b / a)
                if ratio > max_ratio:
                    max_ratio, lj_month = ratio, recs[i][0]
        regime_switch = switch_month is not None
        level_jump_flag = max_ratio > level_jump
        flag = regime_switch or level_jump_flag
        rec = {"flag": bool(flag), "regime_switch": bool(regime_switch),
               "level_jump": bool(level_jump_flag), "switch_month": switch_month,
               "regimes": regimes, "level_jump_ratio": round(float(max_ratio), 3),
               "level_jump_month": lj_month, "n_months": len(recs)}
        if flag:
            magdesc = "功率归一电流" if power_norm else "电压raw-level"
            parts = ([f"派生↔独立切换@{switch_month}"] if regime_switch else []) + \
                    ([f"{magdesc}×{max_ratio:.1f}@{lj_month}"] if level_jump_flag else [])
            rec["note"] = "通道语义中途变更 (契约级): " + " + ".join(parts) + " → 跨点不可比, 须按段切窗 (WARN)"
        out[t] = rec
    return out


# ==================== 机舱风速计校核 (W 三轴: 相对 / power耦合 / 二阶矩) ====================

# ★纪律 (EXP-W3-01, 状态表附带纪律洞的生产载体): 生产判 freeze/variance **必用 sensor_wind_var_verdict
# (二阶矩轴), 勿信 mean-ratio 轴** — mean 轴对 freeze/variance 的 gym"检出"是 confound 假因
# (freeze = 夏季冻值≠冬季 fleet 中位的季节偏, per台 ratio_shift −0.08~+0.50 方向乱, onset 换季即翻盲;
# variance = min_ws 非对称削低尾抬中位 [frac<min_ws 2%→13%, 中位+0.8] + 右偏膨胀抬中位, mean保真≠median保真),
# 脆/不可迁。二阶矩轴测真离差变 (freeze z 11-17 / variance z~10, 右因), 对 onset 季节/min_ws 地板不漂。
_WIND_FREEZE_VAR_DISCIPLINE = (
    "判 freeze/variance 必用 sensor_wind_var_verdict (二阶矩轴), 勿信 mean-ratio 轴 "
    "(mean 轴检出=季节/min_ws confound 假因, 脆/不可迁; EXP-W3-01)")


def _fleet_median_ratio(df, ws, ok, time_col, min_fleet_n):
    """fleet 同刻中位参照 + per行 ratio + **参照有效时刻掩码** (RV-1 W 审 F1 逮): per台时间戳错位 →
    同刻组只剩单台 → 参照=自身 → ratio≡1 = **静默假阴** (输出"看似干净"零弃权)。
    修法保守 (薄并回, 不改 gym 估计量): ratio 计算与 gym 逐位一致 (稀疏时刻照算); 另返回
    valid = 同刻有效贡献台 ≥ min_fleet_n 掩码, 调用方按**窗内 valid 占比** < 0.5 响亮弃权
    (对齐数据占比≈1 → 分数逐位不变; 全错位 → 占比 0 → abstain)。"""
    g = df.assign(_w=np.where(ok, ws, np.nan)).groupby(df[time_col])["_w"]
    fleet = g.transform("median").to_numpy()
    cnt = g.transform("count").to_numpy()
    with np.errstate(invalid="ignore", divide="ignore"):
        ratio = np.where(ok & (fleet > 0), ws / fleet, np.nan)
    return fleet, ratio, (cnt >= min_fleet_n)


def sensor_wind_fleet_verdict(
    df: pd.DataFrame, *, ws_col: str, time_col: str, turbine_col: str,
    gen_flag_col: str | None = None, min_ws: float = 3.0, ratio_thresh: float = 0.05,
    early_frac: float = 0.35, late_frac: float = 0.35, min_win_n: int = 50,
    min_fleet_n: int = 4,
) -> dict:
    """per-机组 **机舱风速计校核 — 相对轴** (机舱 ws / fleet 同时刻中位 ws 的早late比移).

    返回 {tid: {"score": |ratio_shift|, "flag": bool, "ratio_shift": signed 比移,
                "n_e"/"n_l": 窗内有效样本, "note"?: 弃权原因}}。

    ★机制 (no black box): 全场同刻见近同风 → fleet 中位 ws = 好参照; r_i(t)=ws_i/fleet中位ws(t);
      score = |median(r|late) − median(r|early)|, flag = score > ratio_thresh (0.05 ≈ 阜山自然
      NTF 台间散度 0.047 量级, EXP-W1-01 预注册锚)。抓 per台 bias/gain/drift onset
      (gym: bias+1 / gain×1.10 / drift+1/60d 均 pctl 94.4 ×3seed)。
    ★可分性: 真风事件 = fleet 共动 → 各台比不变 → 不 flag (clean fp 0; 真风空间共动 misattr 0,
      EXP-W1-01 H4)。
    ★诚实界:
      - **对 fleet-common 偏结构盲** (全场同偏 → fleet 中位同移 → 比不变; gym fleetbias 全场+1
        fleet_ratio_shift 0.0002 = 盲, 预期同温度 fleet-relative 教义) → fleet-common onset 检出
        用 sensor_wind_ntf_verdict (power 耦合轴)。
      - 只检 **onset 变化**; t0 起恒定偏 (装机即偏) 不可检 (早late 差分教义); drift↔step-bias
        不可分 (60d 漂 late 窗前完成 → 同签名; 无 onset 轴, EXP-W3-01 诚实注)。
      - **freeze/variance 勿用本轴**: 判 freeze/variance 必用 sensor_wind_var_verdict (二阶矩轴), 勿信 mean-ratio 轴 (mean 轴检出=季节/min_ws confound 假因, 脆/不可迁; EXP-W3-01) — mean 轴对二者的"检出"是 confound。
      - 检出地板 = 自然 NTF 台间散度 (~0.047, 阜山); 单台偏须超此。相对轴, 需 fleet ≥ min_fleet_n 台。
      - **前提: 时间戳须跨台对齐** (10min 栅格; RV-1 逮): fleet 参照按同刻中位聚合, per台时间戳
        错位 → 同刻贡献台<min_fleet_n → 参照有效时刻占比 <0.5 → **响亮弃权** note (非静默 ratio≡1
        假"干净"); ratio 数值本身与 gym 逐位一致 (稀疏时刻照算, 闸只管弃权语义)。**混合错位定量**
        (RV-1 复核实测): 占比贴 0.5 下沿时 shift 被稀释 ~×3 (0.139→0.047), 真 bias 可漏 —
        方向 = 漏检非假阳 (安全); 占比 <0.5 由闸兜住。
      - **批次污染上界 (RV-1 逮, 承 memory relative-criterion-batch-blindness)**: fleet 中位击穿点 =
        50%; 同向坏台占比 ≳1/3 起健康台可现**镜像反号 shift 假阳** (实测 2/4 台 bias+2 → 健康台 −0.15
        flag)。flag 台占比 >~1/3 或出现对称反号 shift → 疑 fleet 参照污染, 按批次场型处理勿逐台判。
    来源: gym W 赛道 sensor_wind_fleet (EXP-W1-01 相对轴成立 → EXP-W3-01 变体补全 → §7 四闸并回
    src/ 2026-07-16, gym 逻辑 as-is 薄搬运)。DP 验收 = tests/sop/test_wind_sensor_dp.py。
    """
    # 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 = {"score": 0.0, "flag": False, "ratio_shift": 0.0, "n_e": 0, "n_l": 0}
    if ws_col not in df.columns:
        return {t: {**NULL, "note": "机舱风速列缺; abstain"} for t in tids}
    if len(tids) < min_fleet_n:
        return {t: {**NULL, "note": f"fleet<{min_fleet_n}台, 相对参照不可估; abstain"} for t in tids}
    df = df.sort_values([turbine_col, time_col], kind="mergesort").reset_index(drop=True)
    gen = ((df[gen_flag_col] == 1).to_numpy() if gen_flag_col and gen_flag_col in df.columns
           else np.ones(len(df), dtype=bool))
    ws = df[ws_col].to_numpy(float)
    tid = df[turbine_col].astype(str)
    ok = gen & np.isfinite(ws) & (ws > min_ws)
    fleet, ratio, valid = _fleet_median_ratio(df, ws, ok, time_col, min_fleet_n)
    early, late = _windows(df[time_col], early_frac, late_frac)
    out = {}
    for t in tids:
        sel = (tid.values == t) & ok
        win = sel & (early | late) & np.isfinite(ratio)
        fv = float(valid[win].mean()) if win.any() else 0.0    # 参照有效时刻占比 (RV-1 F1 闸)
        re = ratio[sel & early]; rl = ratio[sel & late]
        re = re[np.isfinite(re)]; rl = rl[np.isfinite(rl)]
        if len(re) < min_win_n or len(rl) < min_win_n:
            out[t] = {**NULL, "n_e": int(len(re)), "n_l": int(len(rl)), "note": "窗内有效样本不足; abstain"}
            continue
        if fv < 0.5:
            out[t] = {**NULL, "n_e": int(len(re)), "n_l": int(len(rl)),
                      "note": f"fleet 参照有效时刻占比 {fv:.2f}<0.5 (同刻贡献台<{min_fleet_n}; "
                              f"时间戳错位/机群过稀) → 参照不可信; abstain"}
            continue
        sh = float(np.median(rl) - np.median(re))
        out[t] = {"score": round(abs(sh), 4), "flag": bool(abs(sh) > ratio_thresh),
                  "ratio_shift": round(sh, 4), "n_e": int(len(re)), "n_l": int(len(rl))}
    return out


def sensor_wind_ntf_verdict(
    df: pd.DataFrame, *, ws_col: str, power_col: str, mast_col: str, time_col: str, turbine_col: str,
    gen_flag_col: str | None = None, min_ws: float = 3.0, ratio_thresh: float = 0.05,
    early_frac: float = 0.35, late_frac: float = 0.35, power_min: float = 50.0,
    n_pc_bins: int = 20, min_bin_n: int = 20, min_valid_bins: int = 5,
    min_mast_n: int = 200, min_win_n: int = 50, min_fleet_n: int = 4,
) -> dict:
    """**机舱风速计校核 — power 耦合绝对轴** (fleet-common 偏 onset 检出, 相对轴/mast 快照双盲的洞).

    返回 {"verdict": "FLEET_COMMON_SHIFT" | "NO_FLEET_COMMON_SHIFT" | "INSUFFICIENT",
          "fleet_ratio_shift": fleet 中位早late比移 (核心量, |·|>ratio_thresh → 王牌 verdict),
          "per_turbine": {tid: {"score", "flag", "ratio_shift", "n_e", "n_l", "note"?}},
          "n_eff": 参与 fleet 中位的有效台数, "n_pc_bins": h(P) 有效功率箱数, "honesty": [...], "note": str}。

    ★机制 (no black box, EXP-W2-01 王牌): mast 窗 (ws_mast 有效期) 内拟合 h(P)=free-stream
      (P 分箱→per箱 ws_mast 中位→单调插值) — **P 与 ws_mast 均 nacelle 独立** → h(P) 是干净外部参照;
      年 round ratio_i(t)=机舱ws_i/h(P_i) 恢复变化检测 (mast 短窗 [阜山 2.5月] 无 pre-onset baseline
      做不到的)。传感器偏 (ws 移, P 不动) → 比移 flag; 真风 (P 与 ws 同动沿功率曲线) → 比稳不 flag。
      fleet_ratio_shift = median(per台早late比移) = fleet-common 信号 (gym fleetbias 全场+1:
      0.1677 全 seed vs 相对轴 0.0002 盲 = 王牌兑现); per台移 = 相对信号 (bias pctl 94.4)。
    ★可分性: power 是 nacelle 独立不变量 — clean (自然 P-ws 耦合) fp 0 全 seed (EXP-W2-01 H3);
      **INV 场景须物理自洽** (真风增 → P 增; EXP-W3-01 逮 gym realwind-008 抬 ws 不抬 P = 场景对
      power 轴不完备 → fp 1.0 是场景缺陷非算法; 生产 DP test 用 P 同动的物理自洽真风场景直锁)。
    ★诚实界 (skill 铁律遵, power-curve-ntf 教义):
      - 只检 fleet-common **onset 变化**; **恒定 t0 fleet-common 仍 INSUFFICIENT** (无变化信号 +
        单 mast 无空间代表性定不了绝对 NTF — fleet 中位比 1.08 vs 0.93 谁对, 无外部 NTF 无从判)。
      - 变化检测 robust to 恒定空间 NTF 散度 (常量不移); h(P) 拟合于 mast 窗, 年 round 外推设
        功率曲线稳 (同型同物理; 大修/换叶片/控制版本改 → 前提破, 调用方须知)。
      - 额定区 P 饱和 → h(P) 对 ws 失敏 (P 封顶不载风信息) → 高风段贡献弱, 信号主来自部分负荷区。
        深度/长时**限电** (P 被压而 ws 不动 = 与传感器偏同签名) bind 超窗内半数小时 → 前提破,
        须先过限电剥离 (curtailment-screen; 浅限电由中位数稳健性阻尼, RV-1 实测 500kW 封顶不破)。
      - **fleet-common 断言须够台数 (RV-1 逮)**: 有效台 < min_fleet_n → INSUFFICIENT (单台/少台的
        比移是个体传感器信号, 断言"批次/固件级共模"= 反向归因); n_eff 随 verdict 报。
      - 需 mast 列 (短窗即可) + power 列; mast 窗不足/功率曲线箱不足 → INSUFFICIENT 显式弃权。
      - **freeze/variance 勿用本轴**: 判 freeze/variance 必用 sensor_wind_var_verdict (二阶矩轴), 勿信 mean-ratio 轴 (mean 轴检出=季节/min_ws confound 假因, 脆/不可迁; EXP-W3-01)。
    来源: gym W 赛道 sensor_wind_ntf (EXP-W2-01 王牌兑现 H1∧H2∧H3 全达 → §7 四闸并回 src/ 2026-07-16,
    gym 逻辑 as-is 薄搬运)。DP 验收 = tests/sop/test_wind_sensor_dp.py。
    """
    HONESTY = [
        "(a) 只检 fleet-common onset 变化; 恒定 t0 fleet-common 偏 INSUFFICIENT (单 mast 无空间代表性, 需外部 NTF/IEC 功率曲线基准)。",
        "(b) h(P) 拟合 mast 窗, 年 round 外推设功率曲线稳; 大修/换叶片/控制版本变更期不可用。",
        "(c) 真风 (P 与 ws 同动) 比稳不 flag = 可分性来自 power 的 nacelle 独立性; INV 验证场景须物理自洽 (风增→P增)。",
        "(d) " + _WIND_FREEZE_VAR_DISCIPLINE + "。",
    ]
    # 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 = {"score": 0.0, "flag": False, "ratio_shift": 0.0, "n_e": 0, "n_l": 0}

    def _insuf(reason, pt=None, n_eff=0, frs=None):
        return {"verdict": "INSUFFICIENT", "fleet_ratio_shift": frs,
                "per_turbine": pt if pt is not None else {t: {**NULL, "note": reason} for t in tids},
                "n_eff": n_eff, "n_pc_bins": 0, "honesty": HONESTY, "note": reason}

    for c, nm in ((ws_col, "机舱风速"), (power_col, "功率"), (mast_col, "mast free-stream")):
        if c not in df.columns:
            return _insuf(f"{nm}列 ({c}) 缺 → power 耦合轴不可建; INSUFFICIENT")
    df = df.sort_values([turbine_col, time_col], kind="mergesort").reset_index(drop=True)
    gen = ((df[gen_flag_col] == 1).to_numpy() if gen_flag_col and gen_flag_col in df.columns
           else np.ones(len(df), dtype=bool))
    ws = df[ws_col].to_numpy(float)
    P = df[power_col].to_numpy(float)
    mast = df[mast_col].to_numpy(float)
    tid = df[turbine_col].astype(str)
    # 1. h(P) = free-stream, mast 窗拟合 (nacelle 独立)
    mok = gen & np.isfinite(mast) & np.isfinite(P) & (mast > min_ws) & (P > power_min)
    if mok.sum() < min_mast_n:
        return _insuf(f"mast 窗有效样本 {int(mok.sum())} < {min_mast_n} → h(P) 不可拟合; INSUFFICIENT")
    Pm, Wm = P[mok], mast[mok]
    edges = np.nanquantile(Pm, np.linspace(0, 1, n_pc_bins + 1))
    edges[0], edges[-1] = -np.inf, np.inf
    binid = np.clip(np.digitize(Pm, edges) - 1, 0, n_pc_bins - 1)
    bc, bw = [], []
    for b in range(n_pc_bins):
        m = binid == b
        if m.sum() >= min_bin_n:
            bc.append(float(np.median(Pm[m]))); bw.append(float(np.median(Wm[m])))
    if len(bc) < min_valid_bins:
        return _insuf(f"功率曲线有效箱 {len(bc)} < {min_valid_bins}; INSUFFICIENT")
    bc, bw = np.array(bc), np.array(bw)
    order = np.argsort(bc); bc, bw = bc[order], bw[order]
    # 2. year-round ratio = 机舱 ws / h(P)
    ok = gen & np.isfinite(ws) & np.isfinite(P) & (ws > min_ws) & (P > power_min)
    implied = np.interp(P, bc, bw, left=bw[0], right=bw[-1])
    with np.errstate(invalid="ignore", divide="ignore"):
        ratio = np.where(ok & (implied > min_ws), ws / implied, np.nan)
    early, late = _windows(df[time_col], early_frac, late_frac)
    pt, shifts = {}, []
    for t in tids:
        sel = (tid.values == t) & ok
        re = ratio[sel & early]; rl = ratio[sel & late]
        re = re[np.isfinite(re)]; rl = rl[np.isfinite(rl)]
        if len(re) < min_win_n or len(rl) < min_win_n:
            pt[t] = {**NULL, "n_e": int(len(re)), "n_l": int(len(rl)), "note": "窗内有效样本不足; abstain"}
            continue
        sh = float(np.median(rl) - np.median(re)); shifts.append(sh)
        pt[t] = {"score": round(abs(sh), 4), "flag": bool(abs(sh) > ratio_thresh),
                 "ratio_shift": round(sh, 4), "n_e": int(len(re)), "n_l": int(len(rl))}
    if len(shifts) < min_fleet_n:                              # RV-1 F2 逮: 少台比移=个体信号,
        frs = round(float(np.median(shifts)), 4) if shifts else None   # 断言"批次共模"=反向归因
        return _insuf(f"有效台 {len(shifts)} < {min_fleet_n} → fleet-common 不可断言 "
                      f"(per台结果照报, 个体信号请读 per_turbine); INSUFFICIENT",
                      pt=pt, n_eff=len(shifts), frs=frs)
    frs = float(np.median(shifts))
    common = abs(frs) > ratio_thresh
    verdict = "FLEET_COMMON_SHIFT" if common else "NO_FLEET_COMMON_SHIFT"
    note = (f"fleet_ratio_shift={frs:+.4f} (n_eff={len(shifts)}台, {'>' if common else '≤'} 阈 {ratio_thresh}): "
            + ("全场机舱风速计共模偏 onset 候选 (传感器批次/固件级, ws 移而 P 不动) — 候选级, "
               "确证须外部 NTF/现场标定" if common
               else "无 fleet-common onset 信号 (恒定 t0 偏本轴不可证, 见 honesty a)"))
    return {"verdict": verdict, "fleet_ratio_shift": round(frs, 4), "per_turbine": pt,
            "n_eff": len(shifts), "n_pc_bins": int(len(bc)), "honesty": HONESTY, "note": note}


def sensor_wind_var_verdict(
    df: pd.DataFrame, *, ws_col: str, time_col: str, turbine_col: str,
    gen_flag_col: str | None = None, min_ws: float = 3.0, studentize_k: float = 3.0,
    early_frac: float = 0.35, late_frac: float = 0.35, min_win_n: int = 50,
    min_fleet_n: int = 4,
) -> dict:
    """per-机组 **机舱风速计校核 — 二阶矩轴** (freeze/variance 的**右因**判据; mean 轴对二者=confound 假因).

    返回 {tid: {"score": |z| (fleet studentized), "flag": bool, "logmad_ratio": log10(D_late/D_early),
                "z": signed z, "note"?: 弃权原因}}。

    ★机制 (no black box): r_t = ws_i / fleet 同刻中位 ws (相对参照 → 去 fleet-common 真风);
      早/late 窗稳健离差 D = MAD(r); raw = log10(D_late/D_early); fleet studentize
      z = (raw − fleet中位) / (1.4826·fleet MAD), flag |z| > k。
      物理: freeze → 分子 const / 分母 varying → 相对离差**涨**; variance (湍流计/杯轴退化离差变)
      → 绕中位噪声涨 → 相对离差**涨** — 两型同向, 单轴双收 (gym: freeze z 11-17 / variance z~10,
      pctl 94.4 ×3seed 右因稳抓)。
    ★为何必用本轴 (纪律, EXP-W3-01 rollout 逮): 判 freeze/variance 必用 sensor_wind_var_verdict (二阶矩轴), 勿信 mean-ratio 轴 (mean 轴检出=季节/min_ws confound 假因, 脆/不可迁; EXP-W3-01) — mean-ratio 轴对 freeze/variance
      也"flag"但假因 (季节偏/min_ws 削尾), 换 onset 季节即翻盲。本轴测真二阶矩, 对 onset 季节/
      min_ws 地板不漂。
    ★可分性: 纯均值移 (真风 fleet 共动/fleet-common 湍流同涨) 由相对参照+studentize 归零
      (clean fp 0, realwind misattr 0, EXP-W3-01 H6)。
    ★诚实界:
      - **正交性部分证伪 (gym 实测)**: 对 per台加性 bias/gain 也漏 z~3 (贴 k=3 边界, 加性偏经
        1/fleet 渗离差; 比真信号小 ~10×) → 作 freeze/variance **检测器**无歧义; 作 1阶↔2阶
        **判别器**边界不清 (勿单凭本轴断言"非均值故障")。
      - freeze onset 首值 < min_ws → late 窗全滤 → 该型漏检 (seed edge, 安全方向: 不误报)。
      - fleet-common 湍流增 (全台 TI 同涨) 由 studentize 归零 → 场级湍流变化本轴盲 (by design)。
      - 相对轴, 需 fleet ≥ min_fleet_n 台有效; 窗内样本/离差不可估 → 显式弃权。**前提: 时间戳须跨台
        对齐** (同 sensor_wind_fleet_verdict; 同刻贡献台数闸 _fleet_median_ratio, 错位 → 响亮弃权)。
    来源: gym W 赛道 sensor_wind_var (EXP-W3-01 §3 预注册补轴, H6 CONFIRMED → §7 四闸并回 src/
    2026-07-16, gym 逻辑 as-is 薄搬运)。DP 验收 = tests/sop/test_wind_sensor_dp.py。
    """
    # 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 = {"score": 0.0, "flag": False, "logmad_ratio": 0.0, "z": 0.0}
    if ws_col not in df.columns:
        return {t: {**NULL, "note": "机舱风速列缺; abstain"} for t in tids}
    df = df.sort_values([turbine_col, time_col], kind="mergesort").reset_index(drop=True)
    gen = ((df[gen_flag_col] == 1).to_numpy() if gen_flag_col and gen_flag_col in df.columns
           else np.ones(len(df), dtype=bool))
    ws = df[ws_col].to_numpy(float)
    tid = df[turbine_col].astype(str)
    ok = gen & np.isfinite(ws) & (ws > min_ws)
    fleet, ratio, valid = _fleet_median_ratio(df, ws, ok, time_col, min_fleet_n)
    early, late = _windows(df[time_col], early_frac, late_frac)

    def _madv(a):
        a = a[np.isfinite(a)]
        return float(np.median(np.abs(a - np.median(a)))) if len(a) else np.nan

    raw = {}
    for t in tids:
        sel = (tid.values == t) & ok
        win = sel & (early | late) & np.isfinite(ratio)
        if win.any() and float(valid[win].mean()) < 0.5:       # 参照有效时刻占比闸 (RV-1 F1, 同 fleet 轴)
            continue                                            # → 该台落入弃权分支 (raw 缺)
        re = ratio[sel & early]; rl = ratio[sel & late]
        ne = int(np.isfinite(re).sum()); nl = int(np.isfinite(rl).sum())
        de = _madv(re); dl = _madv(rl)
        if ne >= min_win_n and nl >= min_win_n and de > 1e-4 and dl > 1e-4:
            raw[t] = float(np.log10(dl) - np.log10(de))
    if len(raw) < min_fleet_n:
        return {t: {**NULL, "note": f"有效台 {len(raw)} < {min_fleet_n}, fleet studentize 不可估; abstain"}
                for t in tids}
    vals = np.array(list(raw.values()))
    med = float(np.median(vals)); scale = max(float(1.4826 * np.median(np.abs(vals - med))), 1e-3)
    out = {}
    for t in tids:
        if t not in raw:
            out[t] = {**NULL, "note": "窗内有效样本/离差不可估; abstain"}
            continue
        z = (raw[t] - med) / scale
        out[t] = {"score": round(abs(z), 4), "flag": bool(abs(z) > studentize_k),
                  "logmad_ratio": round(raw[t], 4), "z": round(z, 3)}
    return out