fix(monitor): 监控三 P1(审计D-1/2/3)——D-1 collect_log_layer跨日志归并改mtime序后到运行整行胜出(旧max(rows)吞后到failed+假resolve双失效:早班rows=100/failed=0压住后到rows=50/failed=3); D-2 main顶层兜底(_run_shift+_crash_exit三段壳:裸崩溃写red health+infra-monitor-crash告警+exit3,旧裸传播+wrapper恒0=SCHTASKS年龄失明最坏22h静默)+check_entry单检隔离(抛错=黄降级data-X-stale对齐strategy先例); D-3 trading_days读侧[:10]归一+None过滤+sorted(双实现分叉收敛,dbbardata datetime历史双格式); 四新测+monitoring spec §5.2/§5.3-④ [vps] [nas]

scripts/data_platform/data_monitor.py

@@ -12,6 +12,7 @@ import json
 import os
 import subprocess
 import sys
+import traceback
 
 _HERE = os.path.dirname(os.path.abspath(__file__))
 _REPO = os.path.dirname(os.path.dirname(_HERE))
@@ -54,6 +55,18 @@ def _ro_conn(path):
 # ---------- 结果层: 逐注册表项(log_only 归过程层, 不入此处) ----------
 
 def check_entry(e, ctx):
+    """注册表项分发; 单检抛错=黄降级不阻断全班(D-2: 对齐 strategy 侧
+    strategy_monitor.check_entry 先例——旧码无隔离, 任一检 KeyError/TypeError
+    =整班崩+wrapper 恒 0=SCHTASKS 年龄判定失明, 最坏 ~22h 仅靠 peer 反盯)。"""
+    try:
+        return _check_entry_impl(e, ctx)
+    except Exception as exc:  # noqa: BLE001
+        return {"key": f"data-{e['name']}-stale", "status": "yellow",
+                "expected": e.get("title", e["name"]),
+                "actual": f"error: {exc!r}", "evidence": {"error": repr(exc)}}
+
+
+def _check_entry_impl(e, ctx):
     # root 语义: "static"→<data_root>/static/<subdir>; "data"→<data_root>/<subdir>
     # (09-26 修正: 首版误拼 data/data/... 致三域恒黄; fixture 曾同镜像故测试假绿)
     if e.get("root") == "data":
@@ -239,6 +252,7 @@ def collect_log_layer(data_root, now, window_h=_LOG_WINDOW_H):
     keep = [p for p in logs
             if (now - dt.datetime.fromtimestamp(os.path.getmtime(p))
                 ).total_seconds() / 3600 <= window_h]
+    keep.sort(key=os.path.getmtime)   # D-1: mtime 序 → 后到运行整行胜出
     stats, sigs_by_type = {}, {}
     for p in keep:
         try:
@@ -247,9 +261,10 @@ def collect_log_layer(data_root, now, window_h=_LOG_WINDOW_H):
         except OSError:
             continue
         for t, s in chk.parse_stat_lines(text).items():
-            prev = stats.get(t)
-            stats[t] = s if prev is None or s.get("rows", 0) >= prev.get("rows", 0) \
-                else prev
+            # D-1(10-05): 旧 max(rows) 归并吞掉后到运行的 failed + 假
+            # resolve(早班 rows=100/failed=0 压住后到 rows=50/failed=3 →
+            # 漏报+既有告警被假自愈双失效)——改后到运行(mtime 新)整行胜出
+            stats[t] = s
         # 10-05: 签名按 [type] 归因(跨类型连坐修), 跨文件按 sig 去重合并
         for t, ss in chk.signatures_by_type(text).items():
             bucket = sigs_by_type.setdefault(t, [])
@@ -438,11 +453,21 @@ def main(argv=None):
     ap.add_argument("--skip-inbox", action="store_true")
     args = ap.parse_args(argv)
 
-    now = _now()
-    today = now.date().isoformat().replace("-", "")
     db_path = args.db
     if args.db == "data/alerts.db" and args.data_root != "data":
         db_path = os.path.join(args.data_root, "alerts.db")
+    # D-2(10-05): 整班顶层兜底 —— 裸崩溃时 wrapper 恒 0 + last_run 新鲜 =
+    # SCHTASKS 年龄判定失明(审计推演最坏 ~22h 仅 peer 反盯); 兜底写 red
+    # health + infra-monitor-crash 告警后 exit 3, 静默窗收敛到本班内
+    try:
+        return _run_shift(args, db_path)
+    except Exception:                            # noqa: BLE001
+        return _crash_exit(args, db_path)
+
+
+def _run_shift(args, db_path):
+    now = _now()
+    today = now.date().isoformat().replace("-", "")
 
     days = chk.trading_days(
         os.path.join(args.data_root, "quant_trading.db"), now.date())
@@ -544,5 +569,33 @@ def main(argv=None):
     return 3 if overall == "red" else 0
 
 
+def _crash_exit(args, db_path):
+    """D-2(10-05): 整班崩溃兜底出口 —— red health + infra-monitor-crash
+    告警 + exit 3(告警/health 任一写失败也不掩原始崩溃事实)。"""
+    now = _now()
+    today = now.date().isoformat().replace("-", "")
+    detail = traceback.format_exc()
+    sys.stderr.write(f"[data-monitor] CRASH 整班崩溃(D-2 兜底):\n{detail}\n")
+    try:
+        conn = store.connect(db_path)
+        store.upsert_alert(conn, _event(
+            "infra-monitor-crash", "red", "data-monitor 整班崩溃(顶层兜底)",
+            detail[-800:], {"data_root": args.data_root, "shift": args.shift},
+            source="infra", today=today))
+        conn.close()
+    except Exception:                            # noqa: BLE001 崩溃中的崩溃
+        pass
+    try:
+        write_health(args.data_root, {
+            "date": now.date().isoformat(), "shift": args.shift, "host": "vps",
+            "generated_at": now.isoformat(timespec="seconds"),
+            "overall": "red", "checks_overall": "red", "alerts_overall": "red",
+            "counts": {"green": 0, "yellow": 0, "red": 0},
+            "alerts_emitted": 1, "checks": [], "crash": detail[-400:]})
+    except Exception:                            # noqa: BLE001
+        pass
+    return 3
+
+
 if __name__ == "__main__":
     sys.exit(main())

scripts/data_platform/monitor_checks.py

@@ -24,7 +24,10 @@ def trading_days(db_path, today, lookback=50):
                 "AND datetime>=? ORDER BY datetime", (cutoff,)).fetchall()
         finally:
             conn.close()
-        return [r[0] for r in rows]
+        # D-3(10-05): 读侧 [:10] 归一+None 过滤, 对齐 strategy_checks 同名
+        # 版(双实现已分叉; dbbardata datetime 历史双格式, 写侧归一不能
+        # 担保读侧形状——脏值曾致 expected_daily 全 per_date 判定崩坏)
+        return sorted({str(r[0])[:10] for r in rows if r[0]})
     except sqlite3.Error:
         return []