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

# -*- coding: utf-8 -*-
"""learn_loop.py — 柱1 P2「判据 n+1 离线学习回路」(RULE-3: 只提案不改阈值, 人裁+版本化+重部署).

飞轮右半的学习件: 修复反馈(故障记录当代理) → 给 agent 候选贴 TP/FP → 每判据代理精度 →
**阈值调整提案**(禁自动应用)。补 backfill 的上游 — backfill 需 CLOSED/REFUTED 结局才出提案,
本模块从故障记录**代理标注**产生这些结局(无真修复台账时的 bootstrap)。

★纪律 (验证程序五次实证 + RULE-3):
- **故障记录=代理非真修复验证**: 故障发生≈候选materialized(proxy-TP), 无故障≈proxy-FP; 弱于真"修完回验"。
- **口径对齐**: 只用与判据同域的故障 (温度判据 vs 主传动链热故障, 剔叶片/电气, 复用 validate_pair.is_thermal)。
- **min-evidence 闸 + RULE-3**: 证据不足/精度不明确 → **不提改阈值**(防过拟合弱反馈); 提案全【参考】人裁。
- **欠功效诚实**: 单场/n小 → 提案标"suggestive 非定论"; 跨场需 n≥3 (同 Tier-1 门槛)。

纯核心 (label_precision / threshold_proposals) 无 I/O; per-farm 加载复用 validate_pair。
"""
import numpy as np


def label_precision(red_turbines, thermal_counts, min_fault=1):
    """代理精度: 候选红台里 有热故障(proxy-TP) vs 无(proxy-FP)。纯。
    red_turbines: set[tid] agent温度红台; thermal_counts: {tid: 严格热故障数} 全台。
    返回 {n_candidates, n_proxy_tp, n_proxy_fp, precision_proxy, base_rate}。"""
    red = {str(t) for t in red_turbines}
    tp = sum(1 for t in red if thermal_counts.get(t, 0) >= min_fault)
    fp = len(red) - tp
    prec = round(tp / len(red), 3) if red else None
    # base rate: 全台有热故障占比 (判精度是否只是撞base rate)
    allt = list(thermal_counts)
    br = round(sum(1 for t in allt if thermal_counts.get(t, 0) >= min_fault) / len(allt), 3) if allt else None
    return {'n_candidates': len(red), 'n_proxy_tp': tp, 'n_proxy_fp': fp,
            'precision_proxy': prec, 'base_rate': br}


def threshold_proposals(judge_stats, precision_floor=0.5, min_candidates=8, lift_min=0.15):
    """判据统计 → 阈值调整提案 (RULE-3: 只【参考】提案, 人裁+版本化+重部署)。纯。
    judge_stats: {judge_id: {precision_proxy, n_candidates, base_rate, farms}}。
    提案闸: 候选≥min_candidates ∧ 精度<floor ∧ 精度未显著超base_rate(lift<lift_min) → 提"该判据过松, 收紧阈值"。
    反之精度好也不提(别乱动稳的判据)。证据不足 → 不提(诚实, 防过拟合)。"""
    props = []
    for jid, s in sorted(judge_stats.items()):
        prec, n, br = s.get('precision_proxy'), s.get('n_candidates', 0), s.get('base_rate')
        if prec is None or n < min_candidates:
            props.append({'judge': jid, 'action': 'no_change', 'verdict': '参考',
                          'reason': f'代理证据不足(候选{n}<{min_candidates})→不提改阈值(RULE-3防过拟合弱反馈)'})
            continue
        lift = (prec - br) if (prec is not None and br is not None) else None
        if prec < precision_floor and (lift is None or lift < lift_min):
            props.append({'judge': jid, 'action': 'propose_tighten', 'verdict': '参考',
                          'reason': f'代理精度{prec}<{precision_floor} ∧ 超base_rate仅{lift}(<{lift_min}) → 提请人裁: '
                                    f'{jid} 阈值(如z_thresh)是否过松需收紧',
                          'evidence': {**s},
                          'action_required': '人裁 + thresholds.yaml版本化 + 重部署 (禁自动应用, RULE-3); '
                                             '⚠故障=代理非真修复验证, 单场/n小=suggestive'})
        else:
            props.append({'judge': jid, 'action': 'no_change', 'verdict': '参考',
                          'reason': f'代理精度{prec}(超base_rate {lift}) 未显著过松 → 不动稳判据'})
    return props


def compose_learn(farm_feedback, judge_id='nbm_residual'):
    """组合 (纯): 多场 {farm: (red, thermal_counts)} → 每判据跨场聚合 + 提案。
    本版温度判据单一(nbm_residual)聚合所有温度红台; 跨场 pooled 精度 + 各场明细。"""
    per_farm, all_red, all_counts = {}, set(), {}
    for farm, (red, counts) in farm_feedback.items():
        lp = label_precision(red, counts)
        per_farm[farm] = lp
        all_red |= {f'{farm}::{t}' for t in red}
        all_counts.update({f'{farm}::{t}': c for t, c in counts.items()})
    pooled = label_precision(all_red, all_counts)
    stats = {judge_id: {**pooled, 'farms': list(per_farm)}}
    props = threshold_proposals(stats)
    return {'judge_stats': stats, 'per_farm': per_farm, 'proposals': props,
            'n_farms': len(per_farm),
            '_rule3': 'RULE-3: 提案=提请人裁证据, 非自动应用; 阈值更新须人裁+版本化+重部署。'
                      '故障记录=代理(非真修复回验); 跨场立需n≥3(现%d场)。' % len(per_farm)}


# ---------------------------------------------------------------------------
if __name__ == '__main__':
    # CLI: python -m src.sop.learn_loop → 用 validate_pair.PAIRED_FARMS 的故障反馈跑 P2 学习回路
    import json
    from src.sop.validate_pair import PAIRED_FARMS, _load_farm
    fb = {}
    for farm, cfg in PAIRED_FARMS.items():
        try:
            red, counts = _load_farm(cfg)
            fb[farm] = (red, counts)
        except Exception as e:
            print(f"  {farm}: 加载失败 {str(e)[:40]}")
    out = compose_learn(fb)
    print("=== 柱1 P2 判据 n+1 学习回路 (RULE-3 只提案) ===")
    for farm, lp in out['per_farm'].items():
        print(f"  {farm}: 候选{lp['n_candidates']}台 proxy-TP{lp['n_proxy_tp']}/FP{lp['n_proxy_fp']} "
              f"精度{lp['precision_proxy']} (base_rate{lp['base_rate']})")
    js = out['judge_stats']['nbm_residual']
    print(f"\n  pooled nbm_residual: 精度{js['precision_proxy']} vs base_rate{js['base_rate']} "
          f"(候选{js['n_candidates']}, {out['n_farms']}场)")
    print("  提案:")
    for p in out['proposals']:
        print(f"    [{p['action']}] {p['reason']}")
    print(f"\n  {out['_rule3']}")