新增故障段模型预测参考（3h滑动聚合概率），全量/实时可用 --with-model，结果与DB含预测字段

models/README.md

@@ -35,6 +35,9 @@ python models/scan_all.py --device 7# 8# 9# --store local
 # 数据库：本地结果与中间文件照常生成，并额外把最终段写入 compressor_alarm
 python models/scan_all.py --device 7# 8# 9# --store db
 
+# 开启模型预测参考：对每个故障段做 3 小时滑动预测，写入概率均值/最大/最小
+python models/scan_all.py --device 7# 8# 9# --store db --with-model
+
 # 也可显式指定时间，或复用已缓存小时中间表
 python models/scan_all.py --start "2025-01-01 00:00:00" --end "2026-09-01 00:00:00"
 python models/scan_all.py --hourly-csv models/datas/hourly/pks_hourly_full.csv
@@ -50,11 +53,21 @@ python models/scan_all.py --hourly-csv models/datas/hourly/pks_hourly_full.csv
 `alarm_des=阶段触发依据`（≤4000）、`alarm_info_json`=对应 CSV 行 JSON、`status=1`。
 全量写库会先删除所选设备的历史压力/振动告警再插入。
 
+### 模型预测参考（`--with-model`，默认关闭）
+
+- 定位：只做**参考数值**，不做任何异常判定，不参与阶段划分。
+- 方式：对故障段内**每个有效运行小时**取以它为结尾的 3 小时窗口滑动预测（段不足 3h 时向前补有效小时），
+  汇总为 `模型异常概率均值 / 最大值 / 最小值`。
+- 输出：`alarm_info_json` 增加上述 3 个字段；`alarm_des` 追加一句
+  `模型预测参考：3小时窗口异常概率 均值x/最大y/最小z（n次）`；本地 CSV 同步增加这 4 列。
+- 训练链路不受影响；开启时若 `models/datas/artifacts/*` 缺失会直接报错（先跑对应 `run_pipeline.py`）。
+
 ## 实时运行
 
 ```bash
 python models/realtime_service.py --once                 # 单次
 python models/realtime_service.py --loop --interval 300  # 每 5 分钟（命令行定时）
+python models/realtime_service.py --once --with-model    # 同时写入模型预测参考
 ```
 
 - 窗口：`end = min(floor(now), floor(最新数据)+1h)`，`start = end - 45 天`。

models/common/alarm_store.py

@@ -47,13 +47,19 @@ def _as_dt(value) -> datetime:
 
 
 def _json(row: dict) -> str:
-    return json.dumps(row, ensure_ascii=False, separators=(",", ":"), default=str)
+    # The explanatory note goes to alarm_des, not to alarm_info_json.
+    data = {key: value for key, value in row.items() if not key.startswith("_") and key != "模型预测参考"}
+    return json.dumps(data, ensure_ascii=False, separators=(",", ":"), default=str)
 
 
 def _targets(device_code: str, kind: str, row: dict) -> list[dict]:
     level = row.get("压力阶段") if kind == "pressure" else row.get("振动阶段")
     start, end = row.get("开始小时"), row.get("结束小时")
-    des = str(row.get("阶段触发依据") or "")[:DES_MAX]
+    des = str(row.get("阶段触发依据") or "")
+    note = row.get("模型预测参考")
+    if note:
+        des = f"{des}；{note}"
+    des = des[:DES_MAX]
     info = _json(row)
     if kind == "pressure":
         parts = [(OIL_PART, OIL_POINT, OIL_ALARM_TYPE)]

models/common/model_predict.py

