guanlan-src/app_ETL/app_ETL_guanlan/builders/baseline_38_build.py

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
r"""三层基线生成端 —— 补上"包内没有生成端"的 `m5_cms_tcm/baseline_38.json` (用户令 2026-09-19)。

## 为什么要写它

用户令：「**所有的计算 均要形成 观澜的源代码**，确保 观澜 在别的电脑安装后，系统运行正常。」
`baseline_38.json` 是**随包快照**（族表 kind=shipped、gen=None，见
`scripts/products_reverse_audit.py::HUMAN_SNAPSHOTS`），旧件的 self/absolute 两层由人工坐实。
它是 `src/windcms/report.py::baseline_card_html()` 的**唯一取数处** —— 件不在位, 报告页第 5 张卡
（三层基线对照）直接返回空串。本脚本把它**从 CMS 窗标量 + 场内实物锚算出来**。

## 产物与消费者 (字段契约)

    outputs/<场>/m5_cms_tcm/baseline_38.json
      meta             built / min_windows / windows / segments_kind / self_stat / layers / caveat /
                       usage_limit / anchor_guard_days / ★数据边界
      absolute_limits  iso_rms_vel, Rms_Vel → {unit:'mm/s', yellow:7.1, red:11.0, source:ISO 20816-3 ...}
                       ★ 标准常数, 不是测量值, 故写死为模块常量 (改进 ISO 不靠重算)
      fleet_baseline   {'<sensor>|<meas>': {段: 全场中位}}   ← 页面查 fleet_baseline[f'{sensor}|{meas}'][latest_window]
      self_baseline    {台: {测点: {标量: 条目}}}

    条目字段 (report.py 逐个读): self_baseline, normal_band[2], latest, latest_window, x_self,
      threshold_self, in_normal_band, n_windows_healthy, n_windows_total, unit, insufficient
      另存 (消费者暂不读, 供人核对): self_baseline_stat, rel_mad, threshold_note, healthy_windows,
      excluded (剔掉的段名) / excluded_detail (剔的理由), recovered_after_anchor, trend_flag, spread, latest_n

    ★ 自基线样本不足 ⇒ `self_baseline: null` + `insufficient: '健康期窗 N < 3, 不给数 (禁用 fleet 值顶替)'`
      —— **绝不**用 fleet 值顶替 (report.py 专门把这条不足列出来, 就是防这个)。

## 单位口径 (与随包旧件一致, ★容易踩)

页面按 `absolute_limits[meas]['unit'] == 'mm/s'` **乘 1000** 显示。CMS 窗索引 (`y_unit` 列) 实测:
页面用的 5 个标量里 `Peak/Rms_HP/rms_200` = `m/s²`、`Rms_Vel/iso_rms_vel` = `m/s` (Kurtosis 页面不展示, y_unit=`1`)
⇒ 本器**存原值** (速度存 m/s), 单位标签按 y_unit 分给: `m/s²` / `mm/s`。
即 0.00282 m/s 存进文件、页面显示 2.82 mm/s。

## 分段 (segment) 口径 —— 与 `scripts/component_history_build.py` **同一套** (两件必须同段)

    len(windows) >= 3   segments_kind='window'  段 = 导出窗 (参考口径: 健康期窗中位, ±锚 120 天护栏)
    len(windows) <  3   segments_kind='week'    段 = 窗内 ISO 周 (本机实况: 只有 w0316 一窗)

★ 本机 (2026-09-19 实测) 只有 1 个窗 `w0316` (2026-03-16 17:27 → 2026-04-21 10:18) ⇒ 走周段 (6 段)。
  **周段散布 ≠ 多窗散布**: 周与周之间工况/负荷更接近 ⇒ 散布更小 ⇒ `threshold_self` 更容易被现实超出。
  这一条写进 `meta.usage_limit`, 不假装等于多窗口径。多窗分支本机**未取数验证**。

## 健康期与锚护栏

健康期段 = 全部段 **减去** 实物锚 ±`ANCHOR_GUARD_DAYS`(=120) 天覆盖的段。
锚来自 `reference/rudong/positive_anchors.json` (场内实物证据: 内窥镜/金相/拆检/闭环), 按
`component_class` 映射到测点 (`gearbox_hs_bearing`→两个高速轴测点, `generator_bearing`→发电机两测点,
`main_bearing`→主轴承两测点)。★ 实测有 4 条锚**没有日期** (WTG03/WTG08/WTG28 主轴, WTG05/WTG32 发电机)
⇒ 无法做时间护栏, 逐条记进 `meta.锚.无日期` 而不是默认忽略。
实测后果: WTG09 (锚 2026-01-23) / WTG27 (2026-01-27) / WTG24 (2025-12-15) 三台的对应测点健康段不足 3
⇒ 出 `insufficient`, 不给数; WTG02 (锚 2025-12-04) 只余 3 个健康段 ⇒ 出数但 `excluded` 里写明剔了哪 3 段。

## 用法

    python scripts/baseline_38_build.py                 # 算并写盘 + 自登记
    python scripts/baseline_38_build.py --dry-run       # 只算不写, 打印漏斗与不足清单
    python scripts/baseline_38_build.py --farm rudong
"""
from __future__ import annotations

