# -*- coding: utf-8 -*-
"""reactive_validation.py — 无功功率控制校核: per机组无功调度偏差 (变流器无功调节异常候选).

裁决对象 = fleet 内 **单机无功 Q (VAr) 显著偏离机群共同调度水平** 的候选 (变流器/无功控制侧) —
与有功 P (W) **物理正交** (Q=VAr 无功 vs P=W 有功)。区别 sensor 三相 imbalance (仪表) / gear (机械)。

来源: gym RQ 赛道 (EXP-RQ1-01 bootstrap TRUE, 五门 H0-H4 齐过: 判别 RCV 1.158 / 无功偏差 recovery 1.0×3seed /
反臆造 0; 真盘 hailesi 台17 median Q −76 kVAr ≈ 8.6× fleet)。gym 分数≠交付, 本件 = 走设计 §7 四闸后的生产实现。

★物理/统计 trace (第一性, no black box):
  健康机群近-unity 功率因数调度 (园区 AVC/PF 指令趋一致) → per台 loaded 态中位无功 Q 近机群共同水平;
  单机变流器无功调节异常 (指令跟踪差/无功环故障) → 该台 Q 强偏 → dev=|medQ−fleet_common| 抬 → flag (候选级筛查)。
  metric = per台 loaded (P>loaded_min) 中位 Q; fleet_common = 各台 medQ 中位; dev = |medQ−fleet_common|;
  |Q|/S (无功占视在功率比) = 调度强度探针 (S=√(P²+Q²))。

★别名前提已核 (承 LESSONS #50/#51, 数据面第0步硬门; RV-1 逮修数): p_reactive 逐台 corr vs 已用通道
  (hailesi 真盘逐台 |corr(Q,P)| 中位 **0.111** / max **0.302**; vs grid电流 中位 0.107) — **≪0.5 → 真分立传感通道**
  (非 actual_torque=P/ω 那类 0.9998 派生别名)。**注**: gym 早报 "0.042/<0.13" 系 pooled/全局 corr (偏乐观), 逐台真值
  0.11/0.30, 结论(分立非别名)稳但**引用取 <0.5 界非具体小数**; 调用方跨场须逐台重核 (新场 Q vs P/S)。

★Q-P 变化判别 (RV-1 逮 + 真盘证, in-data 部分排调度混杂): per台 med Q **随功率 P 强变** (Q≈P·tanφ, 台17
  binQ −10→−54kVAr/P升) = **恒定功率因数/电压droop 调度签名** → 偏调度侧 (印证候选非确诊); **平** (正常台 ±3kVAr
  across P) = 恒偏 (config/故障型)。本件出 qp_span (P分箱 med Q 极差) + dispatch_like → **像变桨电机温度信道那样
  in-data 部分辨调度vs故障** (但 suggestive 非 dispositive, 终判仍需 Q-setpoint 日志)。

★诚实界 (headline confound + 3原则审):
  - **调度混杂 (headline, 最重要)**: 无功部分是**电网/园区调度指令** (AVC/PF setpoint) → fleet-relative 天然离群
    **可能是执行逐台 Q 指令 (非故障)** 而非变流器异常 → **候选非确诊, 分离控制差异 vs 故障需园区 Q-setpoint 日志**
    (无则 INSUFFICIENT-by-归因)。检出器效度 (注入 recovery / 反臆造, gym H3/H4) 与调度**无关** (对合成偏置有效),
    但真盘天然离群的**归因**承此混杂 (同 fatigue TI-exposure / yaw environment / 电机 measurement-side 型: 信号真、归因需外部数据)。
  - **相对非绝对**: fleet-relative 排序/离群, 非绝对无功健康值; 全场一致偏 (整场同 AVC 指令) → fleet_common 吸收 → **批次盲**
    (承 relative-criterion-batch-blindness) → 批次嫌疑场须绝对基准/园区指令对照。
  - **loaded 条件化**: idle/停机态 Q = standby 无功 (非发电调度) → 必 loaded (P>loaded_min) 条件化; loaded 样本不足 → 弃权。
  - **符号双向**: Q 可正 (过励供无功) 或负 (欠励吸无功); dev=|medQ−fleet| 两向离群皆捕; sign_hint 报偏向。
  - 需 reactive_col + power_col; turbine id 列全空 → INSUFFICIENT (承 yaw turbine_hash 坏列, §48 F3)。

DP 验收 = tests/sop/test_reactive_power_dp.py。纯核心无 I/O 可测; per-farm 加载在调用方。
"""
from __future__ import annotations