@@ -0,0 +1,108 @@
+"""Model prediction as a *reference* for fault segments (no verdict).
+
+For each fault segment produced by the rule algorithm, slide a 3-hour valid-run
+window ending at every valid hour of the segment and score it with the existing
+XGBoost model.  Only aggregate probabilities are reported (mean / max / min);
+the model never decides whether a fault exists.
+
+Training is untouched: this module only reuses the same window builder and
+feature functions so the scored features match training exactly.
+"""
+from __future__ import annotations
+
+import json
+from pathlib import Path
+
+import joblib
+import numpy as np
+
+from . import paths, windowing
+
+WINDOW_HOURS = 3
+MODEL_FIELDS = ["模型异常概率均值", "模型异常概率最大值", "模型异常概率最小值"]
+NOTE_FIELD = "模型预测参考"
+
+_BUNDLE_CACHE: dict[str, tuple] = {}
+
+
+def _assets(kind: str):
+    if kind == "pressure":
+        from models.generalized import features_pressure
+        from models.generalized.build_datasets import pressure_features
+
+        return features_pressure, pressure_features
+    from models.vibration import features_vibration
+    from models.vibration.build_datasets import vibration_features
+
+    return features_vibration, vibration_features
+
+
+def load_bundle(kind: str):
+    if kind in _BUNDLE_CACHE:
+        return _BUNDLE_CACHE[kind]
+    model_dir = paths.PRESSURE_MODELS if kind == "pressure" else paths.VIBRATION_MODELS
+    model_path = model_dir / f"{kind}_model.joblib"
+    meta_path = model_dir / f"{kind}_metadata.json"
+    if not model_path.is_file() or not meta_path.is_file():
+        pipeline = "generalized" if kind == "pressure" else "vibration"
+        raise RuntimeError(
+            f"未找到 {kind} 模型产物：{model_dir}。请先运行 "
+            f"python models/{pipeline}/run_pipeline.py --window-hours 3"
+        )
+    bundle = joblib.load(model_path)
+    metadata = json.loads(meta_path.read_text(encoding="utf-8"))
+    _BUNDLE_CACHE[kind] = (bundle, metadata)
+    return bundle, metadata
+
+
+def _window_index(hourly_rows: list[dict], kind: str) -> dict[str, list[tuple[str, dict]]]:
+    _, feature_fn = _assets(kind)
+    samples = windowing.build_samples(hourly_rows, kind, [], WINDOW_HOURS, feature_fn)
+    by_unit: dict[str, list[tuple[str, dict]]] = {}
+    for sample in samples:
+        by_unit.setdefault(sample["unit"], []).append((sample["end_hour"], sample))
+    for series in by_unit.values():
+        series.sort()
+    return by_unit
+
+
+def _probability(bundle, feature_module, sample) -> float:
+    names = bundle["feature_names"]
+    features = feature_module.generalized_features(sample, bundle["unit_baselines"])
+    matrix = np.asarray([[features.get(name) for name in names]], dtype=float)
+    matrix = bundle["imputer"].transform(matrix)
+    return float(bundle["model"].predict_proba(matrix)[0, 1])
+
+
+def annotate(kind: str, hourly_rows: list[dict], segments: list[dict]) -> dict:
+    """Add model probability aggregates + note to each segment in place."""
+    bundle, _metadata = load_bundle(kind)
+    feature_module, _feature_fn = _assets(kind)
+    by_unit = _window_index(hourly_rows, kind)
+    scored = 0
+    for segment in segments:
+        unit = segment.get("机组")
+        start, end = segment.get("开始小时"), segment.get("结束小时")
+        probabilities = []
+        if start and end:
+            for end_hour, sample in by_unit.get(unit, []):
+                if start <= end_hour <= end:
+                    probabilities.append(_probability(bundle, feature_module, sample))
+        if probabilities:
+            mean = float(np.mean(probabilities))
+            high = float(np.max(probabilities))
+            low = float(np.min(probabilities))
+            segment["模型异常概率均值"] = round(mean, 6)
+            segment["模型异常概率最大值"] = round(high, 6)
+            segment["模型异常概率最小值"] = round(low, 6)
+            segment[NOTE_FIELD] = (
+                f"模型预测参考：{WINDOW_HOURS}小时窗口异常概率 "
+                f"均值{mean:.4f}/最大{high:.4f}/最小{low:.4f}（{len(probabilities)}次）"
+            )
+            scored += 1
+        else:
+            segment[NOTE_FIELD] = "模型预测参考：无可用3小时窗口，数据不足"
+    return {"kind": kind, "segments": len(segments), "scored": scored}
+
+
+__all__ = ["WINDOW_HOURS", "MODEL_FIELDS", "NOTE_FIELD", "load_bundle", "annotate"]

models/common/verify.py

@@ -57,12 +57,34 @@ METRIC_TOLERANCE = 1e-3
 THRESHOLD_TOLERANCE = 1e-6
 
 
+def _csv_header_rows(path: Path):
+    with path.open(encoding="utf-8-sig", newline="") as handle:
+        reader = csv.DictReader(handle)
+        return list(reader.fieldnames or []), list(reader)
+
+
 def compare_files(reference: Path, candidate: Path) -> dict:
