feat(db): 每日DB在线备份(07:50,留14天) + 管道审计malformed重试一次再告警(瞬态WAL损坏防误报)

deploy/profile-scripts/db_daily_backup.py

@@ -0,0 +1,50 @@
+#!/usr/bin/env python3
+"""db_daily_backup.py — mofin.db 每日在线备份（sqlite3 backup API，一致性快照）
+
+背景（2026-07-21）：I/O 风暴导致 WAL 帧损坏，出现 transient
+"database disk image is malformed"。DB 无定期备份 = 单点风险。
+
+策略：
+- 每日 07:50（premarket 前）备份到 /home/hmo/MoFin/data/backups/
+- sqlite3 .backup API：在线一致性快照，不锁库
+- 保留最近 14 天，自动清理
+"""
+import sqlite3, os, glob
+from datetime import datetime
+
+SRC = "/home/hmo/MoFin/data/mofin.db"
+DST_DIR = "/home/hmo/MoFin/data/backups"
+KEEP = 14
+
+
+def main():
+    os.makedirs(DST_DIR, exist_ok=True)
+    stamp = datetime.now().strftime("%Y%m%d")
+    dst = os.path.join(DST_DIR, f"mofin_{stamp}.db")
+
+    src = sqlite3.connect(f"file:{SRC}?mode=ro", uri=True, timeout=30)
+    dst_conn = sqlite3.connect(dst, timeout=60)
+    src.backup(dst_conn)
+    dst_conn.close()
+    src.close()
+
+    size = os.path.getsize(dst) / 1024 / 1024
+    print(f"[BACKUP] {dst} ({size:.1f}MB)")
+
+    # 完整性验证（备份不可信直到验证过）
+    conn = sqlite3.connect(f"file:{dst}?mode=ro", uri=True, timeout=30)
+    ok = conn.execute("PRAGMA quick_check").fetchone()[0] == "ok"
+    conn.close()
+    print(f"[VERIFY] quick_check: {'ok' if ok else 'FAIL!'}")
+
+    # 清理旧备份
+    backups = sorted(glob.glob(os.path.join(DST_DIR, "mofin_*.db")))
+    for old in backups[:-KEEP]:
+        os.remove(old)
+        print(f"[CLEAN] 删除旧备份 {os.path.basename(old)}")
+
+    return 0 if ok else 1
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

deploy/profile-scripts/verify_reassess_pipeline.py

@@ -93,8 +93,22 @@ def run():
     checks = []
 
     # 1. price_monitor 最近运行时间
+    try:
+        conn = None
+        last_err = None
+        # malformed 可能是 I/O 风暴下的瞬态 WAL 损坏（2026-07-21 事件）：
+        # checkpoint 后自愈。重试一次再告警，避免误报轰炸
+        for _attempt in range(2):
             try:
                 conn = sqlite3.connect("/home/hmo/MoFin/data/mofin.db")
+                conn.execute("SELECT 1 FROM live_prices LIMIT 1").fetchone()
+                break
+            except Exception as e:
+                last_err = e
+                import time as _t
+                _t.sleep(3)
+        if conn is None:
+            raise last_err
         lp = conn.execute("SELECT MAX(updated_at) FROM live_prices").fetchone()[0]
         if lp:
             lp_dt = datetime.fromisoformat(lp) if isinstance(lp, str) else lp