补交功图特征分析脚本与说明

backend/indicator_features/common_residual_analysis.py

@@ -0,0 +1,196 @@
+#!/usr/bin/env python3
+"""Separate common operating-condition changes from local diagram changes."""
+from __future__ import annotations
+
+import argparse
+import csv
+import sys
+from collections import defaultdict
+from datetime import datetime, timedelta
+from pathlib import Path
+
+import numpy as np
+
+BACKEND = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(BACKEND))
+from event_bins import EVENTS  # noqa: E402
+
+BAD = {("7", "四", "盖侧"), ("7", "六", "盖侧"), ("8", "六", "盖侧")}
+SCALARS = {
+    "p_suction": "吸气压力", "p_discharge": "排气压力", "p_span": "压力幅度",
+    "suction_frac": "吸气占比", "discharge_frac": "排气占比",
+    "compression_slope": "压缩斜率", "expansion_slope": "膨胀斜率",
+    "corner_sharpness": "拐角变化",
+}
+
+
+def load(path: str) -> list[dict]:
+    with open(path, encoding="utf-8-sig", newline="") as handle:
+        rows = list(csv.DictReader(handle))
+    for row in rows:
+        row["time"] = datetime.fromisoformat(row["sample_time"])
+        row["span"] = float(row.get("p_span") or 0)
+        for key in list(SCALARS) + [k for k in row if k.startswith("prof_")]:
+            try:
+                row[key] = float(row[key])
+            except (TypeError, ValueError):
+                row[key] = np.nan
+    return rows
+
+
+def median(rows: list[dict], key: str) -> float | None:
+    values = [r[key] for r in rows if np.isfinite(r[key])]
+    return float(np.median(values)) if values else None
+
+
+def profile_median(rows: list[dict], names: list[str]) -> np.ndarray | None:
+    if not rows:
+        return None
+    values = np.asarray([[r[n] for n in names] for r in rows], dtype=float)
+    values = values[np.all(np.isfinite(values), axis=1)]
+    return np.median(values, axis=0) if len(values) else None
+
+
+def pct(value: float | None) -> str:
+    return "" if value is None else f"{value:.5f}"
+
+
+def main() -> None:
+    parser = argparse.ArgumentParser(description="公共工况变化与局部功图残差")
+    parser.add_argument("--csv", nargs="+", required=True)
+    parser.add_argument("--unit", nargs="+", required=True)
+    parser.add_argument("--pre-days", type=int, default=30)
+    parser.add_argument("--post-days", type=int, default=30)
+    parser.add_argument("--outdir", required=True)
+    parser.add_argument("--event", nargs=3, action="append", metavar=("UNIT", "DATE", "KIND"),
+                        help="只分析指定事件；可重复传入")
+    args = parser.parse_args()
+
+    by_unit = {unit: load(path) for unit, path in zip(args.unit, args.csv)}
+    outdir = Path(args.outdir)
+    outdir.mkdir(parents=True, exist_ok=True)
+    all_rows: list[dict] = []
+    summary: list[dict] = []
+    normal: list[dict] = []
+
+    events = [tuple(item) for item in args.event] if args.event else EVENTS
+    for unit, date_text, kind in events:
+        source = [r for r in by_unit.get(unit, []) if r["span"] >= 0.3]
+        t0 = datetime.fromisoformat(date_text)
+        pre_start = t0 - timedelta(days=args.pre_days)
+        post_end = t0 + timedelta(days=args.post_days)
+        points = sorted({(r["cylinder"], r["side"]) for r in source if (unit, r["cylinder"], r["side"]) not in BAD})
+        profiles = sorted(k for k in source[0] if k.startswith("prof_")) if source else []
+        deltas: dict[tuple[str, str], np.ndarray] = {}
+        profile_pairs = {}
+        scalar_changes: dict[tuple[str, str], dict[str, float]] = {}
+        for cylinder, side in points:
+            rows = [r for r in source if r["cylinder"] == cylinder and r["side"] == side]
+            pre = [r for r in rows if pre_start <= r["time"] < t0]
+            post = [r for r in rows if t0 <= r["time"] <= post_end]
+            if not pre or not post:
+                continue
+            before = profile_median(pre, profiles)
+            after = profile_median(post, profiles)
+            if before is None or after is None:
+                continue
+            delta = after - before
+            deltas[(cylinder, side)] = delta
+            profile_pairs[(cylinder, side)] = (before, after, len(pre), len(post))
+            changes = {}
+            for key in SCALARS:
+                a, b = median(pre, key), median(post, key)
+                changes[key] = np.nan if a is None or b is None else b - a
+            scalar_changes[(cylinder, side)] = changes
+
+        if not deltas:
+            summary.append({"机组": unit, "事件": date_text, "故障类型": kind, "可比较测点": 0,
+                            "公共形状变化": "", "局部残差最大": "", "判断": "修前无有效功图"})
+            continue
+
+        matrix = np.vstack(list(deltas.values()))
+        common = np.median(matrix, axis=0)
+        common_norm = float(np.linalg.norm(common))
+        local_scores = {}
+        for point, delta in deltas.items():
+            residual = delta - common
+            local = float(np.linalg.norm(residual))
+            total = float(np.linalg.norm(delta))
+            local_scores[point] = (local, total, residual)
+            cylinder, side = point
+            before, after, pre_n, post_n = profile_pairs[point]
+            for idx, name in enumerate(profiles):
+                all_rows.append({
+                    "机组": unit, "事件": date_text, "故障类型": kind,
+                    "测点": f"{cylinder}缸{side}", "角度": f"{idx * 360 / len(profiles):.1f}°",
+                    "前后变化": round(float(after[idx] - before[idx]), 6),
+                    "公共变化": round(float(common[idx]), 6),
+                    "局部残差": round(float(after[idx] - before[idx] - common[idx]), 6),
+                    "修前周期": pre_n, "修后周期": post_n,
+                })
+
+        ranked = sorted(local_scores.items(), key=lambda item: item[1][0], reverse=True)
+        top = ranked[0]
+        ratio = top[1][0] / max(common_norm, 1e-12)
+        judgment = "局部差异优先" if ratio >= 1.2 else "公共工况变化优先"
+        summary.append({
+            "机组": unit, "事件": date_text, "故障类型": kind,
+            "可比较测点": len(deltas), "公共形状变化": round(common_norm, 4),
+            "局部残差最大": f"{top[0][0]}缸{top[0][1]}",
+            "最大局部残差": round(top[1][0], 4), "该点总变化": round(top[1][1], 4),
+            "局部/公共": round(ratio, 2), "判断": judgment,
+        })
+
+        # Normal candidates: closest to the pre-event median profile for each point.
+        for point, (before, after, pre_n, post_n) in profile_pairs.items():
+            cylinder, side = point
+            rows = [r for r in source if r["cylinder"] == cylinder and r["side"] == side]
+            pre_rows = [r for r in rows if pre_start <= r["time"] < t0]
+            distances = []
+            for row in pre_rows:
+                vector = np.asarray([row[n] for n in profiles], dtype=float)
+                if np.all(np.isfinite(vector)):
+                    distances.append((float(np.linalg.norm(vector - before)), row))
+            if distances:
+                distance, row = min(distances, key=lambda item: item[0])
+                local, total, _ = local_scores[point]
+                normal.append({
+                    "机组": unit, "事件": date_text, "测点": f"{cylinder}缸{side}",
+                    "类别": "正常候选", "日期": row["sample_time"],
+                    "距修前模板": round(distance, 4), "事件后该点变化": round(total, 4),
+                    "局部残差": round(local, 4),
+                })
+
+        # One compact scalar table per event and point.
+        for point, changes in scalar_changes.items():
+            for key, label in SCALARS.items():
+                all_rows.append({
+                    "机组": unit, "事件": date_text, "故障类型": kind,
+                    "测点": f"{point[0]}缸{point[1]}", "角度": label,
+                    "前后变化": "" if np.isnan(changes[key]) else round(changes[key], 6),
+                    "公共变化": "", "局部残差": "", "修前周期": "", "修后周期": "",
+                })
+
+    def write(path: Path, rows: list[dict]) -> None:
+        fields = []
+        for row in rows:
+            for field in row:
+                if field not in fields:
+                    fields.append(field)
+        with path.open("w", encoding="utf-8-sig", newline="") as handle:
+            writer = csv.DictWriter(handle, fieldnames=fields)
+            writer.writeheader()
+            writer.writerows(rows)
+
+    write(outdir / "三事件_公共与局部变化.csv", all_rows)
+    write(outdir / "三事件_公共工况判断.csv", summary)
+    write(outdir / "三事件_正常功图候选.csv", normal)
+    print(f"公共/局部变化 -> {outdir / '三事件_公共与局部变化.csv'}")
+    print(f"事件判断 -> {outdir / '三事件_公共工况判断.csv'}")
+    print(f"正常候选 -> {outdir / '三事件_正常功图候选.csv'}")
+    for row in summary:
+        print(row)
+
+
+if __name__ == "__main__":
+    main()

backend/indicator_features/event_trend_summary.py

@@ -0,0 +1,84 @@
+#!/usr/bin/env python3
+"""Six-window event trend summary for sampled cycle profiles and pressure levels."""
+from __future__ import annotations
+
+import argparse
+import csv
+import sys
+from datetime import datetime, timedelta
+from pathlib import Path
+
+import numpy as np
+
+BACKEND = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(BACKEND))
+from event_bins import EVENTS  # noqa: E402
+
+BAD = {("7", "四", "盖侧"), ("7", "六", "盖侧"), ("8", "六", "盖侧")}
+
+
+def load(path):
+    with open(path, encoding="utf-8-sig", newline="") as handle:
+        rows = list(csv.DictReader(handle))
+    for row in rows:
+        row["time"] = datetime.fromisoformat(row["sample_time"])
+        for key in ("p_suction", "p_discharge", "p_span"):
+            row[key] = float(row[key]) if row.get(key) else np.nan
+        for key in row:
+            if key.startswith("prof_"):
+                row[key] = float(row[key]) if row[key] else np.nan
+    return rows
+
+
+def med(rows, key):
+    values = [r[key] for r in rows if np.isfinite(r[key])]
+    return float(np.median(values)) if values else np.nan
+
+
+def main():
+    parser = argparse.ArgumentParser(description="事件前后六段趋势")
+    parser.add_argument("--csv", nargs="+", required=True)
+    parser.add_argument("--unit", nargs="+", required=True)
+    parser.add_argument("--out", required=True)
+    parser.add_argument("--event", nargs=3, action="append", metavar=("UNIT", "DATE", "KIND"),
+                        help="只分析指定事件；可重复传入")
+    args = parser.parse_args()
+    data = {u: load(p) for u, p in zip(args.unit, args.csv)}
+    rows_out = []
+    events = [tuple(item) for item in args.event] if args.event else EVENTS
+    for unit, date_text, kind in events:
+        source = [r for r in data.get(unit, []) if (unit, r["cylinder"], r["side"]) not in BAD and r["p_span"] >= 0.3]
+        t0 = datetime.fromisoformat(date_text)
+        profile_names = sorted(k for k in source[0] if k.startswith("prof_")) if source else []
+        for cylinder, side in sorted({(r["cylinder"], r["side"]) for r in source}):
+            point = [r for r in source if r["cylinder"] == cylinder and r["side"] == side]
+            base = None
+            for index, (lo, hi) in enumerate([(-30, -20), (-20, -10), (-10, 0), (0, 10), (10, 20), (20, 30)]):
+                a, b = t0 + timedelta(days=lo), t0 + timedelta(days=hi)
+                selected = [r for r in point if a <= r["time"] < b]
+                profile = np.asarray([[r[k] for k in profile_names] for r in selected], dtype=float)
+                profile = np.median(profile, axis=0) if len(profile) and np.all(np.isfinite(profile), axis=1).any() else None
+                if profile is not None and base is None:
+                    base = profile
+                distance = float(np.linalg.norm(profile - base)) if profile is not None and base is not None else np.nan
+                rows_out.append({
+                    "机组": unit, "事件": date_text, "故障类型": kind,
+                    "测点": f"{cylinder}缸{side}", "时间段": f"{lo}~{hi}天",
+                    "周期数": len(selected),
+                    "吸气压力": med(selected, "p_suction"),
+                    "排气压力": med(selected, "p_discharge"),
+                    "压力幅度": med(selected, "p_span"),
+                    "相对首段形状距离": "" if not np.isfinite(distance) else round(distance, 5),
+                })
+    out = Path(args.out)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    fields = list(rows_out[0])
+    with out.open("w", encoding="utf-8-sig", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=fields)
+        writer.writeheader()
+        writer.writerows(rows_out)
+    print(f"事件趋势 -> {out}（{len(rows_out)} 行）")
+
+
+if __name__ == "__main__":
+    main()

backend/indicator_features/finalize_pseudo_online.py

