#!/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") STATIC_DIR = Path("/home/hmo/web-dashboard/static") 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", ] # 数据实体作用说明 TABLES_DESC = { "holdings": "当前持仓（权威源）", "holding_strategies": "每只股票的完整策略参数", "portfolio_summary": "总资产/现金/仓位汇总", "portfolio_state": "组合状态快照（只读派生）", "strategy_evaluations": "策略重评历史记录", "strategy_feedback": "策略效果反馈", "watchlist_stocks": "自选股列表", "candidates": "潜力股候选池（小果扫描产出）", "live_prices": "所有持仓+自选最新实时价", "price_events": "价格区间突破事件日志", "market_snapshots": "大盘指数快照（每10分）", "sector_snapshots": "行业板块数据", "sector_signals": "行业信号（趋势检测产出）", "signal_news": "信号相关新闻", "macro_raw_news": "宏观新闻原始数据", "macro_context_log": "宏观上下文（大盘偏向/指数）", "stocks": "全量股票代码", "stock_daily": "日线行情", "stock_weekly": "周线行情", "stock_monthly": "月线行情", "stock_fundamentals": "基本面数据(PE/PB)", "stock_sectors": "股票行业映射", "capital_flow_cache": "资金流缓存", "xiaoguo_scan_tracker": "小果扫描跟踪", "advice_timeline": "建议执行时间线", "accuracy_stats": "建议准确率统计", "todos": "自愈任务队列", "health_check_log": "健康检查日志", "cash_log": "资金变动记录", "mtf_cache": "多周期均线缓存", "state_meta": "系统状态元数据", } JSON_DESC = { "decisions.json": "策略决策（DB→JSON同步，兼容层）", "portfolio.json": "持仓汇总（兼容层）", "market.json": "市场概况数据", "xiaoguo_insights.json": "小果分析洞察", "candidate_pool.json": "潜力股候选池完整数据", "zone_breach.json": "价格区间突破状态", "strategy_staleness_report.json": "策略过期报告", "alerts.json": "告警列表", "macro_risk_state.json": "宏观风险状态（采集器写入）", "capital_flow_cache.json": "资金流缓存", "multi_tf_cache.json": "多周期均线缓存", "macro_context.json": "宏观上下文JSON（旧兼容层）", "system_inventory.json": "全量系统清单", "mofin_health.json": "健康监控数据", } now = datetime.now() def load_cron_jobs(): jobs = [] seen = set() for jf in CRON_FILES: profile_tag = "position-analyst" if "position-analyst" in str(jf) else "default" try: for j in json.load(open(jf)).get("jobs", []): jid = j.get("id", "") if jid in seen: continue seen.add(jid) j["profile"] = profile_tag 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 match_cron(cron_jobs, name_keywords): """匹配cron任务列表，返回匹配的cron信息列表（空格归一化后匹配）""" matches = [] for j in cron_jobs: jname = j.get("name", "").replace(" ", "").replace("\u00a0", "") # 去空格再比 if isinstance(name_keywords, str): if name_keywords.replace(" ", "") in jname: matches.append(j) elif isinstance(name_keywords, (list, tuple)): clean_kws = [k.replace(" ", "").replace("\u00a0", "") for k in name_keywords] if any(kw in jname for kw in clean_kws): matches.append(j) elif callable(name_keywords): if name_keywords(j): matches.append(j) # 去重（相同name只保留一条） seen = set() deduped = [] for j in matches: n = j.get("name", "") if n not in seen: seen.add(n) deduped.append(j) return deduped def build_feature_tree(cron_jobs, db_stats): # 硬编码分类规则：标签→匹配关键词 rules = { "市场快照": ["市场数据采集"], "宏观新闻": ["宏观采集"], "价格监控": ["价格监控"], "小果扫描": ["小果独立扫描"], "资金流采集": ["资金流"], "宏观上下文刷新": ["宏观上下文刷新"], "策略重评": ["策略重评"], "持仓自选新鲜度检查": ["策略时效性检查"], "自选买入区提醒": ["自选买入区提醒"], "策略评估": ["策略评估"], "分支自成长": ["分支自成长"], "元自成长": ["元自成长"], "MoFin盘前中监控": ["MoFin盘前中监控"], "MoFin午后监控": ["MoFin午后监控"], "cron报告推XMPP": ["cron报告推XMPP"], "开盘简报": ["开盘简报"], "收盘简报": ["收盘简报"], "市场精选推荐": ["市场精选推荐"], "小果情感分析": ["小果情感分析"], "系统全局审计": ["系统全局审计"], "全局cron健康监控": ["全局cron健康监控"], "重评管道审计": ["重评管道审计"], "健康监控数据采集": ["健康监控数据采集"], "持仓基本面复查": ["分析师-持仓复查"], "策略复盘": ["策略复盘"], "宏观风险扫描": ["宏观风险扫描"], "宏观风险信号消费": ["宏观风险信号消费"], "跨市场背离检测": ["跨市场背离检测"], "自愈执行器": ["自愈执行器"], "策略质量门禁": ["策略质量门禁"], "自选自动清理": ["自选自动清理"], "建议对账": ["建议对账"], "宏观新闻采集": ["宏观新闻采集"], "数据治理": ["数据治理"], "盘前热点扫描": ["盘前热点扫描"], "数据同步": ["数据同步"], "小果市场筛选": ["小果市场筛选"], "芯碁微装": ["芯碁微装"], "宏观新闻采集-周末": ["宏观新闻采集-周末"], "硬编码扫描": ["硬编码扫描"], "系统体检": ["系统体检"], "盘中自检": ["盘中自检"], "记忆守卫": ["记忆守卫"], "数据治理": ["数据治理"], "自选股自动重评": ["自选股自动重评"], "state.db真空整理": ["真空整理"], "300308": ["300308"], "多周期缓存": ["多周期缓存"], "元自成长": ["元自成长"], } # 自动归类：未被任何规则匹配的cron按名称关键词归入类别 # 关键词必须够精确，避免误归类 AUTO_CATEGORIES = [ ("数据采集", ["市场数据", "宏观采集", "新闻采集", "价格监控", "资金流采集", "小果独立扫描", "上下文刷新"]), ("策略分析", ["策略评估", "策略时效性", "重评", "买入区提醒", "自成长", "策略复盘", "分支"]), ("推荐推送", ["简报", "推送", "推荐", "XMPP", "开盘", "收盘"]), ("风险监控", ["宏观风险", "背离检测", "信号消费"]), ("自检/审计", ["系统全局审计", "健康监控", "管道审计", "系统体检", "盘中自检", "记忆守卫", "硬编码扫描", "治理"]), ("执行/修复", ["自愈执行", "门禁", "清理", "对账", "TODO"]), ("持仓监控", ["300308", "芯碁微装", "多周期缓存", "自选股自动重评"]), ("系统服务", ["真空整理"]), ] matched_names = set() # 记录已匹配的cron name def attach_pipes(node, parent_cat=None): nonlocal matched_names label = node.get("label", "") keywords = rules.get(label) pipes = [] if keywords: matched = match_cron(cron_jobs, keywords) for j in matched: n = j.get("name", "") matched_names.add(n) pipes = [{ "name": j.get("name", ""), "script": j.get("script", ""), "schedule": j.get("schedule", {}).get("display", str(j.get("schedule", ""))), "status": j.get("last_status", "unknown"), "last_run": (j.get("last_run_at", "") or "")[:16] if j.get("last_run_at") else "", "type": "no_agent" if j.get("no_agent") else "LLM", "profile": j.get("profile", "?"), } for j in matched] if pipes: node["pipes"] = pipes if node.get("children"): for c in node["children"]: attach_pipes(c, parent_cat or label) def make_cron_node(j): return { "label": f"{j.get('name','?')} ({j.get('script','LLM')})", "status": j.get("last_status", "unknown"), "pipes": [{ "name": j.get("name", ""), "script": j.get("script", ""), "schedule": j.get("schedule", {}).get("display", str(j.get("schedule", ""))), "status": j.get("last_status", "unknown"), "last_run": (j.get("last_run_at", "") or "")[:16] if j.get("last_run_at") else "", "type": "no_agent" if j.get("no_agent") else "LLM", "profile": j.get("profile", "?"), }] } tree = { "label": "MoFin 系统", "status": "ok", "children": [ {"label": "数据采集", "status": "ok", "children": [ {"label": "市场快照", "status": "ok"}, {"label": "宏观新闻", "status": "ok"}, {"label": "价格监控", "status": "ok"}, {"label": "小果扫描", "status": "ok"}, {"label": "资金流采集", "status": "ok"}, {"label": "宏观上下文刷新", "status": "ok"}, ]}, {"label": "策略分析", "status": "ok", "children": [ {"label": "策略重评", "status": "ok"}, {"label": "持仓自选新鲜度检查", "status": "ok"}, {"label": "自选买入区提醒", "status": "ok"}, {"label": "策略评估", "status": "ok"}, {"label": "分支自成长", "status": "ok"}, {"label": "元自成长", "status": "ok"}, ]}, {"label": "推荐推送", "status": "ok", "children": [ {"label": "MoFin盘前中监控", "status": "ok"}, {"label": "MoFin午后监控", "status": "ok"}, {"label": "cron报告推XMPP", "status": "ok"}, {"label": "开盘简报", "status": "ok"}, {"label": "收盘简报", "status": "ok"}, {"label": "市场精选推荐", "status": "ok"}, ]}, {"label": "风险监控", "status": "ok", "children": [ {"label": "宏观风险扫描", "status": "ok"}, {"label": "宏观风险信号消费", "status": "ok"}, {"label": "跨市场背离检测", "status": "ok"}, ]}, {"label": "自检/审计", "status": "ok", "children": [ {"label": "系统全局审计", "status": "ok"}, {"label": "全局cron健康监控", "status": "ok"}, {"label": "重评管道审计", "status": "ok"}, {"label": "健康监控数据采集", "status": "ok"}, ]}, {"label": "执行/修复", "status": "ok", "children": [ {"label": "自愈执行器", "status": "ok"}, {"label": "策略质量门禁", "status": "ok"}, {"label": "自选自动清理", "status": "ok"}, {"label": "建议对账", "status": "ok"}, ]}, {"label": "持仓复查", "status": "ok", "children": [ {"label": "持仓基本面复查", "status": "ok"}, {"label": "策略复盘", "status": "ok"}, ]}, {"label": "信号消费", "status": "ok", "children": [ {"label": "小果情感分析", "status": "ok"}, {"label": "宏观风险信号消费-盘中", "status": "ok"}, ]}, ], } attach_pipes(tree) # 收集所有未被任何规则匹配的cron，按名称自动归入类别 unmatched = [j for j in cron_jobs if j.get("name", "") not in matched_names] # 按自动归类分组 cat_map = {} for j in unmatched: name = j.get("name", "") assigned = False for cat_name, keywords in AUTO_CATEGORIES: if any(kw in name for kw in keywords): cat_map.setdefault(cat_name, []).append(j) assigned = True break if not assigned: cat_map.setdefault("未分类", []).append(j) # 将自动归类的cron追加到已有分类或创建新分类 for cat_name, jobs in sorted(cat_map.items()): # 如果该分类已存在于树中，追加到其children found = None for child in tree["children"]: if child["label"] == cat_name: found = child break if found: existing_labels = {c["label"] for c in found.get("children", [])} for j in jobs: lbl = j.get("name", "?") if lbl not in existing_labels: found["children"].append(make_cron_node(j)) existing_labels.add(lbl) else: tree["children"].append({ "label": cat_name, "status": "ok", "children": [make_cron_node(j) for j in jobs], }) return tree def build_report(): cron_jobs = load_cron_jobs() db_stats = get_db_stats() flows = scan_data_flows() script_health = check_scripts() # ── 功能树（只显示知微的cron）── zhiwei_crons = [j for j in cron_jobs if j.get("profile") == "position-analyst" or j.get("name") in [ "cron-推XMPP中继", "数据同步-dashboard", "记忆守卫-每日", "市场数据采集" ]] feature_tree = build_feature_tree(zhiwei_crons, db_stats) # 递归计算节点状态 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, "desc": TABLES_DESC.get(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, "desc": JSON_DESC.get(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, "profile": j.get("profile", "?"), }) # ── 写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) # 也写到static目录供dashboard直接serve with open(STATIC_DIR / "mofin_health.json", "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()