diff --git a/scripts/preflight_verify.py b/scripts/preflight_verify.py
new file mode 100644
index 00000000..0fdbe001
--- /dev/null
+++ b/scripts/preflight_verify.py
@@ -0,0 +1,188 @@
+#!/usr/bin/env python3
+"""preflight_verify.py — 开盘前钉对钉验证
+
+在 morning_health_check 之后运行（8:30），专测今天踩过的坑：
+  1. MoFin ↔ cron 脚本同步
+  2. 关键脚本可正常导入（修 import 遗漏）
+  3. 关键函数可正常调用（修 NameError / 参数不匹配）
+  4. DB 现金/资产公式一致性
+  5. 关键 no_agent 脚本可执行（exit=0）
+
+输出格式：PASS/FAIL 逐项。任意 FAIL → 写 TODO + 推报警。
+"""
+import sys, os, json, sqlite3, subprocess
+from pathlib import Path
+from datetime import datetime
+
+MOFIN = Path("/home/hmo/MoFin")
+DATA = MOFIN / "data"
+DB_PATH = DATA / "mofin.db"
+PROFILE_SCRIPTS = Path("/home/hmo/.hermes/profiles/position-analyst/scripts")
+
+results = {"pass": 0, "fail": 0, "items": []}
+def check(name, ok, detail=""):
+    results["items"].append({"name": name, "ok": bool(ok), "detail": detail})
+    if ok:
+        results["pass"] += 1
+        print(f"  ✅ {name}")
+    else:
+        results["fail"] += 1
+        print(f"  ❌ {name}")
+        if detail:
+            print(f"      {detail}")
+
+# ── 1. 同步检查 ──
+def check_sync():
+    r = subprocess.run(["bash", str(MOFIN/"scripts/deploy_sync.sh"), "--check"],
+                       capture_output=True, text=True, timeout=15)
+    ok = "一致" in r.stdout
+    check("脚本同步: MoFin↔cron目录", ok, r.stdout.strip())
+
+# ── 2. 关键脚本导入检查 ──
+SCRIPTS = [
+    "price_monitor", "strategy_review", "collect_evaluation_data",
+    "data_governance", "market_screener", "stock_quote",
+    "strategy_evaluator", "mo_data", "mofin_db",
+]
+def check_imports():
+    for name in SCRIPTS:
+        path = PROFILE_SCRIPTS / f"{name}.py"
+        if not path.exists():
+            check(f"导入检查: {name}", False, f"文件不存在: {path}")
+            continue
+        # 检查是否有明显的语法/依赖错误
+        r = subprocess.run([sys.executable, "-c", f"import sys; sys.path.insert(0, '{PROFILE_SCRIPTS}'); import {name}; print('ok')"],
+                          capture_output=True, text=True, timeout=10)
+        ok = r.returncode == 0
+        err = r.stderr.strip()[:200] if not ok else ""
+        check(f"导入检查: {name}", ok, err)
+
+# ── 3. DB 完整性检查 ──
+def check_db():
+    if not DB_PATH.exists():
+        check("DB 文件存在", False, f"找不到 {DB_PATH}")
+        return
+    try:
+        conn = sqlite3.connect(str(DB_PATH))
+        pragma = conn.execute("PRAGMA integrity_check").fetchone()[0]
+        check("DB 完整性", pragma == "ok", pragma)
+        
+        # 检查核心表是否存在
+        core_tables = ["portfolio_summary", "holdings", "holding_strategies", "cash_log", "live_prices"]
+        missing = []
+        for t in core_tables:
+            r = conn.execute(f"SELECT count(*) FROM sqlite_master WHERE type='table' AND name='{t}'").fetchone()[0]
+            if r == 0: missing.append(t)
+        check("核心表完整", len(missing)==0, f"缺失: {missing}" if missing else "")
+        
+        # 现金/资产公式检查
+        try:
+            ps = conn.execute("SELECT id, cash, frozen_cash, total_mv, total_assets, updated_at FROM portfolio_summary ORDER BY id DESC LIMIT 1").fetchone()
+            if ps:
+                # total_assets = total_mv + cash + frozen_cash
+                expected = ps[3] + ps[1] + ps[2] if ps[3] else 0
+                actual = ps[4]
+                if expected > 0 and actual:
+                    diff = abs(actual - expected)
+                    ok = diff < 1.0  # 最多差1元
+                    check(f"资产公式: total_mv({ps[3]:.0f})+cash({ps[1]:.0f})+frozen({ps[2]:.0f})={expected:.0f} vs total_assets({actual:.0f})",
+                          ok, f"差异{diff:.2f}")
+                else:
+                    check("资产公式验证", False, f"total_mv或total_assets为空: mv={ps[3]}, assets={ps[4]}")
+            else:
+                check("资产公式验证", False, "portfolio_summary无数据")
+        except Exception as e:
+            check("资产公式验证", False, str(e))
+        
+        conn.close()
+    except Exception as e:
+        check("DB 检查异常", False, str(e))
+
+# ── 4. 关键 no_agent 脚本快速执行检查 ──
+SMOKE_SCRIPTS = [
+    ("stock_quote.py", ["--code", "000001", "--once"]),
+    ("strategy_review.py", None),  # 不加参数运行
+]
+def check_smoke():
+    for name, args in SMOKE_SCRIPTS:
+        cmd = [sys.executable, str(PROFILE_SCRIPTS / name)]
+        if args: cmd.extend(args)
+        r = subprocess.run(cmd, capture_output=True, text=True, timeout=60)
+        ok = r.returncode == 0
+        err = r.stderr.strip()[:200] if not ok else ""
+        out = r.stdout.strip()[:100] if ok else ""
+        check(f"冒烟测试: {name}", ok, err or out)
+
+# ── 5. 关键cron时间戳检查 ──
+def check_cron_timestamps():
+    cron_path = Path("/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json")
+    if not cron_path.exists():
+        check("cron时间戳", False, "jobs.json不存在")
+        return
+    with open(cron_path) as f:
+        jobs = json.load(f).get("jobs", [])
+    now = datetime.now()
+    stale_count = 0
+    for j in jobs:
+        if not j.get("enabled", True):
+            continue
+        name = j.get("name", "?")
+        last = j.get("last_run_at", "")
+        status = j.get("last_status", "")
+        if not last or not status:
+            continue
+        # 非周任务检查是否在48小时内跑过
+        try:
+            last_dt = datetime.fromisoformat(last)
+            days_gone = (now - last_dt).total_seconds() / 86400
+            # 周任务可接受7天，日任务可接受2天
+            limit = 7 if any(k in name for k in ["每周", "周末", "周六"]) else 2
+            if days_gone > limit and status != "ok":
+                stale_count += 1
+        except:
+            pass
+    check("cron过期检查", stale_count == 0, f"{stale_count}个cron长时间未正常运行" if stale_count else "")
+
+if __name__ == "__main__":
+    print("🔍 MoFin 开盘前验证", datetime.now().strftime("%m-%d %H:%M"))
+    print("=" * 40)
+    check_sync()
+    check_imports()
+    check_db()
+    check_smoke()
+    
+    # 汇总
+    print()
+    total = results["pass"] + results["fail"]
+    status = "✅ 全部通过" if results["fail"] == 0 else f"❌ {results['fail']}/{total} 项失败"
+    summary = {
+        "timestamp": datetime.now().isoformat(),
+        "total": total,
+        "pass": results["pass"],
+        "fail": results["fail"],
+        "status": "pass" if results["fail"] == 0 else "fail",
+        "items": results["items"],
+    }
+    summary_path = DATA / "preflight_result.json"
+    with open(summary_path, "w") as f:
+        json.dump(summary, f, ensure_ascii=False, indent=2)
+    print(f"{status} | {summary_path}")
+    
+    # FAIL 项写 TODO
+    if results["fail"] > 0:
+        failures = [i["name"] for i in results["items"] if not i["ok"]]
+        todo_text = f"开盘前验证发现 {results['fail']} 项失败: {'; '.join(failures)}"
+        print(f"\n⚠️ 已记录TODO: {todo_text}")
+        # 尝试写mofin.db todos
+        try:
+            conn = sqlite3.connect(str(DB_PATH))
+            conn.execute("""
+                INSERT OR REPLACE INTO todos (id, content, status, source, created_at, fix_action)
+                VALUES (?, ?, 'pending', 'preflight', datetime('now'), 'check preflight_result.json and fix issues')
+            """, (f"preflight-{datetime.now().strftime('%Y%m%d')}", todo_text))
+            conn.commit()
+            conn.close()
+        except:
+            pass
+    
+    sys.exit(0 if results["fail"] == 0 else 1)