fix(monitor): 0928 开盘巡检三修——口径噪声根治 ①row-band 豁免改按文件日期(latest)判: lag 型 per_date(margin/margin_sse T-1) exp=ok[-1-lag]=确认日, 旧条件对闸门空写非确认日 0 行文件永不豁免(0927 margin_sse 20260925 空文件假黄实案), weekly 同型 ②schtask weekday 限 26h→72h: weekday 任务周五晚→周一晚班次跨度 69h, 26h 限周一早班必假黄(0928 ak-events-retry 57h/qmt-probe0915 71h 实案), 真死>72h 仍黄+next_run=N/A red 兜底 ③nas-dmsk mtime 80h→季频披露锚: 五域季频披露驱动 mtime 无结构意义(Q2 水位 06-30 停滞正常), 改最新 dt= 分区≥最近已过披露截止季末; 另 bs-daily rc=1(eod 腿 baostock 网络错)与 ak-weekly rc=1(share_capital failed)为正确告警不改; spec §20.15.2 同步; tests 619 绿(+4 新用例) [vps] [nas]

docs/superpowers/specs/2026-07-21-data-source-fusion-design.md

@@ -1789,7 +1789,7 @@ schema 特殊低估交集教训）；限速纪律沿 §19.9 采集纪律表；VP
 
 分类学四层：**调度层**（任务在岗+班次延迟）、**过程层**（统计行分诊+熔断连累检测——周班某类型熔断会跳过后续整批类型）、**结果层**（注册表逐类型新鲜度+行数带宽+交易日历对齐防节假日误报）、**衍生层**（前夜红项自愈回看+xcheck findings 趋势消费[ NAS 侧 ]+磁盘水位）。
 
-**结果层缺失分级（定案，09-26 中秋反转案）**：交易日历=确认集（dbbardata 000300 实证）∪近 15 日工作日假定集（防 dbbardata 晚间源滞后塌全体期望）。缺失遇未确认日=黄（节假日/合并窗/源滞后三态容忍），**不因时间升红**——2026-09-25 周五中秋休市实案（9-25~9-27 连休）：时间基线升级（越过次日 ready 即红）会在 0926~0927 晚把全 per_date 域连坐误红，三源（baostock/sina/EM）同缺一日本就是休市形态而非源滞后；确认日缺失=红（跨域互证=真是交易日）。真漏采兜底=确认集推进（市场恢复交易后 exp 随 confirmed 前移，latest 落后已确认 exp → 确认日红接手）+调度层 schtask 在岗+过程层统计行，纵深分工。**09-27 审计终局 P1 四修**：①kind 归并修复——绿路径统一返 stale key 而 rows key 只在黄时存在，rows 类黄绿班永不 resolve（sw_industry/block_trade 悬死实案）→ 绿班 passed 恒收 `data-{name}-rows`（全 kind）；②row-band 豁免——未确认日 0 行空文件（节假日闸门产物）不告 rows（block_trade band 下限>0 误黄实案），交易日 0 行仍黄保持感知；③weekly 锚 confirmed 过滤——`expected_weekly` 周五锚须在确认集（假定集假周五 0925/1002 免疫，锚自动回上一真周五），**confirmed 空集降级不过滤防 exp=None 静默假绿**；④health 拆 `checks_overall`/`alerts_overall`（overall 保兼容取最坏）——防 checks 全绿而 alerts 层 26h 窗旧红回声时晨检被 overall=red 误导航。
+**结果层缺失分级（定案，09-26 中秋反转案）**：交易日历=确认集（dbbardata 000300 实证）∪近 15 日工作日假定集（防 dbbardata 晚间源滞后塌全体期望）。缺失遇未确认日=黄（节假日/合并窗/源滞后三态容忍），**不因时间升红**——2026-09-25 周五中秋休市实案（9-25~9-27 连休）：时间基线升级（越过次日 ready 即红）会在 0926~0927 晚把全 per_date 域连坐误红，三源（baostock/sina/EM）同缺一日本就是休市形态而非源滞后；确认日缺失=红（跨域互证=真是交易日）。真漏采兜底=确认集推进（市场恢复交易后 exp 随 confirmed 前移，latest 落后已确认 exp → 确认日红接手）+调度层 schtask 在岗+过程层统计行，纵深分工。**09-27 审计终局 P1 四修**：①kind 归并修复——绿路径统一返 stale key 而 rows key 只在黄时存在，rows 类黄绿班永不 resolve（sw_industry/block_trade 悬死实案）→ 绿班 passed 恒收 `data-{name}-rows`（全 kind）；②row-band 豁免——未确认日 0 行空文件（节假日闸门产物）不告 rows（block_trade band 下限>0 误黄实案），交易日 0 行仍黄保持感知；③weekly 锚 confirmed 过滤——`expected_weekly` 周五锚须在确认集（假定集假周五 0925/1002 免疫，锚自动回上一真周五），**confirmed 空集降级不过滤防 exp=None 静默假绿**；④health 拆 `checks_overall`/`alerts_overall`（overall 保兼容取最坏）——防 checks 全绿而 alerts 层 26h 窗旧红回声时晨检被 overall=red 误导航。**09-28 开盘巡检三修（monitor 口径噪声根治）**：①row-band 豁免改按**文件日期（latest）**判而非 exp——lag 型 per_date（margin/margin_sse T-1，registry `lag=1`）的 exp=ok[-1-lag]=确认日，旧豁免条件对闸门空写的非确认日 0 行文件（0927 `20260925_margin_sse.parquet` 假黄实案）永不成立；weekly 同型（假周五空写文件）。②schtask weekday 限 26h→**72h**——weekday 任务（ak-events-retry 20:30/qmt-probe0915 09:15）周五晚跑完→周一晚班次跨度 69h，26h 限周一早班必假黄（0928 实案 57h/71h）；真死 >72h 仍黄+next_run=N/A→red 兜底。③NAS `nas-dmsk` mtime 80h→**季频披露锚**——dmsk 五域季频披露驱动（无新披露不写盘，Q2 报告期水位 06-30 停滞属正常节奏），改「最新 dt= 分区 ≥ 最近已过法定披露截止的季末」（Q1→04-30/半年报→09-30/三季报→10-31/年报→次年 04-30；五域取 max，三表法定必披露为门槛主体，express/forecast 条件披露只助推 latest 不抬门槛）。另：bs-daily `last_result=1`（0927 eod 腿 baostock 网络错误 10002007，fund 腿独立照跑、窗口无新交易日无数据损失）与 ak-weekly `rc=1`（share_capital failed 群）为**正确告警**非口径问题，不改。
 
 #### 20.15.3 两班调度与产物
 

