feat(HEL-435): 盘后未出数时晚间自动重试并提供安全补跑

Co-authored-by: multica-agent <github@multica.ai>

backend/data/datahub/bridge.py

@@ -208,6 +208,10 @@ class DatahubBridge:
             raise DatahubError("STALE", f"{dataset} data is stale")
         if dataset in EMPTY_FAIL_DATASETS and not rows:
             raise DatahubError("EMPTY", f"{dataset} returned no rows")
+        coverage = meta.get("coverage") if isinstance(meta.get("coverage"), dict) else {}
+        if meta.get("incomplete") is True or coverage.get("complete") is False:
+            missing = coverage.get("missing_count")
+            raise DatahubError("INCOMPLETE", f"{dataset} range is incomplete missing={missing}")
 
     def _require_fresh(self, response: DatahubResponse, dataset: str) -> DatahubResponse:
         self._validate_usable(dataset, list(response.data or []) if isinstance(response.data, list) else [], response)

tests/test_datahub_bridge.py

@@ -126,7 +126,7 @@ class DatahubBridgeTests(unittest.TestCase):
         self.assertEqual(calendar[0]["is_open"], 1)
         self.assertEqual(calendar_client.paths, [])
 
-    def test_fallback_on_down_401_timeout_empty_unpublished_and_stale(self) -> None:
+    def test_fallback_on_down_401_timeout_empty_unpublished_stale_and_incomplete(self) -> None:
         cases = [
             DatahubError("UNAVAILABLE", "down"),
             DatahubError("UNAUTHORIZED", "401"),
@@ -134,6 +134,7 @@ class DatahubBridgeTests(unittest.TestCase):
             DatahubError("EMPTY", "no rows"),
             DatahubError("DATASET_NOT_PUBLISHED", "not ready"),
             DatahubError("STALE", "old"),
+            DatahubError("INCOMPLETE", "truncated"),
         ]
         for error in cases:
             with self.subTest(error=error.code):
@@ -144,6 +145,16 @@ class DatahubBridgeTests(unittest.TestCase):
                         data=[dict(HUB_DAILY)],
                         meta={"stale": True, "staleness_seconds": 999999},
                     ))
+                elif error.code == "INCOMPLETE":
+                    client = FakeClient(response=DatahubResponse(
+                        data=[dict(HUB_DAILY)],
+                        meta={
+                            "stale": False,
+                            "staleness_seconds": 0,
+                            "incomplete": True,
+                            "coverage": {"complete": False, "missing_count": 80},
+                        },
+                    ))
                 else:
                     client = FakeClient(error=error)
                 legacy = FakeLegacy([LEGACY_DAILY])
@@ -235,6 +246,27 @@ class DatahubBridgeTests(unittest.TestCase):
         self.assertIsInstance(client, DatahubAwareTushareClient)
         self.assertFalse(gateway.datahub.settings.any_enabled())
 
+    def test_stock_detail_range_query_is_not_silently_accepted_when_incomplete(self) -> None:
+        source = (ROOT / "backend" / "data" / "providers" / "tushare_stocks.py").read_text(encoding="utf-8")
+        self.assertIn('"daily"', source)
+        self.assertIn("start_date", source)
+        self.assertIn("end_date", source)
+        client = FakeClient(
+            response=DatahubResponse(
+                data=[dict(HUB_DAILY)],
+                meta={"stale": False, "staleness_seconds": 0, "incomplete": True, "coverage": {"complete": False, "missing_count": 89}},
+            )
+        )
+        legacy = FakeLegacy([LEGACY_DAILY])
+        wrapped = DatahubAwareTushareClient(legacy, DatahubBridge(flags(daily=(True, False)), client))
+        rows = wrapped.query(
+            "daily",
+            {"ts_code": "600000.SH", "start_date": "20240301", "end_date": "20240902"},
+            "ts_code,amount",
+        )
+        self.assertEqual(rows[0]["amount"], 2000.0)
+        self.assertEqual(len(legacy.calls), 1)
+
     def test_features_do_not_import_datahub_client(self) -> None:
         violations = []
         for path in (ROOT / "backend" / "features").rglob("*.py"):

xiaobai-datahub/README.md

@@ -63,6 +63,22 @@ python -m unittest discover -s tests -v
 
 不调用真实 Tushare；用内存/临时库和假适配器。
 
+## 历史回补
+
+交易日历默认从 `20160101` 拉到今天后 30 天；盘前 `precheck` 与手动回补都走同一 UPSERT，可重复执行。
+
+网站实际使用的指数（上证、深成、创业板、沪深300）按交易日增量发布，默认覆盖 260 个交易日（大于现有 90 天窗口，并覆盖智能选股基准回看）。已发布日期默认跳过。
+
+```bash
+cd xiaobai-datahub
+python -m datahub history-backfill
+# 可选：--calendar-start 20160101 --index-days 260 --force
+```
+
+管理后台也可手动跑 `history_backfill` 任务，或 `POST /admin/api/backfill` 且 `dataset=history`、确认词 `history:full`。
+
+区间接口在 `meta.coverage` / `meta.incomplete` 标明覆盖是否完整；网站只读接入把不完整区间视为不可用并回旧链路。个股日 K 的 90 天区间查询依赖已核实，本阶段不回补全市场历史。
+
 ## 备份
 
 每日 00:40 任务把 `datahub.db` 备份到 `data/backups/`（保留 14 份）。也可手动：

xiaobai-datahub/admin/app.js

