feat(HEL-463): 接入剩余行情数据到 datahub

扩展盘后正式集（涨跌停/人气/龙虎榜/板块日线）与盘中观察 API（报价/指数/分时），网站 bridge 按开关接入并回退旧链路；问天改为按数据依赖跟随开关，不再整栈强制旧路径。

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

.env.example

@@ -7,6 +7,9 @@ TUSHARE_TOKEN=your_tushare_token_here
 
 # Optional xiaobai-datahub client. All DATAHUB_READ_* / DATAHUB_SHADOW_* flags
 # default off in config/datahub.config.json, so the website keeps using Tushare.
+# Extended datasets (HEL-463): LIMIT_EVENTS POPULARITY DRAGON_TIGER SECTOR_DAILY
+# QUOTES INDEX_QUOTES INTRADAY — plus first-batch CALENDAR STOCKS DAILY INDEX_DAILY
+# VALUATION MONEYFLOW AUCTION STATUS.
 DATAHUB_BASE_URL=http://127.0.0.1:8766
 DATAHUB_TOKEN=
 

backend/data/datahub/bridge.py

@@ -21,11 +21,14 @@ from backend.data.providers.tushare_client import TushareClient
 
 LOGGER = logging.getLogger("xiaobai.datahub")
 ShadowSink = Callable[[dict[str, Any]], None]
-EMPTY_FAIL_DATASETS = {"stocks", "daily", "index_daily", "valuation", "moneyflow", "auction"}
+EMPTY_FAIL_DATASETS = {
+    "stocks", "daily", "index_daily", "valuation", "moneyflow", "auction",
+    "limit_events", "sector_daily",
+}
 
 
 def looks_like_heaven(module_name: str, filename: str = "") -> bool:
-    """问天调用栈识别。问天未永久冻结，只是本阶段仍走旧 Tushare 链路。"""
+    """问天调用栈识别（诊断用）。问天按数据集依赖接入，不再整栈强制旧链路。"""
     path = filename.replace("\\", "/")
     return module_name.startswith("backend.features.heaven") or "/features/heaven/" in path
 
@@ -96,8 +99,8 @@ class DatahubBridge:
         legacy_query: Callable[..., list[dict[str, Any]]],
     ) -> list[dict[str, Any]]:
         dataset = API_TO_DATASET.get(api_name)
-        # 问天允许后续纳入 datahub；首批只读接入仍保持旧链路，避免误切。
+        # 问天按实际数据依赖接入：已映射到 hub 的 API 跟随开关；未映射的继续旧链路。
-        if not dataset or self.heaven_guard():
+        if not dataset:
             return legacy_query(api_name, params, fields)
         flags = self.settings.flags(dataset)
         if not flags.read and not flags.shadow:
@@ -108,7 +111,7 @@ class DatahubBridge:
         hub_error: str | None = None
         hub_canonical: list[dict[str, Any]] = []
         try:
-            response = self._fetch_dataset(dataset, params or {})
+            response = self._fetch_dataset(dataset, params or {}, api_name=api_name)
             hub_canonical = self._extract_rows(dataset, response, params or {})
             hub_rows = to_native_rows(dataset, hub_canonical)
             hub_meta = dict(response.meta)
@@ -136,7 +139,7 @@ class DatahubBridge:
             return project_fields(hub_rows, fields)
         return legacy_query(api_name, params, fields)
 
-    def _fetch_dataset(self, dataset: str, params: dict[str, Any]) -> DatahubResponse:
+    def _fetch_dataset(self, dataset: str, params: dict[str, Any], api_name: str = "") -> DatahubResponse:
         date = yyyymmdd(params.get("trade_date") or params.get("date"))
         start = yyyymmdd(params.get("start_date") or params.get("from") or date)
         end = yyyymmdd(params.get("end_date") or params.get("to") or date)
@@ -153,6 +156,10 @@ class DatahubBridge:
             "valuation": self.client.valuation,
             "moneyflow": self.client.moneyflow,
             "auction": self.client.auction,
+            "limit_events": self.client.limit_events,
+            "popularity": self.client.popularity,
+            "dragon_tiger": self.client.dragon_tiger,
+            "sector_daily": self.client.sectors,
         }
         fetcher = fetchers[dataset]
         query: dict[str, Any] = {}
@@ -167,6 +174,23 @@ class DatahubBridge:
                 query["to"] = end
         if dataset == "daily":
             query["adjust"] = "none"
+        if dataset == "limit_events":
+            limit_type = str(params.get("limit_type") or "").strip().upper()
+            if limit_type:
+                query["limit_type"] = limit_type
+        if dataset == "popularity":
+            if api_name == "ths_hot":
+                query["source"] = "ths"
+            elif api_name == "dc_hot":
+                query["source"] = "dc"
+        if dataset == "sector_daily":
+            family = {
+                "ths_daily": "ths",
+                "dc_index": "dc",
+                "sw_daily": "sw",
+            }.get(api_name, "")
+            if family:
+                query["family"] = family
         return self._paginate(fetcher, query)
 
     def _paginate(self, fetcher: Callable[..., DatahubResponse], params: dict[str, Any]) -> DatahubResponse:

backend/data/datahub/client.py

@@ -60,6 +60,27 @@ class DatahubClient:
     def auction(self, **params: Any) -> DatahubResponse:
         return self.get("/v1/auction", params)
 