scripts/data_platform/data_monitor.py

@@ -84,8 +84,11 @@ def check_entry(e, ctx):
         rows = chk.parquet_rows(got[1]) if got else None
         lo, hi = e["row_band"]
         # 未确认日 0 行空文件(节假日闸门产物)不告 row-band (block_trade
-        # band 下限>0 对 0925 空文件误黄实案); 交易日 0 行仍黄保持感知
-        if not (rows == 0 and exp not in ctx["confirmed_days"]) \
+        # band 下限>0 对 0925 空文件误黄实案); 交易日 0 行仍黄保持感知.
+        # 0928 修正: 豁免按文件日期(latest)判而非 exp——lag 型(margin/
+        # margin_sse T-1) exp=ok[-1-lag]=确认日, 旧条件对闸门空写的非
+        # 确认日空文件永不豁免(0927 margin_sse 20260925 空文件假黄实案)
+        if not (rows == 0 and latest 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)
@@ -103,7 +106,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 not (rows == 0 and exp not in ctx["confirmed_days"]) \
+        # 同 per_date 0928 修正: 豁免按文件日期判(假周五闸门空写 0 行文件)
+        if not (rows == 0 and latest 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)
@@ -287,7 +291,9 @@ def log_only_results(entries, stats, sigs, today):
 def schtask_results(snapshot, now):
     results = []
     today = now.date().isoformat().replace("-", "")
-    limits = {"daily": 26 if now.weekday() < 5 else 80, "weekday": 26,
+    # weekday=72h: 周五晚班→周一晚班次跨度 69h+容差(0928 早班 ak-events-retry
+    # 57h/qmt-probe0915 71h 假黄实案); 真死 >72h 仍黄, next_run=N/A→red 兜底
+    limits = {"daily": 26 if now.weekday() < 5 else 80, "weekday": 72,
               "weekly": 8 * 24, "monthly": 40 * 24, "always": None}
     for name, cadence in reg.SCHTASKS:
         key = f"infra-task-{name}"

scripts/nas_sync/nas_monitor_check.py

@@ -79,6 +79,35 @@ def _log_has_today_fail(log_path, now):
                for line in text.splitlines()[-200:])
 
 