try:
    from app_common.app_common_guanlan.api import install_root as _install_root
except ImportError:                     # 理论不可达；包结构异常时回退到按位置上跳
    from pathlib import Path as _P
    def _install_root(_f): return _P(_f).resolve().parents[3]

import argparse
import json
import pathlib
import sys
import time

ROOT = _install_root(__file__)      # 模块化后按标记找安装根（原 parents[1] 已不成立）
sys.path.insert(0, str(ROOT))

# ★GBK 控制台/日志重定向不再因 '²' '⇒' 这类字符抛 UnicodeEncodeError (2026-09-19 远端实逮:
#   `baseline_38_build.py` 打印 'm/s²' 时在 cp936 下 rc=1, 整条恢复链少一件产物)。
#   只把**不可编码字符替换掉**, 不改流编码 —— 中文照旧可读。
try:
    sys.stdout.reconfigure(errors='replace')
except Exception:
    pass

import numpy as np                                            # noqa: E402
import pandas as pd                                           # noqa: E402
import pyarrow.parquet as pq                                  # noqa: E402

from src import paths as P                                    # noqa: E402
from src.derived_manifest import record as dm_record           # noqa: E402
from src.windscada.config import farm as ws_farm               # noqa: E402
from src.windcms import data as CD                             # noqa: E402
from src.windcms.config import SENSORS, SENSOR_CN              # noqa: E402
from src.windcms.config import farm as cms_farm                # noqa: E402

# ── 页面要的标量集: 与 src/windcms/report.py::_BL_MEAS **逐字一致** (顺序即报告卡里的行序) ──
# 有绝对标准的速度量在前 (iso_rms_vel/Rms_Vel), 冲击量次之; Kurtosis/CrestFactor 判别力弱, 页面不展示。
MEAS_BL = ['iso_rms_vel', 'Rms_Vel', 'Peak', 'Rms_HP', 'rms_200']

# ── 绝对限值: ISO 20816-3 标准常数 (不是测量值 ⇒ 写死; 改进标准号才需要动这里) ──
ISO_YELLOW, ISO_RED = 7.1, 11.0
ISO_SRC = 'ISO 20816-3 (机组 >300kW 刚性安装) 区界 B/C 与 C/D'
ABSOLUTE_LIMITS = {
    'iso_rms_vel': dict(unit='mm/s', yellow=ISO_YELLOW, red=ISO_RED, source=ISO_SRC),
    'Rms_Vel': dict(unit='mm/s', yellow=ISO_YELLOW, red=ISO_RED,
                    source='同 iso_rms_vel (同为速度 RMS; 频带口径以厂商配置为准)'),
}
# y_unit → 存值单位标签 (见 docstring「单位口径」); 速度类标 mm/s (页面 ×1000 显示), 与旧件同约定
UNIT_OF_YUNIT = {'m/s': 'mm/s', 'm/s²': 'm/s²', 'm/s2': 'm/s²', '1': '-', '': '-'}

MIN_WINDOWS = 3                    # 健康期段数门 (旧件 min_windows=3)
ANCHOR_GUARD_DAYS = 120            # 实物锚 ±120 天不进健康期 (旧件 anchor_guard_days=120)
FLEET_MIN_TURBINES = 5             # fleet 中位至少要有几台贡献 (不足则该段不出 fleet 值)
RECOVER_FRAC = 0.5                 # recovered_after_anchor: 锚后中位 ≤ 锚前中位 × 0.5 判"已恢复"
CONTAMINATED_FLAG = '自基线可能已污染 (健康期窗单调上升, 基线中心被抬高)'   # 旧件同措辞
THRESHOLD_NOTE = ('max(center×max(1+2.0×relMAD, 1.3), 健康期最大×1.05) — 由该台自身散布定非拍脑袋倍数, '
                  '且保证历史正常波动不自触发')