-    return {
-        "reference": str(reference),
-        "candidate": str(candidate),
-        "identical": reference.read_bytes() == candidate.read_bytes(),
-    }
+    """Byte-compare when identical; otherwise compare the reference columns as a subset.
+
+    The model-enabled run appends model-reference columns, so the base algorithm
+    columns are compared while extra columns are allowed.
+    """
+    base = {"reference": str(reference), "candidate": str(candidate)}
+    if reference.read_bytes() == candidate.read_bytes():
+        return {**base, "identical": True, "mode": "byte"}
+    ref_header, ref_rows = _csv_header_rows(reference)
+    cand_header, cand_rows = _csv_header_rows(candidate)
+    missing = [column for column in ref_header if column not in cand_header]
+    extra = [column for column in cand_header if column not in ref_header]
+    if missing or len(ref_rows) != len(cand_rows):
+        return {**base, "identical": False, "mode": "subset", "missing_columns": missing,
+                "row_counts": [len(ref_rows), len(cand_rows)]}
+    identical = all(
+        ref_rows[index].get(column) == cand_rows[index].get(column)
+        for index in range(len(ref_rows))
+        for column in ref_header
+    )
+    return {**base, "identical": identical, "mode": "subset", "extra_columns": extra}
 
 
 def check_algorithm():

models/generalized/pressure_algorithm.py

@@ -308,10 +308,11 @@ def run(rows: list[dict], configs: dict[int, dict]):
     return hourly_levels, hour_segments, cycles
 
 
-def write_results(output_dir: Path, hourly_levels, hour_segments, cycles):
+def write_results(output_dir: Path, hourly_levels, hour_segments, cycles, extra_fields=()):
     output_dir = Path(output_dir)
+    stage_fields = [field for field in HOUR_STAGE_FIELDS if field not in extra_fields] + list(extra_fields)
     io.write_csv(output_dir / "润滑油压力小时等级.csv", io.clean_rows(hourly_levels), HOURLY_LEVEL_FIELDS, {field: field for field in HOURLY_LEVEL_FIELDS})
-    io.write_csv(output_dir / "润滑油压力小时阶段段.csv", io.clean_rows(hour_segments), HOUR_STAGE_FIELDS, {field: field for field in HOUR_STAGE_FIELDS})
+    io.write_csv(output_dir / "润滑油压力小时阶段段.csv", io.clean_rows(hour_segments), stage_fields, {field: field for field in stage_fields})
     io.write_csv(output_dir / "润滑油压力演化周期.csv", io.clean_rows(cycles), CYCLE_FIELDS, {field: field for field in CYCLE_FIELDS})
 
 

models/markdown/全量与实时运行及告警落库.md

@@ -29,6 +29,9 @@ python models/scan_all.py --device 7# 8# 9# --store local
 # 数据库：本地结果照常生成，并额外写入 compressor_alarm
 python models/scan_all.py --device 7# 8# 9# --store db
 
+# 开启模型预测参考（3 小时滑动预测，写入概率均值/最大/最小）
+python models/scan_all.py --device 7# 8# 9# --store db --with-model
+
 # 显式时间或复用小时中间表
 python models/scan_all.py --start "2025-01-01 00:00:00" --end "2026-09-01 00:00:00"
 python models/scan_all.py --hourly-csv models/datas/hourly/pks_hourly_full.csv
@@ -60,11 +63,26 @@ python models/scan_all.py --hourly-csv models/datas/hourly/pks_hourly_full.csv
 - 结构迁移：`models/migrations/alter_compressor_alarm_alarm_des_4000.sql`
   （`ALTER TABLE compressor_alarm MODIFY alarm_des VARCHAR(4000)`）。
 
+## 3.1 模型预测参考（`--with-model`，默认关闭）
+
+- 定位：模型只提供**预测参考数值**，不做异常判定，不影响故障段的产生与等级。
+- 方式：对每个故障段内**每个有效运行小时**取“结尾 3 个有效运行小时”窗口滑动预测一次
+  （段不足 3 小时时窗口向前补有效小时）；整段无法组成窗口则记“数据不足”。
+- 汇总：`模型异常概率均值`、`模型异常概率最大值`、`模型异常概率最小值`。
+- 输出：
+  - `alarm_info_json` 只增加上述 3 个概率字段；
+  - `alarm_des` 追加 `模型预测参考：3小时窗口异常概率 均值x/最大y/最小z（n次）`；
+  - 本地 `润滑油压力小时阶段段.csv` / `振动异常段.csv` 同步增加这 3 列 + `模型预测参考` 列。
+- 训练链路（`build_datasets` / `split` / `train`）完全不受影响；预测复用同一窗口与特征函数。
+- 开启时若 `models/datas/artifacts/*` 缺失会直接报错，需先运行对应 `run_pipeline.py`。
+- 默认关闭，不带该参数时结果与 golden 逐字节一致；带该参数时校验按基础列子集比对。
+
 ## 4. 实时运行与合并规则
 
 ```bash
 python models/realtime_service.py --once                    # 单次
 python models/realtime_service.py --loop --interval 300     # 每 5 分钟（命令行定时）
+python models/realtime_service.py --once --with-model       # 同时写入模型预测参考
 ```
 
 - 每次 tick：`end = min(floor(本机当前时间), floor(最新数据) + 1h)`，

