fix(monitor): 1010 二审吸收八修——dmsk半年报08-31勘正+margin_sse同步+strategy两班入岗+Disabled豁免+ETF守卫单边死+ISO格式校验+mirror fallback+FakeDate helper [vps] [nas]

P2-1 半年报法定截止 09-30→08-31(向宽松错30天检出延迟),代码+spec+runbook三处同步;
P2-2 margin_sse (200,4000)→(200,8000)预防性同步(SSE子集~3300贴脸);
P1-2 SCHTASKS 补 strategy-monitor morn/eve(监控自身死亡全链静默缺口),watchlist测试钉17;
P2-6 schtask 主动禁用(Disabled/已禁用)跳过,qmt-relogin 0929/30检修暂停误红实案;
P3-6 xt_eod 守卫抓单边板块死+raise态(NotImplementedError不再裸崩),+3单测;
P3-7 dmsk 分区名只认ISO格式防紧凑格式静默错序;P3-11 mirror start无日期置None防DONE错归日;
P3-12 FakeDate 三份拷贝提取_anchor_today+文件头日期教条修订(禁裸墙钟,锚日选非周期端点);
P3-13 spec带宽表margin/seats行对齐registry现值。

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

scripts/data_platform/data_monitor.py

@@ -306,6 +306,11 @@ def schtask_results(snapshot, now):
         next_run = info.get("next_run", "")
         last_run = info.get("last_run", "")
         last_result = info.get("last_result", "")
+        # 1010 二审 P2-6: 主动禁用(Disabled)不告——qmt-relogin 0929/30 连两日
+        # 红实案=运维检修暂停(防与手动登录打架), N/A→red 兜底只应抓真死
+        st = info.get("status", "")
+        if "禁用" in st or "disabled" in st.lower():
+            continue
         bad = []
         if "N/A" in next_run and cadence != "always":
             bad.append("next_run=N/A(一次性死)")

scripts/data_platform/monitor_registry.py

@@ -38,7 +38,7 @@ _PER_DATE = [
     _pd("xueqiu_hot", (5000, 6500)),
     _pd("hot_rank", (99, 101), ready="18:30", producer="nas"),
     _pd("margin", (200, 8000), ready="21:00", lag=1),  # 实测 6051(0928 文件, 0929 班采); 两融标的扩容涨破旧 4000 上限致 0929/30 连黄
-    _pd("margin_sse", (200, 4000), ready="21:00", lag=1),
+    _pd("margin_sse", (200, 8000), ready="21:00", lag=1),  # SSE 子集 ~3300 贴脸旧 4000 上限(1010 二审 P2-2, 与全市场同受扩容驱动)
     _pd("dragon_tiger_seats", (200, 2000)),  # 实测 670~810(0923~28); 旧(0,400)上限低致真交易日必黄(0928 670 实案)
     _pd("nb_daily", (30, 150)),
 ]
@@ -127,7 +127,9 @@ def _dir_entries():
 REGISTRY = (_PER_DATE + _WEEKLY + _MONTH + _YEAR + _dir_entries()
             + _DB_SAMPLE + _MISC)
 