INS_TMPL = '健康期窗 {n} < 3, 不给数 (禁用 fleet 值顶替)'
WANT_COLS = ['turbine', 'sensor_name', 'meas_name', 'trigger_time', 'scalar_value', 'ds_size', 'y_unit']

# 锚的部件类 → 测点 (只映射该部件类**自己的**测点, 不扩大到整机)
ANCHOR_CLASS_SENSORS = {
    'gearbox_hs_bearing': ('Gear_HS_rotor_side', 'Gear_HS_generator_side'),
    'gearbox_planet_bearing': ('Gear_planet',),
    'gearbox_ims_bearing': ('Gear_IMS',),
    'generator_bearing': ('Generator_DE', 'Generator_NDE'),
    'main_bearing': ('Main_bearing_front', 'Main_bearing_rear'),
}


def _r(v, nd=6):
    if v is None:
        return None
    v = float(v)
    return float(f'{v:.{nd}g}') if np.isfinite(v) else None


def _round(v, nd):
    if v is None:
        return None
    v = float(v)
    return round(v, nd) if np.isfinite(v) else None


def _py(o):
    """numpy 标量 → python; NaN/Inf → None (JSON 严格可解析)。"""
    if isinstance(o, dict):
        return {k: _py(v) for k, v in o.items()}
    if isinstance(o, (list, tuple)):
        return [_py(v) for v in o]
    if isinstance(o, (np.integer,)):
        return int(o)
    if isinstance(o, (np.floating, float)):
        f = float(o)
        return f if np.isfinite(f) else None
    if isinstance(o, (np.bool_,)):
        return bool(o)
    return o


def load_scalars(cfg, meas):
    """各 CMS 窗的标量子集 (口径同 `src.windcms/data.py`, 但不落缓存); 附带 y_unit 用于定单位标签。"""
    ws = CD.windows(cfg)
    spans = CD.window_spans(cfg)
    parts, used = [], {}
    for w, info in ws.items():
        p = pathlib.Path(info['index'])
        names = set(pq.ParquetFile(p).schema_arrow.names)
        cols = [c for c in WANT_COLS if c in names]
        d = pd.read_parquet(p, columns=cols)
        for c in ('ds_size', 'scalar_value'):
            if c not in d.columns:
                d[c] = np.nan
        if 'y_unit' not in d.columns:
            d['y_unit'] = ''
        d = d[(d.ds_size == 1) & d.meas_name.isin(meas) & d.sensor_name.isin(SENSORS)].copy()
        d['window'] = w
        parts.append(d)
        sp = spans.get(w) or {}
        used[w] = dict(index=P.rel(p), rows_scalar=int(len(d)), rows_all=int(sp.get('rows') or 0),
                       span_min=str(sp.get('t_min') or '')[:10], span_max=str(sp.get('t_max') or '')[:10],
                       turbines=int(d['turbine'].nunique()))
    if not parts:
        return pd.DataFrame(columns=WANT_COLS + ['window', 't', 'val']), used
    d = pd.concat(parts, ignore_index=True)
    d['t'] = pd.to_datetime(d['trigger_time'], errors='coerce')
    d['val'] = pd.to_numeric(d['scalar_value'], errors='coerce')
    d = d[d['val'].notna() & d['t'].notna()].copy()
    return d, used


def mark_segments(d, n_win, span_min=None):
    """加 `seg` 列 → (kind, {段: 段日起日}, 段序, {段: 段结束日})。口径见 docstring。"""
    span_min = span_min or {}
    if n_win >= 3:
        d['seg'] = d['window'].astype(str)
        kind = 'window'
    else:
        iso = d['t'].dt.isocalendar()
        d['seg'] = iso['year'].astype(int).astype(str) + '-W' + iso['week'].astype(int).astype(str).str.zfill(2)
        kind = 'week'
    g = d.groupby('seg')['t'].min()
    order = [s for s, _ in sorted(g.items(), key=lambda kv: kv[1])]
    if kind == 'week':
        seg_date = {s: str(pd.Timestamp.fromisocalendar(int(s[:4]), int(s[6:]), 1).date()) for s in order}
        seg_end = {s: str((pd.Timestamp.fromisocalendar(int(s[:4]), int(s[6:]), 1) + pd.Timedelta(days=6)).date())
                   for s in order}
    else:
        seg_date = {s: str(span_min.get(s) or str(g[s])[:10])[:10] for s in order}
        seg_end = {s: str(d[d.seg == s]['t'].max())[:10] for s in order}
    return kind, seg_date, order, seg_end


