From bcf922fb6a6e6e28ef71d618b98d9fad97014de5 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E7=9F=A5=E5=BE=AE?= Date: Wed, 8 Jul 2026 11:44:05 +0800 Subject: [PATCH] =?UTF-8?q?feat:=20=E5=81=A5=E5=BA=B7=E7=9B=91=E6=8E=A7?= =?UTF-8?q?=E4=B8=89Tab=20+=20DB=E2=86=92JSON=E5=90=8C=E6=AD=A5=20+=20deci?= =?UTF-8?q?sions.json=2019=E6=9D=A1?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit --- scripts/mofin_health.py | 225 +++++++++++++++++++++++++++++++++++++ scripts/sync_db_to_json.py | 26 +++++ 2 files changed, 251 insertions(+) create mode 100644 scripts/mofin_health.py create mode 100644 scripts/sync_db_to_json.py diff --git a/scripts/mofin_health.py b/scripts/mofin_health.py new file mode 100644 index 00000000..54a1f648 --- /dev/null +++ b/scripts/mofin_health.py @@ -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() diff --git a/scripts/sync_db_to_json.py b/scripts/sync_db_to_json.py new file mode 100644 index 00000000..43ab5933 --- /dev/null +++ b/scripts/sync_db_to_json.py @@ -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}")