@@ -0,0 +1,107 @@
+#!/usr/bin/env python3
+"""Merge final pseudo-online event replays and unlabelled controls."""
+from __future__ import annotations
+
+import csv
+from pathlib import Path
+
+ROOT=Path("cache/indicator")
+OUT=ROOT/"伪在线检测"
+EVENT_DIRS=[ROOT/"伪在线检测_7号2025final",ROOT/"伪在线检测_7号2026final",
+            ROOT/"伪在线检测_4号final",ROOT/"伪在线检测_6号final"]
+HEALTH=ROOT/"伪在线健康段v2"
+
+
+def load(path):
+    with path.open(encoding="utf-8-sig",newline="") as handle:return list(csv.DictReader(handle))
+
+
+def write(path,rows,fields=None):
+    fields=list(dict.fromkeys(key for row in rows for key in row)) if rows else list(fields or [])
+    with path.open("w",encoding="utf-8-sig",newline="") as handle:
+        writer=csv.DictWriter(handle,fieldnames=fields);writer.writeheader();writer.writerows(rows)
+
+
+def main():
+    OUT.mkdir(parents=True,exist_ok=True)
+    details=[];bands=[];validations=[]
+    for folder in EVENT_DIRS:
+        details+=load(folder/"伪在线_窗口检测明细.csv")
+        bands+=load(folder/"伪在线_角度带告警.csv")
+        validations+=load(folder/"伪在线_事件验证.csv")
+    write(OUT/"伪在线_窗口检测明细.csv",details)
+    write(OUT/"伪在线_角度带告警.csv",bands)
+    write(OUT/"伪在线_事件验证.csv",validations)
+
+    health=[]
+    for unit in ("7","8","9"):
+        path=HEALTH/f"伪在线_未标注对照触发明细_{unit}号.csv"
+        if path.exists():health+=load(path)
+    write(OUT/"伪在线_未标注对照触发明细.csv",health,["机组","样本标签"])
+    control_summary=[]
+    for unit in sorted({row["机组"] for row in health}):
+        rows=[row for row in health if row["机组"]==unit]
+        point_days={(row["测点"],row["日期"]) for row in rows}
+        candidate={(row["测点"],row["日期"]) for row in rows if row["等级"] in ("候选","持续预警","未分类异常")}
+        persistent={(row["测点"],row["日期"]) for row in rows if row["等级"]=="持续预警"}
+        control_summary.append({"机组":unit,"标签":"未标注对照，非确认健康","日期数":len({row["日期"] for row in rows}),
+            "测点数":len({row["测点"] for row in rows}),"测点日数":len(point_days),
+            "候选测点日数":len(candidate),"候选测点日率":round(len(candidate)/max(1,len(point_days)),4),
+            "持续预警测点日数":len(persistent),"持续预警测点日率":round(len(persistent)/max(1,len(point_days)),4),
+            "说明":"7日运行片段；8号片段距维修17天，不保证健康。用于粗略候选压力测试，不是确认误报率。"})
+    write(OUT/"伪在线_未标注对照统计.csv",control_summary,["机组","标签"])
+
+    common_by_angle={}
+    grouped={}
+    for row in details:
+        if row["等级"] not in ("候选","持续预警"):continue
+        angle=int(row["角度箱"]);grouped.setdefault(angle,[]).append(row)
+    for angle,rows in grouped.items():
+        events=sorted({row["事件"] for row in rows})
+        if len(events)<2:continue
+        common_by_angle[angle]={"角度范围":f"{angle}-{angle+10}°","事件数":len(events),"事件":";".join(events),
+            "测点":";".join(sorted({row["测点"] for row in rows})),
+            "候选窗":sum(row["等级"]=="候选" for row in rows),"持续预警窗":sum(row["等级"]=="持续预警" for row in rows),
+            "历史签名匹配窗":sum(row.get("历史事件签名匹配")=="1" for row in rows),
+            "未分类异常窗":sum(row["等级"]=="未分类异常" for row in rows),
+            "最大快RMS":round(max(float(row["综合RMS"]) for row in rows),3),
+            "最大慢RMS":round(max(float(row["慢通道综合RMS"]) for row in rows),3)}
+    common=sorted(common_by_angle.values(),key=lambda row:(-row["事件数"],-row["持续预警窗"],-row["候选窗"]))
+    write(OUT/"伪在线_跨事件共性.csv",common,["角度范围","事件数"])
+
+    lines=["# 伪在线双通道故障检测验证", "", "## 回放设置", "",
+        "- 快通道：长历史基线，每测点每天粗采最多3周期；24小时检测窗口使用全部周期并分4个6小时子窗。",
+        "- 慢通道：本次运行段最早3个有效日期建立固定基线；后续有效日检测。",
+        "- 两路均使用周期平台归一化、测点自身稳健Z和单机公共变化扣除。",
+        "- 历史签名只用于标记相似性；未匹配异常仍保留为未分类异常，不静默丢弃。",
+        "- 全周期大范围异常单列为通道质控，不计入局部气阀候选。", ""]
+    for event in sorted({row["事件"] for row in validations}):
+        rows=[row for row in validations if row["事件"]==event]
+        rows.sort(key=lambda row:(-int(row["持续预警窗口数"]),-float(row["最大综合RMS"])))
+        lines += [f"## {event}","",f"- 验证候选角度带：{len(rows)}。"]
+        for i,row in enumerate(rows[:12],1):
+            lines.append(f"{i}. {row['测点']} {row['角度带']}：首次候选{row['首次候选日期']}，提前{row['提前日数']}天，持续预警{row['持续预警窗口数']}窗，快通道RMS {float(row['最大综合RMS']):.1f}。")
+        lines.append("")
+    lines += ["## 目标特征验证", "",
+        "- 7号2026一缸盖侧240~270°：5月17日左右相邻箱形成候选，5月19日前后达到持续条件，较5月21日事件提前约2~4天。",
+        "- 7号2026二缸盖侧140~150°：慢通道能在事件前触发，但长历史健康对照触发率高，当前只能视为慢漂移候选，尚未证明故障特异性。",
+        "- 6号三缸轴侧160~180°：快通道在事件前约1~3天触发。",
+        "- 4号出现多测点候选；候选分散，需降低误报后再判断哪些与故障相关。",
+        "- 7号2025二缸盖侧全周期异常被标为通道质控，不进入局部故障签名；事件附近缺测，不能证明预测了8月或10月故障。", "",
+        "## 未标注运行段压力测试", ""]
+    for row in control_summary:
+        lines.append(f"- {row['机组']}号：{row['日期数']}日、{row['测点数']}测点；候选测点日率{row['候选测点日率']:.1%}，持续预警测点日率{row['持续预警测点日率']:.1%}。这是未标注片段触发率，不是真实误报率。")
+    lines += ["", "## 判定", "",
+        "当前实现仍不能上线。健康压力测试中7/8/9号触发率偏高，且慢通道会捕捉季节、负荷和运行段状态转换。检测器可以发现事件前的局部变化，但还没有足够证据判断这些变化是否具有故障特异性。",
+        "正式阈值冻结前，需要更长、确认无维修/故障的负样本；建立事件级而不是10度箱级的误报控制；对传感器异常和跨运行段工况切换设置独立状态分类。", "",
+        "## 推荐下一版规则", "",
+        "1. 保留快慢双通道，慢通道只在每个新运行段开始时重置，不能用全历史模板。",
+        "2. 先做整通道质控：近零幅、异常覆盖超过半周期、异常平台量程，剔除后再做局部检测。",
+        "3. 对候选角度带要求至少两个特征族、连续时间支持和局部公共变化残差；历史签名匹配只加分，不作为唯一条件。",
+        "4. 误报按事件/机组/运行段计数，不能把重叠10度箱和相邻24小时窗口当独立误报。",
+        "5. 目前未标注对照每台仅7日，且8号段距维修17日；需更多独立健康运行段后再估计生产误报率。"]
+    (OUT/"伪在线检测结论.md").write_text("\n".join(lines)+"\n",encoding="utf-8")
+    print("merged",len(details),len(bands),len(validations),len(health),len(common))
+
+
+if __name__=="__main__":main()

backend/indicator_features/local_evolution_analysis.py

