#!/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 # ── 消息通道统一路由(broadcast/xmpp by delivery) ── try: from messenger import install_stdio_hook as _msh _msh() except Exception: pass 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": "健康监控数据", } # ── 2026-08-12 三层架构层归属映射（采集→加工→使用，老莫定）── # 数据表层归属：collect=采集层原始数据 / process=加工层衍生数据 / biz=使用层业务数据 TABLE_LAYER = { # 采集层原始数据（采集任务直接写入） "stock_daily": "collect", "stock_weekly": "collect", "stock_monthly": "collect", "stock_news": "collect", "stock_fundamentals": "collect", "stock_capital_flow": "collect", "market_snapshots": "collect", "sector_snapshots": "collect", "macro_raw_news": "collect", "macro_context_log": "collect", "live_prices": "collect", "capital_flow_cache": "collect", "stocks": "collect", "stock_sectors": "collect", "stock_sectors_em": "collect", "price_events": "collect", # 加工层衍生数据（加工任务由原始数据算出） "stock_indicators": "process", "market_indicators": "process", "sector_index_daily": "process", "mtf_cache": "process", "portfolio_state": "process", # 其余默认 biz（使用层业务数据：candidates/holdings/strategy_*/watchlist 等） } # 任务层归属（按脚本名关键词匹配，优先级 collect>process>use>monitor，默认 use） PIPELINE_LAYER_RULES = [ ("collect", ["daily_kline", "news_collector", "fundamentals_full", "fundamentals_refresh", "market_watch", "price_monitor", "capital_flow_collector", "macro_context_collector", "collect_evaluation_data", "refresh_macro_context", "sector_enrich", "import_holding", "market_data", "mofin_collect"]), ("process", ["factor_engine", "sector_index_builder", "refresh_mtf_cache", "multi_timeframe"]), ("use", ["scanner", "screener", "candidate_filter", "promote_candidates", "reassess", "market_insight", "strategy_effectiveness", "strategy_review", "premarket_full_review", "staleness", "review_needed", "stale_push", "mr_scanner", "s2_scanner", "predictive_oversold", "accumulation", "watchlist_12d"]), ("monitor", ["health", "audit", "verify", "governance", "guardian", "watchdog", "repair", "spiral", "deploy_guard", "todo_executor", "backup", "vacuum", "gateway", "clean_watchlist", "xmpp", "brief", "signal_consumer", "divergence", "monitor", "hygiene", "hardcode", "preflight"]), ] def _pipeline_layer(script, name): """按脚本名/任务名关键词匹配层归属（collect>process>use>monitor，默认 use）""" s = (script or "").lower() + " " + (name or "").lower() for layer, kws in PIPELINE_LAYER_RULES: if any(k in s for k in kws): return layer return "use" 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 j["source"] = "hermes" # 2026-08-12 cron来源标记：统一 hermes 体系 jobs.append(j) except: pass return jobs def load_system_crontab_tasks(): """扫描系统 crontab 里 MoFin 相关任务（2026-08-12 老莫要求：显著提示违规系统 cron）。 开发原则：自动任务必须部署在 hermes（position-analyst），禁止系统 crontab。 返回含 MoFin/profile-scripts 路径的 crontab 条目，source='system_crontab'（违规警告）。""" import subprocess as _sp tasks = [] try: out = _sp.run(["crontab", "-l"], capture_output=True, text=True, timeout=10).stdout for line in out.splitlines(): line = line.strip() if not line or line.startswith("#"): continue # 只关心 MoFin 相关（含 MoFin 或 profile-scripts 路径） if "MoFin" not in line and "profile-scripts" not in line: continue # 提取调度（前 5 个字段或 @reboot）和命令/脚本 script = "" m = line.split("python3")[-1].split("python")[-1].strip() import re as _re sm = _re.search(r'([\w./-]*profile-scripts/([\w.-]+\.py)|([\w.-]+\.py))', line) if sm: script = sm.group(2) or sm.group(3) or "" sched = " ".join(line.split()[:5]) if not line.startswith("@") else line.split()[0] tasks.append({ "name": f"[系统cron] {script or line[:40]}", "script": script, "schedule": sched, "status": "unknown", "last_run": "", "profile": "system", "source": "system_crontab", # 违规：应迁移到 hermes "no_agent": True, "type": "no_agent", "enabled": True, }) except Exception: pass return tasks 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股/港股/美股指数对）", "mofin_news": "写入外部财经常规新闻采集结果", }, "readers": { "macro_signal_consumer": "读取原始宏观新闻和信号，判定风险等级并生成建议", "system_audit": "读取信号表行数/更新时间，审计数据管道是否畅通", "intraday_health_check": "读取最新信号，检查盘中是否有新的风险信号到达", "server": "读取信号数据供Web Dashboard展示", }, }, "holdings": { "summary": "当前持仓表，是系统最核心的数据表之一。import_holding_xls从券商文件导入持仓，mofin_db在价格刷新时更新市值。下游脚本读取持仓做策略分析和推送。", "writers": { "mofin_db": "写入price_monitor刷新后的持仓最新市值（通过write_holdings_batch）", "import_holding_xls": "从券商holding.xls导入最新持仓数量/成本/市值", }, "readers": { "price_monitor": "高频价格监控+操作区间触发重评", "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()读取策略数据", "price_monitor": "检测操作区间(买入/止损/止盈)并触发重评", }, }, "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": { "market_regime": "温区判断(替代原detect_scenario)", "divergence_detector": "读取多市场指数数据做背离检测", "system_audit": "审计数据采集是否正常", }, }, "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": { "strategy_effectiveness": "每周对账时写入对账结果", }, "readers": { "strategy_effectiveness": "读取历史建议做对账", "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": { }, "readers": { "server": "读取扫描状态供Dashboard展示", }, }, "state_meta": { "summary": "状态元数据表，记录各服务的状态追踪信息（如扫描偏移量/最新处理ID）。", "writers": { }, "readers": { }, }, } 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", "children": [ {"label": "MoFin盘前中监控", "status": "ok"}, {"label": "MoFin午后监控", "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", "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"], } 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" wcnt, rcnt = len(writers), len(readers) # 2026-08-19 多写方风险（老莫：数据篡改风险的根源） # 写方>=3 → 高风险（多个模块竞争写同一实体，可能互相覆写/篡改） if wcnt >= 3: risk = "high" risk_note = f"⚠️多写方{wcnt}(竞争写,可能篡改)" elif wcnt == 0: risk = "orphan" risk_note = "无写方(死实体)" elif rcnt == 0: risk = "write_only" risk_note = "写无读(幽灵)" elif wcnt == 2: risk = "watch" risk_note = f"{wcnt}写方(需关注)" else: risk = "ok" risk_note = "" entities.append({ "name": tname, "desc": TABLES_DESC.get(tname, ""), "layer": TABLE_LAYER.get(tname, "biz"), # 2026-08-12 三层架构层归属 "rows": cnt, "readers": readers[:10], "writers": writers[:10], "writer_count": wcnt, "reader_count": rcnt, "multi_writer": wcnt >= 3, "risk": risk, "risk_note": risk_note, "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" or wcnt >= 3, "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) # 输出缺失检测（2026-08-10）：hermes失败时last_status仍ok， # 检查"今天应按调度运行但输出目录无当日文件"的LLM cron output_missing = False output_missing_reason = "" if not no_agent and j.get("id") and j.get("enabled", True): try: _job_id = j.get("id") _out_dir = Path('/home/hmo/.hermes/profiles/position-analyst/cron/output') / _job_id if _out_dir.exists(): _today = now.strftime("%Y-%m-%d") _today_files = list(_out_dir.glob(f"*{_today}*")) # 修正2026-08-10：不要求 last_run 今天（失败后 last_run 停在旧日期） # 只要求：工作日 + 调度含工作日 + 今天无输出 _sched = str(j.get("schedule", "")) _weekday_only = ("1-5" in _sched or "mon" in _sched.lower()) _is_weekday = now.weekday() < 5 # 如果 job 今天已运行(last_run 今天)，不报 output_missing（SILENT 是正常的） _last_run_date = str(j.get('last_run_at', ''))[:10] _ran_today = (_last_run_date == _today) if not _today_files and (_weekday_only and _is_weekday) and not _ran_today: output_missing = True output_missing_reason = f"今天({_today})工作日应按调度运行但输出目录无文件(last_run={str(j.get('last_run_at',''))[:16]})" except Exception: pass pipelines.append({ "name": name, "type": "no_agent" if no_agent else "LLM", "script": script, "layer": _pipeline_layer(script, name), # 2026-08-12 三层架构层归属 "source": j.get("source", "hermes"), # 2026-08-12 cron来源（hermes 统一体系） "schedule": schedule, "status": status, "last_run": last_run, "profile": j.get("profile", "?"), "delivery": j.get("delivery", "broadcast"), # 消息通道：broadcast/xmpp/both "output_missing": output_missing, "output_missing_reason": output_missing_reason, }) if output_missing: feature_tree["status"] = "fail" feature_tree.setdefault("warn_detail", {})[name] = output_missing_reason # ── 2026-08-12 系统 crontab 违规任务（老莫要求：显著提示。开发原则：自动任务统一 hermes，禁止系统 crontab）── _sys_tasks = load_system_crontab_tasks() for _t in _sys_tasks: _t["layer"] = _pipeline_layer(_t.get("script", ""), _t.get("name", "")) _t["output_missing"] = False _t["output_missing_reason"] = "" pipelines.append(_t) if _sys_tasks: feature_tree.setdefault("warn_detail", {})["system_crontab"] = ( f"发现 {len(_sys_tasks)} 个系统 crontab 部署的 MoFin 任务（违规：应迁移到 hermes）") # ── 自检体系状态（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: pass 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'] = [] # ── LLM端点健康检查（2026-08-10新增：监控LLM调用失败，防静默故障）── # 背景：opencode.ai deepseek-v4-flash 端点故障时，gateway进程活着但LLM全挂， # 监控只查端口/进程发现不了。扫描hermes错误日志弥补。 llm_health = {"ok": True, "errors_recent": [], "scan_window_min": 60} _LLM_ERR_PATTERNS = [ "RemoteProtocolError", "Stream stale", "empty stream", "peer closed connection", "API call failed", # 2026-08-11 修复：移除 "finish_reason"——它是正常完成标志（INFO级别 Turn ended）， # 把它当错误导致连续误报"LLM端点故障"。真正的失败才报（RemoteProtocolError/Stream stale等）。 ] try: _err_logs = [ Path('/home/hmo/.hermes/profiles/position-analyst/logs/errors.log'), Path('/home/hmo/.hermes/profiles/position-analyst/logs/agent.log'), ] _recent_errs = [] _scan_cutoff = (now.timestamp() - 60 * llm_health["scan_window_min"]) for _lf in _err_logs: if not _lf.exists(): continue try: _lines = _lf.read_text(encoding='utf-8', errors='ignore').splitlines()[-5000:] for _line in _lines: # 提取时间戳（行首 YYYY-MM-DD HH:MM:SS） if len(_line) < 19: continue try: _ts = datetime.strptime(_line[:19], "%Y-%m-%d %H:%M:%S") except Exception: continue if _ts.timestamp() < _scan_cutoff: continue if any(_p in _line for _p in _LLM_ERR_PATTERNS): # 2026-08-25 只数最终失败：hermes 一次抖动打3条WARNING(attempt1/2/3)， # 重试成功不算故障——attempt 1/2 的"API call failed"和"Retrying"跳过 if "Retrying API call" in _line: continue import re as _re2 _att = _re2.search(r"attempt (\d+)/(\d+)", _line) if _att and _att.group(1) != _att.group(2): continue # 非最终次失败=重试中，跳过 _recent_errs.append({ "log": _lf.name, "time": _ts.strftime("%H:%M"), "msg": _line[20:180], }) except Exception: pass # 去重+限量 _seen = set() _uniq = [] for _e in _recent_errs: _k = _e["time"] + _e["msg"][:50] if _k not in _seen: _seen.add(_k) _uniq.append(_e) llm_health["errors_recent"] = _uniq[:10] llm_health["error_count"] = len(_uniq) llm_health["ok"] = len(_uniq) == 0 self_check["llm_health"] = llm_health # 若LLM故障，功能树根节点标 fail if not llm_health["ok"]: feature_tree["status"] = "fail" # 失败告警（2026-08-10新增）：LLM故障推XMPP，不只标红 try: import sys as _sys if '/home/hmo/.hermes/profiles/position-analyst/scripts' not in _sys.path: _sys.path.insert(0, '/home/hmo/.hermes/profiles/position-analyst/scripts') from alert_helper import notify, INFO _first = llm_health["errors_recent"][0] if llm_health["errors_recent"] else {} notify("LLM端点故障", f"最近{llm_health['scan_window_min']}分钟{llm_health['error_count']}次调用失败" f"（{_first.get('log','?')} {_first.get('time','?')}: {_first.get('msg','')[:80]}）。" f"gateway进程可能正常但LLM服务不可用，请检查opencode.ai端点。", INFO) except Exception as _ae: print(f"[LLM_MON] 告警发送失败: {_ae}", file=sys.stderr) feature_tree.setdefault("warn_detail", {})["llm"] = ( f"LLM端点最近{llm_health['scan_window_min']}分钟{llm_health['error_count']}次错误" f"（{llm_health['errors_recent'][0]['log'] if llm_health['errors_recent'] else '?'}）") except Exception as _e: self_check["llm_health"] = {"ok": True, "error": str(_e)[:100]} # ── 2026-08-12 三层架构健康检查（采集/加工/使用三层各自健康，老莫定）── def _layer_health(): health = {} try: _c = sqlite3.connect("/home/hmo/MoFin/data/mofin.db", timeout=10) # 采集层健康：原始数据新鲜度 + 覆盖范围 collect = {} for tname, sql in { "stock_daily": "SELECT MAX(date), COUNT(DISTINCT code) FROM stock_daily WHERE length(code)=6", "stock_news": "SELECT MAX(date), COUNT(*) FROM stock_news", "stock_fundamentals": "SELECT MAX(updated_at), COUNT(*) FROM stock_fundamentals", "sector_index_daily": "SELECT MAX(date), COUNT(DISTINCT sector) FROM sector_index_daily", }.items(): try: r = _c.execute(sql).fetchone() collect[tname] = {"latest": r[0], "count": r[1]} except Exception as e: collect[tname] = {"latest": None, "count": 0, "error": str(e)[:50]} health["collect"] = collect # 加工层健康：衍生指标新鲜度 + 覆盖范围 process = {} for tname, sql in { "stock_indicators": "SELECT MAX(date), COUNT(DISTINCT code) FROM stock_indicators", "market_indicators": "SELECT MAX(date), COUNT(*) FROM market_indicators", }.items(): try: r = _c.execute(sql).fetchone() process[tname] = {"latest": r[0], "count": r[1]} except Exception as e: process[tname] = {"latest": None, "count": 0, "error": str(e)[:50]} health["process"] = process # 使用层健康：业务数据 + 候选管道 r1 = _c.execute("SELECT COUNT(*), MAX(created_at) FROM candidates").fetchone() r2 = _c.execute("SELECT COUNT(*) FROM holding_strategies WHERE status='active'").fetchone() health["use"] = { "candidates": {"count": r1[0], "latest": r1[1]}, "active_strategies": r2[0], } _c.close() except Exception as e: health["error"] = str(e)[:100] # 任务状态按层汇总（pipelines 已有 layer 字段） layer_tasks = {} for p in pipelines: layer = p.get("layer", "use") layer_tasks.setdefault(layer, {"ok": 0, "error": 0, "total": 0}) layer_tasks[layer]["total"] += 1 if p.get("status") == "ok": layer_tasks[layer]["ok"] += 1 elif p.get("status") in ("error", "fail"): layer_tasks[layer]["error"] += 1 health["tasks"] = layer_tasks return health self_check["layers_health"] = _layer_health() # ── 写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()