From bed64509926045193feee5909332c29c35dddc99 Mon Sep 17 00:00:00 2001 From: multica-agent Date: Fri, 4 Sep 2026 21:36:20 +0800 Subject: [PATCH] =?UTF-8?q?feat(HEL-457):=20=E4=BC=B0=E5=80=BC=E5=AD=97?= =?UTF-8?q?=E6=AE=B5=E7=BA=A7=E8=B4=A8=E9=87=8F=E9=97=A8=E3=80=81=E8=82=A1?= =?UTF-8?q?=E7=A5=A8=E4=B8=BB=E6=A1=A3=E6=AF=8F=E6=97=A5=E5=8F=91=E5=B8=83?= =?UTF-8?q?=E5=92=8C=E8=B5=84=E9=87=91=E6=B5=81=E5=8E=86=E5=8F=B2=E5=9B=9E?= =?UTF-8?q?=E8=A1=A5?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit - field_gates 按数据集配置关键字段非空率下限/非有限比例/相对上一批次的塌陷保护， 字段大面积为空的批次拒发并保留上一正式批次，可读失败原因入 batches.error - 股票主档交易日 20:00/23:10 自动刷新并发布版本化快照（eod_stocks + publications）， 覆盖新上市/简称变化/N前缀摘除；/v1/stocks 携带 batch_id/published_at，无变化跳过 - moneyflow 历史回补（默认 60 交易日，跳过已发布日期）；未发布点查返回 available_from/available_to 与 history_not_backfilled 标记，缺失不再静默 - eod-refresh 新增 --force --dataset 安全重发（仍走全部质量门，上一批次可回滚） - 保持 HEL-435 盘后重试机制；新增 22 项测试覆盖字段拒发/正常通过/旧批保留/ 主档新增改名/资金流覆盖/重复执行幂等 Co-authored-by: multica-agent --- xiaobai-datahub/README.md | 32 ++ .../config/hub-quality.config.json | 28 +- xiaobai-datahub/datahub/admin_api.py | 2 + xiaobai-datahub/datahub/cli.py | 62 +++- xiaobai-datahub/datahub/db.py | 16 + xiaobai-datahub/datahub/pipeline.py | 298 ++++++++++++++++-- xiaobai-datahub/datahub/scheduler.py | 10 +- xiaobai-datahub/datahub/serving.py | 47 ++- xiaobai-datahub/datahub/settings.py | 11 + xiaobai-datahub/tests/fixtures.py | 2 +- xiaobai-datahub/tests/test_eod_retry.py | 38 ++- .../tests/test_moneyflow_backfill.py | 128 ++++++++ xiaobai-datahub/tests/test_quality_gates.py | 253 +++++++++++++++ xiaobai-datahub/tests/test_stocks_refresh.py | 168 ++++++++++ 14 files changed, 1058 insertions(+), 37 deletions(-) create mode 100644 xiaobai-datahub/tests/test_moneyflow_backfill.py create mode 100644 xiaobai-datahub/tests/test_quality_gates.py create mode 100644 xiaobai-datahub/tests/test_stocks_refresh.py diff --git a/xiaobai-datahub/README.md b/xiaobai-datahub/README.md index 6cea588..f440af9 100644 --- a/xiaobai-datahub/README.md +++ b/xiaobai-datahub/README.md @@ -79,6 +79,38 @@ python -m datahub history-backfill 区间接口在 `meta.coverage` / `meta.incomplete` 标明覆盖是否完整；网站只读接入把不完整区间视为不可用并回旧链路。个股日 K 的 90 天区间查询依赖已核实，本阶段不回补全市场历史。 +## 估值字段级质量门 + +`hub-quality.config.json` 的 `field_gates` 按数据集配置关键字段：非空率下限（支持按字段覆盖，如 `dv_ttm` 合法高空值）、非有限值比例上限、以及相对上一已发布批次的非空率塌陷保护。字段大面积为空的批次会被拒绝发布、保留上一份正常正式数据，失败原因逐字段写入 `batches.error` / `quality_json`。被拒后数据集仍视为缺失，盘后自动重试（HEL-435 机制）会继续尝试直到成功或截止。配置对任意数据集生效，不写死单日或单字段。 + +## 股票主档每日刷新与发布 + +交易日 20:00 与 23:10（`stocks_refresh_times` 可配）自动刷新股票主档并发布版本化快照（`eod_stocks` + `publications.dataset='stocks'`），覆盖当日新上市、证券简称变化和上市首日 N/C 前缀摘除；无变化则跳过，重复执行幂等。`/v1/stocks` 从最新已发布快照提供数据并带 `batch_id` / `published_at`；`/v1/datasets/status` 同步展示 stocks 状态。 + +```bash +cd xiaobai-datahub +python -m datahub stocks-refresh # 手动触发；--force 无变化也重发 +``` + +## 资金流历史回补 + +网站会沿真实调用链查最近若干交易日的 moneyflow（个股详情任意日期点查 + 智能选股最近 5 个交易日），默认回补最近 60 个交易日（`moneyflow_history_trading_days` 可配，已发布日期自动跳过）。点查未覆盖的历史日期返回 `DATASET_NOT_PUBLISHED` 并附 `available_from` / `available_to`（低于下界时 `reason=history_not_backfilled`），网站据此明确回退旧链路，不会静默拿到半截数据。 + +```bash +cd xiaobai-datahub +python -m datahub moneyflow-backfill # --trading-days 60 --end-date --force 可选 +``` + +## 盘后补跑与强制重发 + +```bash +cd xiaobai-datahub +python -m datahub eod-refresh --trade-date 20260904 # 只补缺失数据集 +python -m datahub eod-refresh --trade-date 20260904 --force --dataset valuation +# 强制重取重发：仍走全部质量门，生成新批次，上一批次保留可回滚 +``` + + ## 备份 每日 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 06892ee..d0fafea 100644 --- a/xiaobai-datahub/config/hub-quality.config.json +++ b/xiaobai-datahub/config/hub-quality.config.json @@ -16,5 +16,31 @@ "index_history_trading_days": 260, "eod_retry_start": "15:15", "eod_retry_interval_minutes": 30, - "eod_retry_cutoff": "23:30" + "eod_retry_cutoff": "23:30", + "moneyflow_history_trading_days": 60, + "stocks_refresh_times": [ + "20:00", + "23:10" + ], + "field_gates": { + "valuation": { + "fields": [ + "turnover_rate", + "volume_ratio", + "total_mv", + "circ_mv", + "pe_ttm", + "pb", + "ps_ttm", + "dv_ttm" + ], + "min_nonnull_rate": 0.9, + "min_nonnull_rate_by_field": { + "pe_ttm": 0.5, + "dv_ttm": 0.3 + }, + "max_nonnull_drop_vs_prev": 0.15, + "max_nonfinite_rate": 0.01 + } + } } diff --git a/xiaobai-datahub/datahub/admin_api.py b/xiaobai-datahub/datahub/admin_api.py index 82a8558..b9bf4e9 100644 --- a/xiaobai-datahub/datahub/admin_api.py +++ b/xiaobai-datahub/datahub/admin_api.py @@ -84,12 +84,14 @@ class AdminAPI: def jobs(self) -> dict[str, Any]: runs = self.db.fetchall("SELECT * FROM job_runs ORDER BY id DESC LIMIT 100") + stocks_times = "/".join(self.pipeline.settings.stocks_refresh_times) or "20:00" return { "jobs": [ {"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": "stocks_refresh", "at": stocks_times, "title": "股票主档刷新与正式发布（新上市/更名，无变化跳过）"}, {"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 de142e0..daea873 100644 --- a/xiaobai-datahub/datahub/cli.py +++ b/xiaobai-datahub/datahub/cli.py @@ -7,6 +7,7 @@ import json import sys from datahub.hub import build_hub +from datahub.pipeline import OFFICIAL_DATASETS from datahub.settings import load_settings from datahub.timeutil import yyyymmdd @@ -20,6 +21,23 @@ def main(argv: list[str] | None = None) -> int: history.add_argument("--force", action="store_true", help="覆盖已发布的指数日期") refresh = sub.add_parser("eod-refresh", help="对指定交易日补跑盘后正式数据（跳过已发布数据集，仍走质量门禁）") refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天") + refresh.add_argument( + "--force", action="store_true", + help="对 --dataset 指定的数据集强制重取重发（生成新批次，保留上一批次可回滚）", + ) + refresh.add_argument( + "--dataset", default=None, + help="配合 --force 使用：只强制重发该数据集（如 valuation）", + ) + stocks_refresh = sub.add_parser("stocks-refresh", help="刷新股票主档并发布正式快照（幂等：无变化则跳过）") + stocks_refresh.add_argument("--trade-date", default=None, help="交易日 YYYYMMDD，默认今天") + stocks_refresh.add_argument("--force", action="store_true", help="即使快照无变化也重新发布") + moneyflow_backfill = sub.add_parser( + "moneyflow-backfill", help="回补资金流历史（默认覆盖网站所需的最近 N 个交易日，跳过已发布日期）", + ) + moneyflow_backfill.add_argument("--end-date", default=None, help="截止交易日 YYYYMMDD，默认今天") + moneyflow_backfill.add_argument("--trading-days", type=int, default=None, help="回补交易日数量，默认配置 moneyflow_history_trading_days") + moneyflow_backfill.add_argument("--force", action="store_true", help="覆盖已发布的资金流日期") args = parser.parse_args(argv) settings = load_settings() @@ -35,16 +53,56 @@ def main(argv: list[str] | None = None) -> int: 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) + if args.force: + datasets = tuple(sorted({args.dataset} & OFFICIAL_DATASETS)) if args.dataset else () + if args.dataset and not datasets: + parser.error(f"unknown dataset: {args.dataset}") + if not datasets: + parser.error("--force requires --dataset (e.g. --dataset valuation)") + result = {} + for dataset in datasets: + result[dataset] = hub.pipeline.run_dataset(dataset, day) + else: + 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)}, + {"force": bool(args.force), "dataset": args.dataset, + **{name: item.get("state") for name, item in result.items() if isinstance(item, dict)}}, ensure_ascii=False, )) + if args.force: + payload = {"trade_date": day, "datasets": result} + json.dump(payload, sys.stdout, ensure_ascii=False, indent=2, default=str) + sys.stdout.write("\n") + return 0 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 + if args.command == "stocks-refresh": + day = yyyymmdd(args.trade_date) if args.trade_date else yyyymmdd() + result = hub.pipeline.refresh_stocks(day, force=args.force) + hub.pipeline.audit("cli", "stocks-refresh", f"stocks:{day}", json.dumps( + {"force": bool(args.force), "state": result.get("state"), "batch_id": result.get("batch_id")}, + ensure_ascii=False, + )) + json.dump(result, sys.stdout, ensure_ascii=False, indent=2, default=str) + sys.stdout.write("\n") + return 0 if result.get("state") != "failed" else 1 + if args.command == "moneyflow-backfill": + result = hub.pipeline.backfill_moneyflow_history( + end_date=args.end_date, + trading_days=args.trading_days, + force=args.force, + ) + hub.pipeline.audit("cli", "moneyflow-backfill", f"moneyflow:{result.get('end')}", json.dumps( + {"published": len(result.get("published") or []), "skipped": len(result.get("skipped") or []), + "failed": len(result.get("failed") or [])}, + ensure_ascii=False, + )) + 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 diff --git a/xiaobai-datahub/datahub/db.py b/xiaobai-datahub/datahub/db.py index 93f430b..bcea373 100644 --- a/xiaobai-datahub/datahub/db.py +++ b/xiaobai-datahub/datahub/db.py @@ -114,6 +114,21 @@ CREATE TABLE IF NOT EXISTS eod_index_bars ( PRIMARY KEY (ts_code, trade_date, batch_id) ) WITHOUT ROWID; +CREATE TABLE IF NOT EXISTS eod_stocks ( + ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, + symbol TEXT, name TEXT, area TEXT, industry TEXT, market TEXT, + list_status TEXT, list_date TEXT, + batch_id TEXT NOT NULL, + PRIMARY KEY (ts_code, trade_date, batch_id) +) WITHOUT ROWID; + +CREATE TABLE IF NOT EXISTS staging_stocks ( + ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, batch_id TEXT NOT NULL, + symbol TEXT, name TEXT, area TEXT, industry TEXT, market TEXT, + list_status TEXT, list_date TEXT, + PRIMARY KEY (batch_id, ts_code, trade_date) +); + CREATE TABLE IF NOT EXISTS staging_bars ( ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, batch_id TEXT NOT NULL, open REAL, high REAL, low REAL, close REAL, pct_chg REAL, @@ -273,6 +288,7 @@ DATASET_TABLES = { "moneyflow": ("eod_moneyflow", "staging_moneyflow"), "auction": ("eod_auction", "staging_auction"), "index_daily": ("eod_index_bars", "staging_index_bars"), + "stocks": ("eod_stocks", "staging_stocks"), } diff --git a/xiaobai-datahub/datahub/pipeline.py b/xiaobai-datahub/datahub/pipeline.py index 35488d0..89bdd0f 100644 --- a/xiaobai-datahub/datahub/pipeline.py +++ b/xiaobai-datahub/datahub/pipeline.py @@ -1,6 +1,7 @@ from __future__ import annotations import json +import math import time from collections.abc import Callable from datetime import timedelta @@ -22,11 +23,21 @@ LOGGER = get_logger() HARD_DATASETS = {"daily", "valuation", "index_daily"} SOFT_DATASETS = {"moneyflow", "auction"} OFFICIAL_DATASETS = HARD_DATASETS | SOFT_DATASETS +STOCKS_DATASET = "stocks" +STOCK_SNAPSHOT_FIELDS = ("ts_code", "symbol", "name", "area", "industry", "market", "list_status", "list_date") EOD_A_DATASETS = ("daily", "valuation", "moneyflow", "auction") EOD_B_DATASETS = ("index_daily",) EMPTY_BATCH_ERROR = "empty official batch: 0 valid rows" STAGING_INSERT = { + "stocks": ( + "INSERT INTO staging_stocks(ts_code,trade_date,batch_id,symbol,name,area,industry,market,list_status,list_date) " + "VALUES (?,?,?,?,?,?,?,?,?,?)", + lambda r, b: ( + r["ts_code"], r["trade_date"], b, r.get("symbol"), r.get("name"), r.get("area"), + r.get("industry"), r.get("market"), r.get("list_status"), r.get("list_date"), + ), + ), "daily": ( "INSERT INTO staging_bars(ts_code,trade_date,batch_id,open,high,low,close,pct_chg,volume,amount,adj_factor) " "VALUES (?,?,?,?,?,?,?,?,?,?,?)", @@ -72,6 +83,11 @@ STAGING_INSERT = { } EOD_COPY = { + "stocks": ( + "INSERT OR REPLACE INTO eod_stocks " + "SELECT ts_code,trade_date,symbol,name,area,industry,market,list_status,list_date,batch_id " + "FROM staging_stocks WHERE batch_id = ?" + ), "daily": ( "INSERT OR REPLACE INTO eod_bars " "SELECT ts_code,trade_date,open,high,low,close,pct_chg,volume,amount,adj_factor,batch_id " @@ -101,6 +117,13 @@ EOD_COPY = { } +def _finite(value: Any) -> bool: + try: + return math.isfinite(float(value)) + except (TypeError, ValueError): + return False + + def _staging_row_count(connection: Any, dataset: str, batch_id: str) -> int: table = DATASET_TABLES[dataset][1] row = connection.execute( @@ -176,23 +199,7 @@ class Pipeline: """, (row["exchange"], row["cal_date"], row["is_open"], row.get("pretrade_date"), fetched_at), ) - for row in stocks: - connection.execute( - """ - INSERT INTO stock_master(ts_code,symbol,name,area,industry,market,list_status,list_date,updated_at) - VALUES (?,?,?,?,?,?,?,?,?) - ON CONFLICT(ts_code) DO UPDATE SET - symbol=excluded.symbol, name=excluded.name, area=excluded.area, - industry=excluded.industry, market=excluded.market, - list_status=excluded.list_status, list_date=excluded.list_date, - updated_at=excluded.updated_at - """, - ( - row["ts_code"], row.get("symbol"), row.get("name"), row.get("area"), - row.get("industry"), row.get("market"), row.get("list_status"), - row.get("list_date"), fetched_at, - ), - ) + self._upsert_stock_master(connection, stocks, fetched_at) return { "calendar": len(calendar), "stocks": len(stocks), @@ -201,6 +208,154 @@ class Pipeline: "calendar_to": end, } + @staticmethod + def _upsert_stock_master(connection: Any, stocks: list[dict[str, Any]], fetched_at: str) -> None: + for row in stocks: + connection.execute( + """ + INSERT INTO stock_master(ts_code,symbol,name,area,industry,market,list_status,list_date,updated_at) + VALUES (?,?,?,?,?,?,?,?,?) + ON CONFLICT(ts_code) DO UPDATE SET + symbol=excluded.symbol, name=excluded.name, area=excluded.area, + industry=excluded.industry, market=excluded.market, + list_status=excluded.list_status, list_date=excluded.list_date, + updated_at=excluded.updated_at + """, + ( + row["ts_code"], row.get("symbol"), row.get("name"), row.get("area"), + row.get("industry"), row.get("market"), row.get("list_status"), + row.get("list_date"), fetched_at, + ), + ) + + def latest_stocks_publication(self, on_or_before: str | None = None) -> dict[str, Any] | None: + if on_or_before: + row = self.db.fetchone( + "SELECT * FROM publications WHERE dataset = ? AND trade_date <= ? ORDER BY trade_date DESC LIMIT 1", + (STOCKS_DATASET, yyyymmdd(on_or_before)), + ) + else: + row = self.db.fetchone( + "SELECT * FROM publications WHERE dataset = ? ORDER BY trade_date DESC LIMIT 1", + (STOCKS_DATASET,), + ) + return row + + def published_stock_snapshot(self, on_or_before: str | None = None) -> tuple[str | None, list[dict[str, Any]]]: + pub = self.latest_stocks_publication(on_or_before) + if not pub: + return None, [] + rows = self.db.fetchall( + f"SELECT {','.join(STOCK_SNAPSHOT_FIELDS)} FROM eod_stocks WHERE batch_id = ? ORDER BY ts_code", + (pub["active_batch"],), + ) + return str(pub["active_batch"]), rows + + def refresh_stocks(self, trade_date: str | None = None, force: bool = False) -> dict[str, Any]: + """Refresh stock master from upstream and publish a versioned snapshot. + + Runs on trading days (scheduler) and via CLI. Idempotent: when the + latest published snapshot already matches the upstream list, nothing + is published; ``force`` re-publishes unconditionally. New listings, + renames (incl. N/C prefix removal) and status changes all flow into + the snapshot, which carries batch_id/published_at metadata. + """ + day = yyyymmdd(trade_date or self.clock()) + rows = self._fetch_dataset(STOCKS_DATASET, day) + with self.db.write() as connection: + self._upsert_stock_master(connection, rows, isoformat(self.clock())) + if not force: + active, snapshot = self.published_stock_snapshot(day) + if active is not None: + upstream = sorted( + tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in rows + ) + published = sorted(tuple(str(row.get(field)) for field in STOCK_SNAPSHOT_FIELDS) for row in snapshot) + if upstream == published: + return { + "dataset": STOCKS_DATASET, + "trade_date": day, + "state": "skipped", + "reason": "unchanged", + "batch_id": active, + "rows": len(snapshot), + } + result = self.run_dataset(STOCKS_DATASET, day, prepared_rows=rows) + self.audit( + "pipeline", "stocks-refresh", f"{STOCKS_DATASET}:{day}", + json.dumps({"batch_id": result["batch_id"], "rows": result["rows"]}, ensure_ascii=False), + ) + return result + + def backfill_moneyflow_history( + self, + end_date: str | None = None, + trading_days: int | None = None, + force: bool = False, + ) -> dict[str, Any]: + """Incrementally publish official moneyflow history for the website window. + + The website queries moneyflow for any navigable trade date (stock + detail point queries) and for the screener's last-5-days window, so + the hub must cover a trailing window of trading days instead of only + days published since go-live. Already published dates are skipped + unless ``force``; per-date failures are recorded without aborting. + """ + end = yyyymmdd(end_date or self.clock()) + limit = int(trading_days or self.settings.moneyflow_history_trading_days) + open_dates = self.open_trade_dates(end, limit) + if not open_dates: + return { + "start": None, + "end": end, + "requested_days": 0, + "published": [], + "skipped": [], + "failed": [{"error": "calendar has no open dates on or before end"}], + "ok": False, + } + start = open_dates[0] + published_dates: set[str] = set() + if not force: + pubs = self.db.fetchall( + "SELECT trade_date FROM publications WHERE dataset = 'moneyflow' AND trade_date >= ? AND trade_date <= ?", + (start, end), + ) + published_dates = {str(row["trade_date"]) for row in pubs} + targets = [day for day in open_dates if day not in published_dates] + skipped = [day for day in open_dates if day in published_dates] + published: list[dict[str, Any]] = [] + failed: list[dict[str, Any]] = [] + for day in targets: + try: + raw = retry_call( + lambda day=day: self._guarded_fetch("moneyflow", {"trade_date": day}), + attempts=self.settings.max_publish_attempts, + base_delay=0.05, + sleeper=lambda _d: time.sleep(_d), + ) + rows = self.adapter.normalize("moneyflow", raw) + result = self.run_dataset("moneyflow", 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)}) + return { + "start": start, + "end": end, + "requested_days": len(open_dates), + "published": published, + "skipped": skipped, + "failed": failed, + "ok": not failed, + } + def open_trade_dates(self, end: str, limit: int) -> list[str]: end = yyyymmdd(end) rows = self.db.fetchall( @@ -464,13 +619,14 @@ class Pipeline: if null_rate >= float(quality.get("null_rate_max") or 0.01): errors.append(f"null rate {null_rate:.4f}") empty = row_n == 0 - if empty and dataset in OFFICIAL_DATASETS: + if empty and (dataset in OFFICIAL_DATASETS or dataset == STOCKS_DATASET): errors.append(EMPTY_BATCH_ERROR) + field_report = self._field_gate(dataset, trade_date, rows, errors) if dataset in SOFT_DATASETS: hard_fail = bool(dup or bad_date or empty) else: - hard_fail = bool(errors) and dataset in HARD_DATASETS - return { + hard_fail = bool(errors) and (dataset in HARD_DATASETS or dataset == STOCKS_DATASET) + report = { "rows": row_n, "listed": listed_n, "ratio": round(ratio, 4), @@ -480,6 +636,103 @@ class Pipeline: "soft_fail": bool(warnings) and not hard_fail, "batch_id": batch_id, } + if field_report is not None: + report["fields"] = field_report + return report + + def _field_gate( + self, + dataset: str, + trade_date: str, + rows: list[dict[str, Any]], + errors: list[str], + ) -> dict[str, Any] | None: + """Config-driven per-field completeness gate. + + Catches field-level half-products (rows complete, key columns empty) + that row-count gates miss: non-null rate floors per field, non-finite + share, and a collapse guard against the previous published batch so + legitimately sparse fields (e.g. dv_ttm) are not false-flagged. + Generic for any dataset configured under quality["field_gates"]. + """ + gate = dict((self.settings.quality.get("field_gates") or {}).get(dataset) or {}) + if not gate or not rows: + return None + fields = [str(item) for item in (gate.get("fields") or []) if str(item)] + if not fields: + return None + min_rate = float(gate.get("min_nonnull_rate") or 0.9) + by_field = {str(k): float(v) for k, v in dict(gate.get("min_nonnull_rate_by_field") or {}).items()} + max_drop = float(gate.get("max_nonnull_drop_vs_prev") or 0.15) + max_nonfinite = float(gate.get("max_nonfinite_rate") or 0.01) + row_n = len(rows) + prev_batch, prev_stats = self._prev_field_stats(dataset, trade_date, fields) + report: dict[str, Any] = {} + for field in fields: + values = [row.get(field) for row in rows] + nulls = sum(1 for value in values if value is None) + nonfinite = sum(1 for value in values if value is not None and not _finite(value)) + rate = (row_n - nulls) / row_n + stats = { + "nonnull": row_n - nulls, + "null": nulls, + "nonnull_rate": round(rate, 4), + "nonfinite": nonfinite, + } + floor = by_field.get(field, min_rate) + if rate < floor: + errors.append( + f"field gate: {dataset}.{field} non-null rate {rate:.4f} < {floor}" + ) + if nonfinite / row_n > max_nonfinite: + errors.append( + f"field gate: {dataset}.{field} non-finite rate {nonfinite / row_n:.4f} > {max_nonfinite}" + ) + prev_rate = prev_stats.get(field) if prev_stats else None + if prev_rate is not None: + stats["prev_nonnull_rate"] = round(prev_rate, 4) + if prev_rate - rate > max_drop: + errors.append( + f"field gate: {dataset}.{field} non-null rate {rate:.4f} dropped > {max_drop:.2f} " + f"vs prev batch {prev_batch} ({prev_rate:.4f})" + ) + report[field] = stats + return report + + def _prev_field_stats( + self, + dataset: str, + trade_date: str, + fields: list[str], + ) -> tuple[str | None, dict[str, float]]: + """Non-null rates per field from the latest earlier published batch.""" + if dataset not in DATASET_TABLES: + return None, {} + table = DATASET_TABLES[dataset][0] + columns = { + str(row["name"]) + for row in self.db.fetchall(f"PRAGMA table_info({table})") + } + usable = [field for field in fields if field in columns] + if not usable: + return None, {} + pub = self.db.fetchone( + """ + SELECT active_batch FROM publications + WHERE dataset = ? AND trade_date < ? ORDER BY trade_date DESC LIMIT 1 + """, + (dataset, trade_date), + ) + if not pub: + return None, {} + batch_id = str(pub["active_batch"]) + selects = ",".join(f"AVG({field} IS NOT NULL) AS {field}" for field in usable) + row = self.db.fetchone( + f"SELECT {selects} FROM {table} WHERE batch_id = ?", + (batch_id,), + ) + stats = {field: float(row[field]) for field in usable if row.get(field) is not None} + return batch_id, stats def publish(self, dataset: str, trade_date: str, batch_id: str, state: str = "published") -> None: copy_sql = EOD_COPY[dataset] @@ -636,6 +889,11 @@ class Pipeline: ) def _fetch_dataset(self, dataset: str, trade_date: str) -> list[dict[str, Any]]: + if dataset == STOCKS_DATASET: + rows = self.adapter.normalize("stocks", self._guarded_fetch("stocks", {"list_status": "L"})) + for row in rows: + row["trade_date"] = trade_date + return rows if dataset == "daily": raw = self._guarded_fetch("daily", {"trade_date": trade_date}) factors = { diff --git a/xiaobai-datahub/datahub/scheduler.py b/xiaobai-datahub/datahub/scheduler.py index c9cbee5..51e2cb3 100644 --- a/xiaobai-datahub/datahub/scheduler.py +++ b/xiaobai-datahub/datahub/scheduler.py @@ -49,6 +49,7 @@ class Scheduler: "eod_a": self._eod_a, "eod_b": self._eod_b, "eod_retry": self._eod_retry, + "stocks_refresh": self._stocks_refresh, "cleanup": self._cleanup, "backup": self._backup, "history_backfill": self._history_backfill, @@ -89,6 +90,8 @@ class Scheduler: ("cleanup", time(0, 30)), ("backup", time(0, 40)), ] + for refresh_at in self.pipeline.settings.stocks_refresh_times: + plan.append(("stocks_refresh", _hhmm(refresh_at))) open_day = is_open_day(self.db, day) for job_id, at in plan: if current < at: @@ -96,7 +99,7 @@ class Scheduler: key = (job_id, day, at.strftime("%H%M")) if key in self._fired: continue - if job_id in {"eod_a", "eod_b"} and not open_day: + if job_id in {"eod_a", "eod_b", "stocks_refresh"} and not open_day: self._fired.add(key) continue self._fired.add(key) @@ -107,7 +110,7 @@ class Scheduler: try: self.run_job(job_id, day) except Exception: - if job_id not in {"eod_a", "eod_b"}: + if job_id not in {"eod_a", "eod_b", "stocks_refresh"}: raise # Keep the tick alive; evening retries take over. LOGGER.exception("scheduled job %s failed for %s", job_id, day) @@ -317,6 +320,9 @@ class Scheduler: def _eod_retry(self, trade_date: str) -> dict[str, Any]: return self.pipeline.run_eod_missing(trade_date) + def _stocks_refresh(self, trade_date: str) -> dict[str, Any]: + return self.pipeline.refresh_stocks(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/serving.py b/xiaobai-datahub/datahub/serving.py index fd6071f..ff8eb71 100644 --- a/xiaobai-datahub/datahub/serving.py +++ b/xiaobai-datahub/datahub/serving.py @@ -135,6 +135,29 @@ class V1API: def stocks(self, updated_since: str, q: dict[str, str]) -> dict[str, Any]: limit, offset = self._page(q) + today = yyyymmdd(now_shanghai()) + batch_id, snapshot = self.pipeline.published_stock_snapshot(today) + if batch_id: + # Formal view: the latest published stock snapshot, with batch + # metadata. Filters are applied in-memory on the snapshot. + pub = self.pipeline.latest_stocks_publication(today) or {} + rows = snapshot + if updated_since: + rows = [] + rows = rows[offset: offset + limit] + return envelope( + rows, + { + "tier": "official", + "trade_date": pub.get("trade_date"), + "published_at": pub.get("published_at"), + "source": "tushare:stock_basic", + "batch_id": batch_id, + "stale": False, + "staleness_seconds": 0, + "state": pub.get("state"), + }, + ) if updated_since: rows = self.db.fetchall( "SELECT * FROM stock_master WHERE updated_at >= ? ORDER BY ts_code LIMIT ? OFFSET ?", @@ -176,7 +199,7 @@ class V1API: def dataset_status(self, date: str) -> dict[str, Any]: trade_date = yyyymmdd(date or now_shanghai()) - datasets = ("daily", "valuation", "moneyflow", "auction", "index_daily") + datasets = ("daily", "valuation", "moneyflow", "auction", "index_daily", "stocks") items = [] for dataset in datasets: pub = self.db.fetchone( @@ -251,7 +274,7 @@ class V1API: raise ApiError( "DATASET_NOT_PUBLISHED", f"{dataset} {start} 尚未发布", - extra={"expected_at": "15:05+08:00"}, + extra=self._unpublished_extra(dataset, start), ) limit, offset = self._page(q) sql = f"SELECT * FROM {table} WHERE trade_date = ? AND batch_id = ?" @@ -281,7 +304,11 @@ class V1API: (dataset, start, end), ) if not pubs: - raise ApiError("DATASET_NOT_PUBLISHED", f"{dataset} {start}-{end} 尚未发布") + raise ApiError( + "DATASET_NOT_PUBLISHED", + f"{dataset} {start}-{end} 尚未发布", + extra=self._unpublished_extra(dataset, end), + ) rows: list[dict[str, Any]] = [] limit, offset = self._page(q) for pub in pubs: @@ -350,6 +377,20 @@ class V1API: offset = max(0, offset) return limit, offset + def _unpublished_extra(self, dataset: str, trade_date: str) -> dict[str, Any]: + """Identifiable coverage info: is this a history gap or today-not-yet?""" + extra: dict[str, Any] = {"expected_at": "15:05+08:00"} + row = self.db.fetchone( + "SELECT MIN(trade_date) AS a, MAX(trade_date) AS b FROM publications WHERE dataset = ?", + (dataset,), + ) + if row and row.get("a"): + extra["available_from"] = row["a"] + extra["available_to"] = row["b"] + if str(trade_date) < str(row["a"]): + extra["reason"] = "history_not_backfilled" + return extra + def _official_meta(self, dataset: str, trade_date: str, source: str) -> dict[str, Any]: pub = self.db.fetchone( "SELECT * FROM publications WHERE dataset = ? AND trade_date = ?", diff --git a/xiaobai-datahub/datahub/settings.py b/xiaobai-datahub/datahub/settings.py index f5e0099..3277899 100644 --- a/xiaobai-datahub/datahub/settings.py +++ b/xiaobai-datahub/datahub/settings.py @@ -56,6 +56,17 @@ class Settings: def index_history_trading_days(self) -> int: return int(self.quality.get("index_history_trading_days") or 260) + @property + def moneyflow_history_trading_days(self) -> int: + return int(self.quality.get("moneyflow_history_trading_days") or 60) + + @property + def stocks_refresh_times(self) -> tuple[str, ...]: + raw = self.quality.get("stocks_refresh_times") or ["20:00", "23:10"] + if isinstance(raw, str): + raw = [raw] + return tuple(str(item) for item in raw) + @property def eod_retry_start(self) -> str: return str(self.quality.get("eod_retry_start") or "15:15") diff --git a/xiaobai-datahub/tests/fixtures.py b/xiaobai-datahub/tests/fixtures.py index 9780423..78db5b4 100644 --- a/xiaobai-datahub/tests/fixtures.py +++ b/xiaobai-datahub/tests/fixtures.py @@ -25,7 +25,7 @@ RAW = { ], "daily_basic": [ {"ts_code": "600000.SH", "trade_date": "20240902", "turnover_rate": 1.2, "volume_ratio": 0.8, "total_mv": 1000.0, "circ_mv": 800.0, "pe_ttm": 5.1, "pb": 0.6, "ps_ttm": 1.1, "dv_ttm": 4.0}, - {"ts_code": "000001.SZ", "trade_date": "20240902", "turnover_rate": 2.2, "volume_ratio": 1.1, "total_mv": 2000.0, "circ_mv": 1500.0, "pe_ttm": 6.2, "pb": 0.7, "ps_ttm": 1.2, "dv_ttm": 3.0}, + {"ts_code": "000001.SZ", "trade_date": "20240902", "turnover_rate": 2.2, "volume_ratio": 1.1, "total_mv": 2000.0, "circ_mv": 1500.0, "pe_ttm": 6.2, "pb": 0.7, "ps_ttm": 1.2, "dv_ttm": None}, ], "adj_factor": [ {"ts_code": "600000.SH", "trade_date": "20240902", "adj_factor": 1.1}, diff --git a/xiaobai-datahub/tests/test_eod_retry.py b/xiaobai-datahub/tests/test_eod_retry.py index eefb523..9b9f77f 100644 --- a/xiaobai-datahub/tests/test_eod_retry.py +++ b/xiaobai-datahub/tests/test_eod_retry.py @@ -60,7 +60,11 @@ class EodRetryTests(unittest.TestCase): 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,)) + placeholders = ",".join("?" for _ in OFFICIAL) + return db.fetchall( + f"SELECT * FROM batches WHERE trade_date = ? AND dataset IN ({placeholders})", + (day, *sorted(OFFICIAL)), + ) @staticmethod def _batch_ids(db: HubDB, day: str) -> set[str]: @@ -71,7 +75,11 @@ class EodRetryTests(unittest.TestCase): 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,)) + placeholders = ",".join("?" for _ in OFFICIAL) + rows = db.fetchall( + f"SELECT dataset FROM publications WHERE trade_date = ? AND dataset IN ({placeholders})", + (day, *sorted(OFFICIAL)), + ) return {str(row["dataset"]) for row in rows} def test_first_empty_then_retry_succeeds(self) -> None: @@ -105,12 +113,12 @@ class EodRetryTests(unittest.TestCase): # success stops all further same-day requests batches_before = len(self._batches(db, day)) - calls_before = len(transport.calls) + eod_calls_before = len(self._eod_calls(transport)) 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) + self.assertEqual(len(self._eod_calls(transport)), eod_calls_before) def test_never_ready_marks_cutoff_failed_and_stops(self) -> None: day = "20240902" @@ -144,9 +152,20 @@ class EodRetryTests(unittest.TestCase): 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) + + def official_batches() -> list[str]: + placeholders = ",".join("?" for _ in OFFICIAL) + return [ + str(row["batch_id"]) + for row in db.fetchall( + f"SELECT batch_id FROM batches WHERE trade_date = ? AND dataset IN ({placeholders})", + (day, *sorted(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) + active_map = {row["dataset"]: row["active_batch"] for row in active if row["dataset"] in OFFICIAL} + batches_before = set(official_batches()) calls_before = self._eod_calls(transport) # container restart: fresh scheduler, missed-time catch-up fires eod_a/eod_b @@ -157,8 +176,11 @@ class EodRetryTests(unittest.TestCase): 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( + {row["dataset"]: row["active_batch"] for row in after if row["dataset"] in OFFICIAL}, + active_map, + ) + self.assertEqual(set(official_batches()), 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") diff --git a/xiaobai-datahub/tests/test_moneyflow_backfill.py b/xiaobai-datahub/tests/test_moneyflow_backfill.py new file mode 100644 index 0000000..fc18f19 --- /dev/null +++ b/xiaobai-datahub/tests/test_moneyflow_backfill.py @@ -0,0 +1,128 @@ +from __future__ import annotations + +import unittest +from datetime import date, timedelta +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.serving import ApiError, V1API +from datahub.settings import Settings +from tests.fixtures import fake_transport + +OPEN_DATES = ["20240826", "20240827", "20240828", "20240829", "20240830", "20240902", "20240903"] +EMPTY_UPSTREAM = {"20240828"} # one date the upstream cannot serve + + +def build_calendar(open_dates: list[str], span_days: int = 16) -> list[dict]: + start = date(int(open_dates[0][:4]), int(open_dates[0][4:6]), int(open_dates[0][6:8])) + rows = [] + open_set = set(open_dates) + for offset in range(span_days): + cursor = start + timedelta(days=offset) + compact = cursor.strftime("%Y%m%d") + rows.append( + {"exchange": "SSE", "cal_date": compact, "is_open": 1 if compact in open_set else 0, "pretrade_date": compact} + ) + return rows + + +def moneyflow_rows(day: str) -> list[dict]: + return [ + { + "ts_code": "600000.SH", "trade_date": day, + "buy_sm_amount": 10 + int(day[-2:]), "sell_sm_amount": 8, "buy_md_amount": 20, "sell_md_amount": 15, + "buy_lg_amount": 30, "sell_lg_amount": 25, "buy_elg_amount": 40, "sell_elg_amount": 35, "net_mf_amount": 17, + }, + { + "ts_code": "000001.SZ", "trade_date": day, + "buy_sm_amount": 11, "sell_sm_amount": 9, "buy_md_amount": 21, "sell_md_amount": 16, + "buy_lg_amount": 31, "sell_lg_amount": 26, "buy_elg_amount": 41, "sell_elg_amount": 36, "net_mf_amount": 18, + }, + ] + + +class MoneyflowHistoryTransport: + def __init__(self) -> None: + self.calendar = build_calendar(OPEN_DATES) + self.moneyflow_fetches: list[str] = [] + + def __call__(self, api_name: str, params: dict, fields: str): + 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 self.calendar if start <= row["cal_date"] <= end] + if api_name == "moneyflow": + day = str(params.get("trade_date") or "") + self.moneyflow_fetches.append(day) + if day in EMPTY_UPSTREAM: + return [] + return moneyflow_rows(day) + return fake_transport(api_name, params, fields) + + +class MoneyflowBackfillTests(unittest.TestCase): + def setUp(self) -> None: + self.transport = MoneyflowHistoryTransport() + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + self.db = HubDB(Path(tmp.name) / "hub.db") + adapter = TushareAdapter("x", transport=self.transport) + settings = Settings( + encryption_key=SecretVault.generate_key(), + db_path=self.db.path, + backup_dir=Path(tmp.name) / "backups", + quality={"max_publish_attempts": 2, "publication_generations": 3}, + ) + self.pipe = Pipeline(self.db, adapter, settings) + self.pipe.ingest_reference("20240903") + self.api = V1API(self.db, self.pipe, settings) + + def test_backfill_publishes_window_and_reports_failures(self) -> None: + result = self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5) + published = [item["trade_date"] for item in result["published"]] + self.assertEqual(published, ["20240829", "20240830", "20240902", "20240903"]) + self.assertEqual(result["failed"][0]["trade_date"], "20240828") + self.assertFalse(result["ok"]) + rows = self.db.fetchall("SELECT * FROM eod_moneyflow WHERE trade_date='20240830'") + self.assertEqual(len(rows), 2) + + def test_backfill_is_idempotent(self) -> None: + self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5) + fetches_after_first = list(self.transport.moneyflow_fetches) + second = self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5) + # only the still-missing date is re-fetched; published dates are skipped + self.assertEqual(self.transport.moneyflow_fetches[len(fetches_after_first):], ["20240828"]) + self.assertEqual(len(second["skipped"]), 4) + + def test_point_query_on_backfilled_date_serves_data(self) -> None: + self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5) + payload = self.api.handle("/v1/moneyflow", {"date": ["20240830"]}) + self.assertEqual(len(payload["data"]), 2) + self.assertEqual(payload["data"][0]["net_mf_amount"], 180000.0) + + def test_unpublished_point_below_window_is_identifiable(self) -> None: + self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5) + with self.assertRaises(ApiError) as ctx: + self.api.handle("/v1/moneyflow", {"date": ["20240801"]}) + extra = ctx.exception.extra + self.assertEqual(extra["available_from"], "20240829") # window starts at the first published date + self.assertEqual(extra["available_to"], "20240903") + self.assertEqual(extra["reason"], "history_not_backfilled") + self.assertEqual(extra["expected_at"], "15:05+08:00") + + def test_range_query_flags_missing_dates(self) -> None: + self.pipe.backfill_moneyflow_history(end_date="20240903", trading_days=5) + payload = self.api.handle("/v1/moneyflow", {"from": ["20240828"], "to": ["20240903"]}) + coverage = payload["meta"]["coverage"] + self.assertFalse(coverage["complete"]) + self.assertEqual(coverage["missing_count"], 1) + self.assertEqual(coverage["missing_sample"], ["20240828"]) + self.assertTrue(payload["meta"]["incomplete"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/xiaobai-datahub/tests/test_quality_gates.py b/xiaobai-datahub/tests/test_quality_gates.py new file mode 100644 index 0000000..c772703 --- /dev/null +++ b/xiaobai-datahub/tests/test_quality_gates.py @@ -0,0 +1,253 @@ +from __future__ import annotations + +import unittest +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, QualityError +from datahub.scheduler import Scheduler +from datahub.settings import Settings +from tests.fixtures import TRADE_DATE, fake_transport +from tests.test_eod_retry import clock_at + +FIELD_GATES = { + "valuation": { + "fields": [ + "turnover_rate", "volume_ratio", "total_mv", "circ_mv", + "pe_ttm", "pb", "ps_ttm", "dv_ttm", + ], + "min_nonnull_rate": 0.9, + "min_nonnull_rate_by_field": {"pe_ttm": 0.5, "dv_ttm": 0.3}, + "max_nonnull_drop_vs_prev": 0.15, + "max_nonfinite_rate": 0.01, + }, +} + + +class ValuationTransport: + """fake_transport with switchable daily_basic degradation modes.""" + + def __init__(self) -> None: + self.mode = "ok" + + def __call__(self, api_name: str, params: dict, fields: str): + rows = fake_transport(api_name, params, fields) + if api_name != "daily_basic": + return rows + trade_date = str(params.get("trade_date") or "") + if trade_date: + rows = [{**row, "trade_date": trade_date} for row in rows] + if self.mode == "ok": + return rows + patched = [] + for row in rows: + item = dict(row) + if self.mode == "fields_all_null": + item["volume_ratio"] = None + item["dv_ttm"] = None + elif self.mode == "vr_all_null": + item["volume_ratio"] = None + elif self.mode == "dv_all_null": + item["dv_ttm"] = None + elif self.mode == "nonfinite": + item["volume_ratio"] = float("inf") + patched.append(item) + return patched + + +def make_pipe(transport, quality_extra=None, clock=None): + tmp = tempfile.TemporaryDirectory() + db = HubDB(Path(tmp.name) / "hub.db") + adapter = TushareAdapter("test-token", transport=transport) + quality = { + "daily_row_ratio": 0.98, + "null_rate_max": 0.01, + "max_publish_attempts": 3, + "publication_generations": 3, + "job_run_retain_days": 90, + "staging_retain_days": 14, + "field_gates": FIELD_GATES, + } + if quality_extra: + quality.update(quality_extra) + settings = Settings( + encryption_key=SecretVault.generate_key(), + api_token="t" * 32, + admin_password="admin-pass", + tushare_token="test-token", + db_path=db.path, + quality=quality, + scheduler_enabled=False, + ) + pipe = Pipeline(db, adapter, settings, clock=clock) + pipe._tmp = tmp + return pipe, db + + +class ValuationFieldGateTests(unittest.TestCase): + def setUp(self) -> None: + self.transport = ValuationTransport() + self.pipe, self.db = make_pipe(self.transport) + self.pipe.ingest_reference(TRADE_DATE) + + def _active(self) -> str | None: + row = self.db.fetchone( + "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?", + (TRADE_DATE,), + ) + return str(row["active_batch"]) if row else None + + def test_normal_batch_with_legit_dv_nulls_passes(self) -> None: + result = self.pipe.run_dataset("valuation", TRADE_DATE) + self.assertEqual(result["state"], "published") + fields = result["quality"]["fields"] + # fixture: 1 of 2 stocks has null dv_ttm → 0.5 non-null ≥ 0.3 floor + self.assertEqual(fields["dv_ttm"]["nonnull_rate"], 0.5) + self.assertEqual(fields["volume_ratio"]["nonnull_rate"], 1.0) + self.assertFalse(result["quality"]["errors"]) + + def test_all_null_fields_rejected_and_prev_batch_kept(self) -> None: + first = self.pipe.run_dataset("valuation", TRADE_DATE) + self.transport.mode = "fields_all_null" + with self.assertRaises(QualityError) as ctx: + self.pipe.run_dataset("valuation", TRADE_DATE) + errors = "; ".join(ctx.exception.report["errors"]) + self.assertIn("field gate: valuation.volume_ratio non-null rate 0.0000 < 0.9", errors) + self.assertIn("field gate: valuation.dv_ttm non-null rate 0.0000 < 0.3", errors) + # previous good publication stays active + self.assertEqual(self._active(), first["batch_id"]) + # rejected batch left staged with readable error + field stats + rejected = self.db.fetchone( + "SELECT * FROM batches WHERE state='staged' AND dataset='valuation' ORDER BY started_at DESC", + ) + self.assertIsNotNone(rejected) + self.assertIn("field gate: valuation.volume_ratio", str(rejected["error"])) + import json + + quality = json.loads(rejected["quality_json"]) + self.assertEqual(quality["fields"]["volume_ratio"]["nonnull"], 0) + self.assertEqual(quality["fields"]["dv_ttm"]["nonnull"], 0) + + def test_volume_ratio_all_null_alone_rejected(self) -> None: + self.pipe.run_dataset("valuation", TRADE_DATE) + self.transport.mode = "vr_all_null" + with self.assertRaises(QualityError): + self.pipe.run_dataset("valuation", TRADE_DATE) + self.assertEqual( + self.db.fetchone( + "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?", + (TRADE_DATE,), + )["active_batch"], + "20240902-valuation-001", + ) + + def test_dv_ttm_all_null_rejected_by_floor_and_collapse(self) -> None: + prev_day = "20240830" + prev = self.pipe.run_dataset("valuation", prev_day) # prev dv nonnull 0.5 + self.transport.mode = "dv_all_null" + with self.assertRaises(QualityError) as ctx: + self.pipe.run_dataset("valuation", TRADE_DATE) + errors = "; ".join(ctx.exception.report["errors"]) + self.assertIn("field gate: valuation.dv_ttm non-null rate 0.0000 < 0.3", errors) + self.assertIn(f"dropped > 0.15 vs prev batch {prev['batch_id']}", errors) + + def test_nonfinite_values_rejected(self) -> None: + self.pipe.run_dataset("valuation", TRADE_DATE) + rows = self.pipe.adapter.normalize( + "valuation", self.pipe._guarded_fetch("valuation", {"trade_date": TRADE_DATE}) + ) + for row in rows: + row["volume_ratio"] = float("inf") + with self.assertRaises(QualityError) as ctx: + self.pipe.run_dataset("valuation", TRADE_DATE, prepared_rows=rows) + errors = "; ".join(ctx.exception.report["errors"]) + self.assertIn("field gate: valuation.volume_ratio non-finite rate 1.0000 > 0.01", errors) + + def test_gate_off_when_not_configured(self) -> None: + pipe, _db = make_pipe(ValuationTransport(), quality_extra={"field_gates": {}}) + pipe.ingest_reference(TRADE_DATE) + pipe.adapter._transport.mode = "fields_all_null" + result = pipe.run_dataset("valuation", TRADE_DATE) + self.assertEqual(result["state"], "published") # legacy behavior when unconfigured + + def test_gate_applies_to_any_configured_dataset(self) -> None: + gates = {"daily": {"fields": ["volume"], "min_nonnull_rate": 0.9, "max_nonfinite_rate": 0.01}} + pipe, _db = make_pipe(ValuationTransport(), quality_extra={"field_gates": gates}) + pipe.ingest_reference(TRADE_DATE) + + def null_volume(api_name, params, fields): + if api_name != "daily": + return fake_transport(api_name, params, fields) + rows = fake_transport(api_name, params, fields) + for row in rows: + row["vol"] = None + return rows + + pipe.adapter._transport = null_volume + with self.assertRaises(QualityError) as ctx: + pipe.run_dataset("daily", TRADE_DATE) + errors = "; ".join(ctx.exception.report["errors"]) + self.assertIn("field gate: daily.volume non-null rate 0.0000 < 0.9", errors) + + +class GateRetryInterplayTests(unittest.TestCase): + def test_rejected_valuation_stays_missing_and_retry_publishes_later(self) -> None: + transport = ValuationTransport() + transport.mode = "fields_all_null" + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + db = HubDB(Path(tmp.name) / "hub.db") + adapter = TushareAdapter("x", transport=transport) + settings = Settings( + encryption_key=SecretVault.generate_key(), + db_path=db.path, + backup_dir=Path(tmp.name) / "backups", + quality={"field_gates": FIELD_GATES, "max_publish_attempts": 2}, + ) + pipe = Pipeline(db, adapter, settings) + pipe.ingest_reference("20240902") + sched = Scheduler(db, pipe) + + sched.tick(clock_at("20240902", 15, 5)) # valuation rejected by field gate + sched.tick(clock_at("20240902", 15, 10)) + self.assertIn("valuation", pipe.missing_official_datasets("20240902")) + self.assertEqual( + pipe.active_batch("valuation", "20240902"), + None, + ) + + transport.mode = "ok" + sched.tick(clock_at("20240902", 15, 45)) # retry passes the gate + self.assertNotIn("valuation", pipe.missing_official_datasets("20240902")) + rows = db.fetchall("SELECT * FROM eod_valuation WHERE trade_date='20240902'") + self.assertTrue(rows) + self.assertTrue(all(row["volume_ratio"] is not None for row in rows)) + + +class ForceRepublishTests(unittest.TestCase): + def test_run_dataset_over_published_keeps_prev_for_rollback(self) -> None: + transport = ValuationTransport() + pipe, db = make_pipe(transport) + pipe.ingest_reference(TRADE_DATE) + first = pipe.run_dataset("valuation", TRADE_DATE) + transport.mode = "vr_all_null" + with self.assertRaises(QualityError): + pipe.run_dataset("valuation", TRADE_DATE) # gate holds: bad re-publish refused + transport.mode = "ok" + second = pipe.run_dataset("valuation", TRADE_DATE) # CLI --force path + self.assertNotEqual(first["batch_id"], second["batch_id"]) + pub = db.fetchone( + "SELECT * FROM publications WHERE dataset='valuation' AND trade_date=?", + (TRADE_DATE,), + ) + self.assertEqual(pub["active_batch"], second["batch_id"]) + self.assertEqual(pub["prev_batch"], first["batch_id"]) + rolled = pipe.rollback("valuation", TRADE_DATE, actor="cli") + self.assertEqual(rolled["active_batch"], first["batch_id"]) + + +if __name__ == "__main__": + unittest.main() diff --git a/xiaobai-datahub/tests/test_stocks_refresh.py b/xiaobai-datahub/tests/test_stocks_refresh.py new file mode 100644 index 0000000..17e5333 --- /dev/null +++ b/xiaobai-datahub/tests/test_stocks_refresh.py @@ -0,0 +1,168 @@ +from __future__ import annotations + +import unittest +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.serving import V1API +from datahub.settings import Settings +from tests.fixtures import TRADE_DATE, fake_transport +from tests.test_eod_retry import clock_at + + +class StockMasterTransport: + """fake_transport with a mutable stock_basic list (new listings / renames).""" + + def __init__(self) -> None: + self.stocks = [ + {"ts_code": "600000.SH", "symbol": "600000", "name": "浦发银行", "area": "上海", "industry": "银行", "market": "主板", "list_status": "L", "list_date": "19991110"}, + {"ts_code": "920071.BJ", "symbol": "920071", "name": "N金钛", "area": "辽宁", "industry": "小金属", "market": "北交所", "list_status": "L", "list_date": "20240901"}, + ] + + def __call__(self, api_name: str, params: dict, fields: str): + if api_name == "stock_basic": + return [dict(row) for row in self.stocks] + return fake_transport(api_name, params, fields) + + def rename_and_add(self) -> None: + for row in self.stocks: + if row["ts_code"] == "920071.BJ": + row["name"] = "金钛股份" # N-prefix removed the day after listing + self.stocks.append( + {"ts_code": "920289.BJ", "symbol": "920289", "name": "N华汇", "area": "广东", "industry": "专用机械", "market": "北交所", "list_status": "L", "list_date": "20240902"} + ) + + +def make_pipe(transport): + tmp = tempfile.TemporaryDirectory() + db = HubDB(Path(tmp.name) / "hub.db") + adapter = TushareAdapter("test-token", transport=transport) + settings = Settings( + encryption_key=SecretVault.generate_key(), + api_token="t" * 32, + admin_password="admin-pass", + tushare_token="test-token", + db_path=db.path, + quality={"max_publish_attempts": 3, "publication_generations": 3}, + scheduler_enabled=False, + ) + pipe = Pipeline(db, adapter, settings) + pipe._tmp = tmp + return pipe, db + + +class StocksRefreshTests(unittest.TestCase): + def setUp(self) -> None: + self.transport = StockMasterTransport() + self.pipe, self.db = make_pipe(self.transport) + self.pipe.ingest_reference(TRADE_DATE) + + def _stocks_api(self) -> dict: + return V1API(self.db, self.pipe, self.pipe.settings).handle("/v1/stocks", {}) + + def test_first_refresh_publishes_snapshot_with_meta(self) -> None: + result = self.pipe.refresh_stocks(TRADE_DATE) + self.assertEqual(result["state"], "published") + self.assertEqual(result["rows"], 2) + self.assertTrue(result["batch_id"].startswith("20240902-stocks-")) + payload = self._stocks_api() + self.assertEqual(payload["meta"]["batch_id"], result["batch_id"]) + self.assertIsNotNone(payload["meta"]["published_at"]) + self.assertEqual(len(payload["data"]), 2) + names = {row["ts_code"]: row["name"] for row in payload["data"]} + self.assertEqual(names["920071.BJ"], "N金钛") + self.assertNotIn("batch_id", payload["data"][0]) + + def test_new_listing_and_rename_publish_new_batch(self) -> None: + first = self.pipe.refresh_stocks(TRADE_DATE) + self.transport.rename_and_add() + second = self.pipe.refresh_stocks(TRADE_DATE) + self.assertEqual(second["state"], "published") + self.assertNotEqual(second["batch_id"], first["batch_id"]) + payload = self._stocks_api() + names = {row["ts_code"]: row["name"] for row in payload["data"]} + self.assertEqual(names["920071.BJ"], "金钛股份") + self.assertIn("920289.BJ", names) + self.assertEqual(names["920289.BJ"], "N华汇") + # stock_master is refreshed too (code resolution stays current) + master = self.db.fetchone("SELECT name FROM stock_master WHERE ts_code='920289.BJ'") + self.assertEqual(master["name"], "N华汇") + + def test_unchanged_refresh_is_idempotent(self) -> None: + first = self.pipe.refresh_stocks(TRADE_DATE) + again = self.pipe.refresh_stocks(TRADE_DATE) + self.assertEqual(again["state"], "skipped") + self.assertEqual(again["reason"], "unchanged") + self.assertEqual(again["batch_id"], first["batch_id"]) + count = self.db.fetchone( + "SELECT COUNT(*) AS n FROM batches WHERE dataset='stocks' AND trade_date=?", + (TRADE_DATE,), + )["n"] + self.assertEqual(count, 1) + + def test_force_republishes_even_unchanged(self) -> None: + first = self.pipe.refresh_stocks(TRADE_DATE) + forced = self.pipe.refresh_stocks(TRADE_DATE, force=True) + self.assertEqual(forced["state"], "published") + self.assertNotEqual(forced["batch_id"], first["batch_id"]) + + def test_snapshot_pinned_until_next_publish(self) -> None: + first = self.pipe.refresh_stocks(TRADE_DATE) + self.transport.rename_and_add() + # upstream changed but no refresh ran: published snapshot is untouched + _, snapshot = self.pipe.published_stock_snapshot(TRADE_DATE) + names = {row["ts_code"]: row["name"] for row in snapshot} + self.assertEqual(names["920071.BJ"], "N金钛") + self.assertNotIn("920289.BJ", names) + self.assertEqual(len(snapshot), 2) + + def test_dataset_status_includes_stocks(self) -> None: + result = self.pipe.refresh_stocks(TRADE_DATE) + payload = V1API(self.db, self.pipe, self.pipe.settings).handle( + "/v1/datasets/status", {"date": [TRADE_DATE]} + ) + by_name = {item["dataset"]: item for item in payload["data"]} + self.assertIn("stocks", by_name) + self.assertEqual(by_name["stocks"]["batch_id"], result["batch_id"]) + self.assertIsNotNone(by_name["stocks"]["published_at"]) + + +class StocksRefreshSchedulingTests(unittest.TestCase): + def _make(self): + tmp = tempfile.TemporaryDirectory() + self.addCleanup(tmp.cleanup) + db = HubDB(Path(tmp.name) / "hub.db") + transport = StockMasterTransport() + adapter = TushareAdapter("x", transport=transport) + settings = Settings( + encryption_key=SecretVault.generate_key(), + db_path=db.path, + backup_dir=Path(tmp.name) / "backups", + quality={"stocks_refresh_times": ["20:00", "23:10"]}, + ) + pipe = Pipeline(db, adapter, settings) + pipe.ingest_reference(TRADE_DATE) + return db, pipe, Scheduler(db, pipe) + + def test_scheduled_refresh_runs_on_open_day(self) -> None: + db, pipe, sched = self._make() + ran = sched.tick(clock_at(TRADE_DATE, 20, 0)) + self.assertIn("stocks_refresh", ran) + ran = sched.tick(clock_at(TRADE_DATE, 23, 10)) + self.assertIn("stocks_refresh", ran) # second slot catches late renames + self.assertIsNotNone(pipe.active_batch("stocks", TRADE_DATE)) + + def test_no_refresh_on_closed_day(self) -> None: + db, _pipe, sched = self._make() + sched.tick(clock_at("20240907", 20, 30)) # fixture: Saturday closed + runs = db.fetchall("SELECT * FROM job_runs WHERE job_id='stocks_refresh'") + self.assertEqual(runs, []) + + +if __name__ == "__main__": + unittest.main()