@@ -0,0 +1,578 @@
+#!/usr/bin/env python3
+"""Find directional 10-degree local feature evolution around repair events."""
+from __future__ import annotations
+
+import argparse
+import csv
+import math
+import sys
+from collections import defaultdict
+from datetime import date, datetime
+from pathlib import Path
+
+import numpy as np
+
+BACKEND = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(BACKEND))
+
+from indicator_features.extract_from_db import (  # noqa: E402
+    connect, fetch_cycles, fetch_file_rows, point_name, sample_per_day,
+)
+from indicator_features.indicator_features import _plateau_levels, smooth  # noqa: E402
+
+CYLINDERS = ("一", "二", "三", "四", "五", "六")
+SIDES = ("盖侧", "轴侧")
+METRICS = ("p50", "slope", "std", "iqr", "d_rms", "d2_abs_max")
+METRIC_CN = {
+    "p50": "局部中位压力", "slope": "局部斜率", "std": "标准差",
+    "iqr": "四分位距", "d_rms": "导数RMS", "d2_abs_max": "最大二阶导",
+}
+FAMILY = {"p50": "位置", "slope": "斜率", "std": "波动", "iqr": "波动",
+          "d_rms": "导数", "d2_abs_max": "导数"}
+BAD = {
+    "4": {("二", "盖侧"), ("四", "盖侧"), ("六", "盖侧")},
+    "6": {("二", "盖侧"), ("四", "盖侧"), ("六", "盖侧")},
+    "7": {("四", "盖侧"), ("六", "盖侧")},
+}
+EVENTS = [
+    {
+        "id": "7号_2025维修过程", "unit": "7", "label": "7号2025停机维修过程",
+        "start": "2025-04-29", "end": "2026-02-17", "event": "2025-08-08",
+        "pre": [("2025-04-29", "2025-05-04"), ("2025-06-11", "2025-06-14"), ("2025-07-25", "2025-07-28")],
+        "pre_segments": [("2025-04-29", "2025-05-04"), ("2025-06-11", "2025-06-14"), ("2025-07-25", "2025-07-28")],
+        "early": [("2025-04-29", "2025-05-04")], "late": [("2025-07-25", "2025-07-28")],
+        "post_base": [("2025-11-19", "2025-11-21")],
+        "post_check": [("2026-02-10", "2026-02-17")],
+        "mode": "跨运行段", "markers": "2025-08-08;2025-10-07",
+    },
+    {
+        "id": "7号_2026-05-21", "unit": "7", "label": "7号2026-05-21气阀事件",
+        "start": "2026-02-10", "end": "2026-06-07", "event": "2026-05-21",
+        "pre": [("2026-05-06", "2026-05-21")],
+        "pre_segments": [("2026-02-10", "2026-02-17"), ("2026-03-12", "2026-04-05"), ("2026-05-06", "2026-05-21")],
+        "early": [("2026-05-06", "2026-05-09")], "late": [("2026-05-18", "2026-05-21")],
+        "post_base": [("2026-05-31", "2026-06-04")],
+        "post_check": [("2026-06-04", "2026-06-07")],
+        "mode": "连续运行段", "markers": "2026-05-21",
+    },
+    {
+        "id": "4号_2026-06-10", "unit": "4", "label": "4号2026-06-10气阀事件",
+        "start": "2026-04-21", "end": "2026-07-31", "event": "2026-06-10",
+        "pre": [("2026-05-29", "2026-05-31"), ("2026-06-06", "2026-06-10")],
+        "pre_segments": [("2026-04-21", "2026-05-02"), ("2026-05-29", "2026-05-31"), ("2026-06-06", "2026-06-10")],
+        "early": [("2026-06-06", "2026-06-08")], "late": [("2026-06-08", "2026-06-10")],
+        "post_base": [("2026-06-11", "2026-06-15")],
+        "post_check": [("2026-06-21", "2026-06-22"), ("2026-06-30", "2026-07-08"),
+                       ("2026-07-11", "2026-07-14"), ("2026-07-22", "2026-07-31")],
+        "mode": "短运行段", "markers": "2026-06-10",
+    },
+    {
+        "id": "6号_2026-04-22_26", "unit": "6", "label": "6号2026-04-22~26气阀维修过程",
+        "start": "2026-04-12", "end": "2026-08-01", "event": "2026-04-22",
+        "pre": [("2026-04-12", "2026-04-22")],
+        "pre_segments": [("2026-04-12", "2026-04-22")],
+        "early": [("2026-04-12", "2026-04-15")], "late": [("2026-04-19", "2026-04-22")],
+        "post_base": [("2026-05-10", "2026-05-17")],
+        "post_check": [("2026-05-17", "2026-05-31"), ("2026-06-21", "2026-07-02"),
+                       ("2026-07-24", "2026-08-01")],
+        "mode": "连续运行段", "markers": "2026-04-22;2026-04-26",
+    },
+]
+
+
+def in_ranges(day: str, ranges) -> bool:
+    return any(start <= day < end for start, end in ranges)
+
+
+def robust_scale(values) -> float:
+    values = np.asarray(values, dtype=float)
+    if values.size == 0:
+        return 0.0
+    scale = float(np.median(np.abs(values - np.median(values))) * 1.4826)
+    return scale if scale > 1e-12 else float(np.std(values))
+
+
+def rankdata(values):
+    values = np.asarray(values)
+    order = np.argsort(values, kind="mergesort")
+    ranks = np.empty(len(values), dtype=float)
+    start = 0
+    while start < len(values):
+        end = start + 1
+        while end < len(values) and values[order[end]] == values[order[start]]:
+            end += 1
+        ranks[order[start:end]] = (start + end - 1) / 2
+        start = end
+    return ranks
+
+
+def spearman(values) -> float:
+    if len(values) < 3:
+        return float("nan")
+    x = np.arange(len(values), dtype=float)
+    y = rankdata(values)
+    if np.std(y) <= 1e-12:
+        return 0.0
+    return float(np.corrcoef(x, y)[0, 1])
+
+
+def normalized_bins(signal):
+    signal = np.asarray(signal, dtype=float)
+    signal = signal[np.isfinite(signal)]
+    if len(signal) < 60:
+        return None
+    pressure = smooth(signal, 21)
+    suction, discharge, _ = _plateau_levels(pressure)
+    span = discharge - suction
+    if not np.isfinite(span) or span < .3:
+        return None
+    normalized = (pressure - suction) / span
+    angles = np.arange(len(normalized), dtype=float) * 360 / len(normalized)
+    result = []
+    for index in range(36):
+        mask = (angles >= index * 10) & (angles < (index + 1) * 10)
+        x, y = angles[mask], normalized[mask]
+        if len(y) < 5:
+            return None
+        slope = float(np.polyfit(x, y, 1)[0])
+        dy = np.diff(y)
+        d2 = np.diff(dy)
+        result.append({
+            "bin": index, "bin_start": index * 10,
+            "p50": float(np.median(y)), "slope": slope, "std": float(np.std(y)),
+            "iqr": float(np.percentile(y, 75) - np.percentile(y, 25)),
+            "d_rms": float(np.sqrt(np.mean(dy ** 2))),
+            "d2_abs_max": float(np.max(np.abs(d2))) if len(d2) else 0.0,
+        })
+    return result
+
+
+def extract(output: Path, per_day: int):
+    grouped = defaultdict(list)
+    connection = connect()
+    fields = ["事件", "机组", "测点", "缸", "腔", "文件ID", "采样时间", "日期", "角度箱"] + list(METRICS)
+    output.parent.mkdir(parents=True, exist_ok=True)
+    with output.open("w", encoding="utf-8-sig", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=fields)
+        writer.writeheader()
+        try:
+            for event in EVENTS:
+                periods = 0
+                for cylinder in CYLINDERS:
+                    for side in SIDES:
+                        if (cylinder, side) in BAD[event["unit"]]:
+                            continue
+                        files = fetch_file_rows(connection, point_name(event["unit"], cylinder, side), event["start"], event["end"])
+                        picked = sample_per_day(files, per_day)
+                        cycles = fetch_cycles(connection, [int(row["id"]) for row in picked], 100) if picked else {}
+                        for row in picked:
+                            features = normalized_bins(cycles.get(int(row["id"]), np.array([])))
+                            if features is None:
+                                continue
+                            periods += 1
+                            day = str(row["sample_time"])[:10]
+                            point = f"{cylinder}缸{side}"
+                            for item in features:
+                                record = {
+                                    "事件": event["id"], "机组": event["unit"], "测点": point,
+                                    "缸": cylinder, "腔": side, "文件ID": int(row["id"]),
+                                    "采样时间": str(row["sample_time"]), "日期": day,
+                                    "角度箱": item["bin_start"], **{metric: item[metric] for metric in METRICS},
+                                }
+                                writer.writerow(record)
+                                for metric in METRICS:
+                                    grouped[(event["id"], point, day, item["bin_start"], metric)].append(item[metric])
+                print(f"{event['label']}: 有效周期 {periods:,}", flush=True)
+        finally:
+            connection.close()
+    return grouped
+
+
+def load_extracted(path: Path):
+    grouped = defaultdict(list)
+    with path.open(encoding="utf-8-sig", newline="") as handle:
+        for row in csv.DictReader(handle):
+            for metric in METRICS:
+                grouped[(row["事件"], row["测点"], row["日期"], int(row["角度箱"]), metric)].append(float(row[metric]))
+    return grouped
+
+
+def baseline_stats(series, ranges):
+    daily_values = [daily for day, daily, _ in series if in_ranges(day, ranges)]
+    cycle_values = [value for day, _, values in series if in_ranges(day, ranges) for value in values]
+    center = float(np.median(daily_values)) if daily_values else float("nan")
+    # Keep dates equal-weighted for the center, but estimate natural variation
+    # from cycles so a 2-3 day baseline cannot produce near-zero scales.
+    scale = robust_scale(cycle_values)
+    return center, scale, len(daily_values), len(cycle_values)
+
+
+def merge_bands(candidates):
+    grouped = defaultdict(list)
+    for row in candidates:
+        if row["可信等级"] == "低":
+            continue
+        start = int(row["角度范围"].split("-")[0])
+        grouped[(row["事件"], row["测点"], row["故障主方向"], row["可信等级"])].append((start, row))
+    bands = []
+    for (event, point, direction, confidence), values in grouped.items():
+        values.sort()
+        runs = []
+        for start, row in values:
+            if runs and start == runs[-1][-1][0] + 10:
+                runs[-1].append((start, row))
+            else:
+                runs.append([(start, row)])
+        # Merge the circular 350-360 and 0-10 runs.
+        if len(runs) > 1 and runs[0][0][0] == 0 and runs[-1][-1][0] == 350:
+            runs[0] = runs[-1] + runs[0]
+            runs.pop()
+        for run in runs:
+            starts = [item[0] for item in run]
+            rows = [item[1] for item in run]
+            if starts[0] > starts[-1]:
+                band = f"{starts[0]}-360°及0-{starts[-1] + 10}°"
+            else:
+                band = f"{starts[0]}-{starts[-1] + 10}°"
+            bands.append({
+                "事件": event, "测点": point, "角度带": band, "故障主方向": direction,
+                "包含10度箱数": len(rows), "可信等级": confidence,
+                "最高评分": max(row["候选评分"] for row in rows),
+                "演化指标数": max(row["演化指标数"] for row in rows),
+                "长期演化指标数": max(row["长期演化指标数"] for row in rows),
+                "恢复指标数": max(row["恢复指标数"] for row in rows),
+                "长期稳定恢复指标数": max(row["长期稳定恢复指标数"] for row in rows),
+                "长期恢复比例平均": max((row["长期恢复比例平均"] for row in rows
+                                         if row["长期恢复比例平均"] != ""), default=""),
+                "局部幅度平均Z": max(row["局部幅度平均Z"] for row in rows),
+                "局部幅度综合RMS": max(row["局部幅度综合RMS"] for row in rows),
+                "最早持续日期": min((row["最早持续日期"] for row in rows if row["最早持续日期"]), default=""),
+                "代表指标": max(rows, key=lambda row: row["候选评分"])["指标摘要"],
+            })
+    bands.sort(key=lambda row: (row["事件"], -row["最高评分"], -row["包含10度箱数"]))
+    return bands
+
+
+def analyze(grouped, daily_output: Path, candidate_output: Path, repeat_output: Path, band_output: Path,
+            segment_output: Path):
+    daily_raw = {key: float(np.median(values)) for key, values in grouped.items()}
+    counts = {key: len(values) for key, values in grouped.items()}
+    event_by_id = {event["id"]: event for event in EVENTS}
+    points_by_event = defaultdict(set)
+    for event_id, point, _, _, _ in daily_raw:
+        points_by_event[event_id].add(point)
+    series_index = defaultdict(list)
+    for (event_id, point, day, angle, metric), values in grouped.items():
+        series_index[(event_id, point, angle, metric)].append((day, daily_raw[(event_id, point, day, angle, metric)], values))
+
+    stats = {}
+    for event in EVENTS:
+        for point in points_by_event[event["id"]]:
+            for metric in METRICS:
+                scales = []
+                provisional = {}
+                for angle in range(0, 360, 10):
+                    series = series_index[(event["id"], point, angle, metric)]
+                    post = baseline_stats(series, event["post_base"])
+                    early = baseline_stats(series, event["early"])
+                    provisional[angle] = (post, early)
+                    scales.extend([post[1], early[1]])
+                positive = [scale for scale in scales if np.isfinite(scale) and scale > 1e-12]
+                floor = .2 * float(np.median(positive)) if positive else 1e-9
+                for angle, (post, early) in provisional.items():
+                    stats[(event["id"], point, angle, metric)] = {
+                        "post_center": post[0], "post_scale": max(post[1], floor), "post_days": post[2],
+                        "post_cycles": post[3],
+                        "early_center": early[0], "early_scale": max(early[1], floor), "early_days": early[2],
+                        "early_cycles": early[3],
+                    }
+
+    daily_rows = []
+    for key, value in daily_raw.items():
+        event_id, point, day, angle, metric = key
+        stat = stats[(event_id, point, angle, metric)]
+        z_post = (value - stat["post_center"]) / stat["post_scale"] if np.isfinite(stat["post_center"]) else float("nan")
+        z_early = (value - stat["early_center"]) / stat["early_scale"] if np.isfinite(stat["early_center"]) else float("nan")
+        if not (np.isfinite(z_post) and np.isfinite(z_early)):
+            continue
+        daily_rows.append({
+            "事件": event_id, "测点": point, "日期": day, "角度箱": angle,
+            "指标": metric, "指标中文": METRIC_CN[metric], "特征族": FAMILY[metric],
+            "每日中位": value, "当日周期数": counts[key],
+            "维修后基线中心": stat["post_center"], "维修后基线尺度": stat["post_scale"],
+            "维修后基线日数": stat["post_days"], "相对维修后Z": z_post,
+            "维修后基线周期数": stat["post_cycles"],
+            "早期基线中心": stat["early_center"], "早期基线尺度": stat["early_scale"],
+            "早期基线日数": stat["early_days"], "早期基线周期数": stat["early_cycles"],
+            "相对早期Z": z_early,
+        })
+    common = defaultdict(list)
+    for row in daily_rows:
+        if np.isfinite(row["相对维修后Z"]):
+            common[(row["事件"], row["日期"], row["角度箱"], row["指标"])].append(row["相对维修后Z"])
+    common = {key: float(np.median(values)) for key, values in common.items()}
+    for row in daily_rows:
+        public = common.get((row["事件"], row["日期"], row["角度箱"], row["指标"]), float("nan"))
+        row["全机公共Z"] = public
+        row["局部残差Z"] = row["相对维修后Z"] - public
+
+    segment_rows = []
+    for event in EVENTS:
+        ranges = [("故障前运行段", start, end) for start, end in event["pre_segments"]]
+        ranges += [("维修后基线", start, end) for start, end in event["post_base"]]
+        ranges += [("维修后长期验证", start, end) for start, end in event["post_check"]]
+        event_rows = [row for row in daily_rows if row["事件"] == event["id"]]
+        keys = sorted({(row["测点"], row["角度箱"], row["指标"], row["指标中文"], row["特征族"])
+                       for row in event_rows})
+        for segment_type, start, end in ranges:
+            for point, angle, metric, metric_cn, family in keys:
+                selected = [row for row in event_rows if row["测点"] == point and row["角度箱"] == angle
+                            and row["指标"] == metric and start <= row["日期"] < end]
+                if not selected:
+                    continue
+                segment_rows.append({
+                    "事件": event["id"], "阶段类型": segment_type, "阶段开始": start, "阶段结束": end,
+                    "测点": point, "角度箱": angle, "指标": metric, "指标中文": metric_cn, "特征族": family,
+                    "有效日期数": len({row["日期"] for row in selected}),
+                    "周期数": sum(int(row["当日周期数"]) for row in selected),
+                    "阶段每日中位的中位": float(np.median([row["每日中位"] for row in selected])),
+                    "相对维修后Z中位": float(np.median([row["相对维修后Z"] for row in selected])),
+                    "全机公共Z中位": float(np.median([row["全机公共Z"] for row in selected])),
+                    "局部残差Z中位": float(np.median([row["局部残差Z"] for row in selected])),
+                })
+
+    daily_output.parent.mkdir(parents=True, exist_ok=True)
+    with daily_output.open("w", encoding="utf-8-sig", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(daily_rows[0]))
+        writer.writeheader(); writer.writerows(daily_rows)
+    with segment_output.open("w", encoding="utf-8-sig", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(segment_rows[0]))
+        writer.writeheader(); writer.writerows(segment_rows)
+
+    rows_index = defaultdict(list)
+    for row in daily_rows:
+        rows_index[(row["事件"], row["测点"], row["角度箱"], row["指标"])].append(row)
+    metric_candidates = []
+    for key, rows in rows_index.items():
+        event_id, point, angle, metric = key
+        event = event_by_id[event_id]
+        rows.sort(key=lambda row: row["日期"])
+        pre = [row for row in rows if in_ranges(row["日期"], event["pre"])]
+        late = [row for row in rows if in_ranges(row["日期"], event["late"])]
+        check = [row for row in rows if in_ranges(row["日期"], event["post_check"])]
+        if len(pre) < 3 or not late:
+            continue
+        late_z = float(np.median([row["相对维修后Z"] for row in late]))
+        late_residual = float(np.median([row["局部残差Z"] for row in late]))
+        direction = 1 if late_z >= 0 else -1
+        directed_late_residual = direction * late_residual
+        directed = [direction * row["相对维修后Z"] for row in pre]
+        directed_local = [direction * row["局部残差Z"] for row in pre]
+        rho = spearman(directed)
+        last_days = directed[-min(5, len(directed)):]
+        persistent = sum(value >= 3 for value in last_days)
+        local_persistent = sum(value >= 2 for value in directed_local[-min(5, len(directed_local)):])
+        late_early_z = float(np.median([direction * row["相对早期Z"] for row in late]))
+        check_z = float(np.median([abs(row["相对维修后Z"]) for row in check])) if check else float("nan")
+        check_segments = []
+        for start, end in event["post_check"]:
+            segment_check = [row for row in rows if start <= row["日期"] < end]
+            if segment_check:
+                check_segments.append(float(np.median([abs(row["相对维修后Z"]) for row in segment_check])))
+        recovery_flags = [value <= 2 or value <= .5 * abs(late_z) for value in check_segments]
+        recovered = bool(recovery_flags and recovery_flags[0])
+        long_recovery_ratio = float(np.mean(recovery_flags)) if recovery_flags else float("nan")
+        long_recovery_stable = bool(recovery_flags and all(recovery_flags))
+        segment_values = []
+        segment_local_values = []
+        segment_labels = []
+        for start, end in event["pre_segments"]:
+            segment_rows = [row for row in rows if start <= row["日期"] < end]
+            if not segment_rows:
+                continue
+            segment_values.append(float(np.median([direction * row["相对维修后Z"] for row in segment_rows])))
+            segment_local_values.append(float(np.median([direction * row["局部残差Z"] for row in segment_rows])))
+            segment_labels.append(f"{start}~{end}")
+        long_rho = spearman(segment_local_values) if len(segment_local_values) >= 3 else float("nan")
+        long_change = segment_local_values[-1] - segment_local_values[0] if len(segment_local_values) >= 2 else float("nan")
+        long_evolution = len(segment_local_values) >= 2 and long_change >= 2 and (
+            len(segment_local_values) == 2 or long_rho >= .5
+        )
+        onset = ""
+        for index in range(len(pre)):
+            window = directed[index:index + 3]
+            if len(window) >= 2 and sum(value >= 3 for value in window) >= 2:
+                onset = pre[index]["日期"]
+                break
+        support = abs(late_z) >= 3 and directed_late_residual >= 2
+        evolution = (persistent >= 2 and rho >= .35) or late_early_z >= 2
+        score = min(abs(late_residual), 12) + min(abs(late_z), 12) * .5
+        score += 3 if evolution else 0
+        score += 3 if recovered else 0
+        score += min(persistent, 3) + min(local_persistent, 3)
+        if event["mode"] != "连续运行段":
+            score -= 2
+        if min(rows[0]["维修后基线日数"], rows[0]["早期基线日数"]) < 3:
+            score -= 2
+        metric_candidates.append({
+            "事件": event_id, "模式": event["mode"], "测点": point, "角度起点": angle,
+            "指标": metric, "指标中文": METRIC_CN[metric], "特征族": FAMILY[metric],
+            "故障方向": "升高" if direction > 0 else "降低", "维修前末段Z": late_z,
+            "维修前末段局部残差Z": late_residual, "定向局部幅度Z": directed_late_residual,
+            "末段相对早期定向Z": late_early_z,
+            "故障前Spearman": rho, "末5日超3Z天数": persistent,
+            "末5日局部超2Z天数": local_persistent, "首次持续日期": onset,
+            "独立维修后验证Z": check_z, "维修后恢复": int(recovered),
+            "后期验证运行段数": len(check_segments),
+            "后期各段绝对Z": ";".join(f"{value:.3f}" for value in check_segments),
+            "长期恢复比例": long_recovery_ratio,
+            "长期恢复稳定": int(long_recovery_stable),
+            "基础支持": int(support), "演化支持": int(evolution), "单指标评分": score,
+            "长期运行段数": len(segment_local_values), "长期段间Spearman": long_rho,
+            "长期段间局部变化Z": long_change, "长期演化支持": int(long_evolution),
+            "长期各段定向Z": ";".join(f"{label}:{value:.3f}" for label, value in zip(segment_labels, segment_local_values)),
+            "维修后基线日数": rows[0]["维修后基线日数"], "早期基线日数": rows[0]["早期基线日数"],
+        })
+
+    groups = defaultdict(list)
+    for row in metric_candidates:
+        if row["基础支持"]:
+            groups[(row["事件"], row["测点"], row["角度起点"])].append(row)
+    candidates = []
+    for (event_id, point, angle), items in groups.items():
+        families = {item["特征族"] for item in items}
+        directions = defaultdict(list)
+        for item in items:
+            directions[item["故障方向"]].append(item)
+        direction, aligned = max(directions.items(), key=lambda pair: len(pair[1]))
+        evolution_n = sum(item["演化支持"] for item in items)
+        long_evolution_n = sum(item["长期演化支持"] for item in items)
+        recovery_n = sum(item["维修后恢复"] for item in items)
+        stable_recovery_n = sum(item["长期恢复稳定"] for item in items)
+        recovery_ratios = [item["长期恢复比例"] for item in items if np.isfinite(item["长期恢复比例"])]
+        directed_amplitudes = [item["定向局部幅度Z"] for item in items]
+        amplitude_mean = float(np.mean(directed_amplitudes))
+        amplitude_rms = float(np.sqrt(np.mean(np.square(directed_amplitudes))))
+        amplitude_median = float(np.median(directed_amplitudes))
+        score = sum(max(0, item["单指标评分"]) for item in items) + 2 * len(families) + 2 * evolution_n + 2 * recovery_n
+        score += 2 * long_evolution_n + min(amplitude_mean, 8) + min(amplitude_rms, 10)
+        candidates.append({
+            "事件": event_id, "模式": items[0]["模式"], "测点": point,
+            "角度范围": f"{angle}-{angle + 10}°", "故障主方向": direction,
+            "支持指标数": len(items), "特征族数": len(families),
+            "演化指标数": evolution_n, "恢复指标数": recovery_n,
+            "长期稳定恢复指标数": stable_recovery_n,
+            "长期恢复比例平均": round(float(np.mean(recovery_ratios)), 3) if recovery_ratios else "",
+            "长期演化指标数": long_evolution_n,
+            "局部幅度平均Z": round(amplitude_mean, 3),
+            "局部幅度中位Z": round(amplitude_median, 3),
+            "局部幅度综合RMS": round(amplitude_rms, 3),
+            "候选评分": round(score, 3),
+            "最早持续日期": min((item["首次持续日期"] for item in items if item["首次持续日期"]), default=""),
+            "指标摘要": ";".join(
+                f"{item['指标中文']}{'↑' if item['故障方向']=='升高' else '↓'}"
+                f"(局部{item['维修前末段局部残差Z']:.1f}Z,ρ={item['故障前Spearman']:.2f},恢复={item['维修后恢复']})"
+                for item in sorted(items, key=lambda item: -item["单指标评分"])
+            ),
+            "可信等级": "高" if items[0]["模式"] == "连续运行段" and len(families) >= 2 and evolution_n >= 2 and recovery_n >= 2 and amplitude_mean >= 2
+                         else "中" if items[0]["模式"] == "连续运行段" and len(families) >= 2 and evolution_n >= 2 and recovery_n >= 1
+                         else "短期线索" if items[0]["模式"] == "短运行段" and len(families) >= 2 and evolution_n >= 2 and recovery_n >= 1 and amplitude_mean >= 2
+                         else "跨段线索" if items[0]["模式"] == "跨运行段" and len(families) >= 2 and long_evolution_n >= 2 and amplitude_mean >= 2
+                         else "低",
+        })
+    candidates.sort(key=lambda row: (row["事件"], -row["候选评分"]))
+    with candidate_output.open("w", encoding="utf-8-sig", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(candidates[0]))
+        writer.writeheader(); writer.writerows(candidates)
+
+    bands = merge_bands(candidates)
+    with band_output.open("w", encoding="utf-8-sig", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(bands[0]))
+        writer.writeheader(); writer.writerows(bands)
+
+    repeats = []
+    by_band = defaultdict(list)
+    for row in candidates:
+        if row["可信等级"] in ("高", "中", "短期线索", "跨段线索"):
+            by_band[row["角度范围"]].append(row)
+    for band, items in by_band.items():
+        events = sorted({item["事件"] for item in items})
+        if len(events) < 2:
+            continue
+        repeats.append({
+            "角度范围": band, "事件数": len(events), "事件": ";".join(events),
+            "测点": ";".join(sorted({item["测点"] for item in items})),
+            "最高评分": max(item["候选评分"] for item in items),
+            "共同特征": ";".join(item["指标摘要"] for item in sorted(items, key=lambda row: -row["候选评分"])[:4]),
+        })
+    repeats.sort(key=lambda row: (-row["事件数"], -row["最高评分"]))
+    with repeat_output.open("w", encoding="utf-8-sig", newline="") as handle:
+        fields = ["角度范围", "事件数", "事件", "测点", "最高评分", "共同特征"]
+        writer = csv.DictWriter(handle, fieldnames=fields); writer.writeheader(); writer.writerows(repeats)
+    return metric_candidates, candidates, repeats, bands
+
+
+def write_report(path: Path, metric_candidates, candidates, repeats, bands):
+    lines = ["# 局部10度特征演化分析", "", "## 方法", "",
+             "- 不使用整周期欧氏距离。每测点每天均匀抽取最多12个周期。",
+             "- 每周期先平滑并按吸排气平台归一化，再在原始键相角度上划分36个10度箱。",
+             "- 核心特征：局部中位压力、斜率、标准差、四分位距、导数RMS、最大二阶导。",
+             "- 同时使用运行段早期基线和维修后稳定基线；逐日结果扣除全机同角度、同指标的公共变化。",
+             "- 候选强调故障前定向演化、局部残差和独立维修后日期的反向恢复。",
+             "- 幅度评估不增加原始特征：对支持指标的定向局部残差计算平均值、中位数和综合RMS。",
+             "- 恢复分为事件后首个验证段的短期恢复，以及所有后续运行段都满足的长期稳定恢复。", ""]
+    for event in EVENTS:
+        rows = [row for row in candidates if row["事件"] == event["id"]]
+        high = [row for row in rows if row["可信等级"] == "高"]
+        medium = [row for row in rows if row["可信等级"] in ("中", "短期线索", "跨段线索")]
+        lines += [f"## {event['label']}", "", f"- 分析模式：{event['mode']}。", f"- 高可信候选：{len(high)}；其他可复核线索：{len(medium)}。"]
+        if event["id"] == "7号_2025维修过程":
+            lines.append("- 维修后基线仅2个日期且中间长期缺测，只能解释为跨段状态线索，不能证明连续故障前演化。")
+        for index, row in enumerate((high + medium)[:12], 1):
+            lines.append(
+                f"{index}. {row['测点']} {row['角度范围']}，{row['故障主方向']}，评分{row['候选评分']:.1f}，"
+                f"{row['可信等级']}可信；幅度平均{row['局部幅度平均Z']:.1f}Z，综合RMS{row['局部幅度综合RMS']:.1f}，"
+                f"近期演化{row['演化指标数']}项，长期跨段演化{row['长期演化指标数']}项，"
+                f"长期稳定恢复{row['长期稳定恢复指标数']}项；{row['指标摘要']}"
+            )
+        event_bands = [row for row in bands if row["事件"] == event["id"]]
+        if event_bands:
+            lines.append("- 相邻角度合并后的主要变化带：" + "；".join(
+                f"{row['测点']} {row['角度带']}({row['故障主方向']}，平均{row['局部幅度平均Z']:.1f}Z，RMS{row['局部幅度综合RMS']:.1f})"
+                for row in event_bands[:10]
+            ) + "。")
+        if not high and not medium:
+            lines.append("- 未找到同时满足局部残差、演化或维修后恢复条件的中高可信候选。")
+        lines.append("")
+    lines += ["## 跨事件重复角度", ""]
+    if repeats:
+        for row in repeats[:20]:
+            lines.append(f"- {row['角度范围']}：{row['事件数']}个事件；{row['事件']}；涉及{row['测点']}。")
+    else:
+        lines.append("- 暂无完全相同10度箱在至少两个事件中形成中高可信候选。")
+    lines += ["", "## 限制", "", "- 角度为键相原始坐标，没有自动相位旋转；不同测点的物理过程区间可能不同。",
+              "- 基线日期较少时稳健尺度仍可能不稳定，结果保留原始方向、日数和独立恢复验证供后续复核。",
+              "- 当前候选是维修相关局部特征，不自动等于通用气阀故障规则。"]
+    path.write_text("\n".join(lines) + "\n", encoding="utf-8")
+
+
+def main():
+    parser = argparse.ArgumentParser(description="10度局部特征双基线演化分析")
+    parser.add_argument("--out-dir", default="cache/indicator/局部演化分析")
+    parser.add_argument("--per-day", type=int, default=12)
+    parser.add_argument("--reuse", action="store_true", help="复用已抽取的逐周期CSV")
+    args = parser.parse_args()
+    out = Path(args.out_dir); out.mkdir(parents=True, exist_ok=True)
+    cycle_path = out / "局部特征_逐周期.csv"
+    grouped = load_extracted(cycle_path) if args.reuse and cycle_path.exists() else extract(cycle_path, args.per_day)
+    metric, candidates, repeats, bands = analyze(
+        grouped, out / "局部特征_逐日.csv", out / "局部演化候选.csv", out / "跨事件重复角度.csv",
+        out / "局部演化角度带.csv", out / "局部特征_逐运行段.csv",
+    )
+    write_report(out / "局部演化结论.md", metric, candidates, repeats, bands)
+    print(f"逐日组合 {len(grouped):,}，单指标候选 {len(metric):,}，角度候选 {len(candidates):,}，角度带 {len(bands):,}，重复角度 {len(repeats):,}")
+    print(f"结果 -> {out}")
+
+
+if __name__ == "__main__":
+    main()

