backend/LabelingPreTraining/predict.py

@@ -1,11 +1,13 @@
 """Step 3: long-running TSPulse anomaly detection poller.
 
 Polls the wave_file table for rows matching ``rpm > 0 AND measurement_type =
-'压力'``, reads only the first 3000 wave_sample rows per file, locates the
-first complete crankshaft cycle, and encodes it with the frozen TSPulse model
-into a 128-dim fingerprint. The fingerprint is compared against the part's
-centroid and radius from the baseline table and ``tspluse_status`` is written
-back as a proportional integer of the centroid distance:
+'压力'``, reads each file's samples with an adaptive multi-stage window
+(``STAGE_LIMITS`` = 3500 -> 7000 -> 14000 rows), escalating until a complete
+crankshaft cycle is found or the file's data is exhausted, and encodes the
+first complete cycle with the frozen TSPulse model into a 128-dim fingerprint.
+The fingerprint is compared against the part's centroid and radius from the
+baseline table and ``tspluse_status`` is written back as a proportional
+integer of the centroid distance:
 
 * 无周期 (no complete cycle found) -> ``-1``
 * 正常 (distance <= 2 x radius)   -> ``0``
@@ -15,18 +17,21 @@ Every processed file is written back, so a full re-run from an empty watermark
 cleanly replaces any previously stored levels.
 
 The samples of a whole batch of files are read with one ``IN (...) AND
-sample_index < 3000`` query and all candidate cycles are encoded in a single
-forward pass, so the network round trips and MPS kernel launches are amortised
-across the batch. Files are processed in ``(sample_time, id)`` ascending order
-and a tuple watermark on those two columns prevents re-processing files that
-have already been seen, so the poller first drains the backlog and then keeps
-watching for newly inserted rows.
+sample_index < {stage_limit}`` query per stage and all candidate cycles are
+encoded in a single forward pass, so the network round trips and MPS kernel
+launches are amortised across the batch. Only files that still report no cycle
+at the current stage are re-fetched at the next, larger window; a file is
+finally marked 无周期 only when no window yields a cycle. Files are processed
+in ``(sample_time, id)`` ascending order and a tuple watermark on those two
+columns prevents re-processing files that have already been seen, so the
+poller first drains the backlog and then keeps watching for newly inserted
+rows.
 
 Usage:
     conda activate tspulse
     python predict.py [--start-time ""] [--start-id 0] [--interval 60]
                       [--batch 100] [--limit 0] [--report-every 100]
-                      [--write-batch 10] [--dry-run]
+                      [--write-batch 100] [--dry-run]
 """
 
 import argparse
@@ -50,7 +55,7 @@ HERE = Path(__file__).resolve().parent
 CHECKPOINT_PATH = HERE / "checkpoints" / "tspulse_frozen.pt"
 BASELINE_PATH = HERE / "baseline" / "baseline.npz"
 
-SAMPLE_LIMIT = 3000
+STAGE_LIMITS = (3500, 7000, 14000)
 MEASUREMENT_TYPE = "压力"
 
 ANOMALY_FACTOR = 2.0
@@ -74,8 +79,8 @@ def load_baseline() -> tuple[dict[str, int], np.ndarray, np.ndarray]:
     return part_index, data["centroids"], data["radii"]
 
 
-def _fetch_chunk(connection, file_ids: list[int]) -> dict[int, np.ndarray]:
-    """Fetch the first SAMPLE_LIMIT rows of a chunk of files in one query."""
+def _fetch_chunk(connection, file_ids: list[int], limit: int) -> dict[int, np.ndarray]:
+    """Fetch the first ``limit`` rows of a chunk of files in one query."""
     if not file_ids:
         return {}
     placeholders = ",".join(["%s"] * len(file_ids))
