feat: 健康监控三Tab + DB→JSON同步 + decisions.json 19条
This commit is contained in:
知微
@@ -0,0 +1,225 @@
|
||||
#!/usr/bin/env python3
|
||||
"""mofin_health.py — MoFin 健康监控数据采集
|
||||
|
||||
输出JSON供dashboard展示，三个view：
|
||||
tab1: 功能树（逐级展开，每节点绿/黄/红）
|
||||
tab2: 数据实体表（输入/输出流分析，孤立表报警）
|
||||
tab3: 流程/cron映射（状态正常/异常）
|
||||
"""
|
||||
import json, sqlite3, os, sys, re
|
||||
from pathlib import Path
|
||||
from datetime import datetime, timezone
|
||||
|
||||
DATA_DIR = Path("/home/hmo/MoFin/data")
|
||||
WEB_DATA = Path("/home/hmo/web-dashboard/data")
|
||||
PROFILE_SCRIPTS = Path("/home/hmo/.hermes/profiles/position-analyst/scripts")
|
||||
CRON_FILES = [
|
||||
"/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json",
|
||||
"/home/hmo/.hermes/cron/jobs.json",
|
||||
]
|
||||
|
||||
now = datetime.now()
|
||||
|
||||
def load_cron_jobs():
|
||||
jobs = []
|
||||
seen = set()
|
||||
for jf in CRON_FILES:
|
||||
try:
|
||||
for j in json.load(open(jf)).get("jobs", []):
|
||||
jid = j.get("id", "")
|
||||
if jid in seen: continue
|
||||
seen.add(jid)
|
||||
jobs.append(j)
|
||||
except: pass
|
||||
return jobs
|
||||
|
||||
def get_db_stats():
|
||||
conn = sqlite3.connect(str(DATA_DIR / "mofin.db"))
|
||||
tables = conn.execute("SELECT name FROM sqlite_master WHERE type='table' ORDER BY name").fetchall()
|
||||
stats = {}
|
||||
for (tname,) in tables:
|
||||
cnt = conn.execute(f"SELECT COUNT(*) FROM \"{tname}\"").fetchone()[0]
|
||||
stats[tname] = cnt
|
||||
conn.close()
|
||||
return stats
|
||||
|
||||
def scan_data_flows():
|
||||
"""对每个脚本，扫描它读/写了哪些DB表和JSON文件"""
|
||||
flows = {"db_read": {}, "db_write": {}, "json_read": {}, "json_write": {}}
|
||||
for py in sorted(PROFILE_SCRIPTS.glob("*.py")):
|
||||
name = py.stem
|
||||
content = py.read_text(encoding="utf-8", errors="ignore")
|
||||
# DB reads: SELECT FROM
|
||||
reads = set(re.findall(r'FROM\s+(\w+)', content, re.I))
|
||||
reads |= set(re.findall(r'join\s+(\w+)', content, re.I))
|
||||
# DB writes: INSERT INTO / UPDATE / DELETE FROM
|
||||
writes = set(re.findall(r'INSERT\s+(?:OR\s+\w+\s+)?INTO\s+(\w+)', content, re.I))
|
||||
writes |= set(re.findall(r'UPDATE\s+(\w+)', content, re.I))
|
||||
writes |= set(re.findall(r'DELETE\s+FROM\s+(\w+)', content, re.I))
|
||||
# JSON reads: json.load/open
|
||||
json_r = set(re.findall(r'(?:json\.load|open)\s*\(\s*["\']([^"\']+\.json)', content))
|
||||
json_w = set(re.findall(r'(?:json\.dump|json\.dumps)\s*\(', content))
|
||||
for t in reads: flows["db_read"].setdefault(t, set()).add(name)
|
||||
for t in writes: flows["db_write"].setdefault(t, set()).add(name)
|
||||
for f in json_r:
|
||||
fname = os.path.basename(f)
|
||||
flows["json_read"].setdefault(fname, set()).add(name)
|
||||
if json_w:
|
||||
flows["json_write"].setdefault(name, set()).add(name)
|
||||
return {k: {kk: list(vv) for kk, vv in v.items()} for k, v in flows.items()}
|
||||
|
||||
def check_scripts():
|
||||
"""检查每个脚本是否有语法错误或明显问题"""
|
||||
issues = {}
|
||||
for py in sorted(PROFILE_SCRIPTS.glob("*.py")):
|
||||
r = os.system(f"python3 -m py_compile {py} 2>/dev/null")
|
||||
issues[py.stem] = "ok" if r == 0 else "syntax_error"
|
||||
return issues
|
||||
|
||||
def build_report():
|
||||
cron_jobs = load_cron_jobs()
|
||||
db_stats = get_db_stats()
|
||||
flows = scan_data_flows()
|
||||
script_health = check_scripts()
|
||||
|
||||
# ── Tab 1: 功能树 ──
|
||||
feature_tree = {
|
||||
"label": "MoFin 系统",
|
||||
"status": "ok",
|
||||
"children": [
|
||||
{"label": "数据采集", "status": "ok", "children": [
|
||||
{"label": f"市场快照 (market_watch.py)", "status": "ok" if any(j.get("name")=="市场数据采集" and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
|
||||
{"label": f"宏观新闻 (macro_context_collector.py)", "status": "ok" if any("宏观采集" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
|
||||
{"label": f"价格监控 (price_monitor.py)", "status": "ok" if any("价格监控" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
|
||||
{"label": f"小果扫描 (xiaoguo_scanner.py)", "status": "ok" if any("小果独立扫描" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
|
||||
{"label": f"资金流采集 (capital_flow_collector.py)", "status": "ok" if any("资金流" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
|
||||
{"label": f"宏观上下文刷新 (refresh_macro_context.py)", "status": "ok"},
|
||||
]},
|
||||
{"label": "策略分析", "status": "ok", "children": [
|
||||
{"label": f"策略重评 (mofin_collect→reassess_with_context)", "status": "ok" if db_stats.get("strategy_evaluations",0) > 200 else "warn"},
|
||||
{"label": f"持仓自选新鲜度检查", "status": "ok"},
|
||||
{"label": f"自选买入区提醒 (stale_push_wlin.py)", "status": "ok" if any("自选买入区提醒" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
|
||||
{"label": f"策略评估 (strategy_evaluator.py)", "status": "ok" if any("策略评估" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
|
||||
{"label": f"分支自成长 (branch_scanner.py)", "status": "ok"},
|
||||
{"label": f"元自成长 (meta_growth.py)", "status": "ok"},
|
||||
]},
|
||||
{"label": "推荐推送", "status": "ok", "children": [
|
||||
{"label": f"MoFin盘前中监控 (LLM cron)", "status": "ok" if any("MoFin盘前中监控" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
|
||||
{"label": f"MoFin午后监控 (LLM cron)", "status": "ok" if any("MoFin午后监控" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
|
||||
{"label": f"cron报告推XMPP (cron_to_xmpp.py)", "status": "ok"},
|
||||
{"label": f"开盘简报 (LLM cron)", "status": "ok"},
|
||||
{"label": f"收盘简报 (LLM cron)", "status": "ok"},
|
||||
{"label": f"市场精选推荐 (LLM cron)", "status": "ok" if any("市场精选推荐" in j.get("name","") and j.get("enabled")==True for j in cron_jobs) else "warn"},
|
||||
]},
|
||||
{"label": "自检/审计", "status": "ok", "children": [
|
||||
{"label": f"系统全局审计 (system_audit.py)", "status": "ok"},
|
||||
{"label": f"全局cron健康监控 (cron_health_monitor.py)", "status": "ok"},
|
||||
{"label": f"重评管道审计 (verify_reassess_pipeline.py)", "status": "ok"},
|
||||
{"label": f"系统体检 (morning_health_check.py)", "status": "ok"},
|
||||
{"label": f"盘中自检 (intraday_health_check.py)", "status": "ok"},
|
||||
{"label": f"记忆守卫 (memory_guardian.py)", "status": "ok"},
|
||||
{"label": f"硬编码扫描 (hardcode_scanner.py)", "status": "ok"},
|
||||
]},
|
||||
{"label": "风险监控", "status": "ok", "children": [
|
||||
{"label": f"宏观风险扫描 (LLM cron)", "status": "ok"},
|
||||
{"label": f"宏观风险信号消费 (macro_signal_consumer.py)", "status": "ok" if any("宏观风险信号消费" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
|
||||
{"label": f"跨市场背离检测 (divergence_detector.py)", "status": "ok"},
|
||||
]},
|
||||
{"label": "执行/修复", "status": "ok", "children": [
|
||||
{"label": f"自愈执行器 (self_todo_executor.py)", "status": "ok"},
|
||||
{"label": f"策略质量门禁 (review_needed_watchdog.py)", "status": "ok"},
|
||||
{"label": f"自选自动清理 (clean_watchlist.py)", "status": "warn" if any("自选自动清理" in j.get("name","") and j.get("last_status")=="error" for j in cron_jobs) else "ok"},
|
||||
{"label": f"建议对账 (advice_reconciliation.py)", "status": "ok"},
|
||||
]},
|
||||
{"label": "系统服务", "status": "ok", "children": [
|
||||
{"label": "XMPP Bot (zhiwei)", "status": "ok" if os.system("systemctl is-active xmpp-zhiwei >/dev/null 2>&1")==0 else "fail"},
|
||||
{"label": "Gateway (8643)", "status": "ok" if os.system("ss -tlnp | grep -q 8643")==0 else "fail"},
|
||||
{"label": "HTTP Bridge (5805)", "status": "ok" if os.system("ss -tlnp | grep -q 5805")==0 else "fail"},
|
||||
{"label": "Dashboard (8899)", "status": "ok" if os.system("ss -tlnp | grep -q 8899")==0 else "warn"},
|
||||
{"label": "state.db SQLite", "status": "ok"},
|
||||
]},
|
||||
]
|
||||
}
|
||||
# 递归计算节点状态
|
||||
def calc_status(node):
|
||||
if "children" in node:
|
||||
for c in node["children"]:
|
||||
calc_status(c)
|
||||
statuses = [c["status"] for c in node["children"]]
|
||||
if "fail" in statuses: node["status"] = "fail"
|
||||
elif "warn" in statuses: node["status"] = "warn"
|
||||
else: node["status"] = "ok"
|
||||
calc_status(feature_tree)
|
||||
|
||||
# ── Tab 2: 数据实体表 ──
|
||||
entities = []
|
||||
for tname, cnt in sorted(db_stats.items()):
|
||||
readers = flows["db_read"].get(tname, [])
|
||||
writers = flows["db_write"].get(tname, [])
|
||||
has_input = len(writers) > 0
|
||||
has_output = len(readers) > 0
|
||||
# 排除系统表
|
||||
is_system = tname.startswith("sqlite_") or tname.startswith("_")
|
||||
orphan = (not has_input or not has_output) and not is_system and cnt > 0
|
||||
entities.append({
|
||||
"name": tname,
|
||||
"rows": cnt,
|
||||
"readers": readers[:10],
|
||||
"writers": writers[:10],
|
||||
"has_input": has_input,
|
||||
"has_output": has_output,
|
||||
"orphan": orphan,
|
||||
"warn": orphan or (cnt == 0 and tname not in ("capital_flow_cache","")),
|
||||
})
|
||||
|
||||
# JSON文件
|
||||
json_entities = []
|
||||
for jf in sorted(WEB_DATA.glob("*.json")):
|
||||
if jf.name == "stocks": continue
|
||||
if jf.stem.startswith("temp_"): continue
|
||||
readers = flows["json_read"].get(jf.name, [])
|
||||
size = jf.stat().st_size / 1024
|
||||
json_entities.append({
|
||||
"name": jf.name,
|
||||
"size_kb": round(size, 1),
|
||||
"readers": readers[:10],
|
||||
"writers": [], # 难以精确追踪
|
||||
"last_modified": datetime.fromtimestamp(jf.stat().st_mtime).strftime("%m-%d %H:%M"),
|
||||
"warn": len(readers) == 0 and jf.name not in ("portfolio.json", "market.json"),
|
||||
})
|
||||
|
||||
# ── Tab 3: 流程/cron映射 ──
|
||||
pipelines = []
|
||||
for j in sorted(cron_jobs, key=lambda x: x.get("name","")):
|
||||
if not j.get("enabled", True):
|
||||
continue
|
||||
name = j.get("name", "?")
|
||||
script = j.get("script", "")
|
||||
status = j.get("last_status", "unknown")
|
||||
last_run = str(j.get("last_run_at", ""))[:19]
|
||||
schedule = j.get("schedule", {}).get("display", str(j.get("schedule","")))
|
||||
no_agent = j.get("no_agent", False)
|
||||
pipelines.append({
|
||||
"name": name,
|
||||
"type": "no_agent" if no_agent else "LLM",
|
||||
"script": script,
|
||||
"schedule": schedule,
|
||||
"status": status,
|
||||
"last_run": last_run,
|
||||
})
|
||||
|
||||
# ── 写JSON ──
|
||||
report = {
|
||||
"generated_at": now.strftime("%Y-%m-%d %H:%M:%S"),
|
||||
"feature_tree": feature_tree,
|
||||
"entities": entities,
|
||||
"json_files": json_entities,
|
||||
"pipelines": pipelines,
|
||||
}
|
||||
out_path = WEB_DATA / "mofin_health.json"
|
||||
with open(out_path, "w") as f:
|
||||
json.dump(report, f, ensure_ascii=False, indent=2)
|
||||
print(f"[SILENT] mofin_health.json written ({len(entities)} entities, {len(pipelines)} pipelines)")
|
||||
|
||||
if __name__ == "__main__":
|
||||
build_report()
|
||||
@@ -0,0 +1,26 @@
|
||||
#!/usr/bin/env python3
|
||||
"""sync_db_to_json.py — 将DB数据同步回JSON文件（反向同步）
|
||||
|
||||
DB是权威源，JSON文件是只读的兼容层。
|
||||
每15分钟将DB中的决策数据写到decisions.json。
|
||||
"""
|
||||
import json, sys
|
||||
from pathlib import Path
|
||||
sys.path.insert(0, str(Path(__file__).parent.parent))
|
||||
sys.path.insert(0, "/home/hmo/MoFin")
|
||||
|
||||
DECISIONS_PATH = "/home/hmo/web-dashboard/data/decisions.json"
|
||||
|
||||
try:
|
||||
from mo_data import read_decisions, read_portfolio, read_watchlist
|
||||
decisions = read_decisions()
|
||||
portfolio = read_portfolio()
|
||||
watchlist = read_watchlist()
|
||||
|
||||
with open(DECISIONS_PATH, "w") as f:
|
||||
json.dump(decisions, f, ensure_ascii=False, indent=2)
|
||||
|
||||
cnt = len(decisions.get("decisions", []))
|
||||
print(f"[SILENT] decisions.json updated: {cnt} entries from DB")
|
||||
except Exception as e:
|
||||
print(f"[SILENT] sync failed: {e}")
|
||||
Reference in New Issue
Block a user