feat(data): 采集器交易日闸门赶 10-1 国庆——sina 官方日历预判(当日缓存跨班共享), 事件类三域(dragon_tiger/block_trade/restricted)非交易日 unit fetch 换空 df 走既有 empty+done 语义, pledge_ratio 剔除休市周五锚(中秋实案 failed 会挂到 10-09 自愈, 10-03 班锚=1002 国庆假同理免疫); 日历失败 None 降级直跑不漏采+覆盖窗外 unit 不动防回补误伤; spec §15.4 登记 [vps]

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

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

@@ -342,6 +342,7 @@ baostock 栈=bs_* 系列,xtdata 栈=daily_update_xtdata,新浪指数=sina_index_
 ### 15.4 语义与防线(全族统一)
 
 - 失败→返 None 不写不标保旧数据(次日重试);真空 df→写空+done;列名零清洗 akshare 原样落盘
+- **交易日闸门**(2026-09-25 中秋休市实案/10-01 国庆预防): EM datacenter 事件类(dragon_tiger/block_trade/restricted)对非交易日查询返 NoneType 异常≠空 df→3 retry 耗尽 failed 计断路器+retry 班整晚空转;pledge_ratio 休市周五锚同理 failed 挂到下个真交易日才自愈。修=akshare_static_download 启动时载 sina 官方交易日历(`trade_calendar.txt` 当日缓存跨班共享),闸门两处——①事件类非交易日 unit 的 fetch 替换为空 df(走既有 empty 写空+done,零新代码路径,日历覆盖窗外 unit 不动防历史回补误伤)②pledge_ratio 枚举剔除非交易日周五。日历加载失败→None 降级直跑(闸门失灵只回到误报,不会漏采)
 - **缺日防线**(static_vintage_check.py 09:00):panel_gap_check 每工作日应有 1 个 parquet(节假日空文件也算有);
   可回补族(zt_pool×3)洞≤42 日历日→告警+回补指引;快照族洞即永久→近窗洞告警查日志、旧洞只入 json 不刷告警
 - **vintage_status.json 的 panel 块**=策略守卫层消费契约:`panel.<type>={n_files,first_date,holes_backfillable,holes_permanent}`

scripts/data_platform/akshare_static_download.py

@@ -121,6 +121,17 @@ _log_dir_env = os.environ.get("AK_STATIC_LOG_DIR")
 LOG_DIR = Path(_log_dir_env) if _log_dir_env else DEFAULT_LOG_DIR
 
 START_DATE_DEFAULT = "2020-01-01"