@@ -89,7 +94,7 @@ def _fetch_chunk(connection, file_ids: list[int]) -> dict[int, np.ndarray]:
             WHERE wave_file_id IN ({placeholders}) AND sample_index < %s
             ORDER BY wave_file_id, sample_index
             """,
-            (*file_ids, SAMPLE_LIMIT),
+            (*file_ids, limit),
         )
         rows = cursor.fetchall()
     grouped: dict[int, tuple[list[float], list[float]]] = {}
@@ -110,8 +115,9 @@ def read_batch_samples(
     reader_connections: list,
     file_ids: list[int],
     workers: int,
+    limit: int,
 ) -> dict[int, np.ndarray]:
-    """Fetch the first SAMPLE_LIMIT rows of every file in one round-trip.
+    """Fetch the first ``limit`` rows of every file in one round-trip.
 
     The batch is split into ``workers`` chunks fetched on parallel persistent
     connections. Returns {file_id: (n, 3) array} with columns
@@ -120,11 +126,11 @@ def read_batch_samples(
     if not file_ids:
         return {}
     if workers <= 1 or len(file_ids) <= workers:
-        return _fetch_chunk(reader_connections[0], file_ids)
+        return _fetch_chunk(reader_connections[0], file_ids, limit)
     chunks = np.array_split(file_ids, min(workers, len(file_ids)))
     with ThreadPoolExecutor(max_workers=len(chunks)) as executor:
         futures = [
-            executor.submit(_fetch_chunk, reader_connections[i], chunk.tolist())
+            executor.submit(_fetch_chunk, reader_connections[i], chunk.tolist(), limit)
             for i, chunk in enumerate(chunks)
         ]
         merged: dict[int, np.ndarray] = {}
@@ -168,7 +174,7 @@ def fetch_batch(
         if watermark_time is None:
             cursor.execute(
                 """
-                SELECT id, point_name, measurement_type, sample_time
+                SELECT id, point_name, measurement_type, sample_time, sample_count
                 FROM wave_file
                 WHERE rpm > 0 AND measurement_type = %s
                 ORDER BY sample_time ASC, id ASC
