1276 lines
60 KiB
Python
1276 lines
60 KiB
Python
#!/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": "健康监控数据",
|
||
}
|
||
|
||
# ── 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",
|
||
"stale_detector", "watchlist_auto_exit", "branch_", "meta_growth", "prune_branches",
|
||
"market_insight", "strategy_evaluator", "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股/港股/美股指数对）",
|
||
"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, ""),
|
||
"layer": TABLE_LAYER.get(tname, "biz"), # 2026-08-12 三层架构层归属
|
||
"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)
|
||
# 输出缺失检测（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
|
||
if not _today_files and (_weekday_only and _is_weekday):
|
||
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", "?"),
|
||
"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:
|
||
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'] = []
|
||
|
||
# ── 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):
|
||
_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()
|