import numpy as np
import pandas as pd


def _robust_hi(vals, cv_floor: float, k_mad: float):
    """fleet 高离群阈: hi = median + k·**地板 scale** (max(1.4826·MAD, cv_floor·|median|)) — 承 yaw §48 F2
    地板 scale 防 MAD≈0 紧 bulk 漏 clean 离群; 批次盲仍保 (全场一致 → 无值超阈)。返回 (median, hi_thresh, robust_cv)。"""
    v = np.asarray([x for x in vals if x is not None and np.isfinite(x)], float)
    if v.size < 5:
        return None, None, None
    med = float(np.median(v)); mad = float(np.median(np.abs(v - med)))
    rcv = (1.4826 * mad) / abs(med) if med != 0 else np.inf
    scale = max(1.4826 * mad, cv_floor * abs(med)) if med != 0 else (1.4826 * mad or float(np.std(v)))
    return med, med + k_mad * scale, round(rcv, 4)


def reactive_power_verdict(
    df: pd.DataFrame, *, reactive_col: str, power_col: str, turbine_col: str,
    loaded_min: float = 200.0, min_loaded_n: int = 500,
    k_mad: float = 3.5, uniform_cv_floor: float = 0.03,
) -> dict:
    """per-机组 **无功功率调度偏差** 校核 (变流器无功调节异常候选; 候选级非确诊).

    返回 {tid: {"med_q": loaded中位无功, "dev": |medQ−fleet_common|, "flag": bool, "z": fleet稳健z,
                "absq_over_s": |Q|/S 调度强度, "sign_hint": over|under (供/吸无功), "n_loaded": 样本, "note"?}}。
    另 {"_fleet": {"fleet_common_q": ..., "hi_thresh": ..., "robust_cv": ..., "verdict": CANDIDATE_RANKING|INSUFFICIENT,
                   "honesty": [...]}}。
    机制: per台 loaded(P>loaded_min) 中位 Q → fleet_common(各台medQ中位) → dev=|medQ−fleet_common|;
      flag = dev > 地板scale稳健阈 (承 yaw §48 F2)。**headline: 调度混杂** — 见模块诚实界。
    """
    HON = [
        "(a) 调度混杂 headline: 无功部分=电网/园区AVC/PF指令 → fleet离群可能执行逐台Q指令非故障 → 候选非确诊, "
        "分离控制差异vs故障需园区Q-setpoint日志(无则INSUFFICIENT-by-归因); 检出器效度(注入recovery)与调度无关但归因承混杂。",
        "(b) 相对非绝对: fleet-relative离群非绝对无功健康值; 全场一致偏(同AVC指令)→fleet_common吸收→批次盲(须绝对基准/指令对照)。",
        "(c) loaded条件化: idle/停机Q=standby非发电调度→必P>loaded_min条件化; loaded样本不足→弃权。",
        "(d) 别名前提(承#50/#51): p_reactive须真分立通道(逐台corr vs P/S<0.5, 非P/ω派生别名); 跨场重核。",
        "(e) 需reactive+power列 ∧ 有效turbine id列(全空→INSUFFICIENT); 符号双向(过励+/欠励−, dev两向捕)。",
    ]
    # 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 = {"med_q": None, "dev": None, "flag": False, "z": None, "absq_over_s": None,
            "sign_hint": None, "n_loaded": 0}

    def _wrap(reason, per=None):
        base = per or {t: {**NULL, "note": reason} for t in tids}
        base["_fleet"] = {"fleet_common_q": None, "hi_thresh": None, "robust_cv": None,
                          "verdict": "INSUFFICIENT", "honesty": HON, "note": reason}
        return base

    if reactive_col not in df.columns or power_col not in df.columns:
        return _wrap("无 reactive/power 列; abstain (INSUFFICIENT)")
    if df[turbine_col].notna().sum() == 0 or (len(tids) == 1 and tids[0] in ("nan", "None", "")):
        return _wrap("turbine id 列全空/坏列 (如 turbine_hash 全 NaN) → 无法分组; INSUFFICIENT")
    Q = pd.to_numeric(df[reactive_col], errors="coerce").to_numpy(float)
    P = pd.to_numeric(df[power_col], errors="coerce").to_numpy(float)
    tv = df[turbine_col].astype(str).to_numpy()
    loaded = np.isfinite(Q) & np.isfinite(P) & (P > loaded_min)
    S = np.sqrt(P ** 2 + Q ** 2)
    per = {}
    for t in tids:
        m = loaded & (tv == t)
        if m.sum() < min_loaded_n:
            per[t] = {**NULL, "n_loaded": int(m.sum()), "note": "loaded(发电态)样本不足; INSUFFICIENT"}
            continue
        medq = float(np.median(Q[m]))
        with np.errstate(all="ignore"):
            aqs = float(np.median(np.abs(Q[m]) / S[m]))
        # Q-P 变化判别 (RV-1): P 分箱 med Q 极差; 强变=调度PF签名(偏调度), 平=恒偏(config/故障型)
        qb = [float(np.median(Q[m & (P >= lo) & (P < hi)]))
              for lo, hi in ((loaded_min, 600.0), (600.0, 1200.0), (1200.0, 1e12))
              if (m & (P >= lo) & (P < hi)).sum() >= 50]
        qp_span = round(max(qb) - min(qb), 1) if len(qb) >= 2 else None
        per[t] = {"med_q": round(medq, 2), "dev": None, "flag": False, "z": None,
                  "absq_over_s": round(aqs, 4), "sign_hint": ("over" if medq > 0 else "under"),
                  "qp_span": qp_span, "dispatch_like": (qp_span is not None and abs(qp_span) > 30.0),
                  "n_loaded": int(m.sum()),
                  "note": ("med_q 随 loaded_min 变 (Q-P 强变工况台, qp_span 大); flag 稳但标量非稳定物理量, "
                           "qp_span 大→调度PF签名偏调度侧" if (qp_span is not None and abs(qp_span) > 30.0) else None)}
    good = {t: per[t]["med_q"] for t in tids if per[t]["med_q"] is not None}
    if len(good) < 5:
        return _wrap("合格台数 <5 → fleet 稳健离群不可估; INSUFFICIENT", per)
    fleet_common = float(np.median(list(good.values())))
    devs = {t: abs(good[t] - fleet_common) for t in good}
    med_d, hi_thr, rcv = _robust_hi(list(devs.values()), uniform_cv_floor, k_mad)
    scale = max(1.4826 * float(np.median(np.abs(np.array(list(devs.values())) - med_d))),
                uniform_cv_floor * abs(med_d) if med_d else 0.0) or 1.0
    for t in good:
        per[t]["dev"] = round(devs[t], 2)
        per[t]["z"] = round((devs[t] - med_d) / scale, 2) if scale > 0 else 0.0
        per[t]["flag"] = bool(devs[t] > hi_thr)
    n_flag = sum(1 for t in good if per[t]["flag"])
    per["_fleet"] = {"fleet_common_q": round(fleet_common, 2), "hi_thresh": round(hi_thr, 2),
                     "robust_cv": rcv, "verdict": "CANDIDATE_RANKING", "honesty": HON,
                     "note": f"无功调度偏差 (候选级, 调度混杂 caveat 见 honesty a): fleet_common={fleet_common:.1f}kVAr; "
                             f"高偏离 flag {n_flag} 台 (归因须 Q-setpoint 日志)。"}
    return per