周期回填改为全量NULL重算并加触发甄别加速、波形UI y轴分类与归一聚合、开机停机状态同步脚本与文档

.gitignore

@@ -15,3 +15,4 @@ web 账号.txt
 backend/LabelingPreTraining/inputdatas/
 *.log
 .backup_old_*/
+cache/

backend/LabelingPreTraining/detect_cycle_index.py

@@ -2,32 +2,45 @@
 
 The original full re-run encoded the first complete cycle of every file but
 never wrote back where that cycle starts and ends. This script re-detects the
-first complete cycle for every file of a given measurement type (rpm > 0,
-selected with ``--measurement-type``) and records the real ``sample_index``
-bounds into two columns:
+first complete cycle for every file that still has **no stored cycle bounds**
+(``cycle_start IS NULL``) and records the real ``sample_index`` bounds into two
+columns:
 
 * ``cycle_start`` / ``cycle_end`` >= 0 : first complete cycle, half-open
   ``[cycle_start, cycle_end)`` in actual sample_index values;
 * ``-1``                             : no complete cycle found (matches the
-  ``tspluse_status = -1`` convention);
-* ``NULL``                           : not yet processed.
-
-It reuses predict.py's adaptive multi-stage batch reading (STAGE_LIMITS =
-3500 -> 7000 -> 14000, parallel reader connections, tuple watermark) but needs
-no model: only the trigger channel is used, so it runs on CPU only. The fetch
-also keeps the real ``sample_index`` per row because ``detect_cycles`` returns
-array offsets, which differ from database indices for sparse columns.
-
-Only rows still holding ``cycle_start IS NULL`` are selected, so the
-``(sample_time, id)`` watermark is naturally idempotent: a file already
-written (0/-1) is never re-selected on resume. Writes are flushed in one
-``CASE id`` UPDATE per ``--write-batch`` rows.
+  ``tspluse_status = -1`` convention).
+
+Any row already carrying a real first cycle (``cycle_start >= 0``) is never
+re-selected. No ``rpm`` / ``device_status`` filter is applied: every file that
+was previously skipped (mostly stopped machines, but also running files with
+unreliable ``rpm = 0``) gets re-detected, across all measurement types by
+default.
+
+Because most pending files never contain a trigger pulse, a cheap screening
+pass runs before detection: one grouped query per batch looks for any
+trigger-high sample (``second_value >= TRIGGER_THRESHOLD``) inside the
+reachable window ``STAGE_LIMITS[-1]``, and only files that pass enter the numpy
+detector. This matches the staged detector's reach exactly — a window with no
+high trigger can never yield a cycle — so the expensive multi-stage reads are
+spent only on the small minority of files that actually contain pulses.
+
+The detection itself reuses predict.py's adaptive multi-stage batch reading
+(STAGE_LIMITS = 3500 -> 7000 -> 14000, parallel reader connections, tuple
+watermark) but needs no model: only the trigger channel is used, so it runs on
+CPU only. The fetch also keeps the real ``sample_index`` per row because
+``detect_cycles`` returns array offsets, which differ from database indices for
+sparse columns.
+
+Only rows holding ``cycle_start IS NULL`` are selected, so the ``(sample_time,
+id)`` watermark keeps the run position and any file that gained a cycle during
+the run is not re-read. Writes are flushed in one ``CASE id`` UPDATE per
+``--write-batch`` rows.
 
 Usage:
     conda activate tspulse
+    python detect_cycle_index.py                              # all types
     python detect_cycle_index.py --measurement-type 位移
-    python detect_cycle_index.py --measurement-type 加速度
-    python detect_cycle_index.py --measurement-type 位移 --measurement-type 加速度
     python detect_cycle_index.py --measurement-type 位移,加速度
     python detect_cycle_index.py [--measurement-type 压力] [--start-time ""]
                                  [--start-id 0] [--batch 100] [--read-workers 4]
@@ -45,7 +58,7 @@ import numpy as np
 BACKEND = Path(__file__).resolve().parents[1]
 sys.path.insert(0, str(BACKEND))
 
-from app.algorithms.cycles import detect_cycles  # noqa: E402
+from app.algorithms.cycles import TRIGGER_THRESHOLD, detect_cycles  # noqa: E402
 import predict  # noqa: E402
 from predict import STAGE_LIMITS  # noqa: E402
 
