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

# -*- coding: utf-8 -*-
r"""★① 置信度**接线**（用户令 2026-10-06）：把 `vib_confidence` 的五轴评分接到现有产物上。

设计要点（刻意保守，防"凑分"）：
- **只读现有产物**：融合表行（`fusion_table`）+ 振动 handoff（`per_turbine` 列表）。**不新增数据源**。
- **缺数的轴显式标 `[INS]`** 并计 0 分，同时在 `notes` 里写明缺什么 —— 不猜、不补、不做默认满分。
- **只加不改**：产出 `置信度` 字段；**不参与**任何判级/等级/结论的生成（接线不影响现有裁决）。
- 五轴取数（能取到什么就取什么）：
  · A 内部证据强度：handoff 里若有逐台"闸/门"通过数则用之；否则 `[INS]`。
  · B 独立源共时确认：融合表行的 `温度判`（非 —）与 `油新鲜=True` 计为**同窗**独立源；两者都无 ⇒ `[INS]`。
  · C 实物锚：handoff 行里若有 `tier/anchor/实物` 字段则映射；否则 `[INS]`。
  · D 样本充分性：融合表 `窗区间` 的窗数 + handoff 行里的记录数（若有）；无窗数 ⇒ 按 1 窗计并标注。
  · E 反证搜索：**当前无产物承载** ⇒ 一律 `[INS]`（这正是"没做过反证搜索 = 0 分"的口径）。
"""
from __future__ import annotations

import re

from . import vib_confidence as VC

