fix(data): 审计终局 P1 监控四修——①kind 归并(绿班 passed 恒收 rows key, rows 黄悬死实案全 kind 自愈) ②row-band 豁免未确认日 0 行空文件(block_trade band 下限误黄) ③expected_weekly 周五锚 confirmed 过滤(假周五 0925/1002 免疫; 空集降级不过滤防静默假绿, factor 交界提醒) ④health 拆 checks_overall/alerts_overall(防晨检被旧红回声误导航) +闸门白名单补 margin/margin_sse(假 T-1 踩坑); TDD 6 新测试 611 全绿+spec §15.4/§20.15.2 登记 [vps]

Co-Authored-By: Claude Code <noreply@anthropic.com>

scripts/data_platform/data_monitor.py

@@ -83,12 +83,16 @@ def check_entry(e, ctx):
                          "unconfirmed_day": exp not in ctx["confirmed_days"]})
         rows = chk.parquet_rows(got[1]) if got else None
         lo, hi = e["row_band"]
-        if rows is None or not (lo <= rows <= hi):
+        # 未确认日 0 行空文件(节假日闸门产物)不告 row-band (block_trade
+        # band 下限>0 对 0925 空文件误黄实案); 交易日 0 行仍黄保持感知
+        if not (rows == 0 and exp not in ctx["confirmed_days"]) \
+                and (rows is None or not (lo <= rows <= hi)):
             return done("rows", "yellow", f"{lo}~{hi}", rows)
         return done("stale", "green", exp, latest)
 
     if kind == "weekly_friday":
-        exp = chk.expected_weekly(ctx["days"], ctx["now"], e.get("ready", "06:00"))
+        exp = chk.expected_weekly(ctx["days"], ctx["now"], e.get("ready", "06:00"),
+                                  confirmed=ctx["confirmed_days"])
         got = chk.latest_stem(os.path.join(root, e["subdir"]), e["pattern"])
         latest = got[0].split("_")[0] if got else None
         if exp is None:
@@ -99,7 +103,8 @@ def check_entry(e, ctx):
             return done("stale", sev, exp, latest)
         lo, hi = e["row_band"]
         rows = chk.parquet_rows(got[1]) if got else None
-        if rows is None or not (lo <= rows <= hi):
+        if not (rows == 0 and exp not in ctx["confirmed_days"]) \
+                and (rows is None or not (lo <= rows <= hi)):
             return done("rows", "yellow", f"{lo}~{hi}", rows)
         return done("stale", "green", exp, latest)
 
@@ -412,6 +417,11 @@ def main(argv=None):
         checks.append(r)
         if r["status"] == "green":
             passed.append(r["key"])
+            # kind 归并修复: 绿路径统一返 stale key, rows key 只在黄时存在
+            # → rows 类黄一旦发生绿班永不 resolve (sw_industry/block_trade
+            # 悬死实案); 绿班恒收 rows key(幂等, 无告警时 resolve 零命中)
+            if e["kind"] in ("per_date", "weekly_friday", "month_bucket"):
+                passed.append(f"data-{e['name']}-rows")
         else:
             kindword = r["key"].rsplit("-", 1)[-1]  # stale/rows/verdict/exists
             events.append(_event(
@@ -471,11 +481,18 @@ def main(argv=None):
               for s in ("green", "yellow", "red")}
     red_events = sum(1 for ev in events if ev["severity"] == "red")
     yellow_events = sum(1 for ev in events if ev["severity"] == "yellow")
+    # 拆分防晨检误导航(🟢-7): checks 绿而 alerts 层 26h 窗旧红回声时,
+    # overall=red 会引导去查已自愈项; 两字段各自可读, overall 保兼容取最坏
+    checks_overall = "red" if counts["red"] else (
+        "yellow" if counts["yellow"] else "green")
+    alerts_overall = "red" if red_events else (
+        "yellow" if yellow_events else "green")
     overall = "red" if (counts["red"] or red_events) else (
         "yellow" if (counts["yellow"] or yellow_events) else "green")
     health = {"date": now.date().isoformat(), "shift": args.shift, "host": "vps",
               "generated_at": now.isoformat(timespec="seconds"),
-              "overall": overall, "counts": counts,
+              "overall": overall, "checks_overall": checks_overall,
+              "alerts_overall": alerts_overall, "counts": counts,
               "alerts_emitted": len(events), "checks": checks}
     write_health(args.data_root, health)
     print(f"[data-monitor] {args.shift} overall={overall} checks={len(checks)}"