feat: F健康三层架构改造step1——entities/pipelines加layer字段(TABLE_LAYER表归属+PIPELINE_LAYER任务归属,采集/加工/使用三层)

deploy/profile-scripts/mofin_health.py

@@ -73,6 +73,52 @@ JSON_DESC = {
     "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():
@@ -856,6 +902,7 @@ def build_report():
         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],
@@ -970,6 +1017,7 @@ def build_report():
             "name": name,
             "type": "no_agent" if no_agent else "LLM",
             "script": script,
+            "layer": _pipeline_layer(script, name),  # 2026-08-12 三层架构层归属
             "schedule": schedule,
             "status": status,
             "last_run": last_run,