feat(HEL-421): 回补历史日历和指数并标记区间不完整

Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>
fix(HEL-417): 配置根日志让影子对比报告落入容器日志
2026-09-02 22:16:05 +08:00 · 2026-09-02 21:53:19 +08:00 · 2026-09-02 18:08:40 +08:00 · 2026-09-02 17:09:28 +08:00 · 2026-09-02 15:56:29 +08:00 · 2026-09-02 12:26:09 +08:00
115 changed files with 7290 additions and 362 deletions
@@ -5,6 +5,11 @@ APP_ENCRYPTION_KEY=
 # the system settings; all accounts use the same backend market snapshot.
 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.
+DATAHUB_BASE_URL=http://127.0.0.1:8766
+DATAHUB_TOKEN=
+
 # Optional iFinD HTTP credential. The backend exchanges it for a short-lived
 # access token and never exposes either token to browsers.
 IFIND_REFRESH_TOKEN=your_ifind_refresh_token_here
@@ -8,6 +8,9 @@ data/*.db
 data/*.db-shm
 data/*.db-wal
 data/backups/
+datahub-data/
+xiaobai-datahub/data/
+xiaobai-datahub/.venv/
 data/*.bak
 data/*.backup
 *.log
@@ -54,7 +54,9 @@ background scheduler
  feature repository mixins; do not add feature queries to it.
 - `backend/jobs/` owns job definitions, locks, retries, idempotency, and persisted run state.
  `backend/jobs/service.py` is the application-facing owner of scheduler start/stop, manual
-  refresh submission, and periodic refresh coordination.
+  refresh submission, and periodic refresh coordination. `backend/jobs/refresh.py` owns
+  whether a dashboard payload is a usable refresh result versus a failed job, and whether
+  after-hours official catch-up is due.
 - `backend/llm/` owns model selection, membership/quota checks, fallback, provider transport,
  streaming rules, and call audit. Feature agents only prepare messages and interpret
  feature-specific results.
@@ -1,252 +1,46 @@
 # 小白复盘

-> 一个给 A 股股民自己用的**盘后复盘工作台**。
-> 收盘以后打开它，把「今天市场到底发生了什么」按情绪、梯队、轮动、竞价、龙虎榜一条条看明白，再决定明天激进还是保守。
+面向 A 股盘后复盘的本地 Web 工作台。收盘后把涨停、炸板、连板梯队、板块轮动、集合竞价、龙虎榜等数据整理成可浏览的复盘界面；不接券商、不代为下单，也不提供个股推荐。

-**一句话定位**：小白复盘是一套可以自己部署、自己掌控数据的 A 股盘后复盘工具。它把散落在各个行情软件里的涨停、炸板、连板梯队、板块轮动、集合竞价、龙虎榜数据，整理成一套**看得懂、能追问、可记录**的复盘界面。
+本目录是唯一正式源码。模块边界见 [ARCHITECTURE.md](ARCHITECTURE.md)，产品与维护文档见 [docs/README.md](docs/README.md)。

-它**不是**行情交易软件，**不接券商、不代为下单**，也不提供任何形式的个股推荐。
+## 主要功能

---
+登录后左侧共 16 个页面，另有一个内嵌页「策略持续跟踪」。交易日快照保存在本地 SQLite 数据库 `data/review.db`。

-## 一、为什么值得试（项目亮点）
+- **情绪周期**：0–100 情绪温度与阶段判定（默认首页）
+- **涨停池 / 炸板池 / 跌停板 / 昨日涨停 / 涨停表现**：封板结构、炸板、跌停与昨日反馈
+- **市场天梯**：按连板高度排列的市场梯队
+- **板块轮动**：近若干交易日板块热力与成分下钻
+- **集合竞价**：盘前生命周期；9:30 后停止更新并冻结为复盘归档。当前数据源提供 9:25 最终竞价快照，不是动态虚拟撮合行情
+- **题材库 / 人气热榜 / 龙虎榜**：题材成分、双榜人气、席位与游资档案
+- **智能选股**（会员）：六阶段策略、精选策略库、自然语言编译为受控公式后的确定性筛选与滚动回测；候选需手动加入后才进入五交易日跟踪
+- **问师**（会员）：按选定的游资思维 Skill 单师对话；新增公开角色时在 `游资skills` 下增加含 `SKILL.md` 的目录，并在 `游资skills/mentor_catalog.json` 登记。管理员私有角色放在 `data/private-mentor-skills`（不进 Git / 镜像）
+- **问天**（会员）：观势 / 观气 / 观心。卦象、干支、节气与气机由本地程序确定性计算，大模型只负责文字解释。此前仅冻结过界面视觉方案，现已解冻；问天可纳入后续数据与功能迁移，本阶段不主动重做视觉。
+- **我的复盘**：手工交易日志、每日复盘、提醒中心与复盘助手；不接券商、不自动下单

-1. **先看情绪，再看个股**
-   用一个 0–100 的「情绪温度」，把当天市场的冷热程度和它处在哪个阶段（冰点 → 修复 → 发酵 → 高潮 → 分化 → 退潮）直接摆在第一屏。先判断环境，再谈个股。
+全局能力：日间 / 夜间主题、股票代码悬停预览日 K 与分时、`Ctrl + K` 全局搜索。图表数据不写入主行情，也不参与情绪、选股或问天计算。

-2. **说的就是短线玩家的话**
-   晋级、断板、连板高度、梯队完整度、封单额、炸板……这些词不用解释，界面上原样呈现，不用再自己在行情软件里一层层翻。
+## 技术栈

-3. **不会写公式，也能按自己的想法筛股票**
-   可以直接写一句人话（比如「连续两天放量上涨且不是 ST」），系统会把它编译成受控公式再执行筛选。技术活儿交给程序，判断权始终在你手上。
+| 层面 | 说明 |
+| --- | --- |
+| 运行时 | Python 3.12；标准库 `ThreadingHTTPServer`，无独立 Web 框架 |
+| 依赖 | `requirements.txt` 仅含 `cryptography`；问天历法使用仓库内 `vendor/lunar_python` |
+| 数据库 | SQLite（WAL），默认文件 `data/review.db` |
+| 前端 | 原生 HTML / CSS / JavaScript，无打包、无构建步骤 |
+| 部署 | Docker / Docker Compose，或本机直接运行 `server.py` |
+| 安全 | 账号密码 scrypt 哈希；行情 Token 与模型密钥用 `APP_ENCRYPTION_KEY` 加密后存库 |

-4. **有问题可以「问」**
-   - **问师**：挑一位游资思维模型，系统自动把当天情绪、梯队、龙虎榜等数据喂给它，单师对话，边看数据边追问。
-   - **问天**：一个偏传统文化视角的模块，分「观势 / 观气 / 观心」三部分，看盘之外也给自己留一段安静时间。
+## 环境要求

-5. **复盘能留下来，而不是第二天就忘**
-   交易日志可以记录当时的买卖逻辑和情绪标签（平静 / 笃定 / 犹豫 / 焦虑 / 冲动），配合每日复盘三问和提醒中心，帮自己看清是不是在情绪化操作。
+- Python 3.12（与 `Dockerfile` 一致）
+- 本机启动：能执行 `python` / `pip`
+- Docker 部署：Docker Engine 24+，Compose v2（`docker compose`）
+- 行情：部署者自行申请并配置 Tushare Pro Token；部分分时优先使用同花顺 iFinD（可选）
+- 问师、问天解释、复盘助手、自然语言编译公式：需配置 OpenAI 兼容接口；未配置时市场数据页仍可用

-6. **数据在自己手里**
-   所有数据存在部署机器本地的 SQLite 数据库里，账号密码用 scrypt 哈希保存，行情 Token 和模型密钥加密存放。没有云端账号，不上传个人交易记录。
-
-7. **一套界面，白天晚上都能看**
-   内置日间 / 夜间两套配色，股票代码上悬停就弹出日 K 与分时小图，`Ctrl + K` 可以全局搜索股票、板块和题材。
-
---
-
-## 二、主要功能
-
-界面左侧一共有 **16 个页面**，另有一个内嵌页面（策略持续跟踪）。下面按分组说明。
-
-### 2.1 市场数据（登录即可用）
-
-| 页面 | 解决什么问题 | 打开后能看到什么 |
-| --- | --- | --- |
-| **情绪周期** | 收盘后说不清市场整体冷热、处在什么阶段 | 0–100 情绪温度、升温 / 降温方向、六阶段判定、由五个维度（市场宽度 20、涨停生态 25、赚钱效应 30、连板结构 15、成交活跃度 10）加权算出 |
-| **涨停池** | 当天封板的票散在各处，看不出封板结构 | 按首板 / 2 板 / 3 板+ 筛选，含首封时间、开板次数、封单额、涨停原因 |
-| **炸板池** | 触板没封住容易被误当成涨停 | 距涨停还差多少、开了几次板、炸板原因 |
-| **跌停板** | 亏钱效应集中在哪不直观 | 连续跌停天数、风险线索、风险行业聚集提示 |
-| **昨日涨停** | 昨天追板的人今天怎么样 | 晋级 / 炸板 / 跌停 / 断板四类结果，加一条「兑现率」（收红口径）摘要 |
-| **涨停表现** | 梯队到底健不健康 | 按连板高度看晋级率、收红率、平均涨幅，并给出当日结论 |
-| **市场天梯** | 连板梯队靠手翻拼不出来 | 从首板逐层排到当天最高板（至少展示 5 层），配市场高度与梯队完整度指标 |
-| **板块轮动** | 热点是延续还是一日游 | 最近 9 个交易日、Top 12 板块的热力矩阵，点板块可跨日高亮看连续性，并可下钻到申万二级成分股 |
-| **集合竞价** | 9:25 的竞价信息太散 | 四个页签：重点异动 / 我的自选 / 全部候选 / 竞价一字；关注分满分 100，可按「超预期 / 符合预期 / 低于预期」筛选；9:30 后停止更新，冻结归档供复盘 |
-| **题材库** | 题材排行和成分股要开好几个网站 | 左边题材排行、右边成分股，题材上悬停直接出日 K / 分时预览 |
-| **人气热榜** | 各家榜单各看各的 | 同花顺 + 东方财富双榜汇总，可切单榜或看「双榜共识」 |
-| **龙虎榜** | 上榜明细和游资动向查起来费劲 | 每日明细、净买入 / 净卖出筛选、活跃游资，点游资可看当日操作，游资档案带席位归类 |
-
-### 2.2 智能功能（会员可用）
-
-| 页面 | 解决什么问题 | 打开后能看到什么 |
-| --- | --- | --- |
-| **智能选股** | 自己写条件门槛高，盘中手翻太累 | 三种模式：① 六阶段策略，当日行情更新后（15:10 之后）自动计算；② **29 套精选策略库**，每套都标注适用环境、失效风险与准入条件；③ 自定义选股，自然语言写条件 → 编译成受控公式 → 确定性筛选执行 → 滚动回测 |
-| **策略持续跟踪** | 选出来的票加完自选就忘了为什么选 | 只有手动加入的候选才进入跟踪，展示 T+1 开盘 / 收盘、T+3、T+5、最大涨幅与最大回撤，到期自动提醒 |
-| **问师** | 复盘时没人讨论、视角单一 | 选一位游资思维模型单师对话，系统按模型流派自动喂情绪、梯队、龙虎榜等对应数据，回答为流式输出并带动态追问 |
-| **问天** | 想换个视角看市场，也想让自己静下来 | **观势**：用市场数据生成三才六爻，用来观察「势」；**观气**：依干支、精确节气、五运六气观察「运」；**观心**：不看盘、不输问题，先准备 1 秒，接着做 5 轮「吸气 3 秒、停顿 2 秒、呼气 4 秒」，再以六次三枚铜钱起卦完成一次问心仪式 |
-| **我的复盘** | 交易散养、不复盘、情绪化操作 | 结构化交易日志（方向、价格、仓位、盈亏、逻辑、执行、情绪、标签）、每日复盘三问、提醒中心、复盘助手（读取你的日志与市场数据，给出分析和条件化计划） |
-
-> 说明：卦象、干支、节气与气机关系全部由本地程序确定性计算，大模型只负责文字解释，不参与起卦，也不会改动结果。
-
-### 2.3 全局能力
-
- 股票 / 板块 / 题材上悬停，弹出日 K 与分时快速预览
- `Ctrl + K` 全局搜索
- 日间 / 夜间双主题一键切换
- 提醒中心集中查看策略跟踪与手工提醒
- 登录页有一组会动的小 K 线小人（红涨绿跌）
-
---
-
-## 三、特色页面展示
-
-下面 25 张图全部是深色主题下的真实页面。普通行情只作带过，重点放在智能选股、问师和问天。
-
-### 3.1 市场一览
-
-情绪周期总览：0–100 温度、当前阶段与五维构成。
-
-![情绪周期总览](docs/images/market-one-glance.png)
-
-市场天梯、板块轮动、龙虎榜三拼带过。
-
-![天梯、板块轮动与龙虎榜](docs/images/market-tools-strip.png)
-
-### 3.2 智能选股
-
-29 套精选策略库。
-
-![精选策略库](docs/images/screener-strategies.png)
-
-单套策略详情：适用环境、失效风险、准入条件与评分权重。
-
-![策略详情](docs/images/screener-strategy-detail.png)
-
-六阶段策略：阶段识别 → 策略匹配 → 执行选股。
-
-![六阶段策略](docs/images/screener-stage.png)
-
-自然语言编译出的受控公式特写。
-
-![受控公式特写](docs/images/screener-dsl.png)
-
-自定义选股结果与回测摘要。
-
-![选股结果与回测](docs/images/screener-results.png)
-
-策略全流程动效。
-
-![策略全流程](docs/images/screener-strategy-flow.gif)
-
-策略持续跟踪：T+1 开盘 / 收盘、后续回看与回撤。
-
-![策略持续跟踪](docs/images/screener-tracking.png)
-
-### 3.3 问师
-
-游资思维模型列表。
-
-![问师模型列表](docs/images/mentor-list.png)
-
-结合当日盘面的流式回答。
-
-![问师回答](docs/images/mentor-answer.gif)
-
-回答尾部的话题追问与输入区。
-
-![问师追问](docs/images/mentor-followup.png)
-
-### 3.4 问天
-
-观势 / 观气 / 观心三观总览。
-
-![问天三观](docs/images/heaven-three-views.png)
-
-观势完整结果：卦象、势值、六爻与行情对应。
-
-![问天观势结果](docs/images/heaven-shi-result.png)
-
-观势卦象展现过程。
-
-![问天观势卦象](docs/images/heaven-shi-hexagram.gif)
-
-观势解势载入动画。
-
-![问天观势载入](docs/images/heaven-shi-loading.gif)
-
-观气主界面：干支、节气与气机关系。
-
-![问天观气](docs/images/heaven-qi.png)
-
-三层气机与客主加临特写。
-
-![问天观气特写](docs/images/heaven-qi-detail.png)
-
-观气解运动画。
-
-![问天观气解运](docs/images/heaven-qi-fortune.gif)
-
-观心入口：五步流程。
-
-![问天观心入口](docs/images/heaven-xin-intro.png)
-
-观心呼吸引导。
-
-![问天观心呼吸](docs/images/heaven-xin-breathing.gif)
-
-观心六次摇钱起卦。
-
-![问天观心起卦](docs/images/heaven-xin-coin.gif)
-
-### 3.5 我的复盘与全局
-
-交易日志与情绪五标签（平静 / 笃定 / 犹豫 / 焦虑 / 冲动）。
-
-![交易日志情绪标签](docs/images/review-journal.png)
-
-复盘助手读取日志与市场数据后流式输出。
-
-![复盘助手](docs/images/review-assistant.gif)
-
-`Ctrl + K` 全局搜索，以及代码悬停弹出的日 K 预览。
-
-![全局搜索与日K预览](docs/images/global-search.png)
-
---
-
-## 四、适合谁用
-
- **每天收盘后不知道从哪看起的散户**：先给一个情绪温度和阶段，再往下拆。
- **做短线、看连板梯队的人**：晋级、断板、高度、完整度都在一个页面里。
- **想按自己的条件筛股票、但不会写公式的人**：说人话就行。
- **知道自己「管不住手」的人**：交易日志 + 情绪标签 + 每日复盘三问，把情绪化操作摆到自己面前。
- **在意数据隐私、想把工具放在自己机器上的人**：私有部署，数据不出本机。
- **对传统文化视角感兴趣的玩家**：问天的观势 / 观气 / 观心提供了另一种看市场的方式。
-
-**不太适合**：想要一键自动下单的人（本项目不接券商）；想要「跟着买就能赚」的人（本项目不提供任何形式的投资建议）。
-
---
-
-## 五、一天怎么用（典型流程）
-
-1. **收盘后 10 分钟：定环境**
-   打开「情绪周期」，看温度多少、在哪个阶段、是升温还是降温。先定下明天的基调是激进还是保守。
-
-2. **看结构：谁在涨、涨到几板**
-   进「市场天梯」看连板梯队和市场高度，再到「涨停池 / 炸板池 / 跌停板」看当天的封板结构与亏钱效应。
-
-3. **看延续：热点有没有搬家**
-   进「板块轮动」，点几个强势板块，看它们在最近 9 天是连续走强还是只有一天行情。
-
-4. **看结果：昨天追板的人今天怎么样**
-   进「昨日涨停」和「涨停表现」，对照梯队健康度，检验自己的判断。
-
-5. **查人气与席位**
-   「人气热榜」看双榜共识，「龙虎榜」看活跃游资当天做了什么。
-
-6. **按自己的条件选一批候选**
-   进「智能选股」，要么用六阶段策略或 29 套精选策略，要么直接写一句人话让它编译成公式。选出来的候选**手动**加入「策略持续跟踪」，后面 T+1 / T+3 / T+5 会自动提醒你回看。
-
-7. **有疑问就问**
-   「问师」按游资思维模型边看数据边追问；想静一静就去「问天」做一次观心。
-
-8. **记下来**
-   在「我的复盘」里写当天交易日志（含情绪标签）和复盘三问。第二天开盘前，先回看昨天的记录。
-
---
-
-## 六、部署与访问
-
-> 本项目为**私有部署的个人工具**，没有官方在线版，也不提供公共账号。
-
- **部署方式**：仓库内提供 `Dockerfile` 与 `compose.yaml`，在已有 Docker 环境的机器上即可启动；也可直接用 Python 运行 `server.py` 在本机启动。
- **访问方式**：部署完成后通过浏览器访问该机器上的服务端口（默认 8765），**具体地址由部署者自己决定**。出于安全考虑，本文档不写任何内网 IP、账号、密码或密钥。
- **数据存放**：所有行情快照、复盘记录与账号数据都存在部署机器本地的 SQLite 数据库文件中（`data/review.db`），随 `data/` 目录一起备份即可。
- **账号**：首次使用先注册，第一个注册的账号自动成为管理员，之后注册的默认为普通用户。
- **数据来源**：行情数据来自 Tushare Pro 等第三方数据接口（部分分时数据优先使用同花顺 iFinD），**需要部署者自己申请并配置 Token**；部分接口的可用性取决于数据源方的授权与积分权限，个别数据缺失时页面会明确提示，不会用估算值顶替。
- **大模型功能**：问师、问天解释、复盘助手、自然语言编译公式等需要配置一个 OpenAI 兼容的接口，由部署者自行配置与管理额度；未配置时这些对话与解释功能不可用，市场数据部分不受影响。
-
-### 本机启动
+## 安装与启动

 仓库根目录即为运行目录（`server.py`、`requirements.txt` 都在根目录）。

@@ -255,94 +49,90 @@ python -m pip install -r requirements.txt
 python server.py
 ```

-默认监听本机回环地址、端口 `8765`（仅本机浏览器可访问）。首次使用先注册账号。主行情不再回退演示数据：盘前、非交易日或临时取数失败时沿用最近真实收盘快照；没有任何真实快照时，页面会提示等待管理员完成首次同步。
+默认监听 `127.0.0.1:8765`（仅本机可访问）。浏览器打开该地址，首次使用先注册账号；第一个账号自动成为管理员，之后注册的默认为普通用户。

-Windows 下若需要后台启动，并把日志、进程号和 Python 缓存统一写入 `runtime/`（不在源码根目录产生运行文件），可使用：
+主行情不再回退演示数据：盘前、非交易日或临时取数失败时沿用最近真实收盘快照；没有任何真实快照时，页面会提示等待管理员完成首次同步。
+
+可选参数：
+
+```bash
+python server.py --host 127.0.0.1 --port 8765
+```
+
+Windows 下若需要后台启动，并把日志、进程号和 Python 缓存写入 `runtime/`（不在源码根目录产生运行文件）：

 ```powershell
 powershell -ExecutionPolicy Bypass -File tools/start_local.ps1
 ```

-### Docker 部署
+该脚本默认端口为 `8797`。统一验收：

-局域网或服务器部署使用仓库根目录的 `Dockerfile` 与 `compose.yaml`。完整的迁移、持久化、防火墙、备份和恢复步骤见 [DOCKER_DEPLOY.md](DOCKER_DEPLOY.md)。
+```bash
+python tools/verify_baseline.py
+```

-### 系统与账号配置
+涉及运行时或前端时再加 `--e2e`（Playwright）。

-管理员通过页面右上角「系统管理」保存公共 Tushare Token、平台主/辅助模型、会员每日额度和后台刷新开关。所有用户读取同一份 SQLite 行情快照，不再分别配置行情 Token。
+## Docker 使用

-`.env` 中的 Tushare 和平台 LLM 配置只用于初始化系统配置，密钥不会返回到浏览器。请将 `.env` 与数据库一起备份，丢失 `APP_ENCRYPTION_KEY` 后无法恢复加密资料。
+局域网或服务器部署使用仓库根目录的 `Dockerfile` 与 `compose.yaml`。容器监听 `8765`，默认以非 root 用户运行，并把宿主机 `./data` 挂到容器内 `/app/data`。

-普通用户在「账号设置」中维护个人资料、查看会员状态和修改密码，不配置个人 LLM。有效会员自动使用平台模型；管理员可在「系统管理」中手动开通、续期、停用会员。平台模型受管理员设置的每日调用次数限制，管理员账号始终可用。
+1. 复制 `.env.example` 为 `.env`，填入 `APP_ENCRYPTION_KEY` 以及行情 / 模型等初始化配置。密钥不会返回到浏览器。
+2. `compose.yaml` 构建时要求带上当前 Git 提交号，避免打出无版本标签的镜像：

-新增公开问师角色时，在 `游资skills` 下增加一个包含 `SKILL.md` 的独立目录，并在 `游资skills/mentor_catalog.json` 中登记。管理员私有角色放在 `data/private-mentor-skills`，该目录不进入 Git 或 Docker 镜像，且只会出现在管理员的问师列表中。
+```bash
+export XIAOBAI_GIT_REV="$(git rev-parse HEAD)"
+export XIAOBAI_GIT_SHORT="$(git rev-parse --short=7 HEAD)"
+docker compose build
+docker compose up -d
+```

-### 相关文档
+3. 检查健康接口：

- [ARCHITECTURE.md](ARCHITECTURE.md) — 模块边界与源码职责
+```bash
+docker compose ps
+curl http://127.0.0.1:8765/api/health
+```
+
+健康响应类似 `{"ok": true, "storage": "sqlite", "account_required": true}`。
+
+完整的迁移、持久化、防火墙、备份、恢复与正式线构建入口见 [DOCKER_DEPLOY.md](DOCKER_DEPLOY.md)。`.env` 必须与数据库成对备份；丢失 `APP_ENCRYPTION_KEY` 后无法恢复加密资料。
+
+## 目录与配置
+
+```text
+server.py            进程入口
+backend/            服务、路由、数据、任务、LLM
+frontend/            无构建前端（shared + pages）
+config/              页面 / API / 任务等注册表
+data/                SQLite 与私有数据（数据库文件不进 Git）
+runtime/             本地日志、PID、缓存（不进 Git）
+tools/               启动、验收与构建辅助脚本
+游资skills/          公开问师角色
+vendor/              本地第三方库（含 lunar-python）
+Dockerfile
+compose.yaml
+.env.example         环境变量模板（复制为 .env 后填写）
+```
+
+管理员通过页面右上角「系统管理」保存公共 Tushare Token、平台主/辅助模型、会员每日额度和后台刷新开关。所有用户读取同一份 SQLite 行情快照。`.env` 中的 Tushare 和平台 LLM 配置只用于初始化系统配置。
+
+普通用户在「账号设置」中维护个人资料、查看会员状态和修改密码，不配置个人 LLM。有效会员使用平台模型；管理员可开通、续期、停用会员。平台模型受每日调用次数限制，管理员账号始终可用。
+
+相关文档：
+
+- [ARCHITECTURE.md](ARCHITECTURE.md) — 模块边界
 - [docs/README.md](docs/README.md) — 交接手册入口
 - [DOCKER_DEPLOY.md](DOCKER_DEPLOY.md) — Docker 部署、备份与恢复
 - [THIRD_PARTY_NOTICES.md](THIRD_PARTY_NOTICES.md) — 第三方授权（含问天历法库）
+- [AGENTS.md](AGENTS.md) — 维护约束

-**技术组成（简要）**
+## 注意事项与免责声明

-| 层面 | 说明 |
-| --- | --- |
-| 后端 | Python 3.12，仅用标准库自带的 HTTP 服务与自研路由，第三方依赖极少 |
-| 数据库 | SQLite（WAL 模式），单文件，易备份 |
-| 前端 | 原生 JavaScript + HTML，无构建工具、无打包步骤，改完刷新即可 |
-| 部署 | Docker / Docker Compose，或本机直接运行 |
-| 安全 | 账号密码 scrypt 哈希存储；行情 Token 与模型密钥使用 `APP_ENCRYPTION_KEY` 加密后存库，密钥与数据库需一起备份 |
-
---
-
-## 七、免责声明
-
- 本项目是一个**个人研究与复盘工具**，所提供的全部数据、指标、候选与文字分析，**均不构成投资建议、证券推荐或买卖要约**。
- 本项目**不接券商、不代为下单**，交易日志只做手工记录与统计，不代表任何实际成交。
- 情绪温度、阶段判定、连板梯队、策略筛选等均为**基于公开数据的统计与规则计算结果**，不预测未来走势，不保证任何收益。历史统计与回测结果不代表未来表现。
- 「问天」模块属于**传统文化视角的娱乐化观察**，卦象、干支、节气与气机关系由本地程序确定性生成，用于换个角度看市场和自我觉察，**不具备预测功能，不得作为投资依据**。
- 行情数据来自第三方接口，可能存在延迟、缺失或口径调整；页面会在数据不可用时明确提示，请以交易所与券商正式披露的数据为准。
+- 本项目是个人研究与复盘工具，全部数据、指标、候选与文字分析均不构成投资建议、证券推荐或买卖要约。
+- 不接券商、不代为下单。交易日志只做手工记录与统计，不代表实际成交。
+- 情绪温度、阶段判定、连板梯队、策略筛选等均为基于公开数据的统计与规则计算，不预测走势，不保证收益。
+- 「问天」属于传统文化视角的观察工具，不具备预测功能，不得作为投资依据。问天不是永久冻结区：此前只冻结过界面视觉方案，现已解冻，后续数据与功能迁移可以纳入。
+- 行情来自第三方接口，可能延迟、缺失或口径调整；不可用时页面会明确提示，请以交易所与券商正式披露为准。
+- 不要把服务端口直接暴露到公网。不要把 Token、密码、密钥、数据库或 `.env` 提交进 Git。
 - 股市有风险，入市需谨慎。投资决策及其后果由使用者本人承担。
-
---
-
-## 附：精简版项目推荐语
-
-> 下面三段可以单独整段复制转发，不需要带图也能看懂。
-
-### 一句话版（约 60 字，适合评论区、群聊快速安利）
-
-> 小白复盘：一个自己部署的 A 股盘后复盘工作台——情绪、梯队、轮动、龙虎榜一页页看明白；说人话就能选股，不荐股、不下单，数据只存自己机器。
-
-### 短版（约 320 字，适合社交平台、群聊、朋友圈）
-
-> 发现一个挺特别的 A 股盘后复盘工具，叫「小白复盘」。
->
-> 它不荐股、也不下单，就专心做一件事：把收盘后散落在各个软件里的信息整理成人看得懂的样子。打开先给你一个 0–100 的情绪温度和「冰点→修复→发酵→高潮→分化→退潮」六个阶段，让你先知道今天是什么环境；然后连板梯队、涨停炸板、板块轮动的 9 天热力图、集合竞价、龙虎榜游资动向，一页一页往下看。
->
-> 最惊喜的是不会写公式也能选股——直接写一句人话，它帮你编译成条件去筛。还有「问师」可以按游资思维跟你对话，「问天」用卦象换个角度看市场、也让你静一静。
->
-> 数据存在自己机器上的本地数据库里，私有部署，不上传。
->
-> 提醒一句：它只是个把信息摆清楚的工具，不荐股也不下单，不构成投资建议。
-
-### 稍长版（约 580 字，适合发论坛、公众号、仓库简介）
-
-> **小白复盘 —— 给 A 股股民自己的盘后复盘工作台**
->
-> 收盘以后，你是怎么复盘的？在好几个软件之间来回切，还是干脆不复盘？
->
-> 小白复盘是一套可以自己部署的 A 股盘后复盘工具。它把涨停、炸板、跌停、连板梯队、板块轮动、集合竞价、龙虎榜这些散落各处的数据，整理成 16 个看得懂的页面：
->
-> - **情绪周期**：0–100 情绪温度 + 六阶段判定，由市场宽度、涨停生态、赚钱效应、连板结构、成交活跃度五个维度加权算出，先看环境再看个股。
-> - **市场天梯**：从首板一路排到当天最高板，配市场高度与梯队完整度。
-> - **板块轮动**：近 9 个交易日 Top 12 板块热力矩阵，点一下就能看出热点是延续还是一日游。
-> - **集合竞价**：关注分满分 100，9:30 后自动冻结归档，专门留给复盘看。
-> - **龙虎榜**：席位明细、活跃游资、游资档案打通，点游资就能看它当天做了什么。
-> - **智能选股**：29 套精选策略（每套都写了适用环境和失效风险），也可以直接写一句人话，让它编译成公式去筛选。
-> - **问师 / 问天 / 我的复盘**：一个陪你按游资思维聊数据，一个用传统卦象换个角度看市场，一个帮你记下每笔交易的逻辑和情绪。
->
-> 它不接券商、不代为下单，也不给出任何个股推荐——只负责把信息摆清楚，判断始终在你自己手里。所有数据存在你自己机器的本地 SQLite 里，私有部署，不上传。
->
-> 项目私有部署、无公共在线版；文中所有功能描述均对应现有实现，不构成投资建议，市场有风险。
@@ -1,11 +1,23 @@
 from __future__ import annotations

 import argparse
+import logging
 from http.server import ThreadingHTTPServer
 from typing import Any


+def configure_logging() -> None:
+    """让 INFO 级结构化日志（含 datahub 影子对比报告）落到容器日志。"""
+    if logging.getLogger().handlers:
+        return
+    logging.basicConfig(
+        level=logging.INFO,
+        format="%(asctime)s %(levelname)s %(name)s %(message)s",
+    )
+
+
 def main(handler_class: type[Any] | None = None, service: Any | None = None) -> None:
+    configure_logging()
    if handler_class is None or service is None:
        from backend.application import RequestHandler, SERVICE

@@ -11,6 +11,8 @@ from backend.features.accounts.security import SecretVault
 def environment_credentials(environment: Mapping[str, str]) -> dict[str, str]:
    return {
        "tushare_token": str(environment.get("TUSHARE_TOKEN") or "").strip(),
+        "datahub_token": str(environment.get("DATAHUB_TOKEN") or "").strip(),
+        "datahub_base_url": str(environment.get("DATAHUB_BASE_URL") or "").strip(),
        "ifind_refresh_token": str(environment.get("IFIND_REFRESH_TOKEN") or "").strip(),
        "ifind_access_token": str(environment.get("IFIND_ACCESS_TOKEN") or "").strip(),
        "platform_llm_primary_api_key": str(
@@ -0,0 +1,15 @@
+from backend.data.datahub.bridge import DatahubAwareTushareClient, DatahubBridge
+from backend.data.datahub.client import DatahubClient, DatahubResponse
+from backend.data.datahub.errors import DatahubError
+from backend.data.datahub.settings import DATASETS, DatahubSettings, DatasetFlags
+
+__all__ = [
+    "DATASETS",
+    "DatahubAwareTushareClient",
+    "DatahubBridge",
+    "DatahubClient",
+    "DatahubError",
+    "DatahubResponse",
+    "DatahubSettings",
+    "DatasetFlags",
+]
@@ -0,0 +1,255 @@
+from __future__ import annotations
+
+import logging
+import sys
+from typing import Any, Callable
+
+from backend.data.datahub.client import DatahubClient, DatahubResponse
+from backend.data.datahub.compare import compare_rows
+from backend.data.datahub.errors import DatahubError
+from backend.data.datahub.native import (
+    API_TO_DATASET,
+    filter_calendar_rows,
+    filter_stock_rows,
+    project_fields,
+    to_native_rows,
+    yyyymmdd,
+)
+from backend.data.datahub.redact import redact_text, redact_value
+from backend.data.datahub.settings import DatahubSettings
+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"}
+
+
+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
+
+
+def caller_is_heaven(depth: int = 24) -> bool:
+    frame = sys._getframe(1)
+    for _ in range(depth):
+        frame = frame.f_back if frame is not None else None
+        if frame is None:
+            return False
+        name = str(frame.f_globals.get("__name__") or "")
+        filename = str(frame.f_code.co_filename or "")
+        if looks_like_heaven(name, filename):
+            return True
+    return False
+
+
+class DatahubBridge:
+    def __init__(
+        self,
+        settings: DatahubSettings,
+        client: DatahubClient,
+        shadow_sink: ShadowSink | None = None,
+        heaven_guard: Callable[[], bool] | None = None,
+    ) -> None:
+        self.settings = settings
+        self.client = client
+        self.shadow_sink = shadow_sink
+        self.heaven_guard = heaven_guard or caller_is_heaven
+
+    def dataset_status(self, trade_date: str) -> list[dict[str, Any]] | None:
+        flags = self.settings.flags("status")
+        if not flags.read and not flags.shadow:
+            return None
+        try:
+            response = self._require_fresh(self.client.dataset_status(yyyymmdd(trade_date)), "status")
+            rows = list(response.data or [])
+            if flags.shadow:
+                self._emit_shadow(compare_rows("status", [], rows, response.meta))
+            if flags.read:
+                return rows
+            return None
+        except Exception as exc:
+            self._log_failure("status", exc)
+            if flags.shadow:
+                self._emit_shadow(compare_rows("status", [], [], {}, self._error_text(exc)))
+            return None
+
+    def batches(self, trade_date: str, dataset: str = "") -> list[dict[str, Any]] | None:
+        flags = self.settings.flags("status")
+        if not flags.read:
+            return None
+        try:
+            response = self._require_fresh(
+                self.client.batches(yyyymmdd(trade_date), dataset),
+                "status",
+            )
+            return list(response.data or [])
+        except Exception as exc:
+            self._log_failure("status", exc)
+            return None
+
+    def query(
+        self,
+        api_name: str,
+        params: dict[str, Any] | None,
+        fields: str,
+        legacy_query: Callable[..., list[dict[str, Any]]],
+    ) -> list[dict[str, Any]]:
+        dataset = API_TO_DATASET.get(api_name)
+        # 问天允许后续纳入 datahub；首批只读接入仍保持旧链路，避免误切。
+        if not dataset or self.heaven_guard():
+            return legacy_query(api_name, params, fields)
+        flags = self.settings.flags(dataset)
+        if not flags.read and not flags.shadow:
+            return legacy_query(api_name, params, fields)
+
+        hub_rows: list[dict[str, Any]] | None = None
+        hub_meta: dict[str, Any] = {}
+        hub_error: str | None = None
+        hub_canonical: list[dict[str, Any]] = []
+        try:
+            response = self._fetch_dataset(dataset, params or {})
+            hub_canonical = self._extract_rows(dataset, response, params or {})
+            hub_rows = to_native_rows(dataset, hub_canonical)
+            hub_meta = dict(response.meta)
+            self._validate_usable(dataset, hub_rows, response)
+        except Exception as exc:
+            hub_error = self._error_text(exc)
+            self._log_failure(dataset, exc)
+
+        if flags.shadow:
+            try:
+                legacy_rows = legacy_query(api_name, params, fields)
+            except Exception as exc:
+                if flags.read and hub_rows is not None and hub_error is None:
+                    self._emit_shadow(compare_rows(dataset, [], hub_canonical, hub_meta, self._error_text(exc)))
+                    return project_fields(hub_rows, fields)
+                raise
+            self._emit_shadow(compare_rows(dataset, legacy_rows, hub_canonical, hub_meta, hub_error))
+            if flags.read and hub_rows is not None and hub_error is None:
+                return project_fields(hub_rows, fields)
+            return legacy_rows
+
+        if flags.read and hub_rows is not None and hub_error is None:
+            return project_fields(hub_rows, fields)
+        return legacy_query(api_name, params, fields)
+
+    def _fetch_dataset(self, dataset: str, params: dict[str, Any]) -> 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)
+        code = str(params.get("ts_code") or params.get("code") or "").strip()
+        if dataset == "calendar":
+            if not start or not end:
+                raise DatahubError("INVALID_ARGUMENT", "calendar requires start_date and end_date")
+            return self.client.calendar(start, end)
+        if dataset == "stocks":
+            return self._paginate(self.client.stocks, {})
+        fetchers = {
+            "daily": self.client.daily_bars,
+            "index_daily": self.client.index_bars,
+            "valuation": self.client.valuation,
+            "moneyflow": self.client.moneyflow,
+            "auction": self.client.auction,
+        }
+        fetcher = fetchers[dataset]
+        query: dict[str, Any] = {}
+        if code:
+            query["code"] = code
+        if date and not (params.get("start_date") or params.get("end_date")):
+            query["date"] = date
+        else:
+            if start:
+                query["from"] = start
+            if end:
+                query["to"] = end
+        if dataset == "daily":
+            query["adjust"] = "none"
+        return self._paginate(fetcher, query)
+
+    def _paginate(self, fetcher: Callable[..., DatahubResponse], params: dict[str, Any]) -> DatahubResponse:
+        limit = self.settings.page_limit
+        offset = 0
+        rows: list[Any] = []
+        meta: dict[str, Any] = {}
+        schema_version = 1
+        while True:
+            page = fetcher(**{**params, "limit": limit, "offset": offset})
+            meta = dict(page.meta)
+            schema_version = page.schema_version
+            data = page.data or []
+            if not isinstance(data, list):
+                raise DatahubError("INTERNAL", "datahub returned a non-list payload")
+            rows.extend(data)
+            if len(data) < limit:
+                break
+            offset += limit
+            if offset > 200_000:
+                break
+        return DatahubResponse(data=rows, meta=meta, schema_version=schema_version)
+
+    def _extract_rows(
+        self,
+        dataset: str,
+        response: DatahubResponse,
+        params: dict[str, Any],
+    ) -> list[dict[str, Any]]:
+        rows = [dict(item) for item in (response.data or [])]
+        if dataset == "calendar":
+            return filter_calendar_rows(rows, params)
+        if dataset == "stocks":
+            return filter_stock_rows(rows, params)
+        return rows
+
+    def _validate_usable(self, dataset: str, rows: list[dict[str, Any]], response: DatahubResponse) -> None:
+        meta = response.meta or {}
+        stale_seconds = int(meta.get("staleness_seconds") or 0)
+        if meta.get("stale") or stale_seconds > self.settings.stale_seconds_max:
+            raise DatahubError("STALE", f"{dataset} data is stale")
+        if dataset in EMPTY_FAIL_DATASETS and not rows:
+            raise DatahubError("EMPTY", f"{dataset} returned no rows")
+        coverage = meta.get("coverage") if isinstance(meta.get("coverage"), dict) else {}
+        if meta.get("incomplete") is True or coverage.get("complete") is False:
+            missing = coverage.get("missing_count")
+            raise DatahubError("INCOMPLETE", f"{dataset} range is incomplete missing={missing}")
+
+    def _require_fresh(self, response: DatahubResponse, dataset: str) -> DatahubResponse:
+        self._validate_usable(dataset, list(response.data or []) if isinstance(response.data, list) else [], response)
+        return response
+
+    def _emit_shadow(self, report: dict[str, Any]) -> None:
+        safe = redact_value(report, secrets=self.settings.secrets())
+        LOGGER.info("datahub shadow %s", safe)
+        if self.shadow_sink is not None:
+            self.shadow_sink(report)
+
+    def _log_failure(self, dataset: str, exc: Exception) -> None:
+        LOGGER.warning(
+            "datahub fallback dataset=%s error=%s",
+            dataset,
+            redact_text(self._error_text(exc), self.settings.secrets()),
+        )
+
+    def _error_text(self, exc: Exception) -> str:
+        if isinstance(exc, DatahubError):
+            text = f"{exc.code}: {exc.message}"
+        else:
+            text = str(exc)
+        return redact_text(text, self.settings.secrets())
+
+
+class DatahubAwareTushareClient:
+    def __init__(self, legacy: TushareClient, bridge: DatahubBridge) -> None:
+        self._legacy = legacy
+        self._bridge = bridge
+
+    def query(
+        self,
+        api_name: str,
+        params: dict[str, Any] | None = None,
+        fields: str = "",
+    ) -> list[dict[str, Any]]:
+        return self._bridge.query(api_name, params, fields, self._legacy.query)
+
+    def __getattr__(self, name: str) -> Any:
+        return getattr(self._legacy, name)
@@ -0,0 +1,184 @@
+from __future__ import annotations
+
+import json
+import logging
+import urllib.error
+import urllib.parse
+import urllib.request
+from dataclasses import dataclass, field
+from typing import Any, Callable
+
+from backend.data.datahub.errors import DatahubError
+from backend.data.datahub.redact import redact_text
+from backend.data.datahub.settings import DatahubSettings
+
+LOGGER = logging.getLogger("xiaobai.datahub")
+UrlOpen = Callable[..., Any]
+
+
+@dataclass(frozen=True)
+class DatahubResponse:
+    data: Any
+    meta: dict[str, Any] = field(default_factory=dict)
+    schema_version: int = 1
+    status: int = 200
+
+
+class DatahubClient:
+    def __init__(
+        self,
+        settings: DatahubSettings,
+        urlopen: UrlOpen = urllib.request.urlopen,
+    ) -> None:
+        self.settings = settings
+        self._urlopen = urlopen
+
+    def health(self) -> DatahubResponse:
+        return self.get("/v1/health")
+
+    def calendar(self, start: str, end: str) -> DatahubResponse:
+        return self.get("/v1/calendar", {"from": start, "to": end})
+
+    def stocks(self, updated_since: str = "", limit: int | None = None, offset: int = 0) -> DatahubResponse:
+        params: dict[str, Any] = {"offset": offset, "limit": limit or self.settings.page_limit}
+        if updated_since:
+            params["updated_since"] = updated_since
+        return self.get("/v1/stocks", params)
+
+    def daily_bars(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/bars/daily", params)
+
+    def index_bars(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/indexes/bars", params)
+
+    def valuation(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/valuation", params)
+
+    def moneyflow(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/moneyflow", params)
+
+    def auction(self, **params: Any) -> DatahubResponse:
+        return self.get("/v1/auction", params)
+
+    def dataset_status(self, date: str) -> DatahubResponse:
+        return self.get("/v1/datasets/status", {"date": date})
+
+    def batches(self, date: str, dataset: str = "") -> DatahubResponse:
+        params: dict[str, Any] = {"date": date}
+        if dataset:
+            params["dataset"] = dataset
+        return self.get("/v1/batches", params)
+
+    def get(self, path: str, params: dict[str, Any] | None = None) -> DatahubResponse:
+        if not self.settings.token:
+            raise DatahubError("NOT_CONFIGURED", "DATAHUB_TOKEN is not configured")
+        query = {
+            key: value
+            for key, value in (params or {}).items()
+            if value is not None and value != ""
+        }
+        url = self.settings.base_url + path
+        if query:
+            url = f"{url}?{urllib.parse.urlencode(query)}"
+        attempts = 1 + max(0, self.settings.retries)
+        last_error: DatahubError | None = None
+        for attempt in range(attempts):
+            try:
+                return self._request(url)
+            except DatahubError as exc:
+                last_error = exc
+                if exc.code not in {"TIMEOUT", "UNAVAILABLE"} or attempt + 1 >= attempts:
+                    raise
+                LOGGER.warning(
+                    "datahub retry %s/%s %s",
+                    attempt + 1,
+                    attempts,
+                    redact_text(str(exc), self.settings.secrets()),
+                )
+        raise last_error or DatahubError("INTERNAL", "datahub request failed")
+
+    def _request(self, url: str) -> DatahubResponse:
+        request = urllib.request.Request(
+            url,
+            headers={
+                "Accept": "application/json",
+                "X-Datahub-Token": self.settings.token,
+                "User-Agent": "XiaobaiReviewDatahub/1.0",
+            },
+            method="GET",
+        )
+        try:
+            with self._urlopen(request, timeout=self.settings.timeout_seconds) as response:
+                status = int(getattr(response, "status", 200) or 200)
+                raw = response.read().decode("utf-8")
+        except TimeoutError as exc:
+            raise DatahubError("TIMEOUT", "datahub request timed out") from exc
+        except urllib.error.HTTPError as exc:
+            body = _read_error_body(exc)
+            raise _http_error(exc.code, body, self.settings.secrets()) from exc
+        except urllib.error.URLError as exc:
+            reason = redact_text(str(getattr(exc, "reason", exc)), self.settings.secrets())
+            if "timed out" in reason.lower():
+                raise DatahubError("TIMEOUT", "datahub request timed out") from exc
+            raise DatahubError("UNAVAILABLE", f"datahub unavailable: {reason}") from exc
+        payload = _parse_json(raw, self.settings.secrets())
+        return _as_response(payload, status, self.settings.secrets())
+
+
+def _parse_json(raw: str, secrets: tuple[str, ...]) -> dict[str, Any]:
+    try:
+        payload = json.loads(raw)
+    except json.JSONDecodeError as exc:
+        raise DatahubError("INTERNAL", "datahub returned invalid json") from exc
+    if not isinstance(payload, dict):
+        raise DatahubError("INTERNAL", "datahub returned a non-object payload")
+    return payload
+
+
+def _as_response(payload: dict[str, Any], status: int, secrets: tuple[str, ...]) -> DatahubResponse:
+    error = payload.get("error")
+    if isinstance(error, dict):
+        raise _mapped_error(str(error.get("code") or "INTERNAL"), str(error.get("message") or "datahub error"), status)
+    if status >= 400:
+        raise DatahubError("UNAVAILABLE", f"datahub http {status}", status)
+    return DatahubResponse(
+        data=payload.get("data"),
+        meta=dict(payload.get("meta") or {}),
+        schema_version=int(payload.get("schema_version") or 1),
+        status=status,
+    )
+
+
+def _http_error(status: int, payload: dict[str, Any], secrets: tuple[str, ...]) -> DatahubError:
+    error = payload.get("error") if isinstance(payload.get("error"), dict) else {}
+    code = str((error or {}).get("code") or "")
+    message = str((error or {}).get("message") or payload.get("message") or f"datahub http {status}")
+    message = redact_text(message, secrets)
+    if status == 401 or code == "UNAUTHORIZED":
+        return DatahubError("UNAUTHORIZED", message, status)
+    if status == 404 or code == "DATASET_NOT_PUBLISHED":
+        return DatahubError("DATASET_NOT_PUBLISHED", message, status)
+    if status == 400 or code == "INVALID_ARGUMENT":
+        return DatahubError("INVALID_ARGUMENT", message, status)
+    if status in {429, 503} or code in {"RATE_LIMITED", "SOURCE_UNAVAILABLE"}:
+        return DatahubError("UNAVAILABLE", message, status)
+    return DatahubError(code or "INTERNAL", message, status)
+
+
+def _mapped_error(code: str, message: str, status: int) -> DatahubError:
+    if code == "STALE_DATA":
+        return DatahubError("STALE", message, status)
+    if code in {"UNAUTHORIZED", "DATASET_NOT_PUBLISHED", "INVALID_ARGUMENT"}:
+        return DatahubError(code, message, status)
+    if code in {"RATE_LIMITED", "SOURCE_UNAVAILABLE"}:
+        return DatahubError("UNAVAILABLE", message, status)
+    return DatahubError(code or "INTERNAL", message, status)
+
+
+def _read_error_body(exc: urllib.error.HTTPError) -> dict[str, Any]:
+    try:
+        raw = exc.read().decode("utf-8")
+        payload = json.loads(raw)
+        return payload if isinstance(payload, dict) else {"message": raw}
+    except Exception:
+        return {"message": str(exc)}
@@ -0,0 +1,134 @@
+from __future__ import annotations
+
+from typing import Any
+
+from backend.data.datahub.native import SCALE_FIELDS, row_key, to_canonical_row, yyyymmdd
+
+NUMERIC_TOLERANCE = 1e-4
+
+
+def compare_rows(
+    dataset: str,
+    legacy_rows: list[dict[str, Any]],
+    hub_rows: list[dict[str, Any]] | None,
+    hub_meta: dict[str, Any] | None = None,
+    hub_error: str | None = None,
+) -> dict[str, Any]:
+    hub = hub_rows or []
+    legacy_map = {row_key(dataset, row): row for row in legacy_rows}
+    hub_map = {row_key(dataset, _align_hub_row(row)): row for row in hub}
+    missing_hub = sorted(key for key in legacy_map if key not in hub_map)
+    missing_legacy = sorted(key for key in hub_map if key not in legacy_map)
+    value_diffs: list[dict[str, Any]] = []
+    unit_conversion: list[dict[str, Any]] = []
+    matched = 0
+    for key, legacy in legacy_map.items():
+        hub_row = hub_map.get(key)
+        if hub_row is None:
+            continue
+        field_report = _compare_fields(dataset, legacy, hub_row)
+        if field_report["unit_conversion"]:
+            unit_conversion.append({"key": list(key), "fields": field_report["unit_conversion"]})
+        if field_report["value_diff"]:
+            value_diffs.append({"key": list(key), "fields": field_report["value_diff"]})
+        if not field_report["unit_conversion"] and not field_report["value_diff"]:
+            matched += 1
+    stale_seconds = int((hub_meta or {}).get("staleness_seconds") or 0)
+    time_skew = bool((hub_meta or {}).get("stale")) or stale_seconds > 0
+    return {
+        "dataset": dataset,
+        "legacy_rows": len(legacy_rows),
+        "hub_rows": len(hub),
+        "matched": matched,
+        "missing_hub": [list(item) for item in missing_hub[:20]],
+        "missing_legacy": [list(item) for item in missing_legacy[:20]],
+        "missing_hub_count": len(missing_hub),
+        "missing_legacy_count": len(missing_legacy),
+        "value_diff_count": len(value_diffs),
+        "unit_conversion_count": len(unit_conversion),
+        "value_diffs": value_diffs[:20],
+        "unit_conversion": unit_conversion[:20],
+        "time_skew": time_skew,
+        "staleness_seconds": stale_seconds,
+        "published_at": (hub_meta or {}).get("published_at"),
+        "trade_date": yyyymmdd((hub_meta or {}).get("trade_date")),
+        "hub_error": hub_error,
+        "equal": (
+            not hub_error
+            and not missing_hub
+            and not missing_legacy
+            and not value_diffs
+            and not unit_conversion
+            and not time_skew
+        ),
+    }
+
+
+def _align_hub_row(row: dict[str, Any]) -> dict[str, Any]:
+    aligned = dict(row)
+    if "volume" in aligned and "vol" not in aligned:
+        aligned["vol"] = aligned.get("volume")
+    return aligned
+
+
+def _compare_fields(dataset: str, legacy: dict[str, Any], hub: dict[str, Any]) -> dict[str, list[dict[str, Any]]]:
+    canonical_legacy = to_canonical_row(dataset, legacy)
+    hub_canonical = _hub_canonical(dataset, hub)
+    native_hub = _align_hub_row(hub)
+    value_diff: list[dict[str, Any]] = []
+    unit_conversion: list[dict[str, Any]] = []
+    keys = (set(canonical_legacy) | set(hub_canonical)) - {"batch_id", "updated_at", "volume"}
+    scales = SCALE_FIELDS.get(dataset) or {}
+    for field in sorted(keys):
+        left = canonical_legacy.get(field)
+        right = hub_canonical.get(field)
+        if _same(left, right):
+            continue
+        native_left = legacy.get(field)
+        hub_raw = native_hub.get(field)
+        if field in scales and _near(_optional(native_left), _optional(hub_raw)):
+            unit_conversion.append(
+                {"field": field, "legacy": native_left, "hub": hub_raw, "reason": "unit_conversion"}
+            )
+            continue
+        value_diff.append({"field": field, "legacy": left, "hub": right, "reason": "value_diff"})
+    return {"value_diff": value_diff, "unit_conversion": unit_conversion}
+
+
+def _hub_canonical(dataset: str, row: dict[str, Any]) -> dict[str, Any]:
+    """Hub API rows are already canonical; only align field names."""
+    aligned = dict(row)
+    if "volume" in aligned and "vol" not in aligned:
+        aligned["vol"] = aligned.get("volume")
+    if dataset == "calendar":
+        is_open = aligned.get("is_open")
+        aligned["is_open"] = 1 if is_open in (True, 1, "1", "Y", "y") else 0
+        aligned["cal_date"] = yyyymmdd(aligned.get("cal_date"))
+        aligned["pretrade_date"] = yyyymmdd(aligned.get("pretrade_date")) or None
+        aligned["exchange"] = str(aligned.get("exchange") or "SSE")
+    return aligned
+
+
+def _same(left: Any, right: Any) -> bool:
+    if left in (None, "") and right in (None, ""):
+        return True
+    if isinstance(left, (int, float)) or isinstance(right, (int, float)):
+        return _near(_optional(left), _optional(right))
+    return str(left or "") == str(right or "")
+
+
+def _near(left: float | None, right: float | None) -> bool:
+    if left is None and right is None:
+        return True
+    if left is None or right is None:
+        return False
+    return abs(left - right) <= max(NUMERIC_TOLERANCE, abs(left) * 1e-9, abs(right) * 1e-9)
+
+
+def _optional(value: Any) -> float | None:
+    if value in (None, ""):
+        return None
+    try:
+        return float(value)
+    except (TypeError, ValueError):
+        return None
@@ -0,0 +1,12 @@
+from __future__ import annotations
+
+
+class DatahubError(RuntimeError):
+    def __init__(self, code: str, message: str, status: int | None = None) -> None:
+        super().__init__(message)
+        self.code = code
+        self.message = message
+        self.status = status
+
+    def __str__(self) -> str:
+        return f"{self.code}: {self.message}"
@@ -0,0 +1,153 @@
+from __future__ import annotations
+
+from typing import Any
+
+from backend.data.numbers import finite_number
+
+
+AMOUNT_THOUSAND_YUAN = 1000.0
+AMOUNT_WAN_YUAN = 10000.0
+VOLUME_LOT = 100.0
+
+API_TO_DATASET = {
+    "trade_cal": "calendar",
+    "stock_basic": "stocks",
+    "daily": "daily",
+    "daily_basic": "valuation",
+    "index_daily": "index_daily",
+    "moneyflow": "moneyflow",
+    "stk_auction": "auction",
+}
+
+SCALE_FIELDS = {
+    "daily": {"vol": VOLUME_LOT, "amount": AMOUNT_THOUSAND_YUAN},
+    "index_daily": {"vol": VOLUME_LOT, "amount": AMOUNT_THOUSAND_YUAN},
+    "valuation": {"total_mv": AMOUNT_WAN_YUAN, "circ_mv": AMOUNT_WAN_YUAN},
+    "moneyflow": {
+        "buy_sm_amount": AMOUNT_WAN_YUAN,
+        "sell_sm_amount": AMOUNT_WAN_YUAN,
+        "buy_md_amount": AMOUNT_WAN_YUAN,
+        "sell_md_amount": AMOUNT_WAN_YUAN,
+        "buy_lg_amount": AMOUNT_WAN_YUAN,
+        "sell_lg_amount": AMOUNT_WAN_YUAN,
+        "buy_elg_amount": AMOUNT_WAN_YUAN,
+        "sell_elg_amount": AMOUNT_WAN_YUAN,
+        "net_mf_amount": AMOUNT_WAN_YUAN,
+    },
+    "auction": {"vol": VOLUME_LOT, "float_share": AMOUNT_WAN_YUAN},
+}
+
+
+def yyyymmdd(value: Any) -> str:
+    return str(value or "").replace("-", "")[:8]
+
+
+def to_native_rows(dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    return [to_native_row(dataset, row) for row in rows]
+
+
+def to_native_row(dataset: str, row: dict[str, Any]) -> dict[str, Any]:
+    if dataset == "calendar":
+        is_open = row.get("is_open")
+        return {
+            "exchange": str(row.get("exchange") or "SSE"),
+            "cal_date": yyyymmdd(row.get("cal_date")),
+            "is_open": 1 if is_open in (True, 1, "1", "Y", "y") else 0,
+            "pretrade_date": yyyymmdd(row.get("pretrade_date")) or None,
+        }
+    converted = dict(row)
+    converted.pop("batch_id", None)
+    if "volume" in converted and "vol" not in converted:
+        converted["vol"] = converted.pop("volume")
+    elif "volume" in converted:
+        converted.pop("volume", None)
+    scales = SCALE_FIELDS.get(dataset) or {}
+    for field, factor in scales.items():
+        if field in converted:
+            converted[field] = _unscale(converted.get(field), factor)
+    if dataset == "stocks":
+        converted.pop("updated_at", None)
+    return converted
+
+
+def to_canonical_row(dataset: str, row: dict[str, Any]) -> dict[str, Any]:
+    if dataset == "calendar":
+        is_open = row.get("is_open")
+        return {
+            "exchange": str(row.get("exchange") or "SSE"),
+            "cal_date": yyyymmdd(row.get("cal_date")),
+            "is_open": 1 if is_open in (True, 1, "1", "Y", "y") else 0,
+            "pretrade_date": yyyymmdd(row.get("pretrade_date")) or None,
+        }
+    converted = dict(row)
+    if "volume" in converted and "vol" not in converted:
+        converted["vol"] = converted.pop("volume")
+    scales = SCALE_FIELDS.get(dataset) or {}
+    for field, factor in scales.items():
+        if field in converted:
+            converted[field] = _scale(converted.get(field), factor)
+    return converted
+
+
+def row_key(dataset: str, row: dict[str, Any]) -> tuple[str, ...]:
+    if dataset == "calendar":
+        return (yyyymmdd(row.get("cal_date")),)
+    if dataset == "stocks":
+        return (str(row.get("ts_code") or "").upper(),)
+    if dataset == "status":
+        return (str(row.get("dataset") or ""), yyyymmdd(row.get("trade_date")))
+    return (str(row.get("ts_code") or "").upper(), yyyymmdd(row.get("trade_date")))
+
+
+def project_fields(rows: list[dict[str, Any]], fields: str) -> list[dict[str, Any]]:
+    keys = [item.strip() for item in str(fields or "").split(",") if item.strip()]
+    if not keys:
+        return rows
+    return [{key: row.get(key) for key in keys} for row in rows]
+
+
+def filter_stock_rows(rows: list[dict[str, Any]], params: dict[str, Any] | None) -> list[dict[str, Any]]:
+    payload = params or {}
+    ts_code = str(payload.get("ts_code") or "").strip().upper()
+    status = str(payload.get("list_status") or "").strip()
+    name = str(payload.get("name") or "").strip()
+    filtered = rows
+    if ts_code:
+        filtered = [row for row in filtered if str(row.get("ts_code") or "").upper() == ts_code]
+    if status:
+        filtered = [row for row in filtered if str(row.get("list_status") or status) == status]
+    if name:
+        filtered = [row for row in filtered if name.casefold() in str(row.get("name") or "").casefold()]
+    return filtered
+
+
+def filter_calendar_rows(rows: list[dict[str, Any]], params: dict[str, Any] | None) -> list[dict[str, Any]]:
+    payload = params or {}
+    if payload.get("is_open") in (1, "1", True):
+        return [row for row in rows if int(row.get("is_open") or 0) == 1]
+    if payload.get("is_open") in (0, "0", False):
+        return [row for row in rows if int(row.get("is_open") or 0) == 0]
+    return rows
+
+
+def _scale(value: Any, factor: float) -> float | None:
+    number = _optional_number(value)
+    if number is None:
+        return None
+    return number * factor
+
+
+def _unscale(value: Any, factor: float) -> float | None:
+    number = _optional_number(value)
+    if number is None or factor == 0:
+        return None
+    return number / factor
+
+
+def _optional_number(value: Any) -> float | None:
+    if value in (None, ""):
+        return None
+    number = finite_number(value, default=float("nan"))
+    if number != number:
+        return None
+    return number
@@ -0,0 +1,39 @@
+from __future__ import annotations
+
+from typing import Any
+
+
+SECRET_HINTS = (
+    "token",
+    "password",
+    "secret",
+    "key",
+    "authorization",
+    "credential",
+    "cookie",
+)
+
+
+def redact_value(value: Any, key: str = "", secrets: tuple[str, ...] = ()) -> Any:
+    lowered = key.lower()
+    if any(part in lowered for part in SECRET_HINTS):
+        return "***"
+    if isinstance(value, dict):
+        return {
+            str(item_key): redact_value(item_value, str(item_key), secrets)
+            for item_key, item_value in value.items()
+        }
+    if isinstance(value, list):
+        return [redact_value(item, key, secrets) for item in value]
+    text = str(value) if value is not None and not isinstance(value, (int, float, bool)) else value
+    if isinstance(text, str):
+        return redact_text(text, secrets)
+    return value
+
+
+def redact_text(text: str, secrets: tuple[str, ...] = ()) -> str:
+    redacted = text
+    for secret in secrets:
+        if secret:
+            redacted = redacted.replace(secret, "***")
+    return redacted
@@ -0,0 +1,120 @@
+from __future__ import annotations
+
+import json
+import os
+from dataclasses import dataclass
+from pathlib import Path
+from typing import Any, Mapping
+
+from backend.bootstrap.config import APP_DIR
+
+
+DATASETS = (
+    "calendar",
+    "stocks",
+    "daily",
+    "index_daily",
+    "valuation",
+    "moneyflow",
+    "auction",
+    "status",
+)
+
+ENV_DATASET = {
+    "calendar": "CALENDAR",
+    "stocks": "STOCKS",
+    "daily": "DAILY",
+    "index_daily": "INDEX_DAILY",
+    "valuation": "VALUATION",
+    "moneyflow": "MONEYFLOW",
+    "auction": "AUCTION",
+    "status": "STATUS",
+}
+
+DEFAULT_CONFIG_PATH = APP_DIR / "config" / "datahub.config.json"
+
+
+def _truthy(value: Any) -> bool:
+    return str(value or "").strip().lower() in {"1", "true", "yes", "on"}
+
+
+def _int(value: Any, default: int) -> int:
+    try:
+        return int(value)
+    except (TypeError, ValueError):
+        return default
+
+
+@dataclass(frozen=True)
+class DatasetFlags:
+    name: str
+    read: bool = False
+    shadow: bool = False
+
+
+@dataclass(frozen=True)
+class DatahubSettings:
+    base_url: str
+    token: str
+    timeout_seconds: int = 8
+    retries: int = 1
+    page_limit: int = 5000
+    stale_seconds_max: int = 86400
+    datasets: dict[str, DatasetFlags] | None = None
+
+    def flags(self, dataset: str) -> DatasetFlags:
+        mapped = self.datasets or {}
+        return mapped.get(dataset) or DatasetFlags(dataset)
+
+    def any_enabled(self) -> bool:
+        return any(item.read or item.shadow for item in (self.datasets or {}).values())
+
+    def secrets(self) -> tuple[str, ...]:
+        return tuple(item for item in (self.token,) if item)
+
+    @classmethod
+    def load(
+        cls,
+        path: Path | None = None,
+        environ: Mapping[str, str] | None = None,
+        credentials: Mapping[str, object] | None = None,
+    ) -> "DatahubSettings":
+        config_path = path or DEFAULT_CONFIG_PATH
+        payload: dict[str, Any] = {}
+        if config_path.is_file():
+            payload = json.loads(config_path.read_text(encoding="utf-8"))
+        env = dict(os.environ if environ is None else environ)
+        creds = dict(credentials or {})
+        dataset_flags: dict[str, DatasetFlags] = {}
+        raw_datasets = payload.get("datasets") or {}
+        for name in DATASETS:
+            item = raw_datasets.get(name) or {}
+            env_key = ENV_DATASET[name]
+            read = _truthy(env.get(f"DATAHUB_READ_{env_key}")) if f"DATAHUB_READ_{env_key}" in env else bool(item.get("read"))
+            shadow = (
+                _truthy(env.get(f"DATAHUB_SHADOW_{env_key}"))
+                if f"DATAHUB_SHADOW_{env_key}" in env
+                else bool(item.get("shadow"))
+            )
+            dataset_flags[name] = DatasetFlags(name, read=read, shadow=shadow)
+        token = str(
+            env.get("DATAHUB_TOKEN")
+            or creds.get("datahub_token")
+            or payload.get("token")
+            or ""
+        ).strip()
+        base_url = str(
+            env.get("DATAHUB_BASE_URL")
+            or creds.get("datahub_base_url")
+            or payload.get("base_url")
+            or "http://127.0.0.1:8766"
+        ).strip().rstrip("/")
+        return cls(
+            base_url=base_url,
+            token=token,
+            timeout_seconds=_int(env.get("DATAHUB_TIMEOUT") or payload.get("timeout_seconds"), 8),
+            retries=max(0, _int(env.get("DATAHUB_RETRIES") or payload.get("retries"), 1)),
+            page_limit=max(1, _int(payload.get("page_limit"), 5000)),
+            stale_seconds_max=max(0, _int(payload.get("stale_seconds_max"), 86400)),
+            datasets=dataset_flags,
+        )
@@ -3,8 +3,10 @@ from __future__ import annotations
 from collections.abc import Callable
 from dataclasses import dataclass
 from datetime import datetime
+from typing import Any

 from backend.data.contracts import DataUsage
+from backend.data.datahub import DatahubAwareTushareClient, DatahubBridge, DatahubClient, DatahubSettings
 from backend.data.policy import DataSourcePolicy
 from backend.data.providers import IfindProvider, TushareProvider
 from backend.data.quality import DataQualityGate, QualityEvidence, QualityReport
@@ -22,6 +24,7 @@ class DataGateway:
    ifind_provider: IfindProvider
    chart_data: MarketChartClient
    realtime_observer: WebRealtimeAggregator
+    datahub: DatahubBridge

    @property
    def ifind(self) -> IfindHttpClient:
@@ -34,7 +37,13 @@ class DataGateway:
    ) -> TushareClient:
        if dataset_id:
            self.policy.assert_allowed(dataset_id, "tushare", usage)
-        return self.tushare_provider.client()
+        return DatahubAwareTushareClient(self.tushare_provider.client(), self.datahub)
+
+    def dataset_status(self, trade_date: str) -> list[dict[str, Any]] | None:
+        return self.datahub.dataset_status(trade_date)
+
+    def batches(self, trade_date: str, dataset: str = "") -> list[dict[str, Any]] | None:
+        return self.datahub.batches(trade_date, dataset)

    def assert_source(self, dataset_id: str, provider_id: str, usage: DataUsage) -> None:
        self.policy.assert_allowed(dataset_id, provider_id, usage)
@@ -64,6 +73,7 @@ class DataGateway:
 def build_data_gateway(
    credentials: dict[str, object],
    tushare_token_supplier: Callable[[], str] | None = None,
+    datahub_settings: DatahubSettings | None = None,
 ) -> DataGateway:
    ifind = IfindHttpClient(
        str(credentials.get("ifind_refresh_token") or ""),
@@ -73,6 +83,8 @@ def build_data_gateway(
        lambda: str(credentials.get("tushare_token") or "")
    )
    policy = DataSourcePolicy.load()
+    settings = datahub_settings or DatahubSettings.load(credentials=credentials)
+    datahub_client = DatahubClient(settings)
    return DataGateway(
        policy=policy,
        quality=DataQualityGate.load(policy),
@@ -80,4 +92,5 @@ def build_data_gateway(
        ifind_provider=IfindProvider(ifind),
        chart_data=MarketChartClient(ifind, EastmoneyChartClient()),
        realtime_observer=WebRealtimeAggregator(),
+        datahub=DatahubBridge(settings, datahub_client),
    )
@@ -79,6 +79,8 @@ class MarketServiceMixin:
        if not force:
            snapshot = self.database.get_snapshot(normalized_date)
            if snapshot and str((snapshot.get("meta") or {}).get("source") or "") != "demo":
+                if self._should_retry_incomplete_snapshot(snapshot, normalized_date):
+                    return self.sync_dashboard(normalized_date)
                snapshot = copy.deepcopy(snapshot)
                if normalized_date != now.strftime("%Y%m%d"):
                    snapshot.setdefault("meta", {}).update(
@@ -97,6 +99,8 @@ class MarketServiceMixin:
                "dashboard_request_v1", normalized_date
            )
            if resolved and str((resolved.get("meta") or {}).get("source") or "") != "demo":
+                if self._should_retry_incomplete_snapshot(resolved, normalized_date):
+                    return self.sync_dashboard(normalized_date)
                resolved = copy.deepcopy(resolved)
                resolved.setdefault("meta", {})["requested_date"] = self._display_compact_date(
                    normalized_date
@@ -138,6 +142,68 @@ class MarketServiceMixin:
    def _display_compact_date(compact: str) -> str:
        return f"{compact[:4]}-{compact[4:6]}-{compact[6:8]}"

+    @staticmethod
+    def _chinese_month_day(value: str) -> str:
+        compact = str(value or "").replace("-", "").replace("/", "")
+        if len(compact) < 8 or not compact[:8].isdigit():
+            return "最近可用交易日"
+        return f"{int(compact[4:6])} 月 {int(compact[6:8])} 日"
+
+    @classmethod
+    def _preparing_display_notice(cls, actual_date: str, requested_date: str) -> str:
+        shown = cls._chinese_month_day(actual_date)
+        requested = str(requested_date or "").replace("-", "")
+        if requested == date.today().strftime("%Y%m%d"):
+            return f"今日数据正在准备，当前展示 {shown}"
+        return f"所选日期数据尚未到齐，当前展示 {shown}"
+
+    @staticmethod
+    def _snapshot_age_seconds(meta: dict[str, Any]) -> float:
+        raw = str(meta.get("updated_at") or "")
+        if not raw:
+            return 10**9
+        try:
+            updated_at = datetime.fromisoformat(raw)
+        except ValueError:
+            return 10**9
+        now = datetime.now().astimezone()
+        if updated_at.tzinfo is None:
+            updated_at = updated_at.replace(tzinfo=now.tzinfo)
+        return (now - updated_at.astimezone(now.tzinfo)).total_seconds()
+
+    def _should_retry_incomplete_snapshot(
+        self, snapshot: dict[str, Any], requested_date: str
+    ) -> bool:
+        if requested_date != date.today().strftime("%Y%m%d"):
+            return False
+        meta = snapshot.get("meta") or {}
+        incomplete = (
+            meta.get("limit_data_source") == "derived"
+            or bool(meta.get("carried_forward"))
+            or str(meta.get("trade_date") or "").replace("-", "") != requested_date
+        )
+        return incomplete and self._snapshot_age_seconds(meta) >= 60
+
+    def _annotate_data_status(self, dashboard: dict[str, Any]) -> dict[str, Any]:
+        meta = dashboard.setdefault("meta", {})
+        notice = str(meta.get("notice") or "")
+        requested = str(meta.get("requested_date") or "").replace("-", "")
+        actual = str(meta.get("trade_date") or "").replace("-", "")
+        if meta.get("limit_data_source") == "derived" and not meta.get("carried_forward"):
+            meta["data_status"] = "partial"
+            meta["display_notice"] = notice or "部分正式数据尚未到齐，当前展示日线推算结果"
+        elif meta.get("carried_forward"):
+            if "非交易日" in notice or "盘前" in notice:
+                meta["data_status"] = "carried"
+                meta["display_notice"] = notice
+            else:
+                meta["data_status"] = "preparing"
+                meta["display_notice"] = self._preparing_display_notice(actual, requested)
+        else:
+            meta["data_status"] = "official"
+            meta.setdefault("display_notice", "")
+        return dashboard
+
    def _carry_dashboard(
        self, snapshot: dict[str, Any], requested_date: str, reason: str
    ) -> dict[str, Any]:
@@ -152,7 +218,7 @@ class MarketServiceMixin:
                "notice": reason,
            }
        )
-        return carried
+        return self._annotate_data_status(carried)

    def _realtime_snapshot_due(
        self,
@@ -197,14 +263,14 @@ class MarketServiceMixin:
                if not self.configured:
                    raise TushareError("公共行情尚未配置")
                dashboard = self._tushare_client().dashboard(normalized_date)
-
-                if (dashboard.get("meta") or {}).get("limit_data_source") == "derived":
-                    raise TushareError(
-                        str((dashboard.get("meta") or {}).get("notice") or "官方涨跌停数据尚未返回")
+                meta = dashboard.setdefault("meta", {})
+                meta["source"] = source
+                meta["requested_date"] = self._display_compact_date(normalized_date)
+                if meta.get("limit_data_source") == "derived":
+                    meta.setdefault(
+                        "notice",
+                        "涨跌停高级接口当日数据尚未更新，已使用日线数据推算。",
                    )
-
-                dashboard["meta"]["source"] = source
-                dashboard["meta"]["requested_date"] = self._display_compact_date(normalized_date)
                dashboard = self._enrich_dashboard_sentiment(dashboard, normalized_date)
                record_count = self._record_count(dashboard)
                actual_date = normalize_date(
@@ -233,8 +299,11 @@ class MarketServiceMixin:
            except TushareError as exc:
                fallback = self.database.get_latest_real_snapshot(normalized_date)
                if fallback:
+                    actual = str((fallback.get("meta") or {}).get("trade_date") or "")
                    carried = self._carry_dashboard(
-                        fallback, normalized_date, f"最新行情暂不可用，沿用最近收盘快照：{exc}"
+                        fallback,
+                        normalized_date,
+                        self._preparing_display_notice(actual, normalized_date),
                    )
                    self.database.finish_sync(
                        sync_id, "fallback", self._record_count(carried), str(exc), "tushare"
@@ -1160,7 +1229,7 @@ class MarketServiceMixin:
            "storage": "sqlite",
            "cached": cached,
        }
-        return result
+        return self._annotate_data_status(result)

    @staticmethod
    def _record_count(dashboard: dict[str, Any]) -> int:
@@ -109,7 +109,14 @@ class HttpTransportMixin:
            return {}
        if length <= 0 or length > 65536:
            raise ValueError("请求内容为空或过大。")
-        return json.loads(self.rfile.read(length).decode("utf-8"))
+        raw = self.rfile.read(length)
+        try:
+            payload = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            raise ValueError("请求不是合法 JSON。") from None
+        if not isinstance(payload, dict):
+            raise ValueError("请求不是合法 JSON。")
+        return payload

    def serve_static(self, request_path: str) -> None:
        relative = unquote(request_path).lstrip("/") or "index.html"
@@ -0,0 +1,46 @@
+from __future__ import annotations
+
+from datetime import datetime, time as dt_time
+
+
+def dashboard_has_usable_data(dashboard: dict[str, object]) -> bool:
+    if not isinstance(dashboard, dict) or dashboard.get("status") == "failed":
+        return False
+    meta = dashboard.get("meta") or {}
+    overview = dashboard.get("overview") or {}
+    if isinstance(meta, dict) and (meta.get("trade_date") or meta.get("carried_forward")):
+        return True
+    return bool(isinstance(overview, dict) and overview)
+
+
+def verified_dashboard_result(dashboard: dict[str, object]) -> dict[str, object]:
+    """Manual refresh and automatic catch-up share this rule.
+
+    Derived limit lists or a previous usable snapshot are not whole-job failures.
+    Only a payload with no displayable market data is recorded as failed.
+    """
+    if dashboard_has_usable_data(dashboard):
+        return dashboard
+    meta = dashboard.get("meta") if isinstance(dashboard, dict) else None
+    notice = ""
+    if isinstance(meta, dict):
+        notice = str(meta.get("notice") or meta.get("display_notice") or "")
+    return {
+        "status": "failed",
+        "error": notice or "未获取到可用行情",
+    }
+
+
+def official_catchup_due(today: str, snapshot: dict[str, object]) -> bool:
+    now = datetime.now().astimezone().time().replace(tzinfo=None)
+    if not (dt_time(15, 5) <= now < dt_time(22, 0)):
+        return False
+    meta = snapshot.get("meta") if isinstance(snapshot.get("meta"), dict) else {}
+    actual = str(meta.get("trade_date") or "").replace("-", "")
+    if (
+        actual == today
+        and meta.get("limit_data_source") != "derived"
+        and not meta.get("carried_forward")
+    ):
+        return False
+    return True
@@ -5,16 +5,7 @@ import time
 from datetime import date

 from backend.bootstrap.config import normalize_date
-
-
-def _verified_dashboard_result(dashboard: dict[str, object]) -> dict[str, object]:
-    meta = dashboard.get("meta") or {}
-    if isinstance(meta, dict) and meta.get("carried_forward"):
-        return {
-            "status": "failed",
-            "error": str(meta.get("notice") or "未获取到所选日期的最新行情"),
-        }
-    return dashboard
+from backend.jobs.refresh import official_catchup_due, verified_dashboard_result


 class JobServiceMixin:
@@ -36,7 +27,7 @@ class JobServiceMixin:
        started = self.jobs.submit(
            "market.refresh",
            key,
-            lambda: _verified_dashboard_result(self.sync_dashboard(normalized)),
+            lambda: verified_dashboard_result(self.sync_dashboard(normalized)),
            {"trade_date": normalized, "trigger": "administrator"},
        )
        return {"started": started, "job_key": key if started else ""}
@@ -54,7 +45,15 @@ class JobServiceMixin:
            self.jobs.submit(
                "market.refresh",
                f"realtime:{today}:{bucket}",
-                lambda: self.sync_dashboard(today),
+                lambda: verified_dashboard_result(self.sync_dashboard(today)),
                {"trade_date": today, "trigger": "realtime-poll"},
            )
+        elif official_catchup_due(today, snapshot):
+            bucket = int(time.time() // 300)
+            self.jobs.submit(
+                "market.refresh",
+                f"catchup:{today}:{bucket}",
+                lambda: verified_dashboard_result(self.sync_dashboard(today)),
+                {"trade_date": today, "trigger": "official-catchup"},
+            )
        self._schedule_automatic_screeners(today, snapshot)
@@ -0,0 +1,45 @@
+# Optional overlay. Does not replace the existing xiaobai-review service.
+# Start later (总工部署时) with:
+#   docker compose -f compose.yaml -f compose.datahub.yaml up -d
+#
+# Required .env keys: DATAHUB_ENCRYPTION_KEY, DATAHUB_TOKEN, DATAHUB_ADMIN_PASSWORD, TUSHARE_TOKEN
+
+services:
+  xiaobai-datahub:
+    build:
+      context: ./xiaobai-datahub
+      dockerfile: Dockerfile
+    image: xiaobai-datahub:local
+    container_name: xiaobai-datahub
+    ports:
+      - "0.0.0.0:8766:8766/tcp"
+    env_file:
+      - ./xiaobai-datahub/.env
+    environment:
+      DATAHUB_ENCRYPTION_KEY: "${DATAHUB_ENCRYPTION_KEY:?DATAHUB_ENCRYPTION_KEY must be set}"
+      DATAHUB_TOKEN: "${DATAHUB_TOKEN:?DATAHUB_TOKEN must be set}"
+      DATAHUB_ADMIN_PASSWORD: "${DATAHUB_ADMIN_PASSWORD:?DATAHUB_ADMIN_PASSWORD must be set}"
+      TUSHARE_TOKEN: "${TUSHARE_TOKEN:-}"
+      DATAHUB_DB_PATH: /app/data/datahub.db
+      DATAHUB_BACKUP_DIR: /app/data/backups
+      TZ: Asia/Shanghai
+      PYTHONUTF8: "1"
+    volumes:
+      - type: bind
+        source: ./datahub-data
+        target: /app/data
+    restart: unless-stopped
+    init: true
+    read_only: true
+    tmpfs:
+      - /tmp:size=64m,mode=1777
+    security_opt:
+      - no-new-privileges:true
+    cap_drop:
+      - ALL
+    stop_grace_period: 30s
+    logging:
+      driver: json-file
+      options:
+        max-size: "10m"
+        max-file: "3"
@@ -12,6 +12,9 @@ These registries describe the approved product surface of the standalone applica
  providers, model entry points, CSS layers, and remaining code hotspots.
 - `data-fields.config.json`: canonical data products, provider eligibility, intended use, and
  known blocked datasets.
+- `datahub.config.json`: optional read-only client for `xiaobai-datahub`. Each dataset has its
+  own `read` / `shadow` flag, all default off. Environment variables `DATAHUB_READ_*` and
+  `DATAHUB_SHADOW_*` can override a single dataset without a master switch.
 - `data-quality.config.json`: freshness, coverage, units, adjustment, point-in-time, and
  fail-closed rules for every canonical data product.
 - `jobs.config.json`: background schedules, dependencies, lock keys, retry policy, timeouts,
@@ -204,6 +204,11 @@
      "path": "backend/data/providers/tushare_client.py",
      "runtime_role": "stable client facade for primary deterministic market data"
    },
+    {
+      "provider": "datahub",
+      "path": "backend/data/datahub/client.py",
+      "runtime_role": "optional official EOD read path behind per-dataset flags"
+    },
    {
      "provider": "ifind",
      "path": "backend/data/providers/ifind_client.py",
@@ -278,6 +283,18 @@
      "owner": "backend/data/providers/tushare.py",
      "compatibility_fallback": "backend/features/market/service.py"
    },
+    {
+      "client": "DatahubClient",
+      "owner": "backend/data/gateway.py"
+    },
+    {
+      "client": "DatahubBridge",
+      "owner": "backend/data/gateway.py"
+    },
+    {
+      "client": "DatahubAwareTushareClient",
+      "owner": "backend/data/gateway.py"
+    },
    {
      "client": "IfindHttpClient",
      "owner": "backend/data/gateway.py"
@@ -313,6 +330,7 @@
    "system_service": "backend/features/system/service.py",
    "account_bridge": "backend/features/accounts/application.py",
    "job_lifecycle": "backend/jobs/service.py",
+    "job_refresh_status": "backend/jobs/refresh.py",
    "feature_routes": "backend/features/*/routes.py"
  },
  "numeric_normalization": [
@@ -455,8 +473,8 @@
    },
    {
      "path": "frontend/shared/shell.css",
-      "bytes": 63659,
-      "lines": 3763
+      "bytes": 63733,
+      "lines": 3767
    },
    {
      "path": "backend/features/heaven/engine.py",
@@ -543,6 +561,11 @@
      "bytes": 14743,
      "lines": 342
    },
