#!/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 "") # ── 7. 提示词一致性检查 ── def check_prompt_sync(): """检查注册的cron prompt版本与jobs.json实际运行的是否一致""" try: reg_path = "/home/hmo/MoFin/data/prompts/registry.json" cron_path = "/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json" if not os.path.exists(reg_path) or not os.path.exists(cron_path): check("提示词一致性", False, "registry或jobs.json不存在") return with open(reg_path) as f: reg = json.load(f) with open(cron_path) as f: crons = json.load(f).get("jobs", []) # 建立cron prompt索引 cron_prompts = {} for j in crons: name = j.get("name", "") pid = name.replace(" ","-").replace("（","-").replace("）","").replace("(","-").replace(")","") if j.get("prompt"): cron_prompts[pid] = {"name": name, "prompt": j["prompt"]} # 逐个检查注册的版本文件是否与cron一致 drift = 0 for p in reg.get("prompts", []): pid = p["id"] if pid not in cron_prompts: continue cv = p.get("current_version", "v1") # 找版本文件路径 content_path = "" for v in p.get("versions", []): if v.get("version") == cv: content_path = v.get("content_path", "") break if not content_path or not os.path.exists(content_path): drift += 1 continue with open(content_path) as f: registered = f.read() actual = cron_prompts[pid]["prompt"] if registered != actual: drift += 1 check("提示词一致性: 注册版≈运行版", drift == 0, f"{drift}个prompt不一致" if drift else "") except Exception as e: check("提示词一致性", False, str(e)) def check_db_locks(): """检查最近24小时是否有 database is locked 错误""" try: import subprocess r = subprocess.run( ["sudo", "journalctl", "-u", "xmpp-zhiwei", "--since", "24 hours ago", "--no-pager"], capture_output=True, text=True, timeout=30 ) locked_lines = [l for l in r.stdout.split("\n") if "database is locked" in l.lower()] check("DB死锁: 24h内lock错误", len(locked_lines) == 0, f"发现{len(locked_lines)}次: {locked_lines[0][:120]}" if locked_lines else "") except Exception as e: check("DB死锁检查", False, str(e)) if __name__ == "__main__": print("🔍 MoFin 开盘前验证", datetime.now().strftime("%m-%d %H:%M")) print("=" * 40) check_sync() check_imports() check_db() check_smoke() check_cron_timestamps() check_db_locks() check_prompt_sync() # 汇总 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)