+# 交易日闸门 (2026-09-25 中秋休市实案 / 10-01 国庆预防): EM datacenter
+# 事件类端点对非交易日查询返 NoneType 异常 (akshare 封装 result=None 取
+# 下标), 3 retry 耗尽 → failed 计断路器 + retry 班整晚空转 (09-25 实证:
+# dragon_tiger/block_trade/restricted ×4 retry 班)。闸门 = sina 官方
+# 交易日历预判, 非交易日 unit 的 fetch 替换为空 df → 走既有「empty 写空
+# +done」语义; pledge_ratio 非交易日周五锚剔除。日历加载失败 → None 降级
+# 直跑 (闸门失灵只回到误报, 不会漏采)。
+TRADE_DAY_GATED_TYPES = frozenset({"dragon_tiger", "block_trade",
+                                    "restricted"})
+TRADE_CALENDAR_CACHE = OUT_DIR / "trade_calendar.txt"
+TRADING_CALENDAR: Optional[set] = None  # main 早期填充; None=闸门降级
 # 报告期循环范围 (近5年×4季, forecast/express/top_holders 用).
 # top_holders per-stock × per-period 会有股票×20期组合.
 REPORT_PERIODS = [
@@ -2128,6 +2139,77 @@ def run_one_type(
     return stats, circuit_triggered
 
 
+# ======================== 交易日闸门 ========================
+
+def load_trading_calendar(cache_path: Optional[Path] = None) -> Optional[set]:
+    """sina 官方交易日历 (ak.tool_trade_date_hist_sina) → {'YYYYMMDD'}.
+
+    当日缓存 (ak-events 四个 retry 班 + ak-weekly 只打一发); 失败/空表
+    返 None = 调用方降级直跑。
+    """
+    cache = Path(cache_path) if cache_path else TRADE_CALENDAR_CACHE
+    today = datetime.date.today()
+    if cache.exists():
+        try:
+            if datetime.date.fromtimestamp(cache.stat().st_mtime) == today:
+                days = {ln.strip() for ln
+                        in cache.read_text(encoding="utf-8").splitlines()
+                        if ln.strip()}
+                if days:
+                    return days
+        except OSError:
+            pass  # 缓存读坏 → 重拉
+    try:
+        df = ak.tool_trade_date_hist_sina()
+        days = {d.strftime("%Y%m%d") for d in df["trade_date"]}
+    except Exception as e:  # noqa: BLE001  日历=闸门增强非正确性必需
+        logger.warning("[trade-calendar] 交易日历拉取失败, 闸门降级直跑: %s", e)
+        return None
+    if not days:
+        logger.warning("[trade-calendar] 交易日历为空, 闸门降级直跑")
+        return None
+    try:
+        cache.parent.mkdir(parents=True, exist_ok=True)
+        tmp = cache.with_name(cache.name + ".tmp")
+        tmp.write_text("\n".join(sorted(days)) + "\n", encoding="utf-8")
+        os.replace(tmp, cache)
+    except OSError as e:
+        logger.warning("[trade-calendar] 缓存写失败 (不影响本次): %s", e)
+    return days
+
+
+def _holiday_empty_fetch(data_type: str, uid: str) -> Callable[[], pd.DataFrame]:
+    def _fetch() -> pd.DataFrame:
+        logger.info("[%s] %s 非交易日 (日历) → 空 df (节假日语义)",
+                    data_type, uid)
+        return pd.DataFrame()
+    return _fetch
+
+
+def gate_units_by_calendar(
+    data_type: str,
+    units: List[Tuple[str, Callable[[], pd.DataFrame]]],
+) -> List[Tuple[str, Callable[[], pd.DataFrame]]]:
+    """交易日闸门: 非交易日 unit 的 fetch 替换为空 df (走 empty 写空+done,
+    不耗 retry/断路器配额)。日历覆盖窗外的 unit 不动 (防历史回补误伤)。"""
+    if data_type not in TRADE_DAY_GATED_TYPES or not TRADING_CALENDAR:
+        return units
+    lo, hi = min(TRADING_CALENDAR), max(TRADING_CALENDAR)
+    out, n_gated = [], 0
+    for uid, fn in units:
+        day = uid.split("_", 1)[0]
+        if (len(day) == 8 and day.isdigit() and lo <= day <= hi
+                and day not in TRADING_CALENDAR):
+            out.append((uid, _holiday_empty_fetch(data_type, uid)))
+            n_gated += 1
+        else:
+            out.append((uid, fn))
+    if n_gated:
+        logger.info("[%s] 交易日闸门: %d 个非交易日 unit → 空 df (empty+done)",
+                    data_type, n_gated)
+    return out
+
+
 # ======================== 各模式 units 构造 ========================
 
 def build_per_stock_units(
@@ -2308,6 +2390,16 @@ def build_pledge_ratio_units(
     下周班窗口重含该周五 → 自动补拉 (滞后最多一周, 自愈)。
     """
     fridays = enumerate_fridays(normalize_date(args.start), normalize_date(args.end))
+    if TRADING_CALENDAR:
+        # 交易日闸门: 休市周五无中登周报 (2026-09-25 中秋实案 → failed 挂
+        # 到 10-09 才自愈; 剔除后 10-03 班锚=1002 国庆假同理免疫)
+        lo, hi = min(TRADING_CALENDAR), max(TRADING_CALENDAR)
+        keep = [f for f in fridays
+                if not (lo <= f <= hi and f not in TRADING_CALENDAR)]
+        if len(keep) != len(fridays):
+            logger.info("[%s] 交易日闸门: 剔除 %d 个非交易日周五 (休市无周报)",
+                        PLEDGE_RATIO, len(fridays) - len(keep))
+        fridays = keep
     units = [
         (f"{f}_{PLEDGE_RATIO}", partial(fetch_pledge_ratio, f))
         for f in fridays
@@ -2469,6 +2561,7 @@ def run_type_dispatch(
             "nb_daily": fetch_nb_daily,
         }
         units = build_per_date_units(t, fetch_map[t], args)
+        units = gate_units_by_calendar(t, units)
         return run_one_type(t, units, args)
 
     if t == NB_HOLD:
@@ -2513,13 +2606,16 @@ def run_type_dispatch(
 
 
 def main() -> None:
-    global logger, LOG_FILE
+    global logger, LOG_FILE, TRADING_CALENDAR
     logger, LOG_FILE = setup_logging()
     args = parse_args()
     start_date = normalize_date(args.start)
     end_date = normalize_date(args.end)
     types = resolve_types(args.types)
 
+    # 交易日闸门 (当日缓存跨班共享; 失败=None 降级直跑)
+    TRADING_CALENDAR = load_trading_calendar()
+
     if getattr(args, "margin_backfill", False):
         logger.info("[margin-backfill] TMC 段慢通道回补启动 (退市史)")
         rc = backfill_margin()

tests/data_platform/test_trading_day_gate.py

@@ -0,0 +1,144 @@
+# -*- coding: utf-8 -*-
+"""交易日闸门测试 (2026-09-25 中秋休市案 / 10-01 国庆预防).
+
+背景: EM datacenter 事件类端点 (dragon_tiger/block_trade/restricted) 对
+非交易日查询返 NoneType 异常 → 3 retry 耗尽 → failed 计断路器 + retry 班
+整晚空转。闸门 = sina 官方交易日历预判, 非交易日 unit fetch 替换为空 df
+→ 走既有「empty 写空+done」语义; pledge_ratio 非交易日周五锚直接剔除。
+日历加载失败 → None 降级直跑 (闸门失灵只回到误报, 不会漏采)。
+"""
+import argparse
+import datetime as dt
+
+import pandas as pd
+import pytest
+
+import akshare_static_download as mod  # noqa: E402  (conftest 已加 sys.path)
+
+
+def _cal(days):
+    return set(days)
+
+
+@pytest.fixture(autouse=True)
+def _reset_calendar(monkeypatch):
+    """每测试结束还原模块级日历 (防跨测试泄漏)。"""
+    monkeypatch.setattr(mod, "TRADING_CALENDAR", None)
+
+
+# ======================== load_trading_calendar ========================
+
+class TestLoadTradingCalendar:
+    def _fake_ak(self, monkeypatch, df=None, raises=None):
+        calls = []
+
+        def fake():
+            calls.append(1)
+            if raises:
+                raise raises
+            return df
+
+        monkeypatch.setattr(mod.ak, "tool_trade_date_hist_sina", fake)
+        return calls
+
+    def test_fetch_writes_cache_and_returns_days(self, monkeypatch, tmp_path):
+        cache = tmp_path / "cal.txt"
+        df = pd.DataFrame({"trade_date": [dt.date(2026, 9, 24),
+                                          dt.date(2026, 9, 25)]})
+        self._fake_ak(monkeypatch, df=df)
+        days = mod.load_trading_calendar(cache)
+        assert days == {"20260924", "20260925"}
+        assert cache.exists() and "20260924" in cache.read_text()
+
+    def test_same_day_cache_hit_skips_network(self, monkeypatch, tmp_path):
+        cache = tmp_path / "cal.txt"
+        cache.write_text("20260924\n20260925\n", encoding="utf-8")
+        # 缓存 mtime=现在=今天 → 命中; mock 打网即炸
+        self._fake_ak(monkeypatch, raises=AssertionError("不该打网"))
+        days = mod.load_trading_calendar(cache)
+        assert days == {"20260924", "20260925"}
+
+    def test_network_failure_returns_none(self, monkeypatch, tmp_path):
+        cache = tmp_path / "cal.txt"
+        self._fake_ak(monkeypatch, raises=RuntimeError("sina 502"))
+        assert mod.load_trading_calendar(cache) is None
+
+    def test_stale_cache_refetches(self, monkeypatch, tmp_path):
+        cache = tmp_path / "cal.txt"
+        cache.write_text("20200101\n", encoding="utf-8")
+        # mtime 拨回 8 天前 → 过期重拉
+        import os
+        old = cache.stat().st_mtime - 8 * 86400
+        os.utime(cache, (old, old))
+        df = pd.DataFrame({"trade_date": [dt.date(2026, 9, 28)]})
+        self._fake_ak(monkeypatch, df=df)
+        assert mod.load_trading_calendar(cache) == {"20260928"}
+
+
+# ======================== gate_units_by_calendar ========================
+
+class TestGateUnits:
+    def test_holiday_unit_fetch_swapped_to_empty(self, monkeypatch):
+        monkeypatch.setattr(mod, "TRADING_CALENDAR",
+                            _cal(["20260924", "20260930"]))
+        def orig():
+            raise AssertionError("非交易日不该打端点")
+        units = [("20260925_dragon_tiger", orig)]
+        gated = mod.gate_units_by_calendar("dragon_tiger", units)
+        assert len(gated) == 1
+        df = gated[0][1]()
+        assert isinstance(df, pd.DataFrame) and df.empty
+
+    def test_trading_day_unit_passthrough(self, monkeypatch):
+        monkeypatch.setattr(mod, "TRADING_CALENDAR",
+                            _cal(["20260924", "20260930"]))
+        def orig():
+            return pd.DataFrame({"x": [1]})
+        units = [("20260924_dragon_tiger", orig)]
+        gated = mod.gate_units_by_calendar("dragon_tiger", units)
+        assert gated[0][1] is orig
+
+    def test_out_of_calendar_range_unit_untouched(self, monkeypatch):
+        # 日历覆盖窗外的历史日 (回补场景) 不 gate, 防误伤丢数据
+        monkeypatch.setattr(mod, "TRADING_CALENDAR",
+                            _cal(["20260924", "20260930"]))
+        def orig():
+            return pd.DataFrame({"x": [1]})
+        units = [("20200101_dragon_tiger", orig)]
+        gated = mod.gate_units_by_calendar("dragon_tiger", units)
+        assert gated[0][1] is orig
+
+    def test_ungated_type_untouched(self, monkeypatch):
+        monkeypatch.setattr(mod, "TRADING_CALENDAR", _cal(["20260924"]))
+        def orig():
+            return pd.DataFrame({"x": [1]})
+        units = [("20260925_zt_pool", orig)]  # zt_pool 空数据=empty 语义已对
+        gated = mod.gate_units_by_calendar("zt_pool", units)
+        assert gated[0][1] is orig
+
+    def test_none_calendar_noop(self):
+        def orig():
+            return pd.DataFrame({"x": [1]})
+        units = [("20260925_dragon_tiger", orig)]
+        assert mod.gate_units_by_calendar("dragon_tiger", units)[0][1] is orig
+
+
+# ======================== pledge_ratio 周五锚闸门 ========================
+
+class TestPledgeRatioFridayGate:
+    def test_holiday_friday_excluded(self, monkeypatch):
+        # 20260918=交易日周五, 20260925=中秋休市周五 → 只剩 0918
+        monkeypatch.setattr(mod, "TRADING_CALENDAR",
+                            _cal(["20260918", "20260930"]))
+        args = argparse.Namespace(start="20260914", end="20260926",
+                                  force=False)
+        units = mod.build_pledge_ratio_units(args)
+        uids = [uid for uid, _ in units]
+        assert uids == ["20260918_pledge_ratio"]
+
+    def test_none_calendar_keeps_all_fridays(self):
+        args = argparse.Namespace(start="20260914", end="20260926",
+                                  force=False)
+        units = mod.build_pledge_ratio_units(args)
+        uids = [uid for uid, _ in units]
+        assert uids == ["20260918_pledge_ratio", "20260925_pledge_ratio"]