hmo/MoFin
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

fix: 功能树每个节点嵌入cron管道信息，修复数据采集节点不显示cron的问题

Browse Source
This commit is contained in:
知微
2026-07-08 16:29:02 +08:00
parent f367c8c94c
commit c614eee8b1
3 changed files with 1051 additions and 346 deletions
Download Patch File Download Diff File Expand all files Collapse all files
scripts/mofin_health.py
+153 -58
View File
@@ -129,70 +129,165 @@ def check_scripts():
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):
# 匹配规则：每个节点名称→匹配的cron名称关键字
rules = {
"市场快照 (market_watch.py)": ["市场数据采集"],
"宏观新闻 (macro_context_collector.py)": ["宏观采集"],
"价格监控 (price_monitor.py)": ["价格监控"],
"小果扫描 (xiaoguo_scanner.py)": ["小果独立扫描"],
"资金流采集 (capital_flow_collector.py)": ["资金流"],
"宏观上下文刷新 (refresh_macro_context.py)": ["宏观上下文刷新"],
"策略重评 (mofin_collect→reassess_with_context)": ["策略重评"],
"持仓自选新鲜度检查": ["策略时效性检查"],
"自选买入区提醒 (stale_push_wlin.py)": ["自选买入区提醒"],
"策略评估 (strategy_evaluator.py)": ["策略评估"],
"分支自成长 (branch_scanner.py)": ["分支自成长"],
"元自成长 (meta_growth.py)": ["元自成长"],
"MoFin盘前中监控 (LLM cron)": ["MoFin盘前中监控"],
"MoFin午后监控 (LLM cron)": ["MoFin午后监控"],
"cron报告推XMPP (cron_to_xmpp.py)": ["cron报告推XMPP"],
"开盘简报 (LLM cron)": ["开盘简报"],
"收盘简报 (LLM cron)": ["收盘简报"],
"市场精选推荐 (LLM cron)": ["市场精选推荐"],
"小果情感分析 (LLM cron)": ["小果情感分析"],
"系统全局审计": ["系统全局审计"],
"全局cron健康监控": ["全局cron健康监控"],
"重评管道审计": ["重评管道审计"],
"健康监控数据采集": ["健康监控数据采集"],
"持仓基本面复查": ["分析师-持仓复查"],
"策略复牌": ["策略复盘"],
"宏观风险扫描 (LLM cron)": ["宏观风险扫描"],
"宏观风险信号消费 (macro_signal_consumer.py)": ["宏观风险信号消费"],
"跨市场背离检测 (divergence_detector.py)": ["跨市场背离检测"],
"自愈执行器 (self_todo_executor.py)": ["自愈执行器"],
"策略质量门禁 (review_needed_watchdog.py)": ["策略质量门禁"],
"自选自动清理 (clean_watchlist.py)": ["自选自动清理"],
"建议对账 (advice_reconciliation.py)": ["建议对账"],
"XMPP Bot (zhiwei)": None,
"Gateway (8643)": None,
"HTTP Bridge (5805)": None,
"Dashboard (8899)": None,
"state.db SQLite": None,
}
def attach_pipes(node):
label = node.get("label", "")
keywords = rules.get(label)
if keywords is not None:
matched = match_cron(cron_jobs, keywords)
node["pipes"] = [{
"name": j.get("name", ""),
"script": j.get("script", ""),
"schedule": 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",
} for j in matched]
if node.get("children"):
for c in node["children"]:
attach_pipes(c)
tree = {
"label": "MoFin 系统",
"status": "ok",
"children": [
{"label": "数据采集", "status": "ok", "children": [
{"label": "市场快照 (market_watch.py)", "status": "ok" if any(j.get("name")=="市场数据采集" and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": "宏观新闻 (macro_context_collector.py)", "status": "ok" if any("宏观采集" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": "价格监控 (price_monitor.py)", "status": "ok" if any("价格监控" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": "小果扫描 (xiaoguo_scanner.py)", "status": "ok" if any("小果独立扫描" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": "资金流采集 (capital_flow_collector.py)", "status": "ok" if any("资金流" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": "宏观上下文刷新 (refresh_macro_context.py)", "status": "ok"},
]},
{"label": "策略分析", "status": "ok", "children": [
{"label": "策略重评 (mofin_collect→reassess_with_context)", "status": "ok" if db_stats.get("strategy_evaluations",0) > 200 else "warn"},
{"label": "持仓自选新鲜度检查", "status": "ok"},
{"label": "自选买入区提醒 (stale_push_wlin.py)", "status": "ok" if any("自选买入区提醒" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": "策略评估 (strategy_evaluator.py)", "status": "ok" if any("策略评估" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": "分支自成长 (branch_scanner.py)", "status": "ok"},
{"label": "元自成长 (meta_growth.py)", "status": "ok"},
]},
{"label": "推荐推送", "status": "ok", "children": [
{"label": "MoFin盘前中监控 (LLM cron)", "status": "ok" if any("MoFin盘前中监控" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": "MoFin午后监控 (LLM cron)", "status": "ok" if any("MoFin午后监控" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": "cron报告推XMPP (cron_to_xmpp.py)", "status": "ok"},
{"label": "开盘简报 (LLM cron)", "status": "ok"},
{"label": "收盘简报 (LLM cron)", "status": "ok"},
{"label": "市场精选推荐 (LLM cron)", "status": "ok" if any("市场精选推荐" in j.get("name","") and j.get("enabled")==True for j in cron_jobs) else "warn"},
]},
{"label": "自检/审计", "status": "ok", "children": [
{"label": "系统全局审计", "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", "children": [
{"label": "小果情感分析 (LLM cron)", "status": "ok"},
]},
]
+ [
{"label": "风险监控", "status": "ok", "children": [
{"label": "宏观风险扫描 (LLM cron)", "status": "ok"},
{"label": "宏观风险信号消费 (macro_signal_consumer.py)", "status": "ok" if any("宏观风险信号消费" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": "跨市场背离检测 (divergence_detector.py)", "status": "ok"},
]},
{"label": "执行/修复", "status": "ok", "children": [
{"label": "自愈执行器 (self_todo_executor.py)", "status": "ok"},
{"label": "策略质量门禁 (review_needed_watchdog.py)", "status": "ok"},
{"label": "自选自动清理 (clean_watchlist.py)", "status": "warn" if any("自选自动清理" in j.get("name","") and j.get("last_status")=="error" for j in cron_jobs) else "ok"},
{"label": "建议对账 (advice_reconciliation.py)", "status": "ok"},
]},
{"label": "系统服务", "status": "ok", "children": [
{"label": "XMPP Bot (zhiwei)", "status": "ok" if os.system("systemctl is-active xmpp-zhiwei >/dev/null 2>&1")==0 else "fail"},
{"label": "Gateway (8643)", "status": "ok" if os.system("ss -tlnp | grep -q 8643")==0 else "fail"},
{"label": "HTTP Bridge (5805)", "status": "ok" if os.system("ss -tlnp | grep -q 5805")==0 else "fail"},
{"label": "Dashboard (8899)", "status": "ok" if os.system("ss -tlnp | grep -q 8899")==0 else "warn"},
{"label": "state.db SQLite", "status": "ok"},
]},
],
}
attach_pipes(tree)
return tree
def build_report():
cron_jobs = load_cron_jobs()
db_stats = get_db_stats()
flows = scan_data_flows()
script_health = check_scripts()
# ── Tab 1: 功能树 ──
feature_tree = {
"label": "MoFin 系统",
"status": "ok",
"children": [
{"label": "数据采集", "status": "ok", "children": [
{"label": f"市场快照 (market_watch.py)", "status": "ok" if any(j.get("name")=="市场数据采集" and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": f"宏观新闻 (macro_context_collector.py)", "status": "ok" if any("宏观采集" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": f"价格监控 (price_monitor.py)", "status": "ok" if any("价格监控" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": f"小果扫描 (xiaoguo_scanner.py)", "status": "ok" if any("小果独立扫描" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": f"资金流采集 (capital_flow_collector.py)", "status": "ok" if any("资金流" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": f"宏观上下文刷新 (refresh_macro_context.py)", "status": "ok"},
]},
{"label": "策略分析", "status": "ok", "children": [
{"label": f"策略重评 (mofin_collect→reassess_with_context)", "status": "ok" if db_stats.get("strategy_evaluations",0) > 200 else "warn"},
{"label": f"持仓自选新鲜度检查", "status": "ok"},
{"label": f"自选买入区提醒 (stale_push_wlin.py)", "status": "ok" if any("自选买入区提醒" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": f"策略评估 (strategy_evaluator.py)", "status": "ok" if any("策略评估" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": f"分支自成长 (branch_scanner.py)", "status": "ok"},
{"label": f"元自成长 (meta_growth.py)", "status": "ok"},
]},
{"label": "推荐推送", "status": "ok", "children": [
{"label": f"MoFin盘前中监控 (LLM cron)", "status": "ok" if any("MoFin盘前中监控" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": f"MoFin午后监控 (LLM cron)", "status": "ok" if any("MoFin午后监控" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": f"cron报告推XMPP (cron_to_xmpp.py)", "status": "ok"},
{"label": f"开盘简报 (LLM cron)", "status": "ok"},
{"label": f"收盘简报 (LLM cron)", "status": "ok"},
{"label": f"市场精选推荐 (LLM cron)", "status": "ok" if any("市场精选推荐" in j.get("name","") and j.get("enabled")==True for j in cron_jobs) else "warn"},
]},
{"label": "自检/审计", "status": "ok", "children": [
{"label": f"系统全局审计 (system_audit.py)", "status": "ok"},
{"label": f"全局cron健康监控 (cron_health_monitor.py)", "status": "ok"},
{"label": f"重评管道审计 (verify_reassess_pipeline.py)", "status": "ok"},
{"label": f"系统体检 (morning_health_check.py)", "status": "ok"},
{"label": f"盘中自检 (intraday_health_check.py)", "status": "ok"},
{"label": f"记忆守卫 (memory_guardian.py)", "status": "ok"},
{"label": f"硬编码扫描 (hardcode_scanner.py)", "status": "ok"},
]},
{"label": "风险监控", "status": "ok", "children": [
{"label": f"宏观风险扫描 (LLM cron)", "status": "ok"},
{"label": f"宏观风险信号消费 (macro_signal_consumer.py)", "status": "ok" if any("宏观风险信号消费" in j.get("name","") and j.get("last_status")=="ok" for j in cron_jobs) else "warn"},
{"label": f"跨市场背离检测 (divergence_detector.py)", "status": "ok"},
]},
{"label": "执行/修复", "status": "ok", "children": [
{"label": f"自愈执行器 (self_todo_executor.py)", "status": "ok"},
{"label": f"策略质量门禁 (review_needed_watchdog.py)", "status": "ok"},
{"label": f"自选自动清理 (clean_watchlist.py)", "status": "warn" if any("自选自动清理" in j.get("name","") and j.get("last_status")=="error" for j in cron_jobs) else "ok"},
{"label": f"建议对账 (advice_reconciliation.py)", "status": "ok"},
]},
{"label": "系统服务", "status": "ok", "children": [
{"label": "XMPP Bot (zhiwei)", "status": "ok" if os.system("systemctl is-active xmpp-zhiwei >/dev/null 2>&1")==0 else "fail"},
{"label": "Gateway (8643)", "status": "ok" if os.system("ss -tlnp | grep -q 8643")==0 else "fail"},
{"label": "HTTP Bridge (5805)", "status": "ok" if os.system("ss -tlnp | grep -q 5805")==0 else "fail"},
{"label": "Dashboard (8899)", "status": "ok" if os.system("ss -tlnp | grep -q 8899")==0 else "warn"},
{"label": "state.db SQLite", "status": "ok"},
]},
]
}
# ── 功能树 ──
feature_tree = build_feature_tree(cron_jobs, db_stats)
# 递归计算节点状态
def calc_status(node):
if "children" in node: