feat(HEL-478): 估值发布后晚间复核并原子追补上游修订

盘后成功发布后继续轻量比对 daily_basic 网站字段，发现修订才走质量门与整组原子切换，避免 17:10 快照落后于晚间上游改写。

Co-authored-by: Cursor <cursoragent@cursor.com>
Co-authored-by: multica-agent <github@multica.ai>

xiaobai-datahub/README.md

@@ -123,6 +123,18 @@ python -m datahub eod-refresh --trade-date 20260904 --force --dataset valuation
 
 管理后台「补数」对盘后正式数据集同样走 `force_republish_boundary`，不会绕过 A/B 整批边界。
 
+## 估值发布后复核与自动追补
+
+Tushare `daily_basic` 会在盘后继续改当日字段。HEL-423 在 2026-09-07 观察到：中枢 17:10 发布 `003021.SZ turnover_rate=1.3565`，21:05 上游/旧链路已是 `1.3572`；其余 7 类观察对象当日一致。日 K、资金流、竞价、指数没有同类晚间修订证据，股票主档已有 20:00/23:10 刷新，因此默认只复核估值，不盲目全量重拉。
+
+窗口（可配）：交易日 **20:00–23:20**，每 30 分钟一次轻量比对（对齐网站 21:00 / 23:30 观察）。只拉取 `daily_basic`，按网站真实请求字段精确比较，无误差豁免。
+
+- 无变化：不产生新批次，状态「已追平」。
+- 发现修订：重新走字段质量门、覆盖检查和 A 组整批原子发布；读者全程只能看到上一完整版本或新完整版本。
+- 上游空 / 接口失败 / 不完整 / 质量门拒绝：保留上一完整版本，状态「复核失败」。
+- 23:20 截止后停止当晚复核；下一自然日盘前对上一交易日再做一次安全追赶。
+- 与 `eod_a` / `eod_retry` 共用互斥锁；容器重启会在窗口内立即补一次。
+
 ## 备份
 
 每日 00:40 任务把 `datahub.db` 备份到 `data/backups/`（保留 14 份）。也可手动：

xiaobai-datahub/admin/app.js

