fix: 管道审计覆盖6条关键数据流+holding_strategies+decisions.json

2026-07-08 11:13:59 +08:00
parent 9f5bc8b004
commit 8a77b43ea1
1 changed files with 64 additions and 11 deletions
@@ -158,21 +158,74 @@ def audit_cache():
        log_issue("编译缓存", "LOW", f"检查失败: {e}")


-# ── 6. 数据管道审计 ──
+# ── 6. 数据管道审计（端到端，逐条trace） ──
 def audit_pipeline():
-    # 检查DB市场数据是否今天更新
+    """遍历所有关键数据管道，检查生产者→存储→消费者链路是否完整"""
+    today = datetime.now().strftime("%Y-%m-%d")
+    conn = sqlite3.connect(str(DATA_DIR / "mofin.db"))
+
+    pipelines = [
+        # 管道名, 生产者, 存储位置, 检查SQL/文件, 新鲜度阈值(天)
+        ("价格数据",  "price_monitor(每2分)", "live_prices.updated_at",
+         "SELECT MAX(updated_at) FROM live_prices", 0.02),   # 30分钟内
+        ("宏观上下文", "refresh_macro_context(每30分)", "macro_context_log.created_at",
+         "SELECT MAX(created_at) FROM macro_context_log", 1), # 1天内
+        ("市场快照",  "market_watch(每10分)", "market_snapshots.created_at",
+         "SELECT MAX(created_at) FROM market_snapshots", 1),
+        ("策略评估",  "reassess_with_context", "strategy_evaluations.created_at",
+         "SELECT MAX(created_at) FROM strategy_evaluations", 2),
+        ("原始新闻",  "macro_context_collector", "macro_raw_news.fetched_at",
+         "SELECT MAX(fetched_at) FROM macro_raw_news", 1),
+        ("风险信号",  "macro_context_collector", "signal_news.created_at",
+         "SELECT MAX(created_at) FROM signal_news", 2),
+    ]
+
+    for name, producer, storage, sql, max_days in pipelines:
+        try:
+            row = conn.execute(sql).fetchone()
+            if not row or not row[0]:
+                log_issue("数据管道", "HIGH", f"{name}: 无数据 ({producer}→{storage})",
+                          fix=f"检查{producer}是否正确运行")
+                continue
+            latest = row[0][:19] if len(row[0]) > 19 else row[0]
+            try:
+                dt = datetime.fromisoformat(latest) if isinstance(latest, str) else latest
+                days_old = (datetime.now() - dt).total_seconds() / 86400
+            except:
+                days_old = 999
+            if days_old > max_days:
+                log_issue("数据管道", "HIGH",
+                          f"{name}: {days_old:.0f}天未更新(阈值{max_days}天) 最后{latest} ({producer}→{storage})",
+                          fix=f"检查{producer}输出和{storage}写入逻辑")
+            else:
+                log_ok("数据管道", f"{name} {days_old*24:.0f}小时前更新 → OK")
+        except Exception as e:
+            log_issue("数据管道", "HIGH", f"{name} 检查失败: {e}")
+
+    # 特殊检查：holding_strategies和decisions.json是否为空
    try:
-        conn = sqlite3.connect(str(DATA_DIR / "mofin.db"))
-        row = conn.execute(
-            "SELECT created_at FROM market_snapshots ORDER BY created_at DESC LIMIT 1"
-        ).fetchone()
-        conn.close()
-        if row and row[0][:10] == datetime.now().strftime("%Y-%m-%d"):
-            log_ok("数据管道", f"市场数据今天更新({row[0]})")
+        hs = conn.execute("SELECT COUNT(*) FROM holding_strategies").fetchone()[0]
+        if hs == 0:
+            log_issue("数据管道", "HIGH", "holding_strategies表为空(策略评估产出未写入)", fix="检查holding_strategies写入逻辑")
        else:
-            log_issue("数据管道", "HIGH", f"市场数据未更新(DB), 最后{row[0] if row else '无数据'}")
+            log_ok("数据管道", f"holding_strategies {hs}条")
    except Exception as e:
-        log_issue("数据管道", "HIGH", f"DB检查失败: {e}")
+        log_issue("数据管道", "HIGH", f"holding_strategies检查失败: {e}")
+
+    conn.close()
+
+    # 检查decisions.json文件
+    try:
+        import json
+        with open(WEB_DATA / "decisions.json") as f:
+            dec = json.load(f)
+        cnt = len(dec.get("decisions", []))
+        if cnt < 5:
+            log_issue("数据管道", "HIGH", f"decisions.json仅{cnt}条决策(异常)", fix="检查decisions.json写入逻辑")
+        else:
+            log_ok("数据管道", f"decisions.json {cnt}条决策")
+    except Exception as e:
+        log_issue("数据管道", "HIGH", f"decisions.json读取失败: {e}")


 # ── 7. 系统服务 ──