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

"""FUSION 振动诊断框架 v2 — 厂商经验值 × 谱线模型 的结构化融合.

═══ v1 自查出五处错, v2 全修 (每处都有如东实证) ═══

★错1 覆盖面凭声明不凭证据 (v1 最严重的错, 且是它自己该逮的那类)
   v1 在 SOURCE_SPEC 里写死 MODEL covers broadband_level, 于是 12# 那次
   模型报"正常"被当成"查过宽带认为正常"; 真相是**模型当时根本没接 rms_200 判据**。
   ⇒ v2: **"正常" 不是可默认的状态**。报正常必须同时登记 checked=[...检查了什么]。
      登记为空 ⇒ 该结论是 INSUFFICIENT 不是 正常。
   (同族 memory: verdict-field-exists-but-unused-in-criteria / parse-failure-read-as-positive-result)

★错2 真分歧一律升级 ⇒ 告警洪水
   如东四方对照实测: 38台里四源一致的**只有09#** ⇒ **分歧是常态不是例外**。
   v1 规则会把约30台升级为"需现场" = 等于没排序。
   ⇒ v2: 升级须同时满足 (a)跨度>=2级 (b)判据**独立**(不同信号链) (c)双方都有登记检查。

★错3 同源判据当独立分歧
   CMS 的 rms_200 与我 L4b 的宽带层**是同一个量**。二者不一致是方法差异不是独立冲突。
   ⇒ v2: signal_id=(测点,判据) 重叠 ⇒ 标 METHOD_DIFF 不升级。
   (memory control-logic-derived-channels-not-independent / alarm-same-sensor-not-independent)

★错4 路由器自身的批次盲
   R3(台间离散<1.35 ⇒ 只能用趋势) 遇批次退化时: 离散同样小、趋势同样一起涨 ⇒ **两条路都盲**。
   而 Indicator_* **没有绝对锚** ⇒ 分不出"紧且健康"与"紧且全坏"。
   ⇒ v2: R3 且无绝对锚 ⇒ 显式挂 UNRESOLVABLE_BATCH_BLIND, 不静默路由。
   (Gemini 双模审第②硬伤在路由器里的复现)

★错5 缺厂商的运维设计 — 这部分是厂商经验值真正值钱的地方, 我原模型完全没有
   Hysteresis=3 (防告警抖动; 我的模型逐窗裁决会翻来翻去)
   TargetCount=15 (最小证据量门; 我没有"样本不够就不判")
   Red/Yellow≈1.4~2.1 (分级响应非二值)
   WTCCode 绑定 (**送达闸** — 报了要有人收到; 如东 631/639 无码 = 报了传不回)
"""
from __future__ import annotations
import numpy as np

EVIDENCE_TYPES = ('spectral_line', 'broadband_level', 'cepstral_indicator',
                  'iso_velocity', 'impact_time_domain', 'data_quality')

LEVELS = ('INSUFFICIENT', '正常', '参考', '候选·记基线', '候选', '准定论·预警', '确诊')
_LI = {v: i for i, v in enumerate(LEVELS)}
_DEFECT_LEVELS = {'候选', '准定论·预警', '确诊'}