+# dmsk 季频锚: 各季末→法定披露截止(年报次年 04-30/半年报 09-30/三季 10-31/
+# 一季 04-30); 截止未过的季度只要求上一季全量在(在披露的季度水位可仍停上季末)
+_QUARTER_DEADLINES = {(3, 31): (4, 30), (6, 30): (9, 30), (9, 30): (10, 31)}
+
+
+def _dmsk_latest_partition(root):
+    """五域(balance/income/cashflow/express/forecast)最新 dt= 分区(ISO)或 None."""
+    best = None
+    for p in glob.glob(os.path.join(root, "*", "dt=*")):
+        d = p.rsplit("dt=", 1)[-1]
+        if best is None or d > best:
+            best = d
+    return best
+
+
+def _dmsk_min_expected(now):
+    """最近一个「法定披露截止已过」的季末(ISO)或 None."""
+    for back in range(0, 400):
+        d = now.date() - dt.timedelta(days=back)
+        key = (d.month, d.day)
+        if key in _QUARTER_DEADLINES:
+            dl_m, dl_d = _QUARTER_DEADLINES[key]
+            if dt.date(d.year, dl_m, dl_d) <= now.date():
+                return d.isoformat()
+        elif key == (12, 31) and dt.date(d.year + 1, 4, 30) <= now.date():
+            return d.isoformat()
+    return None
+
+
 def build_checks(now):
     """返回 (checks 列表[health 用], events 列表[非绿项转告警])."""
     checks = []
@@ -93,11 +122,15 @@ def build_checks(now):
         add(f"nas-corpus_{lane}", f"corpus {lane} 新鲜",
             freshness(age, 34), {"age_h": round(age, 1) if age is not None else None})
 
-    # dmsk 三表 dt 分区(07:10 班)——事件驱动域: 无新披露不写盘(09-26 实证两日
-    # lane exit=0 而 mtime 48h+), mtime 弱信号, 阈值放宽到 80h(漏两个日窗才黄)
-    age = dir_age_hours(DMSK_ROOT, now)
-    add("nas-dmsk", "dmsk 三表分区新鲜", freshness(age, 80),
-        {"age_h": round(age, 1) if age is not None else None})
+    # dmsk 三表 dt 分区(07:10 班)——季频披露驱动域: 无新披露不写盘, mtime 无
+    # 结构意义(09-27 实证 Q2 报告期水位 06-30 停滞, 80h mtime 必假黄). 0928
+    # 改季频锚: 最新分区 dt ≥ 最近已过披露截止的季末; 五域取 max(三表法定
+    # 必披露, express/forecast 条件披露只助推 latest 不抬门槛)
+    latest_dt = _dmsk_latest_partition(DMSK_ROOT)
+    min_exp = _dmsk_min_expected(now)
+    add("nas-dmsk", "dmsk 三表季频水位",
+        "green" if (latest_dt and min_exp and latest_dt >= min_exp) else "yellow",
+        {"latest_dt": latest_dt, "min_expected": min_exp})
 
     # hot_rank: 昨交易日文件 + .pushed marker + 当日 cron.log FAIL 行
     want_day = last_weekday(now.date())

tests/data_platform/test_data_monitor.py

@@ -377,3 +377,61 @@ def test_schtask_layer_parses_snapshot(tree, tmp_path, monkeypatch):
     assert keys["infra-task-sanguo-ak-weekly"]["severity"] == "yellow"  # 周班 15d 前>8d 限
     missing = [e for e in events if e["check_key"] == "infra-task-sanguo-bs-daily"]
     assert missing and missing[0]["severity"] == "red"    # 快照缺任务