models/realtime_service.py

@@ -30,14 +30,14 @@ from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
-from models.common import alarm_store, db, devices, io, queries
+from models.common import alarm_store, db, devices, io, model_predict, queries
 from models.generalized import pressure_algorithm
 from models.vibration import vibration_algorithm
 
 REALTIME_CHUNK_DAYS = 7
 
 
-def run_tick(now: datetime, device_list: list[str], lookback_days: int, write: bool, journal: list | None) -> dict:
+def run_tick(now: datetime, device_list: list[str], lookback_days: int, write: bool, journal: list | None, with_model: bool = False) -> dict:
     device_batches = devices.load_device_batches(device_list)
     unit_by_batch = devices.batch_to_unit(device_batches)
     batch_device = devices.batch_to_device(device_batches)
@@ -53,6 +53,9 @@ def run_tick(now: datetime, device_list: list[str], lookback_days: int, write: b
     rows = queries.fetch_all_hourly(start, end, REALTIME_CHUNK_DAYS, unit_by_batch, configs, sorted(unit_by_batch))
     hourly_levels, hour_segments, cycles = pressure_algorithm.run(rows, configs)
     events = vibration_algorithm.run(rows, configs)
+    if with_model:
+        model_predict.annotate("pressure", rows, hour_segments)
+        model_predict.annotate("vibration", rows, events)
     alarms = alarm_store.build_alarms(io.clean_rows(hour_segments), io.clean_rows(events), batch_device)
     result = alarm_store.merge_realtime(alarms, start, end, journal=journal if write else None, dry_run=not write)
     return {
@@ -86,6 +89,7 @@ def main() -> None:
     parser.add_argument("--loop", action="store_true", help="按 --interval 循环运行")
     parser.add_argument("--interval", type=int, default=300, help="循环间隔秒，默认 300")
     parser.add_argument("--simulate", default=None, help="逗号分隔的时间列表，冒烟测试")
+    parser.add_argument("--with-model", action="store_true", help="对故障段做模型滑动预测（默认关闭）")
     parser.add_argument("--no-rollback", action="store_true", help="冒烟测试后不回滚数据库")
     args = parser.parse_args()
     device_list = devices.normalize_devices(args.device)
@@ -96,7 +100,7 @@ def main() -> None:
         times = [queries.parse_dt(item.strip()) for item in args.simulate.split(",") if item.strip()]
         journal: list = []
         for moment in times:
-            summary = run_tick(moment, device_list, args.lookback_days, write=True, journal=journal)
+            summary = run_tick(moment, device_list, args.lookback_days, write=True, journal=journal, with_model=args.with_model)
             print(format_tick(summary), flush=True)
             print("  当前告警分布：" + snapshot(device_list), flush=True)
         if not args.no_rollback:
@@ -107,11 +111,11 @@ def main() -> None:
     if args.loop:
         print(f"实时服务启动，每 {args.interval} 秒运行一次；Ctrl+C 退出。", flush=True)
         while True:
-            summary = run_tick(datetime.now(), device_list, args.lookback_days, write=True, journal=None)
+            summary = run_tick(datetime.now(), device_list, args.lookback_days, write=True, journal=None, with_model=args.with_model)
             print(format_tick(summary), flush=True)
             time.sleep(args.interval)
 
-    summary = run_tick(datetime.now(), device_list, args.lookback_days, write=True, journal=None)
+    summary = run_tick(datetime.now(), device_list, args.lookback_days, write=True, journal=None, with_model=args.with_model)
     print(format_tick(summary))
     print("当前告警分布：" + snapshot(device_list))
 

models/results/verification/verification_report.json

@@ -7,35 +7,52 @@
         "file": "润滑油压力小时等级.csv",
         "reference": "/Users/zhouhao/Code/Python/Compressor/models/datas/reference/润滑油压力小时等级.csv",
         "candidate": "/Users/zhouhao/Code/Python/Compressor/models/results/pressure/润滑油压力小时等级.csv",
-        "identical": true
+        "identical": true,
+        "mode": "byte"
       },
       {
         "group": "pressure",
         "file": "润滑油压力小时阶段段.csv",
         "reference": "/Users/zhouhao/Code/Python/Compressor/models/datas/reference/润滑油压力小时阶段段.csv",
         "candidate": "/Users/zhouhao/Code/Python/Compressor/models/results/pressure/润滑油压力小时阶段段.csv",
-        "identical": true
+        "identical": true,
+        "mode": "subset",
+        "extra_columns": [
+          "模型异常概率均值",
+          "模型异常概率最大值",
+          "模型异常概率最小值",
+          "模型预测参考"
+        ]
       },
       {
         "group": "pressure",
         "file": "润滑油压力演化周期.csv",
         "reference": "/Users/zhouhao/Code/Python/Compressor/models/datas/reference/润滑油压力演化周期.csv",
         "candidate": "/Users/zhouhao/Code/Python/Compressor/models/results/pressure/润滑油压力演化周期.csv",
-        "identical": true
+        "identical": true,
+        "mode": "byte"
       },
       {
         "group": "vibration",
         "file": "振动异常段.csv",
         "reference": "/Users/zhouhao/Code/Python/Compressor/models/datas/reference/振动异常段.csv",
         "candidate": "/Users/zhouhao/Code/Python/Compressor/models/results/vibration/振动异常段.csv",
-        "identical": true
+        "identical": true,
+        "mode": "subset",
+        "extra_columns": [
+          "模型异常概率均值",
+          "模型异常概率最大值",
+          "模型异常概率最小值",
+          "模型预测参考"
+        ]
       },
       {
         "group": "common",
         "file": "报警阈值配置.csv",
         "reference": "/Users/zhouhao/Code/Python/Compressor/models/datas/reference/报警阈值配置.csv",
         "candidate": "/Users/zhouhao/Code/Python/Compressor/models/results/common/报警阈值配置.csv",
-        "identical": true
+        "identical": true,
+        "mode": "byte"
       }
     ]
   },