# ══════════════════════════════════════════════════════════════════
# ① 判据路由器 — 判据用 level 还是趋势由 fleet 离散度定, 不是固定的
# ══════════════════════════════════════════════════════════════════
def route_criterion(per_turbine_values, *, rpm_values=None, iso_anchor=None,
                    spread_min=1.35, dim_corr_max=0.99, min_turbines=10):
    v = np.asarray(list(per_turbine_values.values() if hasattr(per_turbine_values, 'values')
                        else per_turbine_values), dtype=float)
    v = v[np.isfinite(v) & (v > 0)]
    out = dict(route=None, spread=np.nan, reason='', flags=[])
    if len(v) < min_turbines:
        return {**out, 'route': 'R0_INSUFFICIENT', 'reason': f'有效台数 {len(v)} < {min_turbines}'}
    if rpm_values is not None:
        r = np.asarray(rpm_values, dtype=float); n = min(len(r), len(v))
        if n >= min_turbines:
            c = np.corrcoef(v[:n], r[:n])[0, 1]
            if np.isfinite(c) and abs(c) > dim_corr_max:
                return {**out, 'route': 'R2_DISABLE',
                        'reason': f'与转速 corr={c:.4f} ⇒ 该量就是转速本身, 阈值与量纲无关'}
    sp = float(np.percentile(v, 90) / np.median(v))
    # ★2026-08-22 双模审 (Gemini③) 采纳: spread_min=1.35 是连续分布上的任意切点 —
    #   如东 14 判据×6 窗 p90/p50 分布 p25 1.11 / p50 1.27 / p75 1.38, 1.35 附近无自然断点 (最大相邻比仅 1.13×)。
    #   改为**连续量**输出: σ_log = ln(p90/p50)/1.2816 (对数正态假设), min_detectable_x = exp(3σ_log)
    #   = 一台要被判"3σ 离群"至少得高出中位多少倍。R3/R4 的切点保留为**[暂行, 单场经验]**参数, 不再当结论。
    #   ★与切点无关的硬结论只有一条: 无绝对锚的相对判据对"批次退化"盲 (UNRESOLVABLE_BATCH_BLIND), 任何 spread 都挂。
    sig = float(np.log(sp) / 1.2816) if sp > 1 else 0.0
    mdx = float(np.exp(3 * sig))
    out.update(spread=sp, sigma_log=sig, min_detectable_x=mdx,
               spread_min_note='[暂行 n=1 场] spread_min 切点为经验值, 无自然断点; 用 min_detectable_x 连续判读')
    if iso_anchor:
        return {**out, 'route': 'R1_ABSOLUTE',
                'reason': f'有国际标准 {iso_anchor[0]} ⇒ 绝对锚**双向**不可被统计值覆盖'}
    flags = ['UNRESOLVABLE_BATCH_BLIND']          # 无绝对锚 ⇒ 批次盲, 与 spread 无关
    if sp < spread_min:
        return {**out, 'route': 'R3_TREND_ONLY', 'flags': flags,
                'reason': (f'台间 p90/p50={sp:.2f} (σ_log={sig:.3f}, 3σ 离群需 ≥{mdx:.2f}× 中位) < 暂行切点 {spread_min}; '
                           f'无绝对锚 ⇒ 分不出"紧且健康"与"紧且全坏"')}
    return {**out, 'route': 'R4_LEVEL_TREND', 'flags': flags,
            'reason': f'台间 p90/p50={sp:.2f} (σ_log={sig:.3f}, 3σ 离群需 ≥{mdx:.2f}× 中位) ≥ 暂行切点 {spread_min}'}


# ══════════════════════════════════════════════════════════════════
# ② 厂商运维设计: 迟滞 + 最小证据量
# ══════════════════════════════════════════════════════════════════
def apply_hysteresis(window_levels, *, h=3):
    """厂商 Hysteresis=3 的通用化: 连续 h 窗同向才改级, 否则保持.

    我原模型逐窗独立裁决 ⇒ 会翻来翻去。这是厂商经验值里真正值钱的一条。
    """
    if not window_levels: return None, []
    cur = window_levels[0]; run = 1; trail = [cur]
    for lv in window_levels[1:]:
        if lv == cur: run += 1
        else:
            run = 1 if not (trail and lv == trail[-1]) else run + 1
            if run >= h: cur = lv; run = 0
        trail.append(cur)
    return cur, trail


def min_evidence_gate(n, *, n_min=30, vendor_n=15):
    """厂商 TargetCount=15。我取 30: 3σ 用 15 样本估 σ 的相对标准误 = 1/sqrt(2(N-1)) = 19%
    ⇒ 阈值本身带 ±57% 抖动。n_min=30 把它压到 ±39%。"""
    if n >= n_min: return True, f'n={n} >= {n_min}'
    if n >= vendor_n: return False, f'n={n} 达厂商 TargetCount={vendor_n} 但 < {n_min} (3σ估计不稳) ⇒ 只记基线不定级'
    return False, f'n={n} < 厂商 TargetCount={vendor_n} ⇒ 证据量不足, 不判'