@@ -179,7 +185,7 @@ def fetch_batch(
         else:
             cursor.execute(
                 """
-                SELECT id, point_name, measurement_type, sample_time
+                SELECT id, point_name, measurement_type, sample_time, sample_count
                 FROM wave_file
                 WHERE rpm > 0 AND measurement_type = %s
                   AND (sample_time > %s OR (sample_time = %s AND id > %s))
@@ -191,6 +197,55 @@ def fetch_batch(
         return cursor.fetchall()
 
 
+def resolve_batch_cycles(
+    rows: list[dict],
+    part_of: dict[int, str],
+    part_index: dict[str, int],
+    reader_connections: list,
+    workers: int,
+    first_stage: int = 0,
+) -> tuple[dict[int, tuple[str, np.ndarray | None]], int]:
+    """Adaptively detect one cycle per file with progressively larger windows.
+
+    The whole batch is first fetched at ``STAGE_LIMITS[first_stage]``. Files
+    that still report 无周期 and have more rows than the current window are
+    re-fetched at the next stage, up to ``STAGE_LIMITS[-1]``; a file is only
+    finally 无周期 when no window yields a cycle. ``first_stage`` lets callers
+    (predict_re) skip cheap windows for files already known to need more data.
+
+    Returns (``{file_id: (status, matrix)}``, error_count). Status is "ok" or a
+    skip reason, mirroring ``prepare_cycle``.
+    """
+    errors = 0
+    resolved: dict[int, tuple[str, np.ndarray | None]] = {}
+    stage_files: dict[int, dict] = {int(row["id"]): row for row in rows}
+    for stage in range(first_stage, len(STAGE_LIMITS)):
+        if not stage_files:
+            break
+        limit = STAGE_LIMITS[stage]
+        samples_map = read_batch_samples(reader_connections, list(stage_files), workers, limit)
+        next_pending: dict[int, dict] = {}
+        for file_id, row in stage_files.items():
+            samples = samples_map.get(file_id)
+            if samples is None or len(samples) == 0:
+                resolved[file_id] = ("无样本", None)
+                continue
+            try:
+                status, matrix = prepare_cycle(samples, part_of[file_id], part_index)
+            except Exception:
+                errors += 1
+                resolved[file_id] = ("错误", None)
+                continue
+            if status == "无周期" and int(row.get("sample_count") or 0) > limit:
+                next_pending[file_id] = row
+            else:
+                resolved[file_id] = (status, matrix)
+        stage_files = next_pending
+    for file_id in stage_files:
+        resolved[file_id] = ("无周期", None)
+    return resolved, errors
+
+
 def compute_status(distance: float, radius: float) -> int:
     """Map a fingerprint distance to a proportional integer status.
 
@@ -327,26 +382,18 @@ def main() -> int:
 
                 file_ids = [int(row["id"]) for row in rows]
                 part_of = {int(row["id"]): f"{row['point_name']}_{row['measurement_type']}" for row in rows}
-                sample_map = read_batch_samples(reader_connections, file_ids, read_workers)
+                resolved, batch_errors = resolve_batch_cycles(
+                    rows, part_of, part_index, reader_connections, read_workers,
+                )
+                errors += batch_errors
 
                 candidates: list[tuple[int, str, np.ndarray]] = []
                 for row in rows:
                     file_id = int(row["id"])
-                    part = part_of[file_id]
-                    samples = sample_map.get(file_id)
-                    if samples is None or len(samples) == 0:
-                        status, matrix = "无样本", None
-                    else:
-                        try:
-                            status, matrix = prepare_cycle(samples, part, part_index)
-                        except Exception:
-                            errors += 1
-                            status, matrix = "错误", None
-
                     total += 1
-
+                    status, matrix = resolved[file_id]
                     if status == "ok":
-                        candidates.append((file_id, part, matrix))
+                        candidates.append((file_id, part_of[file_id], matrix))
                     elif status == "无周期":
                         stats["无周期"] += 1
                         write_buffer.append((file_id, -1))

backend/LabelingPreTraining/predict_re.py

@@ -0,0 +1,226 @@
+"""Re-run TSPulse detection for files currently marked 无周期 (tspluse_status = -1).
+
+The original full re-run used a fixed 3000-sample prefix per file. Files that
+start mid-revolution have too few trigger pulses inside that prefix for the
+zero-marker segmentation to find a complete cycle, so they were mislabelled -1
+even though the client and the training generator (which read the whole file)
+detect them normally.
+
+This script re-processes ONLY the rows still holding ``tspluse_status = -1``,
+using predict.py's adaptive multi-stage window but starting at the 7000-sample
+stage (these files already failed the smaller windows). Files that still yield
+no cycle stay -1; the rest are written their corrected status (0 or the
+proportional anomaly value).
+
+The query filters on ``tspluse_status = -1``, so the ``(sample_time, id)``
+watermark is naturally idempotent: a file already fixed is no longer selected
+on resume. Run it once after the full re-run finishes; no column reset needed.
+
+Usage:
+    conda activate tspulse
+    python predict_re.py [--start-time ""] [--start-id 0] [--batch 100]
+                         [--limit 0] [--report-every 100] [--write-batch 100]
+                         [--dry-run]
+"""
+
+import argparse
+import sys
+
+import numpy as np
+import torch
+
+import predict
+from predict import (
+    MEASUREMENT_TYPE,
+    STAGE_LIMITS,
+    compute_status,
+    encode_batch,
+    flush_status,
+    load_baseline,
+    load_model,
+    resolve_batch_cycles,
+)
+
+
+def fetch_batch(
+    connection,
+    watermark_time: object | None,
+    watermark_id: int,
+    batch: int,
+) -> list[dict]:
+    with connection.cursor() as cursor:
+        if watermark_time is None:
+            cursor.execute(
+                """
+                SELECT id, point_name, measurement_type, sample_time, sample_count
+                FROM wave_file
+                WHERE rpm > 0 AND measurement_type = %s AND tspluse_status = -1
+                ORDER BY sample_time ASC, id ASC
+                LIMIT %s
+                """,
+                (MEASUREMENT_TYPE, batch),
+            )
+        else:
+            cursor.execute(
+                """
+                SELECT id, point_name, measurement_type, sample_time, sample_count
+                FROM wave_file
+                WHERE rpm > 0 AND measurement_type = %s AND tspluse_status = -1
+                  AND (sample_time > %s OR (sample_time = %s AND id > %s))
+                ORDER BY sample_time ASC, id ASC
+                LIMIT %s
+                """,
+                (MEASUREMENT_TYPE, watermark_time, watermark_time, watermark_id, batch),
+            )
+        return cursor.fetchall()
+
+
+def main() -> int:
+    parser = argparse.ArgumentParser(description="重新处理 tspluse_status = -1 的文件")
+    parser.add_argument("--start-time", type=str, default="",
+                        help="起始水位时间(含)，留空从头开始")
+    parser.add_argument("--start-id", type=int, default=0,
+                        help="起始水位 id(含)")
+    parser.add_argument("--batch", type=int, default=100, help="每轮读取的文件数")
+    parser.add_argument("--read-workers", type=int, default=4, help="并行读样本的连接数")
+    parser.add_argument("--limit", type=int, default=0, help="最多处理文件数，0 表示不限")
+    parser.add_argument("--report-every", type=int, default=100, help="每 N 个文件打印一次汇总")
+    parser.add_argument("--write-batch", type=int, default=100,
+                        help="回写攒满 N 条才批量写一次")
+    parser.add_argument("--dry-run", action="store_true", help="只预测不回写数据库")
+    args = parser.parse_args()
+
+    device = predict.get_device()
+    print(f"设备: {predict.describe(device)}")
+    model, mean, std = load_model(device)
+    part_index, centroids, radii = load_baseline()
+    print(f"基准表: {len(part_index)} 个部位, 重新处理 tspluse_status = -1 的文件, "
+          f"自适应窗口: {STAGE_LIMITS[1:]} (跳过 {STAGE_LIMITS[0]})")
+
+    start_time = args.start_time.strip()
+    watermark_time: object | None = start_time or None
+    watermark_id = args.start_id
+    total = 0
+    stats = {"正常": 0, "异常": 0, "无周期": 0}
+    skipped = {"无基准": 0, "坏数据": 0, "周期过长": 0, "无样本": 0}
+    errors = 0
+    last_reported = 0
+    write_buffer: list[tuple[int, int]] = []
+
+    read_workers = max(1, min(args.read_workers, args.batch))
+    reader_connections = [predict.get_connection() for _ in range(read_workers)]
+
+    def flush(connection) -> None:
+        if args.dry_run:
+            write_buffer.clear()
+        else:
+            flush_status(connection, write_buffer)
+
+    def summary_text() -> str:
+        return (
+            f"已处理 {total}, 正常 {stats['正常']}, 异常 {stats['异常']}, "
+            f"无周期 {stats['无周期']}, 跳过 {sum(skipped.values())}({dict(skipped)}), "
+            f"错误 {errors}, 水位 sample_time={watermark_time} id={watermark_id}"
+        )
+
+    print(f"水位起点: sample_time={start_time or '从头'}, id={args.start_id}, "
+          f"批大小: {args.batch}, 读并行: {read_workers} 连接"
+          + (", 干跑(不回写)" if args.dry_run else ""))
+
+    try:
+        while True:
+            connection = predict.get_connection()
+            try:
+                rows = fetch_batch(connection, watermark_time, watermark_id, args.batch)
+                if not rows:
+                    flush(connection)
+                    connection.close()
+                    print(f"无待处理的 -1 文件，处理完成 [{summary_text()}]")
+                    return 0
+
+                file_ids = [int(row["id"]) for row in rows]
+                part_of = {int(row["id"]): f"{row['point_name']}_{row['measurement_type']}" for row in rows}
+                resolved, batch_errors = resolve_batch_cycles(
+                    rows, part_of, part_index, reader_connections, read_workers,
+                    first_stage=1,
+                )
+                errors += batch_errors
+
+                candidates: list[tuple[int, str, np.ndarray]] = []
+                for row in rows:
+                    file_id = int(row["id"])
+                    total += 1
+                    status, matrix = resolved[file_id]
+                    if status == "ok":
+                        candidates.append((file_id, part_of[file_id], matrix))
+                    elif status == "无周期":
+                        stats["无周期"] += 1
+                        write_buffer.append((file_id, -1))
+                    else:
+                        if status in skipped:
+                            skipped[status] += 1
+                        write_buffer.append((file_id, 0))
+
+                if candidates:
+                    fingerprints = encode_batch(
+                        model, mean, std, device, [item[2] for item in candidates]
+                    )
+                    for (file_id, part, _matrix), fingerprint in zip(candidates, fingerprints):
+                        index = part_index[part]
+                        distance = float(np.linalg.norm(fingerprint - centroids[index]))
+                        radius = float(radii[index])
+                        status = compute_status(distance, radius)
+                        stats["异常" if status > 0 else "正常"] += 1
+                        write_buffer.append((file_id, status))
+                        if len(write_buffer) >= args.write_batch:
+                            flush(connection)
+
+                if total // args.report_every > last_reported:
+                    last_reported = total // args.report_every
+                    print(f"  汇总: {summary_text()}", flush=True)
+
+                if args.limit and total >= args.limit:
+                    flush(connection)
+                    watermark_time = rows[-1]["sample_time"]
+                    watermark_id = int(rows[-1]["id"])
+                    connection.close()
+                    print(f"已达 --limit={args.limit}，退出")
+                    print(f"  汇总: {summary_text()}")
+                    return 0
+
+                flush(connection)
+                watermark_time = rows[-1]["sample_time"]
+                watermark_id = int(rows[-1]["id"])
+            finally:
+                try:
+                    connection.close()
+                except Exception:
+                    pass
+
+    except KeyboardInterrupt:
+        if write_buffer and not args.dry_run:
+            try:
+                with predict.get_connection() as connection:
+                    flush_status(connection, write_buffer)
+            except Exception:
+                pass
+        for connection in reader_connections:
+            try:
+                connection.close()
+            except Exception:
+                pass
+        print(f"\n已中断。{summary_text()}")
+        if watermark_time is not None:
+            print(f"续跑命令: python predict_re.py --start-time \"{watermark_time}\" --start-id {watermark_id}")
+        return 0
+
+    for connection in reader_connections:
+        try:
+            connection.close()
+        except Exception:
+            pass
+    return 0
+
+
+if __name__ == "__main__":
+    sys.exit(main())

backend/app/main.py

@@ -118,6 +118,7 @@ def time_points(
     max_time: str | None = None,
     include_stopped: bool = False,
     min_status: int | None = Query(default=None, ge=0),
+    status_filter: list[str] = Query(default=[]),
     _auth: str = Depends(require_auth),
 ) -> dict[str, Any]:
     try:
@@ -128,6 +129,7 @@ def time_points(
             max_time,
             include_stopped,
             min_status,
+            status_filter,
         )
     except ValueError as error:
         raise HTTPException(status_code=400, detail=str(error)) from error

backend/app/services/data_service.py

@@ -215,10 +215,15 @@ class DataService:
         max_time: str | None,
         include_stopped: bool = False,
         min_status: int | None = None,
+        status_filter: list[str] | None = None,
     ) -> dict[str, Any]:
         if not point_name.strip():
             raise ValueError("机组与部位不能为空")
         types = _validate_measurement_types(measurement_types)
+        status_filters = {value for value in (status_filter or [])}
+        unknown = status_filters - {"abnormal", "no_cycle"}
+        if unknown:
+            raise ValueError(f"不支持的状态筛选：{'、'.join(sorted(unknown))}")
         start = _parse_time(min_time)
         end = _parse_time(max_time)
         if start and end and start > end:
@@ -233,9 +238,15 @@ class DataService:
             params: list[Any] = [point_name, *types]
             if not include_stopped:
                 clauses.append("rpm > 0")
-            if min_status is not None and min_status > 0:
+            if min_status is not None and min_status > 0 and "no_cycle" not in status_filters:
                 clauses.append("tspluse_status >= %s")
                 params.append(min_status)
+            if "abnormal" in status_filters and "no_cycle" in status_filters:
+                clauses.append("(tspluse_status > 0 OR tspluse_status = -1)")
+            elif "abnormal" in status_filters:
+                clauses.append("tspluse_status > 0")
+            elif "no_cycle" in status_filters:
+                clauses.append("tspluse_status = -1")
             if start:
                 clauses.append("sample_time >= %s")
                 params.append(start)
@@ -255,18 +266,51 @@ class DataService:
                         params,
                     )
                     rows = cursor.fetchall()