-# 调度层在岗检查(spec §20.15.1 表后注: 14 个 schtask; retry 一次性死案即此类)
+# 调度层在岗检查(spec §20.15.1 表后注: 17 个 schtask; retry 一次性死案即此类)
+# 1010 二审 P1-2: strategy-monitor 两班补进名单——否则监控自身被禁用/损坏时
+# alerts/health 停更且全链静默(在岗检查正是为此立的设计)
 SCHTASKS = [
     ("sanguo-qmt-relogin", "daily"),
     ("sanguo-qmt-sentinel", "always"),
@@ -144,4 +146,6 @@ SCHTASKS = [
     ("sanguo-ak-vintage", "daily"),
     ("sanguo-index", "monthly"),
     ("sanguo-idx-daily", "daily"),
+    ("sanguo-strategy-monitor-morn", "daily"),
+    ("sanguo-strategy-monitor-eve", "daily"),
 ]

scripts/data_platform/xt_eod.py

@@ -69,10 +69,19 @@ def main():
     log(f"start window ETF/基金={etf_start} 北交所={bj_start}~{end} (full_bj={args.full_bj})")
 
     # 沪深 ETF/基金 (xtata 独占, baostock 不覆盖)
-    etf_codes = list(set(
-        (xd.get_stock_list_in_sector("沪深ETF") or []) +
-        (xd.get_stock_list_in_sector("沪深基金") or [])
-    ))
+    # 1010 二审 P3-6: 板块调用逐个 try/except(桥死时 compat 对非沪深A股
+    # raise NotImplementedError, 0929 晨两崩实案)+单边空检测——旧守卫只抓
+    # 双板块全空, 单边死(ETF 空/基金在)静默漏采半边 universe.
+    etf_dead = []
+    etf_lists = {}
+    for sec in ("沪深ETF", "沪深基金"):
+        try:
+            etf_lists[sec] = list(xd.get_stock_list_in_sector(sec) or [])
+        except NotImplementedError as e:
+            etf_lists[sec] = []
+            etf_dead.append(f"{sec}:{type(e).__name__}")
+            log(f"WARN sector {sec} raise: {e}")
+    etf_codes = list(set(etf_lists["沪深ETF"] + etf_lists["沪深基金"]))
     # 北交所 920xxx (中证2000 成份股, baostock 不覆盖, xtata 独占)
     bj_codes = []
     try:
@@ -95,7 +104,9 @@ def main():
     # (本地 sqlite 恒 39)掩护合并守卫 → rc=0 静默漏采 ETF 一整晚(0928 全
     # ETF 停 0924 实案, 靠监控 dbbardata_daily RED 才抓到). ETF 清单空仍写
     # bj, 但 rc=1 抬出码 → schtask last_result=1 触发监控告警.
-    etf_empty = not etf_codes
+    # 1010 P3-6: 单边板块空也算死(正常日两板块均非空: ETF ~1000+/基金 ~1200).
+    etf_empty = not etf_codes or not etf_lists["沪深ETF"] \
+        or not etf_lists["沪深基金"] or bool(etf_dead)
     if args.limit:
         u = u[:args.limit]
 
@@ -174,7 +185,10 @@ def main():
     log(f"DONE ok={ok} empty={empty} fail={fail} rows={rows}"
         f"{' [DRY-RUN]' if args.dry_run else ''}")
     if etf_empty:
-        log("FATAL ETF/基金 universe EMPTY (桥板块接口死?) — 本班只写北交所, rc=1")
+        dead = "; ".join(etf_dead) if etf_dead else \
+            f"ETF={len(etf_lists['沪深ETF'])} 基金={len(etf_lists['沪深基金'])}"
+        log(f"FATAL ETF/基金 universe 异常({dead}) — 本班未覆盖全部 ETF/基金"
+            f"{'(dry-run 未写)' if args.dry_run else ', 仅北交所已写'}, rc=1")
         return 1
     return 0
 

scripts/nas_sync/nas_monitor_check.py

@@ -11,6 +11,7 @@ import datetime as dt
 import glob
 import json
 import os
+import re
 
 CORPUS_ROOT = "/volume1/stock/corpus"
 DMSK_ROOT = "/volume1/stock/fundamentals/dmsk"
@@ -79,16 +80,24 @@ def _log_has_today_fail(log_path, now):
                for line in text.splitlines()[-200:])
 
 
-# dmsk 季频锚: 各季末→法定披露截止(年报次年 04-30/半年报 09-30/三季 10-31/
+# dmsk 季频锚: 各季末→法定披露截止(年报次年 04-30/半年报 08-31/三季 10-31/
 # 一季 04-30); 截止未过的季度只要求上一季全量在(在披露的季度水位可仍停上季末)
-_QUARTER_DEADLINES = {(3, 31): (4, 30), (6, 30): (9, 30), (9, 30): (10, 31)}
+# (1010 二审 P2-1: 半年报法定=上半年结束后 2 个月内=08-31, 原写 09-30 向
+#  宽松错——Q2 缺失最晚延迟 30 天才检出)
+_QUARTER_DEADLINES = {(3, 31): (4, 30), (6, 30): (8, 31), (9, 30): (10, 31)}
 
 
 def _dmsk_latest_partition(root):
-    """五域(balance/income/cashflow/express/forecast)最新 dt= 分区(ISO)或 None."""
+    """五域(balance/income/cashflow/express/forecast)最新 dt= 分区(ISO)或 None.
+
+    只认 YYYY-MM-DD 零填充格式(1010 二审 P3-7): 裸字典序对上游改
+    dt=20260630 之类紧凑格式会静默错序, 非标格式一律忽略."""
+    iso = re.compile(r"^\d{4}-\d{2}-\d{2}$")
     best = None
     for p in glob.glob(os.path.join(root, "*", "dt=*")):
         d = p.rsplit("dt=", 1)[-1]
+        if not iso.match(d):
+            continue
         if best is None or d > best:
             best = d
     return best
@@ -163,6 +172,9 @@ def build_checks(now):
             done_days = set()
             for line in fh:
                 if "vps-mirror start" in line:
+                    # start 行无日期(格式变化)→cur_day 置 None 而非沿用旧值,
+                    # 防 DONE 错归日(1010 二审 P3-11)
+                    cur_day = None
                     for part in line.split():
                         if len(part) == 10 and part[:2] == "20" and part[4] == "-":
                             cur_day = part