@@ -105,6 +105,7 @@ async function render() {
     const data = await api("/admin/api/overview");
     $("phase").textContent = data.session_phase;
     const eod = data.eod_status || {};
+    const rev = data.revision_status || {};
     const eodLabels = {
       pending_first_attempt: "等待首次尝试",
       waiting_upstream: "等待上游",
@@ -112,6 +113,14 @@ async function render() {
       cutoff_failed: "已截止失败",
       closed_day: "休市",
     };
+    const revLabels = {
+      waiting_review: "等待复核",
+      review_failed: "复核失败",
+      aligned: "已追平",
+      cutoff: "已截止",
+      pending_publish: "待发布",
+      closed_day: "休市",
+    };
     const eodExtra = [];
     if (eod.state === "waiting_upstream") {
       eodExtra.push(`已试 ${eod.attempts} 次`);
@@ -121,12 +130,16 @@ async function render() {
     if (eod.state === "cutoff_failed" && eod.missing_datasets) {
       eodExtra.push(`缺 ${esc(eod.missing_datasets.join(","))}`);
     }
+    const revExtra = [];
+    if (rev.detail) revExtra.push(esc(String(rev.detail)));
+    if (rev.window) revExtra.push(esc(String(rev.window)));
     page.innerHTML = `
       <div class="cards">
         <div class="card"><div class="muted">交易日</div><strong>${esc(data.trade_date)}</strong></div>
         <div class="card"><div class="muted">阶段</div><strong>${esc(data.session_phase)}</strong></div>
         <div class="card"><div class="muted">今日发布</div><strong>${data.publications.length}</strong></div>
         <div class="card"><div class="muted">盘后补跑</div><strong>${esc(eodLabels[eod.state] || eod.state || "-")}</strong><div class="muted">${eodExtra.join(" · ")}</div></div>
+        <div class="card"><div class="muted">估值复核</div><strong>${esc(revLabels[rev.state] || rev.state || "-")}</strong><div class="muted">${revExtra.join(" · ")}</div></div>
         <div class="card"><div class="muted">异常批次</div><strong class="${data.anomalies.length ? "fail" : "ok"}">${data.anomalies.length}</strong></div>
       </div>
       <h2>最近调用</h2>

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

@@ -17,6 +17,10 @@
   "eod_retry_start": "15:15",
   "eod_retry_interval_minutes": 30,
   "eod_retry_cutoff": "23:30",
+  "revision_review_datasets": ["valuation"],
+  "revision_review_start": "20:00",
+  "revision_review_interval_minutes": 30,
+  "revision_review_cutoff": "23:20",
   "moneyflow_history_trading_days": 60,
   "stocks_refresh_times": [
     "20:00",

xiaobai-datahub/datahub/admin_api.py

@@ -39,6 +39,7 @@ class AdminAPI:
             "session_phase": session_phase(now_shanghai(), is_open),
             "is_open_day": is_open,
             "eod_status": self.scheduler.eod_status(today),
+            "revision_status": self.scheduler.revision_status(today),
             "publications": pubs,
             "anomalies": failed,
             "recent_calls": _public_calls(calls),
@@ -91,6 +92,7 @@ class AdminAPI:
                 {"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": "eod_revise", "at": "20:00-23:20", "title": "估值发布后复核（轻量比对，有修订才整组原子追补）"},
                 {"id": "stocks_refresh", "at": stocks_times, "title": "股票主档刷新与正式发布（新上市/更名，无变化跳过）"},
                 {"id": "history_backfill", "at": "manual", "title": "回补历史日历与指数日 K"},
                 {"id": "cleanup", "at": "00:30", "title": "清理 staging / 日志"},

xiaobai-datahub/datahub/db.py

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

xiaobai-datahub/datahub/pipeline.py

@@ -15,6 +15,12 @@ from datahub.governance.ratelimit import TokenBucket
 from datahub.governance.retry import RetryError, retry_call
 from datahub.logutil import get_logger
 from datahub.normalize import finite_number, normalize_daily
+from datahub.revision import (
+    compare_fields,
+    diff_published_vs_upstream,
+    official_table,
+    revision_datasets,
+)
 from datahub.settings import Settings
 from datahub.timeutil import add_days, isoformat, now_shanghai, yyyymmdd
 
@@ -687,6 +693,240 @@ class Pipeline:
             return self.run_eod_batch_b(trade_date, force=True)
         raise ValueError(f"dataset is not part of an EOD release boundary: {dataset}")
 
+    def published_official_rows(self, dataset: str, trade_date: str) -> list[dict[str, Any]]:
+        day = yyyymmdd(trade_date)
+        batch_id = self.active_batch(dataset, day)
+        if not batch_id:
+            return []
+        fields = compare_fields(dataset)
+        table = official_table(dataset)
+        if fields:
+            columns = ",".join(fields)
+            return self.db.fetchall(
+                f"SELECT {columns} FROM {table} WHERE batch_id = ?",
+                (batch_id,),
+            )
+        return self.db.fetchall(f"SELECT * FROM {table} WHERE batch_id = ?", (batch_id,))
+
+    def compare_revision(self, dataset: str, trade_date: str) -> dict[str, Any]:
+        """Light fetch of one revision-risk dataset vs the published official rows."""
+        day = yyyymmdd(trade_date)
+        published = self.published_official_rows(dataset, day)
+        if not published:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "skipped",
+                "reason": "not_published",
+            }
+        try:
+            upstream = self._fetch_dataset(dataset, day)
+        except Exception as exc:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "upstream_error",
+                "error": str(exc),
+            }
+        if not upstream:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "upstream_empty",
+                "error": "revision review upstream empty",
+                "published_rows": len(published),
+                "upstream_rows": 0,
+            }
+        listed = self.db.fetchone(
+            "SELECT COUNT(*) AS n FROM stock_master WHERE list_status = 'L'",
+        )
+        listed_n = int((listed or {}).get("n") or 0)
+        floor = float(self.settings.quality.get("daily_row_ratio") or 0.98)
+        if listed_n and len(upstream) / listed_n < floor:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "incomplete",
+                "error": (
+                    f"revision review incomplete: upstream {len(upstream)} "
+                    f"/ listed {listed_n} < {floor}"
+                ),
+                "published_rows": len(published),
+                "upstream_rows": len(upstream),
+            }
+        if len(upstream) < len(published) * floor:
+            return {
+                "dataset": dataset,
+                "trade_date": day,
+                "changed": False,
+                "state": "failed",
+                "reason": "incomplete",
+                "error": (
+                    f"revision review incomplete: upstream {len(upstream)} "
+                    f"< published {len(published)} * {floor}"
+                ),
+                "published_rows": len(published),
+                "upstream_rows": len(upstream),
+            }
+        compared = diff_published_vs_upstream(dataset, published, upstream)
+        compared["trade_date"] = day
+        compared["state"] = "changed" if compared["changed"] else "unchanged"
+        compared["reason"] = "revised" if compared["changed"] else "unchanged"
+        return compared
+
+    def review_published_revisions(self, trade_date: str) -> dict[str, Any]:
+        """Evening/morning catch-up: compare website fields, republish only on change.
+
+        Unchanged → no new batch. Changed → full A/B boundary quality gate +
+        atomic switch (HEL-459/460/461). Empty/failed/incomplete upstream keeps
+        the previous complete official version.
+        """
+        day = yyyymmdd(trade_date)
+        results: dict[str, Any] = {}
+        for dataset in revision_datasets(self.settings.quality):
+            compared = self.compare_revision(dataset, day)
+            if compared.get("state") == "skipped":
+                results[dataset] = compared
+                continue
+            if compared.get("state") == "failed":
+                results[dataset] = compared
+                LOGGER.warning(
+                    "revision review kept previous official version",
+                    extra={
+                        "hub": {
+                            "dataset": dataset,
+                            "trade_date": day,
+                            "reason": compared.get("reason"),
+                            "event": "revision_review_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": compared.get("reason"),
+                            "error": compared.get("error"),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            if not compared.get("changed"):
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "aligned",
+                    "reason": "unchanged",
+                    "published_rows": compared.get("published_rows"),
+                    "upstream_rows": compared.get("upstream_rows"),
+                }
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps({"state": "aligned", "reason": "unchanged"}, ensure_ascii=False),
+                )
+                continue
+            LOGGER.info(
+                "revision review detected upstream rewrite, republishing boundary",
+                extra={
+                    "hub": {
+                        "dataset": dataset,
+                        "trade_date": day,
+                        "diffs": compared.get("diffs"),
+                        "event": "revision_review_changed",
+                    }
+                },
+            )
+            try:
+                published = self.force_republish_boundary(dataset, day)
+            except Exception as exc:
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "reason": "republish_error",
+                    "error": str(exc),
+                    "diffs": compared.get("diffs"),
+                }
+                LOGGER.warning(
+                    "revision republish failed, previous official version keeps serving",
+                    extra={
+                        "hub": {
+                            "dataset": dataset,
+                            "trade_date": day,
+                            "reason": str(exc),
+                            "event": "revision_review_failed",
+                        }
+                    },
+                )
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {"state": "failed", "reason": "republish_error", "error": str(exc)},
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            failures = self.eod_failures(published)
+            if failures:
+                results.update(published)
+                results[dataset] = {
+                    **(published.get(dataset) or {}),
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "reason": "quality_gate",
+                    "error": "; ".join(failures),
+                    "diffs": compared.get("diffs"),
+                }
+                self.audit(
+                    "pipeline", "revision-review", f"{dataset}:{day}",
+                    json.dumps(
+                        {
+                            "state": "failed",
+                            "reason": "quality_gate",
+                            "error": "; ".join(failures),
+                            "diffs": compared.get("diffs"),
+                        },
+                        ensure_ascii=False,
+                    ),
+                )
+                continue
+            results.update(published)
+            results["review"] = {
+                "dataset": dataset,
+                "trade_date": day,
+                "state": "aligned",
+                "reason": "revised",
+                "diffs": compared.get("diffs"),
+                "missing_codes": compared.get("missing_codes"),
+                "extra_codes": compared.get("extra_codes"),
+            }
+            self.audit(
+                "pipeline", "revision-review", f"{dataset}:{day}",
+                json.dumps(
+                    {
+                        "state": "aligned",
+                        "reason": "revised",
+                        "diffs": compared.get("diffs"),
+                        "switched": sorted(
+                            name for name, item in published.items()
+                            if isinstance(item, dict) and item.get("state") == "published"
+                        ),
+                    },
+                    ensure_ascii=False,
+                ),
+            )
+        return results
+
     def run_release_group(
         self,
         datasets: tuple[str, ...],

xiaobai-datahub/datahub/revision.py

@@ -0,0 +1,111 @@
+"""Post-publish revision review for datasets whose upstream may rewrite T-day fields.
+
+HEL-423 field evidence, not a whitelist of tolerated diffs:
+
+- 2026-09-07 valuation/daily_basic: hub published 003021.SZ turnover_rate=1.3565
+  at 17:10; website legacy and a direct Tushare read at 21:05 both showed 1.3572.
+  The other seven observed objects (daily, moneyflow, auction, stocks, status,
+  index_daily, calendar) matched. Hub had already stopped the day after the
+  first successful publish, so the revision never self-healed.
+- 2026-09-02: same dataset, opposite direction (hub already held the later
+  value). Confirms daily_basic is rewritten after the first complete dump.
+
+Daily bars, moneyflow, auction and index_daily have no same-evening field
+revision evidence. Stocks already refreshes at 20:00/23:10. Review therefore
+fetches only configured revision-risk datasets (default: valuation) and
+compares the website-requested field set. No numeric tolerance.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from datahub.db import DATASET_TABLES
+from datahub.normalize import VALUATION_FIELDS
+from datahub.numbers import finite_number, round4
+
+# Datasets with proven same-evening upstream rewrites. Config may replace this
+# list; it must not silently expand to a full EOD re-pull.
+DEFAULT_REVISION_DATASETS = ("valuation",)
+
+# Website daily_basic request (HEL-423): ts_code/trade_date plus the eight
+# value fields used by the old link and field_gates.
+WEBSITE_COMPARE_FIELDS: dict[str, tuple[str, ...]] = {
+    "valuation": VALUATION_FIELDS,
+}
+
+REVISION_STATES = ("waiting_review", "review_failed", "aligned", "cutoff")
+
+
+def revision_datasets(quality: dict[str, Any] | None) -> tuple[str, ...]:
+    raw = (quality or {}).get("revision_review_datasets")
+    if isinstance(raw, (list, tuple)) and raw:
+        names = tuple(str(item) for item in raw if str(item))
+        if names:
+            return names
+    return DEFAULT_REVISION_DATASETS
+
+
+def compare_fields(dataset: str) -> tuple[str, ...]:
+    fields = WEBSITE_COMPARE_FIELDS.get(dataset)
+    if fields:
+        return fields
+    gate = {}
+    return tuple(str(item) for item in (gate.get("fields") or []) if str(item))
+
+
+def _norm_value(field: str, value: Any) -> Any:
+    if field in {"ts_code", "trade_date"}:
+        return str(value or "")
+    number = round4(finite_number(value))
+    return number
+
+
+def row_signature(row: dict[str, Any], fields: tuple[str, ...]) -> tuple[Any, ...]:
+    return tuple(_norm_value(field, row.get(field)) for field in fields)
+
+
+def diff_published_vs_upstream(
+    dataset: str,
+    published: list[dict[str, Any]],
+    upstream: list[dict[str, Any]],
+    *,
+    max_diffs: int = 20,
+) -> dict[str, Any]:
+    """Exact compare on website-requested fields. No tolerance / exemption."""
+    fields = compare_fields(dataset)
+    if not fields:
+        fields = tuple(sorted({key for row in published + upstream for key in row if key != "batch_id"}))
+    pub_map = {str(row.get("ts_code") or "").upper(): row for row in published}
+    up_map = {str(row.get("ts_code") or "").upper(): row for row in upstream}
+    missing = sorted(code for code in pub_map if code not in up_map)
+    extra = sorted(code for code in up_map if code not in pub_map)
+    diffs: list[dict[str, Any]] = []
+    for code in sorted(set(pub_map) & set(up_map)):
+        left = row_signature(pub_map[code], fields)
+        right = row_signature(up_map[code], fields)
+        if left == right:
+            continue
+        for field, old, new in zip(fields, left, right):
+            if old == new:
+                continue
+            diffs.append({"ts_code": code, "field": field, "published": old, "upstream": new})
+            if len(diffs) >= max_diffs:
+                break
+        if len(diffs) >= max_diffs:
+            break
+    changed = bool(diffs or missing or extra)
+    return {
+        "changed": changed,
+        "dataset": dataset,
+        "fields": list(fields),
+        "published_rows": len(published),
+        "upstream_rows": len(upstream),
+        "missing_codes": missing[:max_diffs],
+        "extra_codes": extra[:max_diffs],
+        "diffs": diffs,
+    }
+
+
+def official_table(dataset: str) -> str:
+    return DATASET_TABLES[dataset][0]

xiaobai-datahub/datahub/scheduler.py

@@ -1,5 +1,6 @@
 from __future__ import annotations
 
+import json
 import threading
 from collections.abc import Callable
 from datetime import datetime, time, timedelta
@@ -8,13 +9,14 @@ from typing import Any
 from datahub.db import HubDB
 from datahub.logutil import get_logger
 from datahub.pipeline import Pipeline
+from datahub.revision import revision_datasets
 from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
 
 LOGGER = get_logger()
 
 JobFn = Callable[[str], Any]
 
-EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry"}
+EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry", "eod_revise"}
 
 
 def is_open_day(db: HubDB, day: str) -> bool:
@@ -27,6 +29,20 @@ def is_open_day(db: HubDB, day: str) -> bool:
     return int(row["is_open"]) == 1
 
 
+def previous_open_day(db: HubDB, day: str) -> str | None:
+    row = db.fetchone(
+        """
+        SELECT cal_date FROM trade_calendar
+        WHERE exchange = 'SSE' AND is_open = 1 AND cal_date < ?
+        ORDER BY cal_date DESC LIMIT 1
+        """,
+        (day,),
+    )
+    if row is None:
+        return None
+    return str(row["cal_date"])
+
+
 def _hhmm(value: str) -> time:
     return datetime.strptime(value, "%H:%M").time()
 
@@ -49,6 +65,7 @@ class Scheduler:
             "eod_a": self._eod_a,
             "eod_b": self._eod_b,
             "eod_retry": self._eod_retry,
+            "eod_revise": self._eod_revise,
             "stocks_refresh": self._stocks_refresh,
             "cleanup": self._cleanup,
             "backup": self._backup,
@@ -118,6 +135,8 @@ class Scheduler:
             if job_id in {"eod_a", "eod_b"}:
                 self._settle_eod(day)
         ran.extend(self._eod_retry_tick(now, day, open_day))
+        ran.extend(self._revision_review_tick(now, day, open_day))
+        ran.extend(self._revision_catchup_tick(now, day))
         return ran
 
     # ------------------------------------------------------------------
@@ -219,6 +238,201 @@ class Scheduler:
             "detail": (row or {}).get("detail"),
         }
 
+    def revision_progress(self, day: str) -> dict[str, Any] | None:
+        return self.db.fetchone("SELECT * FROM revision_progress WHERE trade_date = ?", (day,))
+
+    def revision_status(self, trade_date: str | None = None, clock: datetime | None = None) -> dict[str, Any]:
+        """等待复核 / 复核失败 / 已追平 / 已截止."""
+        day = yyyymmdd(trade_date or now_shanghai(clock))
+        row = self.revision_progress(day)
+        open_day = is_open_day(self.db, day)
+        published = self._revision_ready(day)
+        if row and row["state"] in {"aligned", "review_failed", "cutoff", "waiting_review"}:
+            state = str(row["state"])
+        elif not open_day:
+            state = "closed_day"
+        elif not published:
+            state = "pending_publish"
+        else:
+            state = "waiting_review"
+        return {
+            "trade_date": day,
+            "is_open_day": open_day,
+            "state": state,
+            "datasets": list(revision_datasets(self.pipeline.settings.quality)),
+            "attempts": int((row or {}).get("attempts") or 0),
+            "last_attempt_at": (row or {}).get("last_attempt_at"),
+            "next_retry_at": (row or {}).get("next_retry_at") if state in {"waiting_review", "review_failed"} else None,
+            "finished_at": (row or {}).get("finished_at"),
+            "catchup_done": bool(int((row or {}).get("catchup_done") or 0)),
+            "detail": (row or {}).get("detail"),
+            "window": f"{self.pipeline.settings.revision_review_start}-{self.pipeline.settings.revision_review_cutoff}",
+        }
+
+    def _revision_ready(self, day: str) -> bool:
+        return all(
+            self.pipeline.active_batch(dataset, day)
+            for dataset in revision_datasets(self.pipeline.settings.quality)
+        )
+
+    def _revision_due(self, now: datetime, row: dict[str, Any] | None) -> bool:
+        if row is None or not row.get("last_attempt_at"):
+            return True
+        try:
+            last = datetime.fromisoformat(str(row["last_attempt_at"]))
+        except ValueError:
+            return True
+        interval = timedelta(minutes=self.pipeline.settings.revision_review_interval_minutes)
+        return now_shanghai(last).replace(tzinfo=None) + interval <= now.replace(tzinfo=None)
+
+    def _revision_review_tick(self, now: datetime, day: str, open_day: bool) -> list[str]:
+        if not open_day or not self._revision_ready(day):
+            return []
+        settings = self.pipeline.settings
+        current = now.time()
+        start = _hhmm(settings.revision_review_start)
+        cutoff = _hhmm(settings.revision_review_cutoff)
+        row = self.revision_progress(day)
+        if current < start:
+            if row is None:
+                self._save_revision_progress(day, state="waiting_review")
+            return []
+        if current >= cutoff:
+            if row is None or row["state"] not in {"aligned", "cutoff"}:
+                detail = "复核窗口已截止"
+                self._save_revision_progress(
+                    day, state="cutoff", finished_at=isoformat(now), detail=detail,
+                )
+                with self.db.write() as connection:
+                    connection.execute(
+                        "INSERT INTO job_runs(job_id, state, started_at, finished_at, error, attempt, detail)"
+                        " VALUES ('eod_revise','failed',?,?,?,?,?)",
+                        (
+                            isoformat(now), isoformat(now), detail,
+                            int((row or {}).get("attempts") or 0), "revision cutoff reached",
+                        ),
+                    )
+            elif row["state"] == "aligned" and not row.get("finished_at"):
+                self._save_revision_progress(day, finished_at=isoformat(now))
+            return []
+        if not self._revision_due(now, row):
+            return []
+        if "eod_revise" not in self.jobs:
+            return []
+        return self._run_revision_job(day, now, catchup=False)
+
+    def _revision_catchup_tick(self, now: datetime, day: str) -> list[str]:
+        prev = previous_open_day(self.db, day)
+        if prev is None or prev >= day:
+            return []
+        if not self._revision_ready(prev):
+            return []
+        row = self.revision_progress(prev)
+        if row and int(row.get("catchup_done") or 0):
+            return []
+        if not self._revision_due(now, row):
+            return []
+        if "eod_revise" not in self.jobs:
+            return []
+        return self._run_revision_job(prev, now, catchup=True)
+
+    def _run_revision_job(self, day: str, now: datetime, catchup: bool) -> list[str]:
+        attempts = int((self.revision_progress(day) or {}).get("attempts") or 0) + 1
+        interval = self.pipeline.settings.revision_review_interval_minutes
+        self._save_revision_progress(
+            day,
+            state="waiting_review",
+            attempts=attempts,
+            last_attempt_at=isoformat(now),
+            next_retry_at=isoformat(now + timedelta(minutes=interval)),
+        )
+        ran: list[str] = []
+        try:
+            out = self.run_job("eod_revise", day)
+        except Exception as exc:
+            LOGGER.warning("revision review failed for %s: %s", day, exc)
+            self._save_revision_progress(
+                day,
+                state="review_failed",
+                detail="复核失败，保留上一完整版本",
+            )
+            ran.append("eod_revise")
+            return ran
+        ran.append("eod_revise")
+        if out.get("state") == "skipped":
+            return ran
+        result = out.get("result") if isinstance(out.get("result"), dict) else {}
+        failed = [
+            name for name, item in result.items()
+            if isinstance(item, dict) and item.get("state") == "failed"
+        ]
+        review = result.get("review") if isinstance(result.get("review"), dict) else None
+        watched = [
+            result[name]
+            for name in revision_datasets(self.pipeline.settings.quality)
+            if isinstance(result.get(name), dict)
+        ]
+        diff_blob = None
+        if review and review.get("diffs"):
+            diff_blob = json.dumps(review.get("diffs"), ensure_ascii=False)
+        else:
+            for item in watched:
+                if item.get("diffs"):
+                    diff_blob = json.dumps(item.get("diffs"), ensure_ascii=False)
+                    break
+        revised = bool(review and review.get("reason") == "revised")
+        matched = any(item.get("reason") == "unchanged" or item.get("state") == "aligned" for item in watched)
+        if failed:
+            self._save_revision_progress(
+                day,
+                state="review_failed",
+                detail="复核失败，保留上一完整版本",
+                last_diff=diff_blob,
+            )
+        elif revised or matched:
+            fields: dict[str, Any] = {
+                "state": "aligned",
+                "finished_at": isoformat(now),
+                "detail": "已追平" if revised else "已追平（无变化）",
+                "last_diff": diff_blob,
+            }
+            if catchup:
+                fields["catchup_done"] = 1
+            self._save_revision_progress(day, **fields)
+        return ran
+
+    def _save_revision_progress(self, day: str, **fields: Any) -> None:
+        columns = [
+            "trade_date", "state", "attempts", "last_attempt_at",
+            "next_retry_at", "finished_at", "catchup_done", "last_diff", "detail", "updated_at",
+        ]
+        with self.db.write() as connection:
+            existing = connection.execute(
+                "SELECT trade_date FROM revision_progress WHERE trade_date = ?",
+                (day,),
+            ).fetchone()
+            if existing is None:
+                payload = {name: None for name in columns}
+                payload.update({
+                    "trade_date": day,
+                    "state": "waiting_review",
+                    "attempts": 0,
+                    "catchup_done": 0,
+                })
+                payload.update(fields)
+                payload["updated_at"] = isoformat()
+                placeholders = ",".join("?" for _ in columns)
+                connection.execute(
+                    f"INSERT INTO revision_progress({','.join(columns)}) VALUES ({placeholders})",
+                    tuple(payload[name] for name in columns),
+                )
+            else:
+                assignments = ", ".join(f"{name} = ?" for name in fields)
+                connection.execute(
+                    f"UPDATE revision_progress SET {assignments}, updated_at = ? WHERE trade_date = ?",
+                    (*fields.values(), isoformat(), day),
+                )
+
     def _record_eod_attempt(self, day: str, now: datetime) -> None:
         row = self.eod_progress(day)
         attempts = int((row or {}).get("attempts") or 0) + 1
@@ -320,6 +534,9 @@ class Scheduler:
     def _eod_retry(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.run_eod_missing(trade_date)
 
+    def _eod_revise(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.review_published_revisions(trade_date)
+
     def _stocks_refresh(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.refresh_stocks(trade_date)
 

xiaobai-datahub/datahub/settings.py

@@ -79,6 +79,20 @@ class Settings:
     def eod_retry_cutoff(self) -> str:
         return str(self.quality.get("eod_retry_cutoff") or "23:30")
 
+    @property
+    def revision_review_start(self) -> str:
+        # Before the 21:00 website shadow observation.
+        return str(self.quality.get("revision_review_start") or "20:00")
+
+    @property
+    def revision_review_interval_minutes(self) -> int:
+        return int(self.quality.get("revision_review_interval_minutes") or 30)
+
+    @property
+    def revision_review_cutoff(self) -> str:
+        # Last light review ~23:00; cutoff before the 23:30 observation.
+        return str(self.quality.get("revision_review_cutoff") or "23:20")
+
 
 def load_settings(
     env: dict[str, str] | None = None,

xiaobai-datahub/tests/test_eod_retry.py

@@ -111,15 +111,24 @@ class EodRetryTests(unittest.TestCase):
         self.assertEqual(progress["state"], "done")
         self.assertEqual(progress["attempts"], 4)  # eod_a + eod_b + 2 retries
 
-        # success stops all further same-day requests
+        # success stops further eod_retry; revision window has not started yet
         batches_before = len(self._batches(db, day))
         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._job_runs(db, "eod_revise")), 0)
         self.assertEqual(len(self._batches(db, day)), batches_before)
         self.assertEqual(len(self._eod_calls(transport)), eod_calls_before)
 
+        # 23:00 is inside the valuation review window: light daily_basic only, no new batch
+        sched.tick(clock_at(day, 23, 0))
+        self.assertEqual(len(self._job_runs(db, "eod_retry")), 2)
+        self.assertEqual(len(self._job_runs(db, "eod_revise")), 1)
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+        extra = [name for name in self._eod_calls(transport)[eod_calls_before:]]
+        self.assertTrue(extra)
+        self.assertTrue(all(name == "daily_basic" for name in extra))
+
     def test_never_ready_marks_cutoff_failed_and_stops(self) -> None:
         day = "20240902"
         db, transport, pipe, sched = self._make(set())
@@ -174,6 +183,7 @@ class EodRetryTests(unittest.TestCase):
         self.assertIn("eod_a", ran)
         self.assertIn("eod_b", ran)
         self.assertNotIn("eod_retry", ran)
+        self.assertIn("eod_revise", ran)
         self.assertEqual(self._published(db, day), OFFICIAL)
         after = db.fetchall("SELECT dataset, active_batch FROM publications WHERE trade_date = ?", (day,))
         self.assertEqual(
@@ -181,7 +191,9 @@ class EodRetryTests(unittest.TestCase):
             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
+        extra = self._eod_calls(transport)[len(calls_before):]
+        self.assertTrue(extra)
+        self.assertTrue(all(name == "daily_basic" for name in extra))
         self.assertEqual(sched2.eod_status(day, clock=clock_at(day, 21, 0))["state"], "done")
 
     def test_restart_with_partial_publish_only_fetches_missing(self) -> None:
@@ -206,6 +218,7 @@ class EodRetryTests(unittest.TestCase):
         for hh, mm in ((15, 5), (15, 10), (15, 40), (16, 10), (20, 0), (23, 40)):
             ran = sched.tick(clock_at(day, hh, mm))
             self.assertNotIn("eod_retry", ran)
+            self.assertNotIn("eod_revise", ran)
         eod_runs = db.fetchall("SELECT * FROM job_runs WHERE job_id LIKE 'eod%'")
         self.assertEqual(eod_runs, [])
         self.assertIsNone(db.fetchone("SELECT * FROM eod_progress WHERE trade_date = ?", (day,)))
@@ -226,12 +239,18 @@ class EodRetryTests(unittest.TestCase):
         self.assertEqual(len(self._batches(db, day)), batches_before)
         self.assertEqual(len(transport.calls), calls_before)
 
+        revised = sched.run_job("eod_revise", day)
+        self.assertEqual(revised["state"], "ok")
+        self.assertEqual(len(self._batches(db, day)), batches_before)
+
         sched._eod_lock.acquire()  # simulate an in-flight EOD job
         try:
             busy = sched.run_job("eod_retry", day)
             self.assertEqual(busy["state"], "skipped")
             busy_a = sched.run_job("eod_a", day)
             self.assertEqual(busy_a["state"], "skipped")
+            busy_r = sched.run_job("eod_revise", day)
+            self.assertEqual(busy_r["state"], "skipped")
         finally:
             sched._eod_lock.release()
         self.assertEqual(len(self._batches(db, day)), batches_before)

xiaobai-datahub/tests/test_revision_review.py

@@ -0,0 +1,333 @@
+from __future__ import annotations
+
+import copy
+import unittest
+from pathlib import Path
+import tempfile
+
+from datahub.adapters.base import AdapterError
+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 datahub.timeutil import SHANGHAI
+from tests.fixtures import RAW, TRADE_DATE, fake_transport
+from tests.test_eod_retry import clock_at
+from tests.test_quality_gates import FIELD_GATES
+
+SAMPLE_DAY = "20260907"
+NEXT_DAY = "20260908"
+SAMPLE_CODE = "003021.SZ"
+
+
+def _dated(row: dict, day: str) -> dict:
+    item = dict(row)
+    if "trade_date" in item:
+        item["trade_date"] = day
+    return item
+
+
+class RevisingTransport:
+    """Fixture transport that can rewrite daily_basic after the first publish."""
+
+    DATE_APIS = {"daily", "daily_basic", "adj_factor", "moneyflow", "stk_auction", "index_daily"}
+
+    def __init__(self, extra_calendar: list[dict] | None = None) -> None:
+        self.calls: list[str] = []
+        self.fail_daily_basic = False
+        self.empty_daily_basic = False
+        self.null_volume_ratio = False
+        self.turnover_by_code: dict[str, float] = {}
+        self.extra_calendar = extra_calendar or []
+
+    def __call__(self, api_name: str, params: dict, fields: str):
+        self.calls.append(api_name)
+        day = str(params.get("trade_date") or "")
+        if api_name == "trade_cal":
+            rows = fake_transport(api_name, params, fields)
+            extra = [
+                row for row in self.extra_calendar
+                if str(params.get("start_date") or "") <= row["cal_date"] <= str(params.get("end_date") or "99999999")
+            ]
+            return rows + extra
+        if self.fail_daily_basic and api_name == "daily_basic":
+            raise AdapterError("tushare daily_basic unavailable")
+        if self.empty_daily_basic and api_name == "daily_basic":
+            return []
+        if api_name == "index_daily":
+            code = params.get("ts_code")
+            rows = [row for row in RAW["index_daily"] if row["ts_code"] == code]
+            if day:
+                rows = [_dated(row, day) for row in rows]
+            return rows
+        rows = fake_transport(api_name, params, fields)
+        if api_name == "stock_basic":
+            rows = list(rows)
+            rows.append({
+                "ts_code": SAMPLE_CODE, "symbol": "003021", "name": "兆威机电",
+                "area": "广东", "industry": "元器件", "market": "主板",
+                "list_status": "L", "list_date": "20201202",
+            })
+            return rows
+        if api_name in self.DATE_APIS:
+            template = RAW.get(api_name) or []
+            if not day:
+                return [_dated(row, TRADE_DATE) for row in template]
+            out = [_dated(row, day) for row in template]
+            extra = copy.deepcopy(template[0])
+            extra["ts_code"] = SAMPLE_CODE
+            extra["trade_date"] = day
+            if api_name == "daily_basic":
+                extra["turnover_rate"] = self.turnover_by_code.get(SAMPLE_CODE, extra.get("turnover_rate"))
+                if self.null_volume_ratio:
+                    extra["volume_ratio"] = None
+                    for row in out:
+                        row["volume_ratio"] = None
+            out.append(extra)
+            if api_name == "daily_basic":
+                for row in out:
+                    code = str(row.get("ts_code") or "")
+                    if code in self.turnover_by_code:
+                        row["turnover_rate"] = self.turnover_by_code[code]
+            return out
+        return rows
+
+
+def make_revision_env(quality_extra: dict | None = None, extra_calendar: list[dict] | None = None):
+    tmp = tempfile.TemporaryDirectory()
+    db = HubDB(Path(tmp.name) / "hub.db")
+    transport = RevisingTransport(extra_calendar=extra_calendar)
+    adapter = TushareAdapter("x", transport=transport)
+    quality = {
+        "daily_row_ratio": 0.5,
+        "null_rate_max": 0.5,
+        "max_publish_attempts": 2,
+        "publication_generations": 3,
+        "field_gates": FIELD_GATES,
+        "revision_review_start": "20:00",
+        "revision_review_interval_minutes": 30,
+        "revision_review_cutoff": "23:20",
+        "revision_review_datasets": ["valuation"],
+    }
+    if quality_extra:
+        quality.update(quality_extra)
+    settings = Settings(
+        encryption_key=SecretVault.generate_key(),
+        api_token="t" * 32,
+        db_path=db.path,
+        backup_dir=Path(tmp.name) / "backups",
+        quality=quality,
+        scheduler_enabled=False,
+    )
+    pipe = Pipeline(db, adapter, settings)
+    sched = Scheduler(db, pipe)
+    return tmp, db, transport, pipe, sched
+
+
+SAMPLE_CALENDAR = [
+    {"exchange": "SSE", "cal_date": SAMPLE_DAY, "is_open": 1, "pretrade_date": "20260906"},
+    {"exchange": "SSE", "cal_date": NEXT_DAY, "is_open": 1, "pretrade_date": SAMPLE_DAY},
+]
+
+
+class RevisionReviewTests(unittest.TestCase):
+    def _publish(self, pipe: Pipeline, day: str) -> None:
+        pipe.ingest_reference(day)
+        pipe.run_eod_batch_a(day)
+        pipe.run_eod_batch_b(day)
+
+    def _turnover(self, db: HubDB, day: str, code: str = SAMPLE_CODE) -> float | None:
+        pub = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (day,),
+        )
+        row = db.fetchone(
+            "SELECT turnover_rate FROM eod_valuation WHERE batch_id=? AND ts_code=?",
+            (pub["active_batch"], code),
+        )
+        return None if row is None else row["turnover_rate"]
+
+    def _batch_ids(self, db: HubDB, day: str) -> set[str]:
+        return {str(row["batch_id"]) for row in db.fetchall("SELECT batch_id FROM batches WHERE trade_date=?", (day,))}
+
+    def test_no_change_does_not_create_a_new_batch(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        before = self._batch_ids(db, TRADE_DATE)
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), before)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertIn("无变化", progress["detail"])
+        status = sched.revision_status(TRADE_DATE, clock=clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(status["state"], "aligned")
+
+    def test_hel423_20260907_single_field_revision_is_caught_up(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env(extra_calendar=SAMPLE_CALENDAR)
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, SAMPLE_DAY)
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+        first = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (SAMPLE_DAY,),
+        )["active_batch"]
+
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        seen: list[float | None] = []
+
+        def watch() -> None:
+            seen.append(self._turnover(db, SAMPLE_DAY))
+
+        pipe.before_commit = watch
+        ran = sched.tick(clock_at(SAMPLE_DAY, 20, 0))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(seen, [1.3565])  # readers still see the previous complete version mid-switch
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3572)
+        second = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (SAMPLE_DAY,),
+        )["active_batch"]
+        self.assertNotEqual(second, first)
+        api = V1API(db, pipe, pipe.settings)
+        payload = api.valuation({"date": SAMPLE_DAY, "code": SAMPLE_CODE})
+        row = next(item for item in payload["data"] if item["ts_code"] == SAMPLE_CODE)
+        self.assertEqual(row["turnover_rate"], 1.3572)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertEqual(progress["detail"], "已追平")
+        audit = db.fetchone(
+            "SELECT * FROM audit_log WHERE action='revision-review' ORDER BY id DESC"
+        )
+        self.assertIn("1.3572", str(audit["detail"]))
+        self.assertIn(SAMPLE_CODE, str(audit["detail"]))
+
+    def test_empty_or_failed_upstream_keeps_previous_version(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        active = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (TRADE_DATE,),
+        )["active_batch"]
+        batches = self._batch_ids(db, TRADE_DATE)
+
+        transport.empty_daily_basic = True
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(
+            db.fetchone("SELECT state FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))["state"],
+            "review_failed",
+        )
+
+        transport.empty_daily_basic = False
+        transport.fail_daily_basic = True
+        sched.tick(clock_at(TRADE_DATE, 20, 30))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), batches)
+
+    def test_quality_gate_rejects_catchup_and_keeps_previous(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        active = db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+            (TRADE_DATE,),
+        )["active_batch"]
+        transport.turnover_by_code[SAMPLE_CODE] = 9.9999
+        transport.null_volume_ratio = True
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        self.assertEqual(
+            db.fetchone(
+                "SELECT active_batch FROM publications WHERE dataset='valuation' AND trade_date=?",
+                (TRADE_DATE,),
+            )["active_batch"],
+            active,
+        )
+        self.assertEqual(
+            db.fetchone("SELECT state FROM revision_progress WHERE trade_date=?", (TRADE_DATE,))["state"],
+            "review_failed",
+        )
+
+    def test_repeat_ticks_after_align_do_not_republish(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, TRADE_DATE)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        after_fix = self._batch_ids(db, TRADE_DATE)
+        sched.tick(clock_at(TRADE_DATE, 20, 10))  # inside interval
+        self.assertEqual(len(db.fetchall("SELECT * FROM job_runs WHERE job_id='eod_revise'")), 1)
+        sched.tick(clock_at(TRADE_DATE, 20, 30))  # next light compare, no change
+        self.assertEqual(self._batch_ids(db, TRADE_DATE), after_fix)
+        self.assertEqual(self._turnover(db, TRADE_DATE), 1.3572)
+
+    def test_restart_catches_up_inside_window(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, TRADE_DATE)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched2 = Scheduler(db, pipe)
+        ran = sched2.tick(clock_at(TRADE_DATE, 21, 0))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(self._turnover(db, TRADE_DATE), 1.3572)
+
+    def test_cutoff_stops_evening_reviews_and_morning_catchup_runs(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env(extra_calendar=SAMPLE_CALENDAR)
+        self.addCleanup(tmp.cleanup)
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3565
+        self._publish(pipe, SAMPLE_DAY)
+        sched.tick(clock_at(SAMPLE_DAY, 23, 25))  # past 23:20 cutoff, no review yet
+        cutoff = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(cutoff["state"], "cutoff")
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+
+        transport.turnover_by_code[SAMPLE_CODE] = 1.3572
+        sched.tick(clock_at(SAMPLE_DAY, 23, 50))  # still same calendar day, no catch-up
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3565)
+
+        ran = sched.tick(clock_at(NEXT_DAY, 8, 45))
+        self.assertIn("eod_revise", ran)
+        self.assertEqual(self._turnover(db, SAMPLE_DAY), 1.3572)
+        progress = db.fetchone("SELECT * FROM revision_progress WHERE trade_date=?", (SAMPLE_DAY,))
+        self.assertEqual(progress["state"], "aligned")
+        self.assertEqual(int(progress["catchup_done"]), 1)
+
+        batches = self._batch_ids(db, SAMPLE_DAY)
+        sched.tick(clock_at(NEXT_DAY, 8, 50))
+        self.assertEqual(self._batch_ids(db, SAMPLE_DAY), batches)
+
+    def test_only_valuation_is_light_fetched(self) -> None:
+        tmp, db, transport, pipe, sched = make_revision_env()
+        self.addCleanup(tmp.cleanup)
+        self._publish(pipe, TRADE_DATE)
+        before = [name for name in transport.calls]
+        sched.tick(clock_at(TRADE_DATE, 20, 0))
+        extra = transport.calls[len(before):]
+        self.assertIn("daily_basic", extra)
+        self.assertNotIn("daily", extra)
+        self.assertNotIn("moneyflow", extra)
+        self.assertNotIn("stk_auction", extra)
+        self.assertNotIn("index_daily", extra)
+
+
+if __name__ == "__main__":
+    unittest.main()