-            return self._group_time_points(rows)
+            reference_rows: list[dict[str, Any]] = []
+            if status_filters and rows:
+                reference_where = [
+                    "point_name = %s",
+                    "measurement_type = '压力'",
+                    "tspluse_status = 0",
+                ]
+                reference_params: list[Any] = [point_name]
+                if not include_stopped:
+                    reference_where.append("rpm > 0")
+                if start:
+                    reference_where.append("sample_time >= %s")
+                    reference_params.append(start)
+                if end:
+                    reference_where.append("sample_time <= %s")
+                    reference_params.append(end)
+                reference_params.append(rows[0]["sample_time"])
+                with get_connection() as connection:
+                    with connection.cursor() as cursor:
+                        cursor.execute(
+                            f"""
+                            SELECT id, point_name, measurement_type, sample_time,
+                                   sample_count, sample_frequency_hz, rpm, tspluse_status
+                            FROM wave_file
+                            WHERE {' AND '.join(reference_where)}
+                            ORDER BY ABS(TIMESTAMPDIFF(SECOND, sample_time, %s)) ASC
+                            LIMIT 1
+                            """,
+                            reference_params,
+                        )
+                        reference_rows = cursor.fetchall()
+            return self._group_time_points(rows), self._group_time_points(reference_rows)
 
         def demo_query():