models/scan_all.py

@@ -29,7 +29,7 @@ from pathlib import Path
 
 sys.path.insert(0, str(Path(__file__).resolve().parents[1]))
 
-from models.common import alarm_store, db, devices, io, paths, queries
+from models.common import alarm_store, db, devices, io, model_predict, paths, queries
 from models.generalized import pressure_algorithm
 from models.vibration import vibration_algorithm
 
@@ -72,6 +72,7 @@ def main() -> None:
     parser.add_argument("--chunk-days", type=int, default=14)
     parser.add_argument("--hourly-csv", default=None, help="复用已落盘的小时中间表，不查询数据库")
     parser.add_argument("--store", choices=("local", "db"), default="local")
+    parser.add_argument("--with-model", action="store_true", help="对故障段做模型滑动预测（默认关闭）")
     parser.add_argument("--dry-run", action="store_true", help="只统计不写库（store=db 时）")
     args = parser.parse_args()
 
@@ -112,8 +113,15 @@ def main() -> None:
     print(f"润滑油压力周期：{len(cycles):,} 个")
     print(f"振动异常段：{len(events):,} 段")
 
-    pressure_algorithm.write_results(paths.PRESSURE_RESULTS, hourly_levels, hour_segments, cycles)
-    vibration_algorithm.write_results(paths.VIBRATION_RESULTS, events)
+    extra_fields = ()
+    if args.with_model:
+        pressure_stats = model_predict.annotate("pressure", rows, hour_segments)
+        vibration_stats = model_predict.annotate("vibration", rows, events)
+        print(f"模型预测：压力 {pressure_stats}，振动 {vibration_stats}")
+        extra_fields = tuple(model_predict.MODEL_FIELDS) + (model_predict.NOTE_FIELD,)
+
+    pressure_algorithm.write_results(paths.PRESSURE_RESULTS, hourly_levels, hour_segments, cycles, extra_fields=extra_fields)
+    vibration_algorithm.write_results(paths.VIBRATION_RESULTS, events, extra_fields=extra_fields)
     print(f"结果已写入本地：{paths.RESULTS}")
 
     if args.store == "db":

models/vibration/vibration_algorithm.py

@@ -312,9 +312,10 @@ def run(rows: list[dict], configs: dict[int, dict]):
     return merge_vibration_signals(signals, configs)
 
 
-def write_results(output_dir: Path, events):
+def write_results(output_dir: Path, events, extra_fields=()):
     output_dir = Path(output_dir)
-    io.write_csv(output_dir / "振动异常段.csv", io.clean_rows(events), VIBRATION_FIELDS, {field: field for field in VIBRATION_FIELDS})
+    fields = [field for field in VIBRATION_FIELDS if field not in extra_fields] + list(extra_fields)
+    io.write_csv(output_dir / "振动异常段.csv", io.clean_rows(events), fields, {field: field for field in fields})
 
 
 def load_hourly(args, selected_batches, unit_by_batch, configs):