diff --git a/deploy/profile-scripts/functional_health_check.py b/deploy/profile-scripts/functional_health_check.py
new file mode 100644
index 00000000..e47315ed
--- /dev/null
+++ b/deploy/profile-scripts/functional_health_check.py
@@ -0,0 +1,216 @@
+#!/usr/bin/env python3
+"""functional_health_check.py — L1 功能健康检查
+
+判据不是"进程活着"，而是"功能是否达成"：每个核心模块检查其
+**输出物的新鲜度和有效性**。输出物不新鲜 = 功能未达成 = 异常。
+
+频率：交易时段每 15 分钟（cron），输出 gateway/logs/functional_health.json。
+责任矩阵 L1。修复由 L3 self_repair 读取本报告执行。
+"""
+import os, sys, json, sqlite3, subprocess
+from datetime import datetime, timedelta
+
+sys.path.insert(0, '/home/hmo/MoFin')
+
+DB = '/home/hmo/MoFin/data/mofin.db'
+OUT = '/home/hmo/MoFin/gateway/logs/functional_health.json'
+
+# ── 功能注册表：模块 → 功能判据 ──────────────────────────────
+# type:
+# db_freshness: DB 表 MAX(col) 距今不超过 max_age（trading=交易时段才检查, daily=每日, always=总是）
+# file_freshness: 文件 mtime 距今不超过 max_age
+# bot_activity: XMPP bot 活动检查（journal）
+# agent_log: gateway agent.log 最近有成功 LLM 调用
+# soft=True: 无输出可能属正常（如区间未触发），降级为 warn 而非 fail
+REGISTRY = [
+ {"module": "price_monitor", "function": "实时价格写入DB(live_prices)",
+ "check": {"type": "db_freshness", "table": "live_prices", "col": "updated_at",
+ "max_age_min": 6, "when": "trading"},
+ "repair": {"action": "rerun_script", "script": "price_monitor.py"}},
+ {"module": "market_watch", "function": "市场快照采集(market_snapshots)",
+ "check": {"type": "db_freshness", "table": "market_snapshots", "col": "created_at",
+ "max_age_min": 40, "when": "trading"},
+ "repair": {"action": "rerun_script", "script": "market_watch.py"}},
+ {"module": "mtf_cache", "function": "多周期均线缓存刷新(mtf_cache)",
+ "check": {"type": "db_freshness", "table": "mtf_cache", "col": "updated_at",
+ "max_age_min": 75, "when": "trading"},
+ "repair": {"action": "rerun_script", "script": "refresh_mtf_cache.py"}},
+ {"module": "macro_context", "function": "宏观上下文刷新(macro_context_log)",
+ "check": {"type": "db_freshness", "table": "macro_context_log", "col": "created_at",
+ "max_age_min": 45, "when": "trading"},
+ "repair": {"action": "rerun_script", "script": "refresh_macro_context.py"}},
+ {"module": "health_collector", "function": "健康数据采集(mofin_health.json)",
+ "check": {"type": "file_freshness", "path": "/home/hmo/web-dashboard/static/mofin_health.json",
+ "max_age_min": 20, "when": "trading"},
+ "repair": {"action": "rerun_script", "script": "mofin_health.py"}},
+ {"module": "premarket", "function": "盘前全量重评(premarket summary)",
+ "check": {"type": "file_freshness", "path": "/tmp/mofin_premarket/summary.json",
+ "max_age_h": 26, "when": "daily"},
+ "repair": {"action": "rerun_script", "script": "premarket_full_review.py"}},
+ {"module": "gateway_llm", "function": "LLM调用链可用(gateway agent.log)",
+ "check": {"type": "agent_log", "max_age_min": 30, "when": "always"},
+ "repair": {"action": "llm_diagnose"}},
+ {"module": "xmpp_bot", "function": "XMPP消息收发(bot journal)",
+ "check": {"type": "bot_activity", "when": "always"},
+ "repair": {"action": "llm_diagnose"}},
+ {"module": "cron_engine", "function": "cron调度引擎本身(有job在最近10min运行)",
+ "check": {"type": "cron_engine", "max_age_min": 12, "when": "trading"},
+ "repair": {"action": "llm_diagnose"}},
+]
+
+
+def is_trading_now(now):
+ return now.weekday() < 5 and 9 <= now.hour <= 16
+
+
+def check_db_freshness(conn, chk, now):
+ try:
+ row = conn.execute(f"SELECT MAX({chk['col']}) FROM {chk['table']}").fetchone()
+ if not row or not row[0]:
+ return "fail", f"表 {chk['table']} 无数据"
+ last = datetime.fromisoformat(str(row[0]).replace("Z", ""))
+ age_min = (now - last).total_seconds() / 60
+ limit = chk.get("max_age_min", chk.get("max_age_h", 24) * 60)
+ if age_min > limit:
+ return ("warn" if chk.get("soft") else "fail"), \
+ f"{chk['table']} 最新记录 {last.strftime('%m-%d %H:%M')}（{age_min/60:.1f}h前，阈值{limit}min）"
+ return "ok", f"{age_min:.0f}min前"
+ except Exception as e:
+ return "fail", f"查询失败: {str(e)[:80]}"
+
+
+def check_file_freshness(chk, now):
+ p = chk["path"]
+ if not os.path.exists(p):
+ return "fail", f"文件不存在: {p}"
+ age_min = (now.timestamp() - os.path.getmtime(p)) / 60
+ limit = chk.get("max_age_min", chk.get("max_age_h", 24) * 60)
+ if age_min > limit:
+ return ("warn" if chk.get("soft") else "fail"), \
+ f"{os.path.basename(p)} {age_min/60:.1f}h 未更新（阈值{limit}min）"
+ return "ok", f"{age_min:.0f}min前"
+
+
+def check_agent_log(chk, now):
+ try:
+ sys.path.insert(0, '/home/hmo/MoFin')
+ from xmpp_logger import _scan_agent_log
+ r = _scan_agent_log(now.timestamp(), "zhiwei")
+ if r["status"] == "ok":
+ age = r.get("age_sec", -1)
+ if age > chk["max_age_min"] * 60:
+ return "warn", f"最近成功LLM调用在 {age//60}min 前（无新流量，可能正常）"
+ return "ok", f"latency={r.get('latency')}"
+ return "fail", f"agent.log 最近调用失败: {r.get('error','?')[:100]}"
+ except Exception as e:
+ return "fail", f"检查失败: {e}"
+
+
+def check_bot_activity(chk, now):
+ try:
+ # 第一判据：服务当前是否 active（权威"现在在不在跑"）
+ r2 = subprocess.run(["systemctl", "is-active", "xmpp-zhiwei"],
+ capture_output=True, text=True, timeout=5)
+ if r2.stdout.strip() != "active":
+ return "fail", "bot 服务非 active（已停止）"
+ # 第二判据：最近 journal 是否处于断线循环
+ r = subprocess.run(
+ ["journalctl", "-u", "xmpp-zhiwei", "--no-pager", "--since", "30 min ago", "-o", "cat"],
+ capture_output=True, timeout=8, text=True)
+ lines = r.stdout
+ if "连接超时" in lines and "就绪" not in lines:
+ return "fail", "bot 处于断线重连循环"
+ if "XMPP 就绪" in lines or "已发送" in lines or "收到" in lines:
+ return "ok", "bot 活动正常"
+ return "ok", "bot 在线空闲（无新消息）"
+ except Exception as e:
+ return "fail", f"检查失败: {e}"
+
+
+def check_cron_engine(chk, now):
+ """cron 引擎：最近 max_age_min 内是否有任何 job 运行过"""
+ try:
+ d = json.load(open('/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json'))
+ jobs = d if isinstance(d, list) else d.get('jobs', [])
+ latest = None
+ for j in jobs:
+ lr = j.get('last_run_at')
+ if lr:
+ try:
+ t = datetime.fromisoformat(lr.replace('Z', '+00:00')).replace(tzinfo=None)
+ if latest is None or t > latest:
+ latest = t
+ except Exception:
+ pass
+ if not latest:
+ return "fail", "所有 job 均无运行记录"
+ age_min = (now - latest).total_seconds() / 60
+ if age_min > chk["max_age_min"]:
+ return "fail", f"最近 job 运行在 {age_min:.0f}min 前，调度引擎疑似停摆"
+ return "ok", f"最近 job 运行于 {age_min:.0f}min 前"
+ except Exception as e:
+ return "fail", f"检查失败: {e}"
+
+
+def main():
+ now = datetime.now()
+ trading = is_trading_now(now)
+ results = []
+ conn = sqlite3.connect(DB, timeout=10)
+
+ for item in REGISTRY:
+ chk = item["check"]
+ when = chk.get("when", "always")
+ if when == "trading" and not trading:
+ results.append({"module": item["module"], "function": item["function"],
+ "status": "skip", "reason": "非交易时段"})
+ continue
+ if when == "daily":
+ # 每日类：只 fail 不 warn，且非交易日放宽到 72h
+ if now.weekday() >= 5:
+ chk = dict(chk)
+ chk["max_age_h"] = 72
+
+ t = chk["type"]
+ if t == "db_freshness":
+ status, reason = check_db_freshness(conn, chk, now)
+ elif t == "file_freshness":
+ status, reason = check_file_freshness(chk, now)
+ elif t == "agent_log":
+ status, reason = check_agent_log(chk, now)
+ elif t == "bot_activity":
+ status, reason = check_bot_activity(chk, now)
+ elif t == "cron_engine":
+ status, reason = check_cron_engine(chk, now)
+ else:
+ status, reason = "fail", f"未知检查类型 {t}"
+
+ results.append({"module": item["module"], "function": item["function"],
+ "status": status, "reason": reason,
+ "repair": item.get("repair")})
+
+ conn.close()
+
+ fails = [r for r in results if r["status"] == "fail"]
+ warns = [r for r in results if r["status"] == "warn"]
+ report = {
+ "generated_at": now.strftime("%Y-%m-%d %H:%M:%S"),
+ "trading_hours": trading,
+ "summary": {"total": len(results), "ok": sum(1 for r in results if r["status"] == "ok"),
+ "warn": len(warns), "fail": len(fails),
+ "skip": sum(1 for r in results if r["status"] == "skip")},
+ "status": "fail" if fails else ("warn" if warns else "ok"),
+ "checks": results,
+ }
+ os.makedirs(os.path.dirname(OUT), exist_ok=True)
+ with open(OUT, "w", encoding="utf-8") as f:
+ json.dump(report, f, ensure_ascii=False, indent=2)
+
+ print(f"功能健康: {report['summary']} status={report['status']}")
+ for r in results:
+ if r["status"] in ("fail", "warn"):
+ print(f" {r['status'].upper()} {r['module']}: {r['reason']}")
+
+
+if __name__ == "__main__":
+ main()
\ No newline at end of file
diff --git a/deploy/profile-scripts/meta_watchdog.py b/deploy/profile-scripts/meta_watchdog.py
new file mode 100644
index 00000000..2e0c24de
--- /dev/null
+++ b/deploy/profile-scripts/meta_watchdog.py
@@ -0,0 +1,123 @@
+#!/usr/bin/env python3
+"""meta_watchdog.py — L4 自检系统的自检（看门狗的看门狗）
+
+检查 L0-L3 各自检组件本身是否在正常运转：
+- L0 agents_health_check: last_health_check.json 是否 <10min
+- L1 functional_health_check: functional_health.json 是否 <20min（交易时段）
+- L2 system_hygiene_audit: hygiene_report.json 是否 <26h（每日）
+- L3 self_repair: repair_state.json 存在性 + cron 是否注册
+- mofin_health 采集: mofin_health.json 是否 <20min（交易时段）
+- XMPP 桥: :5805 是否可发（self_repair 的报备通道）
+
+任何一层死了 → 推 XMPP 点名（这是最后的兜底，必须直达用户）。
+频率：每小时（cron）。输出 gateway/logs/meta_watchdog.json。
+"""
+import os, sys, json, subprocess
+from datetime import datetime
+
+OUT = '/home/hmo/MoFin/gateway/logs/meta_watchdog.json'
+
+LAYERS = [
+ {"layer": "L0 agents_health_check", "file": "/home/hmo/MoFin/gateway/temp/last_health_check.json",
+ "max_age_min": 10, "when": "always",
+ "repair": "crontab */5 agents_health_check.py 停摆，检查系统 crontab"},
+ {"layer": "L1 functional_health", "file": "/home/hmo/MoFin/gateway/logs/functional_health.json",
+ "max_age_min": 25, "when": "trading",
+ "repair": "L1 cron 停摆，检查 hermes cron 引擎"},
+ {"layer": "L2 hygiene_audit", "file": "/home/hmo/MoFin/gateway/logs/hygiene_report.json",
+ "max_age_min": 26 * 60, "when": "always",
+ "repair": "L2 每日审计未跑，检查 hermes cron"},
+ {"layer": "L1.5 mofin_health采集", "file": "/home/hmo/web-dashboard/static/mofin_health.json",
+ "max_age_min": 25, "when": "trading",
+ "repair": "mofin_health.py 采集停摆"},
+ {"layer": "L3 self_repair", "file": "/home/hmo/MoFin/gateway/logs/repair_log.jsonl",
+ "max_age_min": None, "when": "meta",
+ "repair": "self_repair cron 未注册"},
+]
+
+
+def is_trading(now):
+ return now.weekday() < 5 and 9 <= now.hour <= 16
+
+
+def main():
+ now = datetime.now()
+ trading = is_trading(now)
+ results = []
+
+ for L in LAYERS:
+ if L["when"] == "trading" and not trading:
+ results.append({"layer": L["layer"], "status": "skip", "reason": "非交易时段"})
+ continue
+ if L["when"] == "meta":
+ # 检查 self_repair 是否注册在 cron
+ try:
+ d = json.load(open('/home/hmo/.hermes/profiles/position-analyst/cron/jobs.json'))
+ jobs = d if isinstance(d, list) else d.get('jobs', [])
+ registered = any(j.get('script') == 'self_repair.py' and j.get('enabled', True) for j in jobs)
+ results.append({"layer": L["layer"],
+ "status": "ok" if registered else "fail",
+ "reason": "已注册" if registered else "未在 cron 注册"})
+ except Exception as e:
+ results.append({"layer": L["layer"], "status": "fail", "reason": str(e)[:60]})
+ continue
+
+ f = L["file"]
+ if not os.path.exists(f):
+ results.append({"layer": L["layer"], "status": "fail",
+ "reason": f"输出物不存在", "repair": L["repair"]})
+ continue
+ age_min = (now.timestamp() - os.path.getmtime(f)) / 60
+ if L["max_age_min"] and age_min > L["max_age_min"]:
+ results.append({"layer": L["layer"], "status": "fail",
+ "reason": f"输出物 {age_min/60:.1f}h 未更新（阈值 {L['max_age_min']}min）",
+ "repair": L["repair"]})
+ else:
+ results.append({"layer": L["layer"], "status": "ok",
+ "reason": f"{age_min:.0f}min 前"})
+
+ # XMPP 桥（报备通道）：只收 POST，GET 会 501，但任何 HTTP 响应都说明进程活着
+ try:
+ import urllib.request
+ urllib.request.urlopen('http://127.0.0.1:5805/', timeout=3)
+ results.append({"layer": "XMPP桥 :5805", "status": "ok", "reason": "可达"})
+ except urllib.error.HTTPError as e:
+ results.append({"layer": "XMPP桥 :5805", "status": "ok", "reason": f"可达(HTTP {e.code})"})
+ except Exception:
+ results.append({"layer": "XMPP桥 :5805", "status": "fail",
+ "reason": "不可达", "repair": "重启 xmpp-zhiwei"})
+
+ fails = [r for r in results if r["status"] == "fail"]
+ report = {
+ "generated_at": now.strftime("%Y-%m-%d %H:%M:%S"),
+ "status": "fail" if fails else "ok",
+ "layers": results,
+ }
+ os.makedirs(os.path.dirname(OUT), exist_ok=True)
+ with open(OUT, 'w', encoding='utf-8') as f:
+ json.dump(report, f, ensure_ascii=False, indent=2)
+
+ print(f"meta_watchdog: {report['status']}")
+ for r in results:
+ icon = {"ok": "✅", "fail": "❌", "skip": "⏭"}[r["status"]]
+ print(f" {icon} {r['layer']}: {r['reason']}")
+
+ if fails:
+ try:
+ import urllib.request
+ lines = [f"🚨 自检系统自检（L4兜底）发现 {len(fails)} 层异常："]
+ for r in fails:
+ lines.append(f"❌ {r['layer']}: {r['reason']}")
+ if r.get('repair'):
+ lines.append(f" → 处置建议: {r['repair']}")
+ payload = json.dumps({'to': 'hmo@yoin.fun', 'body': '\n'.join(lines), 'type': 'chat'}).encode()
+ req = urllib.request.Request('http://127.0.0.1:5805/', data=payload,
+ headers={'Content-Type': 'application/json'})
+ urllib.request.urlopen(req, timeout=5)
+ print(' 📨 已推 XMPP（兜底直达）')
+ except Exception as e:
+ print(f' XMPP 失败: {e}')
+
+
+if __name__ == '__main__':
+ main()
\ No newline at end of file
diff --git a/deploy/profile-scripts/mofin_health.py b/deploy/profile-scripts/mofin_health.py
index efdfaf8e..18916fbf 100644
--- a/deploy/profile-scripts/mofin_health.py
+++ b/deploy/profile-scripts/mofin_health.py
@@ -1,976 +1,1018 @@
-#!/usr/bin/env python3
-"""mofin_health.py — MoFin 健康监控数据采集
-
-输出JSON供dashboard展示，三个view：
- tab1: 功能树（逐级展开，每节点绿/黄/红）
- tab2: 数据实体表（输入/输出流分析，孤立表报警）
- tab3: 流程/cron映射（状态正常/异常）
-"""
-import json, os, sys, re
-import sqlite3
-from pathlib import Path
-from datetime import datetime, timezone
-from mofin_db import get_conn
-
-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 = get_conn()
- 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
-
-
-# ── 功能树描述 ──
-NODE_DESC = {
- "数据采集": "从腾讯/东财/小果采集原始行情、新闻、资金流数据",
- "策略分析": "策略评估、新鲜度检查、重评和成长分析",
- "推荐推送": "生成简报、推荐并推送到XMPP",
- "风险监控": "宏观风险信号、跨市场背离检测",
- "自检/审计": "系统健康检查、监控采集、审计",
- "执行/修复": "自愈系统、门禁跟进、清理修复",
- "持仓复查": "持仓基本面复查和策略复盘",
- "信号消费": "消费小果情感分析和宏观风险信号",
- "系统服务": "系统维护（如DB真空整理）",
- "持仓监控": "特定持仓（300308/芯碁微装）盘中监控",
- "市场快照": "每10分钟采集全市场板块和指数快照",
- "宏观新闻": "采集宏观新闻和财经资讯",
- "价格监控": "每2分钟刷新持仓/自选实时价格→写入live_prices",
- "小果扫描": "小果独立扫描潜在机会",
- "资金流采集": "盘中采集板块资金流向",
- "宏观上下文刷新": "刷新大盘指数/市场情绪",
- "策略重评": "价格偏离买入区或策略过期时自动重评",
- "持仓自选新鲜度检查": "检查策略是否过期或价格严重偏离",
- "自选买入区提醒": "自选进入买入区时推送提醒",
- "策略评估": "每日/每周策略效果评估",
- "分支自成长": "策略分支探索和剪枝",
- "元自成长": "系统元层级自我进化",
- "MoFin盘前中监控": "上午盘中实时监控+推送",
- "MoFin午后监控": "下午盘中实时监控+推送",
- "cron报告推XMPP": "cron报告通过XMPP推送到手机",
- "开盘简报": "每日开盘前市场简报",
- "收盘简报": "每日收盘后市场简报",
- "市场精选推荐": "每日全市场潜力股精推",
- "宏观风险扫描": "从新闻中检测系统性风险",
- "宏观风险信号消费": "消费宏观风险信号并生成建议",
- "跨市场背离检测": "检测A股/港股/美股指数背离",
- "系统全局审计": "7维度系统全面审计",
- "全局cron健康监控": "监控所有cron的运行状态",
- "重评管道审计": "审计策略重评管道完整性",
- "健康监控数据采集": "采集健康数据供Dashboard展示",
- "自愈执行器": "每10分钟自动处理TODO列表",
- "策略质量门禁": "新策略必须通过9维验证才能写入",
- "自选自动清理": "开盘前清理过期自选数据",
- "建议对账": "每周对账校验建议准确性",
- "持仓基本面复查": "每周持仓基本面深度复查",
- "策略复盘": "每日策略执行复盘",
- "小果情感分析": "收盘后对持仓/自选做新闻情感分析",
- "宏观风险信号消费-盘中": "盘中消费宏观风险信号",
- "小果市场筛选": "全市场扫描值得关注的板块和个股",
- "芯碁微装": "芯碁微装午后价格监控",
- "300308": "300308午后紧盯+入场信号监控",
- "硬编码扫描": "扫描脚本中的硬编码参数",
- "系统体检": "开盘前系统全面体检",
- "盘中自检": "盘中高频自检",
- "记忆守卫": "每日记忆清理和优化",
- "数据治理": "每周数据清理和归档",
- "自选股自动重评": "周末自动重评自选股策略",
- "多周期缓存": "刷新MA5/MA20/MA60等技术指标缓存",
- "数据同步": "同步数据到Dashboard",
- "盘前热点扫描": "盘前扫描市场热点",
- "宏观新闻采集": "采集宏观新闻",
- "宏观新闻采集-周末": "周末宏观新闻采集",
- "state.db真空整理": "DB真空整理维护",
- "分支剪枝-每日": "修剪已失效的策略分支",
- "自选股自动重评-周末": "周末批量重评自选股策略",
- "系统健康检查-开盘前": "开盘前检查所有核心组件是否正常",
- "多周期缓存刷新-开盘前": "开盘前刷新技术指标缓存",
- "MoFin 系统常规体检-开盘前": "开盘前8:00全面系统体检",
- "开盘前钉对钉验证": "开盘前15项验证(脚本同步/DB完整性/资产公式)",
- "cron-推XMPP中继": "将cron输出通过XMPP中继推送",
- "小果信号消费-盘中": "盘中消费小果扫描信号",
- "硬编码扫描-每日": "扫描脚本中的硬编码参数",
- "盘中自检-高频": "每15分钟盘中自检",
- "数据治理-每周": "每周数据治理",
- "记忆守卫-每日": "每日记忆优化",
- "300308入场信号紧盯": "300308入场信号(13:00-14:00)",
- "300308午后紧盯": "300308午后监控(13:00-15:00含止损)",
- "多周期缓存刷新-盘中": "盘中刷新技术指标缓存",
- "知识萃取-盘后": "盘后从分析报告中萃取可复用知识",
- "区间维护": "每30分钟维护买入区",
- "知微洞察生成": "生成每日市场洞察(15:35)",
- "小果市场筛选-全市场": "小果筛选全市场关注板块",
- "数据同步-dashboard": "同步数据到Dashboard",
- "state.db真空整理-每周": "每周DB真空整理",
- "未分类": "未被规则匹配的cron自动归入此",
-}
-
-# ── 数据流详细描述 ──
-# 每张表说明：存什么 + 谁写入(为什么+写什么) + 谁读取(为什么+读什么) + 综合总结
-FLOW_DETAIL = {
- "signal_news": {
- "summary": "全系统信号/新闻的统一存储表，所有宏观分析、风险扫描、小果分析的输出汇聚地，也是下游消费脚本的输入源。7个写入方汇聚不同来源信号，5个读取方按需消费。",
- "writers": {
- "macro_context_collector": "写入宏观新闻原始数据（标题+摘要+分类），供后续风险扫描消费",
- "xiaoguo_news_processor": "写入小果LLM处理后的新闻情感分析结果",
- "macro_signal_consumer": "写入宏观风险信号判定结果（等级+来源+建议）",
- "divergence_detector": "写入跨市场背离检测信号（A股/港股/美股指数对）",
- "xiaoguo_signal_consumer": "写入小果扫描发现的个股/板块信号",
- "mofin_news": "写入外部财经常规新闻采集结果",
- "xiaoguo_scanner": "写入小果独立扫描的市场机会信号",
- },
- "readers": {
- "macro_signal_consumer": "读取原始宏观新闻和信号，判定风险等级并生成建议",
- "system_audit": "读取信号表行数/更新时间，审计数据管道是否畅通",
- "intraday_health_check": "读取最新信号，检查盘中是否有新的风险信号到达",
- "xiaoguo_signal_consumer": "读取小果相关信号，生成买入/卖出建议",
- "server": "读取信号数据供Web Dashboard展示",
- },
- },
- "holdings": {
- "summary": "当前持仓表，是系统最核心的数据表之一。import_holding_xls从券商文件导入持仓，mofin_db在价格刷新时更新市值。下游脚本读取持仓做策略分析和推送。",
- "writers": {
- "mofin_db": "写入price_monitor刷新后的持仓最新市值（通过write_holdings_batch）",
- "import_holding_xls": "从券商holding.xls导入最新持仓数量/成本/市值",
- },
- "readers": {
- "stale_push_wlin": "读取持仓列表+最新价格，检查是否进入买入区/触发止损",
- "mofin_db": "内部读取（get_price_from_db等函数）",
- "system_audit": "读取持仓总数/品种分布，审计持仓完整性",
- "server": "读取持仓数据供Web Dashboard展示",
- "prepare_report_data": "读取持仓数据用于生成分析报告",
- "mo_data": "通过read_portfolio()读取持仓结构化数据",
- },
- },
- "portfolio_summary": {
- "summary": "组合汇总表（id=1单行），记录总资产=持股市值+可用资金+冻结资金。每笔导入或价格刷新后更新。",
- "writers": {
- "mofin_db": "价格监控刷新总市值后更新total_mv/total_assets",
- "import_holding_xls": "导入持仓后更新cash/frozen/total_assets",
- },
- "readers": {
- "import_holding_xls": "读取当前汇总信息，验证导入后是否正确",
- "mo_data": "通过read_portfolio()读取组合汇总",
- "price_monitor": "读取当前现金/市值，计算总资产变动",
- "prepare_report_data": "读取总资产/现金数据用于报告",
- "server": "读取汇总数据供Dashboard展示",
- },
- },
- "holding_strategies": {
- "summary": "策略数据表，记录每只持仓/自选股的策略配置（买入价/止损/止盈/目标价/分析维度等）。多写入方按各自职责更新不同字段。",
- "writers": {
- "data_governance": "归档过期策略、修复异常策略数据",
- "sync_decisions_to_db": "从JSON同步策略到DB",
- "mofin_db": "策略写入（内部函数）",
- "strategy_review": "策略复盘后更新执行结果和评级",
- },
- "readers": {
- "data_governance": "读取所有活跃策略，检查缺失和异常",
- "per_stock_reassess": "读取个股策略配置，判断是否需要重评",
- "mo_data": "通过read_decisions()读取策略数据",
- "stale_push_wlin": "读取买入区/止损/止盈配置，检查价格触发",
- },
- },
- "live_prices": {
- "summary": "实时价格缓存表，price_monitor每2分钟写入全量持仓/自选价格。所有脚本必须通过mo_data.get_price()读取——先读此表，无数据才调API。单一写入、多方读取。",
- "writers": {
- "mofin_db": "price_monitor调用write_live_prices写入最新价格",
- "mo_data": "get_price()兜底时从API拉取价格后写回此表",
- },
- "readers": {
- "mo_data": "get_price()/get_prices_batch()优先从此表读取价格",
- "mofin_db": "内部读取（get_price_from_db）",
- "system_audit": "读取价格更新时间和数据量",
- "verify_reassess_pipeline": "验证重评管道是否有最新价格",
- },
- },
- "price_events": {
- "summary": "价格触发事件表，价格进入/离开买入区或触发止损止盈时记录事件。用于审计和重评触发。",
- "writers": {
- "mofin_db": "price_monitor检测到价格区间变化时写入事件记录",
- },
- "readers": {
- "mofin_db": "查询历史事件判断是否触发重评",
- },
- },
- "cash_log": {
- "summary": "资金流水表，每次资金变动（入金/出金/冻结/解冻）记录一条日志。审计用。",
- "writers": {
- "mofin_db": "通过write_cash_log记录资金变动",
- "mo_data": "write_cash_log函数入口",
- },
- "readers": {
- "prepare_report_data": "读取现金变动历史用于报告",
- "mofin_db": "内部查询最近流水",
- },
- },
- "market_snapshots": {
- "summary": "市场快照表，market_watch每10分钟采集全市场大盘指数+板块涨跌+上涨下跌家数。下游用于判断市场情绪。",
- "writers": {
- "mofin_db": "market_watch采集后写入快照数据",
- },
- "readers": {
- "market_screener": "读取最新板块快照，判断热点板块",
- "prepare_report_data": "读取市场情绪数据用于报告",
- "mofin_db": "内部查询最新快照",
- "system_audit": "审计数据新鲜度",
- },
- },
- "sector_snapshots": {
- "summary": "板块快照表，market_watch按板块写入涨跌/领涨股/资金流向。market_screener据此判断行业热点。",
- "writers": {
- "mofin_db": "market_watch采集后写入各板块数据",
- },
- "readers": {
- "market_screener": "读取板块涨跌排名，筛选热点行业",
- "strategy_lifecycle": "读取板块数据用于策略生命周期管理",
- "mofin_db": "内部查询",
- "trend_detector": "读取板块趋势数据用于趋势检测",
- },
- },
- "sector_signals": {
- "summary": "板块信号表，多源汇聚的板块级别信号（新闻情感+趋势+资金流向）。用于判断行业轮动。",
- "writers": {
- "mofin_news": "写入新闻分析得出的板块信号",
- "xiaoguo_news_processor": "写入小果LLM分析的板块情感信号",
- "trend_detector": "写入技术面趋势检测到的板块信号",
- },
- "readers": {
- "server": "读取供Dashboard展示",
- "mofin_news": "读取已有信号做增量更新",
- "xiaoguo_news_processor": "读取已有信号避免重复写入",
- "trend_detector": "读取信号辅助趋势判定",
- },
- },
- "macro_context_log": {
- "summary": "宏观上下文日志，refresh_macro_context每30分钟采集大盘指数/市场情绪/资金面数据。下游多个脚本按需读取最新宏观状态。",
- "writers": {
- "refresh_macro_context": "每30分钟采集上证/深证/创业板/恒指等指数+情绪指标",
- },
- "readers": {
- "stale_push_wlin": "读取大盘情绪用于策略推送的宏观背景",
- "divergence_detector": "读取多市场指数数据做背离检测",
- "system_audit": "审计数据采集是否正常",
- "xiaoguo_signal_consumer": "读取宏观情绪辅助信号判定",
- },
- },
- "macro_raw_news": {
- "summary": "宏观新闻原始数据表，macro_context_collector采集的未经处理的财经新闻。供后续清洗和分析。",
- "writers": {
- "macro_context_collector": "从财经网站采集原始新闻标题+URL+摘要",
- },
- "readers": {
- "macro_context_collector": "读取最近新闻hash避免重复采集",
- "system_audit": "审计新闻采集量",
- },
- },
- "accuracy_stats": {
- "summary": "策略准确率统计表，strategy_review复盘后写入各策略的正确/错误/待定计数。",
- "writers": {
- "strategy_review": "策略复盘后更新准确率统计",
- },
- "readers": {
- "mofin_db": "读取统计结果用于报告",
- },
- },
- "advice_timeline": {
- "summary": "建议时间线表，记录每条推送建议的时间/内容/状态。用于审计和对账。",
- "writers": {
- "advice_reconciliation": "每周对账时写入对账结果",
- },
- "readers": {
- "advice_reconciliation": "读取历史建议做对账",
- "mofin_db": "内部查询",
- },
- },
- "candidate_score_history": {
- "summary": "候选股评分历史表，记录每次全市场筛选时对候选股的评分。用于评分变化追踪。",
- "writers": {
- "mofin_db": "market_screener筛选结果写入评分记录",
- },
- "readers": {
- "mofin_db": "查询评分历史供展示",
- },
- },
- "candidates": {
- "summary": "候选股池表，market_screener筛选出的值得关注的个股。包含评分/买入区/止损/目标价。",
- "writers": {
- "mofin_db": "market_screener写入候选股",
- "market_screener": "直接写入候选股列表",
- },
- "readers": {
- "mofin_db": "读取候选股数据供展示和后续处理",
- },
- },
- "capital_flow_cache": {
- "summary": "资金流向缓存表，capital_flow_collector采集的板块资金流入流出数据。",
- "writers": {
- "mofin_db": "写入板块资金流向数据",
- },
- "readers": {
- "mofin_db": "读取缓存数据",
- },
- },
- "health_check_log": {
- "summary": "健康检查日志表，morning_health_check每次运行记录检查结果。用于追踪系统健康历史。",
- "writers": {
- "morning_health_check": "每日开盘前体检后写入检查结果",
- },
- "readers": {
- "morning_health_check": "读取历史检查结果比较变化",
- },
- },
- "mtf_cache": {
- "summary": "多周期技术指标缓存，refresh_mtf_cache计算MA5/MA20/MA60/支撑阻力位等。下游技术分析脚本从缓存读取避免重复计算。",
- "writers": {
- "multi_timeframe": "计算并写入多周期MA/支撑阻力位",
- "mofin_db": "内部写入函数",
- },
- "readers": {
- "multi_timeframe": "读取已有缓存判断是否需要刷新",
- "technical_analysis": "读取MA/支撑阻力位用于技术分析",
- "mofin_db": "内部读取",
- },
- },
- "stock_fundamentals": {
- "summary": "基本面数据表，存储PE/PB/ROE/市值等财务指标。",
- "writers": {
- "mofin_db": "基本面数据采集后写入",
- },
- "readers": {
- "strategy_lifecycle": "读取基本面数据用于策略评估",
- },
- },
- "stock_sectors": {
- "summary": "股票-板块映射表，记录每只股票所属行业板块。多脚本用于行业分类和板块归因。",
- "writers": {
- "mofin_db": "股票行业分类数据写入",
- },
- "readers": {
- "xiaoguo_news_processor": "按行业分类新闻",
- "mofin_news": "按行业归类新闻",
- "mofin_db": "内部查询",
- "strategy_lifecycle": "读取行业信息用于策略决策",
- },
- },
- "stocks": {
- "summary": "全量股票代码表，所有A股/港股基础信息。供各脚本按code查询股票名称/市场。",
- "writers": {
- "mofin_db": "初始化时导入全量股票代码",
- },
- "readers": {
- "mofin_news": "按股票代码查找新闻",
- "xiaoguo_news_processor": "按股票代码过滤新闻",
- "mofin_db": "内部查询",
- "trend_detector": "按股票代码获取数据",
- },
- },
- "strategy_evaluations": {
- "summary": "策略评估结果表，策略评估脚本每次运行记录评估得分/等级/评语。",
- "writers": {
- "mofin_collect": "策略评估前采集数据并写入评估结果",
- },
- "readers": {
- "verify_reassess_pipeline": "读取评估结果验证管道完整性",
- "mofin_db": "内部查询",
- "system_audit": "审计评估是否按时执行",
- },
- },
- "strategy_feedback": {
- "summary": "策略反馈表，记录用户对建议的反馈（采纳/忽略/修改）。用于策略自学习。",
- "writers": {
- "mofin_db": "写入反馈数据",
- "server": "通过Web提交反馈后写入",
- },
- "readers": {
- "mofin_db": "读取反馈用于分析和展示",
- },
- },
- "todos": {
- "summary": "待办事项表，各脚本发现异常时写入TODO，self_todo_executor每10分钟执行修复。异常发现→自动修复的闭环。",
- "writers": {
- "morning_health_check": "体检发现异常写入TODO",
- "intraday_health_check": "盘中自检发现异常写入TODO",
- "strategy-staleness-check": "策略过期检测写入TODO",
- "self_todo_executor": "执行完成后更新TODO状态",
- "preflight_verify": "开盘前验证失败写入TODO",
- },
- "readers": {
- "morning_health_check": "读取待处理的TODO",
- "self_todo_executor": "读取待处理的TODO并执行fix_action",
- "strategy-staleness-check": "读取TODO避免重复写入",
- "intraday_health_check": "读取TODO检查自愈进度",
- },
- },
- "watchlist_stocks": {
- "summary": "自选股表，系统自动维护的观察列表。与持仓表分离，用于跟踪潜在买入机会。",
- "writers": {
- "per_stock_reassess": "策略重评时更新自选状态",
- "mofin_db": "内部写入函数",
- },
- "readers": {
- "per_stock_reassess": "读取自选列表做重评",
- "stock_quote": "读取自选代码拉取行情",
- "mo_alphasift_bridge": "读取自选供Alpha分析",
- "mo_data": "通过read_watchlist()读取自选数据",
- },
- },
- "xiaoguo_scan_tracker": {
- "summary": "小果扫描追踪表，记录每次小果扫描的状态/耗时/结果数量。用于监控小果服务健康。",
- "writers": {
- "xiaoguo_scanner": "每次扫描完成后写入状态和统计",
- },
- "readers": {
- "server": "读取扫描状态供Dashboard展示",
- "xiaoguo_scanner": "读取上次扫描时间判断是否需要全量扫描",
- },
- },
- "state_meta": {
- "summary": "状态元数据表，记录各服务的状态追踪信息（如扫描偏移量/最新处理ID）。",
- "writers": {
- "xiaoguo_scanner": "写入扫描进度偏移量",
- },
- "readers": {
- "xiaoguo_scanner": "读取上次处理位置继续增量处理",
- },
- },
-}
-
-
-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", "")
- # 附加描述（自动带脚本名的节点去掉括号内容匹配）
- desc_key = label.split(" (")[0] if " (" in label else label
- if desc_key in NODE_DESC:
- node["desc"] = NODE_DESC[desc_key]
- 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):
- name = j.get("name", "?")
- desc_key = name.split(" (")[0] if " (" in name else name
- return {
- "label": f"{name} ({j.get('script','LLM')})",
- "desc": NODE_DESC.get(desc_key, ""),
- "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, [])
- # 扫描器漏检的手动补录写入方
- _manual_writers = {
- "candidates": ["mofin_db", "market_screener"],
- "candidate_score_history": ["mofin_db"],
- "strategy_feedback": ["mofin_db", "server"],
- "stock_daily": ["mofin_db"],
- "stock_weekly": ["mofin_db"],
- "stock_monthly": ["mofin_db"],
- }
- _manual_readers = {
- "stock_weekly": ["multi_timeframe"],
- "stock_monthly": ["multi_timeframe"],
- "watchlist_log": ["watchlist_auto_exit", "mofin_db"],
- }
- if not writers and tname in _manual_writers:
- writers = _manual_writers[tname]
- if not readers and tname in _manual_readers:
- readers = _manual_readers[tname]
-
- # 数据流详细描述
- flow_detail = FLOW_DETAIL.get(tname, {})
-
- has_input = len(writers) > 0
- has_output = len(readers) > 0
- # 排除系统表
- is_system = tname.startswith("sqlite_") or tname.startswith("_")
- if is_system:
- continue
- # 数据流状态：healthy / write_only / read_only / orphan
- if has_input and has_output:
- flow_status = "healthy"
- elif has_input and not has_output:
- flow_status = "write_only"
- elif not has_input and has_output:
- flow_status = "read_only"
- else:
- flow_status = "orphan"
- 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": flow_status in ("orphan", "read_only", "write_only"),
- "flow_status": flow_status,
- "warn": flow_status != "healthy",
- "flow_detail": flow_detail,
- })
-
- # JSON文件
- # 已迁移到DB的旧JSON文件：不再报"无读取方"假警报，真实健康信号看DB表新鲜度
- MIGRATED_TO_DB = {
- "multi_tf_cache.json": "mtf_cache",
- "macro_context.json": "macro_context_log",
- "market.json": "market_snapshots",
- "live_prices.json": "live_prices",
- "price_history.json": "price_events",
- "macro_risk_state.json": "macro_context_log",
- }
- 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
- migrated = MIGRATED_TO_DB.get(jf.name)
- desc = JSON_DESC.get(jf.name, "")
- if migrated:
- desc = (desc + " " if desc else "") + f"(已迁移到DB表 {migrated}，此为遗留文件)"
- json_entities.append({
- "name": jf.name,
- "desc": desc,
- "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")
- and not migrated),
- "migrated_to_db": migrated or None,
- })
-
- # ── DB表新鲜度：真实数据管道健康信号（替代对遗留JSON文件的mtime检查）──
- # 注意：活跃数据在 /home/hmo/MoFin/data/mofin.db（live_prices/mtf_cache 今日有写入），
- # 不用 get_conn()（它指向 web-dashboard 的库，那边部分表是旧的）
- db_freshness = []
- FRESHNESS_TABLES = [
- ("mtf_cache", "updated_at", "多周期均线缓存"),
- ("macro_context_log", "created_at", "宏观上下文"),
- ("market_snapshots", "created_at", "市场快照"),
- ("live_prices", "updated_at", "实时价格"),
- ("price_events", "created_at", "价格事件"),
- ]
- try:
- _fc = sqlite3.connect("/home/hmo/MoFin/data/mofin.db", timeout=10)
- for tname, tcol, label in FRESHNESS_TABLES:
- try:
- row = _fc.execute(
- f"SELECT MAX({tcol}) FROM {tname}").fetchone()
- if row and row[0]:
- last_dt = datetime.fromisoformat(str(row[0]).replace("Z", ""))
- age_h = (now - last_dt).total_seconds() / 3600
- db_freshness.append({
- "table": tname, "label": label,
- "last_record": last_dt.strftime("%m-%d %H:%M"),
- "age_hours": round(age_h, 1),
- "warn": age_h > 24,
- })
- else:
- db_freshness.append({"table": tname, "label": label,
- "last_record": None, "age_hours": -1, "warn": True})
- except Exception:
- pass # 表不存在或列名不同，跳过
- _fc.close()
- except Exception:
- pass
-
- # ── 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,
- "db_freshness": db_freshness,
- }
- 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()
+#!/usr/bin/env python3
+"""mofin_health.py — MoFin 健康监控数据采集
+
+输出JSON供dashboard展示，三个view：
+ tab1: 功能树（逐级展开，每节点绿/黄/红）
+ tab2: 数据实体表（输入/输出流分析，孤立表报警）
+ tab3: 流程/cron映射（状态正常/异常）
+"""
+import json, os, sys, re
+import sqlite3
+from pathlib import Path
+from datetime import datetime, timezone
+from mofin_db import get_conn
+
+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 = get_conn()
+ 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
+
+
+# ── 功能树描述 ──
+NODE_DESC = {
+ "数据采集": "从腾讯/东财/小果采集原始行情、新闻、资金流数据",
+ "策略分析": "策略评估、新鲜度检查、重评和成长分析",
+ "推荐推送": "生成简报、推荐并推送到XMPP",
+ "风险监控": "宏观风险信号、跨市场背离检测",
+ "自检/审计": "系统健康检查、监控采集、审计",
+ "执行/修复": "自愈系统、门禁跟进、清理修复",
+ "持仓复查": "持仓基本面复查和策略复盘",
+ "信号消费": "消费小果情感分析和宏观风险信号",
+ "系统服务": "系统维护（如DB真空整理）",
+ "持仓监控": "特定持仓（300308/芯碁微装）盘中监控",
+ "市场快照": "每10分钟采集全市场板块和指数快照",
+ "宏观新闻": "采集宏观新闻和财经资讯",
+ "价格监控": "每2分钟刷新持仓/自选实时价格→写入live_prices",
+ "小果扫描": "小果独立扫描潜在机会",
+ "资金流采集": "盘中采集板块资金流向",
+ "宏观上下文刷新": "刷新大盘指数/市场情绪",
+ "策略重评": "价格偏离买入区或策略过期时自动重评",
+ "持仓自选新鲜度检查": "检查策略是否过期或价格严重偏离",
+ "自选买入区提醒": "自选进入买入区时推送提醒",
+ "策略评估": "每日/每周策略效果评估",
+ "分支自成长": "策略分支探索和剪枝",
+ "元自成长": "系统元层级自我进化",
+ "MoFin盘前中监控": "上午盘中实时监控+推送",
+ "MoFin午后监控": "下午盘中实时监控+推送",
+ "cron报告推XMPP": "cron报告通过XMPP推送到手机",
+ "开盘简报": "每日开盘前市场简报",
+ "收盘简报": "每日收盘后市场简报",
+ "市场精选推荐": "每日全市场潜力股精推",
+ "宏观风险扫描": "从新闻中检测系统性风险",
+ "宏观风险信号消费": "消费宏观风险信号并生成建议",
+ "跨市场背离检测": "检测A股/港股/美股指数背离",
+ "系统全局审计": "7维度系统全面审计",
+ "全局cron健康监控": "监控所有cron的运行状态",
+ "重评管道审计": "审计策略重评管道完整性",
+ "健康监控数据采集": "采集健康数据供Dashboard展示",
+ "自愈执行器": "每10分钟自动处理TODO列表",
+ "策略质量门禁": "新策略必须通过9维验证才能写入",
+ "自选自动清理": "开盘前清理过期自选数据",
+ "建议对账": "每周对账校验建议准确性",
+ "持仓基本面复查": "每周持仓基本面深度复查",
+ "策略复盘": "每日策略执行复盘",
+ "小果情感分析": "收盘后对持仓/自选做新闻情感分析",
+ "宏观风险信号消费-盘中": "盘中消费宏观风险信号",
+ "小果市场筛选": "全市场扫描值得关注的板块和个股",
+ "芯碁微装": "芯碁微装午后价格监控",
+ "300308": "300308午后紧盯+入场信号监控",
+ "硬编码扫描": "扫描脚本中的硬编码参数",
+ "系统体检": "开盘前系统全面体检",
+ "盘中自检": "盘中高频自检",
+ "记忆守卫": "每日记忆清理和优化",
+ "数据治理": "每周数据清理和归档",
+ "自选股自动重评": "周末自动重评自选股策略",
+ "多周期缓存": "刷新MA5/MA20/MA60等技术指标缓存",
+ "数据同步": "同步数据到Dashboard",
+ "盘前热点扫描": "盘前扫描市场热点",
+ "宏观新闻采集": "采集宏观新闻",
+ "宏观新闻采集-周末": "周末宏观新闻采集",
+ "state.db真空整理": "DB真空整理维护",
+ "分支剪枝-每日": "修剪已失效的策略分支",
+ "自选股自动重评-周末": "周末批量重评自选股策略",
+ "系统健康检查-开盘前": "开盘前检查所有核心组件是否正常",
+ "多周期缓存刷新-开盘前": "开盘前刷新技术指标缓存",
+ "MoFin 系统常规体检-开盘前": "开盘前8:00全面系统体检",
+ "开盘前钉对钉验证": "开盘前15项验证(脚本同步/DB完整性/资产公式)",
+ "cron-推XMPP中继": "将cron输出通过XMPP中继推送",
+ "小果信号消费-盘中": "盘中消费小果扫描信号",
+ "硬编码扫描-每日": "扫描脚本中的硬编码参数",
+ "盘中自检-高频": "每15分钟盘中自检",
+ "数据治理-每周": "每周数据治理",
+ "记忆守卫-每日": "每日记忆优化",
+ "300308入场信号紧盯": "300308入场信号(13:00-14:00)",
+ "300308午后紧盯": "300308午后监控(13:00-15:00含止损)",
+ "多周期缓存刷新-盘中": "盘中刷新技术指标缓存",
+ "知识萃取-盘后": "盘后从分析报告中萃取可复用知识",
+ "区间维护": "每30分钟维护买入区",
+ "知微洞察生成": "生成每日市场洞察(15:35)",
+ "小果市场筛选-全市场": "小果筛选全市场关注板块",
+ "数据同步-dashboard": "同步数据到Dashboard",
+ "state.db真空整理-每周": "每周DB真空整理",
+ "未分类": "未被规则匹配的cron自动归入此",
+}
+
+# ── 数据流详细描述 ──
+# 每张表说明：存什么 + 谁写入(为什么+写什么) + 谁读取(为什么+读什么) + 综合总结
+FLOW_DETAIL = {
+ "signal_news": {
+ "summary": "全系统信号/新闻的统一存储表，所有宏观分析、风险扫描、小果分析的输出汇聚地，也是下游消费脚本的输入源。7个写入方汇聚不同来源信号，5个读取方按需消费。",
+ "writers": {
+ "macro_context_collector": "写入宏观新闻原始数据（标题+摘要+分类），供后续风险扫描消费",
+ "xiaoguo_news_processor": "写入小果LLM处理后的新闻情感分析结果",
+ "macro_signal_consumer": "写入宏观风险信号判定结果（等级+来源+建议）",
+ "divergence_detector": "写入跨市场背离检测信号（A股/港股/美股指数对）",
+ "xiaoguo_signal_consumer": "写入小果扫描发现的个股/板块信号",
+ "mofin_news": "写入外部财经常规新闻采集结果",
+ "xiaoguo_scanner": "写入小果独立扫描的市场机会信号",
+ },
+ "readers": {
+ "macro_signal_consumer": "读取原始宏观新闻和信号，判定风险等级并生成建议",
+ "system_audit": "读取信号表行数/更新时间，审计数据管道是否畅通",
+ "intraday_health_check": "读取最新信号，检查盘中是否有新的风险信号到达",
+ "xiaoguo_signal_consumer": "读取小果相关信号，生成买入/卖出建议",
+ "server": "读取信号数据供Web Dashboard展示",
+ },
+ },
+ "holdings": {
+ "summary": "当前持仓表，是系统最核心的数据表之一。import_holding_xls从券商文件导入持仓，mofin_db在价格刷新时更新市值。下游脚本读取持仓做策略分析和推送。",
+ "writers": {
+ "mofin_db": "写入price_monitor刷新后的持仓最新市值（通过write_holdings_batch）",
+ "import_holding_xls": "从券商holding.xls导入最新持仓数量/成本/市值",
+ },
+ "readers": {
+ "stale_push_wlin": "读取持仓列表+最新价格，检查是否进入买入区/触发止损",
+ "mofin_db": "内部读取（get_price_from_db等函数）",
+ "system_audit": "读取持仓总数/品种分布，审计持仓完整性",
+ "server": "读取持仓数据供Web Dashboard展示",
+ "prepare_report_data": "读取持仓数据用于生成分析报告",
+ "mo_data": "通过read_portfolio()读取持仓结构化数据",
+ },
+ },
+ "portfolio_summary": {
+ "summary": "组合汇总表（id=1单行），记录总资产=持股市值+可用资金+冻结资金。每笔导入或价格刷新后更新。",
+ "writers": {
+ "mofin_db": "价格监控刷新总市值后更新total_mv/total_assets",
+ "import_holding_xls": "导入持仓后更新cash/frozen/total_assets",
+ },
+ "readers": {
+ "import_holding_xls": "读取当前汇总信息，验证导入后是否正确",
+ "mo_data": "通过read_portfolio()读取组合汇总",
+ "price_monitor": "读取当前现金/市值，计算总资产变动",
+ "prepare_report_data": "读取总资产/现金数据用于报告",
+ "server": "读取汇总数据供Dashboard展示",
+ },
+ },
+ "holding_strategies": {
+ "summary": "策略数据表，记录每只持仓/自选股的策略配置（买入价/止损/止盈/目标价/分析维度等）。多写入方按各自职责更新不同字段。",
+ "writers": {
+ "data_governance": "归档过期策略、修复异常策略数据",
+ "sync_decisions_to_db": "从JSON同步策略到DB",
+ "mofin_db": "策略写入（内部函数）",
+ "strategy_review": "策略复盘后更新执行结果和评级",
+ },
+ "readers": {
+ "data_governance": "读取所有活跃策略，检查缺失和异常",
+ "per_stock_reassess": "读取个股策略配置，判断是否需要重评",
+ "mo_data": "通过read_decisions()读取策略数据",
+ "stale_push_wlin": "读取买入区/止损/止盈配置，检查价格触发",
+ },
+ },
+ "live_prices": {
+ "summary": "实时价格缓存表，price_monitor每2分钟写入全量持仓/自选价格。所有脚本必须通过mo_data.get_price()读取——先读此表，无数据才调API。单一写入、多方读取。",
+ "writers": {
+ "mofin_db": "price_monitor调用write_live_prices写入最新价格",
+ "mo_data": "get_price()兜底时从API拉取价格后写回此表",
+ },
+ "readers": {
+ "mo_data": "get_price()/get_prices_batch()优先从此表读取价格",
+ "mofin_db": "内部读取（get_price_from_db）",
+ "system_audit": "读取价格更新时间和数据量",
+ "verify_reassess_pipeline": "验证重评管道是否有最新价格",
+ },
+ },
+ "price_events": {
+ "summary": "价格触发事件表，价格进入/离开买入区或触发止损止盈时记录事件。用于审计和重评触发。",
+ "writers": {
+ "mofin_db": "price_monitor检测到价格区间变化时写入事件记录",
+ },
+ "readers": {
+ "mofin_db": "查询历史事件判断是否触发重评",
+ },
+ },
+ "cash_log": {
+ "summary": "资金流水表，每次资金变动（入金/出金/冻结/解冻）记录一条日志。审计用。",
+ "writers": {
+ "mofin_db": "通过write_cash_log记录资金变动",
+ "mo_data": "write_cash_log函数入口",
+ },
+ "readers": {
+ "prepare_report_data": "读取现金变动历史用于报告",
+ "mofin_db": "内部查询最近流水",
+ },
+ },
+ "market_snapshots": {
+ "summary": "市场快照表，market_watch每10分钟采集全市场大盘指数+板块涨跌+上涨下跌家数。下游用于判断市场情绪。",
+ "writers": {
+ "mofin_db": "market_watch采集后写入快照数据",
+ },
+ "readers": {
+ "market_screener": "读取最新板块快照，判断热点板块",
+ "prepare_report_data": "读取市场情绪数据用于报告",
+ "mofin_db": "内部查询最新快照",
+ "system_audit": "审计数据新鲜度",
+ },
+ },
+ "sector_snapshots": {
+ "summary": "板块快照表，market_watch按板块写入涨跌/领涨股/资金流向。market_screener据此判断行业热点。",
+ "writers": {
+ "mofin_db": "market_watch采集后写入各板块数据",
+ },
+ "readers": {
+ "market_screener": "读取板块涨跌排名，筛选热点行业",
+ "strategy_lifecycle": "读取板块数据用于策略生命周期管理",
+ "mofin_db": "内部查询",
+ "trend_detector": "读取板块趋势数据用于趋势检测",
+ },
+ },
+ "sector_signals": {
+ "summary": "板块信号表，多源汇聚的板块级别信号（新闻情感+趋势+资金流向）。用于判断行业轮动。",
+ "writers": {
+ "mofin_news": "写入新闻分析得出的板块信号",
+ "xiaoguo_news_processor": "写入小果LLM分析的板块情感信号",
+ "trend_detector": "写入技术面趋势检测到的板块信号",
+ },
+ "readers": {
+ "server": "读取供Dashboard展示",
+ "mofin_news": "读取已有信号做增量更新",
+ "xiaoguo_news_processor": "读取已有信号避免重复写入",
+ "trend_detector": "读取信号辅助趋势判定",
+ },
+ },
+ "macro_context_log": {
+ "summary": "宏观上下文日志，refresh_macro_context每30分钟采集大盘指数/市场情绪/资金面数据。下游多个脚本按需读取最新宏观状态。",
+ "writers": {
+ "refresh_macro_context": "每30分钟采集上证/深证/创业板/恒指等指数+情绪指标",
+ },
+ "readers": {
+ "stale_push_wlin": "读取大盘情绪用于策略推送的宏观背景",
+ "divergence_detector": "读取多市场指数数据做背离检测",
+ "system_audit": "审计数据采集是否正常",
+ "xiaoguo_signal_consumer": "读取宏观情绪辅助信号判定",
+ },
+ },
+ "macro_raw_news": {
+ "summary": "宏观新闻原始数据表，macro_context_collector采集的未经处理的财经新闻。供后续清洗和分析。",
+ "writers": {
+ "macro_context_collector": "从财经网站采集原始新闻标题+URL+摘要",
+ },
+ "readers": {
+ "macro_context_collector": "读取最近新闻hash避免重复采集",
+ "system_audit": "审计新闻采集量",
+ },
+ },
+ "accuracy_stats": {
+ "summary": "策略准确率统计表，strategy_review复盘后写入各策略的正确/错误/待定计数。",
+ "writers": {
+ "strategy_review": "策略复盘后更新准确率统计",
+ },
+ "readers": {
+ "mofin_db": "读取统计结果用于报告",
+ },
+ },
+ "advice_timeline": {
+ "summary": "建议时间线表，记录每条推送建议的时间/内容/状态。用于审计和对账。",
+ "writers": {
+ "advice_reconciliation": "每周对账时写入对账结果",
+ },
+ "readers": {
+ "advice_reconciliation": "读取历史建议做对账",
+ "mofin_db": "内部查询",
+ },
+ },
+ "candidate_score_history": {
+ "summary": "候选股评分历史表，记录每次全市场筛选时对候选股的评分。用于评分变化追踪。",
+ "writers": {
+ "mofin_db": "market_screener筛选结果写入评分记录",
+ },
+ "readers": {
+ "mofin_db": "查询评分历史供展示",
+ },
+ },
+ "candidates": {
+ "summary": "候选股池表，market_screener筛选出的值得关注的个股。包含评分/买入区/止损/目标价。",
+ "writers": {
+ "mofin_db": "market_screener写入候选股",
+ "market_screener": "直接写入候选股列表",
+ },
+ "readers": {
+ "mofin_db": "读取候选股数据供展示和后续处理",
+ },
+ },
+ "capital_flow_cache": {
+ "summary": "资金流向缓存表，capital_flow_collector采集的板块资金流入流出数据。",
+ "writers": {
+ "mofin_db": "写入板块资金流向数据",
+ },
+ "readers": {
+ "mofin_db": "读取缓存数据",
+ },
+ },
+ "health_check_log": {
+ "summary": "健康检查日志表，morning_health_check每次运行记录检查结果。用于追踪系统健康历史。",
+ "writers": {
+ "morning_health_check": "每日开盘前体检后写入检查结果",
+ },
+ "readers": {
+ "morning_health_check": "读取历史检查结果比较变化",
+ },
+ },
+ "mtf_cache": {
+ "summary": "多周期技术指标缓存，refresh_mtf_cache计算MA5/MA20/MA60/支撑阻力位等。下游技术分析脚本从缓存读取避免重复计算。",
+ "writers": {
+ "multi_timeframe": "计算并写入多周期MA/支撑阻力位",
+ "mofin_db": "内部写入函数",
+ },
+ "readers": {
+ "multi_timeframe": "读取已有缓存判断是否需要刷新",
+ "technical_analysis": "读取MA/支撑阻力位用于技术分析",
+ "mofin_db": "内部读取",
+ },
+ },
+ "stock_fundamentals": {
+ "summary": "基本面数据表，存储PE/PB/ROE/市值等财务指标。",
+ "writers": {
+ "mofin_db": "基本面数据采集后写入",
+ },
+ "readers": {
+ "strategy_lifecycle": "读取基本面数据用于策略评估",
+ },
+ },
+ "stock_sectors": {
+ "summary": "股票-板块映射表，记录每只股票所属行业板块。多脚本用于行业分类和板块归因。",
+ "writers": {
+ "mofin_db": "股票行业分类数据写入",
+ },
+ "readers": {
+ "xiaoguo_news_processor": "按行业分类新闻",
+ "mofin_news": "按行业归类新闻",
+ "mofin_db": "内部查询",
+ "strategy_lifecycle": "读取行业信息用于策略决策",
+ },
+ },
+ "stocks": {
+ "summary": "全量股票代码表，所有A股/港股基础信息。供各脚本按code查询股票名称/市场。",
+ "writers": {
+ "mofin_db": "初始化时导入全量股票代码",
+ },
+ "readers": {
+ "mofin_news": "按股票代码查找新闻",
+ "xiaoguo_news_processor": "按股票代码过滤新闻",
+ "mofin_db": "内部查询",
+ "trend_detector": "按股票代码获取数据",
+ },
+ },
+ "strategy_evaluations": {
+ "summary": "策略评估结果表，策略评估脚本每次运行记录评估得分/等级/评语。",
+ "writers": {
+ "mofin_collect": "策略评估前采集数据并写入评估结果",
+ },
+ "readers": {
+ "verify_reassess_pipeline": "读取评估结果验证管道完整性",
+ "mofin_db": "内部查询",
+ "system_audit": "审计评估是否按时执行",
+ },
+ },
+ "strategy_feedback": {
+ "summary": "策略反馈表，记录用户对建议的反馈（采纳/忽略/修改）。用于策略自学习。",
+ "writers": {
+ "mofin_db": "写入反馈数据",
+ "server": "通过Web提交反馈后写入",
+ },
+ "readers": {
+ "mofin_db": "读取反馈用于分析和展示",
+ },
+ },
+ "todos": {
+ "summary": "待办事项表，各脚本发现异常时写入TODO，self_todo_executor每10分钟执行修复。异常发现→自动修复的闭环。",
+ "writers": {
+ "morning_health_check": "体检发现异常写入TODO",
+ "intraday_health_check": "盘中自检发现异常写入TODO",
+ "strategy-staleness-check": "策略过期检测写入TODO",
+ "self_todo_executor": "执行完成后更新TODO状态",
+ "preflight_verify": "开盘前验证失败写入TODO",
+ },
+ "readers": {
+ "morning_health_check": "读取待处理的TODO",
+ "self_todo_executor": "读取待处理的TODO并执行fix_action",
+ "strategy-staleness-check": "读取TODO避免重复写入",
+ "intraday_health_check": "读取TODO检查自愈进度",
+ },
+ },
+ "watchlist_stocks": {
+ "summary": "自选股表，系统自动维护的观察列表。与持仓表分离，用于跟踪潜在买入机会。",
+ "writers": {
+ "per_stock_reassess": "策略重评时更新自选状态",
+ "mofin_db": "内部写入函数",
+ },
+ "readers": {
+ "per_stock_reassess": "读取自选列表做重评",
+ "stock_quote": "读取自选代码拉取行情",
+ "mo_alphasift_bridge": "读取自选供Alpha分析",
+ "mo_data": "通过read_watchlist()读取自选数据",
+ },
+ },
+ "xiaoguo_scan_tracker": {
+ "summary": "小果扫描追踪表，记录每次小果扫描的状态/耗时/结果数量。用于监控小果服务健康。",
+ "writers": {
+ "xiaoguo_scanner": "每次扫描完成后写入状态和统计",
+ },
+ "readers": {
+ "server": "读取扫描状态供Dashboard展示",
+ "xiaoguo_scanner": "读取上次扫描时间判断是否需要全量扫描",
+ },
+ },
+ "state_meta": {
+ "summary": "状态元数据表，记录各服务的状态追踪信息（如扫描偏移量/最新处理ID）。",
+ "writers": {
+ "xiaoguo_scanner": "写入扫描进度偏移量",
+ },
+ "readers": {
+ "xiaoguo_scanner": "读取上次处理位置继续增量处理",
+ },
+ },
+}
+
+
+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", "")
+ # 附加描述（自动带脚本名的节点去掉括号内容匹配）
+ desc_key = label.split(" (")[0] if " (" in label else label
+ if desc_key in NODE_DESC:
+ node["desc"] = NODE_DESC[desc_key]
+ 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):
+ name = j.get("name", "?")
+ desc_key = name.split(" (")[0] if " (" in name else name
+ return {
+ "label": f"{name} ({j.get('script','LLM')})",
+ "desc": NODE_DESC.get(desc_key, ""),
+ "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, [])
+ # 扫描器漏检的手动补录写入方
+ _manual_writers = {
+ "candidates": ["mofin_db", "market_screener"],
+ "candidate_score_history": ["mofin_db"],
+ "strategy_feedback": ["mofin_db", "server"],
+ "stock_daily": ["mofin_db"],
+ "stock_weekly": ["mofin_db"],
+ "stock_monthly": ["mofin_db"],
+ }
+ _manual_readers = {
+ "stock_weekly": ["multi_timeframe"],
+ "stock_monthly": ["multi_timeframe"],
+ "watchlist_log": ["watchlist_auto_exit", "mofin_db"],
+ }
+ if not writers and tname in _manual_writers:
+ writers = _manual_writers[tname]
+ if not readers and tname in _manual_readers:
+ readers = _manual_readers[tname]
+
+ # 数据流详细描述
+ flow_detail = FLOW_DETAIL.get(tname, {})
+
+ has_input = len(writers) > 0
+ has_output = len(readers) > 0
+ # 排除系统表
+ is_system = tname.startswith("sqlite_") or tname.startswith("_")
+ if is_system:
+ continue
+ # 数据流状态：healthy / write_only / read_only / orphan
+ if has_input and has_output:
+ flow_status = "healthy"
+ elif has_input and not has_output:
+ flow_status = "write_only"
+ elif not has_input and has_output:
+ flow_status = "read_only"
+ else:
+ flow_status = "orphan"
+ 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": flow_status in ("orphan", "read_only", "write_only"),
+ "flow_status": flow_status,
+ "warn": flow_status != "healthy",
+ "flow_detail": flow_detail,
+ })
+
+ # JSON文件
+ # 已迁移到DB的旧JSON文件：不再报"无读取方"假警报，真实健康信号看DB表新鲜度
+ MIGRATED_TO_DB = {
+ "multi_tf_cache.json": "mtf_cache",
+ "macro_context.json": "macro_context_log",
+ "market.json": "market_snapshots",
+ "live_prices.json": "live_prices",
+ "price_history.json": "price_events",
+ "macro_risk_state.json": "macro_context_log",
+ }
+ 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
+ migrated = MIGRATED_TO_DB.get(jf.name)
+ desc = JSON_DESC.get(jf.name, "")
+ if migrated:
+ desc = (desc + " " if desc else "") + f"(已迁移到DB表 {migrated}，此为遗留文件)"
+ json_entities.append({
+ "name": jf.name,
+ "desc": desc,
+ "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")
+ and not migrated),
+ "migrated_to_db": migrated or None,
+ })
+
+ # ── DB表新鲜度：真实数据管道健康信号（替代对遗留JSON文件的mtime检查）──
+ # 注意：活跃数据在 /home/hmo/MoFin/data/mofin.db（live_prices/mtf_cache 今日有写入），
+ # 不用 get_conn()（它指向 web-dashboard 的库，那边部分表是旧的）
+ db_freshness = []
+ FRESHNESS_TABLES = [
+ ("mtf_cache", "updated_at", "多周期均线缓存"),
+ ("macro_context_log", "created_at", "宏观上下文"),
+ ("market_snapshots", "created_at", "市场快照"),
+ ("live_prices", "updated_at", "实时价格"),
+ ("price_events", "created_at", "价格事件"),
+ ]
+ try:
+ _fc = sqlite3.connect("/home/hmo/MoFin/data/mofin.db", timeout=10)
+ for tname, tcol, label in FRESHNESS_TABLES:
+ try:
+ row = _fc.execute(
+ f"SELECT MAX({tcol}) FROM {tname}").fetchone()
+ if row and row[0]:
+ last_dt = datetime.fromisoformat(str(row[0]).replace("Z", ""))
+ age_h = (now - last_dt).total_seconds() / 3600
+ db_freshness.append({
+ "table": tname, "label": label,
+ "last_record": last_dt.strftime("%m-%d %H:%M"),
+ "age_hours": round(age_h, 1),
+ "warn": age_h > 24,
+ })
+ else:
+ db_freshness.append({"table": tname, "label": label,
+ "last_record": None, "age_hours": -1, "warn": True})
+ except Exception:
+ pass # 表不存在或列名不同，跳过
+ _fc.close()
+ except Exception:
+ pass
+
+ # ── 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", "?"),
+ })
+
+ # ── 自检体系状态（L1功能健康/L3修复记录/L4元监控/L2卫生）──
+ self_check = {}
+ LOGS = Path('/home/hmo/MoFin/gateway/logs')
+ try:
+ fh = json.loads((LOGS / 'functional_health.json').read_text(encoding='utf-8'))
+ self_check['functional'] = {
+ 'generated_at': fh.get('generated_at'), 'status': fh.get('status'),
+ 'summary': fh.get('summary'), 'checks': fh.get('checks', []),
+ }
+ except Exception:
+ self_check['functional'] = None
+ try:
+ mw = json.loads((LOGS / 'meta_watchdog.json').read_text(encoding='utf-8'))
+ self_check['meta_watchdog'] = {
+ 'generated_at': mw.get('generated_at'), 'status': mw.get('status'),
+ 'layers': mw.get('layers', []),
+ }
+ except Exception:
+ self_check['meta_watchdog'] = None
+ try:
+ hy = json.loads((LOGS / 'hygiene_report.json').read_text(encoding='utf-8'))
+ self_check['hygiene'] = {
+ 'generated_at': hy.get('generated_at'), 'status': hy.get('status'),
+ 'issue_count': hy.get('issue_count'), 'issues': hy.get('issues', [])[:10],
+ }
+ except Exception:
+ self_check['hygiene'] = None
+ try:
+ repairs = []
+ rp = LOGS / 'repair_log.jsonl'
+ if rp.exists():
+ for line in rp.read_text(encoding='utf-8').splitlines()[-10:]:
+ try:
+ repairs.append(json.loads(line))
+ except Exception:
+ pass
+ repairs.reverse()
+ self_check['recent_repairs'] = repairs
+ except Exception:
+ self_check['recent_repairs'] = []
+
+ # ── 写JSON ──
+ report = {
+ "generated_at": now.strftime("%Y-%m-%d %H:%M:%S"),
+ "feature_tree": feature_tree,
+ "entities": entities,
+ "json_files": json_entities,
+ "pipelines": pipelines,
+ "db_freshness": db_freshness,
+ "self_check": self_check,
+ }
+ 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()
diff --git a/deploy/profile-scripts/self_repair.py b/deploy/profile-scripts/self_repair.py
new file mode 100644
index 00000000..50299da5
--- /dev/null
+++ b/deploy/profile-scripts/self_repair.py
@@ -0,0 +1,255 @@
+#!/usr/bin/env python3
+"""self_repair.py — L3 LLM 修复循环（报备制）
+
+读取 L1 functional_health.json + L2 hygiene_report.json 的失败项，
+调用 LLM 诊断并**直接执行修复**（先斩后奏），然后记录日志并 XMPP 报备。
+
+安全设计：
+- LLM 只能从白名单动作中选择（不许任意执行代码）
+- 每个模块每天最多自动修复 2 次（防修复循环）
+- LLM 不可用时降级为规则默认动作（rerun_script）
+- 所有动作记录 repair_log.jsonl
+
+频率：每 30 分钟（cron）。
+"""
+import os, sys, json, subprocess, sqlite3, time
+from datetime import datetime
+
+sys.path.insert(0, '/home/hmo/MoFin')
+
+FUNCTIONAL_REPORT = '/home/hmo/MoFin/gateway/logs/functional_health.json'
+HYGIENE_REPORT = '/home/hmo/MoFin/gateway/logs/hygiene_report.json'
+REPAIR_LOG = '/home/hmo/MoFin/gateway/logs/repair_log.jsonl'
+REPAIR_STATE = '/home/hmo/MoFin/gateway/logs/repair_state.json'
+GATEWAY = 'http://127.0.0.1:8643/v1/chat/completions'
+SCRIPTS_DIR = '/home/hmo/.hermes/profiles/position-analyst/scripts'
+MAX_REPAIRS_PER_MODULE_PER_DAY = 2
+
+WHITELIST_ACTIONS = """
+可执行的白名单动作（只能选其一，不许自创）：
+1. {"action": "rerun_script", "script": "<脚本名>"} — 立即重跑指定 cron 脚本（限注册表内的脚本）
+2. {"action": "restart_service", "service": "<服务名>"} — systemctl 重启（限: hermes-gateway-zhiwei, xmpp-zhiwei, mofin-dashboard）
+3. {"action": "sync_links"} — 重建脚本硬链接（修复 cron 跑旧代码）
+4. {"action": "switch_llm_key"} — 切换 LLM key（429 限流时）
+5. {"action": "none", "reason": "<原因>"} — 判断为无需动作（如非交易时段的正常空闲）
+"""
+
+
+def log_repair(entry):
+ os.makedirs(os.path.dirname(REPAIR_LOG), exist_ok=True)
+ with open(REPAIR_LOG, 'a', encoding='utf-8') as f:
+ f.write(json.dumps(entry, ensure_ascii=False) + '\n')
+
+
+def load_state():
+ try:
+ return json.load(open(REPAIR_STATE))
+ except Exception:
+ return {}
+
+
+def save_state(s):
+ with open(REPAIR_STATE, 'w') as f:
+ json.dump(s, f)
+
+
+def repairs_today(state, module):
+ today = datetime.now().strftime('%Y-%m-%d')
+ return state.get(today, {}).get(module, 0)
+
+
+def bump_repair(state, module):
+ today = datetime.now().strftime('%Y-%m-%d')
+ state.setdefault(today, {})
+ state[today][module] = state[today].get(module, 0) + 1
+ # 只保留最近 3 天
+ for k in list(state.keys()):
+ if k < (datetime.now().strftime('%Y-%m-%d'))[:8] + '17':
+ del state[k]
+ save_state(state)
+
+
+def collect_failures():
+ failures = []
+ try:
+ fh = json.load(open(FUNCTIONAL_REPORT))
+ for c in fh.get('checks', []):
+ if c.get('status') == 'fail':
+ failures.append({
+ 'source': 'functional', 'module': c['module'],
+ 'function': c['function'], 'reason': c['reason'],
+ 'repair_hint': c.get('repair'),
+ })
+ except Exception:
+ pass
+ try:
+ hy = json.load(open(HYGIENE_REPORT))
+ for i in hy.get('issues', []):
+ failures.append({
+ 'source': 'hygiene', 'module': i.get('type'),
+ 'function': i.get('file') or i.get('job') or i.get('table', '?'),
+ 'reason': json.dumps(i, ensure_ascii=False)[:200],
+ 'repair_hint': {'action': 'llm_diagnose'},
+ })
+ except Exception:
+ pass
+ return failures
+
+
+def run_action(action, module):
+ """执行白名单动作，返回 (ok, detail)"""
+ act = action.get('action')
+ try:
+ if act == 'rerun_script':
+ script = action.get('script', '')
+ # 只允许注册表内脚本
+ if not script.endswith('.py') or '/' in script or '..' in script:
+ return False, f'非法脚本名: {script}'
+ path = os.path.join(SCRIPTS_DIR, script)
+ if not os.path.exists(path):
+ return False, f'脚本不存在: {script}'
+ r = subprocess.run(['python3', path], capture_output=True, text=True,
+ timeout=600, cwd=SCRIPTS_DIR)
+ tail = (r.stdout or r.stderr)[-150:]
+ return r.returncode == 0, f'exit={r.returncode} {tail}'
+
+ elif act == 'restart_service':
+ svc = action.get('service', '')
+ allowed = {'hermes-gateway-zhiwei': ['sudo', '-n', 'systemctl', 'restart', svc],
+ 'xmpp-zhiwei': ['sudo', '-n', 'systemctl', 'restart', svc],
+ 'mofin-dashboard': ['sudo', '-n', 'systemctl', 'restart', svc]}
+ if svc not in allowed:
+ return False, f'服务不在白名单: {svc}'
+ subprocess.Popen(allowed[svc], stdout=subprocess.DEVNULL,
+ stderr=subprocess.DEVNULL, start_new_session=True)
+ return True, f'{svc} 重启已触发(异步)'
+
+ elif act == 'sync_links':
+ r = subprocess.run(['bash', '/home/hmo/MoFin/deploy/profile-scripts/sync_profile_scripts.sh'],
+ capture_output=True, text=True, timeout=60)
+ return True, r.stdout.strip()
+
+ elif act == 'switch_llm_key':
+ sys.path.insert(0, '/home/hmo/MoFin')
+ from xmpp_logger import best_key, switch_key, KEY_TO_PROVIDER, current_provider
+ bk = best_key()
+ if not bk:
+ return False, '无可用 key'
+ target = KEY_TO_PROVIDER.get(bk['key_id'])
+ if not target or target == current_provider():
+ return False, f'已在最优 key ({bk["key_id"]})'
+ r = switch_key(bk['key_id'])
+ return r.get('switched', False), json.dumps(r, ensure_ascii=False)[:150]
+
+ elif act == 'none':
+ return True, f'无需动作: {action.get("reason", "")}'
+
+ return False, f'未知动作: {act}'
+ except subprocess.TimeoutExpired:
+ return False, '动作超时(600s)'
+ except Exception as e:
+ return False, f'动作异常: {str(e)[:120]}'
+
+
+def llm_diagnose(failure):
+ """调 LLM 诊断并返回白名单动作。LLM 不可用 → 返回 None（走规则默认）"""
+ hint = failure.get('repair_hint') or {}
+ default_action = hint if hint.get('action') and hint['action'] != 'llm_diagnose' else None
+
+ prompt = f"""你是 MoFin 系统的自动修复工程师。一个健康检查发现了功能异常，请诊断并给出一个修复动作。
+
+【异常信息】
+模块: {failure['module']}
+功能: {failure['function']}
+异常表现: {failure['reason']}
+来源: {failure['source']}
+
+【背景】
+- 系统: Linux 246, MoFin 股票分析系统
+- cron 脚本目录: {SCRIPTS_DIR}
+- 数据库: /home/hmo/MoFin/data/mofin.db
+- 当前时间: {datetime.now().strftime('%Y-%m-%d %H:%M')}（注意是否交易时段，非交易时段部分管道停跑属正常）
+
+{WHITELIST_ACTIONS}
+
+请只输出一个 JSON 动作（不要其他文字）。如果是非交易时段的正常空闲，选 none。"""
+
+ try:
+ payload = json.dumps({
+ 'model': 'deepseek-v4-flash',
+ 'messages': [{'role': 'user', 'content': prompt}],
+ 'max_tokens': 300, 'stream': False,
+ }).encode()
+ import urllib.request
+ req = urllib.request.Request(GATEWAY, data=payload, headers={
+ 'Content-Type': 'application/json', 'Authorization': 'Bearer hermes123',
+ 'X-Hermes-Session-Id': 'self-repair'})
+ resp = urllib.request.urlopen(req, timeout=120)
+ text = json.loads(resp.read())['choices'][0]['message']['content']
+ # 提取 JSON
+ import re
+ m = re.search(r'\{[^{}]*"action"[^{}]*\}', text)
+ if m:
+ action = json.loads(m.group(0))
+ if action.get('action') in ('rerun_script', 'restart_service', 'sync_links',
+ 'switch_llm_key', 'none'):
+ return action, text[:200]
+ except Exception as e:
+ print(f' LLM 诊断不可用: {str(e)[:80]}')
+ return (default_action, None) if default_action else (None, None)
+
+
+def xmpp_report(lines):
+ try:
+ import urllib.request
+ payload = json.dumps({'to': 'hmo@yoin.fun', 'body': '\n'.join(lines), 'type': 'chat'}).encode()
+ req = urllib.request.Request('http://127.0.0.1:5805/', data=payload,
+ headers={'Content-Type': 'application/json'})
+ urllib.request.urlopen(req, timeout=5)
+ except Exception as e:
+ print(f'XMPP 失败: {e}')
+
+
+def main():
+ now = datetime.now()
+ print('🔧 self_repair', now.strftime('%H:%M'))
+ failures = collect_failures()
+ if not failures:
+ print(' ✅ 无失败项')
+ return
+
+ state = load_state()
+ reports = []
+ for f in failures:
+ module = f['module']
+ if repairs_today(state, module) >= MAX_REPAIRS_PER_MODULE_PER_DAY:
+ print(f' ⏭ {module}: 今日已修复 {MAX_REPAIRS_PER_MODULE_PER_DAY} 次，跳过（防循环）')
+ continue
+
+ print(f' 🔍 {module}: {f["reason"][:60]}')
+ action, llm_note = llm_diagnose(f)
+ if action is None:
+ print(f' 无可用动作，跳过')
+ log_repair({'ts': now.isoformat(), 'module': module, 'reason': f['reason'],
+ 'action': 'skip_no_action', 'ok': None})
+ continue
+
+ ok, detail = run_action(action, module)
+ bump_repair(state, module)
+ entry = {
+ 'ts': now.isoformat(), 'module': module, 'function': f['function'],
+ 'reason': f['reason'], 'action': action, 'ok': ok, 'detail': detail[:300],
+ 'llm_note': llm_note,
+ }
+ log_repair(entry)
+ icon = '✅' if ok else '❌'
+ print(f' {icon} {action.get("action")} -> {detail[:80]}')
+ reports.append(f'{icon} [{module}] {action.get("action")}: {detail[:80]}')
+
+ if reports:
+ xmpp_report(['🔧 自检系统自动修复报备：', f'发现 {len(failures)} 个功能异常，已直接处理：'] + reports +
+ ['', '详情: gateway/logs/repair_log.jsonl'])
+
+
+if __name__ == '__main__':
+ main()
\ No newline at end of file
diff --git a/static/mofin_health.html b/static/mofin_health.html
index 05c39322..1017877b 100644
--- a/static/mofin_health.html
+++ b/static/mofin_health.html
@@ -1,328 +1,390 @@
-
-
-MoFin 健康监控
-
-
-📊 MoFin 系统健康监控
-加载中...
-
-
🌳 功能树
-
🔧 全部流程/Cron
-
🗃️ 数据实体
-
🔀 数据流
-
📊 MoFin 系统健康监控
+加载中...
+
+
🌳 功能树
+
🔧 全部流程/Cron
+
🗃️ 数据实体
+
🔀 数据流
+
🩺 自检体系
+