-            return self._demo_time_points(point_name, types, start, end)
+            return self._demo_time_points(point_name, types, start, end), []
 
-        points, source = self._run_with_fallback(database_query, demo_query)
+        result, source = self._run_with_fallback(database_query, demo_query)
+        points, reference = result
         return {
             "source": source,
             "pointName": point_name,
             "measurementTypes": types,
             "total": len(points),
             "points": points,
+            "referencePoints": reference,
             "notice": self._source_notice(source),
         }
 

frontend/src/App.vue

@@ -17,9 +17,12 @@ const windowSize = ref(4)
 const maxPoints = ref(200000)
 const noSampling = ref(false)
 const includeStopped = ref(false)
+const abnormalOnly = ref(false)
+const noCycleOnly = ref(false)
 const minStatus = ref<number | null>(null)
 const ruler = ref<{ min: number; max: number } | null>(null)
 const timePoints = ref<TimePoint[]>([])
+const referencePoints = ref<TimePoint[]>([])
 const startIndex = ref(0)
 const waveData = ref<WaveWindowResponse | null>(null)
 const chartMode = ref<'split' | 'merge'>('split')
@@ -56,8 +59,20 @@ const selectedOptionRows = computed<QueryOption[]>(() => (
     row.pointName === selectedPointName.value && selectedTypes.value.includes(row.measurementType)
   )) ?? []
 ))