backend/indicator_features/make_trend_dashboard.py

@@ -0,0 +1,165 @@
+#!/usr/bin/env python3
+"""Render actual sampled waveforms for selected events without changing the original page."""
+from __future__ import annotations
+
+import argparse
+import csv
+from datetime import datetime, timedelta
+from pathlib import Path
+
+import numpy as np
+
+from make_event_dashboard import html
+from extract_from_db import connect, fetch_cycles
+from indicator_features.indicator_features import (
+    CYLINDER_BORE_MM, _plateau_levels, smooth, volume_from_angle,
+)
+from new_unit_events import collect_events
+
+EVENTS = [
+    ("7号 2026-05-21", "7", "2026-05-21", "高密度特征_7号_2026.csv", 30, 30),
+    ("8号 2026-03-29/04-01 合并事件", "8", "2026-03-30", "高密度特征_8号_2026.csv", 30, 30),
+    ("8号 2025同期（后段延长至60天）", "8", "2025-03-30", "高密度特征_8号_2025同期.csv", 30, 60),
+    ("7号 2025-08-08（7月前段 / 11月后段）", "7", "2025-08-08", "高密度特征_7号_2025气阀事件.csv", 30, 110),
+    ("7号 2025-10-07（7月前段 / 11月后段）", "7", "2025-10-07", "高密度特征_7号_2025气阀事件.csv", 90, 50),
+    ("7号 2025停机维修合并过程（4~7月回溯 / 11月恢复）", "7", "2025-07-28", "高密度特征_7号_2025维修前回溯.csv", 120, 120),
+]
+BAD = {("7", "四", "盖侧"), ("7", "六", "盖侧"), ("8", "六", "盖侧")}
+
+
+def build_data(root: Path):
+    bins = 90
+    data = {"angle": (np.arange(bins) * 360 / bins).tolist(), "machines": {}}
+    connection = connect()
+    cycle_cache = {}
+    try:
+        for label, unit, date, filename, pre_days, post_days in EVENTS:
+            t0 = datetime.fromisoformat(date)
+            with (root / filename).open(encoding="utf-8-sig", newline="") as handle:
+                rows = list(csv.DictReader(handle))
+            if filename in ("高密度特征_7号_2025维修前回溯.csv", "高密度特征_7号_2025气阀事件.csv"):
+                with (root / "高密度特征_7号_2025事件后11月.csv").open(encoding="utf-8-sig", newline="") as handle:
+                    rows.extend(csv.DictReader(handle))
+            rows = [r for r in rows if
+                    t0 - timedelta(days=pre_days) <= datetime.fromisoformat(r["sample_time"]) < t0 + timedelta(days=post_days + 1)
+                    and float(r.get("p_span") or 0) >= 0.3
+                    and (unit, r["cylinder"], r["side"]) not in BAD]
+            points = sorted({(r["cylinder"], r["side"]) for r in rows})
+            # Select one cylinder per event. A dropdown exposes both chambers without
+            # presenting unrelated cylinders as peers or duplicate fault samples.
+            cylinder = "一" if any(c == "一" for c, _ in points) else "三"
+            chosen = [r for r in rows if r["cylinder"] == cylinder]
+            if filename == "高密度特征_7号_2025维修前回溯.csv":
+                chosen = [r for r in rows if (r["cylinder"], r["side"]) in
+                          {("三", "盖侧"), ("三", "轴侧"), ("四", "轴侧"), ("五", "盖侧") }]
+            ids = [int(r["wave_file_id"]) for r in chosen]
+            missing = [i for i in ids if i not in cycle_cache]
+            if missing:
+                cycle_cache.update(fetch_cycles(connection, missing, 100))
+            event = {"label": label, "sides": {}}
+            for cylinder, side in sorted({(r["cylinder"], r["side"]) for r in chosen}):
+                samples = []
+                for row in chosen:
+                    if (row["cylinder"], row["side"]) != (cylinder, side):
+                        continue
+                    signal = cycle_cache.get(int(row["wave_file_id"]))
+                    if signal is None or len(signal) < 60 or not np.all(np.isfinite(signal)):
+                        continue
+                    p = smooth(signal, 21)
+                    suction, discharge, _ = _plateau_levels(p)
+                    span = discharge - suction
+                    if span < 0.3:
+                        continue
+                    indices = np.arange(bins) * len(p) / bins
+                    absolute = np.interp(indices, np.arange(len(p)), p)
+                    profile = (absolute - suction) / span
+                    volume = np.interp(indices, np.arange(len(p)), volume_from_angle(len(p), CYLINDER_BORE_MM[cylinder]))
+                    time = datetime.fromisoformat(row["sample_time"])
+                    samples.append({
+                        "date": row["sample_time"], "day": (time.date() - t0.date()).days,
+                        "phase": "修前" if time < t0 else "修后",
+                        "profile": profile.round(6).tolist(), "absolute": absolute.round(6).tolist(),
+                        "volume": volume.round(6).tolist(),
+                    })
+                samples.sort(key=lambda s: s["date"])
+                if not samples:
+                    continue
+                pre = [s for s in samples if s["phase"] == "修前"]
+                post = [s for s in samples if s["phase"] == "修后"]
+                reference = pre or post
+                if filename == "高密度特征_7号_2025维修前回溯.csv":
+                    reference = [s for s in samples if s["date"][:7] == "2025-04"] or reference
+                template = np.median([s["profile"] for s in reference], axis=0)
+                for sample in samples:
+                    sample["distance"] = round(float(np.linalg.norm(np.asarray(sample["profile"]) - template)), 5)
+
+                def median(items, key):
+                    return np.median([s[key] for s in items], axis=0).round(6).tolist() if items else []
+
+                event["sides"][f"{cylinder}缸{side}"] = {
+                    "samples": samples, "counts": {"pre": len(pre), "post": len(post)},
+                    "preMedian": median(pre, "profile"), "postMedian": median(post, "profile"),
+                    "preAbsoluteMedian": median(pre, "absolute"), "postAbsoluteMedian": median(post, "absolute"),
+                    "trend": [{"date": s["date"], "day": s["day"], "distance": s["distance"]} for s in samples],
+                }
+                if filename == "高密度特征_7号_2025维修前回溯.csv":
+                    # Compare the last running period with restart, but measure the
+                    # lookback trend against the earliest available April period.
+                    july = [s for s in samples if s["date"][:7] == "2025-07"]
+                    event["sides"][f"{cylinder}缸{side}"]["preMedian"] = median(july, "profile")
+                    event["sides"][f"{cylinder}缸{side}"]["preAbsoluteMedian"] = median(july, "absolute")
+            coverage = sorted({s["date"][:10] for d in event["sides"].values() for s in d["samples"]})
+            if coverage:
+                event["label"] += f" · 实际功图 {coverage[0]} 至 {coverage[-1]}"
+            data["machines"][label] = event
+        data["machines"].update(collect_events(connection, per_day=6, bins=bins))
+    finally:
+        connection.close()
+    return data
+
+
+def make_html(data):
+    page = html(data)
+    # Reuse the style and actual-waveform plots without editing the source dashboard.
+    page = page.replace("一缸 P-V 功图事件前后对比", "事件功图扩展对比")
+    page = page.replace("图例在图外，可逐时间点勾选", "图例在图外，按四类勾选")
+    page = page.replace("P-V图例可按时间点单独勾选", "功图按前段/后段及中位线分组勾选")
+    page = page.replace("<div id=\"app\"></div>",
+                        '<label>选择事件：<select id="event-select"></select></label>'
+                        '<p class="note">每次只展示一个选项。两个旧选项保留7月与11月的三缸两侧对照，'
+                        '使用7月模板计算形状距离；10月选项的前段实际也是7月，不是10月采样。'
+                        '两个旧选项共用同一批周期，仅横轴相对日期不同，不能作为独立维修证据。'
+                        '7号2025年的维修合并为一次停机过程，7月28日仅作末次运行后的时间分界，不是维修日期。'
+                        '该过程粗蓝线为7月末运行段中位，红线为11月恢复段中位；细蓝线包含4~7月回溯周期。'
+                        '回溯选项的形状距离以4月模板为参照，其他选项仍以事件前模板为参照，数值不可混比。'
+                        '新增4号和6号事件先扫描全部有效测点；本页保留全部测点，候选排名按扣除全机共同变化后的局部残差计算。'
+                        '6号4月22日至26日视为同一维修过程，过程内周期不进入前后模板。'
+                        '5号因当前数据库没有压力功图而暂不展示。展示测点并非已确认异常缸，缺测期间不推断变化发生时刻。</p>'
+                        '<div id="app"></div>')
+    page = page.replace("<script>renderPage(DATA);</script>", """<script>
+renderPage(DATA);
+const sections=Array.from(document.querySelectorAll('.machine'));
+const selector=document.getElementById('event-select');
+Object.keys(DATA.machines).forEach((name,index)=>{const option=document.createElement('option');option.value=index;option.textContent=name;selector.append(option)});
+function selectEvent(){sections.forEach((section,index)=>section.hidden=index!==Number(selector.value));requestAnimationFrame(()=>charts.forEach(chart=>chart.resize()))}
+selector.addEventListener('change',selectEvent);selectEvent();
+</script>""")
+    return page
+
+
+def main():
+    parser = argparse.ArgumentParser(description="生成实际周期的事件功图扩展网页")
+    parser.add_argument("--root", default="cache/indicator")
+    parser.add_argument("--out", required=True)
+    args = parser.parse_args()
+    data = build_data(Path(args.root))
+    out = Path(args.out)
+    out.parent.mkdir(parents=True, exist_ok=True)
+    out.write_text(make_html(data), encoding="utf-8")
+    print(f"对比网页 -> {out}")
+    for event in data["machines"].values():
+        print(event["label"], {point: d["counts"] for point, d in event["sides"].items()})
+
+
+if __name__ == "__main__":
+    main()

backend/indicator_features/new_unit_events.py