+
+
+def _drop_confirmed_0925(root):
+    """把 09-25 从确认集剔除(中秋休市形): dbbardata 删该日 + 保留假定集语义."""
+    conn = sqlite3.connect(root / "quant_trading.db")
+    conn.execute("DELETE FROM dbbardata WHERE datetime='2026-09-25'")
+    conn.commit()
+    conn.close()
+
+
+def test_gate_empty_lag_file_on_unconfirmed_day_skips_row_band(tree, tmp_path):
+    """0928 修正: lag 型 per_date(margin/margin_sse, T-1)闸门空写非确认日
+    0 行文件, 豁免须按文件日期(latest)判——旧条件按 exp(=ok[-1-lag]=确认日)
+    永不豁免, 0927 margin_sse 20260925 空文件假黄实案."""
+    _drop_confirmed_0925(tree)
+    d = tree / "static" / "margin_sse"
+    for f in d.glob("*.parquet"):
+        f.unlink()
+    _mk_parquet(d / "20260925_margin_sse.parquet", 0)
+    db = tmp_path / "alerts.db"
+    assert _run(tree, db) == 0
+    conn = store.connect(str(db))
+    rows, _ = store.list_alerts(conn)
+    assert not any(r["check_key"] == "data-margin_sse-rows" for r in rows)
+
+
+def test_gate_empty_friday_file_on_unconfirmed_day_skips_row_band(tree, tmp_path):
+    """weekly 同型: 假周五(0925 中秋)闸门空写 0 行文件, 豁免按文件日期判."""
+    _drop_confirmed_0925(tree)
+    d = tree / "static" / "pledge_ratio"
+    for f in d.glob("*.parquet"):
+        f.unlink()
+    _mk_parquet(d / "20260925_pledge_ratio.parquet", 0)
+    db = tmp_path / "alerts.db"
+    assert _run(tree, db) == 0
+    conn = store.connect(str(db))
+    rows, _ = store.list_alerts(conn)
+    assert not any(r["check_key"] == "data-pledge_ratio-rows" for r in rows)
+
+
+def test_weekday_task_survives_weekend_gap():
+    """weekday 任务周五晚跑完→周一早班查(57~71h)不再误黄(0928 早班
+    ak-events-retry 57h/qmt-probe0915 71h 实案); 真死 >72h 仍黄."""
+    snap = {"sanguo-ak-events-retry": {"next_run": "2026/9/28 20:30:00",
+                                       "last_run": "2026/9/25 23:30:00",
+                                       "last_result": "0"},
+            "sanguo-qmt-probe0915": {"next_run": "2026/9/28 9:15:00",
+                                     "last_run": "2026/9/25 9:15:00",
+                                     "last_result": "0"}}
+    now = dt.datetime(2026, 9, 28, 8, 5)
+    res = dm.schtask_results(snap, now)
+    keys = {e["check_key"] for e in res}
+    assert "infra-task-sanguo-ak-events-retry" not in keys   # 57h<72h
+    assert "infra-task-sanguo-qmt-probe0915" not in keys     # 71h<72h
+    snap["sanguo-ak-events-retry"]["last_run"] = "2026/9/23 23:30:00"  # 109h
+    res = dm.schtask_results(snap, now)
+    assert any(e["check_key"] == "infra-task-sanguo-ak-events-retry"
+               for e in res)

tests/data_platform/test_nas_monitor_check.py

@@ -52,3 +52,31 @@ def test_log_has_today_fail(tmp_path):
     ok_log = tmp_path / "ok.log"
     ok_log.write_text(f"=== {day} push ok ===\n", encoding="utf-8")
     assert nmc._log_has_today_fail(str(ok_log), dt.datetime(2026, 9, 25, 16)) is False
+
+
+def _mk_dmsk_partition(root, dt_str):
+    p = root / "balance" / f"dt={dt_str}"
+    p.mkdir(parents=True, exist_ok=True)
+    (p / "part-0.parquet").touch()
+
+
+def test_dmsk_quarterly_cadence(tmp_path, monkeypatch):
+    """季频披露驱动域(0928 根治): Q2 报告期水位 06-30 在披露间隙(09-28,
+    Q2 截止 09-30 未到)=绿; 截止过后(10-02)仍停 Q1=黄——mtime 80h 口径对
+    季频域必假黄(0927 nas-dmsk 实案)."""
+    monkeypatch.setattr(nmc, "DMSK_ROOT", str(tmp_path))
+    _mk_dmsk_partition(tmp_path, "2026-06-30")
+    checks, _ = nmc.build_checks(dt.datetime(2026, 9, 28, 16))
+    dmsk = [c for c in checks if c["key"] == "nas-dmsk"][0]
+    assert dmsk["status"] == "green"
+    assert dmsk["evidence"]["latest_dt"] == "2026-06-30"
+    # Q2 截止(09-30)已过: 仍只有 Q1 分区 → 黄(法定披露季缺失)
+    q1 = tmp_path / "balance" / "dt=2026-03-31"
+    q1.mkdir(parents=True, exist_ok=True)
+    (q1 / "part-0.parquet").touch()
+    import shutil
+    shutil.rmtree(tmp_path / "balance" / "dt=2026-06-30")
+    checks, _ = nmc.build_checks(dt.datetime(2026, 10, 2, 16))
+    dmsk = [c for c in checks if c["key"] == "nas-dmsk"][0]
+    assert dmsk["status"] == "yellow"
+    assert dmsk["evidence"]["min_expected"] == "2026-06-30"