def load_anchors(cfg):
    """场内实物锚 → (逐条锚表, 说明)。缺件/无日期都如实记, 不默认忽略。"""
    p = cfg.get('anchors')
    if not p or not pathlib.Path(p).exists():
        return pd.DataFrame(columns=['turbine', 'date', 'sensors']), dict(path=P.rel(p) if p else None, exists=False,
                                                                         note='锚件不在位 ⇒ 无护栏 (健康期=全部段)')
    a = json.loads(pathlib.Path(p).read_text(encoding='utf-8'))
    recs, nodate = [], []
    for x in a.get('anchors', []):
        cls = str(x.get('component_class') or '')
        sens = ANCHOR_CLASS_SENSORS.get(cls)
        if sens is None:                                     # 未知类: 不猜, 记下来
            nodate.append(dict(turbine=x.get('turbine'), component_class=cls, why='部件类未登记 → 未映射测点'))
            continue
        dt_ = x.get('date')
        if not dt_:
            nodate.append(dict(turbine=x.get('turbine'), component_class=cls, why='锚无日期 → 无法做时间护栏'))
            continue
        recs.append(dict(turbine=x.get('turbine'), date=str(dt_)[:10], sensors=list(sens),
                         component_class=cls, tier=x.get('tier')))
    info = dict(path=P.rel(p), exists=True, n_anchors=len(a.get('anchors', [])), guard_days=ANCHOR_GUARD_DAYS,
                n_used=len(recs), 无日期=len(nodate), tiers=sorted({str(x.get('tier')) for x in a.get('anchors', [])}),
                未用锚=nodate)
    return pd.DataFrame(recs), info


def healthy_segments(order, seg_date, seg_end, anchors, turbine, sensor):
    """该 (台, 测点) 的健康期段 + 被剔段 (锚日 ±guard_days 覆盖到的段)。"""
    ex = []
    ad = anchors[(anchors['turbine'] == turbine) & (anchors['sensors'].map(lambda L: sensor in L))] \
        if len(anchors) else anchors
    for _, a in (ad.iterrows() if len(ad) else []):
        t = pd.Timestamp(a['date'])
        lo = str((t - pd.Timedelta(days=ANCHOR_GUARD_DAYS)).date())
        hi = str((t + pd.Timedelta(days=ANCHOR_GUARD_DAYS)).date())
        for s in order:                                      # 段与 [lo,hi] 有交叠即剔 (段是闭区间)
            if not (seg_end[s] < lo or seg_date[s] > hi):
                ex.append((s, f'{a["date"]}±{ANCHOR_GUARD_DAYS}d({a["component_class"]})'))
    ex_map = dict(ex)
    keep = [s for s in order if s not in ex_map]
    return keep, ex_map