-const selectedWindowPoints = computed(() => timePoints.value.slice(startIndex.value, startIndex.value + windowSize.value))
-const endIndex = computed(() => Math.min(timePoints.value.length, startIndex.value + windowSize.value))
+const selectedWindowPoints = computed(() => {
+  const slice = timePoints.value.slice(startIndex.value, startIndex.value + stripWindowSize.value)
+  return hasReference.value ? [...referencePoints.value, ...slice] : slice
+})
+const hasReference = computed(() => referencePoints.value.length > 0)
+const stripWindowSize = computed(() => Math.max(0, windowSize.value - (hasReference.value ? 1 : 0)))
+const maxStart = computed(() => Math.max(0, timePoints.value.length - stripWindowSize.value))
+const endIndex = computed(() => Math.min(timePoints.value.length, startIndex.value + stripWindowSize.value))
+const statusFilter = computed(() => {
+  const list: string[] = []
+  if (abnormalOnly.value) list.push('abnormal')
+  if (noCycleOnly.value) list.push('no_cycle')
+  return list
+})
 const currentCycles = computed(() => waveData.value?.cycles ?? [])
 const currentSource = computed(() => waveData.value?.source ?? timePointsSource.value)
 const timePointsSource = ref<'database' | 'demo'>('demo')
@@ -148,18 +163,21 @@ async function loadTimePoints() {
       minTime: toApiTime(minTime.value),
       maxTime: toApiTime(maxTime.value),
       includeStopped: includeStopped.value,
-      minStatus: minStatus.value,
+      minStatus: noCycleOnly.value ? null : minStatus.value,
+      statusFilter: statusFilter.value,
     })
     timePoints.value = result.points
