206 lines
8.2 KiB
Python
206 lines
8.2 KiB
Python
#!/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 "")
|
||
|
||
# ── 6. DB死锁检测 ──
|
||
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()
|
||
|
||
# 汇总
|
||
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)
|