# ══════════════════════════════════════════════════════════════════
# ③ 跨体系融合 — 覆盖凭证据 / 排同源 / 按跨度升级
# ══════════════════════════════════════════════════════════════════
def _checked_types(v):
    return {c.get('evidence_type') for c in v.get('checked', []) if c.get('evidence_type')}

def _signals(v):
    """物理信号身份 — 用显式 signal 字段, **不用判据名**。
    CMS 的 'rms_200' 与我的 'L1b_rms200_同档分位' 名字不同但**是同一个量**;
    按名字比会把同源判据当独立分歧 (v2 首跑在 12# 上实逮)。"""
    return {c.get('signal') or f"{c.get('sensor')}/{c.get('criterion')}"
            for c in v.get('checked', [])}

def _driver_ev(v):
    """驱动该体系定级的证据类型。盲区检验必须用它, 不能用 checked 全集之差 ——
    12# 实逮: CMS 判红靠 rms_200(宽带) 而全集之差算出 cepstral_indicator,
    于是报'模型盲在倒谱'——可 CMS 的红根本不是倒谱给的。"""
    d = v.get('driver')
    if d: return {d} if isinstance(d, str) else set(d)
    return _checked_types(v)

def normalize_verdict(v):
    """★错1 修: '正常' 不是可默认的状态。没登记检查项 ⇒ 降为 INSUFFICIENT。"""
    v = dict(v)
    if not v.get('checked'):
        if v.get('level') in ('正常', '参考'):
            v['_downgraded'] = f"原报「{v['level']}」但未登记任何检查项 ⇒ 降为 INSUFFICIENT"
            v['level'] = 'INSUFFICIENT'
    return v


