From c9892050c34beac47cd704025ea0cff9ed17f1af Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?=E6=80=BB=E5=B7=A5?= Date: Thu, 3 Sep 2026 22:30:15 +0800 Subject: [PATCH] =?UTF-8?q?feat(HEL-435):=20=E7=9B=98=E5=90=8E=E6=9C=AA?= =?UTF-8?q?=E5=87=BA=E6=95=B0=E6=97=B6=E6=99=9A=E9=97=B4=E8=87=AA=E5=8A=A8?= =?UTF-8?q?=E9=87=8D=E8=AF=95=E5=B9=B6=E6=8F=90=E4=BE=9B=E5=AE=89=E5=85=A8?= =?UTF-8?q?=E8=A1=A5=E8=B7=91?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit Co-authored-by: multica-agent --- xiaobai-datahub/admin/app.js | 18 ++ .../config/hub-quality.config.json | 5 +- xiaobai-datahub/datahub/admin_api.py | 2 + xiaobai-datahub/datahub/cli.py | 15 ++ xiaobai-datahub/datahub/db.py | 11 + xiaobai-datahub/datahub/pipeline.py | 60 ++++- xiaobai-datahub/datahub/scheduler.py | 201 +++++++++++++++- xiaobai-datahub/datahub/settings.py | 12 + xiaobai-datahub/tests/test_eod_retry.py | 219 ++++++++++++++++++ 9 files changed, 532 insertions(+), 11 deletions(-) create mode 100644 xiaobai-datahub/tests/test_eod_retry.py diff --git a/xiaobai-datahub/admin/app.js b/xiaobai-datahub/admin/app.js index 8f29b7f..119796f 100644 --- a/xiaobai-datahub/admin/app.js +++ b/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 = `
交易日
${esc(data.trade_date)}
阶段
${esc(data.session_phase)}
今日发布
${data.publications.length}
+
盘后补跑
${esc(eodLabels[eod.state] || eod.state || "-")}
${eodExtra.join(" · ")}
异常批次
${data.anomalies.length}

最近调用

diff --git a/xiaobai-datahub/config/hub-quality.config.json b/xiaobai-datahub/config/hub-quality.config.json index cc9285b..06892ee 100644 --- a/xiaobai-datahub/config/hub-quality.config.json +++ b/xiaobai-datahub/config/hub-quality.config.json @@ -13,5 +13,8 @@ "list_limit_default": 5000, "list_limit_max": 5000, "calendar_start": "20160101", - "index_history_trading_days": 260 + "index_history_trading_days": 260, + "eod_retry_start": "15:15", + "eod_retry_interval_minutes": 30, + "eod_retry_cutoff": "23:30" } diff --git a/xiaobai-datahub/datahub/admin_api.py b/xiaobai-datahub/datahub/admin_api.py index fc5824c..82a8558 100644 --- a/xiaobai-datahub/datahub/admin_api.py +++ b/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,7 @@ 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 备份"}, diff --git a/xiaobai-datahub/datahub/cli.py b/xiaobai-datahub/datahub/cli.py index 03bf328..de142e0 100644 --- a/xiaobai-datahub/datahub/cli.py +++ b/xiaobai-datahub/datahub/cli.py @@ -8,6 +8,7 @@ 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: @@ -17,6 +18,8 @@ def main(argv: list[str] | None = None) -> int: 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() @@ -30,6 +33,18 @@ def main(argv: list[str] | None = None) -> int: 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 diff --git a/xiaobai-datahub/datahub/db.py b/xiaobai-datahub/datahub/db.py index dec81f9..93f430b 100644 --- a/xiaobai-datahub/datahub/db.py +++ b/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, diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index eb2aa74..35488d0 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/xiaobai-datahub/datahub/pipeline.py @@ -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 = { @@ -379,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 = {} - for dataset in ("daily", "valuation", "moneyflow", "auction"): - results[dataset] = self.run_dataset(dataset, trade_date) - return results + return self._run_eod_datasets(EOD_A_DATASETS, trade_date) 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 diff --git a/xiaobai-datahub/datahub/scheduler.py b/xiaobai-datahub/datahub/scheduler.py index de05360..c9cbee5 100644 --- a/xiaobai-datahub/datahub/scheduler.py +++ b/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,6 +48,7 @@ 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, @@ -42,6 +56,7 @@ class Scheduler: 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(): @@ -63,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)), @@ -85,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: @@ -103,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=?", @@ -126,6 +314,9 @@ 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) diff --git a/xiaobai-datahub/datahub/settings.py b/xiaobai-datahub/datahub/settings.py index 24be368..f5e0099 100644 --- a/xiaobai-datahub/datahub/settings.py +++ b/xiaobai-datahub/datahub/settings.py @@ -56,6 +56,18 @@ class Settings: 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, diff --git a/xiaobai-datahub/tests/test_eod_retry.py b/xiaobai-datahub/tests/test_eod_retry.py new file mode 100644 index 0000000..eefb523 --- /dev/null +++ b/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()