@@ -116,6 +129,56 @@ def read_batch_samples(
     return merged
 
 
+SCREEN_LIMIT = STAGE_LIMITS[-1]
+
+
+def _fetch_trigger_chunk(connection, file_ids: list[int], limit: int) -> set[int]:
+    """Return file_ids that have at least one trigger-high sample in the window."""
+    if not file_ids:
+        return set()
+    placeholders = ",".join(["%s"] * len(file_ids))
+    with connection.cursor() as cursor:
+        cursor.execute(
+            f"""
+            SELECT DISTINCT wave_file_id
+            FROM wave_sample
+            WHERE wave_file_id IN ({placeholders})
+              AND sample_index < %s
+              AND second_value >= %s
+            """,
+            (*file_ids, limit, TRIGGER_THRESHOLD),
+        )
+        return {int(row["wave_file_id"]) for row in cursor.fetchall()}
+
+
+def screen_trigger_files(
+    reader_connections: list,
+    file_ids: list[int],
+    workers: int,
+) -> set[int]:
+    """Return the subset of ``file_ids`` that contain any trigger-high sample.
+
+    A single grouped query reads only the trigger channel (``second_value``)
+    within the reachable detection window (``SCREEN_LIMIT``). Files without a
+    single ``second_value >= TRIGGER_THRESHOLD`` sample can never produce a
+    complete cycle, so they are decided (-1) without the numpy detector.
+    """
+    if not file_ids:
+        return set()
+    if workers <= 1 or len(file_ids) <= workers:
+        return _fetch_trigger_chunk(reader_connections[0], file_ids, SCREEN_LIMIT)
+    chunks = np.array_split(file_ids, min(workers, len(file_ids)))
+    with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
+        futures = [
+            executor.submit(_fetch_trigger_chunk, reader_connections[i], chunk.tolist(), SCREEN_LIMIT)
+            for i, chunk in enumerate(chunks)
+        ]
+        merged: set[int] = set()
+        for future in futures:
+            merged |= future.result()
+    return merged
+
+
 def first_cycle_bounds(samples: np.ndarray) -> tuple[int, int] | None:
     """Return ``(cycle_start, cycle_end)`` in real sample_index, half-open."""
     cycles, _ = detect_cycles(samples)
@@ -201,7 +264,7 @@ def fetch_batch(
                 """
                 SELECT id, point_name, measurement_type, sample_time, sample_count
                 FROM wave_file
-                WHERE rpm > 0 AND measurement_type = %s AND cycle_start IS NULL
+                WHERE measurement_type = %s AND cycle_start IS NULL
                 ORDER BY sample_time ASC, id ASC
                 LIMIT %s
                 """,
@@ -212,7 +275,7 @@ def fetch_batch(
                 """
                 SELECT id, point_name, measurement_type, sample_time, sample_count
                 FROM wave_file
-                WHERE rpm > 0 AND measurement_type = %s AND cycle_start IS NULL
+                WHERE measurement_type = %s AND cycle_start IS NULL
                   AND (sample_time > %s OR (sample_time = %s AND id > %s))
                 ORDER BY sample_time ASC, id ASC
                 LIMIT %s
@@ -222,10 +285,24 @@ def fetch_batch(
         return cursor.fetchall()
 
 
+def list_pending_types(connection) -> list[str]:
+    """Return every measurement_type that still has files without cycle bounds."""
+    with connection.cursor() as cursor:
+        cursor.execute(
+            """
+            SELECT DISTINCT measurement_type AS mt
+            FROM wave_file
+            WHERE cycle_start IS NULL
+            ORDER BY mt
+            """
+        )
+        return [str(row["mt"]) for row in cursor.fetchall() if row["mt"] is not None]
+
+
 def main() -> int:
     parser = argparse.ArgumentParser(description="回写首个完整周期的 sample_index 边界")
     parser.add_argument("--measurement-type", action="append", default=[],
-                        help="要回写的测量类型，可多次指定或用逗号分隔（默认压力）")
+                        help="要回写的测量类型，可多次指定或用逗号分隔（默认自动取全部测量类型）")
     parser.add_argument("--start-time", type=str, default="",
                         help="起始水位时间(含)，留空从头开始")
     parser.add_argument("--start-id", type=int, default=0,
@@ -240,11 +317,17 @@ def main() -> int:
     args = parser.parse_args()
 
     measurement_types: list[str] = []
-    for raw in (args.measurement_type or ["压力"]):
+    for raw in args.measurement_type:
         for part in str(raw).split(","):
             part = part.strip()
             if part and part not in measurement_types:
                 measurement_types.append(part)
+    if not measurement_types:
+        with predict.get_connection() as connection:
+            measurement_types = list_pending_types(connection)
+        if not measurement_types:
+            print("没有待回写的测量类型（无 device_status = 1 且周期缺失/为 -1 的文件），退出")
+            return 0
 
     start_time = args.start_time.strip()
     read_workers = max(1, min(args.read_workers, args.batch))
@@ -254,7 +337,7 @@ def main() -> int:
     watermark_time: object | None = start_time or None
     watermark_id = args.start_id
     total = 0
-    stats = {"有周期": 0, "无周期": 0}
+    stats = {"有周期": 0, "无周期": 0, "屏掉(无触发)": 0}
     errors = 0
     last_reported = 0
     write_buffer: list[tuple[int, int, int]] = []
@@ -270,7 +353,7 @@ def main() -> int:
             watermark_time = start_time or None
             watermark_id = args.start_id
             total = 0
-            stats = {"有周期": 0, "无周期": 0}
+            stats = {"有周期": 0, "无周期": 0, "屏掉(无触发)": 0}
             errors = 0
             last_reported = 0
             write_buffer = []
@@ -284,7 +367,7 @@ def main() -> int:
             def summary_text() -> str:
                 return (
                     f"{measurement_type}: 已处理 {total}, 有周期 {stats['有周期']}, "
-                    f"无周期 {stats['无周期']}, 错误 {errors}, "
+                    f"无周期 {stats['无周期']} (其中屏掉无触发 {stats['屏掉(无触发)']}), 错误 {errors}, "
                     f"水位 sample_time={watermark_time} id={watermark_id}"
                 )
 
@@ -299,9 +382,16 @@ def main() -> int:
                         print(f"无待处理文件，完成 [{summary_text()}]")
                         break
 
-                    resolved, batch_errors = resolve_batch_bounds(
-                        rows, reader_connections, read_workers,
+                    file_ids = [int(row["id"]) for row in rows]
+                    trigger_ids = screen_trigger_files(reader_connections, file_ids, read_workers)
+                    detect_rows = [row for row in rows if int(row["id"]) in trigger_ids]
+                    resolved: dict[int, tuple[int, int] | None] = {
+                        int(row["id"]): None for row in rows if int(row["id"]) not in trigger_ids
+                    }
+                    batch_resolved, batch_errors = resolve_batch_bounds(
+                        detect_rows, reader_connections, read_workers,
                     )
+                    resolved.update(batch_resolved)
                     errors += batch_errors
 
                     for row in rows:
@@ -310,6 +400,8 @@ def main() -> int:
                         bounds = resolved[file_id]
                         if bounds is None:
                             stats["无周期"] += 1
+                            if file_id not in trigger_ids:
+                                stats["屏掉(无触发)"] = stats.get("屏掉(无触发)", 0) + 1
                             write_buffer.append((file_id, -1, -1))
                         else:
                             stats["有周期"] += 1

backend/app/services/data_service.py

@@ -1,6 +1,7 @@
 from __future__ import annotations
 
 import re
+from bisect import bisect_left
 from collections import OrderedDict
 from datetime import datetime, timedelta
 from functools import lru_cache
@@ -47,6 +48,8 @@ ANNOTATION_LABELS = ("正常", "异常")
 # PKS 全场点位：机组号 -> pks_long_sample.import_batch_id
 # 7号机=30、8号机=31、9号机=32。
 UNIT_BATCH = {"7": 30, "8": 31, "9": 32}
+# 全场点位(PKS)每个时间点的曲线窗口：以采样时刻为中心的前后各 7.5 分钟。
+PKS_WINDOW_SECONDS = 15 * 60
 _SITE_POINT_PATTERN = re.compile(r"^YSJ([789])_([1-9]|1[0-9]|2[0-9]|3[0-9]|4[0-1])$")
 
 # Extra samples fetched past cycle_end so detect_cycles can see the zero marker
@@ -361,19 +364,67 @@ class DataService:
                 )
 
     @staticmethod
-    def _build_site_series(site_points: list[str], points: list[dict[str, Any]]) -> dict[str, Any]:
-        """把时间点上的 siteValues 整理成每周期一个点的曲线系列。"""
+    def _build_site_series(
+        site_points: list[str],
+        unit: str | None,
+        points: list[dict[str, Any]],
+    ) -> dict[str, Any]:
+        """把每个时间点扩展为前后各 7.5 分钟的 pks 5s 曲线段。
+
+        每个选中点位(PKS)在该窗口内的每个时间点不再只画单值，而是以该
+        时间点采样时刻为中心，取 [t-7.5min, t+7.5min) 的原始 5s 数据映射
+        到该时间点在 x 轴占据的格子（索引 slot ~ slot+1）内连成一小段曲线。
+        相邻时间点若恰好间隔 15 分钟，则相邻窗口首尾衔接、无重叠。
+        数据按列做一次整段范围查询，再按时间点二分切段。
+        """
+        columns = [
+            (item_name, column)
+            for item_name in site_points
+            if (column := _site_point_column(item_name, unit)) is not None
+        ]
+        centers: list[tuple[int, datetime]] = []
+        for index, point in enumerate(points):
+            try:
+                timestamp = datetime.strptime(point["sampleTime"], "%Y-%m-%d %H:%M:%S")
+            except (TypeError, ValueError):
+                continue
+            centers.append((index, timestamp))
+        if not columns or not centers:
+            return {"points": site_points, "series": []}
+        half = timedelta(seconds=PKS_WINDOW_SECONDS // 2)
+        query_start = min(timestamp for _index, timestamp in centers) - half
+        query_end = max(timestamp for _index, timestamp in centers) + half
         series = []
-        for item_name in site_points:
-            data = []
-            for index, point in enumerate(points):
-                value = (point.get("siteValues") or {}).get(item_name)
-                if value is not None:
+        for item_name, column in columns:
+            with get_connection() as connection:
+                with connection.cursor() as cursor:
+                    cursor.execute(
+                        f"SELECT sample_time, `{column}` AS value FROM pks_long_sample "
+                        "WHERE import_batch_id = %s AND sample_time >= %s AND sample_time < %s "
+                        "ORDER BY sample_time",
+                        (UNIT_BATCH[unit], query_start, query_end),
+                    )
+                    rows = cursor.fetchall()
+            sample_times = [row["sample_time"] for row in rows]
+            values = [row["value"] for row in rows]
+            data: list[dict[str, Any]] = []
+            for index, center in centers:
+                low = bisect_left(sample_times, center - half)
+                high = bisect_left(sample_times, center + half)
+                for pos in range(low, high):
+                    raw_value = values[pos]
+                    if raw_value is None:
+                        continue
+                    value = float(raw_value)
+                    if not np.isfinite(value):
+                        continue
+                    x = index + 0.5 + (sample_times[pos] - center).total_seconds() / PKS_WINDOW_SECONDS
                     data.append(
                         {
-                            "value": [index, float(value)],
-                            "x": index,
-                            "sampleTime": point["sampleTime"],
+                            "value": [x, value],
+                            "x": x,
+                            "rawValue": value,
+                            "sampleTime": _time_string(sample_times[pos]),
                         },
                     )
             series.append({"itemName": item_name, "data": data})
@@ -582,10 +633,15 @@ class DataService:
             )
 
         result, source = self._run_with_fallback(database_query, demo_query)
-        # 全场点位(PKS)系列：仅首个周期模式叠加，每周期一个点。
+        # 全场点位(PKS)系列：每周期(PKS 5s)一小段、跨时间点连续的 15 分钟窗口曲线。
         if first_cycle_only and site_points:
             try:
-                result["siteSeries"] = self._build_site_series(site_points, result["points"])
+                unit = _unit_number(device_part)
+                result["siteSeries"] = (
+                    self._build_site_series(site_points, unit, result["points"])
+                    if unit is not None
+                    else {"points": [], "series": []}
+                )
             except Exception:
                 result["siteSeries"] = {"points": [], "series": []}
         else:

backend/diagnose_drift.py

@@ -0,0 +1,265 @@
+from __future__ import annotations
+
+import os
+import sys
+from time import monotonic
+
+import numpy as np
+import pandas as pd
+import pymysql
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from app.config import settings  # noqa: E402
+
+BATCHES = (30, 31, 32)
+COLUMN = "YSJ_1"
+RNG = np.random.default_rng(7)
+
+
+def connect() -> pymysql.connections.Connection:
+    return pymysql.connect(
+        host=settings.db_host,
+        port=settings.db_port,
+        user=settings.db_user,
+        password=settings.db_password,
+        database=settings.db_name,
+        charset="utf8mb4",
+        cursorclass=pymysql.cursors.SSCursor,
+        connect_timeout=settings.db_connect_timeout,
+        read_timeout=3600,
+        write_timeout=120,
+    )
+
+
+LEVEL_CACHE = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "cache", "pks_levels.csv"
+)
+
+
+def wall_offset() -> int:
+    """Return seconds to add to UNIX epoch so naive pandas shows local wall-clock times."""
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT sample_time, UNIX_TIMESTAMP(sample_time) AS u "
+                           "FROM pks_long_sample LIMIT 1")
+            wall, uni = cursor.fetchone()
+    finally:
+        connection.close()
+    wall_epoch = int(pd.Timestamp(wall).tz_localize(None).value // 10 ** 9)
+    return wall_epoch - int(uni)
+
+
+def load_levels(offset: int) -> pd.DataFrame:
+    """1-min bucket AVERAGE level per batch, from a single server-side aggregation (cached)."""
+    if os.path.exists(LEVEL_CACHE):
+        frame = pd.read_csv(LEVEL_CACHE)
+        frame["time"] = pd.to_datetime(frame["time"])
+        print(f"   level buckets loaded from cache: {len(frame):,} rows")
+        return frame
+    t0 = monotonic()
+    connection = connect()
+    rows = []
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT import_batch_id AS b, UNIX_TIMESTAMP(sample_time) DIV 60 AS bk, "
+                f"COUNT(*) AS cnt, AVG(`{COLUMN}`) AS lvl "
+                "FROM pks_long_sample GROUP BY import_batch_id, bk ORDER BY import_batch_id, bk"
+            )
+            while True:
+                chunk = cursor.fetchmany(200_000)
+                if not chunk:
+                    break
+                rows.extend(chunk)
+    finally:
+        connection.close()
+    frame = pd.DataFrame(rows, columns=["b", "bk", "cnt", "lvl"])
+    frame["time"] = pd.to_datetime(frame["bk"] * 60 + offset, unit="s")
+    bad = frame[frame["cnt"] != 60]
+    if len(bad):
+        print(f"   WARN: {len(bad)} buckets with cnt!=60\n{bad.head(10).to_string()}")
+    frame[["b", "bk", "cnt", "lvl", "time"]].to_csv(LEVEL_CACHE, index=False)
+    print(f"   level buckets loaded: {len(frame):,} rows ({frame['b'].nunique()} batches) "
+          f"{monotonic()-t0:.0f}s")
+    return frame
+
+
+def sample_raw_days() -> pd.DataFrame:
+    """Fetch raw 5s rows of ~24 random days per batch to gauge noise/quantization/gaps."""
+    t0 = monotonic()
+    days = pd.date_range("2025-04-01", "2026-04-30", freq="D")
+    picks = days[RNG.choice(len(days), size=24, replace=False)].sort_values()
+    connection = connect()
+    parts = []
+    try:
+        with connection.cursor() as cursor:
+            for d in picks:
+                start = d.strftime("%Y-%m-%d 00:00:00")
+                end = d.strftime("%Y-%m-%d 23:59:55")
+                cursor.execute(
+                    "SELECT import_batch_id AS b, "
+                    f"UNIX_TIMESTAMP(sample_time) AS t, `{COLUMN}` AS v "
+                    "FROM pks_long_sample WHERE sample_time BETWEEN %s AND %s",
+                    (start, end),
+                )
+                while True:
+                    chunk = cursor.fetchmany(100_000)
+                    if not chunk:
+                        break
+                    parts.extend(chunk)
+    finally:
+        connection.close()
+    frame = pd.DataFrame(parts, columns=["b", "t", "v"])
+    frame = frame.dropna(subset=["v"]).sort_values(["b", "t"]).reset_index(drop=True)
+    print(f"   raw sampled days: {len(frame):,} rows over {len(picks)} days "
+          f"({monotonic()-t0:.0f}s)")
+    return frame
+
+
+def noise_stats(raw: pd.DataFrame) -> dict:
+    out = {}
+    for b, g in raw.groupby("b"):
+        dv = np.abs(np.diff(g["v"].to_numpy()))
+        pos = dv[dv > 1e-12]
+        med = float(np.median(dv))
+        mad0 = float(np.median(np.abs(dv - med)))
+        out[int(b)] = {
+            "n_pairs": int(len(dv)),
+            "pct_eq0": round(float((dv <= 1e-12).mean()), 4),
+            "adj_diff": {
+                "median": med,
+                "mad": mad0,
+                "delta(med+3*mad)": round(med + 3 * mad0, 4),
+                "min_pos": float(pos.min()) if pos.size else 0.0,
+                "p1_pos": round(float(np.percentile(pos, 1)), 5) if pos.size else 0.0,
+                "p50_pos": round(float(np.percentile(pos, 50)), 5) if pos.size else 0.0,
+                "p99_pos": round(float(np.percentile(pos, 99)), 5) if pos.size else 0.0,
+            },
+            "value": {
+                "min": round(float(g["v"].min()), 3),
+                "p1": round(float(g["v"].quantile(0.01)), 3),
+                "p50": round(float(g["v"].median()), 3),
+                "p99": round(float(g["v"].quantile(0.99)), 3),
+                "max": round(float(g["v"].max()), 3),
+            },
+        }
+    return out
+
+
+def series_per_batch(levels: pd.DataFrame) -> dict[int, pd.Series]:
+    out = {}
+    for b, g in levels.groupby("b"):
+        s = g.set_index("time")["lvl"]
+        s = s[~s.index.duplicated(keep="first")].sort_index()
+        out[int(b)] = s
+    return out
+
+
+def run_stats(values: np.ndarray, tau: float, minutes: int = 5) -> dict:
+    mask = np.abs(values) > tau
+    if mask.size == 0:
+        return {"count": 0, "frac": 0.0, "top": []}
+    diff = np.diff(mask.astype(np.int8))
+    starts = np.flatnonzero(diff == 1) + 1
+    ends = np.flatnonzero(diff == -1) + 1
+    if mask[0]:
+        starts = np.concatenate(([0], starts))
+    if mask[-1]:
+        ends = np.concatenate((ends, [len(mask)]))
+    if len(starts) == 0 or len(ends) == 0:
+        return {"count": 0, "frac": float(mask.mean()), "top": []}
+    ends = ends[: len(starts)]
+    durs = (ends - starts) * minutes
+    peaks = [float(np.max(np.abs(values[s:e]))) for s, e in zip(starts, ends)]
+    order = np.argsort(peaks)[::-1][:6]
+    top = [{"peak": round(peaks[i], 3), "dur_min": int(durs[i])} for i in order]
+    return {
+        "count": int(len(durs)),
+        "durs_min": [int(x) for x in durs],
+        "frac": round(float(mask.mean()), 4),
+        "top": top,
+    }
+
+
+def main() -> None:
+    out: dict = {}
+    coverage = {}
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT import_batch_id AS b, COUNT(*) AS n, COUNT(YSJ_1) AS nn, "
+                "MIN(sample_time) AS mn, MAX(sample_time) AS mx, "
+                "COUNT(DISTINCT sample_time) AS dd FROM pks_long_sample GROUP BY import_batch_id"
+            )
+            for row in cursor.fetchall():
+                b, n, nn, mn, mx, dd = row
+                span = int((mx - mn).total_seconds() // 5) + 1
+                coverage[int(b)] = {
+                    "rows": int(n), "non_null": int(nn), "distinct_t": int(dd),
+                    "span_5s_slots": span, "gap_free": n == dd == span,
+                }
+    finally:
+        connection.close()
+    out["coverage"] = coverage
+    print("coverage:", out["coverage"], flush=True)
+
+    offset = wall_offset()
+    print(f"wall offset = {offset}s (UTC -> local)")
+    levels = load_levels(offset)
+    per_batch = series_per_batch(levels)
+
+    raw = sample_raw_days()
+    out["noise_and_quantization"] = noise_stats(raw)
+
+    self_stats: dict[int, dict] = {}
+    for b in BATCHES:
+        s5 = per_batch[b].resample("5min").median().dropna()
+        local6h = s5.rolling(145, center=True, min_periods=20).median()
+        ref10d = s5.rolling(5761, center=True, min_periods=100).median()
+        e = (local6h - ref10d).dropna().to_numpy()
+        med_e = float(np.median(e))
+        sigma_e = 1.4826 * float(np.median(np.abs(e - med_e)))
+        self_stats[b] = {
+            "med": round(med_e, 4), "sigma": round(sigma_e, 4), "tau4": round(4 * sigma_e, 4),
+            "max_abs": round(float(np.abs(e).max()), 4),
+            "flag": {f"{k}s": run_stats(e, k * sigma_e) for k in (3, 4, 5, 6)},
+            "daily_level_p50": round(float(per_batch[b].resample("1D").median().median()), 4),
+            "daily_level_minmax": [
+                round(float(per_batch[b].resample("1D").median().min()), 3),
+                round(float(per_batch[b].resample("1D").median().max()), 3),
+            ],
+        }
+        print(f"self b{b}: med={self_stats[b]['med']} sigma={self_stats[b]['sigma']} "
+              f"tau4={self_stats[b]['tau4']} max={self_stats[b]['max_abs']} "
+              f"frac@4s={self_stats[b]['flag']['4s']['frac']} runs@4s={self_stats[b]['flag']['4s']['count']}")
+    out["self_ref"] = self_stats
+
+    s5_all = {b: per_batch[b].resample("5min").median() for b in BATCHES}
+    frame = pd.DataFrame(s5_all).dropna()
+    cross = {}
+    for u in BATCHES:
+        others = [o for o in BATCHES if o != u]
+        ref = frame[others].median(axis=1)
+        d = (frame[u] - ref).to_numpy()
+        med_d = float(np.median(d))
+        sigma_d = 1.4826 * float(np.median(np.abs(d - med_d)))
+        cross[int(u)] = {
+            "med": round(med_d, 4), "sigma": round(sigma_d, 4), "tau4": round(4 * sigma_d, 4),
+            "max_abs": round(float(np.abs(d).max()), 4),
+            "flag": {f"{k}s": run_stats(d, k * sigma_d) for k in (3, 4, 5, 6)},
+        }
+        print(f"cross b{u}: med={cross[u]['med']} sigma={cross[u]['sigma']} "
+              f"tau4={cross[u]['tau4']} max={cross[u]['max_abs']} "
+              f"frac@4s={cross[u]['flag']['4s']['frac']} runs@4s={cross[u]['flag']['4s']['count']}")
+    out["cross_ref"] = cross
+    out["cross_grid"] = {"common_5min": int(len(frame)),
+                         "start": str(frame.index.min()), "end": str(frame.index.max())}
+
+    print("\n===== JSON SUMMARY =====")
+    print(__import__("json").dumps(out, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

backend/diagnose_drift_running.py

@@ -0,0 +1,228 @@
+from __future__ import annotations
+
+import os
+import sys
+from time import monotonic
+
+import numpy as np
+import pandas as pd
+import pymysql
+
+sys.path.insert(0, os.path.dirname(os.path.abspath(__file__)))
+from app.config import settings  # noqa: E402
+
+BATCHES = (30, 31, 32)
+CACHE = os.path.join(
+    os.path.dirname(os.path.dirname(os.path.abspath(__file__))), "cache", "pks_minute_levels.csv"
+)
+SPEED_THRESHOLD = 300.0  # rpm: running gate
+
+
+def connect() -> pymysql.connections.Connection:
+    return pymysql.connect(
+        host=settings.db_host,
+        port=settings.db_port,
+        user=settings.db_user,
+        password=settings.db_password,
+        database=settings.db_name,
+        charset="utf8mb4",
+        cursorclass=pymysql.cursors.SSCursor,
+        connect_timeout=settings.db_connect_timeout,
+        read_timeout=3600,
+        write_timeout=120,
+    )
+
+
+def wall_offset() -> int:
+    connection = connect()
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute("SELECT sample_time, UNIX_TIMESTAMP(sample_time) AS u "
+                           "FROM pks_long_sample LIMIT 1")
+            wall, uni = cursor.fetchone()
+    finally:
+        connection.close()
+    return int(pd.Timestamp(wall).value // 10 ** 9) - int(uni)
+
+
+def load_minute_levels(offset: int) -> pd.DataFrame:
+    if os.path.exists(CACHE):
+        f = pd.read_csv(CACHE, parse_dates=["time"])
+        print(f"   loaded minute levels from cache: {len(f):,} rows")
+        return f
+    t0 = monotonic()
+    connection = connect()
+    rows = []
+    try:
+        with connection.cursor() as cursor:
+            cursor.execute(
+                "SELECT import_batch_id AS b, UNIX_TIMESTAMP(sample_time) DIV 60 AS bk, "
+                "COUNT(*) AS cnt, AVG(YSJ_1) AS lev, AVG(YSJ_41) AS rpm "
+                "FROM pks_long_sample GROUP BY import_batch_id, bk "
+                "ORDER BY import_batch_id, bk"
+            )
+            while True:
+                chunk = cursor.fetchmany(200_000)
+                if not chunk:
+                    break
+                rows.extend(chunk)
+    finally:
+        connection.close()
+    f = pd.DataFrame(rows, columns=["b", "bk", "cnt", "lev", "rpm"])
+    f["time"] = pd.to_datetime(f["bk"] * 60 + offset, unit="s")
+    f[["b", "bk", "cnt", "lev", "rpm", "time"]].to_csv(CACHE, index=False)
+    print(f"   aggregated minute levels: {len(f):,} rows in {monotonic()-t0:.0f}s")
+    return f
+
+
+def mad_std(x: np.ndarray) -> float:
+    med = float(np.median(x))
+    return 1.4826 * float(np.median(np.abs(x - med))), med
+
+
+def flag_summary(values: np.ndarray, tau: float, minute_step: int = 1) -> dict:
+    mask = np.abs(values) > tau
+    if mask.size == 0:
+        return {"count": 0, "frac": 0.0, "top": []}
+    diff = np.diff(mask.astype(np.int8))
+    starts = np.flatnonzero(diff == 1) + 1
+    ends = np.flatnonzero(diff == -1) + 1
+    if mask[0]:
+        starts = np.concatenate(([0], starts))
+    if mask[-1]:
+        ends = np.concatenate((ends, [len(mask)]))
+    if len(starts) == 0 or len(ends) == 0:
+        return {"count": 0, "frac": round(float(mask.mean()), 4), "top": []}
+    ends = ends[: len(starts)]
+    durs = (ends - starts) * minute_step
+    peaks = [float(np.max(np.abs(values[s:e]))) for s, e in zip(starts, ends)]
+    order = np.argsort(peaks)[::-1][:5]
+    return {
+        "count": int(len(durs)),
+        "durs_min": [int(x) for x in durs],
+        "frac": round(float(mask.mean()), 4),
+        "top": [{"peak": round(peaks[i], 2), "dur_min": int(durs[i])} for i in order],
+    }
+
+
+def build_run_blocks(times: pd.DatetimeIndex, running: np.ndarray) -> np.ndarray:
+    """block id per minute; merge running minutes with gaps <= 6 min."""
+    idx_run = np.flatnonzero(running)
+    if len(idx_run) == 0:
+        return np.full(len(times), -1, dtype=np.int64)
+    block = np.full(len(times), -1, dtype=np.int64)
+    bid = 0
+    prev_t = None
+    for i in idx_run:
+        if prev_t is None or (times[i] - prev_t).total_seconds() > 360:
+            bid += 1
+        block[i] = bid
+        prev_t = times[i]
+    return block
+
+
+def main() -> None:
+    offset = wall_offset()
+    f = load_minute_levels(offset)
+    per: dict[int, pd.DataFrame] = {b: g.sort_values("time").set_index("time") for b, g in f.groupby("b")}
+
+    result: dict = {}
+    for b in BATCHES:
+        df = per[b].copy()
+        df["running"] = (df["rpm"] > SPEED_THRESHOLD).to_numpy()
+        df["block"] = build_run_blocks(df.index, df["running"].to_numpy())
+        dur = df.groupby("block")["lev"].transform("size")
+        df["warmup"] = df.groupby("block").cumcount() < 120  # first 2h of each run block
+        valid = df["running"] & ~df["warmup"] & df["lev"].notna()
+        n_run = int(df["running"].sum())
+        n_block = int((df["block"] >= 0).groupby(df["block"]).ngroups)
+
+        run_levels = df.loc[df["running"], "lev"]
+        rpm_run = df.loc[df["running"], "rpm"]
+
+        # self-reference on valid (running, non-warmup) minutes
+        ser = df["lev"].where(df["running"])
+        local = ser.rolling("6h", min_periods=20).median()
+        ref = ser.rolling("20d", min_periods=200).median()
+        e_series = (local - ref).loc[valid].dropna()
+        e = e_series.to_numpy()
+
+        if len(e) == 0:
+            self_info = None
+        else:
+            sig, med = mad_std(e)
+            self_info = {
+                "n_min": int(len(e)), "med": round(med, 3), "sigma": round(sig, 3),
+                "tau4": round(4 * sig, 3), "max_abs": round(float(np.abs(e).max()), 3),
+                "flag": {f"{k}s": flag_summary(e, k * sig) for k in (3, 4, 5)},
+            }
+
+        # plateau (steady level) per run block and drift of plateau across blocks
+        plateau = (df.loc[valid, "lev"].groupby(df.loc[valid, "block"]).median())
+        pstart = df.loc[df["running"]].groupby("block")["rpm"].apply(lambda s: s.index[0])
+        prev_ref = plateau.rolling(30, min_periods=8).median().shift(1)
+        dev = (plateau - prev_ref).dropna()
+        drift = None
+        if len(dev) > 3:
+            sig_d, med_d = mad_std(dev.to_numpy())
+            flagged = dev[np.abs(dev) > 4 * sig_d]
+            drift = {
+                "n_blocks": int(len(plateau)),
+                "blocks_per_month": round(len(plateau) / 13.0, 1),
+                "plateau_p50": round(float(plateau.median()), 2),
+                "plateau_minmax": [round(float(plateau.min()), 2), round(float(plateau.max()), 2)],
+                "dev_sigma": round(sig_d, 3),
+                "dev_mad_med": round(med_d, 3),
+                "blocks_over_4sigma": int(len(flagged)),
+                "top_blocks": [
+                    {"start": str(idx)[:16], "plateau": round(float(pl), 2),
+                     "dev": round(float(dv), 2)}
+                    for idx, (pl, dv) in flagged.head(5).items()
+                ],
+            }
+
+        stats = {
+            "n_run_minutes": n_run, "run_minutes_share": round(n_run / len(df), 4),
+            "n_run_blocks": n_block, "blocks_over_24h": int((dur >= 1440).sum()),
+            "rpm_run_p50": round(float(rpm_run.median()), 0),
+            "run_level_p50": round(float(run_levels.median()), 2),
+            "run_level_minmax": [round(float(run_levels.min()), 1), round(float(run_levels.max()), 1)],
+        }
+        result[str(b)] = {"gating": stats, "self_within_run": self_info, "plateau_drift": drift}
+        print(f"[{b}] run_min={n_run:,} ({stats['run_minutes_share']:.1%}) blocks={n_block} "
+              f"rpm_p50={stats['rpm_run_p50']:.0f} run_level_p50={stats['run_level_p50']}")
+        if self_info:
+            print(f"   within-run self: sigma={self_info['sigma']} tau4={self_info['tau4']} "
+                  f"max={self_info['max_abs']} frac@4s={self_info['flag']['4s']['frac']} "
+                  f"runs@4s={self_info['flag']['4s']['count']}")
+        if drift:
+            print(f"   plateau drift: blocks={drift['n_blocks']} dev_sigma={drift['dev_sigma']} "
+                  f"over_4s={drift['blocks_over_4sigma']}")
+
+    # cross-machine on common running minutes (all three machines running)
+    frame = pd.DataFrame({str(b): per[b]["lev"] for b in BATCHES})
+    frame["rpm"] = per[30]["rpm"]
+    common = (frame["30"] > 0) & (frame["31"] > 0) & (frame["32"] > 0) & (frame["rpm"] > SPEED_THRESHOLD)
+    sub = frame.loc[common, [str(b) for b in BATCHES]]
+    cross = {}
+    if len(sub) > 1000:
+        for u in BATCHES:
+            others = [str(o) for o in BATCHES if o != u]
+            ref = sub[others].median(axis=1)
+            d = (sub[str(u)] - ref).to_numpy()
+            sig, med = mad_std(d)
+            cross[str(u)] = {
+                "n_min": int(len(d)), "med_bias": round(med, 3), "sigma": round(sig, 3),
+                "tau4": round(4 * sig, 3), "max_abs": round(float(np.abs(d).max()), 3),
+                "flag": {f"{k}s": flag_summary(d, k * sig) for k in (3, 4, 5)},
+            }
+            print(f"[cross u={u}] common_run_min={len(d):,} bias={med:.2f} sigma={sig:.3f} "
+                  f"frac@4s={cross[str(u)]['flag']['4s']['frac']} runs@4s={cross[str(u)]['flag']['4s']['count']}")
+    result["cross_common_running"] = cross
+
+    print("\n===== JSON SUMMARY =====")
+    print(__import__("json").dumps(result, indent=2, ensure_ascii=False))
+
+
+if __name__ == "__main__":
+    main()

backend/openclose/schema_device_status.sql

@@ -0,0 +1,31 @@
+-- =============================================================
+-- 开机停机表 + 运行状态列 DDL（手动执行一次即可）
+-- 说明：重复执行会因“列已存在 / 表已存在”报错，可忽略；
+--      如需重建列请先自行 DROP COLUMN。
+-- 数据填充/更新请运行 sync_device_status.py（脚本内不含 DDL）
+-- =============================================================
+
+-- 1) 开机停机表（由 开机停机表/*.xls 逐日记录合并成连续周期）
+CREATE TABLE IF NOT EXISTS machine_run_status (
+  id               BIGINT UNSIGNED NOT NULL AUTO_INCREMENT COMMENT '自增主键',
+  unit_no          INT             NOT NULL                COMMENT '机组：7/8/9',
+  import_batch_id  BIGINT UNSIGNED NOT NULL                COMMENT 'pks导入批次：30/31/32',
+  status           TINYINT         NOT NULL                COMMENT '设备状态：1开机 0关机',
+  status_start     DATETIME        NOT NULL                COMMENT '状态开始时间(年月日 时分秒)',
+  status_end       DATETIME        NOT NULL                COMMENT '状态结束时间(年月日 时分秒，开区间)',
+  start_category   VARCHAR(50)     NULL                    COMMENT '停机/启动类别（开始类别）',
+  run_hours        DECIMAL(10,2)   NULL                    COMMENT '运行时间：关机段=0，开机段=状态开始到结束小时数',
+  cum_run_hours    DECIMAL(12,2)   NULL                    COMMENT '累计运行时间（段内最大累计值）',
+  PRIMARY KEY (id),
+  KEY idx_unit_time  (unit_no, status_start, status_end),
+  KEY idx_batch_time (import_batch_id, status_start, status_end)
+) ENGINE=InnoDB DEFAULT CHARSET=utf8mb4 COLLATE=utf8mb4_0900_ai_ci
+  COMMENT='开机停机表(合并自开机停机表/*.xls)';
+
+-- 2) 长周期采样数据：增加设备状态列（0关机 1开机）
+ALTER TABLE pks_long_sample
+  ADD COLUMN device_status TINYINT NOT NULL DEFAULT 0 COMMENT '设备状态：0关机 1开机';
+
+-- 3) 波形文件表：增加设备状态列（0关机 1开机）
+ALTER TABLE wave_file
+  ADD COLUMN device_status TINYINT NOT NULL DEFAULT 0 COMMENT '设备状态：0关机 1开机';

backend/openclose/sync_device_status.py

@@ -0,0 +1,370 @@
+#!/usr/bin/env python3
+# -*- coding: utf-8 -*-
+"""
+开机停机表合并 + 数据表 device_status 全量覆盖脚本。
+
+职责（脚本内不含 DDL，建表/加列请先执行 schema_device_status.sql）：
+  1. 解析 开机停机表/{7,8,9}号机.xls，将“按天逐条”记录合并成连续的开/关机时间周期；
+  2. 重建 machine_run_status 表（DELETE 全部 + 重插，可重复执行）；
+  3. 全量覆盖 pks_long_sample.device_status（import_batch_id 30/31/32 匹配 7/8/9 号机）；
+  4. 全量覆盖 wave_file.device_status（按 device_part/point_name 文本识别机组后按时间匹配）。
+
+用法：
+  python sync_device_status.py                 # 全流程
+  python sync_device_status.py --dry-run        # 只解析Excel并打印合并段，不写库
+  python sync_device_status.py --smoke          # 冒烟：pks仅 batch=30 且 2025-04-01~04-05，wave同窗
+  python sync_device_status.py --skip-pks --skip-wave   # 只重建开机停机表
+"""
+from __future__ import annotations
+
+import argparse
+import importlib.util
+import math
+import sys
+from datetime import datetime, timedelta
+from pathlib import Path
+
+import xlrd
+import pymysql
+from pymysql.constants import CLIENT
+
+ROOT = Path(__file__).resolve().parent
+PROJECT_ROOT = ROOT.parents[1] if ROOT.name == "openclose" else ROOT
+EXCEL_DIR = PROJECT_ROOT / "开机停机表"
+STATUS_TABLE = "machine_run_status"
+
+# 机组 -> (import_batch_id)
+UNIT_BATCH = {7: 30, 8: 31, 9: 32}
+UNIT_FILES = {u: f"{u}号机.xls" for u in (7, 8, 9)}
+
+CHUNK_DAYS = 15          # pks 清零分块天数
+PKS_GLOBAL_MIN = datetime(2025, 4, 1, 0, 0, 0)
+# 上界取次日 00:00，保证把末行 2026-05-01 23:59:55 也纳入 [min,max) 更新范围
+PKS_GLOBAL_MAX = datetime(2026, 5, 2, 0, 0, 0)
+
+
+def load_settings():
+    spec = importlib.util.spec_from_file_location("app_config", PROJECT_ROOT / "backend" / "app" / "config.py")
+    mod = importlib.util.module_from_spec(spec)
+    spec.loader.exec_module(mod)
+    return mod.settings
+
+
+def connect(settings):
+    return pymysql.connect(
+        host=settings.db_host, port=settings.db_port, user=settings.db_user,
+        password=settings.db_password, database=settings.db_name, charset="utf8mb4",
+        autocommit=False, client_flag=CLIENT.FOUND_ROWS,
+        cursorclass=pymysql.cursors.DictCursor,
+    )
+
+
+# ---------------------------------------------------------------- Excel 解析
+def parse_xls(path: Path):
+    """返回该机 (unit_no, [记录])，记录为 dict；同时收集问题警告。"""
+    wb = xlrd.open_workbook(str(path))
+    sh = wb.sheet_by_index(0)
+    rows = [sh.row_values(r) for r in range(sh.nrows)]
+    header = [str(x).strip() for x in rows[0]]
+    idx = {name: i for i, name in enumerate(header)}
+    unit_no = int(path.stem.replace("号机", ""))
+    records = []
+    problems = []
+
+    for r_i, raw in enumerate(rows[1:], start=2):
+        if not raw or not any(str(c).strip() for c in raw):
+            continue
+        status_raw = str(raw[idx["设备状态"]]).strip()
+        if status_raw not in ("开机", "关机"):
+            problems.append(f"行{r_i}：无法识别状态 {status_raw!r}，已跳过")
+            continue
+        try:
+            base = xlrd.xldate_as_datetime(raw[idx["时间"]], wb.datemode)
+        except Exception:
+            problems.append(f"行{r_i}：时间无法解析 {raw[idx['时间']]!r}，已跳过")
+            continue
+        tp_raw = raw[idx["时间点"]] if idx["时间点"] is not None else 1.0
+        try:
+            tp = float(tp_raw)
+        except (TypeError, ValueError):
+            tp = 1.0
+            problems.append(f"行{r_i}：时间点无法解析 {tp_raw!r}，按 24:00 处理")
+        if tp >= 1.0:
+            dt = base + timedelta(days=1)
+        else:
+            dt = base + timedelta(seconds=int(round(tp * 86400)))
+        start_cat = str(raw[idx["启动类别"]]).strip() if idx["启动类别"] is not None else ""
+        stop_cat = str(raw[idx["停机类别"]]).strip() if idx["停机类别"] is not None else ""
+        if start_cat == "nan":
+            start_cat = ""
+        if stop_cat == "nan":
+            stop_cat = ""
+        try:
+            cum = float(raw[idx["累计运行时间"]])
+        except (TypeError, ValueError):
+            cum = None
+        records.append({
+            "unit": unit_no,
+            "dt": dt,
+            "status": 1 if status_raw == "开机" else 0,
+            "start_cat": start_cat,
+            "stop_cat": stop_cat,
+            "cum": cum,
+        })
+
+    records.sort(key=lambda r: r["dt"])
+    return unit_no, records, problems
+
+
+def build_periods(records):
+    """把逐条记录合并为连续状态周期。
+
+    状态只在“相邻记录状态发生变化”处切段；状态相同则继续合并。
+    最后一个周期结束于 Excel 末行时刻（状态结束时间截到末行）。
+    """
+    periods = []
+    changes = []          # 状态发生变化的记录（含首条），作为每段的开启点
+    for rec in records:
+        if not changes or rec["status"] != changes[-1]["status"]:
+            changes.append(rec)
+    if not changes:
+        return periods
+    last_time = records[-1]["dt"]
+    for i, chg in enumerate(changes):
+        start = chg["dt"]
+        end = changes[i + 1]["dt"] if i + 1 < len(changes) else last_time
+        if end <= start:
+            continue
+        status = chg["status"]
+        cat = chg["start_cat"] if status == 1 else chg["stop_cat"]
+        cum_vals = [r["cum"] for r in records if start <= r["dt"] < end and r["cum"] is not None]
+        cum = max(cum_vals) if cum_vals else None
+        run_hours = round((end - start).total_seconds() / 3600.0, 2) if status == 1 else 0.0
+        periods.append({
+            "unit": chg["unit"],
+            "import_batch_id": UNIT_BATCH[chg["unit"]],
+            "status": status,
+            "status_start": start,
+            "status_end": end,
+            "start_category": cat or None,
+            "run_hours": run_hours,
+            "cum_run_hours": round(cum, 2) if cum is not None else None,
+        })
+    return periods
+
+
+def validate_periods(periods):
+    warns = []
+    per_unit = {}
+    for p in periods:
+        per_unit.setdefault(p["unit"], []).append(p)
+    for u, ps in sorted(per_unit.items()):
+        on_days = sum((p["status_end"] - p["status_start"]).total_seconds() for p in ps if p["status"] == 1) / 86400.0
+        off_days = sum((p["status_end"] - p["status_start"]).total_seconds() for p in ps if p["status"] == 0) / 86400.0
+        prev = None
+        for p in ps:
+            if prev and p["status"] == prev["status"]:
+                warns.append(f"{u}号机：相邻两段状态未交替 {prev['status_start']}->{p['status_start']}")
+            if prev and p["status_start"] < prev["status_end"]:
+                warns.append(f"{u}号机：时间段重叠 {prev['status_start']}..{prev['status_end']} 与 {p['status_start']}")
+            prev = p
+        print(f"  {u}号机：{len(ps)} 段，开机 {on_days:.1f} 天 / 关机 {off_days:.1f} 天，"
+              f"区间 {ps[0]['status_start']:%Y-%m-%d %H:%M} ~ {ps[-1]['status_end']:%Y-%m-%d %H:%M}")
+    return warns
+
+
+# ---------------------------------------------------------------- DB 更新
+def rebuild_status_table(conn, periods):
+    cur = conn.cursor()
+    cur.execute(f"DELETE FROM {STATUS_TABLE}")
+    sql = (f"INSERT INTO {STATUS_TABLE} "
+           f"(unit_no, import_batch_id, status, status_start, status_end, "
+           f"start_category, run_hours, cum_run_hours) "
+           f"VALUES (%(unit)s, %(import_batch_id)s, %(status)s, %(status_start)s, "
+           f"%(status_end)s, %(start_category)s, %(run_hours)s, %(cum_run_hours)s)")
+    data = [
+        {
+            "unit": p["unit"],
+            "import_batch_id": p["import_batch_id"],
+            "status": p["status"],
+            "status_start": p["status_start"],
+            "status_end": p["status_end"],
+            "start_category": p["start_category"],
+            "run_hours": p["run_hours"],
+            "cum_run_hours": p["cum_run_hours"],
+        }
+        for p in periods
+    ]
+    cur.executemany(sql, data)
+    conn.commit()
+    print(f"machine_run_status 重建完成，共 {cur.rowcount} 条（已先清空）")
+
+
+def fetch_periods(conn):
+    cur = conn.cursor()
+    cur.execute(f"SELECT unit_no, status, status_start, status_end FROM {STATUS_TABLE} "
+                f"ORDER BY unit_no, status_start")
+    per = {}
+    for r in cur.fetchall():
+        per.setdefault(r["unit_no"], []).append(r)
+    return per
+
+
+def _clip(s, e, lo, hi):
+    return max(s, lo), min(e, hi)
+
+
+def _grid_count(a, b):
+    """区间 [a,b) 内落在 5 秒采样网格上的点数（网格=epoch秒可被5整除）。"""
+    a_s, b_s = int(a.timestamp()), int(b.timestamp())
+    k0 = math.ceil(a_s / 5)
+    k1 = math.floor((b_s - 1) / 5)
+    return max(0, k1 - k0 + 1)
+
+
+def update_pks(conn, per, lo=None, hi=None, only_batch=None, chunk_days=CHUNK_DAYS):
+    cur = conn.cursor()
+    lo = lo or PKS_GLOBAL_MIN
+    hi = hi or PKS_GLOBAL_MAX
+    for unit in sorted(UNIT_BATCH):
+        batch = UNIT_BATCH[unit]
+        if only_batch is not None and batch != only_batch:
+            continue
+        on_periods = [p for p in per.get(unit, []) if p["status"] == 1]
+        t = lo
+        n_clear = 0
+        while t < hi:
+            c1 = min(t + timedelta(days=chunk_days), hi)
+            cur.execute("UPDATE pks_long_sample SET device_status=0 "
+                        "WHERE import_batch_id=%s AND sample_time>=%s AND sample_time<%s",
+                        (batch, t, c1))
+            n_clear += cur.rowcount
+            conn.commit()
+            t = c1
+        expected = 0
+        for p in on_periods:
+            s, e = _clip(p["status_start"], p["status_end"], lo, hi)
+            if e <= s:
+                continue
+            expected += _grid_count(s, e)
+            cur.execute("UPDATE pks_long_sample SET device_status=1 "
+                        "WHERE import_batch_id=%s AND sample_time>=%s AND sample_time<%s",
+                        (batch, s, e))
+            conn.commit()
+        cur.execute("SELECT COUNT(*) n FROM pks_long_sample WHERE import_batch_id=%s AND device_status=1",
+                    (batch,))
+        actual = cur.fetchone()["n"]
+        print(f"  pks batch{batch}({unit}号机)：清零 {n_clear} 行，置1后 {actual} 行（理论 {expected}，差 {actual - expected}）")
+
+
+def _unit_expr():
+    """由文本识别机组（7/8/9号机）；识别不到返回 NULL。"""
+    p = "CONCAT(IFNULL(device_part,''),' ',IFNULL(point_name,''),' ',IFNULL(device_code,''))"
+    return (
+        "CASE "
+        f"WHEN {p} LIKE '%9号机组%' THEN 9 "
+        f"WHEN {p} LIKE '%8号机组%' THEN 8 "
+        f"WHEN {p} LIKE '%7号机组%' THEN 7 "
+        "ELSE NULL END"
+    )
+
+
+def update_wave(conn, per, lo=None, hi=None):
+    cur = conn.cursor()
+    if lo and hi:
+        wnd = "sample_time>=%s AND sample_time<%s"
+        args = (lo, hi)
+        cur.execute("UPDATE wave_file SET device_status=0 WHERE " + wnd, args)
+    else:
+        args = ()
+        cur.execute("UPDATE wave_file SET device_status=0")
+    print(f"  wave_file 已清零 {cur.rowcount} 行")
+    conn.commit()
+
+    # 注：SQL 内含字面 %(LIKE)，不能用 %s 参数格式化，窗口条件直接内联
+    expr = _unit_expr()
+    sql_set = (
+        "UPDATE wave_file w "
+        "JOIN machine_run_status m ON m.status=1 "
+        "  AND w.sample_time>=m.status_start AND w.sample_time<m.status_end "
+        f"  AND ({expr})=m.unit_no "
+        "SET w.device_status=1 "
+    )
+    if lo and hi:
+        sql_set += (f"WHERE w.sample_time>='{lo:%Y-%m-%d %H:%M:%S}' "
+                    f"AND w.sample_time<'{hi:%Y-%m-%d %H:%M:%S}' ")
+    cur.execute(sql_set)
+    conn.commit()
+    print(f"  wave_file 置1完成 {cur.rowcount} 行")
+
+
+# ---------------------------------------------------------------- main
+def main():
+    ap = argparse.ArgumentParser(description="开机停机表合并 + device_status 全量覆盖")
+    ap.add_argument("--dry-run", action="store_true", help="只解析Excel并打印，不写库")
+    ap.add_argument("--smoke", action="store_true", help="冒烟：pks仅batch=30，窗口2025-04-01~04-05")
+    ap.add_argument("--skip-status-table", action="store_true", help="跳过重建 machine_run_status")
+    ap.add_argument("--skip-pks", action="store_true", help="跳过 pks_long_sample 更新")
+    ap.add_argument("--skip-wave", action="store_true", help="跳过 wave_file 更新")
+    args = ap.parse_args()
+
+    all_periods = []
+    all_problems = []
+    for u in (7, 8, 9):
+        f = EXCEL_DIR / UNIT_FILES[u]
+        if not f.exists():
+            print(f"[跳过] 缺少 {f.name}")
+            continue
+        unit, records, problems = parse_xls(f)
+        all_problems += [f"{unit}号机：{x}" for x in problems]
+        periods = build_periods(records)
+        all_periods += periods
+        print(f"{unit}号机.xls：{len(records)} 条原始记录 -> {len(periods)} 个合并周期")
+    for p in all_problems:
+        print("  [警告]", p)
+    if not all_periods:
+        print("没有可用的合并周期，退出")
+        sys.exit(1)
+
+    print("---- 合并结果预览（每台 前2/后2 段）----")
+    for p in all_periods[:2]:
+        print("  ", p)
+    print("   ...")
+    for p in all_periods[-2:]:
+        print("  ", p)
+
+    print("---- 连续性/交替性校验 ----")
+    for w in validate_periods(all_periods):
+        print("  [警告]", w)
+
+    if args.dry_run:
+        print("[dry-run] 结束，未写库")
+        return
+
+    settings = load_settings()
+    conn = connect(settings)
+    try:
+        if not args.skip_status_table:
+            rebuild_status_table(conn, all_periods)
+        per = fetch_periods(conn)
+
+        smoke_lo = smoke_hi = None
+        if args.smoke:
+            smoke_lo = datetime(2025, 4, 29)
+            smoke_hi = datetime(2025, 5, 4)
+            print("---- 冒烟模式：pks batch=30，窗口 2025-04-29 ~ 2025-05-04 ----")
+
+        if not args.skip_pks:
+            print("---- 更新 pks_long_sample.device_status ----")
+            update_pks(conn, per, lo=smoke_lo, hi=smoke_hi, only_batch=30 if args.smoke else None)
+
+        if not args.skip_wave:
+            print("---- 更新 wave_file.device_status ----")
+            update_wave(conn, per, lo=smoke_lo, hi=smoke_hi)
+    finally:
+        conn.close()
+    print("完成")
+
+
+if __name__ == "__main__":
+    main()

frontend/src/components/WaveChart.vue

@@ -45,6 +45,81 @@ const pvPhaseColors = [
 const modalChartWidth = 1060
 const modalChartHeight = 440
 
+type RowKind = 'device' | 'site' | 'siteAgg' | 'second' | 'volume'
+type RowMeta = { name: string; kind: RowKind; siteSeq: number }
+
+const SITE_AGG_NAME = '全场点位（归一化）'
+const SITE_AGG_PREFIX = '全场点位·'
+
+// PKS 全场点位按物理分类规范 y 轴：同分类所有子图用同一刻度步长(step)，
+// 每个子图范围由各自数据自适应起点/终点；波动过小时用 minSpan 保证最小显示幅度。
+type SiteScaleRule = { unit: string; step: number; minSpan: number }
+const SITE_SCALE_RULES: Record<string, SiteScaleRule> = {
+  温度: { unit: '℃', step: 5, minSpan: 10 },
+  压力: { unit: 'MPa', step: 1, minSpan: 2 },
+  振动: { unit: 'mm/s', step: 0.5, minSpan: 2 },
+  电流: { unit: 'A', step: 10, minSpan: 20 },
+  频率: { unit: 'Hz', step: 5, minSpan: 10 },
+  阀位: { unit: '%', step: 5, minSpan: 10 },
+}
+
+function siteScaleRuleFor(description: string | undefined): SiteScaleRule | null {
+  const text = description ?? ''
+  if (!text) return null
+  if (text.includes('温度') || text.includes('RTD')) return SITE_SCALE_RULES['温度']
+  if (text.includes('压力')) return SITE_SCALE_RULES['压力']
+  if (text.includes('振动')) return SITE_SCALE_RULES['振动']
+  if (text.includes('电流')) return SITE_SCALE_RULES['电流']
+  if (text.includes('频率')) return SITE_SCALE_RULES['频率']
+  if (text.includes('阀位')) return SITE_SCALE_RULES['阀位']
+  return null
+}
+
+function siteAxisRange(rule: SiteScaleRule, values: number[]): { min: number; max: number; interval: number } | null {
+  const finite = values.filter(Number.isFinite)
+  if (!finite.length) return null
+  let min = Math.min(...finite)
+  let max = Math.max(...finite)
+  if (!(max > min)) {
+    min -= rule.minSpan / 2
+    max += rule.minSpan / 2
+  }
+  if (max - min < rule.minSpan) {
+    const center = (max + min) / 2
+    min = center - rule.minSpan / 2
+    max = center + rule.minSpan / 2
+  }
+  min = Math.floor(min / rule.step) * rule.step
+  max = Math.ceil(max / rule.step) * rule.step
+  if (!(max > min)) max = min + rule.step
+  return { min, max, interval: rule.step }
+}
+
+const SPLIT_ROW_HEIGHT = 120
+const SPLIT_ROW_GAP = 14
+const CHART_TOP_INSET = 32
+const CHART_BOTTOM_INSET = 68
+const MERGE_CHART_HEIGHT = 560
+const EMPTY_CHART_HEIGHT = 480
+
+function buildRowMetas(data: WaveWindowResponse): RowMeta[] {
+  const metas: RowMeta[] = data.devicePoints.map((name) => ({ name, kind: 'device' as const, siteSeq: -1 }))
+  const siteMetas: RowMeta[] = []
+  ;(data.siteSeries?.series.filter((series) => series.data.length) ?? []).forEach((series, index) => {
+    siteMetas.push({ name: series.itemName, kind: 'site' as const, siteSeq: index })
+  })
+  metas.push(...siteMetas)
+  if (siteMetas.length > 1) metas.push({ name: SITE_AGG_NAME, kind: 'siteAgg' as const, siteSeq: -1 })
+  if (data.secondSeries.data.length) metas.push({ name: '周期数据', kind: 'second' as const, siteSeq: -1 })
+  if (data.volumeSeries.data.length || data.volumeSeries.info) metas.push({ name: '体积', kind: 'volume' as const, siteSeq: -1 })
+  return metas
+}
+
+function splitTotalHeight(rowCount: number) {
+  if (rowCount <= 0) return EMPTY_CHART_HEIGHT
+  return CHART_TOP_INSET + CHART_BOTTOM_INSET + rowCount * SPLIT_ROW_HEIGHT + (rowCount - 1) * SPLIT_ROW_GAP
+}
+
 function niceAxisExtent(values: number[]): [number, number] {
   const dataMin = Math.min(...values)
   const dataMax = Math.max(...values)
@@ -59,6 +134,14 @@ function niceAxisExtent(values: number[]): [number, number] {
   return [Math.floor(min / interval) * interval, Math.ceil(max / interval) * interval]
 }
 
+function rowColor(data: WaveWindowResponse, name: string): string {
+  const meta = buildRowMetas(data).find((item) => item.name === name)
+  if (meta?.kind === 'site') return SITE_POINT_COLORS[meta.siteSeq % SITE_POINT_COLORS.length]
+  const seriesMatch = data.series.find((series) => series.devicePoint === name)
+  if (seriesMatch) return seriesMatch.color
+  return colors[name] ?? '#60717b'
+}
+
 function fixedYAxis(devicePoint: string) {
   if (props.mode === 'merge') return null
   const series = props.data?.series.find((item) => item.devicePoint === devicePoint)
@@ -67,13 +150,15 @@ function fixedYAxis(devicePoint: string) {
 }
 
 function splitExtraCount(data: WaveWindowResponse): number {
-  let count = 0
-  if (data.secondSeries.data.length) count += 1
-  if (data.volumeSeries.data.length || data.volumeSeries.info) count += 1
-  if (data.siteSeries?.series.some((series) => series.data.length)) count += 1
-  return count
+  return Math.max(0, buildRowMetas(data).length - data.devicePoints.length)
 }
 
+const layoutHeightPx = computed(() => {
+  if (props.mode === 'merge') return props.data ? MERGE_CHART_HEIGHT : EMPTY_CHART_HEIGHT
+  if (!props.data) return EMPTY_CHART_HEIGHT
+  return splitTotalHeight(buildRowMetas(props.data).length)
+})
+
 const plottedPointCount = computed(() => (
   (props.data?.series.reduce((total, series) => total + series.data.length, 0) ?? 0)
   + (props.data?.secondSeries.data.length ?? 0)
@@ -175,7 +260,7 @@ function annotationAreas(annotations: Annotation[]): any[] {
 function buildAnnotationOverlaySeries(annotations: Annotation[]): echarts.SeriesOption[] {
   const data = props.data
   if (!data) return []
-  const gridCount = props.mode === 'merge' ? 1 : data.devicePoints.length + splitExtraCount(data)
+  const gridCount = props.mode === 'merge' ? 1 : buildRowMetas(data).length
   const areas = annotationAreas(annotations)
   return Array.from({ length: gridCount }, (_, index) => ({
     id: `annotation-overlay-${index}`,
@@ -262,19 +347,15 @@ function buildOption(zoomMode: 'initial' | 'keep' = 'initial') {
   const showVolume = data.volumeSeries.data.length > 0 || data.volumeSeries.info != null
   const siteSeries = data.siteSeries?.series.filter((series) => series.data.length) ?? []
   const showSite = siteSeries.length > 0
-  const extraRows: string[] = []
-  if (showSite) extraRows.push('全场点位')
-  if (showSecond) extraRows.push('周期数据')
-  if (showVolume) extraRows.push('体积')
-  const rows = props.mode === 'merge' ? ['合并信号'] : [...devicePoints, ...extraRows]
+  const metas = props.mode === 'merge' ? [] : buildRowMetas(data)
+  const rows = props.mode === 'merge' ? ['合并信号'] : metas.map((meta) => meta.name)
   const gridCount = rows.length
-  const chartHeight = chartElement.value?.clientHeight || 640
-  const topInset = 32
-  const bottomInset = 68
-  const rowGap = props.mode === 'merge' ? 0 : 14
+  const topInset = CHART_TOP_INSET
+  const bottomInset = CHART_BOTTOM_INSET
+  const rowGap = props.mode === 'merge' ? 0 : SPLIT_ROW_GAP
   const rowHeight = props.mode === 'merge'
-    ? Math.max(260, chartHeight - topInset - bottomInset)
-    : Math.max(76, Math.floor((chartHeight - topInset - bottomInset - rowGap * (gridCount - 1)) / gridCount))
+    ? Math.max(260, MERGE_CHART_HEIGHT - topInset - bottomInset)
+    : SPLIT_ROW_HEIGHT
   const grid = Array.from({ length: gridCount }, (_, index) => ({
     left: 70,
     right: 22,
@@ -302,28 +383,46 @@ function buildOption(zoomMode: 'initial' | 'keep' = 'initial') {
     splitLine: { show: false },
   }))
   const yAxes = rows.map((rowName, index) => {
-    const isDevicePoint = (devicePoints as readonly string[]).includes(rowName)
-    const seriesColor = isDevicePoint
-      ? (data.series.find((series) => series.devicePoint === rowName)?.color ?? '#60717b')
-      : (colors[rowName] ?? '#60717b')
+    const meta = metas.find((item) => item.name === rowName)
+    const isDevicePoint = meta?.kind === 'device'
+    const seriesColor = rowColor(data, rowName)
     const fixed = isDevicePoint ? fixedYAxis(rowName) : null
     const isDisplacement = isDevicePoint
       ? data.series.find((series) => series.devicePoint === rowName)?.measurementType === '位移'
       : false
+    let siteRule: SiteScaleRule | null = null
+    let siteRange: { min: number; max: number; interval: number } | null = null
+    let axisName = rowName
+    if (meta?.kind === 'site') {
+      const rule = siteScaleRuleFor(props.siteDescriptions?.[rowName])
+      if (rule) {
+        siteRule = rule
+        axisName = `${rowName} ${rule.unit}`
+        const siteValues = data.siteSeries?.series.find((series) => series.itemName === rowName)?.data
+          .map((item) => item.value[1]) ?? []
+        siteRange = siteAxisRange(rule, siteValues)
+      }
+    } else if (meta?.kind === 'siteAgg') {
+      siteRange = { min: 0, max: 1, interval: 0.2 }
+    }
     return {
       type: 'value' as const,
       gridIndex: index,
-      name: rowName,
+      name: axisName,
       nameLocation: 'middle' as const,
-      nameGap: 48,
+      nameGap: 52,
       nameTextStyle: { color: seriesColor, fontWeight: 600 },
       axisLine: { show: true, lineStyle: { color: seriesColor } },
-      axisLabel: { color: '#71808a', fontSize: 11 },
+      axisLabel: {
+        color: '#71808a',
+        fontSize: 11,
+        formatter: meta?.kind === 'siteAgg' ? (value: number) => Number(value).toFixed(1) : undefined,
+      },
       splitLine: { show: true, lineStyle: { color: '#e7edf0', width: 1 } },
-      scale: isDisplacement,
-      min: fixed?.min,
-      max: fixed?.max,
-      interval: fixed?.interval,
+      scale: meta?.kind === 'site' ? !siteRule : isDisplacement,
+      min: siteRange?.min ?? fixed?.min,
+      max: siteRange?.max ?? fixed?.max,
+      interval: siteRange?.interval ?? fixed?.interval,
     }
   })
   const series: echarts.SeriesOption[] = []
@@ -400,10 +499,10 @@ function buildOption(zoomMode: 'initial' | 'keep' = 'initial') {
           type: 'line',
           xAxisIndex: 0,
           yAxisIndex: 0,
-          showSymbol: true,
-          symbol: 'circle',
-          symbolSize: 6,
-          lineStyle: { width: 1.5, color, opacity: 0.8 },
+          showSymbol: false,
+          connectNulls: false,
+          sampling: 'lttb' as const,
+          lineStyle: { width: 1.4, color, opacity: 0.8 },
           itemStyle: { color },
           data: siteItem.data
             .filter((item) => Number.isFinite(item.x) && Number.isFinite(item.value[1]))
@@ -412,85 +511,108 @@ function buildOption(zoomMode: 'initial' | 'keep' = 'initial') {
       })
     }
   } else {
-    devicePoints.forEach((devicePoint, index) => {
-      const source = data.series.find((series) => series.devicePoint === devicePoint)
-      const color = source?.color ?? colors[source?.measurementType ?? '加速度']
-      series.push({
-        name: devicePoint,
-        type: 'line',
-        z: 10,
-        xAxisIndex: index,
-        yAxisIndex: index,
-        showSymbol: false,
-        connectNulls: false,
-        sampling: 'lttb' as const,
-        lineStyle: { width: 2.5, color, cap: 'round', join: 'round' },
-        itemStyle: { color },
-        data: source?.data
-          .filter((item) => Number.isFinite(item.x) && Number.isFinite(item.rawValue))
-          .map((item) => [item.x, item.rawValue] as [number, number]) ?? [],
-        markArea: { silent: true, data: periodBackground },
-        markLine: index === 0 ? { silent: true, symbol: 'none', data: triggerLines(data.triggerXs) } : undefined,
-      })
-    })
-    if (showSite) {
-      const siteIndex = devicePoints.length
-      siteSeries.forEach((siteItem, index) => {
-        const color = SITE_POINT_COLORS[index % SITE_POINT_COLORS.length]
+    metas.forEach((meta, index) => {
+      if (meta.kind === 'device') {
+        const source = data.series.find((series) => series.devicePoint === meta.name)
+        const color = source?.color ?? colors[source?.measurementType ?? '加速度']
         series.push({
-          name: siteItem.itemName,
+          name: meta.name,
           type: 'line',
           z: 10,
-          xAxisIndex: siteIndex,
-          yAxisIndex: siteIndex,
-          showSymbol: true,
-          symbol: 'circle',
-          symbolSize: 6,
-          lineStyle: { width: 1.5, color, opacity: 0.8 },
+          xAxisIndex: index,
+          yAxisIndex: index,
+          showSymbol: false,
+          connectNulls: false,
+          sampling: 'lttb' as const,
+          lineStyle: { width: 2.5, color, cap: 'round', join: 'round' },
+          itemStyle: { color },
+          data: source?.data
+            .filter((item) => Number.isFinite(item.x) && Number.isFinite(item.rawValue))
+            .map((item) => [item.x, item.rawValue] as [number, number]) ?? [],
+          markArea: { silent: true, data: periodBackground },
+          markLine: index === 0 ? { silent: true, symbol: 'none', data: triggerLines(data.triggerXs) } : undefined,
+        })
+      } else if (meta.kind === 'site') {
+        const siteItem = siteSeries.find((series) => series.itemName === meta.name)
+        if (!siteItem) return
+        const color = SITE_POINT_COLORS[meta.siteSeq % SITE_POINT_COLORS.length]
+        series.push({
+          name: meta.name,
+          type: 'line',
+          z: 10,
+          xAxisIndex: index,
+          yAxisIndex: index,
+          showSymbol: false,
+          connectNulls: false,
+          sampling: 'lttb' as const,
+          lineStyle: { width: 1.6, color, opacity: 0.9 },
           itemStyle: { color },
           data: siteItem.data
             .filter((item) => Number.isFinite(item.x) && Number.isFinite(item.value[1]))
             .map((item) => [item.x, item.value[1]] as [number, number]),
           markArea: { silent: true, data: periodBackground },
         })
-      })
-    }
-    if (showSecond) {
-      const secondIndex = devicePoints.length + (showSite ? 1 : 0)
-      series.push({
-        name: '周期数据',
-        type: 'line',
-        z: 10,
-        xAxisIndex: secondIndex,
-        yAxisIndex: secondIndex,
-        showSymbol: false,
-        connectNulls: false,
-        lineStyle: { width: 3, color: colors['周期数据'], cap: 'round', join: 'round' },
-        areaStyle: { color: 'rgba(245, 108, 108, 0.16)' },
-        itemStyle: { color: colors['周期数据'] },
-        data: data.secondSeries.data
-          .filter((item) => Number.isFinite(item.x) && Number.isFinite(item.rawValue))
-          .map((item) => [item.x, item.rawValue] as [number, number]),
-        markArea: { silent: true, data: periodBackground },
-      })
-    }
-    if (showVolume) {
-      const volumeIndex = devicePoints.length + (showSite ? 1 : 0) + (showSecond ? 1 : 0)
-      series.push({
-        name: '体积',
-        type: 'line',
-        z: 10,
-        xAxisIndex: volumeIndex,
-        yAxisIndex: volumeIndex,
-        showSymbol: false,
-        lineStyle: { width: 2, color: colors['体积'] },
-        itemStyle: { color: colors['体积'] },
-        data: data.volumeSeries.data
-          .filter((item) => Number.isFinite(item.x) && Number.isFinite(item.volume))
-          .map((item) => [item.x, item.volume] as [number, number]),
-        markArea: { silent: true, data: periodBackground },
-      })
-    }
+      } else if (meta.kind === 'siteAgg') {
+        siteSeries.forEach((siteItem) => {
+          const siteValues = siteItem.data
+            .filter((item) => Number.isFinite(item.value[1]))
+            .map((item) => item.value[1])
+          if (!siteValues.length) return
+          const siteMin = Math.min(...siteValues)
+          const siteMax = Math.max(...siteValues)
+          const siteSpan = siteMax - siteMin
+          const siteIndex = siteSeries.findIndex((series) => series.itemName === siteItem.itemName)
+          const color = SITE_POINT_COLORS[siteIndex % SITE_POINT_COLORS.length]
+          series.push({
+            name: `${SITE_AGG_PREFIX}${siteItem.itemName}`,
+            type: 'line',
+            z: 9,
+            xAxisIndex: index,
+            yAxisIndex: index,
+            showSymbol: false,
+            connectNulls: false,
+            sampling: 'lttb' as const,
+            lineStyle: { width: 1.2, color, opacity: 0.85 },
+            itemStyle: { color },
+            data: siteItem.data
+              .filter((item) => Number.isFinite(item.x) && Number.isFinite(item.value[1]))
+              .map((item) => [item.x, siteSpan > 0 ? (item.value[1] - siteMin) / siteSpan : 0.5] as [number, number]),
+            markArea: { silent: true, data: periodBackground },
+          })
+        })
+      } else if (meta.kind === 'second') {
+        series.push({
+          name: '周期数据',
+          type: 'line',
+          z: 10,
+          xAxisIndex: index,
+          yAxisIndex: index,
+          showSymbol: false,
+          connectNulls: false,
+          lineStyle: { width: 3, color: colors['周期数据'], cap: 'round', join: 'round' },
+          areaStyle: { color: 'rgba(245, 108, 108, 0.16)' },
+          itemStyle: { color: colors['周期数据'] },
+          data: data.secondSeries.data
+            .filter((item) => Number.isFinite(item.x) && Number.isFinite(item.rawValue))
+            .map((item) => [item.x, item.rawValue] as [number, number]),
+          markArea: { silent: true, data: periodBackground },
+        })
+      } else if (meta.kind === 'volume') {
+        series.push({
+          name: '体积',
+          type: 'line',
+          z: 10,
+          xAxisIndex: index,
+          yAxisIndex: index,
+          showSymbol: false,
+          lineStyle: { width: 2, color: colors['体积'] },
+          itemStyle: { color: colors['体积'] },
+          data: data.volumeSeries.data
+            .filter((item) => Number.isFinite(item.x) && Number.isFinite(item.volume))
+            .map((item) => [item.x, item.volume] as [number, number]),
+        })
+      }
+    })
   }
 
   const zoom = zoomMode === 'keep' ? readCurrentZoom(data) : initialZoom(data)
@@ -516,15 +638,23 @@ function buildOption(zoomMode: 'initial' | 'keep' = 'initial') {
         params.forEach((item) => {
           const value = item.value?.[1]
           if (value !== undefined && value !== null) {
-            const seriesName = props.siteDescriptions?.[item.seriesName] || item.seriesName
-            lines.push(`<span style="color:${item.color}">●</span> ${seriesName}: ${Number(value).toPrecision(7)}`)
+            const fullName = String(item.seriesName ?? '')
+            const isAgg = fullName.startsWith(SITE_AGG_PREFIX)
+            const base = isAgg ? fullName.slice(SITE_AGG_PREFIX.length) : fullName
+            const seriesName = props.siteDescriptions?.[base] || base
+            lines.push(
+              `<span style="color:${item.color}">●</span> ${seriesName}${isAgg ? ' (归一化)' : ''}: `
+              + `${Number(value).toPrecision(isAgg ? 3 : 7)}`,
+            )
           }
         })
         return lines.join('<br/>')
       },
     },
     legend: {
-      data: [...devicePoints, ...extraRows, ...siteSeries.map((series) => series.itemName)],
+      data: props.mode === 'merge'
+        ? [...devicePoints, ...siteSeries.map((series) => series.itemName), ...(showSecond ? ['周期数据'] : []), ...(showVolume ? ['体积'] : [])]
+        : metas.filter((meta) => meta.kind !== 'siteAgg').map((meta) => meta.name),
       formatter: (name: string) => props.siteDescriptions?.[name] || name,
       top: 0,
       left: 70,
@@ -794,7 +924,7 @@ onBeforeUnmount(() => {
         <canvas :ref="(el) => setPvCanvas(el, rowIndex)" class="pv-preview-canvas" :aria-label="row.title"></canvas>
       </div>
     </div>
-    <div ref="chartElement" class="wave-chart" :class="{ 'is-merge': mode === 'merge' }"></div>
+    <div ref="chartElement" class="wave-chart" :class="{ 'is-merge': mode === 'merge' }" :style="{ height: `${layoutHeightPx}px` }"></div>
     <div v-if="annotations?.length" class="annotation-nav">
       <div class="annotation-nav-head">
         <span class="annotation-nav-title">标注导航</span>

异常检测方案讨论.md

@@ -0,0 +1,48 @@
+# PKS 点位异常检测 —— 方案讨论稿（v0.2）
+
+> 状态：待讨论，未实施。记录已达成共识 + 备选思路 + 遗留问题。
+
+## 一、已确认的基础事实
+
+- 分析单元 = `(import_batch_id, YSJ_k)`，30/31/32 ↔ 7/8/9 号机。
+- 41 列物理量不同：温度℃ / 压力MPa / 振动mm/s / 电流A / 频率Hz / 阀位% / 转速RPM，**阈值一律逐列自学习，不跨列统一**。
+- 剔除备用测点：YSJ 33 / 34 / 35（备用RTD、备用AI）。
+- 运行状态权威来源 = `开机停机表/8号机.xls、9号机.xls`（事件级日志：时间=日期，时间点=当天小数时刻，状态=开机/关机）；7号机暂无表。
+- 运行状态退路（表缺失或不可信时）：YSJ_41 转速(≈995=开机) + wave_file 时间戳。
+- 采样：pks_long_sample 主键 (import_batch_id, sample_time)，5s 一条；窗口 2025-04-01 ~ 2026-05-01，覆盖完整无空。
+- 季节/慢水平变化处理：采用**滚动参照**（最近 N 个运行日的中位数 + MAD 作正常带），随季节漂移，不追坏段。
+
+## 二、数据读取（性能）
+
+- 不做逐列全量 20M 行扫描。
+- 一次性服务端分钟级聚合全部启用列（AVG/MIN/MAX/COUNT + 运行标记），落 `cache/`，之后 38 列全部在本地分钟级序列上分析。
+
+## 三、推荐方案：运行期多列异常初筛（分层）
+
+1. **运行时间轴**：解析开机停机表 → 每机运行区间；7号用 rpm/wave_file 推断；启停过渡段单独标记。
+2. **点位字典**：按 site_point 建 YSJ_n → 单位/描述/类别；明显标错的 unit 以实测修正；剔除 33-35。
+3. **每列正常模型**：只统计运行段 → 运行日指标（日水平中位数、波动跨度、日内突变强度）；滚动参照 + MAD 得季节自适应正常带。
+4. **两层事件识别**：
+   - 慢速/水平类：日级 z 分（|z|>4 且持续若干运行日）。
+   - 快速/突发类：分钟级速率/越界，该列自校准 + 滞回定界。
+   - 可选补充：同点跨机差分（三台同跑时差掉环境，只留单机偏差）。
+5. **排序与人工确认**：每列每年出 top-N；跨机同向变化降权（判环境/负荷），单机单独优先；前端高亮原始曲线；人工标注回写，坏段不进正常模型。
+
+## 四、备选/待议：时间窗口无监督模型
+
+- 结论：能做但**不能靠它抵抗季节**。季节鲁棒性取决于“参照”：
+  - 无参照（学全量分布）→ 扛不住季节；
+  - 全局最近邻 → 夏季窗口会被去年同期掩护，反而漏检；
+  - 局部/近期参照 + 先季节去趋势 → 扛得住快变。
+- 慢漂移在统计上无法与“合法慢变化”区分，换任何无监督模型也一样，需外部参照或标注兜底。
+- 纯无监督的坑：间歇运行需先按运行态硬分割；38 列异质需标准化/分组；分数不可解释、无天然段边界、需滞回定界；超参校准成本高。
+- 建议角色：作为**组合层**（对“去趋势残差 + 每列统计量”做多测点联合异常检测 + 排序），不是季节与运行态问题的替代品。
+
+## 五、遗留问题（下次讨论）
+
+1. 开机后过渡段（爬坡）算正常物理还是异常候选？建议：稳态分析排除，启动特性单独一条。
+2. 8号机窗口内仅约 96 天开机（长停偶开），确认是真实工况还是采集问题（决定“正常模型”可学多少运行日）。
+3. 初筛范围：38 列全含，还是先温度+压力两类跑通？
+4. 输出形态：事件库 + 每列 top-N 清单/图表，还是整表 excel？
+5. 人工标注回写是否第一版就纳入闭环？
+6. 是否需要先解析 8/9号 正式运行区间、7号 rpm 推断区间并做一致性核对，再进入建模？