def fuse(verdicts, *, escalate_span=2, positive_anchor=False, anchors=None, component_class=None):
    """多判据体系裁决。返回覆盖限定的结论, 不是一个裸等级。"""
    vs = [normalize_verdict(v) for v in verdicts]
    vs = [v for v in vs if v.get('level') in _LI]
    if not vs:
        return dict(level='INSUFFICIENT', mechanism='NO_VALID_VERDICT', rationale='无有效裁决输入',
                    covered=[], uncovered=list(EVIDENCE_TYPES), dissent=[], downgraded=[])
    covered = set().union(*[_checked_types(v) for v in vs]) if vs else set()
    down = [f"{v['source']}: {v['_downgraded']}" for v in vs if v.get('_downgraded')]
    real = [v for v in vs if v['level'] != 'INSUFFICIENT']
    if not real:
        return dict(level='INSUFFICIENT', mechanism='ALL_INSUFFICIENT',
                    rationale='所有体系都无登记检查项', covered=sorted(covered),
                    uncovered=sorted(set(EVIDENCE_TYPES) - covered), dissent=[], downgraded=down)
    hi = max(real, key=lambda v: _LI[v['level']]); lo = min(real, key=lambda v: _LI[v['level']])
    span = _LI[hi['level']] - _LI[lo['level']]
    base = dict(covered=sorted(covered), uncovered=sorted(set(EVIDENCE_TYPES) - covered),
                dissent=[f"{v['source']}:{v['level']}" for v in real], downgraded=down)

    def cap(lv, mech, why, driver=None):
        # ★封顶只封**自定门限**的结论 (ISO/厂商已校准门限不封)。
        # ★2026-08-23: 全局 positive_anchor → 按部件类 anchor_cap (登记表 reference/<场>/positive_anchors.json)
        # ★2026-08-23 修: 校准与否只看**驱动该定级的证据类型**的登记项 (告警证据力). 原写法 any(全部登记项) 被 ISO 项 (每台都登记) 豁免,
        #   CMS 由未整定 rms_200/Indicator 判红的台永远不封顶 (12# 实逮: 红告警 rms_200 → 准定论·预警, 而裁决为 候选).
        _dev = _driver_ev(driver) if driver else set()
        _drv_items = [c for c in (driver or {}).get('checked', []) if not _dev or c.get('evidence_type') in _dev]
        cal = any(c.get('calibrated') for c in _drv_items)
        unrepro = [c.get('criterion') for c in _drv_items if c.get('reproducible') is False]
        if unrepro:
            why += f" (驱动判据定义不可复现: {sorted(set(unrepro))})"
        if lv in _DEFECT_LEVELS and _LI[lv] > _LI['候选'] and not cal and not positive_anchor:
            if anchors is not None:
                from sop.discriminators import anchor_cap as _ac
                ntypes = len({c.get('evidence_type') for c in (driver or {}).get('checked', [])
                              if c.get('evidence_type') and c.get('value') is not None})
                ci = _ac(component_class, anchors, n_evidence_types=ntypes)
                if ci['cap'] in ('准定论·预警', '定论'):
                    why += f" (部件类锚 tier={ci['tier']}: {ci['reason']} ⇒ 不封顶)"
                    return dict(level=lv, mechanism=mech + '+ANCHOR_OK', rationale=why, **base)
                return dict(level='候选', mechanism=mech + '+ANCHOR_CAP',
                            rationale=why + f" ★部件类锚 tier={ci['tier']} ({ci['reason']}) ⇒ 封顶「候选」", **base)
            return dict(level='候选', mechanism=mech + '+ANCHOR_CAP',
                        rationale=why + ' ★结论由**未校准的自定相对门限**驱动且无正样本锚 ⇒ 封顶「候选」', **base)
        if not cal and lv in _DEFECT_LEVELS and _LI[lv] > _LI['候选']:
            pass  # (anchors/positive_anchor 分支已处理)
        if cal and lv in _DEFECT_LEVELS:
            why += ' (门限已校准: ISO/厂商定值 ⇒ 不受无正样本锚封顶约束)'
        return dict(level=lv, mechanism=mech, rationale=why, **base)

    if span == 0:
        return cap(hi['level'], 'AGREE',
                   f'{len(real)} 个体系一致'+(f" (另 {len(down)} 个因未登记检查项降为 INSUFFICIENT)" if down else ''), hi)
    # 盲区检验: 高者的证据类型, 低者查过没有
    hi_ev = _driver_ev(hi) - _checked_types(lo)
    if hi_ev:
        return cap(hi['level'], 'LOW_IS_BLIND',
                   (f"{lo['source']}判「{lo['level']}」但它**未检查**驱动证据类型 {sorted(hi_ev)} "
                    f"(它查的是 {sorted(_checked_types(lo))}) ⇒ 其'正常'在此无信息、不得用于销案 "
                    f"⇒ 按 {hi['source']}"), hi)
    # ★错3: 同源判据 ⇒ 方法差异不升级
    if _signals(hi) & _signals(lo):
        return cap(hi['level'], 'METHOD_DIFF_SAME_SIGNAL',
                   (f"两者算的是**同一个量** {sorted(_signals(hi) & _signals(lo))} ⇒ 非独立, "
                    f"分歧属**方法差异**(阈值口径不同)不属独立冲突 ⇒ 不升级, 按较高者, "
                    f"并挂『需复核口径』"), hi)
    # ★错2: 按跨度升级
    if span >= escalate_span:
        up = LEVELS[min(_LI[hi['level']] + 1, len(LEVELS) - 1)]
        return cap(up, 'TRUE_DISAGREEMENT_ESCALATE',
                   (f"{hi['source']}「{hi['level']}」vs {lo['source']}「{lo['level']}」跨度{span}级, "
                    f"信号链独立且双方都有登记检查 ⇒ 真分歧, 不取任一, 升级 "
                    f"(分歧点信息量 > 任一单侧结论)"), hi)
    return cap(hi['level'], 'MINOR_DISAGREEMENT',
               f"跨度仅{span}级 < {escalate_span} ⇒ 不升级, 按较高者", hi)