+    referencePoints.value = result.referencePoints ?? []
     timePointsSource.value = result.source
     startIndex.value = Math.max(
       0,
-      Math.min(firstRunningIndex(), Math.max(0, timePoints.value.length - windowSize.value)),
+      Math.min(firstRunningIndex(), Math.max(0, timePoints.value.length - stripWindowSize.value)),
     )
     chartDirty.value = true
     void loadAnnotations()
   } catch (error) {
     timePoints.value = []
+    referencePoints.value = []
     errorMessage.value = error instanceof Error ? error.message : '时间点读取失败'
   } finally {
     timePointsLoading.value = false
@@ -227,13 +245,13 @@ function onTimeChange() {
 }
 
 function onStartIndexChange(value: number) {
-  startIndex.value = Math.max(0, Math.min(value, Math.max(0, timePoints.value.length - windowSize.value)))
+  startIndex.value = Math.max(0, Math.min(value, maxStart.value))
   markChartDirty()
 }
 
 function onWindowSizeChange(value: number) {
   windowSize.value = Math.max(1, Math.min(200, Math.round(value || 1)))
-  startIndex.value = Math.min(startIndex.value, Math.max(0, timePoints.value.length - windowSize.value))
+  startIndex.value = Math.min(startIndex.value, maxStart.value)
   markChartDirty()
 }
 
@@ -243,8 +261,8 @@ function resetWindow() {
 }
 
 function pageWindow(direction: -1 | 1) {
-  const next = startIndex.value + direction * windowSize.value
-  startIndex.value = Math.max(0, Math.min(next, Math.max(0, timePoints.value.length - windowSize.value)))
+  const next = startIndex.value + direction * stripWindowSize.value
+  startIndex.value = Math.max(0, Math.min(next, maxStart.value))
   markChartDirty()
 }
 
@@ -444,6 +462,12 @@ watch(noSampling, () => {
 watch(includeStopped, () => {
   if (initialized.value) scheduleTimePoints()
 })
+watch(abnormalOnly, () => {
+  if (initialized.value) scheduleTimePoints()
+})
+watch(noCycleOnly, () => {
+  if (initialized.value) scheduleTimePoints()
+})
 watch(minStatus, () => {
   if (initialized.value) scheduleTimePoints()
 })
@@ -674,9 +698,11 @@ onBeforeUnmount(() => {
             <label class="field field-check">
               <span class="field-label">运行状态</span>
               <el-checkbox v-model="includeStopped" class="query-checkbox" size="large">停机</el-checkbox>
+              <el-checkbox v-model="abnormalOnly" class="query-checkbox" size="large">异常</el-checkbox>
+              <el-checkbox v-model="noCycleOnly" class="query-checkbox" size="large">无周期</el-checkbox>
             </label>
             <label class="field field-status">
-              <span class="field-label">质心距离 <em>异常级别 ≥ 输入值</em></span>
+              <span class="field-label">质心距离</span>
               <el-input-number
                 v-model="minStatus"
                 class="query-control"
@@ -686,8 +712,10 @@ onBeforeUnmount(() => {
                 :step="1"
                 :step-strictly="true"
                 controls-position="right"
+                :disabled="queryLoading || noCycleOnly"
                 clearable
                 placeholder="可空"
+                :title="noCycleOnly ? '勾选无周期时忽略质心距离' : undefined"
               />
             </label>
           </div>
@@ -700,7 +728,7 @@ onBeforeUnmount(() => {
           :points="timePoints"
           :measurement-types="selectedTypes"
           :start-index="startIndex"
-          :window-size="windowSize"
+          :window-size="stripWindowSize"
           :loading="timePointsLoading"
           :min-time="minTime"
           :max-time="maxTime"
@@ -712,6 +740,9 @@ onBeforeUnmount(() => {
           <div class="selection-readout">
             <span class="selection-accent"></span>
             <span>当前窗口覆盖 <strong>{{ selectedWindowPoints.length }}</strong> 个时间点</span>
+            <span v-if="hasReference" class="reference-badge" title="始终保留一个 status=0 的正常数据用于对比，固定不随翻页变化">
+              正常对比：{{ formatDateTime(referencePoints[0]?.sampleTime) }}
+            </span>
           </div>
           <div class="selection-actions">
             <el-button class="ghost-button" plain :disabled="!startIndex" @click="resetWindow">回到起点</el-button>

frontend/src/api.ts

@@ -73,6 +73,7 @@ export function fetchTimePoints(params: {
   maxTime?: string
   includeStopped?: boolean
   minStatus?: number | null
+  statusFilter?: string[]
 }) {
   const search = new URLSearchParams({ point_name: params.pointName })
   params.measurementTypes.forEach((value) => search.append('measurement_types', value))
@@ -80,6 +81,7 @@ export function fetchTimePoints(params: {
   if (params.maxTime) search.set('max_time', params.maxTime)
   if (params.includeStopped) search.set('include_stopped', 'true')
   if (params.minStatus) search.set('min_status', String(params.minStatus))
+  ;(params.statusFilter ?? []).forEach((value) => search.append('status_filter', value))
   return request<TimePointsResponse>(`/api/time-points?${search.toString()}`)
 }
 

frontend/src/styles.css

@@ -117,12 +117,15 @@ h2 { font-size: 18px; line-height: 1.35; font-weight: 600; }
 .query-control .el-tag { font-size: 13px; }
 .window-number .el-input__inner { text-align: center; }
 .query-button.el-button { width: 100%; height: 40px; margin: 0; border-radius: 4px; font-size: 14px; }
-.query-row-2 { display: flex; align-items: flex-end; gap: 16px; grid-column: 1 / -1; }
+.query-row-2 { display: flex; align-items: flex-start; gap: 16px; grid-column: 1 / -1; }
+.query-row-2 .field-check,
+.query-row-2 .field-status { min-height: 68px; }
 .query-row-2 .field-check .field-label,
 .query-row-2 .field-status .field-label { white-space: nowrap; }
-.query-checkbox.el-checkbox { height: 40px; margin-right: 0; }
+.query-checkbox.el-checkbox { height: 40px; margin-right: 20px; display: inline-flex; align-items: center; }
+.query-checkbox.el-checkbox:last-child { margin-right: 0; }
 .query-checkbox .el-checkbox__label { font-size: 14px; color: #303133; }
-.field-status { width: 240px; max-width: 100%; }
+.field-status { width: 120px; max-width: 100%; }
 .data-alert { margin-top: 16px; }
 
 .selection-panel { padding: 20px 24px 18px; }
@@ -152,7 +155,17 @@ h2 { font-size: 18px; line-height: 1.35; font-weight: 600; }
 .selection-actions { display: flex; gap: 8px; flex-shrink: 0; }
 .selection-readout strong { color: #303133; font-family: ui-monospace, SFMono-Regular, Menlo, Monaco, Consolas, monospace; font-weight: 500; }
 .selection-accent { width: 7px; height: 7px; background: #409eff; border-radius: 50%; }
-.jump-select { width: 220px; }
+.reference-badge {
+  flex-shrink: 0;
+  padding: 1px 8px;
+  border: 1px solid #4d9e6f;
+  border-radius: 10px;
+  color: #2e7d32;
+  background: #ecf7f1;
+  font-size: 12px;
+  white-space: nowrap;
+}
+.jump-select { width: 280px; }
 .jump-select .el-select__wrapper { min-height: 32px; }
 .ghost-button.chart-query { font-weight: 500; }
 .ghost-button.chart-query.is-dirty {

frontend/src/types.ts

@@ -44,6 +44,7 @@ export type TimePointsResponse = {
   measurementTypes: MeasurementType[]
   total: number
   points: TimePoint[]
+  referencePoints?: TimePoint[]
   notice: string | null
 }