@@ -104,11 +104,29 @@ async function render() {
   if (state.page === "overview") {
     const data = await api("/admin/api/overview");
     $("phase").textContent = data.session_phase;
+    const eod = data.eod_status || {};
+    const eodLabels = {
+      pending_first_attempt: "等待首次尝试",
+      waiting_upstream: "等待上游",
+      done: "已成功",
+      cutoff_failed: "已截止失败",
+      closed_day: "休市",
+    };
+    const eodExtra = [];
+    if (eod.state === "waiting_upstream") {
+      eodExtra.push(`已试 ${eod.attempts} 次`);
+      if (eod.next_retry_at) eodExtra.push(`下次重试 ${esc(String(eod.next_retry_at).replace("T", " ").slice(11, 16))}`);
+      if (eod.missing_datasets && eod.missing_datasets.length) eodExtra.push(`缺 ${esc(eod.missing_datasets.join(","))}`);
+    }
+    if (eod.state === "cutoff_failed" && eod.missing_datasets) {
+      eodExtra.push(`缺 ${esc(eod.missing_datasets.join(","))}`);
+    }
     page.innerHTML = `
       <div class="cards">
         <div class="card"><div class="muted">交易日</div><strong>${esc(data.trade_date)}</strong></div>
         <div class="card"><div class="muted">阶段</div><strong>${esc(data.session_phase)}</strong></div>
         <div class="card"><div class="muted">今日发布</div><strong>${data.publications.length}</strong></div>
+        <div class="card"><div class="muted">盘后补跑</div><strong>${esc(eodLabels[eod.state] || eod.state || "-")}</strong><div class="muted">${eodExtra.join(" · ")}</div></div>
         <div class="card"><div class="muted">异常批次</div><strong class="${data.anomalies.length ? "fail" : "ok"}">${data.anomalies.length}</strong></div>
       </div>
       <h2>最近调用</h2>

xiaobai-datahub/config/hub-quality.config.json

@@ -11,5 +11,10 @@
   "publication_generations": 3,
   "tushare_rate_per_minute": 300,
   "list_limit_default": 5000,
-  "list_limit_max": 5000
+  "list_limit_max": 5000,
+  "calendar_start": "20160101",
+  "index_history_trading_days": 260,
+  "eod_retry_start": "15:15",
+  "eod_retry_interval_minutes": 30,
+  "eod_retry_cutoff": "23:30"
 }

xiaobai-datahub/datahub/__main__.py

@@ -0,0 +1,4 @@
+from datahub.cli import main
+
+if __name__ == "__main__":
+    raise SystemExit(main())

xiaobai-datahub/datahub/adapters/tushare.py

@@ -44,7 +44,10 @@ DATASET_API = {
     "auction": "stk_auction",
 }
 
-DEFAULT_INDEX_CODES = ("000001.SH", "399001.SZ", "399006.SZ", "000300.SH")
+# Website actual index usage: market cards / 90-day charts (SH/SZ/CYB) plus
+# screener 沪深300 benchmark (lookback up to 260 trading days).
+WEBSITE_INDEX_CODES = ("000001.SH", "399001.SZ", "399006.SZ", "000300.SH")
+DEFAULT_INDEX_CODES = WEBSITE_INDEX_CODES
 
 
 class TushareAdapter(MarketAdapter):

xiaobai-datahub/datahub/admin_api.py

@@ -38,6 +38,7 @@ class AdminAPI:
             "trade_date": today,
             "session_phase": session_phase(now_shanghai(), is_open),
             "is_open_day": is_open,
+            "eod_status": self.scheduler.eod_status(today),
             "publications": pubs,
             "anomalies": failed,
             "recent_calls": _public_calls(calls),
@@ -88,6 +89,8 @@ class AdminAPI:
                 {"id": "precheck", "at": "08:45", "title": "盘前预检"},
                 {"id": "eod_a", "at": "15:05", "title": "盘后批 A daily/valuation/moneyflow/auction"},
                 {"id": "eod_b", "at": "15:10", "title": "盘后批 B index_daily"},
+                {"id": "eod_retry", "at": "15:15-23:30", "title": "盘后未出数自动重试（每 30 分钟，成功即停）"},
+                {"id": "history_backfill", "at": "manual", "title": "回补历史日历与指数日 K"},
                 {"id": "cleanup", "at": "00:30", "title": "清理 staging / 日志"},
                 {"id": "backup", "at": "00:40", "title": "SQLite 备份"},
             ],
@@ -130,12 +133,17 @@ class AdminAPI:
         return result
 
     def backfill(self, dataset: str, trade_date: str, password: str, confirm: str, actor: str) -> dict[str, Any]:
-        self._dangerous(password, confirm, f"{dataset}:{trade_date}")
+        day = yyyymmdd(trade_date or now_shanghai())
-        if dataset == "reference":
+        if dataset == "history":
-            result = self.pipeline.ingest_reference(trade_date)
+            self._dangerous(password, confirm, "history:full")
+            result = self.pipeline.backfill_history(day)
         else:
-            result = self.pipeline.run_dataset(dataset, trade_date)
+            self._dangerous(password, confirm, f"{dataset}:{day}")
-        self.pipeline.audit(actor, "backfill", f"{dataset}:{trade_date}", json.dumps({"ok": True}))
+            if dataset == "reference":
+                result = self.pipeline.ingest_reference(day)
+            else:
+                result = self.pipeline.run_dataset(dataset, day)
+        self.pipeline.audit(actor, "backfill", f"{dataset}:{day}", json.dumps({"ok": True}))
         return result
 
     def _dangerous(self, password: str, confirm: str, expected: str) -> None:

xiaobai-datahub/datahub/cli.py

@@ -0,0 +1,53 @@
+"""Command-line entry for one-shot datahub operations."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+
+from datahub.hub import build_hub
+from datahub.settings import load_settings
+from datahub.timeutil import yyyymmdd
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="xiaobai-datahub CLI")
+    sub = parser.add_subparsers(dest="command", required=True)
+    history = sub.add_parser("history-backfill", help="回补 2016 年起交易日历和网站所用指数日 K")
+    history.add_argument("--calendar-start", default=None, help="日历起点，默认配置 calendar_start")
+    history.add_argument("--index-days", type=int, default=None, help="指数回补交易日数量，默认 260")
+    history.add_argument("--force", action="store_true", help="覆盖已发布的指数日期")
+    refresh = sub.add_parser("eod-refresh", help="对指定交易日补跑盘后正式数据（跳过已发布数据集，仍走质量门禁）")
+    refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天")
+    args = parser.parse_args(argv)
+
+    settings = load_settings()
+    hub = build_hub(settings)
+    if args.command == "history-backfill":
+        result = hub.pipeline.backfill_history(
+            calendar_start=args.calendar_start,
+            index_days=args.index_days,
+            force=args.force,
+        )
+        json.dump(result, sys.stdout, ensure_ascii=False, indent=2, default=str)
+        sys.stdout.write("\n")
+        return 0 if result.get("ok") else 1
+    if args.command == "eod-refresh":
+        day = yyyymmdd(args.trade_date) if args.trade_date else yyyymmdd()
+        result = hub.pipeline.run_eod_missing(day)
+        hub.pipeline.audit("cli", "eod-refresh", f"eod:{day}", json.dumps(
+            {name: item.get("state") for name, item in result.items() if isinstance(item, dict)},
+            ensure_ascii=False,
+        ))
+        missing = hub.pipeline.missing_official_datasets(day)
+        payload = {"trade_date": day, "datasets": result, "missing_after": missing}
+        json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, default=str)
+        sys.stdout.write("\n")
+        return 0 if not missing else 1
+    parser.error(f"unknown command: {args.command}")
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

xiaobai-datahub/datahub/coverage.py

@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+from typing import Any, Iterable
+
+from datahub.db import HubDB
+from datahub.timeutil import iter_yyyymmdd, yyyymmdd
+
+MISSING_SAMPLE_LIMIT = 10
+
+
+def coverage_payload(
+    *,
+    kind: str,
+    start: str,
+    end: str,
+    expected: Iterable[str],
+    available: Iterable[str],
+    extra: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    start = yyyymmdd(start)
+    end = yyyymmdd(end)
+    expected_list = sorted({yyyymmdd(item) for item in expected if item})
+    available_set = {yyyymmdd(item) for item in available if item}
+    missing = [item for item in expected_list if item not in available_set]
+    payload: dict[str, Any] = {
+        "kind": kind,
+        "complete": not missing,
+        "requested_from": start,
+        "requested_to": end,
+        "available_from": min(available_set) if available_set else None,
+        "available_to": max(available_set) if available_set else None,
+        "expected_count": len(expected_list),
+        "available_count": len(available_set),
+        "missing_count": len(missing),
+        "missing_sample": missing[:MISSING_SAMPLE_LIMIT],
+    }
+    if extra:
+        payload.update(extra)
+    return payload
+
+
+def calendar_coverage(db: HubDB, start: str, end: str, exchange: str = "SSE") -> dict[str, Any]:
+    start = yyyymmdd(start)
+    end = yyyymmdd(end)
+    expected = list(iter_yyyymmdd(start, end))
+    rows = db.fetchall(
+        "SELECT cal_date FROM trade_calendar WHERE exchange = ? AND cal_date >= ? AND cal_date <= ?",
+        (exchange, start, end),
+    )
+    return coverage_payload(
+        kind="calendar",
+        start=start,
+        end=end,
+        expected=expected,
+        available=(row["cal_date"] for row in rows),
+        extra={"exchange": exchange},
+    )
+
+
+def published_range_coverage(
+    db: HubDB,
+    dataset: str,
+    start: str,
+    end: str,
+    ts_code: str = "",
+    table: str = "",
+) -> dict[str, Any]:
+    start = yyyymmdd(start)
+    end = yyyymmdd(end)
+    calendar = calendar_coverage(db, start, end)
+    open_rows = db.fetchall(
+        """
+        SELECT cal_date FROM trade_calendar
+        WHERE exchange = 'SSE' AND is_open = 1 AND cal_date >= ? AND cal_date <= ?
+        ORDER BY cal_date
+        """,
+        (start, end),
+    )
+    expected_open = [row["cal_date"] for row in open_rows]
+    pubs = db.fetchall(
+        """
+        SELECT trade_date, active_batch FROM publications
+        WHERE dataset = ? AND trade_date >= ? AND trade_date <= ?
+        ORDER BY trade_date
+        """,
+        (dataset, start, end),
+    )
+    published_dates = [row["trade_date"] for row in pubs]
+    available = list(published_dates)
+    extra: dict[str, Any] = {
+        "dataset": dataset,
+        "calendar_complete": calendar["complete"],
+        "calendar_missing_count": calendar["missing_count"],
+    }
+    if ts_code and table and pubs:
+        present_code: list[str] = []
+        for pub in pubs:
+            hit = db.fetchone(
+                f"SELECT 1 AS ok FROM {table} WHERE trade_date = ? AND batch_id = ? AND ts_code = ? LIMIT 1",
+                (pub["trade_date"], pub["active_batch"], ts_code),
+            )
+            if hit:
+                present_code.append(pub["trade_date"])
+        available = present_code
+        extra["code"] = ts_code
+    payload = coverage_payload(
+        kind="published_range",
+        start=start,
+        end=end,
+        expected=expected_open,
+        available=available,
+        extra=extra,
+    )
+    if not calendar["complete"]:
+        payload["complete"] = False
+        payload["calendar_missing_sample"] = calendar["missing_sample"]
+    return payload
+
+
+def point_coverage(trade_date: str, dataset: str = "") -> dict[str, Any]:
+    day = yyyymmdd(trade_date)
+    payload = coverage_payload(
+        kind="point",
+        start=day,
+        end=day,
+        expected=[day],
+        available=[day],
+        extra={"dataset": dataset} if dataset else None,
+    )
+    return payload

xiaobai-datahub/datahub/db.py

@@ -212,6 +212,17 @@ CREATE TABLE IF NOT EXISTS job_runs (
   detail TEXT
 );
 
+CREATE TABLE IF NOT EXISTS eod_progress (
+  trade_date TEXT PRIMARY KEY,
+  state TEXT NOT NULL,
+  attempts INTEGER NOT NULL DEFAULT 0,
+  last_attempt_at TEXT,
+  next_retry_at TEXT,
+  finished_at TEXT,
+  detail TEXT,
+  updated_at TEXT NOT NULL
+);
+
 CREATE TABLE IF NOT EXISTS audit_log (
   id INTEGER PRIMARY KEY AUTOINCREMENT,
   actor TEXT NOT NULL,

xiaobai-datahub/datahub/pipeline.py

@@ -7,7 +7,7 @@ from datetime import timedelta
 from typing import Any
 
 from datahub.adapters.base import AdapterError
-from datahub.adapters.tushare import DEFAULT_INDEX_CODES, TushareAdapter
+from datahub.adapters.tushare import DEFAULT_INDEX_CODES, WEBSITE_INDEX_CODES, TushareAdapter
 from datahub.db import DATASET_TABLES, HubDB
 from datahub.governance.circuit import CircuitBreaker
 from datahub.governance.ratelimit import TokenBucket
@@ -22,6 +22,8 @@ LOGGER = get_logger()
 HARD_DATASETS = {"daily", "valuation", "index_daily"}
 SOFT_DATASETS = {"moneyflow", "auction"}
 OFFICIAL_DATASETS = HARD_DATASETS | SOFT_DATASETS
+EOD_A_DATASETS = ("daily", "valuation", "moneyflow", "auction")
+EOD_B_DATASETS = ("index_daily",)
 EMPTY_BATCH_ERROR = "empty official batch: 0 valid rows"
 
 STAGING_INSERT = {
@@ -141,11 +143,22 @@ class Pipeline:
         seq = int((row or {}).get("n") or 0) + 1
         return f"{trade_date}-{dataset}-{seq:03d}"
 
-    def ingest_reference(self, trade_date: str | None = None) -> dict[str, Any]:
+    def ingest_reference(
-        """Refresh trade calendar (window) and stock master. Not versioned by batch."""
+        self,
+        trade_date: str | None = None,
+        start: str | None = None,
+        end: str | None = None,
+    ) -> dict[str, Any]:
+        """Refresh trade calendar and stock master. Not versioned by batch.
+
+        Calendar defaults to 2016-01-01 through today+30 so a 5-year website
+        query is not silently truncated. UPSERT makes repeats safe.
+        """
         day = yyyymmdd(trade_date or self.clock())
-        start = add_days(day, -400)
+        start = yyyymmdd(start or self.settings.calendar_start)
-        end = add_days(day, 30)
+        end = yyyymmdd(end or add_days(day, 30))
+        if start > end:
+            start, end = end, start
         calendar = self.adapter.normalize(
             "calendar",
             self._guarded_fetch("calendar", {"exchange": "SSE", "start_date": start, "end_date": end}),
@@ -180,9 +193,150 @@ class Pipeline:
                         row.get("list_date"), fetched_at,
                     ),
                 )
-        return {"calendar": len(calendar), "stocks": len(stocks), "trade_date": day}
+        return {
+            "calendar": len(calendar),
+            "stocks": len(stocks),
+            "trade_date": day,
+            "calendar_from": start,
+            "calendar_to": end,
+        }
 
-    def run_dataset(self, dataset: str, trade_date: str, attempts: int | None = None) -> dict[str, Any]:
+    def open_trade_dates(self, end: str, limit: int) -> list[str]:
+        end = yyyymmdd(end)
+        rows = self.db.fetchall(
+            """
+            SELECT cal_date FROM trade_calendar
+            WHERE exchange = 'SSE' AND is_open = 1 AND cal_date <= ?
+            ORDER BY cal_date DESC
+            LIMIT ?
+            """,
+            (end, max(1, int(limit))),
+        )
+        return sorted(str(row["cal_date"]) for row in rows)
+
+    def backfill_history(
+        self,
+        trade_date: str | None = None,
+        calendar_start: str | None = None,
+        index_days: int | None = None,
+        codes: tuple[str, ...] | None = None,
+        force: bool = False,
+    ) -> dict[str, Any]:
+        """Idempotent calendar + website-index history backfill."""
+        day = yyyymmdd(trade_date or self.clock())
+        calendar = self.ingest_reference(day, start=calendar_start)
+        index = self.backfill_index_history(
+            end_date=day,
+            trading_days=index_days,
+            codes=codes,
+            force=force,
+        )
+        return {"calendar": calendar, "index_daily": index, "ok": bool(index.get("ok"))}
+
+    def backfill_index_history(
+        self,
+        end_date: str | None = None,
+        trading_days: int | None = None,
+        codes: tuple[str, ...] | None = None,
+        force: bool = False,
+    ) -> dict[str, Any]:
+        """Incrementally publish official index bars for website index codes.
+
+        One range fetch per code, then per-day publish. Already published dates
+        are skipped unless ``force``. Failures are recorded and do not roll back
+        successful days.
+        """
+        end = yyyymmdd(end_date or self.clock())
+        limit = int(trading_days or self.settings.index_history_trading_days)
+        codes = tuple(codes or WEBSITE_INDEX_CODES)
+        open_dates = self.open_trade_dates(end, limit)
+        if not open_dates:
+            return {
+                "start": None,
+                "end": end,
+                "codes": list(codes),
+                "requested_days": 0,
+                "published": [],
+                "skipped": [],
+                "failed": [{"error": "calendar has no open dates on or before end"}],
+                "ok": False,
+            }
+        start = open_dates[0]
+        complete_dates = set() if force else self._index_dates_with_all_codes(start, end, codes)
+        targets = [day for day in open_dates if day not in complete_dates]
+        skipped = [day for day in open_dates if day in complete_dates]
+        by_date: dict[str, list[dict[str, Any]]] = {day: [] for day in targets}
+        failed: list[dict[str, Any]] = []
+        for ts_code in codes:
+            try:
+                raw = retry_call(
+                    lambda code=ts_code: self._guarded_fetch(
+                        "index_daily",
+                        {"ts_code": code, "start_date": start, "end_date": end},
+                    ),
+                    attempts=self.settings.max_publish_attempts,
+                    base_delay=0.05,
+                    sleeper=lambda _d: time.sleep(_d),
+                )
+                for row in self.adapter.normalize("index_daily", raw):
+                    day = str(row.get("trade_date") or "")
+                    if day in by_date:
+                        by_date[day].append(row)
+            except Exception as exc:
+                failed.append({"ts_code": ts_code, "error": str(exc)})
+        published: list[dict[str, Any]] = []
+        for day in targets:
+            rows = by_date.get(day) or []
+            try:
+                result = self.run_dataset("index_daily", day, prepared_rows=rows)
+                published.append(
+                    {
+                        "trade_date": day,
+                        "batch_id": result["batch_id"],
+                        "rows": result["rows"],
+                        "state": result["state"],
+                    }
+                )
+            except Exception as exc:
+                failed.append({"trade_date": day, "error": str(exc), "rows": len(rows)})
+        return {
+            "start": start,
+            "end": end,
+            "codes": list(codes),
+            "requested_days": len(open_dates),
+            "published": published,
+            "skipped": skipped,
+            "failed": failed,
+            "ok": not failed,
+        }
+
+    def _index_dates_with_all_codes(self, start: str, end: str, codes: tuple[str, ...]) -> set[str]:
+        pubs = self.db.fetchall(
+            """
+            SELECT trade_date, active_batch FROM publications
+            WHERE dataset = 'index_daily' AND trade_date >= ? AND trade_date <= ?
+            """,
+            (start, end),
+        )
+        needed = set(codes)
+        complete: set[str] = set()
+        for pub in pubs:
+            rows = self.db.fetchall(
+                "SELECT DISTINCT ts_code FROM eod_index_bars WHERE trade_date = ? AND batch_id = ?",
+                (pub["trade_date"], pub["active_batch"]),
+            )
+            have = {str(row["ts_code"]) for row in rows}
+            if needed <= have:
+                complete.add(str(pub["trade_date"]))
+        return complete
+
+    def run_dataset(
+        self,
+        dataset: str,
+        trade_date: str,
+        attempts: int | None = None,
+        prepared_rows: list[dict[str, Any]] | None = None,
+    ) -> dict[str, Any]:
         trade_date = yyyymmdd(trade_date)
         batch_id = self.next_batch_id(dataset, trade_date)
         max_attempts = attempts or self.settings.max_publish_attempts
@@ -190,12 +344,15 @@ class Pipeline:
         rows: list[dict[str, Any]] = []
         try:
             self._set_batch(batch_id, dataset, trade_date, "fetching", 1)
-            rows = retry_call(
+            if prepared_rows is None:
-                lambda: self._fetch_dataset(dataset, trade_date),
+                rows = retry_call(
-                attempts=max_attempts,
+                    lambda: self._fetch_dataset(dataset, trade_date),
-                base_delay=0.05,
+                    attempts=max_attempts,
-                sleeper=lambda _d: None if attempts == 1 else time.sleep(_d),
+                    base_delay=0.05,
-            )
+                    sleeper=lambda _d: None if attempts == 1 else time.sleep(_d),
+                )
+            else:
+                rows = list(prepared_rows)
             self._stage(dataset, batch_id, rows)
             self._set_batch(batch_id, dataset, trade_date, "staged", 1, rows_in=len(rows), rows_out=len(rows))
             self._set_batch(batch_id, dataset, trade_date, "validating", 1)
@@ -224,14 +381,62 @@ class Pipeline:
             self._set_batch(batch_id, dataset, trade_date, "failed", 1, error=str(exc), finished=True)
             raise
 
+    def missing_official_datasets(self, trade_date: str) -> list[str]:
+        """Official datasets without an active publication for the date."""
+        day = yyyymmdd(trade_date)
+        placeholders = ",".join("?" for _ in OFFICIAL_DATASETS)
+        rows = self.db.fetchall(
+            f"SELECT dataset FROM publications WHERE trade_date = ? AND dataset IN ({placeholders})",
+            (day, *sorted(OFFICIAL_DATASETS)),
+        )
+        published = {str(row["dataset"]) for row in rows}
+        return [dataset for dataset in sorted(OFFICIAL_DATASETS) if dataset not in published]
+
+    def run_eod_missing(self, trade_date: str) -> dict[str, Any]:
+        """Fetch/publish every official dataset still missing for the date.
+
+        Idempotent: datasets with an existing publication are skipped, so
+        repeats never overwrite the current official batch. Per-dataset
+        failures are collected instead of aborting the remaining datasets.
+        """
+        return self._run_eod_datasets(tuple(sorted(OFFICIAL_DATASETS)), trade_date)
+
     def run_eod_batch_a(self, trade_date: str) -> dict[str, Any]:
-        results = {}
+        return self._run_eod_datasets(EOD_A_DATASETS, trade_date)
-        for dataset in ("daily", "valuation", "moneyflow", "auction"):
-            results[dataset] = self.run_dataset(dataset, trade_date)
-        return results
 
     def run_eod_batch_b(self, trade_date: str) -> dict[str, Any]:
-        return {"index_daily": self.run_dataset("index_daily", trade_date)}
+        return self._run_eod_datasets(EOD_B_DATASETS, trade_date)
+
+    def _run_eod_datasets(self, datasets: tuple[str, ...], trade_date: str) -> dict[str, Any]:
+        day = yyyymmdd(trade_date)
+        results: dict[str, Any] = {}
+        for dataset in datasets:
+            if self.active_batch(dataset, day) is not None:
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "skipped",
+                    "reason": "already_published",
+                }
+                continue
+            try:
+                results[dataset] = self.run_dataset(dataset, day)
+            except Exception as exc:
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "error": str(exc),
+                }
+        return results
+
+    @staticmethod
+    def eod_failures(results: dict[str, Any]) -> list[str]:
+        return [
+            f"{name}: {item.get('error')}"
+            for name, item in results.items()
+            if isinstance(item, dict) and item.get("state") == "failed"
+        ]
 
     def validate(self, dataset: str, batch_id: str, trade_date: str, rows: list[dict[str, Any]]) -> dict[str, Any]:
         quality = self.settings.quality

xiaobai-datahub/datahub/scheduler.py

@@ -2,7 +2,7 @@ from __future__ import annotations
 
 import threading
 from collections.abc import Callable
-from datetime import datetime, time
+from datetime import datetime, time, timedelta
 from typing import Any
 
 from datahub.db import HubDB
@@ -14,6 +14,8 @@ LOGGER = get_logger()
 
 JobFn = Callable[[str], Any]
 
+EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry"}
+
 
 def is_open_day(db: HubDB, day: str) -> bool:
     row = db.fetchone(
@@ -25,8 +27,19 @@ def is_open_day(db: HubDB, day: str) -> bool:
     return int(row["is_open"]) == 1
 
 
+def _hhmm(value: str) -> time:
+    return datetime.strptime(value, "%H:%M").time()
+
+
 class Scheduler:
-    """Calendar-driven in-process scheduler. Non-trading days skip EOD fetches."""
+    """Calendar-driven in-process scheduler. Non-trading days skip EOD fetches.
+
+    EOD datasets that failed to publish (e.g. upstream not ready at 15:05)
+    are retried automatically every ``eod_retry_interval_minutes`` between
+    ``eod_retry_start`` and ``eod_retry_cutoff``. Progress is persisted in
+    ``eod_progress`` so a container restart catches up instead of waiting
+    for the next day, and completed days are never re-fetched.
+    """
 
     def __init__(self, db: HubDB, pipeline: Pipeline, jobs: dict[str, JobFn] | None = None) -> None:
         self.db = db
@@ -35,12 +48,15 @@ class Scheduler:
             "precheck": self._precheck,
             "eod_a": self._eod_a,
             "eod_b": self._eod_b,
+            "eod_retry": self._eod_retry,
             "cleanup": self._cleanup,
             "backup": self._backup,
+            "history_backfill": self._history_backfill,
         }
         self._stop = threading.Event()
         self._thread: threading.Thread | None = None
         self._fired: set[tuple[str, str, str]] = set()
+        self._eod_lock = threading.Lock()
 
     def start(self, interval_seconds: float = 30.0) -> None:
         if self._thread and self._thread.is_alive():
@@ -62,9 +78,9 @@ class Scheduler:
             self._thread.join(timeout)
 
     def tick(self, clock: datetime | None = None) -> list[str]:
-        now = clock or now_shanghai()
+        now = now_shanghai(clock)
         day = yyyymmdd(now)
-        current = now.timetz() if False else now.time()
+        current = now.time()
         ran: list[str] = []
         plan = [
             ("precheck", time(8, 45)),
@@ -84,14 +100,183 @@ class Scheduler:
                 self._fired.add(key)
                 continue
             self._fired.add(key)
-            self.run_job(job_id, day)
+            if job_id in {"eod_a", "eod_b"}:
+                # Record the attempt before running: even a crash must not
+                # hide that today's first EOD try already happened.
+                self._record_eod_attempt(day, now)
+            try:
+                self.run_job(job_id, day)
+            except Exception:
+                if job_id not in {"eod_a", "eod_b"}:
+                    raise
+                # Keep the tick alive; evening retries take over.
+                LOGGER.exception("scheduled job %s failed for %s", job_id, day)
             ran.append(job_id)
+            if job_id in {"eod_a", "eod_b"}:
+                self._settle_eod(day)
+        ran.extend(self._eod_retry_tick(now, day, open_day))
         return ran
 
+    # ------------------------------------------------------------------
+    # EOD retry window
+    # ------------------------------------------------------------------
+
+    def _eod_retry_tick(self, now: datetime, day: str, open_day: bool) -> list[str]:
+        if not open_day:
+            return []
+        settings = self.pipeline.settings
+        current = now.time()
+        start = _hhmm(settings.eod_retry_start)
+        cutoff = _hhmm(settings.eod_retry_cutoff)
+        interval = timedelta(minutes=settings.eod_retry_interval_minutes)
+        missing = self.pipeline.missing_official_datasets(day)
+        row = self.eod_progress(day)
+
+        if not missing:
+            if row is None or row["state"] != "done":
+                self._save_eod_progress(day, state="done", finished_at=isoformat(now))
+            return []
+        if current < start:
+            return []
+        if row and row["state"] == "cutoff_failed":
+            return []
+        if current >= cutoff:
+            detail = "截止时间已到，缺失数据集: " + ",".join(missing)
+            self._save_eod_progress(day, state="cutoff_failed", finished_at=isoformat(now), detail=detail)
+            with self.db.write() as connection:
+                connection.execute(
+                    "INSERT INTO job_runs(job_id, state, started_at, finished_at, error, attempt, detail)"
+                    " VALUES ('eod_retry','failed',?,?,?,?,?)",
+                    (isoformat(now), isoformat(now), detail, int((row or {}).get("attempts") or 0), "eod cutoff reached"),
+                )
+            LOGGER.warning(
+                "eod retry window closed without data",
+                extra={"hub": {"trade_date": day, "missing": missing, "reason": "eod_cutoff"}},
+            )
+            return []
+        last = None
+        if row and row["last_attempt_at"]:
+            try:
+                last = datetime.fromisoformat(str(row["last_attempt_at"]))
+            except ValueError:
+                last = None
+        if last is not None and now_shanghai(last).replace(tzinfo=None) + interval > now.replace(tzinfo=None):
+            return []
+        if "eod_retry" not in self.jobs:
+            return []
+        self._record_eod_attempt(day, now)
+        ran = []
+        try:
+            self.run_job("eod_retry", day)
+        except Exception:
+            # job_runs already carries the failure; the window keeps retrying.
+            LOGGER.warning("eod retry failed for %s", day, exc_info=True)
+        ran.append("eod_retry")
+        self._settle_eod(day)
+        return ran
+
+    def _settle_eod(self, day: str) -> None:
+        """Flip the day to done as soon as every official dataset is published."""
+        if not self.pipeline.missing_official_datasets(day):
+            row = self.eod_progress(day)
+            if row is None or row["state"] != "done":
+                self._save_eod_progress(day, state="done", finished_at=isoformat())
+
+    def eod_progress(self, day: str) -> dict[str, Any] | None:
+        return self.db.fetchone("SELECT * FROM eod_progress WHERE trade_date = ?", (day,))
+
+    def eod_status(self, trade_date: str | None = None, clock: datetime | None = None) -> dict[str, Any]:
+        """Human/admin facing view: 等待上游 / 下次重试 / 已成功 / 已截止失败."""
+        day = yyyymmdd(trade_date or now_shanghai(clock))
+        now = now_shanghai(clock)
+        row = self.eod_progress(day)
+        open_day = is_open_day(self.db, day)
+        missing = self.pipeline.missing_official_datasets(day)
+        if row and row["state"] == "done":
+            state = "done"
+        elif not open_day:
+            state = "closed_day"
+        elif not missing:
+            state = "done"
+        elif row and row["state"] == "cutoff_failed":
+            state = "cutoff_failed"
+        elif now.time() < _hhmm("15:05"):
+            state = "pending_first_attempt"
+        else:
+            state = "waiting_upstream"
+        return {
+            "trade_date": day,
+            "is_open_day": open_day,
+            "state": state,
+            "missing_datasets": missing,
+            "attempts": int((row or {}).get("attempts") or 0),
+            "last_attempt_at": (row or {}).get("last_attempt_at"),
+            "next_retry_at": (row or {}).get("next_retry_at") if state == "waiting_upstream" else None,
+            "finished_at": (row or {}).get("finished_at"),
+            "detail": (row or {}).get("detail"),
+        }
+
+    def _record_eod_attempt(self, day: str, now: datetime) -> None:
+        row = self.eod_progress(day)
+        attempts = int((row or {}).get("attempts") or 0) + 1
+        interval = self.pipeline.settings.eod_retry_interval_minutes
+        self._save_eod_progress(
+            day,
+            state="waiting_upstream",
+            attempts=attempts,
+            last_attempt_at=isoformat(now),
+            next_retry_at=isoformat(now + timedelta(minutes=interval)),
+        )
+
+    def _save_eod_progress(self, day: str, **fields: Any) -> None:
+        columns = [
+            "trade_date", "state", "attempts", "last_attempt_at",
+            "next_retry_at", "finished_at", "detail", "updated_at",
+        ]
+        with self.db.write() as connection:
+            existing = connection.execute(
+                "SELECT trade_date FROM eod_progress WHERE trade_date = ?",
+                (day,),
+            ).fetchone()
+            if existing is None:
+                payload = {name: None for name in columns}
+                payload.update({"trade_date": day, "state": "waiting_upstream", "attempts": 0})
+                payload.update(fields)
+                payload["updated_at"] = isoformat()
+                placeholders = ",".join("?" for _ in columns)
+                connection.execute(
+                    f"INSERT INTO eod_progress({','.join(columns)}) VALUES ({placeholders})",
+                    tuple(payload[name] for name in columns),
+                )
+            else:
+                assignments = ", ".join(f"{name} = ?" for name in fields)
+                connection.execute(
+                    f"UPDATE eod_progress SET {assignments}, updated_at = ? WHERE trade_date = ?",
+                    (*fields.values(), isoformat(), day),
+                )
+
+    # ------------------------------------------------------------------
+    # Job execution
+    # ------------------------------------------------------------------
+
     def run_job(self, job_id: str, trade_date: str) -> dict[str, Any]:
         fn = self.jobs.get(job_id)
         if fn is None:
             raise KeyError(job_id)
+        if job_id in EOD_JOB_IDS:
+            if not self._eod_lock.acquire(blocking=False):
+                return {
+                    "job_id": job_id,
+                    "state": "skipped",
+                    "detail": "another EOD job is already running",
+                }
+            try:
+                return self._run_job(fn, job_id, trade_date)
+            finally:
+                self._eod_lock.release()
+        return self._run_job(fn, job_id, trade_date)
+
+    def _run_job(self, fn: JobFn, job_id: str, trade_date: str) -> dict[str, Any]:
         started = isoformat()
         run_id = None
         with self.db.write() as connection:
@@ -102,6 +287,10 @@ class Scheduler:
             run_id = cur.lastrowid
         try:
             result = fn(trade_date) or {}
+            if isinstance(result, dict):
+                failures = self.pipeline.eod_failures(result) if job_id in EOD_JOB_IDS else []
+                if failures:
+                    raise RuntimeError("; ".join(failures))
             with self.db.write() as connection:
                 connection.execute(
                     "UPDATE job_runs SET state=?, finished_at=?, rows_out=?, detail=? WHERE id=?",
@@ -125,6 +314,12 @@ class Scheduler:
     def _eod_b(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.run_eod_batch_b(trade_date)
 
+    def _eod_retry(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_missing(trade_date)
+
+    def _history_backfill(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.backfill_history(trade_date)
+
     def _cleanup(self, trade_date: str) -> dict[str, Any]:
         result = self.pipeline.cleanup()
         if now_shanghai().weekday() == 6:

xiaobai-datahub/datahub/serving.py

@@ -6,6 +6,7 @@ from urllib.parse import parse_qs
 
 from datahub import SCHEMA_VERSION
 from datahub.codes import resolve_code
+from datahub.coverage import calendar_coverage, point_coverage, published_range_coverage
 from datahub.db import HubDB
 from datahub.normalize import qfq_bar
 from datahub.numbers import finite_number
@@ -129,7 +130,8 @@ class V1API:
             }
             for row in rows
         ]
-        return envelope(items, self._official_meta("calendar", end if items else start, source="tushare:trade_cal"))
+        meta = self._official_meta("calendar", end if items else start, source="tushare:trade_cal")
+        return envelope(items, attach_coverage(meta, calendar_coverage(self.db, start, end)))
 
     def stocks(self, updated_since: str, q: dict[str, str]) -> dict[str, Any]:
         limit, offset = self._page(q)
@@ -272,7 +274,7 @@ class V1API:
                 "staleness_seconds": 0,
                 "state": pub["state"],
             }
-            return envelope(rows, meta)
+            return envelope(rows, attach_coverage(meta, point_coverage(start, dataset)))
         # multi-day: walk published dates
         pubs = self.db.fetchall(
             "SELECT * FROM publications WHERE dataset = ? AND trade_date >= ? AND trade_date <= ? ORDER BY trade_date",
@@ -294,17 +296,28 @@ class V1API:
         if adjust == "qfq" and dataset == "daily":
             sliced = self._apply_qfq(sliced)
         last = pubs[-1]
+        coverage = published_range_coverage(
+            self.db,
+            dataset,
+            start,
+            end,
+            ts_code=ts_code,
+            table=table,
+        )
         return envelope(
             sliced,
-            {
+            attach_coverage(
-                "tier": "official",
+                {
-                "trade_date": last["trade_date"],
+                    "tier": "official",
-                "published_at": last["published_at"],
+                    "trade_date": last["trade_date"],
-                "source": source,
+                    "published_at": last["published_at"],
-                "batch_id": last["active_batch"],
+                    "source": source,
-                "stale": False,
+                    "batch_id": last["active_batch"],
-                "staleness_seconds": 0,
+                    "stale": False,
-            },
+                    "staleness_seconds": 0,
+                },
+                coverage,
+            ),
         )
 
     def _apply_qfq(self, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
@@ -359,6 +372,13 @@ def add_default(days: int) -> str:
     return (now_shanghai() + timedelta(days=days)).strftime("%Y%m%d")
 
 
+def attach_coverage(meta: dict[str, Any], coverage: dict[str, Any]) -> dict[str, Any]:
+    merged = dict(meta)
+    merged["coverage"] = coverage
+    merged["incomplete"] = not bool(coverage.get("complete"))
+    return merged
+
+
 def parse_query(raw: str) -> dict[str, list[str]]:
     return parse_qs(raw, keep_blank_values=True)
 

xiaobai-datahub/datahub/settings.py

@@ -48,6 +48,26 @@ class Settings:
     def list_limit_max(self) -> int:
         return int(self.quality.get("list_limit_max") or 5000)
 
+    @property
+    def calendar_start(self) -> str:
+        return str(self.quality.get("calendar_start") or "20160101")
+
+    @property
+    def index_history_trading_days(self) -> int:
+        return int(self.quality.get("index_history_trading_days") or 260)
+
+    @property
+    def eod_retry_start(self) -> str:
+        return str(self.quality.get("eod_retry_start") or "15:15")
+
+    @property
+    def eod_retry_interval_minutes(self) -> int:
+        return int(self.quality.get("eod_retry_interval_minutes") or 30)
+
+    @property
+    def eod_retry_cutoff(self) -> str:
+        return str(self.quality.get("eod_retry_cutoff") or "23:30")
+
 
 def load_settings(
     env: dict[str, str] | None = None,

xiaobai-datahub/datahub/timeutil.py

@@ -58,6 +58,16 @@ def add_days(trade_date: str, days: int) -> str:
     return (parse_trade_date(trade_date) + timedelta(days=days)).strftime("%Y%m%d")
 
 
+def iter_yyyymmdd(start: str, end: str):
+    cursor = parse_trade_date(start)
+    last = parse_trade_date(end)
+    if cursor > last:
+        return
+    while cursor <= last:
+        yield cursor.strftime("%Y%m%d")
+        cursor += timedelta(days=1)
+
+
 def utc_timestamp(value: Any) -> str:
     if isinstance(value, datetime):
         return isoformat(value)

xiaobai-datahub/tests/fixtures.py

@@ -51,7 +51,17 @@ RAW = {
 def fake_transport(api_name: str, params: dict, fields: str):
     if api_name == "index_daily":
         code = params.get("ts_code")
-        return [row for row in RAW["index_daily"] if row["ts_code"] == code]
+        rows = [row for row in RAW["index_daily"] if row["ts_code"] == code]
+        trade_date = str(params.get("trade_date") or "")
+        start = str(params.get("start_date") or "")
+        end = str(params.get("end_date") or "")
+        if trade_date:
+            rows = [row for row in rows if row["trade_date"] == trade_date]
+        if start:
+            rows = [row for row in rows if row["trade_date"] >= start]
+        if end:
+            rows = [row for row in rows if row["trade_date"] <= end]
+        return rows
     if api_name == "trade_cal":
         start = str(params.get("start_date") or "")
         end = str(params.get("end_date") or "99999999")

xiaobai-datahub/tests/test_api.py

@@ -107,6 +107,9 @@ class ApiContractTests(unittest.TestCase):
             self.assertIn("data", body)
             self.assertIn("meta", body)
             self.assertIn("tier", body["meta"])
+            if "calendar" in path or "bars" in path or "indexes" in path or "valuation" in path or "moneyflow" in path or "auction" in path:
+                self.assertIn("coverage", body["meta"])
+                self.assertIn("incomplete", body["meta"])
 
     def test_qfq_matches_formula(self) -> None:
         _, none = self._get(f"/v1/bars/daily?date={TRADE_DATE}&code=600000.SH&adjust=none", token=self.token)

xiaobai-datahub/tests/test_eod_retry.py

@@ -0,0 +1,219 @@
+from __future__ import annotations
+
+import unittest
+from datetime import datetime
+from pathlib import Path
+import tempfile
+
+from datahub.adapters.tushare import TushareAdapter
+from datahub.crypto import SecretVault
+from datahub.db import HubDB
+from datahub.pipeline import Pipeline
+from datahub.scheduler import Scheduler
+from datahub.settings import Settings
+from datahub.timeutil import SHANGHAI
+from tests.fixtures import fake_transport
+
+OFFICIAL = {"daily", "valuation", "moneyflow", "auction", "index_daily"}
+
+
+class DelayedTransport:
+    """Upstream that only returns rows for dates it has "published" yet."""
+
+    DATE_APIS = {"daily", "daily_basic", "adj_factor", "moneyflow", "stk_auction", "index_daily"}
+
+    def __init__(self, ready_dates: set[str]) -> None:
+        self.ready = set(ready_dates)
+        self.calls: list[str] = []
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        self.calls.append(api_name)
+        if api_name in self.DATE_APIS:
+            trade_date = str(params.get("trade_date") or "")
+            if trade_date and trade_date not in self.ready:
+                return []
+        return fake_transport(api_name, params, fields)
+
+
+def clock_at(day: str, hh: int, mm: int) -> datetime:
+    return datetime(int(day[:4]), int(day[4:6]), int(day[6:8]), hh, mm, tzinfo=SHANGHAI)
+
+
+class EodRetryTests(unittest.TestCase):
+    def _make(self, ready_dates: set[str]):
+        tmp = tempfile.TemporaryDirectory()
+        self.addCleanup(tmp.cleanup)
+        db = HubDB(Path(tmp.name) / "hub.db")
+        transport = DelayedTransport(ready_dates)
+        adapter = TushareAdapter("x", transport=transport)
+        settings = Settings(
+            encryption_key=SecretVault.generate_key(),
+            db_path=db.path,
+            backup_dir=Path(tmp.name) / "backups",
+        )
+        pipe = Pipeline(db, adapter, settings)
+        pipe.ingest_reference("20240902")
+        sched = Scheduler(db, pipe)
+        return db, transport, pipe, sched
+
+    def _job_runs(self, db: HubDB, job_id: str) -> list[dict]:
+        return db.fetchall("SELECT * FROM job_runs WHERE job_id = ? ORDER BY id", (job_id,))
+
+    def _batches(self, db: HubDB, day: str) -> list[dict]:
+        return db.fetchall("SELECT * FROM batches WHERE trade_date = ?", (day,))
+
+    @staticmethod
+    def _batch_ids(db: HubDB, day: str) -> set[str]:
+        return {str(row["batch_id"]) for row in db.fetchall("SELECT batch_id FROM batches WHERE trade_date = ?", (day,))}
+
+    @staticmethod
+    def _eod_calls(transport: DelayedTransport) -> list[str]:
+        return [name for name in transport.calls if name in DelayedTransport.DATE_APIS]
+
+    def _published(self, db: HubDB, day: str) -> set[str]:
+        rows = db.fetchall("SELECT dataset FROM publications WHERE trade_date = ?", (day,))
+        return {str(row["dataset"]) for row in rows}
+
+    def test_first_empty_then_retry_succeeds(self) -> None:
+        day = "20240902"
+        db, transport, pipe, sched = self._make(set())
+
+        sched.tick(clock_at(day, 15, 5))   # eod_a: upstream empty -> failed
+        sched.tick(clock_at(day, 15, 10))  # eod_b: upstream empty -> failed
+        self.assertEqual(self._published(db, day), set())  # quality gate held
+
+        sched.tick(clock_at(day, 15, 20))  # inside window, but <30min since 15:10
+        self.assertEqual(self._job_runs(db, "eod_retry"), [])
+        status = sched.eod_status(day, clock=clock_at(day, 15, 20))
+        self.assertEqual(status["state"], "waiting_upstream")
+        self.assertTrue(status["next_retry_at"])
+        self.assertEqual(status["missing_datasets"], sorted(OFFICIAL))
+
+        sched.tick(clock_at(day, 15, 40))  # retry #1, still empty
+        runs = self._job_runs(db, "eod_retry")
+        self.assertEqual(len(runs), 1)
+        self.assertEqual(runs[0]["state"], "failed")
+        self.assertEqual(self._published(db, day), set())
+
+        transport.ready.add(day)
+        sched.tick(clock_at(day, 16, 10))  # retry #2 succeeds
+        self.assertEqual(self._published(db, day), OFFICIAL)
+        self.assertEqual(sched.eod_status(day, clock=clock_at(day, 16, 10))["state"], "done")
+        progress = db.fetchone("SELECT * FROM eod_progress WHERE trade_date = ?", (day,))
+        self.assertEqual(progress["state"], "done")
+        self.assertEqual(progress["attempts"], 4)  # eod_a + eod_b + 2 retries
+
+        # success stops all further same-day requests
+        batches_before = len(self._batches(db, day))
+        calls_before = len(transport.calls)
+        sched.tick(clock_at(day, 17, 0))
+        sched.tick(clock_at(day, 23, 0))
+        self.assertEqual(len(self._job_runs(db, "eod_retry")), 2)
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+        self.assertEqual(len(transport.calls), calls_before)
+
+    def test_never_ready_marks_cutoff_failed_and_stops(self) -> None:
+        day = "20240902"
+        db, transport, pipe, sched = self._make(set())
+        sched.tick(clock_at(day, 15, 5))
+        sched.tick(clock_at(day, 15, 10))
+        sched.tick(clock_at(day, 15, 40))
+        sched.tick(clock_at(day, 16, 10))
+        sched.tick(clock_at(day, 23, 29))
+        self.assertEqual(len(self._job_runs(db, "eod_retry")), 3)
+
+        sched.tick(clock_at(day, 23, 35))  # past cutoff 23:30
+        status = sched.eod_status(day, clock=clock_at(day, 23, 35))
+        self.assertEqual(status["state"], "cutoff_failed")
+        cutoff_runs = [r for r in self._job_runs(db, "eod_retry") if "截止" in str(r["error"])]
+        self.assertEqual(len(cutoff_runs), 1)
+        self.assertEqual(self._published(db, day), set())
+
+        attempts = db.fetchone("SELECT attempts FROM eod_progress WHERE trade_date = ?", (day,))["attempts"]
+        sched.tick(clock_at(day, 23, 59))
+        self.assertEqual(
+            db.fetchone("SELECT attempts FROM eod_progress WHERE trade_date = ?", (day,))["attempts"],
+            attempts,
+        )
+        self.assertEqual(len(self._job_runs(db, "eod_retry")), 4)  # 3 retries + 1 cutoff record
+        self.assertEqual(self._published(db, day), set())
+
+    def test_restart_catches_up_without_overwriting(self) -> None:
+        day = "20240902"
+        db, transport, pipe, sched = self._make({day})
+        sched.tick(clock_at(day, 15, 5))   # eod_a publishes 4 datasets
+        sched.tick(clock_at(day, 15, 10))  # eod_b publishes index
+        self.assertEqual(self._published(db, day), OFFICIAL)
+        active = db.fetchall("SELECT dataset, active_batch FROM publications WHERE trade_date = ?", (day,))
+        active_map = {row["dataset"]: row["active_batch"] for row in active}
+        batches_before = self._batch_ids(db, day)
+        calls_before = self._eod_calls(transport)
+
+        # container restart: fresh scheduler, missed-time catch-up fires eod_a/eod_b
+        sched2 = Scheduler(db, pipe)
+        ran = sched2.tick(clock_at(day, 21, 0))
+        self.assertIn("eod_a", ran)
+        self.assertIn("eod_b", ran)
+        self.assertNotIn("eod_retry", ran)
+        self.assertEqual(self._published(db, day), OFFICIAL)
+        after = db.fetchall("SELECT dataset, active_batch FROM publications WHERE trade_date = ?", (day,))
+        self.assertEqual({row["dataset"]: row["active_batch"] for row in after}, active_map)
+        self.assertEqual(self._batch_ids(db, day), batches_before)  # no duplicate batches
+        self.assertEqual(self._eod_calls(transport), calls_before)  # no duplicate upstream EOD calls
+        self.assertEqual(sched2.eod_status(day, clock=clock_at(day, 21, 0))["state"], "done")
+
+    def test_restart_with_partial_publish_only_fetches_missing(self) -> None:
+        day = "20240902"
+        db, transport, pipe, sched = self._make({day})
+        sched.tick(clock_at(day, 15, 5))  # eod_a publishes 4; container "crashes" before eod_b
+        self.assertEqual(self._published(db, day), {"daily", "valuation", "moneyflow", "auction"})
+        batches_before = self._batch_ids(db, day)
+
+        sched2 = Scheduler(db, pipe)
+        ran = sched2.tick(clock_at(day, 15, 20))  # restart: eod_b catch-up, eod_a all skipped
+        self.assertIn("eod_b", ran)
+        self.assertEqual(self._published(db, day), OFFICIAL)
+        new_ids = self._batch_ids(db, day) - batches_before
+        new_datasets = {str(b["dataset"]) for b in self._batches(db, day) if str(b["batch_id"]) in new_ids}
+        self.assertEqual(new_datasets, {"index_daily"})
+        self.assertEqual(sched2.eod_status(day, clock=clock_at(day, 15, 20))["state"], "done")
+
+    def test_closed_day_skips_all_eod_work(self) -> None:
+        day = "20240907"  # closed in fixture calendar
+        db, transport, pipe, sched = self._make(set())
+        for hh, mm in ((15, 5), (15, 10), (15, 40), (16, 10), (20, 0), (23, 40)):
+            ran = sched.tick(clock_at(day, hh, mm))
+            self.assertNotIn("eod_retry", ran)
+        eod_runs = db.fetchall("SELECT * FROM job_runs WHERE job_id LIKE 'eod%'")
+        self.assertEqual(eod_runs, [])
+        self.assertIsNone(db.fetchone("SELECT * FROM eod_progress WHERE trade_date = ?", (day,)))
+        self.assertEqual(self._published(db, day), set())
+        self.assertEqual(sched.eod_status(day, clock=clock_at(day, 20, 0))["state"], "closed_day")
+
+    def test_duplicate_and_concurrent_execution_are_safe(self) -> None:
+        day = "20240902"
+        db, transport, pipe, sched = self._make({day})
+        sched.tick(clock_at(day, 15, 5))
+        sched.tick(clock_at(day, 15, 10))
+        self.assertEqual(self._published(db, day), OFFICIAL)
+        batches_before = len(self._batches(db, day))
+        calls_before = len(transport.calls)
+
+        out = sched.run_job("eod_retry", day)  # manual duplicate run
+        self.assertEqual(out["state"], "ok")
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+        self.assertEqual(len(transport.calls), calls_before)
+
+        sched._eod_lock.acquire()  # simulate an in-flight EOD job
+        try:
+            busy = sched.run_job("eod_retry", day)
+            self.assertEqual(busy["state"], "skipped")
+            busy_a = sched.run_job("eod_a", day)
+            self.assertEqual(busy_a["state"], "skipped")
+        finally:
+            sched._eod_lock.release()
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_history_backfill.py

@@ -0,0 +1,236 @@
+from __future__ import annotations
+
+import unittest
+from datetime import date, timedelta
+
+from datahub.coverage import calendar_coverage, point_coverage, published_range_coverage
+from datahub.serving import V1API
+from tests.fixtures import TRADE_DATE, fake_transport
+from tests.test_pipeline import make_pipeline
+
+
+def history_transport(open_dates: list[str], extra_closed: list[str] | None = None):
+    open_set = set(open_dates)
+    start = date(int(open_dates[0][:4]), int(open_dates[0][4:6]), int(open_dates[0][6:8]))
+    end = date(int(open_dates[-1][:4]), int(open_dates[-1][4:6]), int(open_dates[-1][6:8]))
+    calendar = []
+    cursor = start
+    while cursor <= end:
+        compact = cursor.strftime("%Y%m%d")
+        calendar.append(
+            {
+                "exchange": "SSE",
+                "cal_date": compact,
+                "is_open": 1 if compact in open_set else 0,
+                "pretrade_date": compact,
+            }
+        )
+        cursor += timedelta(days=1)
+    for day in extra_closed or []:
+        calendar.append(
+            {"exchange": "SSE", "cal_date": day, "is_open": 0, "pretrade_date": open_dates[0]}
+        )
+    index_codes = ("000001.SH", "399001.SZ", "399006.SZ", "000300.SH")
+    index_rows = []
+    for ts_code in index_codes:
+        for day in open_dates:
+            index_rows.append(
+                {
+                    "ts_code": ts_code,
+                    "trade_date": day,
+                    "open": 100,
+                    "high": 101,
+                    "low": 99,
+                    "close": 100.5,
+                    "pct_chg": 0.1,
+                    "vol": 10.0,
+                    "amount": 20.0,
+                }
+            )
+
+    def transport(api_name, params, fields):
+        if api_name == "trade_cal":
+            start = str(params.get("start_date") or "")
+            end = str(params.get("end_date") or "99999999")
+            return [row for row in calendar if start <= row["cal_date"] <= end]
+        if api_name == "index_daily":
+            code = params.get("ts_code")
+            rows = [row for row in index_rows if row["ts_code"] == code]
+            trade_date = str(params.get("trade_date") or "")
+            start = str(params.get("start_date") or "")
+            end = str(params.get("end_date") or "")
+            if trade_date:
+                rows = [row for row in rows if row["trade_date"] == trade_date]
+            if start:
+                rows = [row for row in rows if row["trade_date"] >= start]
+            if end:
+                rows = [row for row in rows if row["trade_date"] <= end]
+            return rows
+        return fake_transport(api_name, params, fields)
+
+    return transport
+
+
+def consecutive_open_days(end: str, count: int) -> list[str]:
+    cursor = date(int(end[:4]), int(end[4:6]), int(end[6:8]))
+    days: list[str] = []
+    while len(days) < count:
+        if cursor.weekday() < 5:
+            days.append(cursor.strftime("%Y%m%d"))
+        cursor -= timedelta(days=1)
+    return sorted(days)
+
+
+class CoverageApiTests(unittest.TestCase):
+    def test_calendar_marks_holes_incomplete(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        api = V1API(pipe.db, pipe, pipe.settings)
+        payload = api.handle("/v1/calendar", {"from": ["20240901"], "to": ["20240907"]})
+        self.assertTrue(payload["meta"]["incomplete"])
+        self.assertFalse(payload["meta"]["coverage"]["complete"])
+        self.assertGreater(payload["meta"]["coverage"]["missing_count"], 0)
+        self.assertIn("20240901", payload["meta"]["coverage"]["missing_sample"])
+
+    def test_calendar_complete_when_every_day_present(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        api = V1API(pipe.db, pipe, pipe.settings)
+        payload = api.handle("/v1/calendar", {"from": ["20240902"], "to": ["20240903"]})
+        self.assertFalse(payload["meta"]["incomplete"])
+        self.assertTrue(payload["meta"]["coverage"]["complete"])
+        self.assertEqual(payload["meta"]["coverage"]["expected_count"], 2)
+        self.assertEqual(len(payload["data"]), 2)
+
+    def test_index_range_incomplete_without_history(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        pipe.run_dataset("index_daily", TRADE_DATE)
+        api = V1API(pipe.db, pipe, pipe.settings)
+        payload = api.handle(
+            "/v1/indexes/bars",
+            {"from": ["20240902"], "to": ["20240903"], "code": ["000001.SH"]},
+        )
+        self.assertTrue(payload["meta"]["incomplete"])
+        self.assertFalse(payload["meta"]["coverage"]["complete"])
+        self.assertEqual(payload["meta"]["coverage"]["available_count"], 1)
+        self.assertIn("20240903", payload["meta"]["coverage"]["missing_sample"])
+
+    def test_index_point_query_stays_complete(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        pipe.run_dataset("index_daily", TRADE_DATE)
+        api = V1API(pipe.db, pipe, pipe.settings)
+        payload = api.handle("/v1/indexes/bars", {"date": [TRADE_DATE], "code": ["000001.SH"]})
+        self.assertFalse(payload["meta"]["incomplete"])
+        self.assertTrue(payload["meta"]["coverage"]["complete"])
+        self.assertEqual(payload["meta"]["coverage"]["kind"], "point")
+
+    def test_daily_range_incomplete_without_stock_history(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        pipe.run_dataset("daily", TRADE_DATE)
+        api = V1API(pipe.db, pipe, pipe.settings)
+        payload = api.handle(
+            "/v1/bars/daily",
+            {"from": ["20240902"], "to": ["20240903"], "code": ["600000.SH"]},
+        )
+        self.assertTrue(payload["meta"]["incomplete"])
+        self.assertFalse(payload["meta"]["coverage"]["complete"])
+
+
+class HistoryBackfillTests(unittest.TestCase):
+    def test_index_history_is_idempotent_and_covers_requested_days(self) -> None:
+        open_dates = consecutive_open_days(TRADE_DATE, 5)
+        pipe, db = make_pipeline(quality={"index_history_trading_days": 5, "calendar_start": open_dates[0]})
+        pipe.adapter._transport = history_transport(open_dates)
+        first = pipe.backfill_history(TRADE_DATE, index_days=5)
+        self.assertTrue(first["ok"])
+        self.assertEqual(first["calendar"]["calendar_from"], open_dates[0])
+        self.assertEqual(first["index_daily"]["requested_days"], 5)
+        self.assertEqual(len(first["index_daily"]["published"]), 5)
+        self.assertEqual(first["index_daily"]["skipped"], [])
+        pubs = db.fetchall("SELECT trade_date FROM publications WHERE dataset='index_daily'")
+        self.assertEqual(sorted(row["trade_date"] for row in pubs), open_dates)
+
+        second = pipe.backfill_index_history(TRADE_DATE, trading_days=5)
+        self.assertTrue(second["ok"])
+        self.assertEqual(second["published"], [])
+        self.assertEqual(second["skipped"], open_dates)
+
+        api = V1API(db, pipe, pipe.settings)
+        payload = api.handle(
+            "/v1/indexes/bars",
+            {"from": [open_dates[0]], "to": [open_dates[-1]], "code": ["000001.SH"]},
+        )
+        self.assertFalse(payload["meta"]["incomplete"])
+        self.assertEqual(payload["meta"]["coverage"]["available_count"], 5)
+        self.assertEqual(len(payload["data"]), 5)
+
+    def test_index_history_retries_failed_dates_without_dropping_success(self) -> None:
+        open_dates = consecutive_open_days(TRADE_DATE, 3)
+        base = history_transport(open_dates)
+
+        def missing_cyb(api_name, params, fields):
+            if api_name == "index_daily" and params.get("ts_code") == "399006.SZ":
+                raise RuntimeError("upstream down")
+            return base(api_name, params, fields)
+
+        pipe, db = make_pipeline(quality={"index_history_trading_days": 3, "max_publish_attempts": 1})
+        pipe.adapter._transport = missing_cyb
+        first = pipe.backfill_history(TRADE_DATE, calendar_start=open_dates[0], index_days=3)
+        self.assertFalse(first["ok"])
+        self.assertTrue(any(item.get("ts_code") == "399006.SZ" for item in first["index_daily"]["failed"]))
+        published_first = {
+            row["trade_date"]
+            for row in db.fetchall("SELECT trade_date FROM publications WHERE dataset='index_daily'")
+        }
+        self.assertEqual(published_first, set(open_dates))
+
+        pipe.adapter._transport = base
+        retry = pipe.backfill_index_history(TRADE_DATE, trading_days=3)
+        self.assertTrue(retry["ok"])
+        self.assertEqual(len(retry["published"]), 3)
+        for day in open_dates:
+            rows = db.fetchall(
+                """
+                SELECT DISTINCT ts_code FROM eod_index_bars
+                WHERE trade_date = ? AND batch_id = (
+                    SELECT active_batch FROM publications
+                    WHERE dataset='index_daily' AND trade_date = ?
+                )
+                """,
+                (day, day),
+            )
+            self.assertEqual({row["ts_code"] for row in rows}, {"000001.SH", "399001.SZ", "399006.SZ", "000300.SH"})
+
+    def test_prepared_rows_skip_upstream_fetch(self) -> None:
+        pipe, _db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        calls = {"n": 0}
+        original = pipe.adapter._transport
+
+        def counting(api_name, params, fields):
+            calls["n"] += 1
+            return original(api_name, params, fields)
+
+        pipe.adapter._transport = counting
+        rows = pipe.adapter.normalize("index_daily", original("index_daily", {"ts_code": "000001.SH", "trade_date": TRADE_DATE}, ""))
+        before = calls["n"]
+        result = pipe.run_dataset("index_daily", TRADE_DATE, prepared_rows=rows)
+        self.assertEqual(result["rows"], 1)
+        self.assertEqual(calls["n"], before)
+
+    def test_coverage_helpers_point_and_calendar(self) -> None:
+        pipe, db = make_pipeline()
+        pipe.ingest_reference(TRADE_DATE)
+        point = point_coverage(TRADE_DATE, "index_daily")
+        self.assertTrue(point["complete"])
+        cal = calendar_coverage(db, "20240902", "20240903")
+        self.assertTrue(cal["complete"])
+        pub = published_range_coverage(db, "index_daily", "20240902", "20240903")
+        self.assertFalse(pub["complete"])
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_pipeline.py

@@ -69,6 +69,7 @@ class PipelineTests(unittest.TestCase):
         pipe, db = make_pipeline()
         ref = pipe.ingest_reference(TRADE_DATE)
         self.assertEqual(ref["stocks"], 2)
+        self.assertEqual(ref["calendar_from"], "20160101")
         result = pipe.run_dataset("daily", TRADE_DATE)
         self.assertEqual(result["state"], "published")
         self.assertEqual(result["rows"], 2)