def build(cfg, farm_name='rudong', dry=False):
    m5 = P.m5(farm_name)
    store = P.store(farm_name)
    t0 = time.time()
    d, used = load_scalars(cfg, MEAS_BL)
    if not len(d):
        print('[X] 一个 CMS 窗索引都没读到 —— 先放数据并跑重算链 (scripts/rebuild_all.py); 本器不写空产物。')
        return None, 4
    kind, seg_date, order, seg_end = mark_segments(d, len(used), {w: v['span_min'] for w, v in used.items()})
    # 单位标签: 按每个 (测点, 标量) 在窗索引里的 y_unit 定 (见 docstring「单位口径」)
    yunit = d.groupby(['sensor_name', 'meas_name'])['y_unit'].agg(lambda s: s.dropna().astype(str).mode().iloc[0]
                                                                 if s.notna().any() else '')
    sg = d.groupby(['turbine', 'sensor_name', 'meas_name', 'seg'])['val'].agg(['median', 'size']).reset_index()
    sg = sg.rename(columns={'median': 'med', 'size': 'n'})
    anchors, anchor_info = load_anchors(cfg)
    turbines = [t for t in ws_farm(farm_name).get('turbines', [])] or sorted(d['turbine'].unique())
    # 段跨度用**数据实际覆盖**区间 (周段按日历算出的段末日会到 04-26, 而数据止 04-21 ——
    # 前者只用于"工单落在哪一段"的判定, 报告里说跨度必须说数据真到的日子)。
    span = [min(v['span_min'] for v in used.values()), max(v['span_max'] for v in used.values())]
    # ── fleet: 每个 (测点, 标量) 每段的**全场中位** (先全算一遍; 早期版本按台循环里数段行数,
    #    每段至多 1 行 ⇒ 计数永远 <5 ⇒ fleet 键全空, 页面 ×fleet 恒为 —) ──
    fleet = {}
    for (sen, ms), sub in sg.groupby(['sensor_name', 'meas_name']):
        for s in order:
            v = sub[sub.seg == s]['med'].to_numpy(dtype=float)
            if len(v) >= FLEET_MIN_TURBINES:
                fleet.setdefault(f'{sen}|{ms}', {})[s] = _r(float(np.median(v)))
    print(f'三层基线生成端 · 场={farm_name}')
    print(f'  窗 {len(used)} 个: ' + '; '.join(
        f'{w}({v["span_min"]}→{v["span_max"]}, 全表 {v["rows_all"]}行; 标量 {v["rows_scalar"]}行)'
        for w, v in used.items()))
    print(f'  标量 (ds_size==1 ∧ {len(MEAS_BL)} 标量 ∧ {len(SENSORS)} 测点): {len(d)} 行 · 台 {d.turbine.nunique()}')
    print(f'  分段: segments_kind={kind} · {len(order)} 段 {order[0]}…{order[-1]} ({span[0]}→{span[1]})')
    print(f'  单位 (y_unit→标签): ' + '; '.join(
        f'{s}|{m}:{yunit.get((s, m), "") or "(该测点无此标量)"}→{UNIT_OF_YUNIT.get(str(yunit.get((s, m), "")), "-")}'
        for s, m in [('Main_bearing_front', 'Peak'), ('Main_bearing_front', 'Rms_Vel'),
                     ('Main_bearing_front', 'rms_200'), ('Generator_DE', 'iso_rms_vel'),
                     ('Gear_planet', 'iso_rms_vel')]))
    print(f'  锚: {anchor_info.get("path")} 共 {anchor_info.get("n_anchors", 0)} 条 · 带日期可用 {anchor_info.get("n_used", 0)} '
          f'· 护栏 ±{ANCHOR_GUARD_DAYS}d · 未用 {anchor_info.get("无日期", 0)} 条')
    for x in anchor_info.get('未用锚', []):
        print(f'    [锚未用] {x.get("turbine")} {x.get("component_class")} — {x.get("why")}')

    self_bl, ins_rows = {}, []
    n_entries = n_ok = 0
    for t in turbines:
        self_bl[t] = {}
        for sen in SENSORS:
            self_bl[t][sen] = {}
            for ms in MEAS_BL:
                sub = sg[(sg.turbine == t) & (sg.sensor_name == sen) & (sg.meas_name == ms)]
                if not len(sub):
                    continue                                  # 该测点无此标量 (如 齿轮箱无 iso_rms_vel): 整条不出现
                n_entries += 1
                ser = sub.set_index('seg')['med'].reindex(order).dropna()
                nser = sub.set_index('seg')['n'].reindex(order).fillna(0)
                healthy, ex_map = healthy_segments(order, seg_date, seg_end, anchors, t, sen)
                hv = [s for s in healthy if s in ser.index]
                unit = UNIT_OF_YUNIT.get(str(yunit.get((sen, ms), '')), '-')
                base = dict(n_windows_total=int(len(ser)), n_windows_healthy=int(len(hv)),
                            excluded=[s for s in order if s in ex_map],           # 剔掉的段名 (旧件同字段名)
                            excluded_detail=[dict(seg=s, why=ex_map[s]) for s in order if s in ex_map],
                            unit=unit, healthy_windows=hv, latest_window=str(ser.index[-1]),
                            latest=_r(float(ser.iloc[-1])), latest_n=int(nser.get(ser.index[-1], 0)))
                if len(hv) < MIN_WINDOWS:
                    self_bl[t][sen][ms] = _py(dict(base, self_baseline=None, self_baseline_stat=None,
                                                   normal_band=None, rel_mad=None, threshold_self=None,
                                                   threshold_note=None, spread=None, x_self=None,
                                                   in_normal_band=None, recovered_after_anchor=None,
                                                   trend_flag=None,
                                                   insufficient=INS_TMPL.format(n=len(hv))))
                    ins_rows.append(dict(turbine=t, sensor=sen, meas=ms, n_healthy=len(hv),
                                         n_total=int(len(ser)), why='锚护栏后健康段不足 3' if ex_map else '总段数不足 3'))
                    continue
                vals = ser.loc[hv].to_numpy(dtype=float)
                center = float(np.median(vals))
                lo, hi = float(vals.min()), float(vals.max())
                mad = float(np.median(np.abs(vals - center)))
                rel_mad = (mad / center) if center else None
                thr = max(center * max(1 + 2.0 * (rel_mad or 0.0), 1.3), hi * 1.05) if center else None
                latest = float(ser.iloc[-1])
                # 锚后是否恢复: 需要该 (台,测点) 有锚且段跨度内锚前/锚后各有段 (本机多窗口径才可能成立)
                rec = None
                ad = anchors[(anchors['turbine'] == t) & (anchors['sensors'].map(lambda L: sen in L))] \
                    if len(anchors) else anchors
                for _, a in (ad.iterrows() if len(ad) else []):
                    pre = [s for s in ser.index if seg_end[s] < a['date']]
                    post = [s for s in ser.index if seg_date[s] > a['date']]
                    if pre and post:
                        pv, qv = float(np.median(ser.loc[pre])), float(np.median(ser.loc[post]))
                        rec = bool(qv <= RECOVER_FRAC * pv) if pv > 0 else None
                        break
                trend = CONTAMINATED_FLAG if (len(vals) >= 3 and np.all(np.diff(vals) > 0)) else None
                self_bl[t][sen][ms] = _py(dict(
                    base, self_baseline=_r(center), self_baseline_stat=f'中位 of 健康期{"窗" if kind == "window" else "段"}中位',
                    normal_band=[_r(lo), _r(hi)], rel_mad=_round(rel_mad, 4), threshold_self=_r(thr),
                    threshold_note=THRESHOLD_NOTE, spread=_round(hi / lo, 4) if lo > 0 else None,
                    x_self=_round(latest / center, 3) if center else None,
                    in_normal_band=bool(lo <= latest <= hi), recovered_after_anchor=rec, trend_flag=trend))
                n_ok += 1
    # (fleet 已在上面按 (测点, 标量, 段) 全场算完 —— 不在这里按台累加, 见那一处注释)

    fleet_bl = {k: dict(sorted(v.items(), key=lambda kv: order.index(kv[0]))) for k, v in sorted(fleet.items())}
    n_fleet_keys = len(fleet_bl)
    missing_fleet = sorted({f'{s}|{m}' for t in self_bl.values() for s, mm in t.items() for m in mm} - set(fleet_bl))
    seg_kind_txt = '导出窗' if kind == 'window' else f'窗内 ISO 周 (本机窗数 {len(used)} < 3 ⇒ 降级; 多窗时自动改回)'
    meta = dict(
        built=time.strftime('%Y-%m-%d'), farm=farm_name, by='scripts/baseline_38_build.py',
        min_windows=MIN_WINDOWS,
        windows={w: v['span_min'] for w, v in used.items()},
        windows_detail={w: dict(rows_all=v['rows_all'], rows_scalar=v['rows_scalar'], span_min=v['span_min'],
                                span_max=v['span_max'], turbines=v['turbines']) for w, v in used.items()},
        segments_kind=kind, segments=seg_date, n_segments=len(order),
        anchor_guard_days=ANCHOR_GUARD_DAYS,
        self_stat=(f'中位 of 健康期{"窗" if kind == "window" else "段"}中位 (+正常带[min,max]+relMAD); '
                   f'健康期 = 全部段 减去 实物锚 ±{ANCHOR_GUARD_DAYS} 天覆盖的段。'
                   f'本机段 = {seg_kind_txt}, 共 {len(order)} 段 ({span[0]}→{span[1]}); '
                   f'段中位 = 该段该 (台,测点,标量) 全部标量记录的中位 (ds_size==1, 每时间戳一条)。'),
        layers='self (变了没) / fleet (比别人高没) / abs (严重不严重) — 三者独立, 只报一个会误导',
        caveat=('自基线假设健康期段代表正常; 慢性劣化台该假设失效 → 见 recovered_after_anchor / trend_flag。'
                f'★ 本机用周段 ⇒ 正常带/relMAD 是**周与周之间**的散布 (周间工况更接近, 散布偏小), '
                f'不等于随包旧件那种"跨月导出窗"的散布; 且 `latest` 是最后一个周段的段中位, '
                f'该段样本可能很薄 (逐条给出 latest_n)。'),
        unit_note=('存值单位 = CMS 窗索引原值 (y_unit): Peak/Rms_HP/rms_200 = m/s², Rms_Vel/iso_rms_vel = m/s, '
                   'Kurtosis = 1; 速度类标签写 mm/s 而页面按 absolute_limits.unit=="mm/s" 乘 1000 显示 '
                   '—— 与随包旧件同一约定 (0.00282 m/s 存值 ↔ 2.82 mm/s 显示)。'),
        usage_limit=(
            '★ 本基线**不作自动报警闸**: 随包旧件 (多窗口径, 2026-08-26 参数扫描) 记录的 ROC 结论是 '
            '误报率压到 6.3% 时检出率仅 14.3% (6/42 锚单元且集中在 20# 一台), 加持续性要求(连续2窗)检出率归零 '
            '—— 多数锚台只有 1~2 个锚前窗。该回测**未在本机重跑** (本机只有 1 窗, 跑不出多窗 ROC), '
            '此处照录以免被当成"已验证的报警线"。'
            f'★ 本机为周段口径 ⇒ 上列阈值由**周段散布**定, 散布比多窗口径更小, 更容易被现实超出; '
            f'正确用法 = ①量化描述(报 x_self 而非"偏高"这类模糊词) ②人工研判输入 ③趋势跟踪积累段数。'
            '自动报警须等段数够做持续性检验。'),
        fleet_min_turbines=FLEET_MIN_TURBINES,
        inputs=dict(scalars=[v['index'] for v in used.values()], anchors=anchor_info.get('path')),
        锚=anchor_info,
        未出fleet键=missing_fleet,
        自基线不足=ins_rows,
    )
    meta['★数据边界'] = (
        f'① 本机只有 {len(used)} 个 CMS 导出色窗 ('
        + '; '.join(f'{w} {v["span_min"]}→{v["span_max"]}' for w, v in used.items())
        + f') ⇒ 分段降级为窗内 ISO 周 ({len(order)} 段, {span[0]}→{span[1]}); '
          f'多窗口径 (健康期"窗"中位) 不可算, 本件 meta.segments_kind 写明实际用的是哪种。'
        f'② 份数: 自基线条目 {n_entries} 条 (台 × 测点 × 该测点**实际存在**的标量), 出数 {n_ok} 条, '
        f'不给数 {len(ins_rows)} 条 (健康期段 < {MIN_WINDOWS}, 明细见 meta.自基线不足; 一律 null + insufficient, '
        f'**不用 fleet 值顶替**)。★ 页面遍历的 5 个标量里, iso_rms_vel 只有发电机两测点有、rms_200 只有主轴承两测点有 '
        f'(y_unit 实测) ⇒ 其余测点这两条**不存在**, 页面 `if not r: continue` 跳过, 本件在 meta.未覆盖标量 里写明。'
        f'③ 锚护栏: {anchor_info.get("n_used", 0)} 条带日期锚已按 ±{ANCHOR_GUARD_DAYS} 天剔段, '
        f'{anchor_info.get("无日期", 0)} 条锚**无日期/类未登记** ⇒ 未做护栏 (逐条见 meta.锚.未用锚)。'
        f'④ fleet 中位要求该段 ≥{FLEET_MIN_TURBINES} 台贡献, 不足则该段不出值 (页面 ×fleet 显示 —)。'
        f'⑤ 本件全部数字由本脚本从 outputs/<场>/m5_cms_tcm/windows/*/index.parquet (data/raw 重算产物) 与 '
        f'reference/<场>/positive_anchors.json 现算, 不含随包旧件的数值搬运。')
    meta['未覆盖标量'] = {s: [m for m in MEAS_BL if m not in {mm for t in self_bl.values() for mm in t.get(s, {})}]
                          for s in SENSORS}
    meta['未覆盖标量'] = {s: v for s, v in meta['未覆盖标量'].items() if v}

    payload = dict(meta=meta, absolute_limits=ABSOLUTE_LIMITS, fleet_baseline=fleet_bl, self_baseline=self_bl)
    txt = json.dumps(payload, ensure_ascii=False, indent=1, allow_nan=False, default=str)
    out = m5 / 'baseline_38.json'
    print(f'\n  条目 {n_entries} 条 · 出数 {n_ok} · 不给数 {len(ins_rows)} ({len({r["turbine"] for r in ins_rows})} 台) '
          f'· fleet 键 {n_fleet_keys} (未出 {len(missing_fleet)})')
    print(f'  self_baseline 台 {len([t for t in self_bl if any(self_bl[t][s] for s in self_bl[t])])} 台 × 测点 {len(SENSORS)} '
          f'· 正常带/阈值样例见下')
    for t, sen, ms in (('WTG01', 'Main_bearing_front', 'Peak'), ('WTG01', 'Gear_HS_rotor_side', 'Rms_Vel'),
                       ('WTG01', 'Generator_DE', 'iso_rms_vel'), ('WTG02', 'Generator_DE', 'Peak')):
        e = (self_bl.get(t) or {}).get(sen, {}).get(ms)
        if e and e.get('self_baseline'):
            print(f'    {t} {SENSOR_CN.get(sen, sen)}|{ms}: 自基线 {e["self_baseline"]} 带 {e["normal_band"]} '
                  f'relMAD {e["rel_mad"]} 阈 {e["threshold_self"]} 末段 {e["latest_window"]}(n={e["latest_n"]}) '
                  f'×自基线 {e["x_self"]} 带内={e["in_normal_band"]} 健康段 {e["n_windows_healthy"]}/{e["n_windows_total"]}'
                  f' 剔 {len(e["excluded"])}')
        elif e:
            print(f'    {t} {SENSOR_CN.get(sen, sen)}|{ms}: 不给数 — {e.get("insufficient")} '
                  f'(健康段 {e["n_windows_healthy"]}/{e["n_windows_total"]}, 剔 {len(e["excluded"])})')
    if ins_rows:
        print('  不给数明细: ' + '; '.join(f'{r["turbine"]} {SENSOR_CN.get(r["sensor"], r["sensor"])}|{r["meas"]} '
                                          f'{r["n_healthy"]}/{r["n_total"]} ({r["why"]})' for r in ins_rows[:8])
              + (' …' if len(ins_rows) > 8 else ''))
    print(f'  产物体积预估 {len(txt.encode("utf-8")) / 1024:.0f} KB ({P.rel(out)}) · 用时 {time.time() - t0:.1f}s')
    if dry:
        print('\n[dry-run] 不写盘、不登记。')
        return payload, 0
    m5.mkdir(parents=True, exist_ok=True)
    out.write_text(txt, encoding='utf-8')
    # ★自登记台账的落点与键 (2026-09-19 实逮): 正本是 `P.out_root(farm)/_derived_manifest.json`,
    #   键必须**相对产物仓根**(其它生成端同一口径)。用 P.store()/P.rel() 会另建一本
    #   `windscada/_derived_manifest.json` 且键带 `outputs/rudong/` 前缀 ⇒ 台账读不到、审计报"未归类"。
    dm_record(P.out_root(farm_name), {out.relative_to(P.out_root(farm_name)).as_posix():'scripts/baseline_38_build.py (CMS 窗标量 → 自基线/fleet/ISO 绝对三层; '
                                 f'段口径 segments_kind={kind}, 实物锚 ±{ANCHOR_GUARD_DAYS}d 护栏, 不足不给数)'},
              by='scripts/baseline_38_build.py')
    print(f'\n[OK] 已写 {P.rel(out)} ({out.stat().st_size / 1024:.0f} KB) · 已自登记')
    return payload, 0


def main() -> int:
    ap = argparse.ArgumentParser(description='三层基线生成端 (m5_cms_tcm/baseline_38.json)')
    ap.add_argument('--farm', default='rudong')
    ap.add_argument('--dry-run', action='store_true', help='只算不写盘 (打印漏斗与不给数清单)')
    args = ap.parse_args()
    cfg = cms_farm(args.farm)
    _, rc = build(cfg, args.farm, dry=args.dry_run)
    return rc


if __name__ == '__main__':
    sys.exit(main())