fix: 统一DB路径+管道审计脚本

mo_data.py DB_PATH: /home/hmo/web-dashboard/data/mofin.db → /home/hmo/MoFin/data/mofin.db
新增 verify_reassess_pipeline.py:管道审计,静默时无声、异常时报警
This commit is contained in:
知微
2026-07-07 10:37:05 +08:00
parent 26e2794cca
commit 7c255d5234
2 changed files with 119 additions and 1 deletions
+1 -1
View File
@@ -16,7 +16,7 @@ JSON 文件已弃用,仅保留为历史备份。
import sqlite3, json
from datetime import datetime
DB_PATH = '/home/hmo/web-dashboard/data/mofin.db'
DB_PATH = '/home/hmo/MoFin/data/mofin.db'
def _get_db():
+118
View File
@@ -0,0 +1,118 @@
#!/usr/bin/env python3
"""verify_reassess_pipeline.py — 重评推送管道审计
检查三个代码级约束是否在正常运转:
1. price_monitor 每2分触发 → 突破检测 → reassess调用
2. stale_detector/stale_push_wlin 自选买入区检测
3. mofin_collect LLM前强制重评
输出:JSON格式管道状态(无异常时静默[SILENT])
"""
import json, os, sys, subprocess
from pathlib import Path
from datetime import datetime, timedelta
BASE = Path(__file__).parent.parent
sys.path.insert(0, str(BASE))
sys.path.insert(0, "/home/hmo/MoFin")
try:
from mofin_db import get_conn
conn = get_conn()
cur = conn.cursor()
except Exception:
conn = None
now = datetime.now()
ok = True
checks = []
# 1. price_monitor 最近运行时间
try:
if conn:
lp = cur.execute("SELECT MAX(updated_at) FROM live_prices").fetchone()[0]
if lp:
lp_dt = datetime.fromisoformat(lp) if isinstance(lp, str) else lp
mins_ago = (now - lp_dt).total_seconds() / 60
status = "ok" if mins_ago < 10 else "warn"
if mins_ago > 15:
status = "fail"
ok = False
checks.append({"check":"price_monitor","status":status,"detail":f"最后更新{mins_ago:.0f}分前 @ {lp}"})
else:
checks.append({"check":"price_monitor","status":"warn","detail":"live_prices无数据"})
except Exception as e:
checks.append({"check":"price_monitor","status":"fail","detail":str(e)})
ok = False
# 2. 今日价格事件(有无zone breach触发)
try:
if conn:
pe = cur.execute("SELECT COUNT(*) FROM price_events WHERE date=date('now')").fetchone()[0]
checks.append({"check":"zone_breaches_today","status":"ok","detail":f"{pe}次突破事件"})
except Exception as e:
checks.append({"check":"zone_breaches","status":"fail","detail":str(e)})
# 3. holding_strategies 是否有策略数据
try:
if conn:
hs = cur.execute("SELECT COUNT(*) FROM holding_strategies").fetchone()[0]
valid = cur.execute("SELECT COUNT(*) FROM holding_strategies WHERE stop_loss IS NOT NULL").fetchone()[0]
status = "ok" if hs > 0 else "warn"
if hs == 0:
ok = False
checks.append({"check":"strategies_in_db","status":status,"detail":f"{hs}条策略(含止损{valid}条)"})
except Exception as e:
checks.append({"check":"strategies","status":"fail","detail":str(e)})
# 4. mofin_collect 今日是否跑过(检查策略新鲜度日志)
try:
if conn:
fresh = cur.execute("""
SELECT COUNT(*) FROM holding_strategies
WHERE date(created_at)=date('now')
""").fetchone()[0]
checks.append({"check":"today_reassessed","status":"ok","detail":f"今日{('reassess刷新'+str(fresh)+'') if fresh else '暂无新重评记录'}"})
except Exception as e:
checks.append({"check":"today_reassess","status":"warn","detail":str(e)})
# 5. XMPP bridge是否在线
try:
import urllib.request
req = urllib.request.Request("http://127.0.0.1:5805/",
data=b'{"to":"hmo@yoin.fun","body":"ping","type":"chat"}',
headers={"Content-Type":"application/json"})
resp = urllib.request.urlopen(req, timeout=3)
result = json.loads(resp.read())
bridge_ok = result.get("ok") == True
checks.append({"check":"xmpp_bridge_5805","status":"ok" if bridge_ok else "fail","detail":"在线" if bridge_ok else "异常"})
if not bridge_ok:
ok = False
except Exception as e:
checks.append({"check":"xmpp_bridge","status":"fail","detail":str(e)})
ok = False
# 6. 验证reassess_with_context可导入
try:
sys.path.insert(0, "/home/hmo/MoFin")
from strategy_lifecycle import reassess_with_context
checks.append({"check":"reassess_module","status":"ok","detail":"reassess_with_context可导入"})
except Exception as e:
checks.append({"check":"reassess_module","status":"fail","detail":str(e)})
ok = False
if conn:
conn.close()
# 输出:无异常静默,有异常报警
result = {
"pipeline": "ok" if ok else "degraded",
"checked_at": now.isoformat(),
"checks": checks
}
if ok:
# 只有异常时才出声
print("[SILENT]")
else:
print(json.dumps(result, ensure_ascii=False, indent=2))