@@ -0,0 +1,147 @@
+"""Sample and rank 4/6 unit valve events without scanning wave_sample_one."""
+from __future__ import annotations
+
+from collections import defaultdict
+from datetime import datetime, timedelta
+
+import numpy as np
+
+from extract_from_db import fetch_cycles, fetch_file_rows, point_name, sample_per_day
+from indicator_features.indicator_features import CYLINDER_BORE_MM, _plateau_levels, smooth, volume_from_angle
+
+CYLINDERS = ("一", "二", "三", "四", "五", "六")
+SIDES = ("盖侧", "轴侧")
+BAD = {
+    "4": {("二", "盖侧"), ("四", "盖侧"), ("六", "盖侧")},
+    "6": {("二", "盖侧"), ("四", "盖侧"), ("六", "盖侧")},
+}
+EVENTS = [
+    {
+        "key": "4号 2026-06-10 气阀事件",
+        "unit": "4", "center": "2026-06-10", "pre_end": "2026-06-10", "post_start": "2026-06-10",
+        "start": "2026-05-11", "end": "2026-07-11", "markers": ["2026-06-10"],
+    },
+    {
+        "key": "6号 2026-04-22~04-26 气阀维修过程",
+        "unit": "6", "center": "2026-04-24", "pre_end": "2026-04-22", "post_start": "2026-04-26",
+        "start": "2026-03-23", "end": "2026-05-27", "markers": ["2026-04-22", "2026-04-26"],
+    },
+]
+
+
+def _median(items, key):
+    return np.median([item[key] for item in items], axis=0).round(6).tolist() if items else []
+
+
+def collect_events(connection, per_day=6, bins=90):
+    result = {}
+    for config in EVENTS:
+        center = datetime.fromisoformat(config["center"])
+        pre_end = datetime.fromisoformat(config["pre_end"])
+        post_start = datetime.fromisoformat(config["post_start"])
+        points = {}
+        for cylinder in CYLINDERS:
+            for side in SIDES:
+                if (cylinder, side) in BAD[config["unit"]]:
+                    continue
+                files = fetch_file_rows(
+                    connection, point_name(config["unit"], cylinder, side), config["start"], config["end"]
+                )
+                picked = sample_per_day(files, per_day)
+                cycles = fetch_cycles(connection, [int(row["id"]) for row in picked], 100) if picked else {}
+                samples = []
+                for row in picked:
+                    signal = cycles.get(int(row["id"]))
+                    if signal is None or len(signal) < 60 or not np.all(np.isfinite(signal)):
+                        continue
+                    pressure = smooth(signal, 21)
+                    suction, discharge, _ = _plateau_levels(pressure)
+                    span = discharge - suction
+                    if not np.isfinite(span) or span < .3:
+                        continue
+                    indices = np.arange(bins) * len(pressure) / bins
+                    absolute = np.interp(indices, np.arange(len(pressure)), pressure)
+                    profile = (absolute - suction) / span
+                    volume = np.interp(
+                        indices, np.arange(len(pressure)),
+                        volume_from_angle(len(pressure), CYLINDER_BORE_MM[cylinder]),
+                    )
+                    time = row["sample_time"]
+                    if not isinstance(time, datetime):
+                        time = datetime.fromisoformat(str(time))
+                    if time < pre_end:
+                        phase = "修前"
+                    elif time >= post_start:
+                        phase = "修后"
+                    else:
+                        phase = "过程内"
+                    samples.append({
+                        "date": time.strftime("%Y-%m-%d %H:%M:%S"),
+                        "day": (time.date() - center.date()).days,
+                        "phase": phase,
+                        "profile": profile.round(6).tolist(),
+                        "absolute": absolute.round(6).tolist(),
+                        "volume": volume.round(6).tolist(),
+                    })
+                samples.sort(key=lambda sample: sample["date"])
+                pre = [sample for sample in samples if sample["phase"] == "修前"]
+                post = [sample for sample in samples if sample["phase"] == "修后"]
+                if not pre or not post:
+                    continue
+                before = np.median([sample["profile"] for sample in pre], axis=0)
+                after = np.median([sample["profile"] for sample in post], axis=0)
+                points[(cylinder, side)] = {
+                    "samples": samples, "pre": pre, "post": post,
+                    "before": before, "after": after, "delta": after - before,
+                }
+
+        common = np.median([point["delta"] for point in points.values()], axis=0)
+        cylinder_scores = defaultdict(list)
+        for (cylinder, _), point in points.items():
+            point["total"] = float(np.linalg.norm(point["delta"]))
+            point["residual"] = float(np.linalg.norm(point["delta"] - common))
+            cylinder_scores[cylinder].append(point["residual"])
+        ranking = sorted(
+            ((cylinder, max(scores)) for cylinder, scores in cylinder_scores.items()),
+            key=lambda item: item[1], reverse=True,
+        )
+        event = {
+            "label": config["key"], "unit": config["unit"], "markers": config["markers"],
+            "ranking": [{"cylinder": cylinder, "residual": round(score, 5)} for cylinder, score in ranking],
+            "sides": {},
+        }
+        for (cylinder, side), point in points.items():
+            template = point["before"]
+            for sample in point["samples"]:
+                sample["distance"] = round(float(np.linalg.norm(np.asarray(sample["profile"]) - template)), 5)
+            name = f"{cylinder}缸{side}"
+            event["sides"][name] = {
+                "samples": point["samples"],
+                "counts": {"pre": len(point["pre"]), "post": len(point["post"])},
+                "preMedian": point["before"].round(6).tolist(),
+                "postMedian": point["after"].round(6).tolist(),
+                "preAbsoluteMedian": _median(point["pre"], "absolute"),
+                "postAbsoluteMedian": _median(point["post"], "absolute"),
+                "trend": [{"date": sample["date"], "day": sample["day"], "distance": sample["distance"]}
+                          for sample in point["samples"]],
+                "totalChange": round(point["total"], 5),
+                "localResidual": round(point["residual"], 5),
+                "markerDays": [
+                    (datetime.fromisoformat(marker).date() - center.date()).days
+                    for marker in config["markers"]
+                ],
+            }
+        coverage = sorted({sample["date"][:10] for point in event["sides"].values() for sample in point["samples"]})
+        event["label"] += f" · 实际功图 {coverage[0]} 至 {coverage[-1]}"
+        result[config["key"]] = event
+    return result
+
+
+def top_cylinder_event(event, limit=3):
+    cylinders = {item["cylinder"] for item in event["ranking"][:limit]}
+    selected = dict(event)
+    selected["sides"] = {
+        name: point for name, point in event["sides"].items()
+        if name[0] in cylinders
+    }
+    return selected

backend/indicator_features/pseudo_online_detection.py