+    {
+      "path": "frontend/shared/dashboard.js",
+      "bytes": 14740,
+      "lines": 316
+    },
    {
      "path": "frontend/shared/admin.js",
      "bytes": 14410,
@@ -558,11 +581,6 @@
      "bytes": 13219,
      "lines": 289
    },
-    {
-      "path": "frontend/shared/dashboard.js",
-      "bytes": 12894,
-      "lines": 274
-    },
    {
      "path": "backend/features/market/insights_auction_data.py",
      "bytes": 12829,
@@ -768,16 +786,16 @@
      "bytes": 2514,
      "lines": 63
    },
+    {
+      "path": "backend/jobs/service.py",
+      "bytes": 2337,
+      "lines": 59
+    },
    {
      "path": "backend/features/mentor/routes.py",
      "bytes": 2299,
      "lines": 57
    },
-    {
-      "path": "backend/jobs/service.py",
-      "bytes": 2219,
-      "lines": 60
-    },
    {
      "path": "backend/features/screener/regime.py",
      "bytes": 2202,
@@ -813,6 +831,11 @@
      "bytes": 1791,
      "lines": 46
    },
+    {
+      "path": "backend/jobs/refresh.py",
+      "bytes": 1728,
+      "lines": 46
+    },
    {
      "path": "backend/features/alerts/routes.py",
      "bytes": 1687,
@@ -2,6 +2,7 @@
  "schema_version": 1,
  "providers": {
    "tushare": {"class": "licensed", "calculation_allowed": true},
+    "datahub": {"class": "licensed", "calculation_allowed": true},
    "ifind": {"class": "licensed", "calculation_allowed": true},
    "eastmoney": {"class": "public_web", "calculation_allowed": false},
    "tencent": {"class": "public_web", "calculation_allowed": false},
@@ -0,0 +1,18 @@
+{
+  "schema_version": 1,
+  "base_url": "http://127.0.0.1:8766",
+  "timeout_seconds": 8,
+  "retries": 1,
+  "page_limit": 5000,
+  "stale_seconds_max": 86400,
+  "datasets": {
+    "calendar": { "read": false, "shadow": false },
+    "stocks": { "read": false, "shadow": false },
+    "daily": { "read": false, "shadow": false },
+    "index_daily": { "read": false, "shadow": false },
+    "valuation": { "read": false, "shadow": false },
+    "moneyflow": { "read": false, "shadow": false },
+    "auction": { "read": false, "shadow": false },
+    "status": { "read": false, "shadow": false }
+  }
+}
@@ -32,4 +32,4 @@

 - 旧文档不能删：被替代的旧文档开头要加一行「⚠️ 本文档已过时，仅留档备查，请勿删除」，再写新版。
 - 用中文大白话写，专业词要带通俗解释，让不懂代码的人也能看懂。
- 「问天」板块是冻结区，任何改动都不许碰；写文档时别误导后来人去改它。
+- 「问天」不是永久冻结区：此前只冻结过界面视觉方案，现已解冻。问天可纳入后续数据与功能迁移，不要再写成“永远不碰”。
@@ -7,6 +7,7 @@
 | 任务 | 说明 | 状态 |
 |---|---|---|
 | 全站视觉统一改造收尾 | 主线。17 个阶段已完成，正在最终验收、代码合并 | 收尾中 |
+| 行情刷新误报与旧数据提示 | HEL-412：高级接口未到齐不再记整次失败；今日正式数据晚到时提示当前展示日期 | 施工中 |
 | 手机端独立重新设计 | 先出视觉/交互规范和技术架构方案，等老板确认后再施工 | 方案送审中 |

 ## 已做完
@@ -29,11 +29,11 @@
 - **智能工具类（3 个）**：智能选股、问师、问天。
 - **个人类（1 个）**：我的复盘。

-其中「问天」是冻结区（见下面的硬规矩）。
+其中「问天」此前只在全站视觉改造阶段冻结过界面方案，现已解冻；问天可以纳入后续数据与功能迁移，但不等于本阶段要重做视觉。

 ## 几条硬规矩（不能破坏的边界）

- 「问天」板块是**冻结区**，任何改动都不许碰它。
+- 「问天」板块**不是永久冻结区**：此前冻结的是界面视觉方案，现已解冻。问天现有功能与界面不要破坏；后续数据与功能迁移可以纳入，不主动重做视觉。
 - **不用假数据冒充真行情**；数据缺失就明说“没有/不可用”，不能编。
 - **每个用户自己的数据互相隔离**（自选、复盘、对话、问天历史等），看不到别人的。
 - **计算由程序确定性完成**（情绪周期、智能选股、问天排盘等），AI 大模型（LLM，就是会聊天的那个 AI）只负责解释或编译自然语言条件，不能改计算结果。
@@ -770,11 +770,33 @@
    scroll.classList.add("m-motion-fade-in");
  }

+  function dashboardFreshnessNotice() {
+    const meta = (state.dashboard && state.dashboard.meta) || {};
+    if (meta.display_notice) return String(meta.display_notice);
+    const requested = String(meta.requested_date || "").replace(/-/g, "");
+    const actual = String(meta.trade_date || "").replace(/-/g, "");
+    const compact = actual;
+    const shown = /^\d{8}$/.test(compact)
+      ? (Number(compact.slice(4, 6)) + " 月 " + Number(compact.slice(6, 8)) + " 日")
+      : "";
+    if (meta.data_status === "preparing" || (meta.carried_forward && actual && requested && actual !== requested)) {
+      return shown ? ("今日数据正在准备，当前展示 " + shown) : "今日数据正在准备，当前展示最近可用数据";
+    }
+    if (meta.data_status === "partial" || meta.limit_data_source === "derived") {
+      return meta.notice || "部分正式数据尚未到齐，当前展示日线推算结果";
+    }
+    return "";
+  }
+
  function renderTopArea(key) {
    const page = document.querySelector(".m-page");
    if (!page) return;
    let top = page.querySelector(".m-top");
    let html = buildStrip();
+    const freshness = dashboardFreshnessNotice();
+    if (freshness) {
+      html = '<div class="m-phase-notice"><strong>' + escapeHtml(freshness) + "</strong></div>" + html;
+    }
    if (key === "market/performance") html += performanceConclusion();
    if (!top) {
      top = document.createElement("div");
@@ -66,11 +66,13 @@ async function startAdminRefresh() {
    const requestedCompact = requestedDate.replaceAll("-", "");
    const actualCompact = actualDate.replaceAll("-", "");
    const updated = formatTimestamp(meta.updated_at);
-    if (actualCompact !== requestedCompact || meta.carried_forward) {
-      const reason = meta.notice ? `；${meta.notice}` : "";
-      setAdminRefreshStatus("warning", `刷新已完成，但没有获取到 ${requestedDate} 的最新行情；当前仍是 ${actualDate || "未知日期"}${reason}`, "triangle-alert");
-      showToast("刷新完成，但未获取到所选日期的最新行情");
-    } else if (meta.notice) {
+    const freshness = dashboardFreshnessMessage(meta);
+    if (freshness || actualCompact !== requestedCompact || meta.carried_forward || meta.limit_data_source === "derived") {
+      setAdminRefreshStatus("warning", freshness || `部分正式数据尚未到齐，当前展示 ${actualDate || "最近可用数据"}`, "triangle-alert");
+      setStatus(freshness || "部分正式数据尚未到齐，当前展示最近可用数据");
+      return;
+    }
+    if (meta.notice) {
      setAdminRefreshStatus("warning", `已刷新到 ${actualDate}（${updated}），但数据源提示：${meta.notice}`, "triangle-alert");
      showToast(`已刷新到 ${actualDate}，请留意数据源提示`);
    } else {
@@ -105,6 +107,37 @@ async function waitForAdminRefresh(jobKey) {
  throw new Error("刷新等待超时，请稍后重试");
 }

+let dashboardCatchupTimer = 0;
+
+function chineseMonthDay(value) {
+  const compact = String(value || "").replaceAll("-", "").replaceAll("/", "");
+  if (!/^\d{8}/.test(compact)) return "";
+  return `${Number(compact.slice(4, 6))} 月 ${Number(compact.slice(6, 8))} 日`;
+}
+
+function dashboardFreshnessMessage(meta = {}) {
+  if (meta.display_notice) return String(meta.display_notice);
+  const requested = String(meta.requested_date || "").replaceAll("-", "");
+  const actual = String(meta.trade_date || "").replaceAll("-", "");
+  const shown = chineseMonthDay(actual);
+  if (meta.data_status === "preparing" || (meta.carried_forward && actual && requested && actual !== requested)) {
+    return shown ? `今日数据正在准备，当前展示 ${shown}` : "今日数据正在准备，当前展示最近可用数据";
+  }
+  if (meta.data_status === "partial" || meta.limit_data_source === "derived") {
+    return meta.notice || "部分正式数据尚未到齐，当前展示日线推算结果";
+  }
+  return "";
+}
+
+function scheduleDashboardCatchup(meta = {}) {
+  window.clearTimeout(dashboardCatchupTimer);
+  const status = String(meta.data_status || "");
+  if (status !== "preparing" && status !== "partial") return;
+  dashboardCatchupTimer = window.setTimeout(() => {
+    loadDashboard(false, true, false);
+  }, 60000);
+}
+
 function applyDashboard(payload, background = false) {
  state.dashboard = payload;
  const selectedDate = payload.meta.requested_date || payload.meta.trade_date;
@@ -112,7 +145,11 @@ function applyDashboard(payload, background = false) {
  document.querySelector("#qiObservationDate").value = selectedDate;
  document.querySelector("#journalDate").value = selectedDate;
  renderDashboard();
-  setStatus(`${dashboardSourceLabel(payload.meta)} · 数据已更新`);
+  const freshness = dashboardFreshnessMessage(payload.meta || {});
+  setStatus(freshness || `${dashboardSourceLabel(payload.meta)} · 数据已更新`);
+  const updatedAt = document.querySelector("#updatedAt");
+  if (updatedAt) updatedAt.dataset.tone = freshness ? "warning" : "ok";
+  scheduleDashboardCatchup(payload.meta || {});
  if (!background) {
    if (state.activeView === "dragonView") loadDragonTiger();
    if (state.activeView === "screenerView") loadScreenerSetup();
@@ -180,7 +217,12 @@ function renderDashboard() {
    }
  }
  updateSentimentGauge(overview.sentiment_score);
-  setText("updatedAt", `${dashboardSourceLabel(meta)} · 更新 ${formatTimestamp(meta.updated_at)}`);
+  const freshness = dashboardFreshnessMessage(meta);
+  setText("updatedAt", freshness
+    ? freshness
+    : `${dashboardSourceLabel(meta)} · 更新 ${formatTimestamp(meta.updated_at)}`);
+  const updatedAt = document.querySelector("#updatedAt");
+  if (updatedAt) updatedAt.dataset.tone = freshness ? "warning" : "ok";

  renderLimitTable();
  renderLadderMini(ladders || []);
@@ -921,6 +921,10 @@ body.sidebar-collapsed .app-main {
  text-align: right;
 }

+.status-bar #updatedAt[data-tone="warning"] {
+  color: var(--warning);
+}
+
 .status-bar .risk-note {
  display: block;

@@ -1,23 +1,223 @@
 from __future__ import annotations

+import copy
+import threading
 import unittest
+from datetime import date, datetime, timedelta, timezone
+from pathlib import Path

-from backend.jobs.service import _verified_dashboard_result
+from backend.features.market.service import MarketServiceMixin
+from backend.jobs.refresh import (
+    dashboard_has_usable_data,
+    official_catchup_due,
+    verified_dashboard_result,
+)
+from backend.data.providers.tushare_transport import TushareError


 class AdminRefreshStatusTests(unittest.TestCase):
-    def test_carried_snapshot_is_reported_as_failed_job(self):
-        result = _verified_dashboard_result(
-            {"meta": {"carried_forward": True, "notice": "官方涨跌停数据尚未返回"}}
+    def test_carried_snapshot_is_usable_not_failed_job(self):
+        result = verified_dashboard_result(
+            {
+                "meta": {
+                    "trade_date": "2026-09-01",
+                    "requested_date": "2026-09-02",
+                    "carried_forward": True,
+                    "notice": "今日数据正在准备，当前展示 9 月 1 日",
+                    "data_status": "preparing",
+                },
+                "overview": {"limit_up_count": 12},
+            }
        )

-        self.assertEqual(result["status"], "failed")
-        self.assertEqual(result["error"], "官方涨跌停数据尚未返回")
+        self.assertNotEqual(result.get("status"), "failed")
+        self.assertEqual(result["meta"]["data_status"], "preparing")
+        self.assertTrue(dashboard_has_usable_data(result))
+
+    def test_derived_limit_snapshot_is_usable_not_failed_job(self):
+        dashboard = {
+            "meta": {
+                "trade_date": "2026-09-02",
+                "limit_data_source": "derived",
+                "notice": "涨跌停高级接口当日数据尚未更新，已使用日线数据推算。",
+                "data_status": "partial",
+            },
+            "overview": {"limit_up_count": 8},
+        }
+
+        self.assertIs(verified_dashboard_result(dashboard), dashboard)

    def test_current_snapshot_is_reported_as_successful_job(self):
        dashboard = {"meta": {"trade_date": "2026-08-28", "carried_forward": False}}

-        self.assertIs(_verified_dashboard_result(dashboard), dashboard)
+        self.assertIs(verified_dashboard_result(dashboard), dashboard)
+
+    def test_empty_payload_is_still_failed(self):
+        result = verified_dashboard_result({"meta": {}, "overview": {}})
+        self.assertEqual(result["status"], "failed")
+
+
+class FakeSyncDatabase:
+    def __init__(self, latest=None):
+        self.latest = latest
+        self.saved = []
+        self.finished = []
+
+    def start_sync(self, *_args, **_kwargs):
+        return 1
+
+    def save_snapshot(self, trade_date, source, payload):
+        self.saved.append((trade_date, source, copy.deepcopy(payload)))
+
+    def save_data_snapshot(self, *_args, **_kwargs):
+        return None
+
+    def finish_sync(self, *args, **kwargs):
+        self.finished.append((args, kwargs))
+
+    def get_latest_real_snapshot(self, *_args, **_kwargs):
+        return copy.deepcopy(self.latest)
+
+    def get_snapshot(self, *_args, **_kwargs):
+        return None
+
+    def get_data_snapshot(self, *_args, **_kwargs):
+        return None
+
+    def reason_overrides(self, *_args, **_kwargs):
+        return {}
+
+
+class FakeDerivedClient:
+    def dashboard(self, trade_date: str):
+        return {
+            "meta": {
+                "trade_date": f"{trade_date[:4]}-{trade_date[4:6]}-{trade_date[6:8]}",
+                "limit_data_source": "derived",
+                "notice": "涨跌停高级接口当日数据尚未更新，已使用日线数据推算。",
+                "updated_at": datetime.now().astimezone().isoformat(timespec="seconds"),
+            },
+            "overview": {"limit_up_count": 3},
+            "limits": [{"code": "000001"}],
+            "broken": [],
+            "down_limits": [],
+            "yesterday_limits": [],
+        }
+
+
+class FakeMissingDailyClient:
+    def dashboard(self, trade_date: str):
+        raise TushareError(f"No daily data returned for {trade_date}")
+
+
+class SyncHarness(MarketServiceMixin):
+    def __init__(self, client, latest=None):
+        self.configured = True
+        self.sync_lock = threading.Lock()
+        self.database = FakeSyncDatabase(latest)
+        self._client = client
+        self.current_user_id = 1
+
+    def _tushare_client(self):
+        return self._client
+
+    def _enrich_dashboard_sentiment(self, dashboard, _trade_date):
+        return dashboard
+
+    def _apply_reason_overrides(self, dashboard):
+        return dashboard
+
+
+class DashboardFreshnessTests(unittest.TestCase):
+    def test_derived_limits_are_kept_as_partial_success(self):
+        today = date.today().strftime("%Y%m%d")
+        harness = SyncHarness(FakeDerivedClient())
+        payload = harness.sync_dashboard(today)
+        meta = payload["meta"]
+
+        self.assertEqual(meta["limit_data_source"], "derived")
+        self.assertEqual(meta["data_status"], "partial")
+        self.assertFalse(meta.get("carried_forward"))
+        self.assertIn("日线数据推算", meta["display_notice"])
+        self.assertEqual(harness.database.finished[0][0][1], "success")
+        self.assertEqual(verified_dashboard_result(payload), payload)
+
+    def test_missing_official_data_keeps_previous_day_with_preparing_notice(self):
+        today = date.today()
+        previous = (today - timedelta(days=1)).strftime("%Y-%m-%d")
+        latest = {
+            "meta": {"trade_date": previous, "source": "tushare"},
+            "overview": {"limit_up_count": 20},
+        }
+        harness = SyncHarness(FakeMissingDailyClient(), latest)
+        payload = harness.sync_dashboard(today.strftime("%Y%m%d"))
+        meta = payload["meta"]
+
+        self.assertTrue(meta["carried_forward"])
+        self.assertEqual(meta["data_status"], "preparing")
+        self.assertIn("今日数据正在准备，当前展示", meta["display_notice"])
+        self.assertIn("月", meta["display_notice"])
+        self.assertNotIn("No daily data", meta["display_notice"])
+        self.assertNotEqual(verified_dashboard_result(payload).get("status"), "failed")
+
+    def test_weekend_carry_is_not_labeled_as_preparing(self):
+        snapshot = {
+            "meta": {"trade_date": "2026-07-24", "source": "tushare", "updated_at": "2026-07-24T15:00:00+08:00"},
+            "overview": {"limit_up_count": 1},
+        }
+        harness = SyncHarness(FakeMissingDailyClient())
+        carried = harness._carry_dashboard(snapshot, "20260725", "非交易日沿用最近交易日收盘行情")
+        self.assertEqual(carried["meta"]["data_status"], "carried")
+        self.assertIn("非交易日", carried["meta"]["display_notice"])
+
+    def test_stale_derived_snapshot_is_retried(self):
+        today = date.today().strftime("%Y%m%d")
+        old = datetime.now(timezone.utc) - timedelta(minutes=5)
+        snapshot = {
+            "meta": {
+                "source": "tushare",
+                "trade_date": f"{today[:4]}-{today[4:6]}-{today[6:8]}",
+                "limit_data_source": "derived",
+                "updated_at": old.isoformat(),
+            },
+            "overview": {"limit_up_count": 1},
+        }
+        harness = SyncHarness(FakeDerivedClient())
+        harness.database.get_snapshot = lambda *_args, **_kwargs: copy.deepcopy(snapshot)
+        payload = harness.get_dashboard(today)
+        self.assertEqual(payload["meta"]["data_status"], "partial")
+        self.assertTrue(harness.database.saved)
+
+    def test_official_catchup_skips_complete_today_snapshot(self):
+        today = date.today().strftime("%Y%m%d")
+        iso = f"{today[:4]}-{today[4:6]}-{today[6:8]}"
+        due = official_catchup_due(
+            today,
+            {"meta": {"trade_date": iso, "limit_data_source": "official"}},
+        )
+        derived_due = official_catchup_due(
+            today,
+            {"meta": {"trade_date": iso, "limit_data_source": "derived"}},
+        )
+        now = datetime.now().astimezone().time().replace(tzinfo=None)
+        if datetime.strptime("15:05", "%H:%M").time() <= now < datetime.strptime("22:00", "%H:%M").time():
+            self.assertFalse(due)
+            self.assertTrue(derived_due)
+        else:
+            self.assertFalse(due)
+            self.assertFalse(derived_due)
+
+
+class FrontendRefreshCopyTests(unittest.TestCase):
+    def test_dashboard_script_distinguishes_partial_from_failure(self):
+        script = (Path(__file__).resolve().parents[1] / "frontend" / "shared" / "dashboard.js").read_text(encoding="utf-8")
+        self.assertIn("今日数据正在准备，当前展示", script)
+        self.assertIn("部分正式数据尚未到齐", script)
+        self.assertIn('job.status === "failed"', script)
+        failed_block = script.split("if (job.status === \"failed\")", 1)[1].split("const query", 1)[0]
+        self.assertIn("后台刷新失败", failed_block)
+        success_block = script.split("const freshness = dashboardFreshnessMessage(meta);", 1)[1]
+        self.assertNotIn("后台刷新失败", success_block.split("} else {", 1)[0])


 if __name__ == "__main__":
@@ -0,0 +1,34 @@
+import logging
+import unittest
+
+from backend.bootstrap.runtime import configure_logging
+
+
+class ConfigureLoggingTest(unittest.TestCase):
+    def setUp(self) -> None:
+        self._saved_handlers = logging.getLogger().handlers[:]
+        self._saved_level = logging.getLogger().level
+        logging.getLogger().handlers.clear()
+
+    def tearDown(self) -> None:
+        logging.getLogger().handlers[:] = self._saved_handlers
+        logging.getLogger().setLevel(self._saved_level)
+
+    def test_configures_root_logger_at_info(self) -> None:
+        configure_logging()
+        root = logging.getLogger()
+        self.assertTrue(root.handlers)
+        self.assertEqual(root.level, logging.INFO)
+        with self.assertLogs("xiaobai.datahub", level="INFO") as captured:
+            logging.getLogger("xiaobai.datahub").info("datahub shadow %s", {"dataset": "daily"})
+        self.assertIn("datahub shadow", captured.output[0])
+
+    def test_keeps_existing_configuration(self) -> None:
+        handler = logging.NullHandler()
+        logging.getLogger().addHandler(handler)
+        configure_logging()
+        self.assertEqual(logging.getLogger().handlers, [handler])
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -67,6 +67,9 @@ class DataGatewayTests(unittest.TestCase):
            "TushareClient": {"backend/features/market/service.py"},
            "TushareProvider": {"backend/data/gateway.py"},
            "WebRealtimeAggregator": {"backend/data/gateway.py"},
+            "DatahubClient": {"backend/data/gateway.py"},
+            "DatahubAwareTushareClient": {"backend/data/gateway.py"},
+            "DatahubBridge": {"backend/data/gateway.py"},
        }
        found = {name: set() for name in owners}
        for path in (root / "backend").rglob("*.py"):
@@ -0,0 +1,287 @@
+from __future__ import annotations
+
+import ast
+import json
+import unittest
+from pathlib import Path
+from typing import Any
+
+from backend.data import build_data_gateway
+from backend.data.datahub.bridge import DatahubAwareTushareClient, DatahubBridge, looks_like_heaven
+from backend.data.datahub.client import DatahubClient, DatahubResponse
+from backend.data.datahub.compare import compare_rows
+from backend.data.datahub.errors import DatahubError
+from backend.data.datahub.native import to_canonical_row, to_native_row
+from backend.data.datahub.settings import DATASETS, DatahubSettings, DatasetFlags
+
+ROOT = Path(__file__).resolve().parents[1]
+TOKEN = "super-secret-datahub-token"
+
+LEGACY_DAILY = {
+    "ts_code": "600000.SH",
+    "trade_date": "20240902",
+    "open": 10.11,
+    "high": 10.25,
+    "low": 10.01,
+    "close": 10.20,
+    "pct_chg": 1.2345,
+    "vol": 1000.0,
+    "amount": 2000.0,
+}
+HUB_DAILY = {
+    "ts_code": "600000.SH",
+    "trade_date": "20240902",
+    "open": 10.11,
+    "high": 10.25,
+    "low": 10.01,
+    "close": 10.20,
+    "pct_chg": 1.2345,
+    "volume": 100000.0,
+    "amount": 2000000.0,
+}
+
+
+class FakeLegacy:
+    def __init__(self, rows: list[dict[str, Any]] | Exception | None = None) -> None:
+        self.token = "legacy-token"
+        self.timeout = 30
+        self.rows = [] if rows is None else rows
+        self.calls: list[tuple[str, dict[str, Any] | None, str]] = []
+
+    def query(self, api_name: str, params: dict[str, Any] | None = None, fields: str = "") -> list[dict[str, Any]]:
+        self.calls.append((api_name, params, fields))
+        if isinstance(self.rows, Exception):
+            raise self.rows
+        return [dict(row) for row in self.rows]
+
+
+class FakeClient(DatahubClient):
+    def __init__(self, error: DatahubError | None = None, response: DatahubResponse | None = None) -> None:
+        super().__init__(DatahubSettings(base_url="http://127.0.0.1:9", token=TOKEN))
+        self.error = error
+        self.response = response or DatahubResponse(
+            data=[dict(HUB_DAILY)],
+            meta={"tier": "official", "trade_date": "20240902", "stale": False, "staleness_seconds": 0},
+        )
+        self.paths: list[str] = []
+
+    def get(self, path: str, params: dict[str, Any] | None = None) -> DatahubResponse:
+        self.paths.append(path)
+        if TOKEN in json.dumps(params or {}) or TOKEN in path:
+            raise AssertionError("token leaked into url")
+        if self.error:
+            raise self.error
+        return self.response
+
+
+def flags(**enabled: tuple[bool, bool]) -> DatahubSettings:
+    datasets = {name: DatasetFlags(name) for name in DATASETS}
+    for name, pair in enabled.items():
+        datasets[name] = DatasetFlags(name, read=pair[0], shadow=pair[1])
+    return DatahubSettings(base_url="http://127.0.0.1:9", token=TOKEN, datasets=datasets)
+
+
+class DatahubBridgeTests(unittest.TestCase):
+    def test_default_config_keeps_legacy_and_does_not_call_datahub(self) -> None:
+        settings = DatahubSettings.load(environ={}, credentials={})
+        self.assertFalse(settings.any_enabled())
+        self.assertTrue(all(not settings.flags(name).read and not settings.flags(name).shadow for name in DATASETS))
+        client = FakeClient(error=DatahubError("INTERNAL", "should not be called"))
+        legacy = FakeLegacy([LEGACY_DAILY])
+        wrapped = DatahubAwareTushareClient(legacy, DatahubBridge(settings, client))
+        rows = wrapped.query("daily", {"trade_date": "20240902"}, "ts_code,close,vol,amount")
+        self.assertEqual(rows[0]["amount"], 2000.0)
+        self.assertEqual(client.paths, [])
+        self.assertEqual(len(legacy.calls), 1)
+
+    def test_each_dataset_has_independent_read_flag(self) -> None:
+        settings = flags(daily=(True, False), auction=(False, False))
+        self.assertTrue(settings.flags("daily").read)
+        self.assertFalse(settings.flags("auction").read)
+        self.assertFalse(any(settings.flags(name).read for name in DATASETS if name != "daily"))
+        source = (ROOT / "config" / "datahub.config.json").read_text(encoding="utf-8")
+        self.assertNotIn("master", source)
+        self.assertNotIn("DATAHUB_READ_ALL", source)
+
+    def test_read_flag_replaces_only_that_dataset_and_converts_units(self) -> None:
+        shadows: list[dict[str, Any]] = []
+        client = FakeClient()
+        legacy = FakeLegacy([LEGACY_DAILY])
+        wrapped = DatahubAwareTushareClient(
+            legacy,
+            DatahubBridge(flags(daily=(True, False)), client, shadow_sink=shadows.append),
+        )
+        rows = wrapped.query("daily", {"trade_date": "20240902"}, "ts_code,vol,amount")
+        self.assertEqual(rows[0]["vol"], 1000.0)
+        self.assertEqual(rows[0]["amount"], 2000.0)
+        self.assertEqual(legacy.calls, [])
+        self.assertEqual(client.paths, ["/v1/bars/daily"])
+        calendar_legacy = FakeLegacy([{"cal_date": "20240902", "is_open": 1}])
+        calendar_client = FakeClient(error=DatahubError("INTERNAL", "nope"))
+        calendar_wrapped = DatahubAwareTushareClient(
+            calendar_legacy,
+            DatahubBridge(flags(daily=(True, False)), calendar_client),
+        )
+        calendar = calendar_wrapped.query("trade_cal", {"start_date": "20240902", "end_date": "20240902"}, "")
+        self.assertEqual(calendar[0]["is_open"], 1)
+        self.assertEqual(calendar_client.paths, [])
+
+    def test_fallback_on_down_401_timeout_empty_unpublished_stale_and_incomplete(self) -> None:
+        cases = [
+            DatahubError("UNAVAILABLE", "down"),
+            DatahubError("UNAUTHORIZED", "401"),
+            DatahubError("TIMEOUT", "late"),
+            DatahubError("EMPTY", "no rows"),
+            DatahubError("DATASET_NOT_PUBLISHED", "not ready"),
+            DatahubError("STALE", "old"),
+            DatahubError("INCOMPLETE", "truncated"),
+        ]
+        for error in cases:
+            with self.subTest(error=error.code):
+                if error.code == "EMPTY":
+                    client = FakeClient(response=DatahubResponse(data=[], meta={"stale": False, "staleness_seconds": 0}))
+                elif error.code == "STALE":
+                    client = FakeClient(response=DatahubResponse(
+                        data=[dict(HUB_DAILY)],
+                        meta={"stale": True, "staleness_seconds": 999999},
+                    ))
+                elif error.code == "INCOMPLETE":
+                    client = FakeClient(response=DatahubResponse(
+                        data=[dict(HUB_DAILY)],
+                        meta={
+                            "stale": False,
+                            "staleness_seconds": 0,
+                            "incomplete": True,
+                            "coverage": {"complete": False, "missing_count": 80},
+                        },
+                    ))
+                else:
+                    client = FakeClient(error=error)
+                legacy = FakeLegacy([LEGACY_DAILY])
+                wrapped = DatahubAwareTushareClient(legacy, DatahubBridge(flags(daily=(True, False)), client))
+                rows = wrapped.query("daily", {"trade_date": "20240902"}, "ts_code,amount")
+                self.assertEqual(rows[0]["amount"], 2000.0)
+                self.assertEqual(len(legacy.calls), 1)
+
+    def test_shadow_compares_without_replacing_and_survives_hub_failure(self) -> None:
+        reports: list[dict[str, Any]] = []
+        client = FakeClient()
+        legacy = FakeLegacy([LEGACY_DAILY])
+        wrapped = DatahubAwareTushareClient(
+            legacy,
+            DatahubBridge(flags(daily=(False, True)), client, shadow_sink=reports.append),
+        )
+        rows = wrapped.query("daily", {"trade_date": "20240902"}, "ts_code,amount,vol")
+        self.assertEqual(rows[0]["amount"], 2000.0)
+        self.assertEqual(len(legacy.calls), 1)
+        self.assertEqual(reports[0]["equal"], True)
+        self.assertEqual(reports[0]["matched"], 1)
+
+        failed = FakeClient(error=DatahubError("UNAVAILABLE", TOKEN))
+        fail_reports: list[dict[str, Any]] = []
+        fail_legacy = FakeLegacy([LEGACY_DAILY])
+        fail_wrapped = DatahubAwareTushareClient(
+            fail_legacy,
+            DatahubBridge(flags(daily=(False, True)), failed, shadow_sink=fail_reports.append),
+        )
+        again = fail_wrapped.query("daily", {"trade_date": "20240902"}, "amount")
+        self.assertEqual(again[0]["amount"], 2000.0)
+        self.assertTrue(fail_reports[0]["hub_error"])
+        self.assertNotIn(TOKEN, json.dumps(fail_reports[0]))
+
+    def test_compare_classifies_unit_conversion_missing_row_and_value_diff(self) -> None:
+        equal = compare_rows("daily", [LEGACY_DAILY], [HUB_DAILY], {"stale": False, "staleness_seconds": 0})
+        self.assertTrue(equal["equal"])
+        unit = compare_rows("daily", [LEGACY_DAILY], [{**HUB_DAILY, "amount": 2000.0, "volume": 1000.0}])
+        self.assertGreater(unit["unit_conversion_count"], 0)
+        missing = compare_rows("daily", [LEGACY_DAILY], [])
+        self.assertEqual(missing["missing_hub_count"], 1)
+        value = compare_rows("daily", [LEGACY_DAILY], [{**HUB_DAILY, "close": 99.0}])
+        self.assertEqual(value["value_diff_count"], 1)
+        skew = compare_rows("daily", [LEGACY_DAILY], [HUB_DAILY], {"stale": False, "staleness_seconds": 12})
+        self.assertTrue(skew["time_skew"])
+
+    def test_native_roundtrip_matches_known_scales(self) -> None:
+        native = to_native_row("daily", HUB_DAILY)
+        self.assertEqual(native["vol"], 1000.0)
+        self.assertEqual(native["amount"], 2000.0)
+        canonical = to_canonical_row("daily", native)
+        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:
+        """问天未永久冻结；首批只读接入仍走旧链路，后续迁移可以纳入。"""
+        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()
+        legacy = FakeLegacy([LEGACY_DAILY])
+        wrapped = DatahubAwareTushareClient(
+            legacy,
+            DatahubBridge(flags(daily=(True, False)), client, heaven_guard=lambda: True),
+        )
+        rows = wrapped.query("daily", {"trade_date": "20240902"}, "amount")
+        self.assertEqual(rows[0]["amount"], 2000.0)
+        self.assertEqual(client.paths, [])
+
+    def test_status_flag_does_not_run_when_off_and_falls_back_when_on(self) -> None:
+        off = DatahubBridge(flags(), FakeClient(error=DatahubError("UNAVAILABLE", "down")))
+        self.assertIsNone(off.dataset_status("20240902"))
+        reports: list[dict[str, Any]] = []
+        failed = DatahubBridge(
+            flags(status=(True, True)),
+            FakeClient(error=DatahubError("UNAUTHORIZED", "nope")),
+            shadow_sink=reports.append,
+        )
+        self.assertIsNone(failed.dataset_status("20240902"))
+        self.assertTrue(reports[0]["hub_error"])
+        ok = DatahubBridge(
+            flags(status=(True, False)),
+            FakeClient(response=DatahubResponse(data=[{"dataset": "daily", "state": "published"}], meta={"stale": False, "staleness_seconds": 0})),
+        )
+        self.assertEqual(ok.dataset_status("20240902")[0]["state"], "published")
+
+    def test_default_gateway_wraps_tushare_without_calling_datahub(self) -> None:
+        gateway = build_data_gateway({}, datahub_settings=flags())
+        client = gateway.tushare()
+        self.assertIsInstance(client, DatahubAwareTushareClient)
+        self.assertFalse(gateway.datahub.settings.any_enabled())
+
+    def test_stock_detail_range_query_is_not_silently_accepted_when_incomplete(self) -> None:
+        source = (ROOT / "backend" / "data" / "providers" / "tushare_stocks.py").read_text(encoding="utf-8")
+        self.assertIn('"daily"', source)
+        self.assertIn("start_date", source)
+        self.assertIn("end_date", source)
+        client = FakeClient(
+            response=DatahubResponse(
+                data=[dict(HUB_DAILY)],
+                meta={"stale": False, "staleness_seconds": 0, "incomplete": True, "coverage": {"complete": False, "missing_count": 89}},
+            )
+        )
+        legacy = FakeLegacy([LEGACY_DAILY])
+        wrapped = DatahubAwareTushareClient(legacy, DatahubBridge(flags(daily=(True, False)), client))
+        rows = wrapped.query(
+            "daily",
+            {"ts_code": "600000.SH", "start_date": "20240301", "end_date": "20240902"},
+            "ts_code,amount",
+        )
+        self.assertEqual(rows[0]["amount"], 2000.0)
+        self.assertEqual(len(legacy.calls), 1)
+
+    def test_features_do_not_import_datahub_client(self) -> None:
+        violations = []
+        for path in (ROOT / "backend" / "features").rglob("*.py"):
+            tree = ast.parse(path.read_text(encoding="utf-8"), filename=str(path))
+            for node in ast.walk(tree):
+                names = []
+                if isinstance(node, ast.Import):
+                    names = [alias.name for alias in node.names]
+                elif isinstance(node, ast.ImportFrom) and node.module:
+                    names = [node.module]
+                for name in names:
+                    if "datahub" in name.split("."):
+                        violations.append(f"{path.relative_to(ROOT)} -> {name}")
+        self.assertEqual(violations, [])
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -0,0 +1,185 @@
+from __future__ import annotations
+
+import io
+import json
+import logging
+import threading
+import unittest
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from urllib.parse import parse_qs, urlparse
+
+from backend.data.datahub.client import DatahubClient
+from backend.data.datahub.errors import DatahubError
+from backend.data.datahub.redact import redact_text
+from backend.data.datahub.settings import DatahubSettings
+
+
+TOKEN = "super-secret-datahub-token"
+
+
+class FakeHubState:
+    def __init__(self) -> None:
+        self.mode = "ok"
+        self.hits = 0
+        self.paths: list[str] = []
+
+
+STATE = FakeHubState()
+
+
+class FakeHubHandler(BaseHTTPRequestHandler):
+    def log_message(self, format: str, *args: object) -> None:
+        return
+
+    def do_GET(self) -> None:  # noqa: N802
+        STATE.hits += 1
+        parsed = urlparse(self.path)
+        STATE.paths.append(parsed.path)
+        token = self.headers.get("X-Datahub-Token", "")
+        if STATE.mode == "timeout":
+            raise TimeoutError("simulated timeout")
+        if token != TOKEN:
+            self._json(401, {"error": {"code": "UNAUTHORIZED", "message": "missing or invalid X-Datahub-Token"}})
+            return
+        if STATE.mode == "unpublished":
+            self._json(404, {"error": {"code": "DATASET_NOT_PUBLISHED", "message": "daily 19990101 尚未发布", "expected_at": "15:05+08:00"}})
+            return
+        if STATE.mode == "empty":
+            self._json(200, {"schema_version": 1, "data": [], "meta": {"tier": "official", "stale": False, "staleness_seconds": 0}})
+            return
+        if STATE.mode == "stale":
+            self._json(200, {"schema_version": 1, "data": [{"ts_code": "600000.SH", "trade_date": "20240902", "close": 10.2, "volume": 100000, "amount": 2000000}], "meta": {"tier": "official", "stale": True, "staleness_seconds": 999999}})
+            return
+        if STATE.mode == "invalid":
+            self.send_response(200)
+            self.send_header("Content-Type", "application/json")
+            self.end_headers()
+            self.wfile.write(b"not-json")
+            return
+        if parsed.path == "/v1/health":
+            self._json(200, {"schema_version": 1, "data": {"status": "ok"}, "meta": {"tier": "official", "source": "datahub", "stale": False, "staleness_seconds": 0}})
+            return
+        if parsed.path == "/v1/calendar":
+            self._json(200, {"schema_version": 1, "data": [{"cal_date": "20240902", "is_open": True, "pretrade_date": "20240830"}], "meta": {"tier": "official", "trade_date": "20240902", "stale": False, "staleness_seconds": 0}})
+            return
+        if parsed.path == "/v1/bars/daily":
+            query = {key: values[-1] for key, values in parse_qs(parsed.query).items()}
+            self._json(200, {
+                "schema_version": 1,
+                "data": [{
+                    "ts_code": "600000.SH",
+                    "trade_date": query.get("date") or "20240902",
+                    "open": 10.11, "high": 10.25, "low": 10.01, "close": 10.20,
+                    "pct_chg": 1.2345, "volume": 100000.0, "amount": 2000000.0, "adj_factor": 1.1,
+                }],
+                "meta": {"tier": "official", "trade_date": "20240902", "stale": False, "staleness_seconds": 0, "source": "tushare:daily"},
+            })
+            return
+        if parsed.path == "/v1/datasets/status":
+            self._json(200, {"schema_version": 1, "data": [{"dataset": "daily", "state": "published", "trade_date": "20240902"}], "meta": {"tier": "official", "stale": False, "staleness_seconds": 0}})
+            return
+        self._json(400, {"error": {"code": "INVALID_ARGUMENT", "message": f"unknown endpoint: {parsed.path}"}})
+
+    def _json(self, status: int, payload: dict) -> None:
+        body = json.dumps(payload).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(body)))
+        self.end_headers()
+        self.wfile.write(body)
+
+
+class DatahubClientTests(unittest.TestCase):
+    @classmethod
+    def setUpClass(cls) -> None:
+        cls.server = ThreadingHTTPServer(("127.0.0.1", 0), FakeHubHandler)
+        cls.thread = threading.Thread(target=cls.server.serve_forever, daemon=True)
+        cls.thread.start()
+        cls.base = f"http://127.0.0.1:{cls.server.server_address[1]}"
+
+    @classmethod
+    def tearDownClass(cls) -> None:
+        cls.server.shutdown()
+        cls.server.server_close()
+
+    def setUp(self) -> None:
+        STATE.mode = "ok"
+        STATE.hits = 0
+        STATE.paths = []
+        self.client = DatahubClient(DatahubSettings(base_url=self.base, token=TOKEN, retries=1, timeout_seconds=2))
+
+    def test_health_envelope(self) -> None:
+        response = self.client.health()
+        self.assertEqual(response.schema_version, 1)
+        self.assertEqual(response.data["status"], "ok")
+        self.assertIn("stale", response.meta)
+
+    def test_missing_and_bad_token_401(self) -> None:
+        missing = DatahubClient(DatahubSettings(base_url=self.base, token=""))
+        with self.assertRaises(DatahubError) as raised:
+            missing.health()
+        self.assertEqual(raised.exception.code, "NOT_CONFIGURED")
+        bad = DatahubClient(DatahubSettings(base_url=self.base, token="wrong"))
+        with self.assertRaises(DatahubError) as raised:
+            bad.health()
+        self.assertEqual(raised.exception.code, "UNAUTHORIZED")
+        self.assertNotIn(TOKEN, str(raised.exception))
+
+    def test_unpublished_and_empty_and_stale_codes(self) -> None:
+        STATE.mode = "unpublished"
+        with self.assertRaises(DatahubError) as raised:
+            self.client.daily_bars(date="19990101")
+        self.assertEqual(raised.exception.code, "DATASET_NOT_PUBLISHED")
+        STATE.mode = "empty"
+        response = self.client.daily_bars(date="20240902")
+        self.assertEqual(response.data, [])
+        STATE.mode = "stale"
+        stale = self.client.daily_bars(date="20240902")
+        self.assertTrue(stale.meta["stale"])
+
+    def test_invalid_json_maps_to_internal(self) -> None:
+        STATE.mode = "invalid"
+        with self.assertRaises(DatahubError) as raised:
+            self.client.health()
+        self.assertEqual(raised.exception.code, "INTERNAL")
+
+    def test_timeout_maps_and_retries(self) -> None:
+        hits = {"n": 0}
+
+        def boom(_request, timeout=None):
+            hits["n"] += 1
+            raise TimeoutError("late")
+
+        client = DatahubClient(
+            DatahubSettings(base_url=self.base, token=TOKEN, retries=1, timeout_seconds=1),
+            urlopen=boom,
+        )
+        with self.assertRaises(DatahubError) as raised:
+            client.health()
+        self.assertEqual(raised.exception.code, "TIMEOUT")
+        self.assertEqual(hits["n"], 2)
+
+    def test_token_never_appears_in_error_text_or_logs(self) -> None:
+        stream = io.StringIO()
+        logger = logging.getLogger("xiaobai.datahub")
+        handler = logging.StreamHandler(stream)
+        logger.addHandler(handler)
+        logger.setLevel(logging.DEBUG)
+        try:
+            with self.assertRaises(DatahubError):
+                DatahubClient(DatahubSettings(base_url=self.base, token="wrong")).health()
+            blob = stream.getvalue() + redact_text("header " + TOKEN, (TOKEN,))
+            self.assertNotIn(TOKEN, blob)
+            self.assertIn("***", redact_text(TOKEN, (TOKEN,)))
+        finally:
+            logger.removeHandler(handler)
+
+    def test_calendar_and_status_contract(self) -> None:
+        calendar = self.client.calendar("20240901", "20240902")
+        self.assertEqual(calendar.data[0]["cal_date"], "20240902")
+        status = self.client.dataset_status("20240902")
+        self.assertEqual(status.data[0]["dataset"], "daily")
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -97,6 +97,7 @@ def code_hotspots() -> list[dict[str, Any]]:
        "backend/features/system/service.py",
        "backend/features/accounts/application.py",
        "backend/jobs/service.py",
+        "backend/jobs/refresh.py",
        "database.py",
        "backend/features/screener/engine.py",
        "backend/features/screener/catalog.py",
@@ -217,6 +218,7 @@ def build() -> dict[str, Any]:
        ),
        "external_data_adapters": [
            {"provider": "tushare", "path": "backend/data/providers/tushare_client.py", "runtime_role": "stable client facade for primary deterministic market data"},
+            {"provider": "datahub", "path": "backend/data/datahub/client.py", "runtime_role": "optional official EOD read path behind per-dataset flags"},
            {"provider": "ifind", "path": "backend/data/providers/ifind_client.py", "runtime_role": "realtime, charts, snapshots, enrichment"},
            {"provider": "eastmoney", "path": "backend/features/market/charts.py", "runtime_role": "display chart fallback"},
            {"provider": "eastmoney", "path": "backend/data/realtime.py", "runtime_role": "isolated realtime observation"},
@@ -235,6 +237,9 @@ def build() -> dict[str, Any]:
        ],
        "provider_construction": [
            {"client": "TushareClient", "owner": "backend/data/providers/tushare.py", "compatibility_fallback": "backend/features/market/service.py"},
+            {"client": "DatahubClient", "owner": "backend/data/gateway.py"},
+            {"client": "DatahubBridge", "owner": "backend/data/gateway.py"},
+            {"client": "DatahubAwareTushareClient", "owner": "backend/data/gateway.py"},
            {"client": "IfindHttpClient", "owner": "backend/data/gateway.py"},
            {"client": "MarketChartClient", "owner": "backend/data/gateway.py"},
            {"client": "WebRealtimeAggregator", "owner": "backend/data/gateway.py"},
@@ -261,6 +266,7 @@ def build() -> dict[str, Any]:
            "system_service": "backend/features/system/service.py",
            "account_bridge": "backend/features/accounts/application.py",
            "job_lifecycle": "backend/jobs/service.py",
+            "job_refresh_status": "backend/jobs/refresh.py",
            "feature_routes": "backend/features/*/routes.py",
        },
        "numeric_normalization": [
@@ -0,0 +1,10 @@
+.git
+.gitignore
+.env
+.env.*
+!.env.example
+__pycache__/
+*.py[cod]
+*.log
+data/
+tests/
@@ -0,0 +1,13 @@
+# Fernet key. Generate with: python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
+DATAHUB_ENCRYPTION_KEY=
+
+# Consumer API token for /v1 (32+ random bytes, shown once). Never log this value.
+DATAHUB_TOKEN=
+
+# Initial admin password for /admin. Forced change on first login.
+DATAHUB_ADMIN_PASSWORD=
+
+# Tushare Pro token. Stored encrypted after first launch; never returned by API or admin pages.
+TUSHARE_TOKEN=
+
+TZ=Asia/Shanghai
@@ -0,0 +1,36 @@
+FROM python:3.12-slim-bookworm
+
+ARG APP_UID=10002
+ARG APP_GID=10002
+
+ENV PYTHONDONTWRITEBYTECODE=1 \
+    PYTHONUNBUFFERED=1 \
+    PYTHONUTF8=1 \
+    PIP_DISABLE_PIP_VERSION_CHECK=1 \
+    TZ=Asia/Shanghai
+
+WORKDIR /app
+
+RUN apt-get update \
+    && DEBIAN_FRONTEND=noninteractive apt-get install -y --no-install-recommends \
+        ca-certificates \
+        tzdata \
+    && groupadd --gid "${APP_GID}" datahub \
+    && useradd --uid "${APP_UID}" --gid "${APP_GID}" --create-home --shell /usr/sbin/nologin datahub \
+    && rm -rf /var/lib/apt/lists/*
+
+COPY requirements.txt ./
+RUN python -m pip install --no-cache-dir -r requirements.txt
+
+COPY --chown=datahub:datahub . .
+RUN mkdir -p /app/data /app/data/backups && chown -R datahub:datahub /app/data
+
+USER datahub
+
+EXPOSE 8766
+STOPSIGNAL SIGINT
+
+HEALTHCHECK --interval=30s --timeout=5s --start-period=20s --retries=3 \
+    CMD ["python", "-c", "import urllib.request; urllib.request.urlopen('http://127.0.0.1:8766/livez', timeout=4).read()"]
+
+CMD ["python", "-u", "server.py", "--host", "0.0.0.0", "--port", "8766"]
@@ -0,0 +1,95 @@
+# xiaobai-datahub
+
+独立行情数据中枢（HEL-382 / P0）。与 `xiaobai-review` 同仓库、不同容器、不共享数据库文件。
+本阶段不部署现网；只提供可本地运行、可自测的底座和盘后正式数据链路。
+
+## 做什么
+
+- SQLite WAL `datahub.db`，容器名 `xiaobai-datahub`，端口 `8766`
+- Tushare 盘后正式数据：交易日历、股票主档、daily、daily_basic、adj_factor、index_daily、moneyflow、stk_auction
+- 暂存 → 校验 → 整批原子发布 → 可回滚
+- `/v1` 稳定接口（`X-Datahub-Token`）
+- `/admin/` 最小管理后台（总览 / 数据源 / 调度 / 发布 / 数据集 / 审计）
+- 东财/腾讯/同花顺/选股宝/AKShare/iFinD 适配器位已预留，本阶段不拉实时源
+
+## 单位口径（相对现站）
+
+现站 `xiaobai-review` 按 Tushare 原始单位入库、展示时再换算。中枢在归一化层一次换算：
+
+| 字段 | Tushare / 现站 | 中枢 canonical |
+|---|---|---|
+| `daily.amount` / `index_daily.amount` | 千元 | 元（×1000） |
+| `daily.vol` / `index_daily.vol` | 手 | 股（×100） |
+| `moneyflow.*_amount` | 万元 | 元（×1e4） |
+| `daily_basic.total_mv` / `circ_mv` | 万元 | 元（×1e4） |
+| `stk_auction.amount` | 元 | 元 |
+
+差异为口径升级，golden 测试按上表对照，不为 0 的字段都有说明。
+
+## 本地启动（不走 Docker）
+
+```bash
+cd xiaobai-datahub
+python -m venv .venv && .venv/bin/pip install -r requirements.txt
+cp .env.example .env
+# 填入 DATAHUB_ENCRYPTION_KEY / DATAHUB_TOKEN / DATAHUB_ADMIN_PASSWORD / TUSHARE_TOKEN
+# 生成 Fernet 密钥：
+# python -c "from cryptography.fernet import Fernet; print(Fernet.generate_key().decode())"
+.venv/bin/python server.py --host 127.0.0.1 --port 8766
+```
+
+- 管理后台：http://127.0.0.1:8766/admin/
+- 存活检查：http://127.0.0.1:8766/livez （无需 token）
+- `/v1/*` 必须带请求头 `X-Datahub-Token`
+
+## Docker（独立 compose，不改现网 review 服务）
+
+```bash
+cd xiaobai-datahub
+cp .env.example .env   # 填密钥
+mkdir -p data
+docker compose build
+docker compose up -d
+```
+
+仓库根目录另有 `compose.datahub.yaml`，供总工以后与现有 `compose.yaml` 叠加部署，本卡不执行现网 `up`。
+
+## 自测
+
+```bash
+cd xiaobai-datahub
+python -m unittest discover -s tests -v
+```
+
+不调用真实 Tushare；用内存/临时库和假适配器。
+
+## 历史回补
+
+交易日历默认从 `20160101` 拉到今天后 30 天；盘前 `precheck` 与手动回补都走同一 UPSERT，可重复执行。
+
+网站实际使用的指数（上证、深成、创业板、沪深300）按交易日增量发布，默认覆盖 260 个交易日（大于现有 90 天窗口，并覆盖智能选股基准回看）。已发布日期默认跳过。
+
+```bash
+cd xiaobai-datahub
+python -m datahub history-backfill
+# 可选：--calendar-start 20160101 --index-days 260 --force
+```
+
+管理后台也可手动跑 `history_backfill` 任务，或 `POST /admin/api/backfill` 且 `dataset=history`、确认词 `history:full`。
+
+区间接口在 `meta.coverage` / `meta.incomplete` 标明覆盖是否完整；网站只读接入把不完整区间视为不可用并回旧链路。个股日 K 的 90 天区间查询依赖已核实，本阶段不回补全市场历史。
+
+## 备份
+
+每日 00:40 任务把 `datahub.db` 备份到 `data/backups/`（保留 14 份）。也可手动：
+
+```bash
+python -c "from pathlib import Path; from datahub.db import HubDB; HubDB(Path('data/datahub.db')).backup_to(Path('data/backups/manual.db'))"
+```
+
+## 安全
+
+- 密钥只以 `configured / 末4位 / 更新时间` 出现在后台，不进日志、不进 `/v1`
+- HTTP 解析失败只记录“请求不是合法 JSON”，不把请求正文、密码或 Token 写入容器日志
+- 回滚、补数需重新输入密码 + 确认词
+- 容器非 root（uid 10002）、read_only、cap_drop ALL
@@ -0,0 +1,268 @@
+const state = { csrf: "", page: "overview" };
+
+function $(id) { return document.getElementById(id); }
+
+async function api(path, options = {}) {
+  const headers = Object.assign({ "Content-Type": "application/json" }, options.headers || {});
+  if (state.csrf && (options.method || "GET") !== "GET") headers["X-CSRF-Token"] = state.csrf;
+  const res = await fetch(path, Object.assign({}, options, { headers, credentials: "same-origin" }));
+  const body = await res.json();
+  if (!res.ok) {
+    const msg = (body.error && body.error.message) || body.error || res.statusText;
+    throw new Error(msg);
+  }
+  return body;
+}
+
+function show(id) {
+  ["login-view", "change-view", "shell"].forEach((key) => { $(key).hidden = key !== id; });
+}
+
+function esc(value) {
+  return String(value ?? "").replace(/[&<>"]/g, (ch) => ({ "&": "&amp;", "<": "&lt;", ">": "&gt;", '"': "&quot;" }[ch]));
+}
+
+function table(headers, rows) {
+  const thead = headers.map((h) => `<th>${esc(h)}</th>`).join("");
+  const body = rows.length
+    ? rows.map((cols) => `<tr>${cols.map((c) => `<td>${c}</td>`).join("")}</tr>`).join("")
+    : `<tr><td colspan="${headers.length}">暂无数据</td></tr>`;
+  return `<table><thead><tr>${thead}</tr></thead><tbody>${body}</tbody></table>`;
+}
+
+async function boot() {
+  try {
+    const session = await api("/admin/api/session");
+    state.csrf = session.csrf;
+    $("who").textContent = session.username;
+    if (session.must_change) { show("change-view"); return; }
+    show("shell");
+    await render();
+  } catch {
+    show("login-view");
+  }
+}
+
+$("login-form").addEventListener("submit", async (event) => {
+  event.preventDefault();
+  const form = new FormData(event.target);
+  $("login-error").hidden = true;
+  try {
+    const result = await api("/admin/api/login", {
+      method: "POST",
+      body: JSON.stringify({ username: form.get("username"), password: form.get("password") }),
+    });
+    state.csrf = result.csrf;
+    if (result.must_change) show("change-view");
+    else { show("shell"); await render(); }
+  } catch (err) {
+    $("login-error").hidden = false;
+    $("login-error").textContent = err.message;
+  }
+});
+
+$("change-form").addEventListener("submit", async (event) => {
+  event.preventDefault();
+  const form = new FormData(event.target);
+  try {
+    await api("/admin/api/change-password", {
+      method: "POST",
+      body: JSON.stringify({ current: form.get("current"), new_password: form.get("new_password") }),
+    });
+    show("shell");
+    await render();
+  } catch (err) {
+    $("change-error").hidden = false;
+    $("change-error").textContent = err.message;
+  }
+});
+
+$("logout-btn").addEventListener("click", async () => {
+  await api("/admin/api/logout", { method: "POST", body: "{}" });
+  show("login-view");
+});
+
+$("theme-btn").addEventListener("click", () => {
+  const root = document.documentElement;
+  const next = root.getAttribute("data-theme") === "night" ? "" : "night";
+  if (next) root.setAttribute("data-theme", next);
+  else root.removeAttribute("data-theme");
+  $("theme-btn").textContent = next ? "日间" : "夜间";
+});
+
+document.querySelectorAll("nav button").forEach((btn) => {
+  btn.addEventListener("click", () => {
+    document.querySelectorAll("nav button").forEach((item) => item.classList.remove("active"));
+    btn.classList.add("active");
+    state.page = btn.dataset.page;
+    render();
+  });
+});
+
+async function render() {
+  const page = $("page");
+  if (state.page === "overview") {
+    const data = await api("/admin/api/overview");
+    $("phase").textContent = data.session_phase;
+    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 class="${data.anomalies.length ? "fail" : "ok"}">${data.anomalies.length}</strong></div>
+      </div>
+      <h2>最近调用</h2>
+      ${table(["时间", "源", "端点", "结果", "耗时"], data.recent_calls.map((row) => [
+        esc(row.created_at), esc(row.provider), esc(row.endpoint),
+        row.ok ? '<span class="ok">成功</span>' : `<span class="fail">${esc(row.error)}</span>`,
+        `${row.latency_ms ?? "-"} ms`,
+      ]))}
+    `;
+    return;
+  }
+  if (state.page === "sources") {
+    const data = await api("/admin/api/sources");
+    page.innerHTML = `<h2>数据源</h2>` + table(
+      ["源", "角色", "状态", "凭据", "操作"],
+      data.items.map((item) => {
+        const cred = item.credential || {};
+        const credText = cred.configured ? `已配置 · ${esc(cred.last4 || "****")}` : "未配置";
+        return [
+          esc(item.provider),
+          esc(item.role),
+          esc((item.health && (item.health.state || item.health.status)) || "-"),
+          credText,
+          `<button data-probe="${esc(item.provider)}">探测一次</button>`,
+        ];
+      }),
+    );
+    page.querySelectorAll("[data-probe]").forEach((btn) => {
+      btn.addEventListener("click", async () => {
+        const result = await api(`/admin/api/sources/${btn.dataset.probe}/probe`, { method: "POST", body: "{}" });
+        alert(JSON.stringify(result));
+        render();
+      });
+    });
+    return;
+  }
+  if (state.page === "jobs") {
+    const data = await api("/admin/api/jobs");
+    page.innerHTML = `
+      <h2>调度任务</h2>
+      ${table(["任务", "时刻", "操作"], data.jobs.map((job) => [
+        `${esc(job.id)} · ${esc(job.title)}`, esc(job.at),
+        `<button data-run="${esc(job.id)}">手动触发</button>`,
+      ]))}
+      <h3>最近运行</h3>
+      ${table(["ID", "任务", "状态", "开始", "结束", "错误"], data.runs.map((row) => [
+        row.id, esc(row.job_id), esc(row.state), esc(row.started_at), esc(row.finished_at), esc(row.error),
+      ]))}
+    `;
+    page.querySelectorAll("[data-run]").forEach((btn) => {
+      btn.addEventListener("click", async () => {
+        const date = prompt("交易日 YYYYMMDD（可留空=今天）", "") || "";
+        await api(`/admin/api/jobs/${btn.dataset.run}/run`, { method: "POST", body: JSON.stringify({ trade_date: date }) });
+        render();
+      });
+    });
+    return;
+  }
+  if (state.page === "release") {
+    const date = new Date().toISOString().slice(0, 10).replace(/-/g, "");
+    const data = await api(`/admin/api/batches?date=${date}`);
+    page.innerHTML = `
+      <h2>盘后发布 ${esc(data.trade_date)}</h2>
+      <div class="toolbar">
+        <label>日期 <input id="rel-date" value="${esc(data.trade_date)}" /></label>
+        <button type="button" id="rel-load">查看</button>
+        <button type="button" id="rel-backfill">补数</button>
+      </div>
+      <h3>当前映射</h3>
+      ${table(["数据集", "活跃批次", "上一批次", "状态", "发布时间", "操作"], data.publications.map((row) => [
+        esc(row.dataset), esc(row.active_batch), esc(row.prev_batch), esc(row.state), esc(row.published_at),
+        row.prev_batch ? `<button class="danger" data-rollback="${esc(row.dataset)}">回滚</button>` : "-",
+      ]))}
+      <h3>批次</h3>
+      ${table(["batch_id", "数据集", "状态", "行数", "错误"], data.batches.map((row) => [
+        esc(row.batch_id), esc(row.dataset), esc(row.state), row.rows_out ?? "", esc(row.error),
+      ]))}
+    `;
+    $bindRelease(page);
+    return;
+  }
+  if (state.page === "datasets") {
+    const data = await api("/admin/api/datasets?date=");
+    page.innerHTML = `
+      <h2>数据集 / 质量 ${esc(data.trade_date)}</h2>
+      ${table(["数据集", "批次", "状态", "发布时间"], data.publications.map((row) => [
+        esc(row.dataset), esc(row.active_batch), esc(row.state), esc(row.published_at),
+      ]))}
+      <h3>源间差异</h3>
+      ${table(["指标", "左", "右", "偏差", "样本"], data.diff_reports.map((row) => [
+        esc(row.metric), esc(row.left_value), esc(row.right_value), esc(row.deviation), row.sample_count ?? "",
+      ]))}
+    `;
+    return;
+  }
+  if (state.page === "audit") {
+    const data = await api("/admin/api/audit");
+    page.innerHTML = `<h2>审计</h2>` + table(
+      ["时间", "操作者", "动作", "对象", "详情"],
+      data.items.map((row) => [esc(row.created_at), esc(row.actor), esc(row.action), esc(row.target), esc(row.detail)]),
+    );
+  }
+}
+
+function $bindRelease(page) {
+  page.querySelector("#rel-load").addEventListener("click", async () => {
+    const date = page.querySelector("#rel-date").value;
+    const data = await api(`/admin/api/batches?date=${encodeURIComponent(date)}`);
+    state.page = "release";
+    // re-render with fetched date by writing location hash
+    history.replaceState(null, "", `#release-${date}`);
+    $("page").innerHTML = renderRelease(data);
+    $bindRelease($("page"));
+  });
+  page.querySelector("#rel-backfill").addEventListener("click", () => dangerous("backfill"));
+  page.querySelectorAll("[data-rollback]").forEach((btn) => {
+    btn.addEventListener("click", () => dangerous("rollback", btn.dataset.rollback));
+  });
+}
+
+function renderRelease(data) {
+  return `
+    <h2>盘后发布 ${esc(data.trade_date)}</h2>
+    <div class="toolbar">
+      <label>日期 <input id="rel-date" value="${esc(data.trade_date)}" /></label>
+      <button type="button" id="rel-load">查看</button>
+      <button type="button" id="rel-backfill">补数</button>
+    </div>
+    <h3>当前映射</h3>
+    ${table(["数据集", "活跃批次", "上一批次", "状态", "发布时间", "操作"], data.publications.map((row) => [
+      esc(row.dataset), esc(row.active_batch), esc(row.prev_batch), esc(row.state), esc(row.published_at),
+      row.prev_batch ? `<button class="danger" data-rollback="${esc(row.dataset)}">回滚</button>` : "-",
+    ]))}
+    <h3>批次</h3>
+    ${table(["batch_id", "数据集", "状态", "行数", "错误"], data.batches.map((row) => [
+      esc(row.batch_id), esc(row.dataset), esc(row.state), row.rows_out ?? "", esc(row.error),
+    ]))}
+  `;
+}
+
+async function dangerous(kind, dataset) {
+  const date = ($("rel-date") && $("rel-date").value) || "";
+  const ds = dataset || prompt("数据集（daily / valuation / moneyflow / auction / index_daily / reference）", "daily");
+  if (!ds) return;
+  const password = prompt("二次确认：输入管理密码");
+  if (!password) return;
+  const confirmWord = `${ds}:${date}`;
+  const typed = prompt(`请输入确认词：${confirmWord}`);
+  const path = kind === "rollback" ? "/admin/api/rollback" : "/admin/api/backfill";
+  await api(path, {
+    method: "POST",
+    body: JSON.stringify({ dataset: ds, trade_date: date, password, confirm: typed }),
+  });
+  render();
+}
+
+boot();
@@ -0,0 +1,53 @@
+<!DOCTYPE html>
+<html lang="zh-CN">
+<head>
+  <meta charset="UTF-8" />
+  <meta name="viewport" content="width=device-width, initial-scale=1" />
+  <title>xiaobai-datahub 管理后台</title>
+  <link rel="stylesheet" href="/admin/styles.css" />
+</head>
+<body>
+  <div id="app">
+    <section id="login-view" class="panel auth-panel">
+      <h1>数据中枢</h1>
+      <p class="muted">内网管理后台，用于查看源状态、调度和盘后发布批次。</p>
+      <form id="login-form">
+        <label>账号 <input name="username" value="hub_admin" autocomplete="username" /></label>
+        <label>密码 <input name="password" type="password" autocomplete="current-password" /></label>
+        <button type="submit">登录</button>
+        <p id="login-error" class="error" hidden></p>
+      </form>
+    </section>
+
+    <section id="change-view" class="panel auth-panel" hidden>
+      <h1>修改初始密码</h1>
+      <form id="change-form">
+        <label>当前密码 <input name="current" type="password" /></label>
+        <label>新密码（至少 8 位） <input name="new_password" type="password" /></label>
+        <button type="submit">保存并继续</button>
+        <p id="change-error" class="error" hidden></p>
+      </form>
+    </section>
+
+    <section id="shell" hidden>
+      <header class="top">
+        <strong>xiaobai-datahub</strong>
+        <span id="phase" class="pill"></span>
+        <span id="who" class="muted"></span>
+        <button type="button" id="theme-btn" class="ghost">夜间</button>
+        <button type="button" id="logout-btn" class="ghost">退出</button>
+      </header>
+      <nav>
+        <button data-page="overview" class="active">总览</button>
+        <button data-page="sources">数据源</button>
+        <button data-page="jobs">调度任务</button>
+        <button data-page="release">盘后发布</button>
+        <button data-page="datasets">数据集</button>
+        <button data-page="audit">审计</button>
+      </nav>
+      <main id="page"></main>
+    </section>
+  </div>
+  <script src="/admin/app.js"></script>
+</body>
+</html>
@@ -0,0 +1,51 @@
+:root {
+  color-scheme: light;
+  --bg: #f4f5f7;
+  --surface: #ffffff;
+  --text: #1f2329;
+  --muted: #646a73;
+  --line: #dee0e3;
+  --action: #3370ff;
+  --danger: #e04536;
+  --ok: #16a34a;
+  --warn: #b45309;
+  --radius: 8px;
+  --pad: 16px;
+  font-family: "Segoe UI", "PingFang SC", "Noto Sans SC", sans-serif;
+}
+:root[data-theme="night"] {
+  color-scheme: dark;
+  --bg: #111318;
+  --surface: #1b1e24;
+  --text: #e8eaed;
+  --muted: #9aa0a6;
+  --line: #2a2f38;
+  --action: #5b8cff;
+}
+* { box-sizing: border-box; }
+body { margin: 0; background: var(--bg); color: var(--text); }
+.panel, header.top, nav, main { background: var(--surface); }
+.auth-panel { max-width: 420px; margin: 12vh auto; padding: 28px; border-radius: var(--radius); border: 1px solid var(--line); }
+label { display: block; margin: 12px 0; }
+input, select { width: 100%; padding: 8px 10px; border: 1px solid var(--line); border-radius: 4px; background: var(--bg); color: var(--text); }
+button { background: var(--action); color: #fff; border: 0; border-radius: 4px; padding: 8px 14px; cursor: pointer; }
+button.ghost { background: transparent; color: var(--text); border: 1px solid var(--line); }
+button.danger { background: var(--danger); }
+.muted { color: var(--muted); }
+.error { color: var(--danger); }
+.top { display: flex; gap: 12px; align-items: center; padding: 10px var(--pad); border-bottom: 1px solid var(--line); }
+nav { display: flex; gap: 4px; padding: 8px var(--pad); border-bottom: 1px solid var(--line); }
+nav button { background: transparent; color: var(--muted); }
+nav button.active { color: var(--action); background: transparent; font-weight: 600; }
+main { padding: var(--pad); min-height: calc(100vh - 96px); }
+.cards { display: grid; grid-template-columns: repeat(auto-fit, minmax(180px, 1fr)); gap: 12px; margin-bottom: 16px; }
+.card { border: 1px solid var(--line); border-radius: var(--radius); padding: 12px; }
+table { width: 100%; border-collapse: collapse; font-size: 13px; }
+th, td { text-align: left; padding: 8px; border-bottom: 1px solid var(--line); vertical-align: top; }
+.pill { font-size: 12px; padding: 2px 8px; border-radius: 999px; border: 1px solid var(--line); }
+.ok { color: var(--ok); }
+.warn { color: var(--warn); }
+.fail { color: var(--danger); }
+.toolbar { display: flex; gap: 8px; flex-wrap: wrap; margin: 12px 0; align-items: end; }
+.toolbar label { margin: 0; }
+dialog { border: 1px solid var(--line); border-radius: var(--radius); background: var(--surface); color: var(--text); padding: 20px; }
@@ -0,0 +1,39 @@
+services:
+  xiaobai-datahub:
+    build:
+      context: .
+      dockerfile: Dockerfile
+    image: xiaobai-datahub:local
+    container_name: xiaobai-datahub
+    ports:
+      - "0.0.0.0:8766:8766/tcp"
+    env_file:
+      - ./.env
+    environment:
+      DATAHUB_ENCRYPTION_KEY: "${DATAHUB_ENCRYPTION_KEY:?DATAHUB_ENCRYPTION_KEY must be set}"
+      DATAHUB_TOKEN: "${DATAHUB_TOKEN:?DATAHUB_TOKEN must be set}"
+      DATAHUB_ADMIN_PASSWORD: "${DATAHUB_ADMIN_PASSWORD:?DATAHUB_ADMIN_PASSWORD must be set}"
+      TUSHARE_TOKEN: "${TUSHARE_TOKEN:-}"
+      DATAHUB_DB_PATH: /app/data/datahub.db
+      DATAHUB_BACKUP_DIR: /app/data/backups
+      TZ: Asia/Shanghai
+      PYTHONUTF8: "1"
+    volumes:
+      - type: bind
+        source: ./data
+        target: /app/data
+    restart: unless-stopped
+    init: true
+    read_only: true
+    tmpfs:
+      - /tmp:size=64m,mode=1777
+    security_opt:
+      - no-new-privileges:true
+    cap_drop:
+      - ALL
+    stop_grace_period: 30s
+    logging:
+      driver: json-file
+      options:
+        max-size: "10m"
+        max-file: "3"
@@ -0,0 +1,17 @@
+{
+  "daily_row_ratio": 0.98,
+  "null_rate_max": 0.01,
+  "cross_check_price_deviation": 0.03,
+  "cross_check_outlier_ratio": 0.05,
+  "index_price_deviation": 0.005,
+  "max_publish_attempts": 5,
+  "staging_retain_days": 14,
+  "job_run_retain_days": 90,
+  "backup_retain": 14,
+  "publication_generations": 3,
+  "tushare_rate_per_minute": 300,
+  "list_limit_default": 5000,
+  "list_limit_max": 5000,
+  "calendar_start": "20160101",
+  "index_history_trading_days": 260
+}
@@ -0,0 +1,4 @@
+"""xiaobai-datahub: independent market-data service for xiaobai-review."""
+
+__version__ = "0.1.0"
+SCHEMA_VERSION = 1
@@ -0,0 +1,4 @@
+from datahub.cli import main
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,15 @@
+from datahub.adapters.akshare import ADAPTER as akshare
+from datahub.adapters.eastmoney import ADAPTER as eastmoney
+from datahub.adapters.ifind import ADAPTER as ifind
+from datahub.adapters.tencent import ADAPTER as tencent
+from datahub.adapters.ths import ADAPTER as ths
+from datahub.adapters.xgb import ADAPTER as xgb
+
+RESERVED = {
+    "eastmoney": eastmoney,
+    "tencent": tencent,
+    "ths": ths,
+    "xgb": xgb,
+    "akshare": akshare,
+    "ifind": ifind,
+}
@@ -0,0 +1,3 @@
+from datahub.adapters.base import ReservedAdapter
+
+ADAPTER = ReservedAdapter("akshare")
@@ -0,0 +1,47 @@
+from __future__ import annotations
+
+from abc import ABC, abstractmethod
+from typing import Any
+
+
+class AdapterError(RuntimeError):
+    pass
+
+
+class MarketAdapter(ABC):
+    """Uniform adapter: probe / fetch / normalize. Realtime adapters may be stubs in P0."""
+
+    name: str = "base"
+
+    @abstractmethod
+    def probe(self) -> dict[str, Any]:
+        """Liveness check. Must not leak credentials."""
+
+    @abstractmethod
+    def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
+        """Return provider-native rows (pre-canonical)."""
+
+    @abstractmethod
+    def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        """Map provider-native rows onto hub canonical fields."""
+
+
+class ReservedAdapter(MarketAdapter):
+    """Placeholder for a later free/licensed source. Does not pull data in P0."""
+
+    def __init__(self, name: str) -> None:
+        self.name = name
+
+    def probe(self) -> dict[str, Any]:
+        return {
+            "provider": self.name,
+            "configured": False,
+            "state": "reserved",
+            "message": "适配器位已预留，本阶段不接入",
+        }
+
+    def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
+        raise AdapterError(f"{self.name} 适配器本阶段未接入")
+
+    def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        return []
@@ -0,0 +1,3 @@
+from datahub.adapters.base import ReservedAdapter
+
+ADAPTER = ReservedAdapter("eastmoney")
@@ -0,0 +1,3 @@
+from datahub.adapters.base import ReservedAdapter
+
+ADAPTER = ReservedAdapter("ifind")
@@ -0,0 +1,3 @@
+from datahub.adapters.base import ReservedAdapter
+
+ADAPTER = ReservedAdapter("tencent")
@@ -0,0 +1,3 @@
+from datahub.adapters.base import ReservedAdapter
+
+ADAPTER = ReservedAdapter("ths")
@@ -0,0 +1,159 @@
+from __future__ import annotations
+
+import json
+import time
+import urllib.error
+import urllib.request
+from typing import Any, Callable
+
+from datahub.adapters.base import AdapterError, MarketAdapter
+from datahub.normalize import (
+    normalize_auction,
+    normalize_calendar,
+    normalize_daily,
+    normalize_index_daily,
+    normalize_moneyflow,
+    normalize_stock,
+    normalize_valuation,
+)
+
+TUSHARE_URL = "http://api.tushare.pro"
+
+TUSHARE_FIELDS = {
+    "trade_cal": "exchange,cal_date,is_open,pretrade_date",
+    "stock_basic": "ts_code,symbol,name,area,industry,market,list_status,list_date",
+    "daily": "ts_code,trade_date,open,high,low,close,pct_chg,vol,amount",
+    "daily_basic": "ts_code,trade_date,turnover_rate,volume_ratio,total_mv,circ_mv,pe_ttm,pb,ps_ttm,dv_ttm",
+    "adj_factor": "ts_code,trade_date,adj_factor",
+    "index_daily": "ts_code,trade_date,open,high,low,close,pct_chg,vol,amount",
+    "moneyflow": (
+        "ts_code,trade_date,buy_sm_amount,sell_sm_amount,buy_md_amount,sell_md_amount,"
+        "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",
+}
+
+DATASET_API = {
+    "calendar": "trade_cal",
+    "stocks": "stock_basic",
+    "daily": "daily",
+    "valuation": "daily_basic",
+    "adj_factor": "adj_factor",
+    "index_daily": "index_daily",
+    "moneyflow": "moneyflow",
+    "auction": "stk_auction",
+}
+
+# Website actual index usage: market cards / 90-day charts (SH/SZ/CYB) plus
+# screener 沪深300 benchmark (lookback up to 260 trading days).
+WEBSITE_INDEX_CODES = ("000001.SH", "399001.SZ", "399006.SZ", "000300.SH")
+DEFAULT_INDEX_CODES = WEBSITE_INDEX_CODES
+
+
+class TushareAdapter(MarketAdapter):
+    name = "tushare"
+
+    def __init__(
+        self,
+        token: str,
+        timeout: int = 30,
+        transport: Callable[[str, dict[str, Any], str], list[dict[str, Any]]] | None = None,
+    ) -> None:
+        self.token = token
+        self.timeout = timeout
+        self._transport = transport
+
+    def probe(self) -> dict[str, Any]:
+        if not self.token:
+            return {"provider": self.name, "configured": False, "state": "unconfigured"}
+        started = time.perf_counter()
+        try:
+            rows = self.fetch("calendar", {"exchange": "SSE", "start_date": "20200102", "end_date": "20200102"})
+        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": "ok" if rows else "empty",
+            "latency_ms": round((time.perf_counter() - started) * 1000),
+        }
+
+    def fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
+        api_name = DATASET_API.get(dataset, dataset)
+        fields = TUSHARE_FIELDS.get(api_name, "")
+        query_params = dict(params)
+        if api_name == "stock_basic" and "list_status" not in query_params:
+            query_params["list_status"] = "L"
+        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_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:
+            rows.extend(self.fetch("index_daily", {"ts_code": ts_code, "trade_date": trade_date}))
+        return rows
+
+    def normalize(self, dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+        mapping = {
+            "calendar": normalize_calendar,
+            "trade_cal": normalize_calendar,
+            "stocks": normalize_stock,
+            "stock_basic": normalize_stock,
+            "daily": normalize_daily,
+            "valuation": normalize_valuation,
+            "daily_basic": normalize_valuation,
+            "moneyflow": normalize_moneyflow,
+            "auction": normalize_auction,
+            "stk_auction": normalize_auction,
+            "index_daily": normalize_index_daily,
+        }
+        fn = mapping.get(dataset)
+        if fn is None:
+            if dataset == "adj_factor":
+                return [
+                    {
+                        "ts_code": str(row.get("ts_code") or "").upper(),
+                        "trade_date": str(row.get("trade_date") or ""),
+                        "adj_factor": row.get("adj_factor"),
+                    }
+                    for row in rows
+                ]
+            raise AdapterError(f"unsupported dataset: {dataset}")
+        return [fn(row) for row in rows]
+
+    def _query(self, api_name: str, params: dict[str, Any], fields: str) -> list[dict[str, Any]]:
+        if self._transport is not None:
+            return self._transport(api_name, params, fields)
+        if not self.token:
+            raise AdapterError("Tushare token 未配置")
+        payload = json.dumps(
+            {"api_name": api_name, "token": self.token, "params": params, "fields": fields}
+        ).encode("utf-8")
+        request = urllib.request.Request(
+            TUSHARE_URL,
+            data=payload,
+            headers={"Content-Type": "application/json", "User-Agent": "XiaobaiDatahub/0.1"},
+            method="POST",
+        )
+        try:
+            with urllib.request.urlopen(request, timeout=self.timeout) as response:
+                result = json.loads(response.read().decode("utf-8"))
+        except json.JSONDecodeError:
+            raise AdapterError("Tushare returned invalid json") from None
+        except (urllib.error.URLError, TimeoutError) as exc:
+            raise AdapterError(f"Tushare request failed: {exc}") from exc
+        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 []
+        return [dict(zip(columns, item)) for item in data.get("items") or []]
@@ -0,0 +1,3 @@
+from datahub.adapters.base import ReservedAdapter
+
+ADAPTER = ReservedAdapter("xgb")
@@ -0,0 +1,168 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from datahub.adapters import RESERVED
+from datahub.auth import AuthService
+from datahub.db import HubDB
+from datahub.pipeline import Pipeline
+from datahub.scheduler import Scheduler
+from datahub.serving import ApiError
+from datahub.timeutil import isoformat, now_shanghai, session_phase, yyyymmdd
+
+
+class AdminAPI:
+    def __init__(self, db: HubDB, pipeline: Pipeline, scheduler: Scheduler, auth: AuthService) -> None:
+        self.db = db
+        self.pipeline = pipeline
+        self.scheduler = scheduler
+        self.auth = auth
+
+    def overview(self) -> dict[str, Any]:
+        today = yyyymmdd(now_shanghai())
+        cal = self.db.fetchone(
+            "SELECT is_open FROM trade_calendar WHERE exchange = 'SSE' AND cal_date = ?",
+            (today,),
+        )
+        is_open = bool(cal and int(cal["is_open"]) == 1)
+        pubs = self.db.fetchall("SELECT * FROM publications WHERE trade_date = ?", (today,))
+        failed = self.db.fetchall(
+            "SELECT * FROM batches WHERE trade_date = ? AND state IN ('failed','staged')",
+            (today,),
+        )
+        calls = self.db.fetchall(
+            "SELECT * FROM src_calls ORDER BY id DESC LIMIT 20",
+        )
+        return {
+            "trade_date": today,
+            "session_phase": session_phase(now_shanghai(), is_open),
+            "is_open_day": is_open,
+            "publications": pubs,
+            "anomalies": failed,
+            "recent_calls": _public_calls(calls),
+            "source_count": len(self.db.fetchall("SELECT provider FROM src_health")),
+        }
+
+    def sources(self) -> dict[str, Any]:
+        health = {f"{row['provider']}:{row['endpoint_class']}": row for row in self.db.fetchall("SELECT * FROM src_health")}
+        items = [
+            {
+                "provider": "tushare",
+                "role": "official",
+                "health": health.get("tushare:pro") or {"state": "unknown"},
+                "credential": self.auth.credential_status("tushare_token") or {"configured": bool(self.pipeline.adapter.token)},
+            }
+        ]
+        for name, adapter in RESERVED.items():
+            items.append(
+                {
+                    "provider": name,
+                    "role": "reserved",
+                    "health": adapter.probe(),
+                    "credential": {"configured": False, "last4": "", "updated_at": ""},
+                }
+            )
+        # Prefer encrypted last4 if stored
+        cred = self.auth.credential_status("tushare_token")
+        if cred.get("configured"):
+            items[0]["credential"] = cred
+        elif self.pipeline.adapter.token:
+            from datahub.crypto import mask_secret
+
+            items[0]["credential"] = {"configured": True, "last4": mask_secret(self.pipeline.adapter.token), "updated_at": ""}
+        return {"items": items}
+
+    def probe(self, provider: str) -> dict[str, Any]:
+        if provider == "tushare":
+            return self.pipeline.adapter.probe()
+        adapter = RESERVED.get(provider)
+        if adapter is None:
+            raise ApiError("INVALID_ARGUMENT", f"unknown provider: {provider}")
+        return adapter.probe()
+
+    def jobs(self) -> dict[str, Any]:
+        runs = self.db.fetchall("SELECT * FROM job_runs ORDER BY id DESC LIMIT 100")
+        return {
+            "jobs": [
+                {"id": "precheck", "at": "08:45", "title": "盘前预检"},
+                {"id": "eod_a", "at": "15:05", "title": "盘后批 A daily/valuation/moneyflow/auction"},
+                {"id": "eod_b", "at": "15:10", "title": "盘后批 B index_daily"},
+                {"id": "history_backfill", "at": "manual", "title": "回补历史日历与指数日 K"},
+                {"id": "cleanup", "at": "00:30", "title": "清理 staging / 日志"},
+                {"id": "backup", "at": "00:40", "title": "SQLite 备份"},
+            ],
+            "runs": runs,
+        }
+
+    def run_job(self, job_id: str, trade_date: str) -> dict[str, Any]:
+        return self.scheduler.run_job(job_id, yyyymmdd(trade_date or now_shanghai()))
+
+    def batches(self, date: str, dataset: str = "") -> dict[str, Any]:
+        trade_date = yyyymmdd(date or now_shanghai())
+        if dataset:
+            rows = self.db.fetchall(
+                "SELECT * FROM batches WHERE trade_date = ? AND dataset = ? ORDER BY started_at",
+                (trade_date, dataset),
+            )
+        else:
+            rows = self.db.fetchall(
+                "SELECT * FROM batches WHERE trade_date = ? ORDER BY started_at",
+                (trade_date,),
+            )
+        pubs = self.db.fetchall("SELECT * FROM publications WHERE trade_date = ?", (trade_date,))
+        return {"trade_date": trade_date, "batches": rows, "publications": pubs}
+
+    def datasets(self, date: str) -> dict[str, Any]:
+        trade_date = yyyymmdd(date or now_shanghai())
+        pubs = self.db.fetchall("SELECT * FROM publications WHERE trade_date = ?", (trade_date,))
+        diffs = self.db.fetchall(
+            "SELECT * FROM diff_reports WHERE trade_date = ? ORDER BY id",
+            (trade_date,),
+        )
+        return {"trade_date": trade_date, "publications": pubs, "diff_reports": diffs}
+
+    def audit(self) -> dict[str, Any]:
+        return {"items": self.db.fetchall("SELECT * FROM audit_log ORDER BY id DESC LIMIT 200")}
+
+    def rollback(self, dataset: str, trade_date: str, password: str, confirm: str, actor: str) -> dict[str, Any]:
+        self._dangerous(password, confirm, f"{dataset}:{trade_date}")
+        result = self.pipeline.rollback(dataset, trade_date, actor=actor)
+        return result
+
+    def backfill(self, dataset: str, trade_date: str, password: str, confirm: str, actor: str) -> dict[str, Any]:
+        day = yyyymmdd(trade_date or now_shanghai())
+        if dataset == "history":
+            self._dangerous(password, confirm, "history:full")
+            result = self.pipeline.backfill_history(day)
+        else:
+            self._dangerous(password, confirm, f"{dataset}:{day}")
+            if dataset == "reference":
+                result = self.pipeline.ingest_reference(day)
+            else:
+                result = self.pipeline.run_dataset(dataset, day)
+        self.pipeline.audit(actor, "backfill", f"{dataset}:{day}", json.dumps({"ok": True}))
+        return result
+
+    def _dangerous(self, password: str, confirm: str, expected: str) -> None:
+        if not self.auth.confirm_password(password):
+            raise ApiError("UNAUTHORIZED", "二次确认密码错误")
+        if confirm.strip() != expected:
+            raise ApiError("INVALID_ARGUMENT", f"确认词必须为 {expected}")
+
+
+def _public_calls(rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    out = []
+    for row in rows:
+        out.append(
+            {
+                "id": row["id"],
+                "provider": row["provider"],
+                "endpoint": row["endpoint"],
+                "ok": bool(row["ok"]),
+                "latency_ms": row["latency_ms"],
+                "error": row["error"],
+                "created_at": row["created_at"],
+            }
+        )
+    return out
@@ -0,0 +1,190 @@
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+import os
+import secrets
+from datetime import timedelta
+from typing import Any
+
+from datahub.crypto import SecretVault, mask_secret
+from datahub.db import HubDB
+from datahub.timeutil import isoformat, now_shanghai
+
+PBKDF2_ROUNDS = 200_000
+SESSION_HOURS = 12
+LOGIN_FAIL_LIMIT = 5
+LOCK_MINUTES = 10
+
+
+def hash_password(password: str, salt: bytes | None = None) -> tuple[str, str]:
+    raw_salt = salt or os.urandom(16)
+    digest = hashlib.pbkdf2_hmac("sha256", password.encode("utf-8"), raw_salt, PBKDF2_ROUNDS, dklen=32)
+    return (
+        base64.urlsafe_b64encode(raw_salt).decode("ascii"),
+        base64.urlsafe_b64encode(digest).decode("ascii"),
+    )
+
+
+def verify_password(password: str, salt_text: str, expected_hash: str) -> bool:
+    try:
+        salt = base64.urlsafe_b64decode(salt_text.encode("ascii"))
+        _, actual = hash_password(password, salt)
+    except (ValueError, TypeError):
+        return False
+    return hmac.compare_digest(actual, expected_hash)
+
+
+def token_hash(token: str) -> str:
+    return hashlib.sha256(token.encode("utf-8")).hexdigest()
+
+
+class AuthService:
+    def __init__(self, db: HubDB, vault: SecretVault, api_token: str, admin_password: str) -> None:
+        self.db = db
+        self.vault = vault
+        self._bootstrap(api_token, admin_password)
+
+    def _bootstrap(self, api_token: str, admin_password: str) -> None:
+        if api_token:
+            existing = self.db.fetchone("SELECT token_hash FROM api_tokens WHERE name = ?", ("review",))
+            hashed = token_hash(api_token)
+            last4 = mask_secret(api_token)
+            if existing is None:
+                self.db.execute(
+                    "INSERT INTO api_tokens(token_hash, name, last4, created_at) VALUES (?,?,?,?)",
+                    (hashed, "review", last4, isoformat()),
+                )
+            elif existing["token_hash"] != hashed:
+                self.db.execute(
+                    "UPDATE api_tokens SET token_hash = ?, last4 = ? WHERE name = ?",
+                    (hashed, last4, "review"),
+                )
+        admin = self.db.fetchone("SELECT id FROM hub_admin WHERE username = ?", ("hub_admin",))
+        if admin is None and admin_password:
+            salt, hashed = hash_password(admin_password)
+            now = isoformat()
+            self.db.execute(
+                """
+                INSERT INTO hub_admin(username, password_salt, password_hash, password_must_change, created_at, updated_at)
+                VALUES (?, ?, ?, 1, ?, ?)
+                """,
+                ("hub_admin", salt, hashed, now, now),
+            )
+
+    def check_api_token(self, supplied: str) -> bool:
+        if not supplied:
+            return False
+        row = self.db.fetchone(
+            "SELECT token_hash FROM api_tokens WHERE token_hash = ? AND revoked_at IS NULL",
+            (token_hash(supplied),),
+        )
+        return row is not None
+
+    def login(self, username: str, password: str) -> dict[str, Any]:
+        user = self.db.fetchone("SELECT * FROM hub_admin WHERE username = ?", (username,))
+        if not user:
+            raise PermissionError("账号或密码错误")
+        now = now_shanghai()
+        locked_until = user.get("locked_until")
+        if locked_until:
+            try:
+                from datetime import datetime
+
+                if datetime.fromisoformat(str(locked_until)) > now:
+                    raise PermissionError("账号已锁定，请稍后再试")
+            except ValueError:
+                pass
+        if not verify_password(password, str(user["password_salt"]), str(user["password_hash"])):
+            fails = int(user["failed_attempts"] or 0) + 1
+            lock = isoformat(now + timedelta(minutes=LOCK_MINUTES)) if fails >= LOGIN_FAIL_LIMIT else None
+            self.db.execute(
+                "UPDATE hub_admin SET failed_attempts = ?, locked_until = ? WHERE id = ?",
+                (fails, lock, user["id"]),
+            )
+            raise PermissionError("账号或密码错误")
+        self.db.execute(
+            "UPDATE hub_admin SET failed_attempts = 0, locked_until = NULL WHERE id = ?",
+            (user["id"],),
+        )
+        session = secrets.token_urlsafe(32)
+        csrf = secrets.token_urlsafe(24)
+        expires = isoformat(now + timedelta(hours=SESSION_HOURS))
+        self.db.execute(
+            "INSERT INTO hub_sessions(token_hash, csrf_token, expires_at, created_at) VALUES (?,?,?,?)",
+            (token_hash(session), csrf, expires, isoformat(now)),
+        )
+        return {
+            "session": session,
+            "csrf": csrf,
+            "must_change": bool(user["password_must_change"]),
+            "expires_at": expires,
+        }
+
+    def session_user(self, raw_token: str) -> dict[str, Any] | None:
+        if not raw_token:
+            return None
+        row = self.db.fetchone(
+            "SELECT * FROM hub_sessions WHERE token_hash = ?",
+            (token_hash(raw_token),),
+        )
+        if not row:
+            return None
+        if str(row["expires_at"]) < isoformat():
+            self.db.execute("DELETE FROM hub_sessions WHERE token_hash = ?", (row["token_hash"],))
+            return None
+        admin = self.db.fetchone("SELECT username, password_must_change FROM hub_admin WHERE username = ?", ("hub_admin",))
+        return {
+            "username": (admin or {}).get("username") or "hub_admin",
+            "csrf_token": row["csrf_token"],
+            "must_change": bool((admin or {}).get("password_must_change")),
+            "token_hash": row["token_hash"],
+        }
+
+    def logout(self, raw_token: str) -> None:
+        if raw_token:
+            self.db.execute("DELETE FROM hub_sessions WHERE token_hash = ?", (token_hash(raw_token),))
+
+    def change_password(self, current: str, new_password: str) -> None:
+        if len(new_password) < 8:
+            raise ValueError("新密码至少 8 位")
+        user = self.db.fetchone("SELECT * FROM hub_admin WHERE username = ?", ("hub_admin",))
+        if not user or not verify_password(current, str(user["password_salt"]), str(user["password_hash"])):
+            raise PermissionError("当前密码错误")
+        salt, hashed = hash_password(new_password)
+        self.db.execute(
+            "UPDATE hub_admin SET password_salt=?, password_hash=?, password_must_change=0, updated_at=? WHERE id=?",
+            (salt, hashed, isoformat(), user["id"]),
+        )
+
+    def confirm_password(self, password: str) -> bool:
+        user = self.db.fetchone("SELECT * FROM hub_admin WHERE username = ?", ("hub_admin",))
+        if not user:
+            return False
+        return verify_password(password, str(user["password_salt"]), str(user["password_hash"]))
+
+    def credential_status(self, name: str) -> dict[str, Any]:
+        row = self.db.fetchone("SELECT last4, updated_at FROM credentials WHERE name = ?", (name,))
+        if not row:
+            return {"configured": False, "last4": "", "updated_at": ""}
+        return {"configured": True, "last4": row["last4"], "updated_at": row["updated_at"]}
+
+    def store_credential(self, name: str, secret: str) -> None:
+        payload = self.vault.encrypt_json({name: secret})
+        self.db.execute(
+            """
+            INSERT INTO credentials(name, encrypted_payload, last4, updated_at)
+            VALUES (?, ?, ?, ?)
+            ON CONFLICT(name) DO UPDATE SET
+              encrypted_payload=excluded.encrypted_payload, last4=excluded.last4, updated_at=excluded.updated_at
+            """,
+            (name, payload, mask_secret(secret), isoformat()),
+        )
+
+    def load_credential(self, name: str) -> str:
+        row = self.db.fetchone("SELECT encrypted_payload FROM credentials WHERE name = ?", (name,))
+        if not row:
+            return ""
+        data = self.vault.decrypt_json(str(row["encrypted_payload"]))
+        return str(data.get(name) or "")
@@ -0,0 +1,38 @@
+"""Command-line entry for one-shot datahub operations."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import sys
+
+from datahub.hub import build_hub
+from datahub.settings import load_settings
+
+
+def main(argv: list[str] | None = None) -> int:
+    parser = argparse.ArgumentParser(description="xiaobai-datahub CLI")
+    sub = parser.add_subparsers(dest="command", required=True)
+    history = sub.add_parser("history-backfill", help="回补 2016 年起交易日历和网站所用指数日 K")
+    history.add_argument("--calendar-start", default=None, help="日历起点，默认配置 calendar_start")
+    history.add_argument("--index-days", type=int, default=None, help="指数回补交易日数量，默认 260")
+    history.add_argument("--force", action="store_true", help="覆盖已发布的指数日期")
+    args = parser.parse_args(argv)
+
+    settings = load_settings()
+    hub = build_hub(settings)
+    if args.command == "history-backfill":
+        result = hub.pipeline.backfill_history(
+            calendar_start=args.calendar_start,
+            index_days=args.index_days,
+            force=args.force,
+        )
+        json.dump(result, sys.stdout, ensure_ascii=False, indent=2, default=str)
+        sys.stdout.write("\n")
+        return 0 if result.get("ok") else 1
+    parser.error(f"unknown command: {args.command}")
+    return 2
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())
@@ -0,0 +1,26 @@
+from __future__ import annotations
+
+from datahub.db import HubDB
+
+
+def resolve_code(db: HubDB, raw: str) -> str | None:
+    text = str(raw or "").strip().upper()
+    if not text:
+        return None
+    if "." in text:
+        row = db.fetchone("SELECT ts_code FROM stock_master WHERE ts_code = ?", (text,))
+        if row:
+            return row["ts_code"]
+        # indices are not always in stock_master
+        return text
+    matches = db.fetchall(
+        "SELECT ts_code FROM stock_master WHERE symbol = ? OR ts_code LIKE ?",
+        (text, f"{text}.%"),
+    )
+    if len(matches) == 1:
+        return matches[0]["ts_code"]
+    if len(matches) > 1:
+        return None
+    # unique exchange guess for 6-digit codes
+    suffix = "SH" if text.startswith("6") or text.startswith("9") else "SZ" if text.startswith(("0", "3")) else "BJ"
+    return f"{text}.{suffix}"
@@ -0,0 +1,130 @@
+from __future__ import annotations
+
+from typing import Any, Iterable
+
+from datahub.db import HubDB
+from datahub.timeutil import iter_yyyymmdd, yyyymmdd
+
+MISSING_SAMPLE_LIMIT = 10
+
+
+def coverage_payload(
+    *,
+    kind: str,
+    start: str,
+    end: str,
+    expected: Iterable[str],
+    available: Iterable[str],
+    extra: dict[str, Any] | None = None,
+) -> dict[str, Any]:
+    start = yyyymmdd(start)
+    end = yyyymmdd(end)
+    expected_list = sorted({yyyymmdd(item) for item in expected if item})
+    available_set = {yyyymmdd(item) for item in available if item}
+    missing = [item for item in expected_list if item not in available_set]
+    payload: dict[str, Any] = {
+        "kind": kind,
+        "complete": not missing,
+        "requested_from": start,
+        "requested_to": end,
+        "available_from": min(available_set) if available_set else None,
+        "available_to": max(available_set) if available_set else None,
+        "expected_count": len(expected_list),
+        "available_count": len(available_set),
+        "missing_count": len(missing),
+        "missing_sample": missing[:MISSING_SAMPLE_LIMIT],
+    }
+    if extra:
+        payload.update(extra)
+    return payload
+
+
+def calendar_coverage(db: HubDB, start: str, end: str, exchange: str = "SSE") -> dict[str, Any]:
+    start = yyyymmdd(start)
+    end = yyyymmdd(end)
+    expected = list(iter_yyyymmdd(start, end))
+    rows = db.fetchall(
+        "SELECT cal_date FROM trade_calendar WHERE exchange = ? AND cal_date >= ? AND cal_date <= ?",
+        (exchange, start, end),
+    )
+    return coverage_payload(
+        kind="calendar",
+        start=start,
+        end=end,
+        expected=expected,
+        available=(row["cal_date"] for row in rows),
+        extra={"exchange": exchange},
+    )
+
+
+def published_range_coverage(
+    db: HubDB,
+    dataset: str,
+    start: str,
+    end: str,
+    ts_code: str = "",
+    table: str = "",
+) -> dict[str, Any]:
+    start = yyyymmdd(start)
+    end = yyyymmdd(end)
+    calendar = calendar_coverage(db, start, end)
+    open_rows = db.fetchall(
+        """
+        SELECT cal_date FROM trade_calendar
+        WHERE exchange = 'SSE' AND is_open = 1 AND cal_date >= ? AND cal_date <= ?
+        ORDER BY cal_date
+        """,
+        (start, end),
+    )
+    expected_open = [row["cal_date"] for row in open_rows]
+    pubs = db.fetchall(
+        """
+        SELECT trade_date, active_batch FROM publications
+        WHERE dataset = ? AND trade_date >= ? AND trade_date <= ?
+        ORDER BY trade_date
+        """,
+        (dataset, start, end),
+    )
+    published_dates = [row["trade_date"] for row in pubs]
+    available = list(published_dates)
+    extra: dict[str, Any] = {
+        "dataset": dataset,
+        "calendar_complete": calendar["complete"],
+        "calendar_missing_count": calendar["missing_count"],
+    }
+    if ts_code and table and pubs:
+        present_code: list[str] = []
+        for pub in pubs:
+            hit = db.fetchone(
+                f"SELECT 1 AS ok FROM {table} WHERE trade_date = ? AND batch_id = ? AND ts_code = ? LIMIT 1",
+                (pub["trade_date"], pub["active_batch"], ts_code),
+            )
+            if hit:
+                present_code.append(pub["trade_date"])
+        available = present_code
+        extra["code"] = ts_code
+    payload = coverage_payload(
+        kind="published_range",
+        start=start,
+        end=end,
+        expected=expected_open,
+        available=available,
+        extra=extra,
+    )
+    if not calendar["complete"]:
+        payload["complete"] = False
+        payload["calendar_missing_sample"] = calendar["missing_sample"]
+    return payload
+
+
+def point_coverage(trade_date: str, dataset: str = "") -> dict[str, Any]:
+    day = yyyymmdd(trade_date)
+    payload = coverage_payload(
+        kind="point",
+        start=day,
+        end=day,
+        expected=[day],
+        available=[day],
+        extra={"dataset": dataset} if dataset else None,
+    )
+    return payload
@@ -0,0 +1,42 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from cryptography.fernet import Fernet, InvalidToken
+
+
+class SecretVault:
+    def __init__(self, key: str) -> None:
+        try:
+            self._fernet = Fernet(key.encode("ascii"))
+        except (ValueError, TypeError) as exc:
+            raise ValueError("DATAHUB_ENCRYPTION_KEY 格式无效。") from exc
+
+    @staticmethod
+    def generate_key() -> str:
+        return Fernet.generate_key().decode("ascii")
+
+    def encrypt_json(self, payload: dict[str, Any]) -> str:
+        raw = json.dumps(payload, ensure_ascii=False, separators=(",", ":")).encode("utf-8")
+        return self._fernet.encrypt(raw).decode("ascii")
+
+    def decrypt_json(self, token: str) -> dict[str, Any]:
+        if not token:
+            return {}
+        try:
+            payload = json.loads(self._fernet.decrypt(token.encode("ascii")).decode("utf-8"))
+        except (InvalidToken, UnicodeDecodeError, json.JSONDecodeError) as exc:
+            raise ValueError("凭据无法解密，请检查 DATAHUB_ENCRYPTION_KEY。") from exc
+        if not isinstance(payload, dict):
+            raise ValueError("凭据格式无效。")
+        return payload
+
+
+def mask_secret(value: str, last_n: int = 4) -> str:
+    text = str(value or "")
+    if not text:
+        return ""
+    if len(text) <= last_n:
+        return "*" * len(text)
+    return ("*" * max(4, len(text) - last_n)) + text[-last_n:]
@@ -0,0 +1,340 @@
+from __future__ import annotations
+
+import sqlite3
+import threading
+from collections.abc import Iterator
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Any
+
+from datahub.timeutil import isoformat
+
+SCHEMA = """
+CREATE TABLE IF NOT EXISTS schema_migrations (
+  version INTEGER PRIMARY KEY,
+  applied_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS credentials (
+  name TEXT PRIMARY KEY,
+  encrypted_payload TEXT NOT NULL,
+  last4 TEXT,
+  updated_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS hub_admin (
+  id INTEGER PRIMARY KEY,
+  username TEXT NOT NULL UNIQUE,
+  password_salt TEXT NOT NULL,
+  password_hash TEXT NOT NULL,
+  password_must_change INTEGER NOT NULL DEFAULT 1,
+  failed_attempts INTEGER NOT NULL DEFAULT 0,
+  locked_until TEXT,
+  created_at TEXT NOT NULL,
+  updated_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS hub_sessions (
+  token_hash TEXT PRIMARY KEY,
+  csrf_token TEXT NOT NULL,
+  expires_at TEXT NOT NULL,
+  created_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS api_tokens (
+  token_hash TEXT PRIMARY KEY,
+  name TEXT NOT NULL,
+  last4 TEXT NOT NULL,
+  created_at TEXT NOT NULL,
+  revoked_at TEXT
+);
+
+CREATE TABLE IF NOT EXISTS trade_calendar (
+  exchange TEXT NOT NULL,
+  cal_date TEXT NOT NULL,
+  is_open INTEGER NOT NULL,
+  pretrade_date TEXT,
+  fetched_at TEXT NOT NULL,
+  PRIMARY KEY (exchange, cal_date)
+);
+
+CREATE TABLE IF NOT EXISTS stock_master (
+  ts_code TEXT PRIMARY KEY,
+  symbol TEXT,
+  name TEXT,
+  area TEXT,
+  industry TEXT,
+  market TEXT,
+  list_status TEXT,
+  list_date TEXT,
+  updated_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS eod_bars (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL,
+  open REAL, high REAL, low REAL, close REAL, pct_chg REAL,
+  volume REAL, amount REAL, adj_factor REAL,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS eod_valuation (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL,
+  turnover_rate REAL, volume_ratio REAL,
+  total_mv REAL, circ_mv REAL,
+  pe_ttm REAL, pb REAL, ps_ttm REAL, dv_ttm REAL,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS eod_moneyflow (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL,
+  buy_sm_amount REAL, sell_sm_amount REAL,
+  buy_md_amount REAL, sell_md_amount REAL,
+  buy_lg_amount REAL, sell_lg_amount REAL,
+  buy_elg_amount REAL, sell_elg_amount REAL,
+  net_mf_amount REAL,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS eod_auction (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL,
+  volume REAL, price REAL, amount REAL, pre_close REAL,
+  turnover_rate REAL, volume_ratio REAL, float_share REAL,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS eod_index_bars (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL,
+  open REAL, high REAL, low REAL, close REAL, pct_chg REAL,
+  volume REAL, amount REAL,
+  batch_id TEXT NOT NULL,
+  PRIMARY KEY (ts_code, trade_date, batch_id)
+) WITHOUT ROWID;
+
+CREATE TABLE IF NOT EXISTS staging_bars (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, batch_id TEXT NOT NULL,
+  open REAL, high REAL, low REAL, close REAL, pct_chg REAL,
+  volume REAL, amount REAL, adj_factor REAL,
+  PRIMARY KEY (batch_id, ts_code, trade_date)
+);
+
+CREATE TABLE IF NOT EXISTS staging_valuation (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, batch_id TEXT NOT NULL,
+  turnover_rate REAL, volume_ratio REAL,
+  total_mv REAL, circ_mv REAL, pe_ttm REAL, pb REAL, ps_ttm REAL, dv_ttm REAL,
+  PRIMARY KEY (batch_id, ts_code, trade_date)
+);
+
+CREATE TABLE IF NOT EXISTS staging_moneyflow (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, batch_id TEXT NOT NULL,
+  buy_sm_amount REAL, sell_sm_amount REAL, buy_md_amount REAL, sell_md_amount REAL,
+  buy_lg_amount REAL, sell_lg_amount REAL, buy_elg_amount REAL, sell_elg_amount REAL,
+  net_mf_amount REAL,
+  PRIMARY KEY (batch_id, ts_code, trade_date)
+);
+
+CREATE TABLE IF NOT EXISTS staging_auction (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, batch_id TEXT NOT NULL,
+  volume REAL, price REAL, amount REAL, pre_close REAL,
+  turnover_rate REAL, volume_ratio REAL, float_share REAL,
+  PRIMARY KEY (batch_id, ts_code, trade_date)
+);
+
+CREATE TABLE IF NOT EXISTS staging_index_bars (
+  ts_code TEXT NOT NULL, trade_date TEXT NOT NULL, batch_id TEXT NOT NULL,
+  open REAL, high REAL, low REAL, close REAL, pct_chg REAL,
+  volume REAL, amount REAL,
+  PRIMARY KEY (batch_id, ts_code, trade_date)
+);
+
+CREATE TABLE IF NOT EXISTS publications (
+  dataset TEXT NOT NULL, trade_date TEXT NOT NULL,
+  active_batch TEXT NOT NULL, prev_batch TEXT,
+  state TEXT NOT NULL,
+  published_at TEXT NOT NULL,
+  PRIMARY KEY (dataset, trade_date)
+);
+
+CREATE TABLE IF NOT EXISTS publication_history (
+  dataset TEXT NOT NULL, trade_date TEXT NOT NULL,
+  batch_id TEXT NOT NULL, published_at TEXT NOT NULL,
+  generation INTEGER NOT NULL,
+  PRIMARY KEY (dataset, trade_date, batch_id)
+);
+
+CREATE TABLE IF NOT EXISTS batches (
+  batch_id TEXT PRIMARY KEY,
+  dataset TEXT NOT NULL,
+  trade_date TEXT NOT NULL,
+  state TEXT NOT NULL,
+  attempt INTEGER DEFAULT 0,
+  rows_in INTEGER,
+  rows_out INTEGER,
+  quality_json TEXT,
+  started_at TEXT,
+  finished_at TEXT,
+  error TEXT
+);
+
+CREATE TABLE IF NOT EXISTS src_health (
+  provider TEXT NOT NULL, endpoint_class TEXT NOT NULL,
+  state TEXT NOT NULL,
+  last_ok_at TEXT, last_error TEXT,
+  consec_failures INTEGER DEFAULT 0,
+  opened_at TEXT,
+  cooldown_until TEXT,
+  PRIMARY KEY (provider, endpoint_class)
+);
+
+CREATE TABLE IF NOT EXISTS src_calls (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  provider TEXT NOT NULL,
+  endpoint TEXT NOT NULL,
+  ok INTEGER NOT NULL,
+  latency_ms INTEGER,
+  error TEXT,
+  created_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS job_runs (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  job_id TEXT NOT NULL,
+  state TEXT NOT NULL,
+  started_at TEXT,
+  finished_at TEXT,
+  rows_in INTEGER,
+  rows_out INTEGER,
+  error TEXT,
+  attempt INTEGER DEFAULT 1,
+  detail TEXT
+);
+
+CREATE TABLE IF NOT EXISTS audit_log (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  actor TEXT NOT NULL,
+  action TEXT NOT NULL,
+  target TEXT,
+  detail TEXT,
+  created_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS rt_cache (
+  cache_key TEXT PRIMARY KEY,
+  payload TEXT NOT NULL,
+  source TEXT NOT NULL,
+  stored_at TEXT NOT NULL,
+  expires_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS last_known_good (
+  cache_key TEXT PRIMARY KEY,
+  payload TEXT NOT NULL,
+  source TEXT NOT NULL,
+  stored_at TEXT NOT NULL
+);
+
+CREATE TABLE IF NOT EXISTS diff_reports (
+  id INTEGER PRIMARY KEY AUTOINCREMENT,
+  trade_date TEXT NOT NULL,
+  metric TEXT NOT NULL,
+  left_source TEXT,
+  right_source TEXT,
+  left_value REAL,
+  right_value REAL,
+  deviation REAL,
+  sample_count INTEGER,
+  created_at TEXT NOT NULL
+);
+
+CREATE INDEX IF NOT EXISTS idx_batches_date ON batches(trade_date, dataset);
+CREATE INDEX IF NOT EXISTS idx_job_runs_job ON job_runs(job_id, started_at);
+CREATE INDEX IF NOT EXISTS idx_src_calls_created ON src_calls(created_at);
+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);
+"""
+
+DATASET_TABLES = {
+    "daily": ("eod_bars", "staging_bars"),
+    "valuation": ("eod_valuation", "staging_valuation"),
+    "moneyflow": ("eod_moneyflow", "staging_moneyflow"),
+    "auction": ("eod_auction", "staging_auction"),
+    "index_daily": ("eod_index_bars", "staging_index_bars"),
+}
+
+
+class ManagedConnection(sqlite3.Connection):
+    def __exit__(self, exc_type, exc_value, traceback):
+        try:
+            return super().__exit__(exc_type, exc_value, traceback)
+        finally:
+            self.close()
+
+
+class HubDB:
+    def __init__(self, path: Path, timeout_seconds: float = 20) -> None:
+        self.path = Path(path)
+        self.path.parent.mkdir(parents=True, exist_ok=True)
+        self.timeout_seconds = timeout_seconds
+        self._write_lock = threading.RLock()
+        self.initialize()
+
+    def connect(self) -> sqlite3.Connection:
+        connection = sqlite3.connect(
+            self.path,
+            timeout=self.timeout_seconds,
+            factory=ManagedConnection,
+        )
+        connection.row_factory = sqlite3.Row
+        connection.execute("PRAGMA journal_mode=WAL")
+        connection.execute("PRAGMA foreign_keys=ON")
+        connection.execute("PRAGMA busy_timeout=20000")
+        connection.execute("PRAGMA synchronous=NORMAL")
+        return connection
+
+    def initialize(self) -> None:
+        with self.connect() as connection:
+            connection.executescript(SCHEMA)
+            row = connection.execute(
+                "SELECT version FROM schema_migrations ORDER BY version DESC LIMIT 1"
+            ).fetchone()
+            if row is None:
+                connection.execute(
+                    "INSERT INTO schema_migrations(version, applied_at) VALUES (1, ?)",
+                    (isoformat(),),
+                )
+
+    @contextmanager
+    def write(self) -> Iterator[sqlite3.Connection]:
+        with self._write_lock:
+            with self.connect() as connection:
+                yield connection
+
+    def fetchall(self, sql: str, params: tuple[Any, ...] = ()) -> list[dict[str, Any]]:
+        with self.connect() as connection:
+            rows = connection.execute(sql, params).fetchall()
+        return [dict(row) for row in rows]
+
+    def fetchone(self, sql: str, params: tuple[Any, ...] = ()) -> dict[str, Any] | None:
+        with self.connect() as connection:
+            row = connection.execute(sql, params).fetchone()
+        return dict(row) if row else None
+
+    def execute(self, sql: str, params: tuple[Any, ...] = ()) -> None:
+        with self.write() as connection:
+            connection.execute(sql, params)
+
+    def executemany(self, sql: str, rows: list[tuple[Any, ...]]) -> None:
+        with self.write() as connection:
+            connection.executemany(sql, rows)
+
+    def backup_to(self, dest: Path) -> None:
+        dest.parent.mkdir(parents=True, exist_ok=True)
+        with self.connect() as source, sqlite3.connect(dest) as target:
+            source.backup(target)
+
+    def vacuum(self) -> None:
+        with self.connect() as connection:
+            connection.execute("VACUUM")
@@ -0,0 +1,13 @@
+from datahub.governance.circuit import CircuitBreaker, CircuitState
+from datahub.governance.lkg import LastKnownGood
+from datahub.governance.ratelimit import TokenBucket
+from datahub.governance.retry import RetryError, retry_call
+
+__all__ = [
+    "CircuitBreaker",
+    "CircuitState",
+    "LastKnownGood",
+    "RetryError",
+    "TokenBucket",
+    "retry_call",
+]
@@ -0,0 +1,107 @@
+from __future__ import annotations
+
+import threading
+import time
+from collections import deque
+from dataclasses import dataclass
+
+
+@dataclass
+class CircuitState:
+    state: str = "closed"  # closed | open | half_open
+    consec_failures: int = 0
+    opened_at: float | None = None
+    cooldown_until: float = 0.0
+    last_error: str = ""
+    last_ok_at: float | None = None
+
+
+class CircuitBreaker:
+    """Sliding-window breaker: 5 consecutive failures or >50% of 60s window → open."""
+
+    def __init__(
+        self,
+        failure_threshold: int = 5,
+        window_seconds: float = 60.0,
+        open_seconds: float = 120.0,
+        max_open_seconds: float = 600.0,
+        clock=time.monotonic,
+    ) -> None:
+        self.failure_threshold = failure_threshold
+        self.window_seconds = window_seconds
+        self.open_seconds = open_seconds
+        self.max_open_seconds = max_open_seconds
+        self._clock = clock
+        self._lock = threading.Lock()
+        self._events: deque[tuple[float, bool]] = deque()
+        self.status = CircuitState()
+        self._open_stretch = open_seconds
+
+    def allow(self) -> bool:
+        with self._lock:
+            self._refresh_locked()
+            if self.status.state == "open":
+                return False
+            if self.status.state == "half_open":
+                # single probe in flight: caller must record success/failure
+                return True
+            return True
+
+    def record_success(self) -> CircuitState:
+        with self._lock:
+            now = self._clock()
+            self._events.append((now, True))
+            self.status.last_ok_at = now
+            self.status.consec_failures = 0
+            self.status.last_error = ""
+            self._open_stretch = self.open_seconds
+            self.status.state = "closed"
+            self.status.opened_at = None
+            self.status.cooldown_until = 0.0
+            return self._copy()
+
+    def record_failure(self, error: str = "") -> CircuitState:
+        with self._lock:
+            now = self._clock()
+            self._events.append((now, False))
+            self.status.consec_failures += 1
+            self.status.last_error = error
+            self._prune_locked(now)
+            failures = sum(1 for _, ok in self._events if not ok)
+            total = len(self._events)
+            rate = (failures / total) if total else 0.0
+            trip = self.status.consec_failures >= self.failure_threshold or (
+                total >= self.failure_threshold and rate > 0.5
+            )
+            if trip:
+                self.status.state = "open"
+                self.status.opened_at = now
+                self.status.cooldown_until = now + self._open_stretch
+                self._open_stretch = min(self.max_open_seconds, self._open_stretch * 2)
+            return self._copy()
+
+    def snapshot(self) -> CircuitState:
+        with self._lock:
+            self._refresh_locked()
+            return self._copy()
+
+    def _refresh_locked(self) -> None:
+        now = self._clock()
+        self._prune_locked(now)
+        if self.status.state == "open" and now >= self.status.cooldown_until:
+            self.status.state = "half_open"
+
+    def _prune_locked(self, now: float) -> None:
+        cutoff = now - self.window_seconds
+        while self._events and self._events[0][0] < cutoff:
+            self._events.popleft()
+
+    def _copy(self) -> CircuitState:
+        return CircuitState(
+            state=self.status.state,
+            consec_failures=self.status.consec_failures,
+            opened_at=self.status.opened_at,
+            cooldown_until=self.status.cooldown_until,
+            last_error=self.status.last_error,
+            last_ok_at=self.status.last_ok_at,
+        )
@@ -0,0 +1,84 @@
+from __future__ import annotations
+
+import json
+from typing import Any
+
+from datahub.db import HubDB
+from datahub.timeutil import isoformat, now_shanghai
+
+
+class LastKnownGood:
+    def __init__(self, db: HubDB) -> None:
+        self.db = db
+
+    def store(self, cache_key: str, payload: Any, source: str) -> None:
+        self.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, isoformat()),
+        )
+
+    def load(self, cache_key: str) -> dict[str, Any] | None:
+        row = self.db.fetchone("SELECT * FROM last_known_good WHERE cache_key = ?", (cache_key,))
+        if not row:
+            return None
+        return {
+            "payload": json.loads(row["payload"]),
+            "source": row["source"],
+            "stored_at": row["stored_at"],
+        }
+
+    def put_rt(self, cache_key: str, payload: Any, source: str, ttl_seconds: int) -> None:
+        now = now_shanghai()
+        expires = isoformat(now.replace(microsecond=0))
+        # expires_at stored as iso; compute by adding ttl via timestamp
+        from datetime import timedelta
+
+        self.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,
+                isoformat(now),
+                isoformat(now + timedelta(seconds=ttl_seconds)),
+            ),
+        )
+        self.store(cache_key, payload, source)
+
+    def get_rt(self, cache_key: str, max_stale_seconds: int | None = None) -> dict[str, Any] | None:
+        row = self.db.fetchone("SELECT * FROM rt_cache WHERE cache_key = ?", (cache_key,))
+        if not row:
+            lkg = self.load(cache_key)
+            if not lkg:
+                return None
+            return {**lkg, "stale": True}
+        stored_at = row["stored_at"]
+        expired = row["expires_at"] < isoformat()
+        result = {
+            "payload": json.loads(row["payload"]),
+            "source": row["source"],
+            "stored_at": stored_at,
+            "stale": expired,
+        }
+        if expired and max_stale_seconds is not None:
+            from datetime import datetime
+
+            try:
+                stored = datetime.fromisoformat(stored_at)
+                age = (now_shanghai() - stored).total_seconds()
+            except ValueError:
+                age = max_stale_seconds + 1
+            if age > max_stale_seconds:
+                return None
+        return result
@@ -0,0 +1,36 @@
+from __future__ import annotations
+
+import threading
+import time
+
+
+class TokenBucket:
+    def __init__(self, rate_per_minute: float, capacity: float | None = None, clock=time.monotonic) -> None:
+        self.rate_per_second = max(0.001, rate_per_minute / 60.0)
+        self.capacity = float(capacity if capacity is not None else rate_per_minute)
+        self._tokens = self.capacity
+        self._updated = clock()
+        self._clock = clock
+        self._lock = threading.Lock()
+
+    def acquire(self, tokens: float = 1.0, block: bool = True) -> bool:
+        while True:
+            with self._lock:
+                now = self._clock()
+                elapsed = max(0.0, now - self._updated)
+                self._tokens = min(self.capacity, self._tokens + elapsed * self.rate_per_second)
+                self._updated = now
+                if self._tokens >= tokens:
+                    self._tokens -= tokens
+                    return True
+                wait = (tokens - self._tokens) / self.rate_per_second
+            if not block:
+                return False
+            time.sleep(min(wait, 0.05))
+
+    @property
+    def remaining(self) -> float:
+        with self._lock:
+            now = self._clock()
+            elapsed = max(0.0, now - self._updated)
+            return min(self.capacity, self._tokens + elapsed * self.rate_per_second)
@@ -0,0 +1,35 @@
+from __future__ import annotations
+
+import time
+from collections.abc import Callable
+from typing import TypeVar
+
+T = TypeVar("T")
+
+
+class RetryError(RuntimeError):
+    def __init__(self, message: str, attempts: int, last_error: BaseException | None = None) -> None:
+        super().__init__(message)
+        self.attempts = attempts
+        self.last_error = last_error
+
+
+def retry_call(
+    fn: Callable[[], T],
+    attempts: int = 5,
+    base_delay: float = 0.2,
+    max_delay: float = 8.0,
+    sleeper: Callable[[float], None] = time.sleep,
+    retry_on: tuple[type[BaseException], ...] = (Exception,),
+) -> T:
+    last: BaseException | None = None
+    for attempt in range(1, max(1, attempts) + 1):
+        try:
+            return fn()
+        except retry_on as exc:
+            last = exc
+            if attempt >= attempts:
+                break
+            delay = min(max_delay, base_delay * (2 ** (attempt - 1)))
+            sleeper(delay)
+    raise RetryError(f"retry exhausted after {attempts} attempts: {last}", attempts, last)
@@ -0,0 +1,250 @@
+from __future__ import annotations
+
+import json
+import mimetypes
+import secrets
+from http import HTTPStatus
+from http.cookies import SimpleCookie
+from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from typing import Any
+from urllib.parse import unquote, urlparse
+
+from datahub.hub import Hub
+from datahub.logutil import configure_logging, get_logger
+from datahub.serving import ApiError, parse_query
+
+LOGGER = get_logger()
+SESSION_COOKIE = "datahub_session"
+
+
+class HubRequestHandler(BaseHTTPRequestHandler):
+    hub: Hub
+
+    def log_message(self, format: str, *args: Any) -> None:
+        LOGGER.info(format % args)
+
+    def do_GET(self) -> None:  # noqa: N802
+        self._dispatch("GET")
+
+    def do_POST(self) -> None:  # noqa: N802
+        self._dispatch("POST")
+
+    def do_OPTIONS(self) -> None:  # noqa: N802
+        self.send_response(HTTPStatus.NO_CONTENT)
+        self.send_header("Allow", "GET, POST, OPTIONS")
+        self.end_headers()
+
+    def _dispatch(self, method: str) -> None:
+        parsed = urlparse(self.path)
+        path = unquote(parsed.path)
+        try:
+            if path in {"/livez", "/healthz"}:
+                self._json({"status": "ok"}, HTTPStatus.OK)
+                return
+            if path.startswith("/v1/"):
+                self._v1(path, parsed.query)
+                return
+            if path.startswith("/admin/api/"):
+                self._admin_api(method, path)
+                return
+            if path.startswith("/admin"):
+                self._admin_static(path)
+                return
+            if path == "/":
+                self.send_response(HTTPStatus.FOUND)
+                self.send_header("Location", "/admin/")
+                self.end_headers()
+                return
+            self._json({"error": {"code": "INVALID_ARGUMENT", "message": "Not found"}}, HTTPStatus.NOT_FOUND)
+        except ApiError as exc:
+            self._json(exc.payload(), exc.status)
+        except PermissionError as exc:
+            self._json({"error": {"code": "UNAUTHORIZED", "message": str(exc)}}, HTTPStatus.UNAUTHORIZED)
+        except ValueError as exc:
+            self._json({"error": {"code": "INVALID_ARGUMENT", "message": str(exc)}}, HTTPStatus.BAD_REQUEST)
+        except Exception:
+            LOGGER.exception("internal error")
+            self._json({"error": {"code": "INTERNAL", "message": "internal error"}}, HTTPStatus.INTERNAL_SERVER_ERROR)
+
+    def _v1(self, path: str, query: str) -> None:
+        token = self.headers.get("X-Datahub-Token", "")
+        if not self.hub.auth.check_api_token(token):
+            self.hub.pipeline.audit("anonymous", "unauthorized", path, "")
+            raise ApiError("UNAUTHORIZED", "missing or invalid X-Datahub-Token")
+        payload = self.hub.api.handle(path, parse_query(query))
+        self._json(payload, HTTPStatus.OK)
+
+    def _admin_api(self, method: str, path: str) -> None:
+        if path == "/admin/api/login" and method == "POST":
+            body = self._read_json()
+            result = self.hub.auth.login(str(body.get("username") or "hub_admin"), str(body.get("password") or ""))
+            self._json(
+                {"ok": True, "must_change": result["must_change"], "csrf": result["csrf"]},
+                HTTPStatus.OK,
+                extra_headers=[self._cookie(result["session"])],
+            )
+            return
+        user = self.hub.auth.session_user(self._cookie_value(SESSION_COOKIE))
+        if not user:
+            raise ApiError("UNAUTHORIZED", "请先登录")
+        if method == "POST" and path != "/admin/api/login":
+            csrf = self.headers.get("X-CSRF-Token", "")
+            if not csrf or not secrets.compare_digest(csrf, str(user["csrf_token"])):
+                raise ApiError("UNAUTHORIZED", "CSRF 校验失败")
+        if path == "/admin/api/logout" and method == "POST":
+            self.hub.auth.logout(self._cookie_value(SESSION_COOKIE))
+            self._json({"ok": True}, HTTPStatus.OK, extra_headers=[self._cookie("", clear=True)])
+            return
+        if path == "/admin/api/session" and method == "GET":
+            self._json({"username": user["username"], "must_change": user["must_change"], "csrf": user["csrf_token"]}, HTTPStatus.OK)
+            return
+        if path == "/admin/api/change-password" and method == "POST":
+            body = self._read_json()
+            self.hub.auth.change_password(str(body.get("current") or ""), str(body.get("new_password") or ""))
+            self.hub.pipeline.audit(user["username"], "change_password", "hub_admin", "")
+            self._json({"ok": True}, HTTPStatus.OK)
+            return
+        if user["must_change"] and path not in {"/admin/api/change-password", "/admin/api/session"}:
+            raise ApiError("UNAUTHORIZED", "请先修改初始密码")
+        if path == "/admin/api/overview" and method == "GET":
+            self._json(self.hub.admin.overview(), HTTPStatus.OK)
+            return
+        if path == "/admin/api/sources" and method == "GET":
+            self._json(self.hub.admin.sources(), HTTPStatus.OK)
+            return
+        if path.startswith("/admin/api/sources/") and path.endswith("/probe") and method == "POST":
+            provider = path.split("/")[4]
+            self._json(self.hub.admin.probe(provider), HTTPStatus.OK)
+            return
+        if path == "/admin/api/jobs" and method == "GET":
+            self._json(self.hub.admin.jobs(), HTTPStatus.OK)
+            return
+        if path.startswith("/admin/api/jobs/") and path.endswith("/run") and method == "POST":
+            job_id = path.split("/")[4]
+            body = self._read_json(allow_empty=True)
+            self._json(self.hub.admin.run_job(job_id, str(body.get("trade_date") or "")), HTTPStatus.OK)
+            return
+        if path == "/admin/api/batches" and method == "GET":
+            query = parse_query(urlparse(self.path).query)
+            date = (query.get("date") or [""])[0]
+            dataset = (query.get("dataset") or [""])[0]
+            self._json(self.hub.admin.batches(date, dataset), HTTPStatus.OK)
+            return
+        if path == "/admin/api/datasets" and method == "GET":
+            query = parse_query(urlparse(self.path).query)
+            self._json(self.hub.admin.datasets((query.get("date") or [""])[0]), HTTPStatus.OK)
+            return
+        if path == "/admin/api/audit" and method == "GET":
+            self._json(self.hub.admin.audit(), HTTPStatus.OK)
+            return
+        if path == "/admin/api/rollback" and method == "POST":
+            body = self._read_json()
+            result = self.hub.admin.rollback(
+                str(body.get("dataset") or ""),
+                str(body.get("trade_date") or ""),
+                str(body.get("password") or ""),
+                str(body.get("confirm") or ""),
+                user["username"],
+            )
+            self._json(result, HTTPStatus.OK)
+            return
+        if path == "/admin/api/backfill" and method == "POST":
+            body = self._read_json()
+            result = self.hub.admin.backfill(
+                str(body.get("dataset") or ""),
+                str(body.get("trade_date") or ""),
+                str(body.get("password") or ""),
+                str(body.get("confirm") or ""),
+                user["username"],
+            )
+            self._json(result, HTTPStatus.OK)
+            return
+        raise ApiError("INVALID_ARGUMENT", f"unknown admin endpoint: {path}")
+
+    def _admin_static(self, path: str) -> None:
+        relative = path[len("/admin"):].lstrip("/") or "index.html"
+        candidate = (self.hub.static_dir / relative).resolve()
+        try:
+            candidate.relative_to(self.hub.static_dir.resolve())
+        except ValueError:
+            self.send_error(HTTPStatus.FORBIDDEN)
+            return
+        if candidate.is_dir():
+            candidate = candidate / "index.html"
+        if not candidate.is_file():
+            candidate = self.hub.static_dir / "index.html"
+        content = candidate.read_bytes()
+        content_type = mimetypes.guess_type(candidate.name)[0] or "application/octet-stream"
+        if content_type.startswith("text/") or content_type in {"application/javascript", "application/json"}:
+            content_type += "; charset=utf-8"
+        self.send_response(HTTPStatus.OK)
+        self.send_header("Content-Type", content_type)
+        self.send_header("Content-Length", str(len(content)))
+        self.send_header("Cache-Control", "no-cache")
+        self.end_headers()
+        self.wfile.write(content)
+
+    def _read_json(self, allow_empty: bool = False) -> dict[str, Any]:
+        length = int(self.headers.get("Content-Length", "0") or 0)
+        if length == 0 and allow_empty:
+            return {}
+        if length <= 0 or length > 65536:
+            raise ValueError("请求内容为空或过大")
+        raw = self.rfile.read(length)
+        try:
+            payload = json.loads(raw.decode("utf-8"))
+        except (UnicodeDecodeError, json.JSONDecodeError):
+            LOGGER.warning("invalid json request body")
+            raise ValueError("请求不是合法 JSON") from None
+        if not isinstance(payload, dict):
+            raise ValueError("请求不是合法 JSON")
+        return payload
+
+    def _cookie_value(self, name: str) -> str:
+        cookie = SimpleCookie()
+        try:
+            cookie.load(self.headers.get("Cookie", ""))
+        except Exception:
+            return ""
+        morsel = cookie.get(name)
+        return morsel.value if morsel else ""
+
+    def _cookie(self, value: str, clear: bool = False) -> str:
+        max_age = 0 if clear else 12 * 3600
+        return f"{SESSION_COOKIE}={value}; Path=/; HttpOnly; SameSite=Strict; Max-Age={max_age}"
+
+    def _json(self, payload: dict[str, Any], status: HTTPStatus, extra_headers: list[str] | None = None) -> None:
+        raw = json.dumps(payload, ensure_ascii=False).encode("utf-8")
+        self.send_response(status)
+        self.send_header("Content-Type", "application/json; charset=utf-8")
+        self.send_header("Content-Length", str(len(raw)))
+        self.send_header("Cache-Control", "no-store")
+        for header in extra_headers or []:
+            self.send_header("Set-Cookie", header)
+        self.end_headers()
+        self.wfile.write(raw)
+
+
+def make_handler(hub: Hub) -> type[HubRequestHandler]:
+    class BoundHandler(HubRequestHandler):
+        pass
+
+    BoundHandler.hub = hub
+    BoundHandler.protocol_version = "HTTP/1.1"
+    return BoundHandler
+
+
+def serve(hub: Hub, host: str, port: int) -> None:
+    configure_logging(hub.settings.log_level)
+    handler = make_handler(hub)
+    server = ThreadingHTTPServer((host, port), handler)
+    hub.start()
+    LOGGER.info("xiaobai-datahub listening", extra={"hub": {"host": host, "port": port}})
+    print(f"xiaobai-datahub is running at http://{host}:{port}/admin/")
+    try:
+        server.serve_forever()
+    except KeyboardInterrupt:
+        pass
+    finally:
+        hub.stop()
+        server.server_close()
@@ -0,0 +1,54 @@
+from __future__ import annotations
+
+from pathlib import Path
+
+from datahub.adapters.tushare import TushareAdapter
+from datahub.admin_api import AdminAPI
+from datahub.auth import AuthService
+from datahub.crypto import SecretVault
+from datahub.db import HubDB
+from datahub.governance.circuit import CircuitBreaker
+from datahub.governance.lkg import LastKnownGood
+from datahub.governance.ratelimit import TokenBucket
+from datahub.pipeline import Pipeline
+from datahub.scheduler import Scheduler
+from datahub.serving import V1API
+from datahub.settings import Settings, load_settings
+
+
+class Hub:
+    def __init__(self, settings: Settings, adapter: TushareAdapter | None = None) -> None:
+        if not settings.encryption_key:
+            raise SystemExit("DATAHUB_ENCRYPTION_KEY 未配置")
+        self.settings = settings
+        self.db = HubDB(settings.db_path)
+        self.vault = SecretVault(settings.encryption_key)
+        self.auth = AuthService(self.db, self.vault, settings.api_token, settings.admin_password)
+        token = settings.tushare_token or self.auth.load_credential("tushare_token")
+        if settings.tushare_token:
+            self.auth.store_credential("tushare_token", settings.tushare_token)
+            token = settings.tushare_token
+        self.adapter = adapter or TushareAdapter(token)
+        self.pipeline = Pipeline(
+            self.db,
+            self.adapter,
+            settings,
+            bucket=TokenBucket(settings.tushare_rate_per_minute),
+            breaker=CircuitBreaker(),
+        )
+        self.lkg = LastKnownGood(self.db)
+        self.scheduler = Scheduler(self.db, self.pipeline)
+        self.api = V1API(self.db, self.pipeline, settings)
+        self.admin = AdminAPI(self.db, self.pipeline, self.scheduler, self.auth)
+        self.static_dir = Path(__file__).resolve().parents[1] / "admin"
+
+    def start(self) -> None:
+        if self.settings.scheduler_enabled:
+            self.scheduler.start()
+
+    def stop(self) -> None:
+        self.scheduler.stop()
+
+
+def build_hub(settings: Settings | None = None) -> Hub:
+    return Hub(settings or load_settings())
@@ -0,0 +1,82 @@
+from __future__ import annotations
+
+import json
+import logging
+import re
+import sys
+import traceback
+from typing import Any
+
+from datahub.timeutil import isoformat
+
+_SECRET_KEYS = (
+    "token", "password", "secret", "key", "authorization", "credential",
+    "tushare_token", "datahub_token", "encryption_key", "cookie",
+)
+_SECRET_JSON = re.compile(
+    r'(?i)("(?:' + "|".join(re.escape(key) for key in _SECRET_KEYS) + r')"\s*:\s*")([^"\\]*(?:\\.[^"\\]*)*)(")'
+)
+
+
+def redact_log_text(text: str) -> str:
+    return _SECRET_JSON.sub(r"\1***\3", str(text))
+
+
+def _redact(value: Any, key: str = "") -> Any:
+    lowered = key.lower()
+    if any(part in lowered for part in _SECRET_KEYS):
+        return "***"
+    if isinstance(value, dict):
+        return {str(item_key): _redact(item_value, str(item_key)) for item_key, item_value in value.items()}
+    if isinstance(value, list):
+        return [_redact(item) for item in value]
+    if isinstance(value, str):
+        return redact_log_text(value)
+    return value
+
+
+def _safe_exc_text(exc_info: tuple[Any, Any, Any]) -> str:
+    exc = exc_info[1]
+    if isinstance(exc, json.JSONDecodeError):
+        return f"JSONDecodeError: invalid json at position {exc.pos}"
+    cause = getattr(exc, "__cause__", None)
+    if isinstance(cause, json.JSONDecodeError):
+        return f"{type(exc).__name__}: invalid json in request"
+    text = "".join(traceback.format_exception(*exc_info))
+    if isinstance(cause, json.JSONDecodeError) and cause.doc:
+        text = text.replace(cause.doc, "")
+    if isinstance(exc, json.JSONDecodeError) and exc.doc:
+        text = text.replace(exc.doc, "")
+    return redact_log_text(text)
+
+
+class JsonFormatter(logging.Formatter):
+    def format(self, record: logging.LogRecord) -> str:
+        payload: dict[str, Any] = {
+            "ts": isoformat(),
+            "level": record.levelname,
+            "logger": record.name,
+            "message": redact_log_text(record.getMessage()),
+        }
+        extra = getattr(record, "hub", None)
+        if isinstance(extra, dict):
+            payload.update(_redact(extra))
+        if record.exc_info:
+            payload["exc"] = _safe_exc_text(record.exc_info)
+        return json.dumps(payload, ensure_ascii=False, default=str)
+
+
+def configure_logging(level: str = "INFO") -> logging.Logger:
+    logger = logging.getLogger("datahub")
+    if logger.handlers:
+        return logger
+    handler = logging.StreamHandler(sys.stdout)
+    handler.setFormatter(JsonFormatter())
+    logger.addHandler(handler)
+    logger.setLevel(getattr(logging, level.upper(), logging.INFO))
+    logger.propagate = False
+    return logger
+
+
+def get_logger() -> logging.Logger:
+    return logging.getLogger("datahub")
@@ -0,0 +1,201 @@
+"""Canonical field normalization for Tushare-native rows.
+
+Units (architecture §7.1):
+- price: 4 decimal REAL
+- pct_chg: percent, 4 decimal REAL
+- volume: shares (Tushare daily/index vol is 手 → ×100)
+- amount: yuan (Tushare daily/index amount is 千元 → ×1000)
+- moneyflow amounts: yuan (Tushare is 万元 → ×1e4)
+- daily_basic total_mv / circ_mv: yuan (Tushare is 万元 → ×1e4)
+- stk_auction.amount is already yuan in Tushare; volume 手 → ×100
+
+Existing xiaobai-review stores Tushare native units and converts at display time.
+Hub converts once at ingest. Golden tests compare hub output against applying
+these same factors to review-native rows.
+"""
+
+from __future__ import annotations
+
+from typing import Any
+
+from datahub.numbers import finite_number, round4
+
+AMOUNT_THOUSAND_YUAN = 1000.0
+AMOUNT_WAN_YUAN = 10000.0
+VOLUME_LOT = 100.0
+
+DAILY_FIELDS = ("ts_code", "trade_date", "open", "high", "low", "close", "pct_chg", "vol", "amount")
+VALUATION_FIELDS = (
+    "ts_code", "trade_date", "turnover_rate", "volume_ratio",
+    "total_mv", "circ_mv", "pe_ttm", "pb", "ps_ttm", "dv_ttm",
+)
+MONEYFLOW_FIELDS = (
+    "ts_code", "trade_date",
+    "buy_sm_amount", "sell_sm_amount", "buy_md_amount", "sell_md_amount",
+    "buy_lg_amount", "sell_lg_amount", "buy_elg_amount", "sell_elg_amount",
+    "net_mf_amount",
+)
+AUCTION_FIELDS = (
+    "ts_code", "trade_date", "vol", "price", "amount", "pre_close",
+    "turnover_rate", "volume_ratio", "float_share",
+)
+INDEX_FIELDS = ("ts_code", "trade_date", "open", "high", "low", "close", "pct_chg", "vol", "amount")
+CALENDAR_FIELDS = ("exchange", "cal_date", "is_open", "pretrade_date")
+STOCK_FIELDS = ("ts_code", "symbol", "name", "area", "industry", "market", "list_status", "list_date")
+
+
+def _code(value: Any) -> str:
+    return str(value or "").strip().upper()
+
+
+def _date(value: Any) -> str:
+    return str(value or "").replace("-", "")[:8]
+
+
+def review_daily_to_canonical(row: dict[str, Any]) -> dict[str, Any]:
+    """Convert a review-stored daily row (Tushare native units) to hub canonical."""
+    return normalize_daily(row)
+
+
+def normalize_daily(row: dict[str, Any], adj_factor: float | None = None) -> dict[str, Any]:
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "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"))),
+        "pct_chg": round4(finite_number(row.get("pct_chg"))),
+        "volume": round4(_scale(row.get("vol"), VOLUME_LOT)),
+        "amount": round4(_scale(row.get("amount"), AMOUNT_THOUSAND_YUAN)),
+        "adj_factor": round4(finite_number(adj_factor if adj_factor is not None else row.get("adj_factor"))),
+    }
+
+
+def normalize_valuation(row: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "turnover_rate": round4(finite_number(row.get("turnover_rate"))),
+        "volume_ratio": round4(finite_number(row.get("volume_ratio"))),
+        "total_mv": round4(_scale(row.get("total_mv"), AMOUNT_WAN_YUAN)),
+        "circ_mv": round4(_scale(row.get("circ_mv"), AMOUNT_WAN_YUAN)),
+        "pe_ttm": round4(finite_number(row.get("pe_ttm"))),
+        "pb": round4(finite_number(row.get("pb"))),
+        "ps_ttm": round4(finite_number(row.get("ps_ttm"))),
+        "dv_ttm": round4(finite_number(row.get("dv_ttm"))),
+    }
+
+
+def normalize_moneyflow(row: dict[str, Any]) -> dict[str, Any]:
+    converted = {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+    }
+    for field in MONEYFLOW_FIELDS[2:]:
+        converted[field] = round4(_scale(row.get(field), AMOUNT_WAN_YUAN))
+    return converted
+
+
+def normalize_auction(row: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "volume": round4(_scale(row.get("vol") if row.get("vol") is not None else row.get("volume"), VOLUME_LOT)),
+        "price": round4(finite_number(row.get("price"))),
+        "amount": round4(finite_number(row.get("amount"))),
+        "pre_close": round4(finite_number(row.get("pre_close"))),
+        "turnover_rate": round4(finite_number(row.get("turnover_rate"))),
+        "volume_ratio": round4(finite_number(row.get("volume_ratio"))),
+        "float_share": round4(_scale(row.get("float_share"), AMOUNT_WAN_YUAN) if row.get("float_share") is not None else None),
+    }
+
+
+def normalize_index_daily(row: dict[str, Any]) -> dict[str, Any]:
+    return {
+        "ts_code": _code(row.get("ts_code")),
+        "trade_date": _date(row.get("trade_date")),
+        "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"))),
+        "pct_chg": round4(finite_number(row.get("pct_chg"))),
+        "volume": round4(_scale(row.get("vol"), VOLUME_LOT)),
+        "amount": round4(_scale(row.get("amount"), AMOUNT_THOUSAND_YUAN)),
+    }
+
+
+def normalize_calendar(row: dict[str, Any]) -> dict[str, Any]:
+    is_open = row.get("is_open")
+    if is_open in (True, "1", 1, "Y", "y"):
+        open_flag = 1
+    elif is_open in (False, "0", 0, "N", "n", None, ""):
+        open_flag = 0
+    else:
+        open_flag = int(is_open)
+    return {
+        "exchange": str(row.get("exchange") or "SSE"),
+        "cal_date": _date(row.get("cal_date") or row.get("calDate")),
+        "is_open": open_flag,
+        "pretrade_date": _date(row.get("pretrade_date")) or None,
+    }
+
+
+def normalize_stock(row: dict[str, Any]) -> dict[str, Any]:
+    ts_code = _code(row.get("ts_code"))
+    symbol = str(row.get("symbol") or "").strip() or (ts_code.split(".")[0] if ts_code else "")
+    return {
+        "ts_code": ts_code,
+        "symbol": symbol,
+        "name": str(row.get("name") or "").strip(),
+        "area": str(row.get("area") or "").strip() or None,
+        "industry": str(row.get("industry") or "").strip() or None,
+        "market": str(row.get("market") or "").strip() or None,
+        "list_status": str(row.get("list_status") or "L").strip() or "L",
+        "list_date": _date(row.get("list_date")) or None,
+    }
+
+
+def apply_qfq(price: float | None, factor: float | None, latest_factor: float | None) -> float | None:
+    if price is None:
+        return None
+    current = factor if factor not in (None, 0) else 1.0
+    latest = latest_factor if latest_factor not in (None, 0) else current
+    return round4(price * current / latest)
+
+
+def qfq_bar(row: dict[str, Any], latest_factor: float | None) -> dict[str, Any]:
+    factor = finite_number(row.get("adj_factor"), 1.0) or 1.0
+    out = dict(row)
+    for field in ("open", "high", "low", "close"):
+        out[field] = apply_qfq(finite_number(row.get(field)), factor, latest_factor)
+    return out
+
+
+NORMALIZERS = {
+    "daily": normalize_daily,
+    "valuation": normalize_valuation,
+    "daily_basic": normalize_valuation,
+    "moneyflow": normalize_moneyflow,
+    "auction": normalize_auction,
+    "stk_auction": normalize_auction,
+    "index_daily": normalize_index_daily,
+    "trade_cal": normalize_calendar,
+    "calendar": normalize_calendar,
+    "stock_basic": normalize_stock,
+    "stocks": normalize_stock,
+}
+
+
+def normalize_rows(dataset: str, rows: list[dict[str, Any]]) -> list[dict[str, Any]]:
+    fn = NORMALIZERS.get(dataset)
+    if fn is None:
+        raise ValueError(f"unknown dataset: {dataset}")
+    return [fn(row) for row in rows]
+
+
+def _scale(value: Any, factor: float) -> float | None:
+    number = finite_number(value)
+    if number is None:
+        return None
+    return number * factor
@@ -0,0 +1,23 @@
+from __future__ import annotations
+
+import math
+from typing import Any
+
+
+def finite_number(value: Any, default: float | None = None) -> float | None:
+    """Return a finite float, or default (None means JSON null)."""
+    if value is None or value == "":
+        return default
+    try:
+        number = float(value)
+    except (TypeError, ValueError):
+        return default
+    if not math.isfinite(number):
+        return default
+    return number
+
+
+def round4(value: float | None) -> float | None:
+    if value is None:
+        return None
+    return round(float(value), 4)
@@ -0,0 +1,701 @@
+from __future__ import annotations
+
+import json
+import time
+from collections.abc import Callable
+from datetime import timedelta
+from typing import Any
+
+from datahub.adapters.base import AdapterError
+from datahub.adapters.tushare import DEFAULT_INDEX_CODES, WEBSITE_INDEX_CODES, TushareAdapter
+from datahub.db import DATASET_TABLES, HubDB
+from datahub.governance.circuit import CircuitBreaker
+from datahub.governance.ratelimit import TokenBucket
+from datahub.governance.retry import RetryError, retry_call
+from datahub.logutil import get_logger
+from datahub.normalize import finite_number, normalize_daily
+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"}
+OFFICIAL_DATASETS = HARD_DATASETS | SOFT_DATASETS
+EMPTY_BATCH_ERROR = "empty official batch: 0 valid rows"
+
+STAGING_INSERT = {
+    "daily": (
+        "INSERT INTO staging_bars(ts_code,trade_date,batch_id,open,high,low,close,pct_chg,volume,amount,adj_factor) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], b, r.get("open"), r.get("high"), r.get("low"),
+            r.get("close"), r.get("pct_chg"), r.get("volume"), r.get("amount"), r.get("adj_factor"),
+        ),
+    ),
+    "valuation": (
+        "INSERT INTO staging_valuation(ts_code,trade_date,batch_id,turnover_rate,volume_ratio,total_mv,circ_mv,pe_ttm,pb,ps_ttm,dv_ttm) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], b, r.get("turnover_rate"), r.get("volume_ratio"),
+            r.get("total_mv"), r.get("circ_mv"), r.get("pe_ttm"), r.get("pb"), r.get("ps_ttm"), r.get("dv_ttm"),
+        ),
+    ),
+    "moneyflow": (
+        "INSERT INTO staging_moneyflow(ts_code,trade_date,batch_id,buy_sm_amount,sell_sm_amount,buy_md_amount,sell_md_amount,buy_lg_amount,sell_lg_amount,buy_elg_amount,sell_elg_amount,net_mf_amount) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], b,
+            r.get("buy_sm_amount"), r.get("sell_sm_amount"), r.get("buy_md_amount"), r.get("sell_md_amount"),
+            r.get("buy_lg_amount"), r.get("sell_lg_amount"), r.get("buy_elg_amount"), r.get("sell_elg_amount"),
+            r.get("net_mf_amount"),
+        ),
+    ),
+    "auction": (
+        "INSERT INTO staging_auction(ts_code,trade_date,batch_id,volume,price,amount,pre_close,turnover_rate,volume_ratio,float_share) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], b, r.get("volume"), r.get("price"), r.get("amount"),
+            r.get("pre_close"), r.get("turnover_rate"), r.get("volume_ratio"), r.get("float_share"),
+        ),
+    ),
+    "index_daily": (
+        "INSERT INTO staging_index_bars(ts_code,trade_date,batch_id,open,high,low,close,pct_chg,volume,amount) "
+        "VALUES (?,?,?,?,?,?,?,?,?,?)",
+        lambda r, b: (
+            r["ts_code"], r["trade_date"], b, r.get("open"), r.get("high"), r.get("low"),
+            r.get("close"), r.get("pct_chg"), r.get("volume"), r.get("amount"),
+        ),
+    ),
+}
+
+EOD_COPY = {
+    "daily": (
+        "INSERT OR REPLACE INTO eod_bars "
+        "SELECT ts_code,trade_date,open,high,low,close,pct_chg,volume,amount,adj_factor,batch_id "
+        "FROM staging_bars WHERE batch_id = ?"
+    ),
+    "valuation": (
+        "INSERT OR REPLACE INTO eod_valuation "
+        "SELECT ts_code,trade_date,turnover_rate,volume_ratio,total_mv,circ_mv,pe_ttm,pb,ps_ttm,dv_ttm,batch_id "
+        "FROM staging_valuation WHERE batch_id = ?"
+    ),
+    "moneyflow": (
+        "INSERT OR REPLACE INTO eod_moneyflow "
+        "SELECT ts_code,trade_date,buy_sm_amount,sell_sm_amount,buy_md_amount,sell_md_amount,"
+        "buy_lg_amount,sell_lg_amount,buy_elg_amount,sell_elg_amount,net_mf_amount,batch_id "
+        "FROM staging_moneyflow WHERE batch_id = ?"
+    ),
+    "auction": (
+        "INSERT OR REPLACE INTO eod_auction "
+        "SELECT ts_code,trade_date,volume,price,amount,pre_close,turnover_rate,volume_ratio,float_share,batch_id "
+        "FROM staging_auction WHERE batch_id = ?"
+    ),
+    "index_daily": (
+        "INSERT OR REPLACE INTO eod_index_bars "
+        "SELECT ts_code,trade_date,open,high,low,close,pct_chg,volume,amount,batch_id "
+        "FROM staging_index_bars WHERE batch_id = ?"
+    ),
+}
+
+
+def _staging_row_count(connection: Any, dataset: str, batch_id: str) -> int:
+    table = DATASET_TABLES[dataset][1]
+    row = connection.execute(
+        f"SELECT COUNT(*) AS n FROM {table} WHERE batch_id = ?",
+        (batch_id,),
+    ).fetchone()
+    return int(row["n"] if row is not None else 0)
+
+
+class QualityError(RuntimeError):
+    def __init__(self, message: str, report: dict[str, Any]) -> None:
+        super().__init__(message)
+        self.report = report
+
+
+class Pipeline:
+    def __init__(
+        self,
+        db: HubDB,
+        adapter: TushareAdapter,
+        settings: Settings,
+        bucket: TokenBucket | None = None,
+        breaker: CircuitBreaker | None = None,
+        before_commit: Callable[[], None] | None = None,
+        clock=None,
+    ) -> None:
+        self.db = db
+        self.adapter = adapter
+        self.settings = settings
+        self.bucket = bucket or TokenBucket(settings.tushare_rate_per_minute)
+        self.breaker = breaker or CircuitBreaker()
+        self.before_commit = before_commit
+        self.clock = clock or now_shanghai
+
+    def next_batch_id(self, dataset: str, trade_date: str) -> str:
+        row = self.db.fetchone(
+            "SELECT COUNT(*) AS n FROM batches WHERE dataset = ? AND trade_date = ?",
+            (dataset, trade_date),
+        )
+        seq = int((row or {}).get("n") or 0) + 1
+        return f"{trade_date}-{dataset}-{seq:03d}"
+
+    def ingest_reference(
+        self,
+        trade_date: str | None = None,
+        start: str | None = None,
+        end: str | None = None,
+    ) -> dict[str, Any]:
+        """Refresh trade calendar and stock master. Not versioned by batch.
+
+        Calendar defaults to 2016-01-01 through today+30 so a 5-year website
+        query is not silently truncated. UPSERT makes repeats safe.
+        """
+        day = yyyymmdd(trade_date or self.clock())
+        start = yyyymmdd(start or self.settings.calendar_start)
+        end = yyyymmdd(end or add_days(day, 30))
+        if start > end:
+            start, end = end, start
+        calendar = self.adapter.normalize(
+            "calendar",
+            self._guarded_fetch("calendar", {"exchange": "SSE", "start_date": start, "end_date": end}),
+        )
+        stocks = self.adapter.normalize("stocks", self._guarded_fetch("stocks", {"list_status": "L"}))
+        fetched_at = isoformat(self.clock())
+        with self.db.write() as connection:
+            for row in calendar:
+                connection.execute(
+                    """
+                    INSERT INTO trade_calendar(exchange, cal_date, is_open, pretrade_date, fetched_at)
+                    VALUES (?, ?, ?, ?, ?)
+                    ON CONFLICT(exchange, cal_date) DO UPDATE SET
+                      is_open=excluded.is_open, pretrade_date=excluded.pretrade_date, fetched_at=excluded.fetched_at
+                    """,
+                    (row["exchange"], row["cal_date"], row["is_open"], row.get("pretrade_date"), fetched_at),
+                )
+            for row in stocks:
+                connection.execute(
+                    """
+                    INSERT INTO stock_master(ts_code,symbol,name,area,industry,market,list_status,list_date,updated_at)
+                    VALUES (?,?,?,?,?,?,?,?,?)
+                    ON CONFLICT(ts_code) DO UPDATE SET
+                      symbol=excluded.symbol, name=excluded.name, area=excluded.area,
+                      industry=excluded.industry, market=excluded.market,
+                      list_status=excluded.list_status, list_date=excluded.list_date,
+                      updated_at=excluded.updated_at
+                    """,
+                    (
+                        row["ts_code"], row.get("symbol"), row.get("name"), row.get("area"),
+                        row.get("industry"), row.get("market"), row.get("list_status"),
+                        row.get("list_date"), fetched_at,
+                    ),
+                )
+        return {
+            "calendar": len(calendar),
+            "stocks": len(stocks),
+            "trade_date": day,
+            "calendar_from": start,
+            "calendar_to": end,
+        }
+
+    def open_trade_dates(self, end: str, limit: int) -> list[str]:
+        end = yyyymmdd(end)
+        rows = self.db.fetchall(
+            """
+            SELECT cal_date FROM trade_calendar
+            WHERE exchange = 'SSE' AND is_open = 1 AND cal_date <= ?
+            ORDER BY cal_date DESC
+            LIMIT ?
+            """,
+            (end, max(1, int(limit))),
+        )
+        return sorted(str(row["cal_date"]) for row in rows)
+
+    def backfill_history(
+        self,
+        trade_date: str | None = None,
+        calendar_start: str | None = None,
+        index_days: int | None = None,
+        codes: tuple[str, ...] | None = None,
+        force: bool = False,
+    ) -> dict[str, Any]:
+        """Idempotent calendar + website-index history backfill."""
+        day = yyyymmdd(trade_date or self.clock())
+        calendar = self.ingest_reference(day, start=calendar_start)
+        index = self.backfill_index_history(
+            end_date=day,
+            trading_days=index_days,
+            codes=codes,
+            force=force,
+        )
+        return {"calendar": calendar, "index_daily": index, "ok": bool(index.get("ok"))}
+
+    def backfill_index_history(
+        self,
+        end_date: str | None = None,
+        trading_days: int | None = None,
+        codes: tuple[str, ...] | None = None,
+        force: bool = False,
+    ) -> dict[str, Any]:
+        """Incrementally publish official index bars for website index codes.
+
+        One range fetch per code, then per-day publish. Already published dates
+        are skipped unless ``force``. Failures are recorded and do not roll back
+        successful days.
+        """
+        end = yyyymmdd(end_date or self.clock())
+        limit = int(trading_days or self.settings.index_history_trading_days)
+        codes = tuple(codes or WEBSITE_INDEX_CODES)
+        open_dates = self.open_trade_dates(end, limit)
+        if not open_dates:
+            return {
+                "start": None,
+                "end": end,
+                "codes": list(codes),
+                "requested_days": 0,
+                "published": [],
+                "skipped": [],
+                "failed": [{"error": "calendar has no open dates on or before end"}],
+                "ok": False,
+            }
+        start = open_dates[0]
+        complete_dates = set() if force else self._index_dates_with_all_codes(start, end, codes)
+        targets = [day for day in open_dates if day not in complete_dates]
+        skipped = [day for day in open_dates if day in complete_dates]
+        by_date: dict[str, list[dict[str, Any]]] = {day: [] for day in targets}
+        failed: list[dict[str, Any]] = []
+        for ts_code in codes:
+            try:
+                raw = retry_call(
+                    lambda code=ts_code: self._guarded_fetch(
+                        "index_daily",
+                        {"ts_code": code, "start_date": start, "end_date": end},
+                    ),
+                    attempts=self.settings.max_publish_attempts,
+                    base_delay=0.05,
+                    sleeper=lambda _d: time.sleep(_d),
+                )
+                for row in self.adapter.normalize("index_daily", raw):
+                    day = str(row.get("trade_date") or "")
+                    if day in by_date:
+                        by_date[day].append(row)
+            except Exception as exc:
+                failed.append({"ts_code": ts_code, "error": str(exc)})
+        published: list[dict[str, Any]] = []
+        for day in targets:
+            rows = by_date.get(day) or []
+            try:
+                result = self.run_dataset("index_daily", day, prepared_rows=rows)
+                published.append(
+                    {
+                        "trade_date": day,
+                        "batch_id": result["batch_id"],
+                        "rows": result["rows"],
+                        "state": result["state"],
+                    }
+                )
+            except Exception as exc:
+                failed.append({"trade_date": day, "error": str(exc), "rows": len(rows)})
+        return {
+            "start": start,
+            "end": end,
+            "codes": list(codes),
+            "requested_days": len(open_dates),
+            "published": published,
+            "skipped": skipped,
+            "failed": failed,
+            "ok": not failed,
+        }
+
+    def _index_dates_with_all_codes(self, start: str, end: str, codes: tuple[str, ...]) -> set[str]:
+        pubs = self.db.fetchall(
+            """
+            SELECT trade_date, active_batch FROM publications
+            WHERE dataset = 'index_daily' AND trade_date >= ? AND trade_date <= ?
+            """,
+            (start, end),
+        )
+        needed = set(codes)
+        complete: set[str] = set()
+        for pub in pubs:
+            rows = self.db.fetchall(
+                "SELECT DISTINCT ts_code FROM eod_index_bars WHERE trade_date = ? AND batch_id = ?",
+                (pub["trade_date"], pub["active_batch"]),
+            )
+            have = {str(row["ts_code"]) for row in rows}
+            if needed <= have:
+                complete.add(str(pub["trade_date"]))
+        return complete
+
+    def run_dataset(
+        self,
+        dataset: str,
+        trade_date: str,
+        attempts: int | None = None,
+        prepared_rows: list[dict[str, Any]] | None = None,
+    ) -> dict[str, Any]:
+        trade_date = yyyymmdd(trade_date)
+        batch_id = self.next_batch_id(dataset, trade_date)
+        max_attempts = attempts or self.settings.max_publish_attempts
+        self._set_batch(batch_id, dataset, trade_date, "scheduled", 0)
+        rows: list[dict[str, Any]] = []
+        try:
+            self._set_batch(batch_id, dataset, trade_date, "fetching", 1)
+            if prepared_rows is None:
+                rows = retry_call(
+                    lambda: self._fetch_dataset(dataset, trade_date),
+                    attempts=max_attempts,
+                    base_delay=0.05,
+                    sleeper=lambda _d: None if attempts == 1 else time.sleep(_d),
+                )
+            else:
+                rows = list(prepared_rows)
+            self._stage(dataset, batch_id, rows)
+            self._set_batch(batch_id, dataset, trade_date, "staged", 1, rows_in=len(rows), rows_out=len(rows))
+            self._set_batch(batch_id, dataset, trade_date, "validating", 1)
+            report = self.validate(dataset, batch_id, trade_date, rows)
+            if report["hard_fail"]:
+                self._reject_batch(batch_id, dataset, trade_date, rows, report)
+                raise QualityError("integrity gate failed", report)
+            self._set_batch(batch_id, dataset, trade_date, "deriving", 1, rows_in=len(rows), rows_out=len(rows), quality=report)
+            self._set_batch(batch_id, dataset, trade_date, "publishing", 1, rows_in=len(rows), rows_out=len(rows), quality=report)
+            state = "degraded" if report["soft_fail"] else "published"
+            self.publish(dataset, trade_date, batch_id, state=state)
+            self._set_batch(
+                batch_id, dataset, trade_date, "published", 1,
+                rows_in=len(rows), rows_out=len(rows), quality=report, finished=True,
+            )
+            return {"batch_id": batch_id, "dataset": dataset, "trade_date": trade_date, "rows": len(rows), "state": state, "quality": report}
+        except RetryError as exc:
+            self._set_batch(batch_id, dataset, trade_date, "failed", max_attempts, error=str(exc), finished=True)
+            raise
+        except QualityError as exc:
+            current = self.db.fetchone("SELECT state FROM batches WHERE batch_id = ?", (batch_id,))
+            if current and current["state"] not in {"staged", "failed"}:
+                self._reject_batch(batch_id, dataset, trade_date, rows, exc.report)
+            raise
+        except Exception as exc:
+            self._set_batch(batch_id, dataset, trade_date, "failed", 1, error=str(exc), finished=True)
+            raise
+
+    def run_eod_batch_a(self, trade_date: str) -> dict[str, Any]:
+        results = {}
+        for dataset in ("daily", "valuation", "moneyflow", "auction"):
+            results[dataset] = self.run_dataset(dataset, trade_date)
+        return results
+
+    def run_eod_batch_b(self, trade_date: str) -> dict[str, Any]:
+        return {"index_daily": self.run_dataset("index_daily", trade_date)}
+
+    def validate(self, dataset: str, batch_id: str, trade_date: str, rows: list[dict[str, Any]]) -> dict[str, Any]:
+        quality = self.settings.quality
+        errors: list[str] = []
+        warnings: list[str] = []
+        listed = self.db.fetchone(
+            "SELECT COUNT(*) AS n FROM stock_master WHERE list_status = 'L'",
+        )
+        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]
+        dup = row_n - len(set(keys))
+        if dup:
+            errors.append(f"duplicate keys: {dup}")
+        bad_date = sum(1 for row in rows if str(row.get("trade_date")) != trade_date)
+        if bad_date:
+            errors.append(f"date mismatch rows: {bad_date}")
+        ratio = (row_n / listed_n) if listed_n else 1.0
+        if dataset == "daily" and listed_n and ratio < float(quality.get("daily_row_ratio") or 0.98):
+            errors.append(f"row ratio {ratio:.4f} < {quality.get('daily_row_ratio')}")
+        null_fields = ("open", "high", "low", "close", "amount") if dataset in {"daily", "index_daily"} else ()
+        if null_fields and rows:
+            nulls = sum(1 for row in rows if any(row.get(field) is None for field in null_fields))
+            null_rate = nulls / row_n
+            if null_rate >= float(quality.get("null_rate_max") or 0.01):
+                errors.append(f"null rate {null_rate:.4f}")
+        empty = row_n == 0
+        if empty and dataset in OFFICIAL_DATASETS:
+            errors.append(EMPTY_BATCH_ERROR)
+        if dataset in SOFT_DATASETS:
+            hard_fail = bool(dup or bad_date or empty)
+        else:
+            hard_fail = bool(errors) and dataset in HARD_DATASETS
+        return {
+            "rows": row_n,
+            "listed": listed_n,
+            "ratio": round(ratio, 4),
+            "errors": errors,
+            "warnings": warnings,
+            "hard_fail": hard_fail,
+            "soft_fail": bool(warnings) and not hard_fail,
+            "batch_id": batch_id,
+        }
+
+    def publish(self, dataset: str, trade_date: str, batch_id: str, state: str = "published") -> None:
+        copy_sql = EOD_COPY[dataset]
+        published_at = isoformat(self.clock())
+        with self.db.write() as connection:
+            rows_out = _staging_row_count(connection, dataset, batch_id)
+            if rows_out <= 0:
+                report = {
+                    "rows": 0,
+                    "errors": [EMPTY_BATCH_ERROR],
+                    "warnings": [],
+                    "hard_fail": True,
+                    "soft_fail": False,
+                    "batch_id": batch_id,
+                    "dataset": dataset,
+                    "trade_date": trade_date,
+                }
+                LOGGER.warning(
+                    "skip official publish for empty batch",
+                    extra={
+                        "hub": {
+                            "dataset": dataset,
+                            "trade_date": trade_date,
+                            "batch_id": batch_id,
+                            "rows_out": rows_out,
+                            "reason": "upstream_empty",
+                        }
+                    },
+                )
+                raise QualityError("empty batch cannot be officially published", report)
+            current = connection.execute(
+                "SELECT active_batch FROM publications WHERE dataset = ? AND trade_date = ?",
+                (dataset, trade_date),
+            ).fetchone()
+            prev = str(current["active_batch"]) if current else None
+            connection.execute(copy_sql, (batch_id,))
+            if self.before_commit:
+                self.before_commit()
+            connection.execute(
+                """
+                INSERT INTO publications(dataset, trade_date, active_batch, prev_batch, state, published_at)
+                VALUES (?, ?, ?, ?, ?, ?)
+                ON CONFLICT(dataset, trade_date) DO UPDATE SET
+                  prev_batch=excluded.prev_batch,
+                  active_batch=excluded.active_batch,
+                  state=excluded.state,
+                  published_at=excluded.published_at
+                """,
+                (dataset, trade_date, batch_id, prev, state, published_at),
+            )
+            max_gen = connection.execute(
+                "SELECT COALESCE(MAX(generation), 0) AS g FROM publication_history WHERE dataset = ? AND trade_date = ?",
+                (dataset, trade_date),
+            ).fetchone()
+            generation = int(max_gen["g"]) + 1
+            connection.execute(
+                "INSERT OR REPLACE INTO publication_history(dataset, trade_date, batch_id, published_at, generation) VALUES (?,?,?,?,?)",
+                (dataset, trade_date, batch_id, published_at, generation),
+            )
+            keep = int(self.settings.quality.get("publication_generations") or 3)
+            stale = connection.execute(
+                """
+                SELECT batch_id FROM publication_history
+                WHERE dataset = ? AND trade_date = ?
+                ORDER BY generation DESC
+                """,
+                (dataset, trade_date),
+            ).fetchall()
+            for row in stale[keep:]:
+                connection.execute(
+                    "DELETE FROM publication_history WHERE dataset = ? AND trade_date = ? AND batch_id = ?",
+                    (dataset, trade_date, row["batch_id"]),
+                )
+
+    def rollback(self, dataset: str, trade_date: str, actor: str = "admin") -> dict[str, Any]:
+        trade_date = yyyymmdd(trade_date)
+        pub = self.db.fetchone(
+            "SELECT * FROM publications WHERE dataset = ? AND trade_date = ?",
+            (dataset, trade_date),
+        )
+        if not pub or not pub.get("prev_batch"):
+            raise ValueError("没有可回滚的上一批次")
+        target = pub["prev_batch"]
+        published_at = isoformat(self.clock())
+        with self.db.write() as connection:
+            connection.execute(
+                """
+                UPDATE publications
+                SET prev_batch = active_batch, active_batch = ?, published_at = ?, state = 'published'
+                WHERE dataset = ? AND trade_date = ?
+                """,
+                (target, published_at, dataset, trade_date),
+            )
+        self.audit(actor, "rollback", f"{dataset}:{trade_date}", json.dumps({"to": target, "from": pub["active_batch"]}))
+        return {"dataset": dataset, "trade_date": trade_date, "active_batch": target, "prev_batch": pub["active_batch"]}
+
+    def active_batch(self, dataset: str, trade_date: str) -> str | None:
+        row = self.db.fetchone(
+            "SELECT active_batch FROM publications WHERE dataset = ? AND trade_date = ?",
+            (dataset, trade_date),
+        )
+        return str(row["active_batch"]) if row else None
+
+    def cleanup(self) -> dict[str, int]:
+        staging_days = int(self.settings.quality.get("staging_retain_days") or 14)
+        job_days = int(self.settings.quality.get("job_run_retain_days") or 90)
+        now = now_shanghai(self.clock())
+        cutoff_staging = add_days(yyyymmdd(now), -staging_days)
+        cutoff_jobs = isoformat(now - timedelta(days=job_days))
+        deleted = 0
+        with self.db.write() as connection:
+            for dataset, (_eod, staging) in DATASET_TABLES.items():
+                cur = connection.execute(
+                    f"DELETE FROM {staging} WHERE trade_date < ?",
+                    (cutoff_staging,),
+                )
+                deleted += cur.rowcount
+            connection.execute("DELETE FROM job_runs WHERE started_at < ?", (cutoff_jobs,))
+            connection.execute("DELETE FROM src_calls WHERE created_at < ?", (cutoff_jobs,))
+        return {"staging_deleted": deleted}
+
+    def audit(self, actor: str, action: str, target: str = "", detail: str = "") -> None:
+        self.db.execute(
+            "INSERT INTO audit_log(actor, action, target, detail, created_at) VALUES (?,?,?,?,?)",
+            (actor, action, target, detail, isoformat(self.clock())),
+        )
+
+    def _reject_batch(
+        self,
+        batch_id: str,
+        dataset: str,
+        trade_date: str,
+        rows: list[dict[str, Any]],
+        report: dict[str, Any],
+    ) -> None:
+        errors = report.get("errors") or []
+        LOGGER.warning(
+            "official batch rejected",
+            extra={
+                "hub": {
+                    "dataset": dataset,
+                    "trade_date": trade_date,
+                    "batch_id": batch_id,
+                    "rows_out": len(rows),
+                    "errors": errors,
+                    "reason": "upstream_empty" if EMPTY_BATCH_ERROR in errors else "integrity_gate",
+                }
+            },
+        )
+        self._set_batch(
+            batch_id, dataset, trade_date, "staged", 1,
+            rows_in=len(rows), rows_out=len(rows),
+            quality=report, error="; ".join(str(item) for item in errors),
+        )
+
+    def _fetch_dataset(self, dataset: str, trade_date: str) -> list[dict[str, Any]]:
+        if dataset == "daily":
+            raw = self._guarded_fetch("daily", {"trade_date": trade_date})
+            factors = {
+                (row["ts_code"], row["trade_date"]): finite_number(row.get("adj_factor"))
+                for row in self._guarded_fetch("adj_factor", {"trade_date": trade_date})
+            }
+            return [
+                normalize_daily(row, adj_factor=factors.get((str(row.get("ts_code") or "").upper(), str(row.get("trade_date") or ""))))
+                for row in raw
+            ]
+        if dataset == "index_daily":
+            rows: list[dict[str, Any]] = []
+            for ts_code in DEFAULT_INDEX_CODES:
+                raw = self._guarded_fetch("index_daily", {"ts_code": ts_code, "trade_date": trade_date})
+                rows.extend(self.adapter.normalize("index_daily", raw))
+            return rows
+        api_dataset = dataset
+        raw = self._guarded_fetch(api_dataset, {"trade_date": trade_date})
+        return self.adapter.normalize(api_dataset, raw)
+
+    def _guarded_fetch(self, dataset: str, params: dict[str, Any]) -> list[dict[str, Any]]:
+        if not self.breaker.allow():
+            raise AdapterError("Tushare circuit open")
+        self.bucket.acquire()
+        started = time.perf_counter()
+        try:
+            # For daily we want RAW tushare rows so adj_factor can be merged later.
+            rows = self.adapter.fetch(dataset, params)
+            latency = round((time.perf_counter() - started) * 1000)
+            self.breaker.record_success()
+            self._log_call(dataset, True, latency, "")
+            self._persist_health("ok")
+            return rows
+        except Exception as exc:
+            latency = round((time.perf_counter() - started) * 1000)
+            self.breaker.record_failure(str(exc))
+            self._log_call(dataset, False, latency, str(exc))
+            self._persist_health("error", str(exc))
+            raise
+
+    def _stage(self, dataset: str, batch_id: str, rows: list[dict[str, Any]]) -> None:
+        sql, mapper = STAGING_INSERT[dataset]
+        with self.db.write() as connection:
+            connection.execute(
+                f"DELETE FROM {DATASET_TABLES[dataset][1]} WHERE batch_id = ?",
+                (batch_id,),
+            )
+            connection.executemany(sql, [mapper(row, batch_id) for row in rows])
+
+    def _set_batch(
+        self,
+        batch_id: str,
+        dataset: str,
+        trade_date: str,
+        state: str,
+        attempt: int,
+        rows_in: int | None = None,
+        rows_out: int | None = None,
+        quality: dict[str, Any] | None = None,
+        error: str | None = None,
+        finished: bool = False,
+    ) -> None:
+        now = isoformat(self.clock())
+        existing = self.db.fetchone("SELECT batch_id FROM batches WHERE batch_id = ?", (batch_id,))
+        payload = json.dumps(quality, ensure_ascii=False) if quality else None
+        with self.db.write() as connection:
+            if existing is None:
+                connection.execute(
+                    """
+                    INSERT INTO batches(batch_id, dataset, trade_date, state, attempt, rows_in, rows_out, quality_json, started_at, finished_at, error)
+                    VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
+                    """,
+                    (batch_id, dataset, trade_date, state, attempt, rows_in, rows_out, payload, now, now if finished else None, error),
+                )
+            else:
+                connection.execute(
+                    """
+                    UPDATE batches SET state=?, attempt=?,
+                      rows_in=COALESCE(?, rows_in), rows_out=COALESCE(?, rows_out),
+                      quality_json=COALESCE(?, quality_json),
+                      finished_at=CASE WHEN ? THEN ? ELSE finished_at END,
+                      error=COALESCE(?, error)
+                    WHERE batch_id = ?
+                    """,
+                    (state, attempt, rows_in, rows_out, payload, 1 if finished else 0, now, error, batch_id),
+                )
+
+    def _log_call(self, endpoint: str, ok: bool, latency_ms: int, error: str) -> None:
+        self.db.execute(
+            "INSERT INTO src_calls(provider, endpoint, ok, latency_ms, error, created_at) VALUES (?,?,?,?,?,?)",
+            ("tushare", endpoint, 1 if ok else 0, latency_ms, error, isoformat(self.clock())),
+        )
+
+    def _persist_health(self, state: str, error: str = "") -> None:
+        snap = self.breaker.snapshot()
+        self.db.execute(
+            """
+            INSERT INTO src_health(provider, endpoint_class, state, last_ok_at, last_error, consec_failures, opened_at, cooldown_until)
+            VALUES (?, ?, ?, ?, ?, ?, ?, ?)
+            ON CONFLICT(provider, endpoint_class) DO UPDATE SET
+              state=excluded.state, last_ok_at=excluded.last_ok_at, last_error=excluded.last_error,
+              consec_failures=excluded.consec_failures, opened_at=excluded.opened_at, cooldown_until=excluded.cooldown_until
+            """,
+            (
+                "tushare", "pro",
+                snap.state,
+                isoformat(self.clock()) if state == "ok" else None,
+                error or snap.last_error,
+                snap.consec_failures,
+                isoformat(self.clock()) if snap.state == "open" else None,
+                None,
+            ),
+        )
@@ -0,0 +1,149 @@
+from __future__ import annotations
+
+import threading
+from collections.abc import Callable
+from datetime import datetime, time
+from typing import Any
+
+from datahub.db import HubDB
+from datahub.logutil import get_logger
+from datahub.pipeline import Pipeline
+from datahub.timeutil import isoformat, now_shanghai, yyyymmdd
+
+LOGGER = get_logger()
+
+JobFn = Callable[[str], Any]
+
+
+def is_open_day(db: HubDB, day: str) -> bool:
+    row = db.fetchone(
+        "SELECT is_open FROM trade_calendar WHERE exchange = 'SSE' AND cal_date = ?",
+        (day,),
+    )
+    if row is None:
+        return True  # unknown calendar: do not skip reference refresh
+    return int(row["is_open"]) == 1
+
+
+class Scheduler:
+    """Calendar-driven in-process scheduler. Non-trading days skip EOD fetches."""
+
+    def __init__(self, db: HubDB, pipeline: Pipeline, jobs: dict[str, JobFn] | None = None) -> None:
+        self.db = db
+        self.pipeline = pipeline
+        self.jobs = jobs or {
+            "precheck": self._precheck,
+            "eod_a": self._eod_a,
+            "eod_b": self._eod_b,
+            "cleanup": self._cleanup,
+            "backup": self._backup,
+            "history_backfill": self._history_backfill,
+        }
+        self._stop = threading.Event()
+        self._thread: threading.Thread | None = None
+        self._fired: set[tuple[str, str, str]] = set()
+
+    def start(self, interval_seconds: float = 30.0) -> None:
+        if self._thread and self._thread.is_alive():
+            return
+
+        def loop() -> None:
+            while not self._stop.wait(interval_seconds):
+                try:
+                    self.tick()
+                except Exception:
+                    LOGGER.exception("scheduler tick failed")
+
+        self._thread = threading.Thread(target=loop, name="datahub-scheduler", daemon=True)
+        self._thread.start()
+
+    def stop(self, timeout: float = 5.0) -> None:
+        self._stop.set()
+        if self._thread and self._thread is not threading.current_thread():
+            self._thread.join(timeout)
+
+    def tick(self, clock: datetime | None = None) -> list[str]:
+        now = clock or now_shanghai()
+        day = yyyymmdd(now)
+        current = now.timetz() if False else now.time()
+        ran: list[str] = []
+        plan = [
+            ("precheck", time(8, 45)),
+            ("eod_a", time(15, 5)),
+            ("eod_b", time(15, 10)),
+            ("cleanup", time(0, 30)),
+            ("backup", time(0, 40)),
+        ]
+        open_day = is_open_day(self.db, day)
+        for job_id, at in plan:
+            if current < at:
+                continue
+            key = (job_id, day, at.strftime("%H%M"))
+            if key in self._fired:
+                continue
+            if job_id in {"eod_a", "eod_b"} and not open_day:
+                self._fired.add(key)
+                continue
+            self._fired.add(key)
+            self.run_job(job_id, day)
+            ran.append(job_id)
+        return ran
+
+    def run_job(self, job_id: str, trade_date: str) -> dict[str, Any]:
+        fn = self.jobs.get(job_id)
+        if fn is None:
+            raise KeyError(job_id)
+        started = isoformat()
+        run_id = None
+        with self.db.write() as connection:
+            cur = connection.execute(
+                "INSERT INTO job_runs(job_id, state, started_at, attempt) VALUES (?,?,?,1)",
+                (job_id, "running", started),
+            )
+            run_id = cur.lastrowid
+        try:
+            result = fn(trade_date) or {}
+            with self.db.write() as connection:
+                connection.execute(
+                    "UPDATE job_runs SET state=?, finished_at=?, rows_out=?, detail=? WHERE id=?",
+                    ("ok", isoformat(), result.get("rows") if isinstance(result, dict) else None, str(result)[:2000], run_id),
+                )
+            return {"job_id": job_id, "result": result, "state": "ok"}
+        except Exception as exc:
+            with self.db.write() as connection:
+                connection.execute(
+                    "UPDATE job_runs SET state=?, finished_at=?, error=? WHERE id=?",
+                    ("failed", isoformat(), str(exc), run_id),
+                )
+            raise
+
+    def _precheck(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.ingest_reference(trade_date)
+
+    def _eod_a(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_batch_a(trade_date)
+
+    def _eod_b(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.run_eod_batch_b(trade_date)
+
+    def _history_backfill(self, trade_date: str) -> dict[str, Any]:
+        return self.pipeline.backfill_history(trade_date)
+
+    def _cleanup(self, trade_date: str) -> dict[str, Any]:
+        result = self.pipeline.cleanup()
+        if now_shanghai().weekday() == 6:
+            self.pipeline.db.vacuum()
+            result["vacuum"] = True
+        return result
+
+    def _backup(self, trade_date: str) -> dict[str, Any]:
+        from pathlib import Path
+
+        dest_dir = Path(self.pipeline.settings.backup_dir)
+        dest = dest_dir / f"datahub-{trade_date}.db"
+        self.pipeline.db.backup_to(dest)
+        keep = int(self.pipeline.settings.quality.get("backup_retain") or 14)
+        backups = sorted(dest_dir.glob("datahub-*.db"))
+        for old in backups[:-keep]:
+            old.unlink(missing_ok=True)
+        return {"path": str(dest.name), "kept": min(len(backups), keep)}
--- a/Show More
+++ b/Show More
Author	SHA1	Message	Date
总工 Cursor multica-agent	a836cda1b2	feat(HEL-421): 回补历史日历和指数并标记区间不完整 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>	2026-09-02 22:16:05 +08:00
总工	25ff6bbe06	fix(HEL-417): 配置根日志让影子对比报告落入容器日志	2026-09-02 21:53:19 +08:00
总工 Cursor multica-agent	5085cacf0d	fix(HEL-412): 刷新降级不再整次失败，并补齐准备中提示手动刷新与自动补跑共用可用数据判定：日线推算或上一交易日快照记为部分/准备中成功，避免前端误报刷新失败。HTTP JSON 解析错误不再把请求正文写入日志。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>	2026-09-02 18:08:40 +08:00
总工 Cursor multica-agent	0d13066386	feat(HEL-402): 接通网站首批只读 datahub 并建立双路对比默认全部读取/影子开关关闭，网站继续走旧 Tushare 链路；开启单项时只替换该类原料并在失败时回旧，问天保持旧路径。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>	2026-09-02 17:09:28 +08:00
总工 Cursor multica-agent	f5dc0f8076	fix(HEL-396): 禁止空数据批次冒充正式发布统一发布入口在有效行数为 0 时不再写成 published，也不推进正式批次指针。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>	2026-09-02 15:56:29 +08:00
总工 Cursor multica-agent	031eefab4d	fix(HEL-386): 清理任务按 ISO 截止时间删除 job_runs/src_calls YYYYMMDD 与 ISO 字符串比较会把同年保留期内记录全部误删。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>	2026-09-02 12:26:09 +08:00
总工 Cursor multica-agent	3498dd7a4b	feat(HEL-382): 搭建 datahub 底座和盘后正式数据链路新增独立 xiaobai-datahub 服务（SQLite WAL、Tushare 盘后发布、/v1 契约和管理后台），不改现站页面与数据链路。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>	2026-09-02 12:05:26 +08:00
总工 Cursor multica-agent	c2ebc0ab91	docs(HEL-353): 恢复普通 README 并移除多媒体展示放弃宣传稿与 docs/images 素材，按当前仓库真实结构重写常规说明。 Co-authored-by: Cursor <cursoragent@cursor.com> Co-authored-by: multica-agent <github@multica.ai>	2026-09-01 21:58:35 +08:00