WINDOW_CLAUSE_DEN = lambda n: max(4, -(-2 * int(n) // 3)) if n and n > 0 else 4   # ceil(2N/3) 起步 4



# ── ★② 真输入装载（2026-10-06）：A 轴 ← L6 过闸谱线表；C 轴 ← 实物锚登记 ───────────────
# 说明：这两样都是**仓内既有产物**（非新增数据源）：
#   A: outputs/rudong/m5_cms_tcm/model_run_l6.parquet（逐台逐线"过闸"结果, 列含 _闸/_依据/定级/绝对锚/_窗）
#      口径映射（保守, 已在 notes 写明）：某台有 ≥1 条 PASS 线 ⇒ 视为"该台走通了全闸"⇒ A 轴按 5 闸计；
#      绝对锚：该台任一行"解封判据"命中 physical_in_window 或"绝对锚"列非「—」⇒ anchor_ok=True。
#   C: reference/rudong/positive_anchors.json 的 anchors（tier ∈ physical_in_window/closed_loop/physical_historical）
#      同一台取**最高档**（20 > 12 > 8）。


def load_axis_inputs(root=None) -> dict:
    """返回 {turbine: {"a":{"gates_passed":int,"anchor_ok":bool,"pass_lines":int,"level":str},
                     "c":{"tier":str,"evidence":str}}}（容错: 缺件则空 dict）。"""
    import json
    import pathlib as _p
    base = _p.Path(root) if root else _p.Path(__file__).resolve().parents[3]
    out: dict = {}
    # A: L6 过闸谱线表
    try:
        import pandas as _pd
        pq = base / "outputs/rudong/m5_cms_tcm/model_run_l6.parquet"
        if pq.exists():
            df = _pd.read_parquet(pq)
            for tid, g in df.groupby("台"):
                key = _norm_tid(tid)
                passes = int((g["_闸"].astype(str).str.upper() == "PASS").sum())
                anchor = bool(g["解封判据"].astype(str).str.contains("physical_in_window").any()
                              or (g.get("绝对锚") is not None
                                  and g["绝对锚"].astype(str).ne("—").any()))
                lv = "候选" if (g["定级"].astype(str) == "候选").any() else "参考"
                out.setdefault(key, {})["a"] = {"gates_passed": 5 if passes else 0,
                                                "anchor_ok": anchor, "pass_lines": passes, "level": lv}
    except Exception:
        pass
    # C: 实物锚登记（同台取最高档）
    try:
        ap = base / "reference/rudong/positive_anchors.json"
        if ap.exists():
            doc = json.loads(ap.read_text(encoding="utf-8"))
            rank = {"physical_in_window": 3, "closed_loop": 2, "physical_historical": 1}
            for a in (doc.get("anchors") or []):
                key = _norm_tid(a.get("turbine"))
                tier = str(a.get("tier") or "")
                cur = (out.setdefault(key, {}).get("c") or {}).get("tier")
                if rank.get(tier, 0) > rank.get(cur or "", 0):
                    out[key]["c"] = {"tier": tier, "evidence": str(a.get("evidence") or "")[:80],
                                     "component_class": str(a.get("component_class") or "")}
    except Exception:
        pass
    # ★common_mode（2026-10-06, 5 窗数据到位后实现）：同期"全场有几台同线通过" ⇒ 该线是否机群共有现象
    try:
        import pandas as _pd
        _pq = base / "outputs/rudong/m5_cms_tcm/model_run_l6.parquet"
        if _pq.exists():
            _d = _pd.read_parquet(_pq)
            if "线" in _d.columns and "台" in _d.columns:
                _g = _d.groupby("线")["台"].nunique()
                _n_turb = int(_d["台"].nunique()) or 1
                # 判据：同线在 >=3 台通过（或覆盖 >=30% 有结论的台）⇒ 视为机群共有现象（共模嫌疑）
                _shared = {str(k): int(v) for k, v in _g.items()
                           if int(v) >= 3 or (int(v) / _n_turb) >= 0.30}
                out["__common_mode__"] = {"ran": True, "n_turbines": _n_turb, "shared_lines": _shared}
    except Exception:
        pass

    # E: 过闸 funnel ⇒ 已跑的"反证类别"（窗口级，非逐台；据实映射，未跑不补）
    try:
        import json as _json
        _sp = base / "outputs/rudong/m5_cms_tcm/model_run_summary.json"
        if _sp.exists():
            _doc = _json.loads(_sp.read_text(encoding="utf-8"))
            _map = {"G4_INHERENT_OR_BATCH": "intrinsic_line",
                    "G8_NOT_FLEET_OUTLIER": "fleet_base_rate",
                    "G6_SELECTIVITY": "alternative_explanation",
                    "G7_PEAK_OFFSET": "alternative_explanation",
                    "G5_INTEGER_ORDER": "alternative_explanation",
                    "G2_NO_LINE": "alternative_explanation"}
            _ran: set = set()
            for _w, _g in (_doc.get("by_gate") or {}).items():
                for _gate, _n in (_g or {}).items():
                    if _gate in _map and int(_n or 0) > 0:
                        _ran.add(_map[_gate])
            out["__checks__"] = sorted(_ran)
    except Exception:
        pass
    return out


def _n_win(frow: dict) -> int:
    """从融合表行的 窗区间 解析窗数（容错；解析不出按 1 窗）。"""
    wr = str(frow.get('窗区间') or '')
    try:
        return max(1, wr.count('[') // 2)
    except Exception:
        return 1


def _pt_map(handoff: dict) -> dict:
    """handoff['per_turbine'] 是列表 ⇒ 归一成 {turbine: row}（容错）。"""
    out = {}
    pt = (handoff or {}).get('per_turbine') or []
    if isinstance(pt, dict):
        return dict(pt)
    for row in pt:
        if not isinstance(row, dict):
            continue
        tid = row.get('turbine') or row.get('id') or row.get('台') or row.get('WTG')
        if tid:
            out[str(tid)] = row
    return out


def _norm_tid(x) -> str:
    s = str(x or '').strip()
    m = re.search(r'(\d+)', s)
    if not m:
        return s
    return 'WTG%02d' % int(m.group(1))


def build_axes(frow: dict, prow: dict | None, axin: dict | None = None) -> dict:
    """按可用数据装配五轴；缺数轴标 [INS] 并计 0。axin = load_axis_inputs() 的逐台真输入。"""
    prow = prow or {}
    axin = axin or {}
    notes: list[str] = []
    ins: list[str] = []

    # ---- A 内部证据强度：优先用 L6 过闸谱线的真输入 ----
    _a = (axin.get('a') or {})
    if _a:
        A = VC.score_internal(int(_a.get('gates_passed') or 0), n_gates=5,
                              anchor_ok=bool(_a.get('anchor_ok')),
                              persist_windows=int(prow.get('persist_windows') or 0),
                              n_windows=int(prow.get('n_windows') or _n_win(frow)))
        notes.append('A 来源=L6 过闸谱线: PASS 线 %s 条 · 绝对锚 %s · 定级 %s'
                     % (_a.get('pass_lines'), _a.get('anchor_ok'), _a.get('level')))
        # 直接跳到 B（跳过下面的 handoff 取数）
        a = None if False else 'FROM_L6'
    else:
        a = None
    for k in ('gates_passed', '闸过', '闸', 'passed_gates', 'n_gates_passed'):
        if k in prow and isinstance(prow[k], (int, float)):
            a = float(prow[k])
            break
    if a is None:
        ins.append('A:本场无 L6 过闸记录且 handoff 无逐台闸过数')
        A = 0.0
    elif a != 'FROM_L6':
        A = VC.score_internal(int(a), n_gates=int(prow.get('n_gates') or 5),
                              anchor_ok=bool(prow.get('anchor_ok') or False),
                              persist_windows=int(prow.get('persist_windows') or 0),
                              n_windows=int(prow.get('n_windows') or _n_win(frow)))

    # ---- B 独立源共时确认（同窗）----
    srcs = set()
    tjudge = str(frow.get('温度判') or '').strip()
    if tjudge and tjudge not in ('—', '-', ''):
        srcs.add('temperature')
    if str(frow.get('油新鲜')).lower() == 'true':
        srcs.add('oil')
    B = VC.score_independent(srcs) if srcs else 0.0
    if not srcs:
        ins.append('B(结构性):同窗独立源不在窗(温度轴判「—」/油样 stale) —— 字段级现实, 非逐台缺陷')

    # ---- C 实物锚 ----
    C = 0.0
    tier = None
    for k in ('tier', 'anchor_tier', '实物锚', 'physical_tier'):
        if prow.get(k):
            tier = str(prow[k])
            break
    if not tier:
        _c = (axin.get('c') or {})
        if _c.get('tier'):
            tier = str(_c['tier'])
            notes.append('C 来源=实物锚登记: %s（%s）' % (tier, _c.get('evidence', '')[:40]))
    if tier:
        C = VC.score_physical(tier)
    else:
        ins.append('C:本台无实物锚登记且 handoff 无该字段')

    # ---- D 样本充分性 ----
    nw = None
    wr = str(frow.get('窗区间') or '')
    if wr:
        nw = max(1, wr.count('[') // 2 or wr.count(',')) if wr else 1
    nrec = 0
    for k in ('n_records', '记录数', 'n'):
        if isinstance(prow.get(k), (int, float)):
            nrec = int(prow[k])
            break
    D = VC.score_sample(nrec, (nw or 1), False, n_windows=(nw or 1))
    if nw is None:
        ins.append('D:无窗区间解析结果')

    # ---- E 反证搜索：用过闸 funnel 的"排除类闸"据实计（未跑的不补）----
    # common_mode 未跑原因（2026-10-06 实测）：L6 过闸 funnel 无共模闸；G4_INHERENT_OR_BATCH 是
    # 「固有线/批次」（fleet 共有谱峰），与「共模」（fleet 同期共同变化）不同义；且共模需 ≥2 窗时间序列。
    _checks = list(axin.get('__checks__') or [])
    _cm = (axin.get('__common_mode__') or {})
    _flag = None
    if _cm.get('ran'):
        _line = str(frow.get('线') or '')
        _shared = _cm.get('shared_lines') or {}
        if _line and _line in _shared:
            _flag = '本线同期在 %d 台通过 ⇒ **机群共有现象（共模嫌疑）**，个体性被削弱' % _shared[_line]
        _checks.append('common_mode')
    if _checks:
        E = VC.score_falsification(tuple(_checks))
        if _flag:
            E = max(0.0, E - 20.0 / max(len(_checks), 1))   # 该类被推翻 ⇒ 扣掉该类得分
        _miss = sorted(set(VC.__dict__.get('CHECKS_TOTAL', () ) or
                           ('fleet_base_rate', 'common_mode', 'intrinsic_line', 'alternative_explanation')) - set(_checks))
        notes.append('E 来源=过闸 funnel%s: 已跑 %s；未跑 %s'
                     % (' + common_mode(5 窗同期跨台)' if _cm.get('ran') else '', '/'.join(_checks), '/'.join(_miss) or '无'))
        if _flag:
            ins.append('E(共模): ' + _flag)
    else:
        E = 0.0
        ins.append('E:无过闸 funnel 产物(没人做过=0 分)')

    axes = {'A_internal': round(A, 1), 'B_independent': round(B, 1),
            'C_physical': round(C, 1), 'D_sample': round(D, 1), 'E_falsification': round(E, 1)}
    return {'axes': axes, 'ins': ins, 'notes': notes}


def score_turbine(frow: dict, prow: dict | None, axin: dict | None = None) -> dict:
    r = build_axes(frow, prow, axin)
    conf = VC.confidence(r['axes'])
    total = conf.get('total') if isinstance(conf, dict) else conf
    band = (conf or {}).get('band') if isinstance(conf, dict) else None
    # ★诊断门槛：A/B/C/E 只要能取到的轴缺件, 分数就**不可用于判断机组状态**
    #   （低分反映的是"输入缺件"而不是"证据弱" —— 防止把缺数据读成"证据不足"）
    hard = ('A:', 'C:', 'E:')   # B 属结构性缺件, 单列
    missing = [s for s in r['ins'] if s.startswith(hard)]
    structural = [s for s in r['ins'] if s.startswith('B(')]
    usable = (len(missing) == 0)
    caveat = ('' if usable else
              '本分**不可用于判断机组状态**：输入端缺件 ' + str(len(missing)) + ' 项（' + '；'.join(missing) + '）。'
              '低分反映"缺少输入"，不等于"证据弱"；补件后重算才有意义。')
    return {'axes': r['axes'], 'total': total, 'band': band,
            'usable': usable, 'missing_inputs': missing, 'structural': structural,
            'caveat': caveat, 'ins': r['ins']}


def attach_confidence(ftab, handoff: dict | None = None):
    """给融合表加一列 置信度（dict），不改任何既有列。

    """
    pm = _pt_map(handoff or {})
    axin = load_axis_inputs()
    out = []
    for _, row in ftab.iterrows():
        d = row.to_dict()
        tid = _norm_tid(d.get('turbine'))
        _ax = dict(axin.get(tid) or {})
        _ax['__checks__'] = axin.get('__checks__') or []   # ★E 轴（窗口级）随行传入
        _c = (pm.get(tid) or pm.get(str(d.get('turbine'))) or {})
        s = score_turbine(d, _c, _ax)
        d['置信度'] = s
        out.append(d)
    import pandas as pd
    return pd.DataFrame(out)