feat: F健康三层改造step3——自检体系加三层架构健康卡片(采集/加工/使用各层数据新鲜度+覆盖+任务状态)

deploy/profile-scripts/mofin_health.py

@@ -1144,6 +1144,62 @@ def build_report():
     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(DB_PATH)
+            # 采集层健康：原始数据新鲜度 + 覆盖范围
+            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"),