diff --git a/backend/data/datahub/bridge.py b/backend/data/datahub/bridge.py index 73ac792..776fd41 100644 --- a/backend/data/datahub/bridge.py +++ b/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) diff --git a/tests/test_datahub_bridge.py b/tests/test_datahub_bridge.py index c76abf8..5200412 100644 --- a/tests/test_datahub_bridge.py +++ b/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"): diff --git a/xiaobai-datahub/README.md b/xiaobai-datahub/README.md index 207afe3..6cea588 100644 --- a/xiaobai-datahub/README.md +++ b/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 份）。也可手动： diff --git a/xiaobai-datahub/config/hub-quality.config.json b/xiaobai-datahub/config/hub-quality.config.json index 3ae6a17..cc9285b 100644 --- a/xiaobai-datahub/config/hub-quality.config.json +++ b/xiaobai-datahub/config/hub-quality.config.json @@ -11,5 +11,7 @@ "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 } diff --git a/xiaobai-datahub/datahub/__main__.py b/xiaobai-datahub/datahub/__main__.py new file mode 100644 index 0000000..0892cbe --- /dev/null +++ b/xiaobai-datahub/datahub/__main__.py @@ -0,0 +1,4 @@ +from datahub.cli import main + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/xiaobai-datahub/datahub/adapters/tushare.py b/xiaobai-datahub/datahub/adapters/tushare.py index 1b9b522..e0c1c11 100644 --- a/xiaobai-datahub/datahub/adapters/tushare.py +++ b/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): diff --git a/xiaobai-datahub/datahub/admin_api.py b/xiaobai-datahub/datahub/admin_api.py index 2c4899b..fc5824c 100644 --- a/xiaobai-datahub/datahub/admin_api.py +++ b/xiaobai-datahub/datahub/admin_api.py @@ -88,6 +88,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": "history_backfill", "at": "manual", "title": "回补历史日历与指数日 K"}, {"id": "cleanup", "at": "00:30", "title": "清理 staging / 日志"}, {"id": "backup", "at": "00:40", "title": "SQLite 备份"}, ], @@ -130,12 +131,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}") - if dataset == "reference": - result = self.pipeline.ingest_reference(trade_date) + day = yyyymmdd(trade_date or now_shanghai()) + if dataset == "history": + self._dangerous(password, confirm, "history:full") + result = self.pipeline.backfill_history(day) else: - result = self.pipeline.run_dataset(dataset, trade_date) - self.pipeline.audit(actor, "backfill", f"{dataset}:{trade_date}", json.dumps({"ok": True})) + self._dangerous(password, confirm, f"{dataset}:{day}") + 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: diff --git a/xiaobai-datahub/datahub/cli.py b/xiaobai-datahub/datahub/cli.py new file mode 100644 index 0000000..03bf328 --- /dev/null +++ b/xiaobai-datahub/datahub/cli.py @@ -0,0 +1,38 @@ +"""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 + + +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="覆盖已发布的指数日期") + 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 + parser.error(f"unknown command: {args.command}") + return 2 + + +if __name__ == "__main__": + raise SystemExit(main()) diff --git a/xiaobai-datahub/datahub/coverage.py b/xiaobai-datahub/datahub/coverage.py new file mode 100644 index 0000000..ef5fddb --- /dev/null +++ b/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 diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index b8f193b..eb2aa74 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/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 @@ -141,11 +141,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]: - """Refresh trade calendar (window) and stock master. Not versioned by batch.""" + def ingest_reference( + 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) - end = add_days(day, 30) + start = yyyymmdd(start or self.settings.calendar_start) + 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 +191,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 +342,15 @@ class Pipeline: rows: list[dict[str, Any]] = [] try: self._set_batch(batch_id, dataset, trade_date, "fetching", 1) - rows = retry_call( - lambda: self._fetch_dataset(dataset, trade_date), - attempts=max_attempts, - base_delay=0.05, - sleeper=lambda _d: None if attempts == 1 else time.sleep(_d), - ) + if prepared_rows is None: + rows = retry_call( + lambda: self._fetch_dataset(dataset, trade_date), + attempts=max_attempts, + 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) diff --git a/xiaobai-datahub/datahub/scheduler.py b/xiaobai-datahub/datahub/scheduler.py index dbefbb1..de05360 100644 --- a/xiaobai-datahub/datahub/scheduler.py +++ b/xiaobai-datahub/datahub/scheduler.py @@ -37,6 +37,7 @@ class Scheduler: "eod_b": self._eod_b, "cleanup": self._cleanup, "backup": self._backup, + "history_backfill": self._history_backfill, } self._stop = threading.Event() self._thread: threading.Thread | None = None @@ -125,6 +126,9 @@ class Scheduler: def _eod_b(self, trade_date: str) -> dict[str, Any]: return self.pipeline.run_eod_batch_b(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: diff --git a/xiaobai-datahub/datahub/serving.py b/xiaobai-datahub/datahub/serving.py index 4c72374..fd6071f 100644 --- a/xiaobai-datahub/datahub/serving.py +++ b/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, - { - "tier": "official", - "trade_date": last["trade_date"], - "published_at": last["published_at"], - "source": source, - "batch_id": last["active_batch"], - "stale": False, - "staleness_seconds": 0, - }, + attach_coverage( + { + "tier": "official", + "trade_date": last["trade_date"], + "published_at": last["published_at"], + "source": source, + "batch_id": last["active_batch"], + "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) diff --git a/xiaobai-datahub/datahub/settings.py b/xiaobai-datahub/datahub/settings.py index f1c65ef..24be368 100644 --- a/xiaobai-datahub/datahub/settings.py +++ b/xiaobai-datahub/datahub/settings.py @@ -48,6 +48,14 @@ 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) + def load_settings( env: dict[str, str] | None = None, diff --git a/xiaobai-datahub/datahub/timeutil.py b/xiaobai-datahub/datahub/timeutil.py index 58faebb..39de2bf 100644 --- a/xiaobai-datahub/datahub/timeutil.py +++ b/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) diff --git a/xiaobai-datahub/tests/fixtures.py b/xiaobai-datahub/tests/fixtures.py index ad2bacc..9780423 100644 --- a/xiaobai-datahub/tests/fixtures.py +++ b/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") diff --git a/xiaobai-datahub/tests/test_api.py b/xiaobai-datahub/tests/test_api.py index 8ec4f19..6e33247 100644 --- a/xiaobai-datahub/tests/test_api.py +++ b/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) diff --git a/xiaobai-datahub/tests/test_history_backfill.py b/xiaobai-datahub/tests/test_history_backfill.py new file mode 100644 index 0000000..d490d71 --- /dev/null +++ b/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() diff --git a/xiaobai-datahub/tests/test_pipeline.py b/xiaobai-datahub/tests/test_pipeline.py index 972af7c..923f972 100644 --- a/xiaobai-datahub/tests/test_pipeline.py +++ b/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)