feat(HEL-457): 估值字段级质量门、股票主档每日发布和资金流历史回补

- 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 <github@multica.ai>
2026-09-04 21:36:20 +08:00
parent c9892050c3
commit bed6450992
14 changed files with 1058 additions and 37 deletions
@@ -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 备份"},
@@ -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

@@ -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"),
 }


@@ -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 = {
@@ -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)

@@ -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 = ?",
@@ -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")