+    def limit_events(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/limit-events", params)
+
+    def popularity(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/popularity", params)
+
+    def dragon_tiger(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/dragon-tiger", params)
+
+    def sectors(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/sectors", params)
+
+    def quotes_latest(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/quotes/latest", params)
+
+    def index_quotes(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/indexes/quotes", params)
+
+    def intraday_points(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/intraday/points", params)
+
     def dataset_status(self, date: str) -> DatahubResponse:
         return self.get("/v1/datasets/status", {"date": date})
 

backend/data/datahub/native.py

@@ -17,6 +17,13 @@ API_TO_DATASET = {
     "index_daily": "index_daily",
     "moneyflow": "moneyflow",
     "stk_auction": "auction",
+    "limit_list_d": "limit_events",
+    "ths_hot": "popularity",
+    "dc_hot": "popularity",
+    "hm_detail": "dragon_tiger",
+    "ths_daily": "sector_daily",
+    "dc_index": "sector_daily",
+    "sw_daily": "sector_daily",
 }
 
 SCALE_FIELDS = {
@@ -35,6 +42,16 @@ SCALE_FIELDS = {
         "net_mf_amount": AMOUNT_WAN_YUAN,
     },
     "auction": {"vol": VOLUME_LOT, "float_share": AMOUNT_WAN_YUAN},
+    "limit_events": {
+        "limit_amount": AMOUNT_WAN_YUAN,
+        "float_mv": AMOUNT_WAN_YUAN,
+        "total_mv": AMOUNT_WAN_YUAN,
+    },
+    "dragon_tiger": {
+        "buy_amount": AMOUNT_WAN_YUAN,
+        "sell_amount": AMOUNT_WAN_YUAN,
+        "net_amount": AMOUNT_WAN_YUAN,
+    },
 }
 
 
@@ -67,6 +84,16 @@ def to_native_row(dataset: str, row: dict[str, Any]) -> dict[str, Any]:
             converted[field] = _unscale(converted.get(field), factor)
     if dataset == "stocks":
         converted.pop("updated_at", None)
+    if dataset == "popularity":
+        # keep hub source; callers filter ths/dc themselves when needed
+        if converted.get("ts_name") and not converted.get("name"):
+            converted["name"] = converted.get("ts_name")
+    if dataset == "dragon_tiger":
+        if converted.get("ts_name") and not converted.get("name"):
+            converted["name"] = converted.get("ts_name")
+    if dataset == "sector_daily":
+        if converted.get("pct_change") is not None and converted.get("pct_chg") is None:
+            converted["pct_chg"] = converted.get("pct_change")
     return converted
 
 
@@ -96,6 +123,30 @@ def row_key(dataset: str, row: dict[str, Any]) -> tuple[str, ...]:
         return (str(row.get("ts_code") or "").upper(),)
     if dataset == "status":
         return (str(row.get("dataset") or ""), yyyymmdd(row.get("trade_date")))
+    if dataset == "limit_events":
+        return (
+            str(row.get("ts_code") or "").upper(),
+            yyyymmdd(row.get("trade_date")),
+            str(row.get("limit_type") or ""),
+        )
+    if dataset == "popularity":
+        return (
+            str(row.get("ts_code") or "").upper(),
+            yyyymmdd(row.get("trade_date")),
+            str(row.get("source") or ""),
+        )
+    if dataset == "dragon_tiger":
+        return (
+            str(row.get("ts_code") or "").upper(),
+            yyyymmdd(row.get("trade_date")),
+            str(row.get("hm_name") or ""),
+        )
+    if dataset == "sector_daily":
+        return (
+            str(row.get("ts_code") or "").upper(),
+            yyyymmdd(row.get("trade_date")),
+            str(row.get("family") or ""),
+        )
     return (str(row.get("ts_code") or "").upper(), yyyymmdd(row.get("trade_date")))
 
 

backend/data/datahub/settings.py

@@ -17,6 +17,13 @@ DATASETS = (
     "valuation",
     "moneyflow",
     "auction",
+    "limit_events",
+    "popularity",
+    "dragon_tiger",
+    "sector_daily",
+    "quotes",
+    "index_quotes",
+    "intraday",
     "status",
 )
 
@@ -28,6 +35,13 @@ ENV_DATASET = {
     "valuation": "VALUATION",
     "moneyflow": "MONEYFLOW",
     "auction": "AUCTION",
+    "limit_events": "LIMIT_EVENTS",
+    "popularity": "POPULARITY",
+    "dragon_tiger": "DRAGON_TIGER",
+    "sector_daily": "SECTOR_DAILY",
+    "quotes": "QUOTES",
+    "index_quotes": "INDEX_QUOTES",
+    "intraday": "INTRADAY",
     "status": "STATUS",
 }
 

config/datahub.config.json

@@ -13,6 +13,13 @@
     "valuation": { "read": false, "shadow": false },
     "moneyflow": { "read": false, "shadow": false },
     "auction": { "read": false, "shadow": false },
+    "limit_events": { "read": false, "shadow": false },
+    "popularity": { "read": false, "shadow": false },
+    "dragon_tiger": { "read": false, "shadow": false },
+    "sector_daily": { "read": false, "shadow": false },
+    "quotes": { "read": false, "shadow": false },
+    "index_quotes": { "read": false, "shadow": false },
+    "intraday": { "read": false, "shadow": false },
     "status": { "read": false, "shadow": false }
   }
 }

tests/test_datahub_bridge.py

@@ -290,8 +290,8 @@ class DatahubBridgeTests(unittest.TestCase):
         self.assertEqual(canonical["vol"], 100000.0)
         self.assertEqual(canonical["amount"], 2000000.0)
 
-    def test_heaven_keeps_legacy_on_first_batch_even_when_read_flag_is_on(self) -> None:
+    def test_heaven_can_use_hub_when_dataset_flag_is_on(self) -> None:
-        """问天未永久冻结；首批只读接入仍走旧链路，后续迁移可以纳入。"""
+        """问天按数据依赖接入：已映射 API 跟随开关，不再整栈强制旧链路。"""
         self.assertTrue(looks_like_heaven("backend.features.heaven.market_context", "backend/features/heaven/market_context.py"))
         self.assertFalse(looks_like_heaven("backend.features.market.service", "backend/features/market/service.py"))
         client = FakeClient()
@@ -302,7 +302,8 @@ class DatahubBridgeTests(unittest.TestCase):
         )
         rows = wrapped.query("daily", {"trade_date": "20240902"}, "amount")
         self.assertEqual(rows[0]["amount"], 2000.0)
-        self.assertEqual(client.paths, [])
+        self.assertEqual(client.paths, ["/v1/bars/daily"])
+        self.assertEqual(legacy.calls, [])
 
     def test_status_flag_does_not_run_when_off_and_falls_back_when_on(self) -> None:
         off = DatahubBridge(flags(), FakeClient(error=DatahubError("UNAVAILABLE", "down")))

xiaobai-datahub/README.md

@@ -6,11 +6,12 @@
 ## 做什么
 
 - SQLite WAL `datahub.db`，容器名 `xiaobai-datahub`，端口 `8766`
-- Tushare 盘后正式数据：交易日历、股票主档、daily、daily_basic、adj_factor、index_daily、moneyflow、stk_auction
+- Tushare 盘后正式数据：交易日历、股票主档、daily、daily_basic、adj_factor、index_daily、moneyflow、stk_auction、limit_list_d、ths_hot/dc_hot、hm_detail、ths_daily/dc_index/sw_daily
+- 盘中观察（provisional）：东财/腾讯指数报价、个股最新价、分时点（`/v1/quotes/latest` `/v1/indexes/quotes` `/v1/intraday/points`）；永不写入 eod_* 正式表
 - 暂存 → 校验 → 整批原子发布 → 可回滚
 - `/v1` 稳定接口（`X-Datahub-Token`）
 - `/admin/` 最小管理后台（总览 / 数据源 / 调度 / 发布 / 数据集 / 审计）
-- 东财/腾讯/同花顺/选股宝/AKShare/iFinD 适配器位已预留，本阶段不拉实时源
+- 同花顺/选股宝/AKShare/iFinD 适配器位仍预留；东财/腾讯已接入盘中观察
 
 ## 单位口径（相对现站）
 
@@ -123,18 +124,6 @@ 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,7 +105,6 @@ 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: "等待上游",
@@ -113,14 +112,6 @@ 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} 次`);
@@ -130,16 +121,12 @@ 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,10 +17,6 @@
   "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/adapters/__init__.py

@@ -1,13 +1,13 @@
 from datahub.adapters.akshare import ADAPTER as akshare
-from datahub.adapters.eastmoney import ADAPTER as eastmoney
+from datahub.adapters.eastmoney import EastmoneyAdapter
 from datahub.adapters.ifind import ADAPTER as ifind
-from datahub.adapters.tencent import ADAPTER as tencent
+from datahub.adapters.tencent import TencentAdapter
 from datahub.adapters.ths import ADAPTER as ths
 from datahub.adapters.xgb import ADAPTER as xgb
 
 RESERVED = {
-    "eastmoney": eastmoney,
+    "eastmoney": EastmoneyAdapter(),
-    "tencent": tencent,
+    "tencent": TencentAdapter(),
     "ths": ths,
     "xgb": xgb,
     "akshare": akshare,

xiaobai-datahub/datahub/adapters/eastmoney.py

@@ -1,3 +1,250 @@
-from datahub.adapters.base import ReservedAdapter
+from __future__ import annotations
 
-ADAPTER = ReservedAdapter("eastmoney")
+import json
+import time
+import urllib.error
+import urllib.parse
+import urllib.request
+from datetime import datetime
+from typing import Any
+
+from datahub.adapters.base import AdapterError, MarketAdapter
+from datahub.numbers import finite_number, round4
+
+EASTMONEY_INDEX_URL = "https://push2.eastmoney.com/api/qt/ulist.np/get"
+EASTMONEY_CLIST_URL = "https://push2.eastmoney.com/api/qt/clist/get"
+TRENDS_URL = "https://push2delay.eastmoney.com/api/qt/stock/trends2/get"
+BROWSER_UA = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"
+)
+INDEX_SECIDS = {
+    "000001.SH": "1.000001",
+    "399001.SZ": "0.399001",
+    "399006.SZ": "0.399006",
+}
+
+
+class EastmoneyAdapter(MarketAdapter):
+    name = "eastmoney"
+
+    def __init__(self, timeout: int = 8) -> None:
+        self.timeout = timeout
+
+    def probe(self) -> dict[str, Any]:
+        started = time.perf_counter()
+        try:
+            rows = self.fetch_indices()
+            state = "ok" if len(rows) == 3 else "empty"
+        except AdapterError as exc:
+            return {
+                "provider": self.name,
+                "configured": True,
+                "state": "error",
+                "message": str(exc),
+                "latency_ms": round((time.perf_counter() - started) * 1000),
+            }
+        return {
+            "provider": self.name,
+            "configured": True,
+            "state": state,
+            "latency_ms": round((time.perf_counter() - started) * 1000),
+        }
+
+    def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
+        if dataset in {"indexes_quotes", "index_quotes"}:
+            return self.fetch_indices()
+        if dataset in {"quotes", "quotes_latest"}:
+            codes = params.get("codes") or []
+            if isinstance(codes, str):
+                codes = [item.strip() for item in codes.split(",") if item.strip()]
+            return self.fetch_quotes(list(codes))
+        raise AdapterError(f"{self.name} unsupported dataset: {dataset}")
+
+    def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        return list(rows)
+
+    def fetch_indices(self) -> list[dict[str, Any]]:
+        payload = self._get_json(
+            EASTMONEY_INDEX_URL,
+            {
+                "secids": "1.000001,0.399001,0.399006",
+                "fltt": "2",
+                "invt": "2",
+                "fields": "f12,f14,f2,f3,f4,f15,f16,f17,f18,f6,f124",
+            },
+            referer="https://quote.eastmoney.com/",
+        )
+        rows = list((payload.get("data") or {}).get("diff") or [])
+        result = []
+        for row in rows:
+            code = str(row.get("f12") or "")
+            if code not in {"000001", "399001", "399006"}:
+                continue
+            epoch = int(finite_number(row.get("f124")) or 0)
+            ts_code = f"{code}.SH" if code.startswith("0") and code == "000001" else f"{code}.SZ"
+            if code == "000001":
+                ts_code = "000001.SH"
+            result.append(
+                {
+                    "ts_code": ts_code,
+                    "code": code,
+                    "name": row.get("f14") or code,
+                    "price": round4(finite_number(row.get("f2"))),
+                    "pct_chg": round4(finite_number(row.get("f3"))),
+                    "change_amount": round4(finite_number(row.get("f4"))),
+                    "open": round4(finite_number(row.get("f17"))),
+                    "high": round4(finite_number(row.get("f15"))),
+                    "low": round4(finite_number(row.get("f16"))),
+                    "previous_close": round4(finite_number(row.get("f18"))),
+                    "amount": round4(finite_number(row.get("f6"))),
+                    "quote_time_epoch": epoch,
+                    "quote_time": (
+                        datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds")
+                        if epoch
+                        else ""
+                    ),
+                    "source": "eastmoney_push2",
+                }
+            )
+        if len(result) != 3:
+            raise AdapterError(f"Eastmoney returned {len(result)}/3 indices")
+        return result
+
+    def fetch_quotes(self, codes: list[str]) -> list[dict[str, Any]]:
+        # Eastmoney clist does not accept arbitrary code lists well; use ulist.np for batches.
+        secids = []
+        for code in codes:
+            ts = str(code or "").upper()
+            symbol = ts.split(".")[0]
+            if ts.endswith(".SH") or symbol.startswith(("5", "6", "9")):
+                secids.append(f"1.{symbol}")
+            else:
+                secids.append(f"0.{symbol}")
+        if not secids:
+            return []
+        payload = self._get_json(
+            EASTMONEY_INDEX_URL,
+            {
+                "secids": ",".join(secids[:60]),
+                "fltt": "2",
+                "invt": "2",
+                "fields": "f12,f14,f2,f3,f4,f15,f16,f17,f18,f5,f6,f8,f124",
+            },
+            referer="https://quote.eastmoney.com/",
+        )
+        rows = list((payload.get("data") or {}).get("diff") or [])
+        result = []
+        for row in rows:
+            symbol = str(row.get("f12") or "")
+            if not symbol:
+                continue
+            ts_code = f"{symbol}.SH" if symbol.startswith(("5", "6", "9")) else f"{symbol}.SZ"
+            epoch = int(finite_number(row.get("f124")) or 0)
+            result.append(
+                {
+                    "ts_code": ts_code,
+                    "name": row.get("f14") or symbol,
+                    "price": round4(finite_number(row.get("f2"))),
+                    "pct_chg": round4(finite_number(row.get("f3"))),
+                    "change_amount": round4(finite_number(row.get("f4"))),
+                    "open": round4(finite_number(row.get("f17"))),
+                    "high": round4(finite_number(row.get("f15"))),
+                    "low": round4(finite_number(row.get("f16"))),
+                    "previous_close": round4(finite_number(row.get("f18"))),
+                    "volume": round4(finite_number(row.get("f5"))),
+                    "amount": round4(finite_number(row.get("f6"))),
+                    "turnover_rate": round4(finite_number(row.get("f8"))),
+                    "quote_time_epoch": epoch,
+                    "quote_time": (
+                        datetime.fromtimestamp(epoch).astimezone().isoformat(timespec="seconds")
+                        if epoch
+                        else ""
+                    ),
+                    "source": "eastmoney_push2",
+                }
+            )
+        return result
+
+    def fetch_intraday(self, ts_code: str) -> dict[str, Any]:
+        code = str(ts_code or "").upper()
+        if code in INDEX_SECIDS:
+            secid = INDEX_SECIDS[code]
+            entity = "index"
+            identifier = code
+        else:
+            symbol = code.split(".")[0]
+            market = "1" if symbol.startswith(("5", "6", "9")) else "0"
+            secid = f"{market}.{symbol}"
+            entity = "stock"
+            identifier = symbol
+        payload = self._get_json(
+            TRENDS_URL,
+            {
+                "secid": secid,
+                "fields1": "f1,f2,f3,f4,f5,f6,f7,f8,f9,f10,f11,f12,f13",
+                "fields2": "f51,f52,f53,f54,f55,f56,f57,f58",
+                "iscr": "0",
+                "ndays": "1",
+            },
+            referer="https://quote.eastmoney.com/",
+        )
+        data = payload.get("data") or {}
+        points = []
+        for raw in data.get("trends") or []:
+            point = _parse_trend(raw)
+            if point:
+                points.append(point)
+        if not points:
+            raise AdapterError("No intraday chart data returned")
+        return {
+            "entity_type": entity,
+            "identifier": identifier,
+            "ts_code": code if "." in code else f"{identifier}.{'SH' if identifier.startswith(('5','6','9')) else 'SZ'}",
+            "name": str(data.get("name") or ""),
+            "code": str(data.get("code") or identifier),
+            "trade_date": points[-1]["date"],
+            "previous_close": round4(finite_number(data.get("preClose"))),
+            "points": points,
+            "source": "eastmoney_trends2",
+        }
+
+    def _get_json(self, url: str, params: dict[str, str], referer: str) -> dict[str, Any]:
+        request_url = f"{url}?{urllib.parse.urlencode(params)}"
+        request = urllib.request.Request(
+            request_url,
+            headers={
+                "Accept": "application/json,text/plain,*/*",
+                "User-Agent": BROWSER_UA,
+                "Referer": referer,
+            },
+            method="GET",
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=self.timeout) as response:
+                return json.loads(response.read().decode("utf-8"))
+        except Exception as exc:
+            raise AdapterError(f"eastmoney request failed: {exc}") from exc
+
+
+def _parse_trend(raw: Any) -> dict[str, Any] | None:
+    text = str(raw or "")
+    parts = text.split(",")
+    if len(parts) < 8:
+        return None
+    stamp = parts[0]
+    try:
+        when = datetime.strptime(stamp, "%Y-%m-%d %H:%M")
+    except ValueError:
+        return None
+    return {
+        "time": when.strftime("%H:%M"),
+        "date": when.strftime("%Y-%m-%d"),
+        "open": round4(finite_number(parts[1])),
+        "close": round4(finite_number(parts[2])),
+        "high": round4(finite_number(parts[3])),
+        "low": round4(finite_number(parts[4])),
+        "avg_price": round4(finite_number(parts[7] if len(parts) > 7 else parts[2])),
+        "volume": round4(finite_number(parts[5])),
+        "amount": round4(finite_number(parts[6])),
+    }

xiaobai-datahub/datahub/adapters/tencent.py

@@ -1,3 +1,99 @@
-from datahub.adapters.base import ReservedAdapter
+from __future__ import annotations
 
-ADAPTER = ReservedAdapter("tencent")
+import time
+import urllib.error
+import urllib.request
+from datetime import datetime
+from typing import Any
+
+from datahub.adapters.base import AdapterError, MarketAdapter
+from datahub.numbers import finite_number, round4
+
+TENCENT_INDEX_URL = "https://qt.gtimg.cn/q=sh000001,sz399001,sz399006"
+BROWSER_UA = (
+    "Mozilla/5.0 (Windows NT 10.0; Win64; x64) "
+    "AppleWebKit/537.36 (KHTML, like Gecko) Chrome/138.0.0.0 Safari/537.36"
+)
+
+
+class TencentAdapter(MarketAdapter):
+    name = "tencent"
+
+    def __init__(self, timeout: int = 8) -> None:
+        self.timeout = timeout
+
+    def probe(self) -> dict[str, Any]:
+        started = time.perf_counter()
+        try:
+            rows = self.fetch_indices()
+            state = "ok" if len(rows) == 3 else "empty"
+        except AdapterError as exc:
+            return {
+                "provider": self.name,
+                "configured": True,
+                "state": "error",
+                "message": str(exc),
+                "latency_ms": round((time.perf_counter() - started) * 1000),
+            }
+        return {
+            "provider": self.name,
+            "configured": True,
+            "state": state,
+            "latency_ms": round((time.perf_counter() - started) * 1000),
+        }
+
+    def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
+        if dataset in {"indexes_quotes", "index_quotes"}:
+            return self.fetch_indices()
+        raise AdapterError(f"{self.name} unsupported dataset: {dataset}")
+
+    def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        return list(rows)
+
+    def fetch_indices(self) -> list[dict[str, Any]]:
+        request = urllib.request.Request(
+            TENCENT_INDEX_URL,
+            headers={"User-Agent": BROWSER_UA, "Referer": "https://gu.qq.com/"},
+            method="GET",
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=self.timeout) as response:
+                raw = response.read().decode("gb18030", errors="ignore")
+        except Exception as exc:
+            raise AdapterError(f"tencent request failed: {exc}") from exc
+        result = []
+        for line in raw.splitlines():
+            if '="' not in line:
+                continue
+            fields = line.split('="', 1)[1].rsplit('";', 1)[0].split("~")
+            if len(fields) < 38:
+                continue
+            code = fields[2]
+            if code not in {"000001", "399001", "399006"}:
+                continue
+            try:
+                quote_time = datetime.strptime(fields[30], "%Y%m%d%H%M%S").astimezone()
+            except ValueError as exc:
+                raise AdapterError(f"Tencent invalid quote time for {code}") from exc
+            ts_code = "000001.SH" if code == "000001" else f"{code}.SZ"
+            result.append(
+                {
+                    "ts_code": ts_code,
+                    "code": code,
+                    "name": fields[1] or code,
+                    "price": round4(finite_number(fields[3])),
+                    "pct_chg": round4(finite_number(fields[32])),
+                    "change_amount": round4(finite_number(fields[31])),
+                    "open": round4(finite_number(fields[5])),
+                    "high": round4(finite_number(fields[33])),
+                    "low": round4(finite_number(fields[34])),
+                    "previous_close": round4(finite_number(fields[4])),
+                    "amount": round4(finite_number(fields[37]) * 10000),
+                    "quote_time_epoch": int(quote_time.timestamp()),
+                    "quote_time": quote_time.isoformat(timespec="seconds"),
+                    "source": "tencent_qt",
+                }
+            )
+        if len(result) != 3:
+            raise AdapterError(f"Tencent returned {len(result)}/3 indices")
+        return result

xiaobai-datahub/datahub/adapters/tushare.py

@@ -11,8 +11,12 @@ from datahub.normalize import (
     normalize_auction,
     normalize_calendar,
     normalize_daily,
+    normalize_dragon_tiger,
     normalize_index_daily,
+    normalize_limit_event,
     normalize_moneyflow,
+    normalize_popularity,
+    normalize_sector_daily,
     normalize_stock,
     normalize_valuation,
 )
@@ -31,6 +35,21 @@ TUSHARE_FIELDS = {
         "buy_lg_amount,sell_lg_amount,buy_elg_amount,sell_elg_amount,net_mf_amount"
     ),
     "stk_auction": "ts_code,trade_date,vol,price,amount,pre_close,turnover_rate,volume_ratio,float_share",
+    "limit_list_d": (
+        "trade_date,ts_code,industry,name,close,pct_chg,amount,limit_amount,"
+        "float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time,"
+        "open_times,up_stat,limit_times,limit_type"
+    ),
+    "ths_hot": "ts_code,ts_name,hot,rank,pct_change,current_price,concept,data_type,trade_date",
+    "dc_hot": "ts_code,ts_name,rank,pct_change,current_price,hot,concept,data_type,trade_date",
+    "hm_detail": "trade_date,ts_code,ts_name,buy_amount,sell_amount,net_amount,hm_name,hm_orgs,tag",
+    "hm_list": "name,desc,orgs",
+    "top_list": "trade_date,ts_code,name,pct_change,reason",
+    "top_inst": "trade_date,ts_code,exalter,buy,buy_rate,sell,sell_rate,net_buy,side,reason",
+    "ths_index": "ts_code,name,count,exchange,list_date,type",
+    "ths_daily": "ts_code,trade_date,open,high,low,close,pre_close,pct_change,vol,turnover_rate",
+    "dc_index": "ts_code,trade_date,name,open,high,low,close,pre_close,pct_change,vol,amount,turnover_rate",
+    "sw_daily": "ts_code,trade_date,name,open,high,low,close,pct_change,vol,amount",
 }
 
 DATASET_API = {
@@ -42,12 +61,15 @@ DATASET_API = {
     "index_daily": "index_daily",
     "moneyflow": "moneyflow",
     "auction": "stk_auction",
+    "limit_events": "limit_list_d",
+    "popularity": "ths_hot",
+    "dragon_tiger": "hm_detail",
+    "sector_daily": "ths_daily",
 }
 
-# Website actual index usage: market cards / 90-day charts (SH/SZ/CYB) plus
-# screener 沪深300 benchmark (lookback up to 260 trading days).
 WEBSITE_INDEX_CODES = ("000001.SH", "399001.SZ", "399006.SZ", "000300.SH")
 DEFAULT_INDEX_CODES = WEBSITE_INDEX_CODES
+LIMIT_TYPES = ("U", "D", "Z")
 
 
 class TushareAdapter(MarketAdapter):
@@ -85,6 +107,14 @@ class TushareAdapter(MarketAdapter):
         }
 
     def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
+        if dataset == "limit_events":
+            return self.fetch_limit_events(str(params.get("trade_date") or ""))
+        if dataset == "popularity":
+            return self.fetch_popularity(str(params.get("trade_date") or ""))
+        if dataset == "dragon_tiger":
+            return self.fetch_dragon_tiger(str(params.get("trade_date") or ""))
+        if dataset == "sector_daily":
+            return self.fetch_sector_daily(str(params.get("trade_date") or ""))
         api_name = DATASET_API.get(dataset, dataset)
         fields = TUSHARE_FIELDS.get(api_name, "")
         query_params = dict(params)
@@ -93,10 +123,67 @@ class TushareAdapter(MarketAdapter):
         if api_name == "trade_cal" and "exchange" not in query_params:
             query_params["exchange"] = "SSE"
         if api_name == "index_daily" and "ts_code" not in query_params:
-            # Caller typically loops codes; a missing code would pull nothing useful.
             query_params.setdefault("ts_code", DEFAULT_INDEX_CODES[0])
         return self._query(api_name, query_params, fields)
 
+    def fetch_limit_events(self, trade_date: str) -> list[dict[str, Any]]:
+        rows: list[dict[str, Any]] = []
+        for limit_type in LIMIT_TYPES:
+            part = self._query(
+                "limit_list_d",
+                {"trade_date": trade_date, "limit_type": limit_type},
+                TUSHARE_FIELDS["limit_list_d"],
+            )
+            for row in part:
+                row = dict(row)
+                row.setdefault("limit_type", limit_type)
+                rows.append(row)
+        return rows
+
+    def fetch_popularity(self, trade_date: str) -> list[dict[str, Any]]:
+        rows: list[dict[str, Any]] = []
+        for api_name, source in (("ths_hot", "ths"), ("dc_hot", "dc")):
+            for row in self._query(api_name, {"trade_date": trade_date}, TUSHARE_FIELDS[api_name]):
+                item = dict(row)
+                item["source"] = source
+                item.setdefault("trade_date", trade_date)
+                rows.append(item)
+        return rows
+
+    def fetch_dragon_tiger(self, trade_date: str) -> list[dict[str, Any]]:
+        details = self._query("hm_detail", {"trade_date": trade_date}, TUSHARE_FIELDS["hm_detail"])
+        top_rows = self._query("top_list", {"trade_date": trade_date}, TUSHARE_FIELDS["top_list"])
+        context = {
+            str(row.get("ts_code") or ""): row
+            for row in top_rows
+            if str(row.get("ts_code") or "")
+        }
+        rows: list[dict[str, Any]] = []
+        for row in details:
+            item = dict(row)
+            stock = context.get(str(item.get("ts_code") or ""), {})
+            if item.get("pct_change") is None and stock.get("pct_change") is not None:
+                item["pct_change"] = stock.get("pct_change")
+            if not item.get("reason") and stock.get("reason"):
+                item["reason"] = stock.get("reason")
+            if not item.get("ts_name") and stock.get("name"):
+                item["ts_name"] = stock.get("name")
+            rows.append(item)
+        return rows
+
+    def fetch_sector_daily(self, trade_date: str) -> list[dict[str, Any]]:
+        rows: list[dict[str, Any]] = []
+        for api_name, family in (("ths_daily", "ths"), ("dc_index", "dc"), ("sw_daily", "sw")):
+            try:
+                part = self._query(api_name, {"trade_date": trade_date}, TUSHARE_FIELDS[api_name])
+            except AdapterError:
+                part = []
+            for row in part:
+                item = dict(row)
+                item["family"] = family
+                rows.append(item)
+        return rows
+
     def fetch_index_daily(self, trade_date: str, codes: tuple[str, ...] = DEFAULT_INDEX_CODES) -> list[dict[str, Any]]:
         rows: list[dict[str, Any]] = []
         for ts_code in codes:
@@ -104,6 +191,17 @@ class TushareAdapter(MarketAdapter):
         return rows
 
     def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        if dataset in {"limit_events", "limit_list_d"}:
+            return [normalize_limit_event(row) for row in rows]
+        if dataset == "popularity":
+            return [normalize_popularity(row, source=str(row.get("source") or "")) for row in rows]
+        if dataset == "dragon_tiger":
+            return [normalize_dragon_tiger(row) for row in rows]
+        if dataset == "sector_daily":
+            return [
+                normalize_sector_daily(row, family=str(row.get("family") or "ths"))
+                for row in rows
+            ]
         mapping = {
             "calendar": normalize_calendar,
             "trade_cal": normalize_calendar,
@@ -148,12 +246,11 @@ class TushareAdapter(MarketAdapter):
         try:
             with urllib.request.urlopen(request, timeout=self.timeout) as response:
                 result = json.loads(response.read().decode("utf-8"))
-        except json.JSONDecodeError:
+        except (urllib.error.URLError, TimeoutError, json.JSONDecodeError) as exc:
-            raise AdapterError("Tushare returned invalid json") from None
+            raise AdapterError(f"Tushare 请求失败: {exc}") from exc
-        except (urllib.error.URLError, TimeoutError) as exc:
+        if result.get("code") not in (0, "0", None):
-            raise AdapterError(f"Tushare request failed: {exc}") from exc
+            raise AdapterError(str(result.get("msg") or f"Tushare error {result.get('code')}"))
-        if result.get("code") != 0:
-            raise AdapterError(result.get("msg") or "Tushare returned an unknown error")
         data = result.get("data") or {}
-        columns = data.get("fields") or []
+        items = data.get("items") or []
-        return [dict(zip(columns, item)) for item in data.get("items") or []]
+        fields_list = data.get("fields") or (fields.split(",") if fields else [])
+        return [dict(zip(fields_list, item)) for item in items]

xiaobai-datahub/datahub/admin_api.py

@@ -39,7 +39,6 @@ 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),
@@ -92,7 +91,6 @@ 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/datasets_ext.py

@@ -0,0 +1,180 @@
+"""Extended EOD datasets beyond the first-batch A/B release groups.
+
+These publish independently (soft): a failure here must not block daily/valuation
+release. Scheduler runs them after the core EOD window.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+# Independent soft datasets (not part of A/B atomic groups).
+EXTENDED_SOFT_DATASETS = {
+    "limit_events",
+    "popularity",
+    "dragon_tiger",
+    "sector_daily",
+}
+
+EXTENDED_SCHEMA = """
+CREATE TABLE IF NOT EXISTS eod_limit_events (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, limit_type TEXT NOT NULL,
+  name TEXT, industry TEXT, close REAL, pct_chg REAL, amount REAL,
+  limit_amount REAL, float_mv REAL, total_mv REAL, turnover_ratio REAL,
+  fd_amount REAL, first_time TEXT, last_time TEXT,
+  open_times INTEGER, up_stat TEXT, limit_times INTEGER,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, limit_type, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS staging_limit_events (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, limit_type TEXT NOT NULL, batch_id TEXT NOT NULL,
+  name TEXT, industry TEXT, close REAL, pct_chg REAL, amount REAL,
+  limit_amount REAL, float_mv REAL, total_mv REAL, turnover_ratio REAL,
+  fd_amount REAL, first_time TEXT, last_time TEXT,
+  open_times INTEGER, up_stat TEXT, limit_times INTEGER,
+  PRIMARY KEY (batch_id, ts_code, trade_date, limit_type)
+);
+
+CREATE TABLE IF NOT EXISTS eod_popularity (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, source TEXT NOT NULL,
+  ts_name TEXT, rank INTEGER, pct_change REAL, current_price REAL,
+  hot REAL, concept TEXT, data_type TEXT,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, source, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS staging_popularity (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, source TEXT NOT NULL, batch_id TEXT NOT NULL,
+  ts_name TEXT, rank INTEGER, pct_change REAL, current_price REAL,
+  hot REAL, concept TEXT, data_type TEXT,
+  PRIMARY KEY (batch_id, ts_code, trade_date, source)
+);
+
+CREATE TABLE IF NOT EXISTS eod_dragon_tiger (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, hm_name TEXT NOT NULL,
+  ts_name TEXT, buy_amount REAL, sell_amount REAL, net_amount REAL,
+  hm_orgs TEXT, tag TEXT, pct_change REAL, reason TEXT,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, hm_name, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS staging_dragon_tiger (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, hm_name TEXT NOT NULL, batch_id TEXT NOT NULL,
+  ts_name TEXT, buy_amount REAL, sell_amount REAL, net_amount REAL,
+  hm_orgs TEXT, tag TEXT, pct_change REAL, reason TEXT,
+  PRIMARY KEY (batch_id, ts_code, trade_date, hm_name)
+);
+
+CREATE TABLE IF NOT EXISTS eod_sector_daily (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, family TEXT NOT NULL,
+  name TEXT, open REAL, high REAL, low REAL, close REAL, pre_close REAL,
+  pct_change REAL, vol REAL, turnover_rate REAL, amount REAL,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, family, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS staging_sector_daily (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, family TEXT NOT NULL, batch_id TEXT NOT NULL,
+  name TEXT, open REAL, high REAL, low REAL, close REAL, pre_close REAL,
+  pct_change REAL, vol REAL, turnover_rate REAL, amount REAL,
+  PRIMARY KEY (batch_id, ts_code, trade_date, family)
+);
+
+CREATE TABLE IF NOT EXISTS sector_master (
+  ts_code TEXT PRIMARY KEY,
+  name TEXT,
+  family TEXT NOT NULL,
+  exchange TEXT,
+  list_date TEXT,
+  member_count INTEGER,
+  type TEXT,
+  updated_at TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_eod_limit_date ON eod_limit_events(trade_date, batch_id);
+CREATE INDEX IF NOT EXISTS idx_eod_pop_date ON eod_popularity(trade_date, batch_id);
+CREATE INDEX IF NOT EXISTS idx_eod_lhb_date ON eod_dragon_tiger(trade_date, batch_id);
+CREATE INDEX IF NOT EXISTS idx_eod_sector_date ON eod_sector_daily(trade_date, family, batch_id);
+"""
+
+EXTENDED_DATASET_TABLES = {
+    "limit_events": ("eod_limit_events", "staging_limit_events"),
+    "popularity": ("eod_popularity", "staging_popularity"),
+    "dragon_tiger": ("eod_dragon_tiger", "staging_dragon_tiger"),
+    "sector_daily": ("eod_sector_daily", "staging_sector_daily"),
+}
+
+EXTENDED_STAGING_INSERT: dict[str, tuple[str, Any]] = {
+    "limit_events": (
+        "INSERT INTO staging_limit_events("
+        "ts_code,trade_date,limit_type,batch_id,name,industry,close,pct_chg,amount,"
+        "limit_amount,float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time,"
+        "open_times,up_stat,limit_times) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], r["limit_type"], b,
+            r.get("name"), r.get("industry"), r.get("close"), r.get("pct_chg"), r.get("amount"),
+            r.get("limit_amount"), r.get("float_mv"), r.get("total_mv"), r.get("turnover_ratio"),
+            r.get("fd_amount"), r.get("first_time"), r.get("last_time"),
+            r.get("open_times"), r.get("up_stat"), r.get("limit_times"),
+        ),
+    ),
+    "popularity": (
+        "INSERT INTO staging_popularity("
+        "ts_code,trade_date,source,batch_id,ts_name,rank,pct_change,current_price,hot,concept,data_type) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], r["source"], b,
+            r.get("ts_name"), r.get("rank"), r.get("pct_change"), r.get("current_price"),
+            r.get("hot"), r.get("concept"), r.get("data_type"),
+        ),
+    ),
+    "dragon_tiger": (
+        "INSERT INTO staging_dragon_tiger("
+        "ts_code,trade_date,hm_name,batch_id,ts_name,buy_amount,sell_amount,net_amount,"
+        "hm_orgs,tag,pct_change,reason) VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], r["hm_name"], b,
+            r.get("ts_name"), r.get("buy_amount"), r.get("sell_amount"), r.get("net_amount"),
+            r.get("hm_orgs"), r.get("tag"), r.get("pct_change"), r.get("reason"),
+        ),
+    ),
+    "sector_daily": (
+        "INSERT INTO staging_sector_daily("
+        "ts_code,trade_date,family,batch_id,name,open,high,low,close,pre_close,"
+        "pct_change,vol,turnover_rate,amount) VALUES (?,?,?,?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], r["family"], b,
+            r.get("name"), r.get("open"), r.get("high"), r.get("low"), r.get("close"),
+            r.get("pre_close"), r.get("pct_change"), r.get("vol"), r.get("turnover_rate"),
+            r.get("amount"),
+        ),
+    ),
+}
+
+EXTENDED_EOD_COPY = {
+    "limit_events": (
+        "INSERT OR REPLACE INTO eod_limit_events "
+        "SELECT ts_code,trade_date,limit_type,name,industry,close,pct_chg,amount,"
+        "limit_amount,float_mv,total_mv,turnover_ratio,fd_amount,first_time,last_time,"
+        "open_times,up_stat,limit_times,batch_id "
+        "FROM staging_limit_events WHERE batch_id = ?"
+    ),
+    "popularity": (
+        "INSERT OR REPLACE INTO eod_popularity "
+        "SELECT ts_code,trade_date,source,ts_name,rank,pct_change,current_price,hot,concept,data_type,batch_id "
+        "FROM staging_popularity WHERE batch_id = ?"
+    ),
+    "dragon_tiger": (
+        "INSERT OR REPLACE INTO eod_dragon_tiger "
+        "SELECT ts_code,trade_date,hm_name,ts_name,buy_amount,sell_amount,net_amount,"
+        "hm_orgs,tag,pct_change,reason,batch_id "
+        "FROM staging_dragon_tiger WHERE batch_id = ?"
+    ),
+    "sector_daily": (
+        "INSERT OR REPLACE INTO eod_sector_daily "
+        "SELECT ts_code,trade_date,family,name,open,high,low,close,pre_close,"
+        "pct_change,vol,turnover_rate,amount,batch_id "
+        "FROM staging_sector_daily WHERE batch_id = ?"
+    ),
+}

xiaobai-datahub/datahub/db.py

@@ -7,9 +7,10 @@ from contextlib import contextmanager
 from pathlib import Path
 from typing import Any
 
+from datahub.datasets_ext import EXTENDED_DATASET_TABLES, EXTENDED_SCHEMA
 from datahub.timeutil import isoformat
 
-SCHEMA = """
+_BASE_SCHEMA = """
 CREATE TABLE IF NOT EXISTS schema_migrations (
   version INTEGER PRIMARY KEY,
   applied_at TEXT NOT NULL
@@ -238,19 +239,6 @@ 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,
@@ -295,6 +283,8 @@ CREATE INDEX IF NOT EXISTS idx_eod_bars_date ON eod_bars(trade_date, batch_id);
 CREATE INDEX IF NOT EXISTS idx_calendar_open ON trade_calendar(is_open, cal_date);
 """
 
+SCHEMA = _BASE_SCHEMA + EXTENDED_SCHEMA
+
 DATASET_TABLES = {
     "daily": ("eod_bars", "staging_bars"),
     "valuation": ("eod_valuation", "staging_valuation"),
@@ -302,6 +292,7 @@ DATASET_TABLES = {
     "auction": ("eod_auction", "staging_auction"),
     "index_daily": ("eod_index_bars", "staging_index_bars"),
     "stocks": ("eod_stocks", "staging_stocks"),
+    **EXTENDED_DATASET_TABLES,
 }
 
 

xiaobai-datahub/datahub/normalize.py

@@ -156,6 +156,95 @@ def normalize_stock(row: dict[str, Any]) -> dict[str, Any]:
     }
 
 
+def normalize_limit_event(row: dict[str, Any]) -> dict[str, Any]:
+    """limit_list_d. float_mv/total_mv/limit_amount are 万元 → yuan; amount/fd_amount already yuan."""
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "limit_type": str(row.get("limit_type") or "").strip().upper() or "U",
+        "name": str(row.get("name") or "").strip() or None,
+        "industry": str(row.get("industry") or "").strip() or None,
+        "close": round4(finite_number(row.get("close"))),
+        "pct_chg": round4(finite_number(row.get("pct_chg"))),
+        "amount": round4(finite_number(row.get("amount"))),
+        "limit_amount": round4(_scale(row.get("limit_amount"), AMOUNT_WAN_YUAN)),
+        "float_mv": round4(_scale(row.get("float_mv"), AMOUNT_WAN_YUAN)),
+        "total_mv": round4(_scale(row.get("total_mv"), AMOUNT_WAN_YUAN)),
+        "turnover_ratio": round4(finite_number(row.get("turnover_ratio"))),
+        "fd_amount": round4(finite_number(row.get("fd_amount"))),
+        "first_time": str(row.get("first_time") or "").strip() or None,
+        "last_time": str(row.get("last_time") or "").strip() or None,
+        "open_times": _optional_int(row.get("open_times")),
+        "up_stat": str(row.get("up_stat") or "").strip() or None,
+        "limit_times": _optional_int(row.get("limit_times")),
+    }
+
+
+def normalize_popularity(row: dict[str, Any], source: str = "") -> dict[str, Any]:
+    src = str(source or row.get("source") or "").strip().lower() or "ths"
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "source": src,
+        "ts_name": str(row.get("ts_name") or row.get("name") or "").strip() or None,
+        "rank": _optional_int(row.get("rank")),
+        "pct_change": round4(
+            finite_number(row.get("pct_change") if row.get("pct_change") is not None else row.get("pct_chg"))
+        ),
+        "current_price": round4(finite_number(row.get("current_price") or row.get("price"))),
+        "hot": round4(finite_number(row.get("hot"))),
+        "concept": str(row.get("concept") or "").strip() or None,
+        "data_type": str(row.get("data_type") or "").strip() or None,
+    }
+
+
+def normalize_dragon_tiger(row: dict[str, Any]) -> dict[str, Any]:
+    """hm_detail amounts are 万元 → yuan."""
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "hm_name": str(row.get("hm_name") or "未命名游资").strip() or "未命名游资",
+        "ts_name": str(row.get("ts_name") or row.get("name") or "").strip() or None,
+        "buy_amount": round4(_scale(row.get("buy_amount"), AMOUNT_WAN_YUAN)),
+        "sell_amount": round4(_scale(row.get("sell_amount"), AMOUNT_WAN_YUAN)),
+        "net_amount": round4(_scale(row.get("net_amount"), AMOUNT_WAN_YUAN)),
+        "hm_orgs": str(row.get("hm_orgs") or "").strip() or None,
+        "tag": str(row.get("tag") or "").strip() or None,
+        "pct_change": round4(finite_number(row.get("pct_change"))),
+        "reason": str(row.get("reason") or "").strip() or None,
+    }
+
+
+def normalize_sector_daily(row: dict[str, Any], family: str = "ths") -> dict[str, Any]:
+    fam = str(family or row.get("family") or "ths").strip().lower()
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "family": fam,
+        "name": str(row.get("name") or "").strip() or None,
+        "open": round4(finite_number(row.get("open"))),
+        "high": round4(finite_number(row.get("high"))),
+        "low": round4(finite_number(row.get("low"))),
+        "close": round4(finite_number(row.get("close"))),
+        "pre_close": round4(finite_number(row.get("pre_close"))),
+        "pct_change": round4(
+            finite_number(row.get("pct_change") if row.get("pct_change") is not None else row.get("pct_chg"))
+        ),
+        "vol": round4(finite_number(row.get("vol"))),
+        "turnover_rate": round4(finite_number(row.get("turnover_rate"))),
+        "amount": round4(finite_number(row.get("amount"))),
+    }
+
+
+def _optional_int(value: Any) -> int | None:
+    if value in (None, ""):
+        return None
+    try:
+        return int(float(value))
+    except (TypeError, ValueError):
+        return None
+
+
 def apply_qfq(price: float | None, factor: float | None, latest_factor: float | None) -> float | None:
     if price is None:
         return None
@@ -184,6 +273,11 @@ NORMALIZERS = {
     "calendar": normalize_calendar,
     "stock_basic": normalize_stock,
     "stocks": normalize_stock,
+    "limit_events": normalize_limit_event,
+    "limit_list_d": normalize_limit_event,
+    "popularity": normalize_popularity,
+    "dragon_tiger": normalize_dragon_tiger,
+    "sector_daily": normalize_sector_daily,
 }
 
 

xiaobai-datahub/datahub/pipeline.py

@@ -9,30 +9,33 @@ from typing import Any
 
 from datahub.adapters.base import AdapterError
 from datahub.adapters.tushare import DEFAULT_INDEX_CODES, WEBSITE_INDEX_CODES, TushareAdapter
+from datahub.datasets_ext import (
+    EXTENDED_EOD_COPY,
+    EXTENDED_SOFT_DATASETS,
+    EXTENDED_STAGING_INSERT,
+)
 from datahub.db import DATASET_TABLES, HubDB
 from datahub.governance.circuit import CircuitBreaker
 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
 
 LOGGER = get_logger()
 
 HARD_DATASETS = {"daily", "valuation", "index_daily"}
-SOFT_DATASETS = {"moneyflow", "auction"}
+SOFT_DATASETS = {"moneyflow", "auction"} | EXTENDED_SOFT_DATASETS
-OFFICIAL_DATASETS = HARD_DATASETS | SOFT_DATASETS
+OFFICIAL_DATASETS = HARD_DATASETS | {"moneyflow", "auction"}  # A/B retry scope unchanged
 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",)
+EOD_C_DATASETS = ("limit_events",)
+EOD_D_DATASETS = ("dragon_tiger",)
+EOD_E_DATASETS = ("sector_daily",)
+EOD_F_DATASETS = ("popularity",)
 EMPTY_BATCH_ERROR = "empty official batch: 0 valid rows"
 
 STAGING_INSERT = {
@@ -86,6 +89,7 @@ STAGING_INSERT = {
             r.get("close"), r.get("pct_chg"), r.get("volume"), r.get("amount"),
         ),
     ),
+    **EXTENDED_STAGING_INSERT,
 }
 
 EOD_COPY = {
@@ -120,6 +124,7 @@ EOD_COPY = {
         "SELECT ts_code,trade_date,open,high,low,close,pct_chg,volume,amount,batch_id "
         "FROM staging_index_bars WHERE batch_id = ?"
     ),
+    **EXTENDED_EOD_COPY,
 }
 
 
@@ -678,255 +683,71 @@ class Pipeline:
     def run_eod_batch_b(self, trade_date: str, force: bool = False) -> dict[str, Any]:
         return self.run_release_group(EOD_B_DATASETS, trade_date, force=force)
 
+    def run_extended_soft(self, datasets: tuple[str, ...], trade_date: str, force: bool = False) -> dict[str, Any]:
+        """Publish extended soft datasets independently (not A/B atomic)."""
+        results: dict[str, Any] = {}
+        day = yyyymmdd(trade_date)
+        for dataset in datasets:
+            if not force and self.active_batch(dataset, day):
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "skipped",
+                    "reason": "already_published",
+                }
+                continue
+            try:
+                rows = self._fetch_dataset(dataset, day)
+                if not rows and dataset in {"popularity", "dragon_tiger"}:
+                    results[dataset] = {
+                        "dataset": dataset,
+                        "trade_date": day,
+                        "state": "skipped",
+                        "reason": "upstream_empty",
+                        "rows": 0,
+                    }
+                    continue
+                results[dataset] = self.run_dataset(dataset, day, prepared_rows=rows)
+            except Exception as exc:
+                results[dataset] = {
+                    "dataset": dataset,
+                    "trade_date": day,
+                    "state": "failed",
+                    "error": str(exc),
+                }
+                LOGGER.exception("extended soft publish failed dataset=%s date=%s", dataset, day)
+        return results
+
+    def run_eod_batch_c(self, trade_date: str, force: bool = False) -> dict[str, Any]:
+        return self.run_extended_soft(EOD_C_DATASETS, trade_date, force=force)
+
+    def run_eod_batch_d(self, trade_date: str, force: bool = False) -> dict[str, Any]:
+        return self.run_extended_soft(EOD_D_DATASETS, trade_date, force=force)
+
+    def run_eod_batch_e(self, trade_date: str, force: bool = False) -> dict[str, Any]:
+        return self.run_extended_soft(EOD_E_DATASETS, trade_date, force=force)
+
+    def run_eod_batch_f(self, trade_date: str, force: bool = False) -> dict[str, Any]:
+        return self.run_extended_soft(EOD_F_DATASETS, trade_date, force=force)
+
     def force_republish_boundary(self, dataset: str, trade_date: str) -> dict[str, Any]:
         """Force-republish the full A/B consistency boundary that owns ``dataset``.
 
         CLI ``eod-refresh --force`` and admin manual backfill must not publish a
         single official member alone — that would mix old and new batches inside
         the same trade date. Naming any A-group member (or stocks) rebuilds the
-        whole A group; naming ``index_daily`` rebuilds B.
+        whole A group; naming ``index_daily`` rebuilds B. Extended soft datasets
+        republish independently.
         """
         name = str(dataset or "").strip()
         if name in EOD_A_DATASETS or name == STOCKS_DATASET:
             return self.run_eod_batch_a(trade_date, force=True)
         if name in EOD_B_DATASETS:
             return self.run_eod_batch_b(trade_date, force=True)
+        if name in EXTENDED_SOFT_DATASETS:
+            return self.run_extended_soft((name,), 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, ...],
@@ -1262,7 +1083,16 @@ class Pipeline:
         )
         listed_n = int((listed or {}).get("n") or 0)
         row_n = len(rows)
-        keys = [(row.get("ts_code"), row.get("trade_date")) for row in rows]
+        if dataset == "limit_events":
+            keys = [(row.get("ts_code"), row.get("trade_date"), row.get("limit_type")) for row in rows]
+        elif dataset == "popularity":
+            keys = [(row.get("ts_code"), row.get("trade_date"), row.get("source")) for row in rows]
+        elif dataset == "dragon_tiger":
+            keys = [(row.get("ts_code"), row.get("trade_date"), row.get("hm_name")) for row in rows]
+        elif dataset == "sector_daily":
+            keys = [(row.get("ts_code"), row.get("trade_date"), row.get("family")) for row in rows]
+        else:
+            keys = [(row.get("ts_code"), row.get("trade_date")) for row in rows]
         dup = row_n - len(set(keys))
         if dup:
             errors.append(f"duplicate keys: {dup}")
@@ -1283,7 +1113,8 @@ class Pipeline:
             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)
+            allow_empty = dataset in {"popularity", "dragon_tiger", "moneyflow", "auction"}
+            hard_fail = bool(dup or bad_date or (empty and not allow_empty))
         else:
             hard_fail = bool(errors) and (dataset in HARD_DATASETS or dataset == STOCKS_DATASET)
         report = {

xiaobai-datahub/datahub/realtime_serve.py

@@ -0,0 +1,188 @@
+"""Provisional (盘中观察) serving: quotes, index quotes, intraday points.
+
+Free sources only. Never writes official eod_* tables. Uses rt_cache + LKG.
+"""
+
+from __future__ import annotations
+
+import json
+import time
+from datetime import datetime
+from typing import Any
+
+from datahub.adapters.eastmoney import EastmoneyAdapter
+from datahub.adapters.tencent import TencentAdapter
+from datahub.codes import resolve_code
+from datahub.db import HubDB
+from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
+
+QUOTE_TTL = 60
+INDEX_TTL = 60
+INTRADAY_TTL = 20
+
+
+class RealtimeApiError(RuntimeError):
+    def __init__(self, code: str, message: str) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+
+
+def _envelope(data: Any, meta: dict[str, Any]) -> dict[str, Any]:
+    from datahub import SCHEMA_VERSION
+
+    return {"schema_version": SCHEMA_VERSION, "data": data, "meta": meta}
+
+
+def fetch_index_quotes(db: HubDB) -> dict[str, Any]:
+    cache_key = "indexes:quotes"
+    cached = _read_cache(db, cache_key)
+    if cached is not None:
+        return cached
+    eastmoney = EastmoneyAdapter()
+    try:
+        rows = eastmoney.fetch_indices()
+        source = "eastmoney:ulist"
+    except Exception:
+        rows = TencentAdapter().fetch_indices()
+        source = "tencent:qt"
+    if len(rows) < 3:
+        raise RealtimeApiError("SOURCE_UNAVAILABLE", "index quotes incomplete")
+    payload = _envelope(
+        rows,
+        {
+            "tier": "provisional",
+            "trade_date": yyyymmdd(now_shanghai()),
+            "source": source,
+            "stale": False,
+            "staleness_seconds": 0,
+            "published_at": isoformat(now_shanghai()),
+        },
+    )
+    _write_cache(db, cache_key, payload, INDEX_TTL, source)
+    return payload
+
+
+def fetch_quotes(db: HubDB, codes: list[str]) -> dict[str, Any]:
+    if not codes:
+        raise RealtimeApiError("INVALID_ARGUMENT", "codes is required")
+    resolved: list[str] = []
+    for code in codes[:60]:
+        item = resolve_code(db, code) or _guess_ts_code(code)
+        if item:
+            resolved.append(item)
+    if not resolved:
+        raise RealtimeApiError("INVALID_ARGUMENT", "no resolvable codes")
+    cache_key = "quotes:" + ",".join(sorted(resolved))
+    cached = _read_cache(db, cache_key)
+    if cached is not None:
+        return cached
+    adapter = EastmoneyAdapter()
+    try:
+        rows = adapter.fetch_quotes(resolved)
+        source = "eastmoney:clist"
+    except Exception as exc:
+        raise RealtimeApiError("SOURCE_UNAVAILABLE", f"quotes unavailable: {exc}") from exc
+    payload = _envelope(
+        rows,
+        {
+            "tier": "provisional",
+            "trade_date": yyyymmdd(now_shanghai()),
+            "source": source,
+            "stale": False,
+            "staleness_seconds": 0,
+            "published_at": isoformat(now_shanghai()),
+        },
+    )
+    _write_cache(db, cache_key, payload, QUOTE_TTL, source)
+    return payload
+
+
+def fetch_intraday(db: HubDB, code: str, date: str = "") -> dict[str, Any]:
+    ts_code = resolve_code(db, code) or _guess_ts_code(code)
+    if not ts_code:
+        raise RealtimeApiError("INVALID_ARGUMENT", f"ambiguous code: {code}")
+    cache_key = f"intraday:{ts_code}:{date or 'today'}"
+    cached = _read_cache(db, cache_key)
+    if cached is not None:
+        return cached
+    adapter = EastmoneyAdapter()
+    try:
+        payload_data = adapter.fetch_intraday(ts_code)
+        source = "eastmoney:trends2"
+    except Exception as exc:
+        raise RealtimeApiError("SOURCE_UNAVAILABLE", f"intraday unavailable: {exc}") from exc
+    payload = _envelope(
+        payload_data,
+        {
+            "tier": "provisional",
+            "trade_date": yyyymmdd(payload_data.get("trade_date") or date or now_shanghai()),
+            "source": source,
+            "stale": False,
+            "staleness_seconds": 0,
+            "published_at": isoformat(now_shanghai()),
+        },
+    )
+    _write_cache(db, cache_key, payload, INTRADAY_TTL, source)
+    return payload
+
+
+def _guess_ts_code(code: str) -> str | None:
+    raw = str(code or "").strip().upper()
+    if "." in raw:
+        return raw
+    if len(raw) == 6 and raw.isdigit():
+        if raw.startswith(("5", "6", "9")):
+            return f"{raw}.SH"
+        return f"{raw}.SZ"
+    return None
+
+
+def _read_cache(db: HubDB, cache_key: str) -> dict[str, Any] | None:
+    row = db.fetchone("SELECT * FROM rt_cache WHERE cache_key = ?", (cache_key,))
+    if not row:
+        return None
+    expires = str(row.get("expires_at") or "")
+    now = isoformat(now_shanghai())
+    if expires and expires < now:
+        return None
+    try:
+        payload = json.loads(row["payload"])
+    except json.JSONDecodeError:
+        return None
+    if isinstance(payload, dict) and isinstance(payload.get("meta"), dict):
+        stored = str(row.get("stored_at") or "")
+        try:
+            age = max(0, int(time.time() - datetime.fromisoformat(stored).timestamp()))
+        except Exception:
+            age = 0
+        payload["meta"]["staleness_seconds"] = age
+        payload["meta"]["stale"] = age > QUOTE_TTL
+    return payload
+
+
+def _write_cache(db: HubDB, cache_key: str, payload: dict[str, Any], ttl: int, source: str) -> None:
+    from datetime import timedelta
+
+    now = now_shanghai()
+    stored = isoformat(now)
+    expires = isoformat(now + timedelta(seconds=ttl))
+    db.execute(
+        """
+        INSERT INTO rt_cache(cache_key, payload, source, stored_at, expires_at)
+        VALUES (?,?,?,?,?)
+        ON CONFLICT(cache_key) DO UPDATE SET
+          payload=excluded.payload, source=excluded.source,
+          stored_at=excluded.stored_at, expires_at=excluded.expires_at
+        """,
+        (cache_key, json.dumps(payload, ensure_ascii=False), source, stored, expires),
+    )
+    db.execute(
+        """
+        INSERT INTO last_known_good(cache_key, payload, source, stored_at)
+        VALUES (?,?,?,?)
+        ON CONFLICT(cache_key) DO UPDATE SET
+          payload=excluded.payload, source=excluded.source, stored_at=excluded.stored_at
+        """,
+        (cache_key, json.dumps(payload, ensure_ascii=False), source, stored),
+    )

xiaobai-datahub/datahub/revision.py

@@ -1,111 +0,0 @@
-"""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,6 +1,5 @@
 from __future__ import annotations
 
-import json
 import threading
 from collections.abc import Callable
 from datetime import datetime, time, timedelta
@@ -9,14 +8,13 @@ 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_revise"}
+EOD_JOB_IDS = {"eod_a", "eod_b", "eod_retry"}
 
 
 def is_open_day(db: HubDB, day: str) -> bool:
@@ -29,20 +27,6 @@ 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()
 
@@ -64,8 +48,11 @@ class Scheduler:
             "precheck": self._precheck,
             "eod_a": self._eod_a,
             "eod_b": self._eod_b,
+            "eod_c": self._eod_c,
+            "eod_d": self._eod_d,
+            "eod_e": self._eod_e,
+            "eod_f": self._eod_f,
             "eod_retry": self._eod_retry,
-            "eod_revise": self._eod_revise,
             "stocks_refresh": self._stocks_refresh,
             "cleanup": self._cleanup,
             "backup": self._backup,
@@ -104,6 +91,10 @@ class Scheduler:
             ("precheck", time(8, 45)),
             ("eod_a", time(15, 5)),
             ("eod_b", time(15, 10)),
+            ("eod_c", time(16, 40)),
+            ("eod_d", time(16, 45)),
+            ("eod_e", time(18, 5)),
+            ("eod_f", time(22, 40)),
             ("cleanup", time(0, 30)),
             ("backup", time(0, 40)),
         ]
@@ -116,7 +107,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", "stocks_refresh"} and not open_day:
+            if job_id in {"eod_a", "eod_b", "eod_c", "eod_d", "eod_e", "eod_f", "stocks_refresh"} and not open_day:
                 self._fired.add(key)
                 continue
             self._fired.add(key)
@@ -127,7 +118,7 @@ class Scheduler:
             try:
                 self.run_job(job_id, day)
             except Exception:
-                if job_id not in {"eod_a", "eod_b", "stocks_refresh"}:
+                if job_id not in {"eod_a", "eod_b", "eod_c", "eod_d", "eod_e", "eod_f", "stocks_refresh"}:
                     raise
                 # Keep the tick alive; evening retries take over.
                 LOGGER.exception("scheduled job %s failed for %s", job_id, day)
@@ -135,8 +126,6 @@ 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
 
     # ------------------------------------------------------------------
@@ -238,201 +227,6 @@ 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
@@ -531,12 +325,21 @@ class Scheduler:
     def _eod_b(self, trade_date: str) -> dict[str, Any]:
         return self.pipeline.run_eod_batch_b(trade_date)
 
+    def _eod_c(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_batch_c(trade_date)
+
+    def _eod_d(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_batch_d(trade_date)
+
+    def _eod_e(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_batch_e(trade_date)
+
+    def _eod_f(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_batch_f(trade_date)
+
     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/serving.py

@@ -73,6 +73,20 @@ class V1API:
             return self.moneyflow(q)
         if path == "/v1/auction":
             return self.auction(q)
+        if path == "/v1/limit-events":
+            return self.limit_events(q)
+        if path == "/v1/popularity":
+            return self.popularity(q)
+        if path == "/v1/dragon-tiger":
+            return self.dragon_tiger(q)
+        if path == "/v1/sectors":
+            return self.sectors(q)
+        if path == "/v1/quotes/latest":
+            return self.quotes_latest(q)
+        if path == "/v1/indexes/quotes":
+            return self.index_quotes(q)
+        if path == "/v1/intraday/points":
+            return self.intraday_points(q)
         if path == "/v1/datasets/status":
             return self.dataset_status(q.get("date") or "")
         if path == "/v1/batches":
@@ -197,9 +211,75 @@ class V1API:
     def auction(self, q: dict[str, str]) -> dict[str, Any]:
         return self._published_rows(dataset="auction", table="eod_auction", q=q, source="tushare:stk_auction")
 
+    def limit_events(self, q: dict[str, str]) -> dict[str, Any]:
+        return self._published_rows(
+            dataset="limit_events",
+            table="eod_limit_events",
+            q=q,
+            source="tushare:limit_list_d",
+            extra_filters={"limit_type": q.get("limit_type") or ""},
+        )
+
+    def popularity(self, q: dict[str, str]) -> dict[str, Any]:
+        return self._published_rows(
+            dataset="popularity",
+            table="eod_popularity",
+            q=q,
+            source="tushare:ths_hot+dc_hot",
+            extra_filters={"source": q.get("source") or ""},
+        )
+
+    def dragon_tiger(self, q: dict[str, str]) -> dict[str, Any]:
+        return self._published_rows(
+            dataset="dragon_tiger",
+            table="eod_dragon_tiger",
+            q=q,
+            source="tushare:hm_detail",
+        )
+
+    def sectors(self, q: dict[str, str]) -> dict[str, Any]:
+        return self._published_rows(
+            dataset="sector_daily",
+            table="eod_sector_daily",
+            q=q,
+            source="tushare:ths_daily+dc_index+sw_daily",
+            extra_filters={"family": q.get("family") or ""},
+        )
+
+    def quotes_latest(self, q: dict[str, str]) -> dict[str, Any]:
+        from datahub.realtime_serve import RealtimeApiError, fetch_quotes
+
+        codes = [item.strip() for item in str(q.get("codes") or "").split(",") if item.strip()]
+        try:
+            return fetch_quotes(self.db, codes)
+        except RealtimeApiError as exc:
+            raise ApiError(exc.code, exc.message) from exc
+
+    def index_quotes(self, q: dict[str, str]) -> dict[str, Any]:
+        from datahub.realtime_serve import RealtimeApiError, fetch_index_quotes
+
+        try:
+            return fetch_index_quotes(self.db)
+        except RealtimeApiError as exc:
+            raise ApiError(exc.code, exc.message) from exc
+
+    def intraday_points(self, q: dict[str, str]) -> dict[str, Any]:
+        from datahub.realtime_serve import RealtimeApiError, fetch_intraday
+
+        code = str(q.get("code") or "").strip()
+        if not code:
+            raise ApiError("INVALID_ARGUMENT", "code is required")
+        try:
+            return fetch_intraday(self.db, code, yyyymmdd(q.get("date") or ""))
+        except RealtimeApiError as exc:
+            raise ApiError(exc.code, exc.message) from exc
+
     def dataset_status(self, date: str) -> dict[str, Any]:
         trade_date = yyyymmdd(date or now_shanghai())
-        datasets = ("daily", "valuation", "moneyflow", "auction", "index_daily", "stocks")
+        datasets = (
+            "daily", "valuation", "moneyflow", "auction", "index_daily", "stocks",
+            "limit_events", "popularity", "dragon_tiger", "sector_daily",
+        )
         items = []
         for dataset in datasets:
             pub = self.db.fetchone(
@@ -244,6 +324,7 @@ class V1API:
         source: str,
         adjust: str = "none",
         default_code: str = "",
+        extra_filters: dict[str, str] | None = None,
     ) -> dict[str, Any]:
         trade_date = q.get("date") or q.get("trade_date") or ""
         code = q.get("code") or default_code
@@ -264,6 +345,7 @@ class V1API:
             if resolved is None:
                 raise ApiError("INVALID_ARGUMENT", f"ambiguous code: {code}")
             ts_code = resolved
+        filters = {key: value for key, value in (extra_filters or {}).items() if value}
         # For a range, use per-date published batch. Single-date is the common path.
         if start == end:
             pub = self.db.fetchone(
@@ -282,6 +364,9 @@ class V1API:
             if ts_code:
                 sql += " AND ts_code = ?"
                 params.append(ts_code)
+            for key, value in filters.items():
+                sql += f" AND {key} = ?"
+                params.append(value)
             sql += " ORDER BY ts_code LIMIT ? OFFSET ?"
             params.extend([limit, offset])
             rows = [dict(row) for row in self.db.fetchall(sql, tuple(params))]
@@ -317,6 +402,9 @@ class V1API:
             if ts_code:
                 sql += " AND ts_code = ?"
                 params.append(ts_code)
+            for key, value in filters.items():
+                sql += f" AND {key} = ?"
+                params.append(value)
             sql += " ORDER BY ts_code"
             rows.extend(self.db.fetchall(sql, tuple(params)))
         sliced = rows[offset: offset + limit]

xiaobai-datahub/datahub/settings.py

@@ -79,20 +79,6 @@ 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/fixtures.py

@@ -45,6 +45,30 @@ RAW = {
         {"ts_code": "600000.SH", "trade_date": "20240902", "vol": 100, "price": 10.15, "amount": 1500000, "pre_close": 10.00, "turnover_rate": 0.1, "volume_ratio": 1.2, "float_share": 2000},
         {"ts_code": "000001.SZ", "trade_date": "20240902", "vol": 80, "price": 11.05, "amount": 1200000, "pre_close": 11.10, "turnover_rate": 0.2, "volume_ratio": 0.9, "float_share": 1800},
     ],
+    "limit_list_d": [
+        {"trade_date": "20240902", "ts_code": "600000.SH", "industry": "银行", "name": "浦发银行", "close": 10.2, "pct_chg": 9.95, "amount": 1e8, "limit_amount": 5000, "float_mv": 800, "total_mv": 1000, "turnover_ratio": 5.0, "fd_amount": 2e7, "first_time": "09:30:01", "last_time": "14:55:00", "open_times": 0, "up_stat": "1/1", "limit_times": 1, "limit_type": "U"},
+    ],
+    "ths_hot": [
+        {"ts_code": "600000.SH", "ts_name": "浦发银行", "hot": 90.0, "rank": 1, "pct_change": 1.2, "current_price": 10.2, "concept": "银行", "data_type": "热股", "trade_date": "20240902"},
+    ],
+    "dc_hot": [
+        {"ts_code": "600000.SH", "ts_name": "浦发银行", "rank": 2, "pct_change": 1.2, "current_price": 10.2, "hot": 80.0, "concept": "银行", "data_type": "A股市场", "trade_date": "20240902"},
+    ],
+    "hm_detail": [
+        {"trade_date": "20240902", "ts_code": "600000.SH", "ts_name": "浦发银行", "buy_amount": 1000, "sell_amount": 200, "net_amount": 800, "hm_name": "测试游资", "hm_orgs": "某某营业部", "tag": "超买"},
+    ],
+    "top_list": [
+        {"trade_date": "20240902", "ts_code": "600000.SH", "name": "浦发银行", "pct_change": 9.95, "reason": "涨幅偏离值达7%"},
+    ],
+    "ths_daily": [
+        {"ts_code": "885811.TI", "trade_date": "20240902", "open": 1000, "high": 1010, "low": 990, "close": 1005, "pre_close": 995, "pct_change": 1.0, "vol": 100, "turnover_rate": 1.2},
+    ],
+    "dc_index": [
+        {"ts_code": "BK0475", "trade_date": "20240902", "name": "银行", "open": 100, "high": 101, "low": 99, "close": 100.5, "pre_close": 99.5, "pct_change": 1.0, "vol": 10, "amount": 1e8, "turnover_rate": 0.5},
+    ],
+    "sw_daily": [
+        {"ts_code": "801780.SI", "trade_date": "20240902", "name": "银行", "open": 2000, "high": 2010, "low": 1990, "close": 2005, "pct_change": 0.8, "vol": 50, "amount": 2e8},
+    ],
 }
 
 
@@ -66,4 +90,9 @@ def fake_transport(api_name: str, params: dict, fields: str):
         start = str(params.get("start_date") or "")
         end = str(params.get("end_date") or "99999999")
         return [row for row in RAW["trade_cal"] if start <= row["cal_date"] <= end]
-    return list(RAW.get(api_name) or [])
+    rows = list(RAW.get(api_name) or [])
+    if api_name == "limit_list_d":
+        limit_type = str(params.get("limit_type") or "")
+        if limit_type:
+            rows = [row for row in rows if str(row.get("limit_type") or "") == limit_type]
+    return rows

xiaobai-datahub/tests/test_eod_retry.py

@@ -111,23 +111,14 @@ class EodRetryTests(unittest.TestCase):
         self.assertEqual(progress["state"], "done")
         self.assertEqual(progress["attempts"], 4)  # eod_a + eod_b + 2 retries
 
-        # success stops further eod_retry; revision window has not started yet
+        # success stops all further same-day requests
         batches_before = len(self._batches(db, day))
         eod_calls_before = len(self._eod_calls(transport))
         sched.tick(clock_at(day, 17, 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.assertEqual(len(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"
@@ -183,7 +174,6 @@ 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(
@@ -191,9 +181,7 @@ class EodRetryTests(unittest.TestCase):
             active_map,
         )
         self.assertEqual(set(official_batches()), batches_before)  # no duplicate batches
-        extra = self._eod_calls(transport)[len(calls_before):]
+        self.assertEqual(self._eod_calls(transport), calls_before)  # no duplicate upstream EOD calls
-        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:
@@ -218,7 +206,6 @@ 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,)))
@@ -239,18 +226,12 @@ 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_extended_eod.py

@@ -0,0 +1,65 @@
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from datahub.adapters.tushare import TushareAdapter
+from datahub.crypto import SecretVault
+from datahub.hub import Hub
+from datahub.settings import Settings
+from tests.fixtures import TRADE_DATE, fake_transport
+
+
+class ExtendedEodTests(unittest.TestCase):
+    def setUp(self) -> None:
+        self.tmp = tempfile.TemporaryDirectory()
+        key = SecretVault.generate_key()
+        settings = Settings(
+            host="127.0.0.1",
+            port=0,
+            encryption_key=key,
+            api_token="k" * 32,
+            admin_password="StartPass1",
+            tushare_token="tushare-secret",
+            db_path=Path(self.tmp.name) / "hub.db",
+            backup_dir=Path(self.tmp.name) / "backups",
+            scheduler_enabled=False,
+            quality={"daily_row_ratio": 0.5, "null_rate_max": 0.5, "list_limit_default": 5000, "list_limit_max": 5000},
+        )
+        adapter = TushareAdapter("tushare-secret", transport=fake_transport)
+        self.hub = Hub(settings, adapter=adapter)
+        self.hub.pipeline.ingest_reference(TRADE_DATE)
+        for dataset in ("daily", "valuation", "moneyflow", "auction", "index_daily"):
+            self.hub.pipeline.run_dataset(dataset, TRADE_DATE)
+
+    def tearDown(self) -> None:
+        self.hub.stop()
+        self.tmp.cleanup()
+
+    def test_extended_soft_datasets_publish_and_serve(self) -> None:
+        results = self.hub.pipeline.run_extended_soft(
+            ("limit_events", "popularity", "dragon_tiger", "sector_daily"),
+            TRADE_DATE,
+        )
+        for name in ("limit_events", "popularity", "dragon_tiger", "sector_daily"):
+            self.assertEqual(results[name]["state"], "published", results[name])
+        api = self.hub.api
+        limits = api.handle("/v1/limit-events", {"date": [TRADE_DATE]})
+        self.assertGreaterEqual(len(limits["data"]), 1)
+        self.assertEqual(limits["meta"]["tier"], "official")
+        pop = api.handle("/v1/popularity", {"date": [TRADE_DATE], "source": ["ths"]})
+        self.assertEqual(pop["data"][0]["source"], "ths")
+        lhb = api.handle("/v1/dragon-tiger", {"date": [TRADE_DATE]})
+        self.assertEqual(lhb["data"][0]["hm_name"], "测试游资")
+        # hub stores 万元→元
+        self.assertEqual(lhb["data"][0]["buy_amount"], 10_000_000.0)
+        sectors = api.handle("/v1/sectors", {"date": [TRADE_DATE], "family": ["ths"]})
+        self.assertEqual(sectors["data"][0]["family"], "ths")
+        status = api.handle("/v1/datasets/status", {"date": [TRADE_DATE]})
+        names = {item["dataset"] for item in status["data"]}
+        self.assertTrue({"limit_events", "popularity", "dragon_tiger", "sector_daily"} <= names)
+
+
+if __name__ == "__main__":
+    unittest.main()

xiaobai-datahub/tests/test_layout.py

@@ -19,11 +19,17 @@ class LayoutTests(unittest.TestCase):
     def test_reserved_adapters_present(self) -> None:
         from datahub.adapters import RESERVED
 
-        for name in ("eastmoney", "tencent", "ths", "xgb", "akshare", "ifind"):
+        for name in ("ths", "xgb", "akshare", "ifind"):
             self.assertIn(name, RESERVED)
             probe = RESERVED[name].probe()
             self.assertEqual(probe["state"], "reserved")
             self.assertFalse(probe["configured"])
+        for name in ("eastmoney", "tencent"):
+            self.assertIn(name, RESERVED)
+            probe = RESERVED[name].probe()
+            # Live free adapters: probe may be ok/error/empty depending on network.
+            self.assertIn(probe["state"], {"ok", "empty", "error"})
+            self.assertTrue(probe["configured"])
 
 
 if __name__ == "__main__":

xiaobai-datahub/tests/test_revision_review.py

@@ -1,333 +0,0 @@
-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()