@@ -0,0 +1,579 @@
+#!/usr/bin/env python3
+"""Pseudo-online local-feature detection with past-only baselines."""
+from __future__ import annotations
+
+import csv
+import argparse
+import math
+import sys
+from collections import defaultdict, deque
+from datetime import date, datetime, timedelta
+from pathlib import Path
+
+import numpy as np
+
+BACKEND = Path(__file__).resolve().parents[1]
+sys.path.insert(0, str(BACKEND))
+
+from indicator_features.extract_from_db import (  # noqa: E402
+    CYLINDERS, SIDES, connect, fetch_cycles, fetch_file_rows, point_name, sample_per_day,
+)
+from indicator_features.local_evolution_analysis import (  # noqa: E402
+    FAMILY, METRICS, METRIC_CN, normalized_bins, robust_scale,
+)
+
+BAD = {
+    "4": {("二", "盖侧"), ("四", "盖侧"), ("六", "盖侧")},
+    "6": {("二", "盖侧"), ("四", "盖侧"), ("六", "盖侧")},
+    "7": {("四", "盖侧"), ("六", "盖侧")},
+}
+FAILURE_TEMPLATES = {
+    "7": {
+        ("一缸盖侧",240,1),("一缸盖侧",250,1),("一缸盖侧",260,1),
+        ("二缸盖侧",140,-1),("三缸轴侧",160,1),("三缸轴侧",170,1),
+        ("三缸盖侧",20,1),("三缸盖侧",30,1),("三缸盖侧",310,-1),("三缸盖侧",320,-1),
+    },
+    "4": {("四缸轴侧",60,-1)},
+    "6": {
+        ("三缸轴侧",160,1),("三缸轴侧",170,1),("三缸轴侧",120,1),("三缸轴侧",130,1),
+        ("一缸轴侧",290,1),("一缸盖侧",220,-1),("三缸轴侧",50,1),
+    },
+}
+EVENTS = [
+    {"id": "7号_2025维修过程", "unit": "7", "event": "2025-08-08",
+     "baseline": ("2025-04-29", "2025-06-14"), "detect": ("2025-07-25", "2025-07-28"),
+     "note": "事件附近缺测，仅验证7月底运行段状态"},
+    {"id": "7号_2026-05-21", "unit": "7", "event": "2026-05-21",
+     "baseline": ("2026-02-10", "2026-04-05"), "detect": ("2026-05-06", "2026-05-22"), "note": ""},
+    {"id": "4号_2026-06-10", "unit": "4", "event": "2026-06-10",
+     "baseline": ("2026-04-21", "2026-05-31"), "detect": ("2026-06-06", "2026-06-11"), "note": ""},
+    {"id": "6号_2026-04-22_26", "unit": "6", "event": "2026-04-22",
+     "baseline": ("2026-04-12", "2026-04-15"), "detect": ("2026-04-15", "2026-04-27"),
+     "note": "历史不足，仅3个基线日期"},
+]
+UNITS = ("4", "6", "7", "8", "9")
+MERGE_GAP = 2
+HEALTH_SEGMENTS = {
+    "7": [("2026-03-12", "2026-04-05")],
+    "8": [("2026-04-18", "2026-05-07")],
+    "9": [("2026-04-01", "2026-04-19")],
+}
+
+
+def calendar_segments(days):
+    segments=[]
+    for day in sorted(days):
+        if segments and (day-segments[-1][1]).days <= MERGE_GAP+1:
+            segments[-1][1]=day
+        else:
+            segments.append([day,day])
+    return [(start,end) for start,end in segments]
+
+
+def date_ranges(event):
+    start,end=event["detect"]
+    cursor=date.fromisoformat(start);stop=date.fromisoformat(end)
+    while cursor<stop:
+        yield cursor.isoformat()
+        cursor+=timedelta(days=1)
+
+
+def process_unit(connection, unit, start, end, per_day_baseline=3):
+    points={}
+    running_days=set()
+    for cylinder in CYLINDERS:
+        for side in SIDES:
+            if (cylinder,side) in BAD.get(unit,set()):continue
+            files=fetch_file_rows(connection,point_name(unit,cylinder,side),start,end)
+            running_days.update(row["sample_time"].date() for row in files if row.get("rpm") and row["rpm"]>0)
+            baseline_files=sample_per_day(files,per_day_baseline)
+            cycles=fetch_cycles(connection,[int(row["id"]) for row in baseline_files],100) if baseline_files else {}
+            baseline_records=[]
+            for row in baseline_files:
+                values=stats_for_cycle(cycles.get(int(row["id"]),np.array([])))
+                if values is None:continue
+                time=row["sample_time"] if isinstance(row["sample_time"],datetime) else datetime.fromisoformat(str(row["sample_time"]))
+                baseline_records.append({"time":time,"day":time.date().isoformat(),"values":values})
+            if baseline_records:
+                points[f"{cylinder}缸{side}"]={"baseline_records":baseline_records,"detection_records":[]}
+    return points,calendar_segments(running_days)
+
+
+def stats_for_cycle(signal):
+    bins = normalized_bins(signal)
+    if bins is None:
+        return None
+    return {(item["bin_start"], metric): item[metric] for item in bins for metric in METRICS}
+
+
+def point_rows(connection, event, cylinder, side, baseline_per_day=3):
+    point = point_name(event["unit"], cylinder, side)
+    baseline_files = sample_per_day(
+        fetch_file_rows(connection, point, event["baseline"][0], event["baseline"][1]), baseline_per_day
+    )
+    detection_files = fetch_file_rows(connection, point, event["detect"][0], event["detect"][1])
+    cycles = fetch_cycles(connection, [int(row["id"]) for row in baseline_files], 100) if baseline_files else {}
+    baseline, detection_meta = [], []
+    baseline_ids = {int(row["id"]) for row in baseline_files}
+    for row in baseline_files + detection_files:
+        file_id = int(row["id"])
+        time = row["sample_time"] if isinstance(row["sample_time"], datetime) else datetime.fromisoformat(str(row["sample_time"]))
+        if file_id in baseline_ids:
+            values = stats_for_cycle(cycles.get(file_id, np.array([])))
+            if values is None: continue
+            baseline.append({"time":time,"day":time.date().isoformat(),"slot":time.hour//6,"values":values})
+        else:
+            detection_meta.append({"id":file_id,"time":time,"day":time.date().isoformat(),"slot":time.hour//6})
+    return baseline, detection_meta
+
+
+def build_baseline(records):
+    per_key_day = defaultdict(list)
+    per_key_cycles = defaultdict(list)
+    for record in records:
+        for key, value in record["values"].items():
+            per_key_day[(key, record["day"])].append(value)
+            per_key_cycles[key].append(value)
+    daily = defaultdict(list)
+    for (key, _), values in per_key_day.items():
+        daily[key].append(float(np.median(values)))
+    provisional = {}
+    for key in [(angle, metric) for angle in range(0, 360, 10) for metric in METRICS]:
+        days = daily.get(key, [])
+        cycles = per_key_cycles.get(key, [])
+        provisional[key] = {
+            "center": float(np.median(days)) if days else float("nan"),
+            "scale": max(robust_scale(days), robust_scale(cycles)),
+            "days": len(days), "cycles": len(cycles),
+        }
+    for metric in METRICS:
+        scales = [provisional[(angle, metric)]["scale"] for angle in range(0, 360, 10)
+                  if provisional[(angle, metric)]["scale"] > 1e-12]
+        floor = .2 * float(np.median(scales)) if scales else 1e-9
+        for angle in range(0, 360, 10):
+            provisional[(angle, metric)]["scale"] = max(provisional[(angle, metric)]["scale"], floor)
+    return provisional
+
+
+def aggregate(records, group_field):
+    grouped = defaultdict(list)
+    for record in records:
+        group = record[group_field] if group_field != "day_slot" else f"{record['day']}#{record['slot']}"
+        for key, value in record["values"].items():
+            grouped[(group, key)].append(value)
+    return {(group, key): float(np.median(values)) for (group, key), values in grouped.items()}
+
+
+def merge_bands(alerts):
+    groups = defaultdict(list)
+    for row in alerts:
+        groups[(row["事件"], row["日期"], row["测点"], row["等级"])].append(row)
+    output = []
+    for key, rows in groups.items():
+        rows.sort(key=lambda row: int(row["角度箱"]))
+        runs = []
+        for row in rows:
+            angle = int(row["角度箱"])
+            if runs and angle == int(runs[-1][-1]["角度箱"]) + 10:
+                runs[-1].append(row)
+            else:
+                runs.append([row])
+        if len(runs) > 1 and int(runs[0][0]["角度箱"]) == 0 and int(runs[-1][-1]["角度箱"]) == 350:
+            runs[0] = runs[-1] + runs[0]; runs.pop()
+        for run in runs:
+            starts = [int(row["角度箱"]) for row in run]
+            band = f"{starts[0]}-360°及0-{starts[-1]+10}°" if starts[0] > starts[-1] else f"{starts[0]}-{starts[-1]+10}°"
+            best = max(run, key=lambda row: float(row["综合RMS"]))
+            output.append({
+                "事件": key[0], "日期": key[1], "测点": key[2], "等级": key[3], "角度带": band,
+                "角度箱数": len(run), "最大支持特征数": max(int(row["支持特征数"]) for row in run),
+                "最大特征族数": max(int(row["支持特征族数"]) for row in run),
+                "最大幅度平均Z": max(float(row["幅度平均Z"]) for row in run),
+                "最大综合RMS": max(float(row["综合RMS"]) for row in run),
+                "最大异常子窗数": max(int(row["异常6小时子窗数"]) for row in run),
+                "代表特征": best["特征摘要"],
+            })
+    return output
+
+
+def run(out_dir: Path, selected_ids=None, slow_channel=False):
+    all_detail, all_bands, validations = [], [], []
+    reuse_dir=Path(str(out_dir)+"_reuse")
+    reuse_dir.mkdir(parents=True,exist_ok=True)
+    connection = connect()
+    try:
+        for event in EVENTS:
+            if selected_ids and event["id"] not in selected_ids:
+                continue
+            point_data = {}
+            for cylinder in CYLINDERS:
+                for side in SIDES:
+                    if (cylinder, side) in BAD[event["unit"]]:
+                        continue
+                    baseline, detection_meta = point_rows(connection, event, cylinder, side)
+                    if baseline and detection_meta:
+                        point_data[f"{cylinder}缸{side}"]={"baseline":build_baseline(baseline),"baseline_records":baseline,
+                                                               "detection_meta":detection_meta}
+            early_baselines={}
+            for point,data in point_data.items():
+                early_days=sorted({record["day"] for record in data["baseline_records"]})[:3]
+                early_records=[record for record in data["baseline_records"] if record["day"] in early_days]
+                early_baselines[point]=build_baseline(early_records) if early_records else {}
+            detection_days=sorted({record["day"] for data in point_data.values() for record in data["detection_meta"]})
+            history=defaultdict(deque)
+            slow_history=defaultdict(deque)
+            event_alerts=[]
+            for day in detection_days:
+                raw_by_point={}
+                slot_z=defaultdict(dict)
+                for point,data in point_data.items():
+                    meta=[record for record in data["detection_meta"] if record["day"]==day]
+                    ids=[record["id"] for record in meta]
+                    cycles=fetch_cycles(connection,ids,100) if ids else {}
+                    records=[]
+                    for record in meta:
+                        values=stats_for_cycle(cycles.get(record["id"],np.array([])))
+                        if values is not None:
+                            records.append({**record,"values":values})
+                    if not records:continue
+                    window_z={}
+                    for angle in range(0,360,10):
+                        for metric in METRICS:
+                            key=(angle,metric);base=data["baseline"][key]
+                            if not np.isfinite(base["center"]):continue
+                            vals=[rec["values"][key] for rec in records if key in rec["values"]]
+                            if not vals:continue
+                            window_z[key]=(float(np.median(vals))-base["center"])/base["scale"]
+                            for slot in range(4):
+                                slot_vals=[rec["values"][key] for rec in records if rec["slot"]==slot and key in rec["values"]]
+                                if slot_vals:slot_z[(slot,key)][point]= (float(np.median(slot_vals))-base["center"])/base["scale"]
+                    raw_by_point[point]=window_z
+                common={}
+                for (slot,key),values in slot_z.items():common[(slot,key)]=float(np.median(list(values.values())))
+                common24={}
+                for key in {(a,m) for point in raw_by_point for a,m in raw_by_point[point]}:
+                    vals=[raw_by_point[point][key] for point in raw_by_point if key in raw_by_point[point]]
+                    common24[key]=float(np.median(vals)) if vals else 0.0
+                by_candidate=defaultdict(list)
+                for point,window_z in raw_by_point.items():
+                    for (angle,metric),z in window_z.items():
+                        row={"point":point,"angle":angle,"metric":metric,"z":z,"local":z-common24[(angle,metric)],"base":point_data[point]["baseline"][(angle,metric)]}
+                        by_candidate[(point,angle)].append(row)
+                abnormal_bins_by_point=defaultdict(set)
+                for (point,angle),rows in by_candidate.items():
+                    if any(abs(row["local"])>=6 for row in rows):
+                        abnormal_bins_by_point[point].add(angle)
+                channel_fault={point for point,bins in abnormal_bins_by_point.items() if len(bins)>=18}
+                for (point,angle),rows in sorted(by_candidate.items(),key=lambda item:(item[0][0],item[0][1])):
+                    supported = [row for row in rows if abs(row["local"]) >= 2]
+                    families = {FAMILY[row["metric"]] for row in supported}
+                    amplitudes = [abs(row["local"]) for row in supported]
+                    mean_amp = float(np.mean(amplitudes)) if amplitudes else 0.0
+                    rms = float(np.sqrt(np.mean(np.square(amplitudes)))) if amplitudes else 0.0
+                    abnormal_slots=0
+                    for slot in range(4):
+                        slot_values=[]
+                        for row in rows:
+                            slot_key=(slot,(angle,row["metric"]))
+                            if point in slot_z.get(slot_key,{}):
+                                slot_values.append(abs(slot_z[slot_key][point]-common.get(slot_key,0.0)))
+                        if sum(value>=2 for value in slot_values)>=2:abnormal_slots+=1
+                    candidate = len(families) >= 2 and mean_amp >= 2 and rms >= 3 and abnormal_slots >= 2
+                    hist = history[(point, angle)]; hist.append(candidate)
+                    if len(hist) > 5: hist.popleft()
+                    persistent = len(hist) >= 3 and sum(hist) >= 3
+
+                    slow_rows=[]
+                    for row in rows:
+                        metric=row["metric"];key=(angle,metric)
+                        early=early_baselines[point].get(key)
+                        if not early or not np.isfinite(early["center"]):continue
+                        current_value=row["z"]*row["base"]["scale"]+row["base"]["center"]
+                        scale=max(early["scale"],row["base"]["scale"]*.2)
+                        slow_z=(current_value-early["center"])/scale
+                        slow_rows.append({**row,"slow_z":slow_z})
+                    slow_public_values=defaultdict(list)
+                    for other_point,other_rows in by_candidate.items():
+                        other_angle=other_point[1]
+                        if other_angle!=angle:continue
+                        for other_row in other_rows:
+                            other_early=early_baselines[other_point[0]].get((angle,other_row["metric"]))
+                            if not other_early or not np.isfinite(other_early["center"]):continue
+                            value=other_row["z"]*other_row["base"]["scale"]+other_row["base"]["center"]
+                            scale=max(other_early["scale"],other_row["base"]["scale"]*.2)
+                            slow_public_values[other_row["metric"]].append((value-other_early["center"])/scale)
+                    slow_supported_rows=[]
+                    for row in slow_rows:
+                        public=float(np.median(slow_public_values[row["metric"]])) if slow_public_values[row["metric"]] else 0.0
+                        row["slow_local"]=row["slow_z"]-public
+                        if abs(row["slow_local"])>=2:slow_supported_rows.append(row)
+                    slow_families={FAMILY[row["metric"]] for row in slow_supported_rows}
+                    slow_amplitudes=[abs(row["slow_local"]) for row in slow_supported_rows]
+                    slow_mean=float(np.mean(slow_amplitudes)) if slow_amplitudes else 0.0
+                    slow_rms=float(np.sqrt(np.mean(np.square(slow_amplitudes)))) if slow_amplitudes else 0.0
+                    slow_candidate=len(slow_families)>=2 and slow_mean>=2 and slow_rms>=3
+                    slow_hist=slow_history[(point,angle)];slow_hist.append(slow_candidate)
+                    if len(slow_hist)>5:slow_hist.popleft()
+                    slow_persistent=len(slow_hist)>=3 and sum(slow_hist)>=3
+                    raw_supported=supported or slow_supported_rows
+                    direction_values=[row["local"] for row in supported] or [row["slow_local"] for row in slow_supported_rows]
+                    direction=1 if direction_values and np.median(direction_values)>=0 else -1
+                    signature_match=any(tp==point and ta==angle and td==direction
+                                        for tp,ta,td in FAILURE_TEMPLATES.get(event["unit"],set()))
+                    candidate_level=(candidate or slow_candidate)
+                    level="通道质控" if point in channel_fault else "持续预警" if ((persistent and candidate) or (slow_persistent and slow_candidate)) else "候选" if candidate_level else "观察" if raw_supported else "正常"
+                    if level in ("候选","持续预警") and not signature_match:level="未分类异常"
+                    base_days=min(row["base"]["days"] for row in rows)
+                    detail={
+                        "事件":event["id"],"机组":event["unit"],"事件日期":event["event"],"日期":day,
+                        "测点":point,"角度箱":angle,"等级":level,
+                        "支持特征数":len(supported),"支持特征族数":len(families),
+                        "幅度平均Z":round(mean_amp,4),"幅度中位Z":round(float(np.median(amplitudes)) if amplitudes else 0,4),
+                        "综合RMS":round(rms,4),"异常6小时子窗数":abnormal_slots,
+                        "最近5窗候选数":sum(hist),"基线有效日数":base_days,
+                        "慢通道支持特征数":len(slow_supported_rows),"慢通道特征族数":len(slow_families),
+                        "慢通道幅度平均Z":round(slow_mean,4),"慢通道综合RMS":round(slow_rms,4),
+                        "慢通道最近5窗候选数":sum(slow_hist),"慢通道持续":int(slow_persistent),
+                        "历史事件签名匹配":int(signature_match),
+                        "基线置信":"低" if base_days<7 else "中" if base_days<20 else "高",
+                        "特征摘要":";".join(f"{METRIC_CN[row['metric']]}{'↑' if row['local']>0 else '↓'}{abs(row['local']):.1f}Z" for row in sorted(supported,key=lambda x:-abs(x['local']))),
+                        "慢通道特征摘要":";".join(f"{METRIC_CN[row['metric']]}{'↑' if row['slow_local']>0 else '↓'}{abs(row['slow_local']):.1f}Z" for row in sorted(slow_supported_rows,key=lambda x:-abs(x['slow_local']))),
+                        "备注":event["note"],
+                    }
+                    all_detail.append(detail)
+                    if level not in ("正常","通道质控"):event_alerts.append(detail)
+            bands = merge_bands(event_alerts); all_bands.extend(bands)
+            event_date = datetime.fromisoformat(event["event"]).date()
+            grouped_bands = defaultdict(list)
+            for row in bands: grouped_bands[(row["测点"], row["角度带"])].append(row)
+            for (point, band), rows in grouped_bands.items():
+                rows.sort(key=lambda row: row["日期"])
+                candidates = [row for row in rows if row["等级"] in ("候选", "持续预警", "未分类异常")]
+                persistent = [row for row in rows if row["等级"] == "持续预警"]
+                if not candidates: continue
+                first = datetime.fromisoformat(candidates[0]["日期"]).date()
+                validations.append({
+                    "事件": event["id"], "事件日期": event["event"], "测点": point, "角度带": band,
+                    "首次候选日期": str(first), "提前日数": (event_date-first).days,
+                    "首次持续预警日期": persistent[0]["日期"] if persistent else "",
+                    "候选窗口数": len(candidates), "持续预警窗口数": len(persistent),
+                    "未分类异常窗口数": sum(row["等级"] == "未分类异常" for row in candidates),
+                    "最大幅度平均Z": max(float(row["最大幅度平均Z"]) for row in candidates),
+                    "最大综合RMS": max(float(row["最大综合RMS"]) for row in candidates),
+                    "备注": event["note"],
+                })
+            if all_detail and any(row["事件"]==event["id"] for row in all_detail):
+                event_detail=[row for row in all_detail if row["事件"]==event["id"]]
+                event_band=[row for row in all_bands if row["事件"]==event["id"]]
+                event_validation=[row for row in validations if row["事件"]==event["id"]]
+                for filename,records in (("伪在线_窗口检测明细.csv",event_detail),
+                                         ("伪在线_角度带告警.csv",event_band),
+                                         ("伪在线_事件验证.csv",event_validation)):
+                    if records:
+                        with (reuse_dir/filename).open("w",encoding="utf-8-sig",newline="") as handle:
+                            writer=csv.DictWriter(handle,fieldnames=list(records[0]));writer.writeheader();writer.writerows(records)
+            print(event["id"], "测点", len(point_data), "明细", len([r for r in all_detail if r['事件']==event['id']]),
+                  "角度带", len(bands), flush=True)
+    finally:
+        connection.close()
+
+    out_dir.mkdir(parents=True, exist_ok=True)
+    def write(name, rows):
+        if not rows: return
+        with (out_dir/name).open("w", encoding="utf-8-sig", newline="") as handle:
+            writer=csv.DictWriter(handle,fieldnames=list(rows[0]));writer.writeheader();writer.writerows(rows)
+    write("伪在线_窗口检测明细.csv", all_detail)
+    write("伪在线_角度带告警.csv", all_bands)
+    write("伪在线_事件验证.csv", validations)
+
+    false_rows=[]
+    for event in EVENTS:
+        rows=[row for row in all_bands if row["事件"]==event["id"]]
+        total_days=len({row["日期"] for row in all_detail if row["事件"]==event["id"]})
+        for point in sorted({row["测点"] for row in rows}):
+            selected=[row for row in rows if row["测点"]==point]
+            false_rows.append({"事件":event["id"],"测点":point,"检测有效日数":total_days,
+                               "观察或以上角度带窗口":len(selected),"候选或以上窗口":sum(row["等级"] in ('候选','持续预警') for row in selected),
+                               "持续预警窗口":sum(row["等级"]=='持续预警' for row in selected)})
+    write("伪在线_误报统计.csv",false_rows)
+    return all_detail, all_bands, validations
+
+
+def run_health_segments(out_dir: Path, units=("7", "8", "9")):
+    """Replay unlabelled controls; triggers are not confirmed false positives."""
+    detail_rows=[]
+    connection=None
+    cache_dir=out_dir/"unlabelled_control_features_v1"
+    cache_dir.mkdir(parents=True,exist_ok=True)
+    keys=[(angle,metric) for angle in range(0,360,10) for metric in METRICS]
+    try:
+        for unit in units:
+            unit_detail_start=len(detail_rows)
+            for segment_start,segment_end in HEALTH_SEGMENTS.get(unit,[]):
+                point_data={}
+                for cylinder in CYLINDERS:
+                    for side in SIDES:
+                        if (cylinder,side) in BAD.get(unit,set()) or (unit=="8" and (cylinder,side)==("六","盖侧")):continue
+                        cache_path=cache_dir/f"{unit}_{cylinder}_{side}_{segment_start}_{segment_end}.npz"
+                        baseline=[];daily=[];slots=[];cycle_count=0
+                        if cache_path.exists():
+                            with np.load(cache_path,allow_pickle=False) as cached:
+                                if list(cached["metrics"])!=list(METRICS):
+                                    raise ValueError(f"Feature schema mismatch: {cache_path}")
+                                for kind,records in (("baseline",baseline),("daily",daily),("slots",slots)):
+                                    for day,slot,values in zip(cached[f"{kind}_days"],cached[f"{kind}_slots"],cached[f"{kind}_values"]):
+                                        records.append({"day":str(day),"slot":int(slot),"values":{key:float(value) for key,value in zip(keys,values) if np.isfinite(value)}})
+                                cycle_count=int(cached["cycle_count"].item())
+                        else:
+                            if connection is None:connection=connect()
+                            files=fetch_file_rows(connection,point_name(unit,cylinder,side),segment_start,segment_end)
+                            by_day=defaultdict(list)
+                            for row in files:
+                                time=row["sample_time"] if isinstance(row["sample_time"],datetime) else datetime.fromisoformat(str(row["sample_time"]))
+                                by_day[time.date().isoformat()].append({**row,"sample_time":time})
+                            valid_days=0
+                            for day,day_files in sorted(by_day.items()):
+                                records=[]
+                                day_files.sort(key=lambda row:(row["sample_time"],int(row["id"])))
+                                for offset in range(0,len(day_files),100):
+                                    batch=day_files[offset:offset+100]
+                                    cycles=fetch_cycles(connection,[int(row["id"]) for row in batch],100)
+                                    for row in batch:
+                                        values=stats_for_cycle(cycles.get(int(row["id"]),np.array([])))
+                                        if values is None:continue
+                                        values={key:value for key,value in values.items() if key in keys and np.isfinite(value)}
+                                        if not values:continue
+                                        time=row["sample_time"]
+                                        records.append({"sample_time":time,"day":day,"slot":time.hour//6,"values":values})
+                                    del cycles
+                                if not records:continue
+                                valid_days+=1
+                                if valid_days<=3:
+                                    baseline.extend(sample_per_day(records,3))
+                                    continue
+                                cycle_count+=len(records)
+                                daily_values=aggregate(records,"day")
+                                daily.append({"day":day,"slot":-1,"values":{key:value for (group,key),value in daily_values.items()}})
+                                slot_values=aggregate(records,"day_slot")
+                                for slot in sorted({record["slot"] for record in records}):
+                                    slots.append({"day":day,"slot":slot,"values":{key:value for (group,key),value in slot_values.items() if group==f"{day}#{slot}"}})
+                            payload={"metrics":np.asarray(METRICS),"cycle_count":np.asarray(cycle_count)}
+                            for kind,records in (("baseline",baseline),("daily",daily),("slots",slots)):
+                                payload[f"{kind}_days"]=np.asarray([record["day"] for record in records],dtype="U10")
+                                payload[f"{kind}_slots"]=np.asarray([record["slot"] for record in records],dtype=int)
+                                payload[f"{kind}_values"]=np.asarray([[record["values"].get(key,np.nan) for key in keys] for record in records],dtype=float).reshape(-1,len(keys))
+                            temporary=cache_path.with_suffix(".tmp.npz")
+                            np.savez_compressed(temporary,**payload)
+                            temporary.replace(cache_path)
+                        if baseline and daily:
+                            point_data[f"{cylinder}缸{side}"]={"baseline":build_baseline(baseline),"daily":daily,"slots":slots,"cycle_count":cycle_count}
+                if not point_data:continue
+                z_rows=[];slot_z=defaultdict(dict)
+                for point,data in point_data.items():
+                    for record in data["daily"]:
+                        for key,value in record["values"].items():
+                            base=data["baseline"][key]
+                            if not np.isfinite(base["center"]):continue
+                            z=(value-base["center"])/base["scale"]
+                            z_rows.append({"point":point,"day":record["day"],"angle":key[0],"metric":key[1],"z":z,"baseline":base})
+                    for record in data["slots"]:
+                        for key,value in record["values"].items():
+                            base=data["baseline"][key]
+                            if not np.isfinite(base["center"]):continue
+                            slot_z[(record["day"],record["slot"],key)][point]=(value-base["center"])/base["scale"]
+                public=defaultdict(list)
+                for row in z_rows:public[(row["day"],row["angle"],row["metric"])].append(row["z"])
+                public={key:float(np.median(values)) for key,values in public.items()}
+                slot_public={key:float(np.median(list(values.values()))) for key,values in slot_z.items()}
+                grouped=defaultdict(list)
+                for row in z_rows:
+                    row["local"]=row["z"]-public[(row["day"],row["angle"],row["metric"])]
+                    grouped[(row["point"],row["day"],row["angle"])].append(row)
+                history=defaultdict(deque)
+                for (point,day,angle),rows in sorted(grouped.items(),key=lambda item:(item[0][1],item[0][0],item[0][2])):
+                    supported=[row for row in rows if abs(row["local"])>=2]
+                    families={FAMILY[row["metric"]] for row in supported}
+                    amps=[abs(row["local"]) for row in supported]
+                    avg=float(np.mean(amps)) if amps else 0.0
+                    rms=float(np.sqrt(np.mean(np.square(amps)))) if amps else 0.0
+                    abnormal_slots=0
+                    for slot in range(4):
+                        slot_vals=[]
+                        for row in rows:
+                            key=(day,slot,(angle,row["metric"]))
+                            if point in slot_z.get(key,{}):slot_vals.append(abs(slot_z[key][point]-slot_public[key]))
+                        if sum(v>=2 for v in slot_vals)>=2:abnormal_slots+=1
+                    candidate=len(families)>=2 and avg>=2 and rms>=3 and abnormal_slots>=2
+                    hist=history[(point,angle)];hist.append(candidate)
+                    if len(hist)>5:hist.popleft()
+                    persistent=len(hist)>=3 and sum(hist)>=3
+                    level="持续预警" if persistent and candidate else "候选" if candidate else "观察" if supported else "正常"
+                    base_days=min(row["baseline"]["days"] for row in rows)
+                    detail_rows.append({
+                        "机组":unit,"运行段":f"{segment_start}~{segment_end}","日期":day,"测点":point,
+                        "样本标签":"unlabelled control",
+                        "备注":"2026-04-18距2026-04-01维修17天，不保证健康" if unit=="8" else "未确认健康，触发不能直接认定为误报",
+                        "角度箱":angle,"等级":level,"支持特征数":len(supported),"支持特征族数":len(families),
+                        "幅度平均Z":round(avg,4),"幅度中位Z":round(float(np.median(amps)) if amps else 0,4),
+                        "综合RMS":round(rms,4),"异常6小时子窗数":abnormal_slots,"最近5窗候选数":sum(hist),
+                        "基线有效日数":base_days,"基线置信":"低" if base_days<7 else "中" if base_days<20 else "高",
+                        "特征摘要":";".join(f"{METRIC_CN[r['metric']]}{'↑' if r['local']>0 else '↓'}{abs(r['local']):.1f}Z" for r in sorted(supported,key=lambda x:-abs(x['local']))),
+                    })
+                print(f"unlabelled control {unit}号 {segment_start}~{segment_end}: 测点{len(point_data)}，全量检测周期{sum(x['cycle_count'] for x in point_data.values()):,}",flush=True)
+            unit_rows=detail_rows[unit_detail_start:]
+            if unit_rows:
+                out_dir.mkdir(parents=True,exist_ok=True)
+                with (out_dir/f"伪在线_未标注对照触发明细_{unit}号.csv").open("w",encoding="utf-8-sig",newline="") as handle:
+                    writer=csv.DictWriter(handle,fieldnames=list(unit_rows[0]));writer.writeheader();writer.writerows(unit_rows)
+    finally:
+        if connection is not None:connection.close()
+    out_dir.mkdir(parents=True,exist_ok=True)
+    with (out_dir/"伪在线_未标注对照触发明细.csv").open("w",encoding="utf-8-sig",newline="") as handle:
+        writer=csv.DictWriter(handle,fieldnames=list(detail_rows[0]) if detail_rows else ["机组","样本标签"]);writer.writeheader();writer.writerows(detail_rows)
+    summary=[]
+    by_unit=defaultdict(list)
+    for row in detail_rows:by_unit[row["机组"]].append(row)
+    for unit,rows in by_unit.items():
+        days=len({row["日期"] for row in rows});points=len({row["测点"] for row in rows})
+        summary.append({"机组":unit,"样本标签":"unlabelled control","对照检测有效日数":days,"测点数":points,"窗口点箱记录":len(rows),
+                        "观察记录":sum(row["等级"]=="观察" for row in rows),
+                        "候选记录":sum(row["等级"]=="候选" for row in rows),
+                        "持续预警记录":sum(row["等级"]=="持续预警" for row in rows),
+                        "候选及以上日测点箱数":len({(row["日期"],row["测点"],row["角度箱"]) for row in rows if row["等级"] in ("候选","持续预警")}),
+                        "持续预警日测点箱数":len({(row["日期"],row["测点"],row["角度箱"]) for row in rows if row["等级"]=="持续预警"})})
+    with (out_dir/"伪在线_未标注对照触发统计.csv").open("w",encoding="utf-8-sig",newline="") as handle:
+        writer=csv.DictWriter(handle,fieldnames=list(summary[0]) if summary else ["机组","样本标签"]);writer.writeheader();writer.writerows(summary)
+    return detail_rows,summary
+
+
+def report(out_dir, detail, bands, validations):
+    lines=["# 伪在线故障检测验证", "", "## 方法", "",
+           "- 历史基线每天粗采最多3周期；24小时检测窗口处理全部有效周期。",
+           "- 周期平台归一化、测点自身历史稳健Z标准化、单机公共变化扣除。",
+           "- 检测不使用维修后数据；维修日期仅用于计算提前量。", ""]
+    for event in EVENTS:
+        rows=[row for row in validations if row["事件"]==event["id"]]
+        rows.sort(key=lambda row:(-int(row["持续预警窗口数"]),-float(row["最大综合RMS"])))
+        lines += [f"## {event['id']}", "", f"- {event['note'] or '按事件前数据回放'}。", f"- 形成候选的测点角度带：{len(rows)}。"]
+        for i,row in enumerate(rows[:20],1):
+            lines.append(f"{i}. {row['测点']} {row['角度带']}：首次候选{row['首次候选日期']}，提前{row['提前日数']}天，持续预警{row['持续预警窗口数']}窗，最大RMS{float(row['最大综合RMS']):.1f}。")
+        lines.append("")
+    lines += ["## 使用限制", "", "- 当前按有数据的自然日形成0~24小时窗口；下一版可改为每小时滚动24小时。",
+              "- 6号基线只有3个日期，所有结果应降级解释。", "- 7号2025事件附近缺测，只能验证运行段状态变化，不能证明事件提前量。"]
+    (out_dir/"伪在线检测结论.md").write_text("\n".join(lines)+"\n",encoding="utf-8")
+
+
+if __name__ == "__main__":
+    parser=argparse.ArgumentParser(description="局部特征伪在线回放")
+    parser.add_argument("--out-dir",default="cache/indicator/伪在线检测")
+    parser.add_argument("--event",action="append",choices=[event["id"] for event in EVENTS])
+    parser.add_argument("--healthy",action="store_true",help="回放无已知故障的健康运行段")
+    parser.add_argument("--unit",action="append",choices=UNITS,help="限制机组，可重复传入")
+    args=parser.parse_args()
+    output=Path(args.out_dir)
+    if args.healthy:
+        run_health_segments(output,tuple(args.unit or ("7","8","9")))
+    else:
+        detail,bands,validations=run(output,set(args.event or []))
+        report(output,detail,bands,validations)
+    print("结果 ->",output)

backend/indicator_features/spring_compare.py

@@ -0,0 +1,62 @@
+"""Daily-weighted spring state comparison, not a test of seasonal causality."""
+import csv
+from collections import defaultdict
+from pathlib import Path
+
+import numpy as np
+
+
+def main():
+    root = Path("cache/indicator")
+    output = []
+    for unit in ("7", "8", "9"):
+        filename = f"高密度特征_{unit}号_2026.csv" if unit != "9" else "高密度特征_9号_2026春季.csv"
+        with (root / filename).open(encoding="utf-8-sig") as handle:
+            rows = list(csv.DictReader(handle))
+        daily = defaultdict(list)
+        for row in rows:
+            day = row["sample_time"][:10]
+            if not "2026-03-01" <= day < "2026-04-20":
+                continue
+            if (unit, row["cylinder"], row["side"]) in {("7", "四", "盖侧"), ("7", "六", "盖侧"), ("8", "六", "盖侧")}:
+                continue
+            if float(row["p_span"]) < .3:
+                continue
+            keys = ["p_suction", "p_discharge", "p_span"] + [f"prof_{i:02d}" for i in range(12)]
+            vector = np.array([float(row[k]) for k in keys])
+            if np.all(np.isfinite(vector)):
+                daily[(row["cylinder"], row["side"], day)].append(vector)
+        for point in sorted({(c, s) for c, s, _ in daily}):
+            periods = []
+            for start, end in [("2026-03-01", "2026-03-21"), ("2026-03-21", "2026-04-01"), ("2026-04-01", "2026-04-20")]:
+                selected = [(day, values) for (c, s, day), values in daily.items() if (c, s) == point and start <= day < end]
+                median = np.median([np.median(values, axis=0) for _, values in selected], axis=0) if selected else None
+                periods.append((selected, median))
+            for label, index in [("3月上中旬至3月下旬", 1), ("3月上中旬至4月", 2)]:
+                before, a = periods[0]
+                after, b = periods[index]
+                if a is None or b is None:
+                    continue
+                delta = b - a
+                result = {"机组": unit, "测点": f"{point[0]}缸{point[1]}", "对照": label,
+                          "前段日期": "/".join(sorted(d for d, _ in before)), "后段日期": "/".join(sorted(d for d, _ in after)),
+                          "前段周期": sum(len(v) for _, v in before), "后段周期": sum(len(v) for _, v in after),
+                          "吸气变化": round(delta[0], 5), "排气变化": round(delta[1], 5),
+                          "幅度变化": round(delta[2], 5), "形状距离12点": round(float(np.linalg.norm(delta[3:])), 5)}
+                output.append(result)
+        print(f"{unit}号")
+        for label in ("3月上中旬至3月下旬", "3月上中旬至4月"):
+            selected = [r for r in output if r["机组"] == unit and r["对照"] == label]
+            print(label, "测点", len(selected))
+            for key in ("吸气变化", "排气变化", "幅度变化"):
+                print(key, "中位", round(float(np.median([r[key] for r in selected])), 5) if selected else None,
+                      "上升", sum(r[key] > 0 for r in selected))
+            print("形状变化", sorted([(r["测点"], r["形状距离12点"]) for r in selected], key=lambda v: -v[1]))
+    with (root / "春季同期跨机组对照.csv").open("w", encoding="utf-8-sig", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(output[0]))
+        writer.writeheader()
+        writer.writerows(output)
+
+
+if __name__ == "__main__":
+    main()

backend/indicator_features/summarize_pseudo_online.py

@@ -0,0 +1,103 @@
+#!/usr/bin/env python3
+"""Summarize merged pseudo-online replay results without database access."""
+from __future__ import annotations
+
+import csv
+from collections import Counter, defaultdict
+from pathlib import Path
+
+
+ROOT = Path("cache/indicator/伪在线检测")
+
+
+def read(name):
+    with (ROOT / name).open(encoding="utf-8-sig", newline="") as handle:
+        return list(csv.DictReader(handle))
+
+
+def write(name, rows):
+    if not rows:
+        return
+    with (ROOT / name).open("w", encoding="utf-8-sig", newline="") as handle:
+        writer = csv.DictWriter(handle, fieldnames=list(rows[0]))
+        writer.writeheader(); writer.writerows(rows)
+
+
+def main():
+    detail = read("伪在线_窗口检测明细.csv")
+    bands = read("伪在线_角度带告警.csv")
+    validation = read("伪在线_事件验证.csv")
+
+    common = []
+    by_angle = defaultdict(list)
+    for row in detail:
+        if row["等级"] in ("候选", "持续预警"):
+            by_angle[int(row["角度箱"])].append(row)
+    for angle, rows in sorted(by_angle.items()):
+        events = sorted({row["事件"] for row in rows})
+        if len(events) < 2:
+            continue
+        common.append({
+            "角度范围": f"{angle}-{angle + 10}°", "事件数": len(events),
+            "事件": ";".join(events), "测点数": len({row["测点"] for row in rows}),
+            "测点": ";".join(sorted({row["测点"] for row in rows})),
+            "候选日数": len({(row["事件"], row["日期"]) for row in rows}),
+            "持续预警记录数": sum(row["等级"] == "持续预警" for row in rows),
+            "最大幅度平均Z": round(max(float(row["幅度平均Z"]) for row in rows), 3),
+            "最大综合RMS": round(max(float(row["综合RMS"]) for row in rows), 3),
+            "共同特征族": ";".join(sorted({part.split("↑")[0].split("↓")[0]
+                                           for row in rows for part in row["特征摘要"].split(";") if part})),
+        })
+    common.sort(key=lambda row: (-row["事件数"], -row["持续预警记录数"], -row["候选日数"]))
+    write("伪在线_跨事件共性.csv", common)
+
+    lines = ["# 伪在线局部故障检测结论", "", "## 回放设置", "",
+             "- 基线使用事件前长日期数据，每测点每天粗采最多3个周期。",
+             "- 检测窗口按有数据的自然日0~24小时处理窗口内全部有效周期，不抽样。",
+             "- 每个24小时窗口拆成4个6小时子窗。",
+             "- 使用周期平台归一化、测点自身历史稳健Z、单机公共变化扣除。",
+             "- 回放检测不使用维修后数据。", ""]
+    for event in sorted({row["事件"] for row in detail}):
+        rows = [row for row in detail if row["事件"] == event]
+        event_bands = [row for row in bands if row["事件"] == event]
+        event_validation = [row for row in validation if row["事件"] == event]
+        counts = Counter(row["等级"] for row in rows)
+        lines += [f"## {event}", "",
+                  f"- 完整日×测点×角度箱记录：{len(rows)}。",
+                  f"- 正常{counts['正常']}，观察{counts['观察']}，候选{counts['候选']}，持续预警{counts['持续预警']}。",
+                  f"- 角度带状态记录：{len(event_bands)}；曾达到候选的测点角度带：{len(event_validation)}。"]
+        ranked = sorted(event_validation, key=lambda row: (-int(row["持续预警窗口数"]),
+                                                            -float(row["最大综合RMS"])))
+        for index, row in enumerate(ranked[:15], 1):
+            sensor = "（疑似整通道异常，排除局部故障）" if row["角度带"] == "0-360°" else ""
+            lines.append(f"{index}. {row['测点']} {row['角度带']}：首次候选{row['首次候选日期']}，"
+                         f"相对事件提前{row['提前日数']}天，持续预警{row['持续预警窗口数']}窗，"
+                         f"最大综合RMS {float(row['最大综合RMS']):.1f}{sensor}。")
+        lines.append("")
+
+    lines += ["## 已知目标命中情况", "",
+              "- 7号2026一缸盖侧240~270°：5月16日进入观察，5月17日三个箱开始形成候选，5月19日形成持续预警，相对5月21日提前约2~4天。",
+              "- 7号2026二缸盖侧140~150°：仅达到观察，没有达到24小时候选；它是数月慢漂移特征，说明必须增加运行段早期基线/慢趋势通道，不能只用长历史24小时幅度阈值。",
+              "- 4号四缸轴侧60~70°：6月7~9日为观察，6月10日达到候选；附近70~80°更早形成持续预警，提示角度带边界应允许相邻箱漂移。",
+              "- 6号三缸轴侧160~180°：4月19日形成候选，4月21日形成持续预警，相对4月22日提前约1~3天。",
+              "- 7号2025二缸盖侧出现0~360°超大异常，是已知传感器卡死污染，不是局部故障命中。", "",
+              "## 方法结论", "",
+              "1. 长历史粗采样基线 + 24小时全周期检测能够检出事件前局部变化，但单通道会漏掉缓慢、小幅、长期单向漂移。",
+              "2. 正式方法必须并行运行快通道和慢通道：快通道使用长历史基线检测24小时突变；慢通道使用本运行段最早日期固定基线检测数日/数周定向漂移。",
+              "3. 角度带不能要求每天边界完全一致，应允许相邻10度箱在前后一天移动一个箱，并按20~40度范围追踪。",
+              "4. 全周期大范围同时异常应转入传感器/工况分类，不进入局部气阀告警。",
+              "5. 当前阈值在4号和6号产生较多候选，尚不能直接上线；必须用更多无故障运行段校准候选日触发率。",
+              "6. 跨机器只比较无量纲Z、特征族、持续性和角度带结构，不比较原始压力或原始特征。", "",
+              "## 建议正式检测条件", "",
+              "- 快通道：24小时内至少两个6小时子窗异常、至少两个特征族、幅度平均不低于2Z、综合RMS不低于3Z。",
+              "- 慢通道：相对运行段早期固定基线连续至少3个有效日期同方向，或最近5个有效日期至少3天达到2Z；不要求单日达到快通道阈值。",
+              "- 持续预警：快通道最近5个有效窗口至少3个候选，或慢通道持续3天以上；同时要求局部残差显著。",
+              "- 传感器/公共工况排除：异常覆盖超过半个周期、多个同级测点同步或绝对压力平台不合理时，降级为数据质量/工况告警。",
+              "- 基线有效日期少于7天只输出探索性结果；7~19天为中置信；20天以上才允许高置信。"]
+    (ROOT / "伪在线检测结论.md").write_text("\n".join(lines) + "\n", encoding="utf-8")
+    print("跨事件共性", len(common), "->", ROOT / "伪在线_跨事件共性.csv")
+    print("结论 ->", ROOT / "伪在线检测结论.md")
+
+
+if __name__ == "__main__":
+    main()

backend/indicator_features/伪在线局部故障检测方案.md

@@ -0,0 +1,116 @@
+# 伪在线局部故障检测方案
+
+## 目标
+
+使用事件发生前可获得的数据建立基线，以0~24小时滚动窗口处理窗口内全部有效周期，验证局部功图特征能否在维修事件前提前触发。维修日期和维修后数据只用于评价，不参与在线基线或事件前判定。
+
+## 归一化
+
+不同机器和不同压力级不能直接比较原始压力或原始特征，采用三层归一化：
+
+1. **周期内工况归一化**：每周期先平滑并识别吸排气平台，计算`(压力-吸气压力)/(排气压力-吸气压力)`。
+2. **测点自身历史标准化**：每个机组、测点、10度箱、特征相对自身过去有效日期计算稳健Z值，不跨机器共用原始中心或尺度。
+3. **单机公共变化扣除**：同一时窗、同角度、同特征在全机有效测点上的Z中位数作为公共变化；测点Z减公共Z得到局部残差。
+
+跨机组只比较无量纲局部残差、持续性、支持特征族和角度带宽度，不直接比较MPa或原始局部特征。
+
+## 历史基线
+
+- 基线只使用检测时刻以前的数据，禁止未来信息泄漏。
+- 基线向前覆盖90~180个日历日；只统计有有效功图的日期，空白时间跳过。
+- 基线粗采样：每测点每天均匀抽取最多3个周期。
+- 每日先取中位，再跨有效日期取中位，保证日期等权。
+- 尺度综合日间MAD、基线周期MAD和全角度尺度下限，防止近零尺度制造虚高Z值。
+- 最近疑似异常日期不立即进入基线；基线只缓慢更新。
+- 基线有效日期不足时降低置信等级，并报告基线日数和周期数。
+
+## 检测窗口
+
+- 每次检测使用过去24小时内全部有效首周期，不抽样。
+- 同时计算0~6、6~12、12~18、18~24小时子窗，用于判断持续性。
+- 输出有效周期数、覆盖小时数和首末采样时间；覆盖不足不能与完整窗口同等级报警。
+- 基础角度箱为10度，相邻同类异常箱合并成20~40度局部变化带；允许跨0/360度合并。
+
+## 原始特征
+
+不扩充原始特征，继续使用：
+
+- 局部中位压力；
+- 局部斜率；
+- 标准差；
+- 四分位距；
+- 导数RMS；
+- 最大二阶导。
+
+## 评估量
+
+每个测点、10度箱计算：
+
+- 相对自身历史基线Z；
+- 全机公共Z；
+- 局部残差Z；
+- 正向和负向支持指标数；
+- 支持特征族数；
+- 局部幅度平均：支持指标绝对局部残差Z的平均；
+- 局部幅度中位；
+- 幅度综合RMS；
+- 24小时内异常子窗数；
+- 最近多个24小时窗口的持续次数。
+
+## 告警分级
+
+### 观察
+
+- 单个10度箱或单个特征超过约2Z；
+- 只出现于一个子窗或覆盖不足。
+
+### 候选
+
+- 至少两个特征族支持；
+- 局部幅度平均约2Z以上；
+- 综合RMS约3Z以上；
+- 至少两个6小时子窗出现；
+- 局部残差不能主要由全机公共变化解释。
+
+### 持续预警
+
+- 最近5个24小时检测窗口至少3个满足候选条件；
+- 方向和角度带基本稳定；
+- 相邻箱形成20~40度局部带，或单个10度箱有多特征强支持。
+
+### 历史故障相似候选
+
+- 达到持续预警；
+- 测点、角度带、特征方向和幅度结构与历史事件模板相近；
+- 只作为人工复核优先级，不作为自动故障定性。
+
+## 伪在线验证
+
+对7号2025维修过程、7号2026-05-21、4号2026-06-10和6号2026-04-22~26按真实时间顺序回放：
+
+1. 每个检测时刻只能使用此前数据建立基线。
+2. 24小时检测窗口使用窗口内全部有效周期。
+3. 记录首次观察、首次候选、首次持续预警、距事件提前量。
+4. 统计事件前无标签区间的候选窗口数量和候选日数量。
+5. 维修后数据只用于检查短期恢复、长期保持或再次偏离。
+6. 7号2025因事件附近长期缺测，只验证各运行段是否被识别为状态变化，不宣称准确预测8月或10月事件。
+
+## 输出列表
+
+必须完整输出而不是只列Top项：
+
+1. `伪在线_窗口检测明细.csv`：每个24小时窗口、测点、角度箱的完整评分。
+2. `伪在线_角度带告警.csv`：相邻10度箱合并后的全部观察/候选/持续告警。
+3. `伪在线_事件验证.csv`：每个事件和变化带的首次触发、提前量、持续性、恢复情况。
+4. `伪在线_误报统计.csv`：按机组、测点、角度带统计非事件邻近窗口触发率。
+5. `伪在线_跨事件共性.csv`：比较无量纲幅度、特征族、方向和持续条件。
+6. `伪在线检测结论.md`：说明哪些特征可用于前期检测，哪些仅是事件后对照或工况敏感特征。
+
+## 判读原则
+
+- 维修后恢复不是在线检测条件，只是离线验证证据。
+- 不将跨长停机的差异称为连续演化。
+- 不将相同角度在不同缸直接解释为相同物理阶段。
+- 不将全机同步变化直接判为局部故障。
+- 不将单日或单周期极端值作为持续预警。
+- 在线基线不得使用未来事件或维修后数据。

backend/indicator_features/基线模板待优化.md

@@ -0,0 +1,31 @@
+# 基线模板待优化
+
+## 当前问题
+
+趋势扩展网页的模板是参考段归一化功图的逐角度中位数，不是经确认的健康功图。
+
+| 选项 | 当前参考段 | 用途与风险 |
+|---|---|---|
+| 7号2026-05-21 | 全部有效事件前周期 | 包含临近事件的变化，可能掩盖早期偏离 |
+| 8号2026合并事件 | 3月15~20日有效周期 | 可代表3月状态，不能证明健康；后段工况可能不同 |
+| 8号2025同期 | 3月13~19日有效周期 | 跨运行段对照，不足以证明连续变化或季节因果 |
+| 7号2025两个旧选项 | 7月25~27日有效周期 | 适合7月与11月对比，不适合发现7月之前的变化；两选项不是独立样本 |
+| 7号2025回溯 | 4月底有效周期 | 日期少，模板代表性待核验 |
+
+形状距离为归一化角度点与模板的欧氏距离。它不保留变化方向，不等于故障严重度；不同模板和不同点数的距离不可直接混比。
+
+## 后续优化
+
+1. 分离两种用途：固定早期参考段用于前期趋势；最后维修前运行段用于维修前后对照。
+2. 冻结并显示每个模板的测点、日期、有效日期数、周期数；扩展窗口不能隐式改变模板。
+3. 每日先取中位功图，再对参考日期取中位，避免采样多的日期主导模板。
+4. 检查参考段自身稳定性；采用留一日期验证，评估参考波动，避免样本参与自己的模板造成过小距离。
+5. 日期过少时只作描述，不能将波动带作为可靠报警阈值。
+6. 同时保留吸气压力、排气压力、压力幅度和局部带符号差值，以区分恢复、工况转换和局部变化。
+7. 7号2026趋势基线优先检查较早5月运行日期，不纳入临近事件段；7号2025回溯先确认4月底与5月初是否同状态，再决定是否合并。
+8. 模板应称为“参考状态模板”，未经健康标签验证不可称“正常模板”。
+
+## 本次决定
+
+仅记录，暂不调整模板或网页。一缸功图交互对比.html保持不动。
+此前“3月底/4月初变化是季节效应”的表述需要进一步验证：同期变化也可能来自工况、维修或传感器，不能仅凭月份重复断言季节因果。

backend/indicator_features/季节交替跨机组证据.md

@@ -0,0 +1,57 @@
+# 季节交替跨机组证据
+
+## 本次范围
+
+- 7号：2026年3月12~4月4日高密度功图；有效测点10个（剔除已知坏道）。
+- 8号：2026年3月15~4月29日高密度功图；有效测点11个（剔除8号六缸盖侧）。
+- 9号：2026年3月1~4月19日高密度功图，共2304条周期特征、12个测点。
+- 2025年3月数据：7号只有4月29日以后，9号只有3月13~14日，无法做完整的2025年3月底/4月初跨年验证。
+- 数据按“每天先取中位，再对日期中位取中位”比较，避免每天抽样数量不同主导结果。
+
+## 2026年3月到4月的变化
+
+| 机组 | 对照 | 吸气压力中位变化 | 排气压力中位变化 | 压力幅度中位变化 | 形状变化特点 |
+|---|---|---:|---:|---:|---|
+| 7号 | 3月上中旬 -> 3月下旬 | +0.0115 | +0.0090 | +0.0085 | 一缸盖/轴侧较大，其他缸较小 |
+| 7号 | 3月上中旬 -> 4月 | +0.0146 | +0.0167 | +0.0075 | 一缸盖/轴侧仍较大，但覆盖短 |
+| 8号 | 3月上中旬 -> 4月 | +0.2550 | +0.3227 | +0.1824 | 多数测点同步上升，一缸额外明显 |
+| 9号 | 3月上中旬 -> 3月下旬 | +0.2452 | +0.3299 | +0.1931 | 12个测点基本都同步上升 |
+| 9号 | 3月上中旬 -> 4月 | +0.2437 | +0.3396 | +0.1862 | 12个测点基本都同步上升 |
+
+## 形状变化排名
+
+归一化功图的12点形状距离仅用于本机内部排序，不与不同机组直接比较。
+
+### 7号
+
+- 一缸轴侧：3月下旬约0.0687；4月约0.0801。
+- 一缸盖侧：3月下旬约0.0370；4月约0.0928。
+- 其余测点大多约0.003~0.018。
+
+### 8号
+
+- 一缸轴侧：约0.5790。
+- 一缸盖侧：约0.5730。
+- 其余测点约0.026~0.077。
+
+### 9号
+
+- 一缸轴侧：3月下旬约0.6398；4月约0.6165。
+- 一缸盖侧：3月下旬约0.6354；4月约0.6792。
+- 三缸/五缸部分测点约0.10~0.15。
+- 其他测点约0.02~0.08。
+
+## 当前判断
+
+1. **9号存在最强的跨机组季节/工况证据。** 2026年3月底到4月，12个测点吸气、排气压力几乎全都同向上升，压力幅度多数上升；3月底和4月的结果一致。这与8号的“全机同步抬高”方向一致。
+2. **8号的季节/工况变化得到9号支持。** 8号不再是孤立案例；但8号一缸的额外形状变化仍然大于其他缸，可能叠加了单缸状态变化，不能全部解释为季节。
+3. **7号也有同期变化，但证据较弱。** 7号3月到4月压力变化方向与8号、9号一致，但绝对幅度小很多，且2026年4月只有4月1~4日有效数据；一缸额外形状变化比较突出，不能只叫季节变化。
+4. **跨机组方向相似，但不能证明纯粹由季节导致。** 共同变化也可能来自同期负荷、介质、压力级运行状态或数据采集状态；目前更准确的名称是“春季同期公共工况变化”。
+5. **9号没有完整2025年同期功图。** 因此目前不能说这是跨年份重复的季节规律，只能说2026年三台机在春季转换窗口出现了相似方向的公共变化。
+
+## 对后续基线的影响
+
+- 8号2026一缸的大变化不能直接全部判为故障；应先减去8号/9号共同的春季公共变化。
+- 7号一缸仍保留为局部候选，因为它相对于7号其他缸的额外变化明显，但需要与公共工况残差比较。
+- 9号应作为新的公共工况参考机组，不应直接作为故障样本。
+- 目前应使用“春季公共工况基线 + 单缸局部残差”而不是单纯事件前模板。