feat: 数据修正功能 + 盘后管道暂停实时行情防竞态 (#87)

* refactor: 「群机器人 Webhook」统一更名为「群推送 Webhook」

"群机器人"易与后续接入的"智能机器人(API 模式)"混淆。
该通道本质是单向往群聊推送消息, 更名为「群推送 Webhook」更准确。

涉及: 飞书/企业微信的 UI 标签、操作指引、后端文档字符串、
错误提示文案(代码逻辑/接口不变)。覆盖 6 个文件, 纯文案改动。

* feat: 数据修正功能 + 盘后管道暂停实时行情防竞态

数据修正/补数据:
- 数据页顶部新增「修正数据」按钮, 弹窗选起始日期重拉到今天
- 复用盘后管道全流程 (维表/A股日K/除权/enriched/指数), 仅日期由用户传入
- run_now() 加 override_start_date 参数, 注入 A股日K + 指数拉取起点
- 新增 /api/kline/repair_daily 端点 (异步 job + 进度轮询)
- 前端 RepairDailyPanel + DatePicker, 默认起始日期为30天前

实时行情暂停机制 (防写盘竞态):
- QuoteService 加 _paused 标志 + pause()/resume()/paused() 上下文管理器
- 盘后管道/数据修正运行期间自动暂停实时行情取数, 防止覆写同一批 parquet
- toggle 端点: 暂停态下禁止开启实时行情 (409)
- 前端开关: 暂停时 disabled + 显示「数据同步运行中，已临时暂停」
- 三处注入 pause: pipeline.py / kline.py(repair_daily) / daily_pipeline.py(定时)

.gitignore

@@ -65,6 +65,7 @@ backend/user_prompts/
 
 # ===== AI IDE =====
 .trae/
+.zcode/
 
 # ===== TypeScript build cache =====
 frontend/tsconfig*.tsbuildinfo

backend/app/api/kline.py

@@ -717,6 +717,90 @@ async def extend_history(request: Request):
         raise HTTPException(status_code=500, detail=str(e)) from e
 
 
+@router.post("/repair_daily")
+async def repair_daily(request: Request):
+    """修正 / 补全日K数据 — 从指定起始日期重拉到今天。
+
+    典型场景: 昨天没看盘 / 服务挂了,本地日K缺了若干天。
+    用户选起始日期,复用盘后管道全流程重拉 [start_date ~ 今天]。
+
+    body: { "start_date": "YYYY-MM-DD" }
+    返回 job_id,可轮询 /api/pipeline/jobs 查看进度。
+    """
+    import asyncio
+    import traceback as _tb
+    from datetime import date as _date
+    try:
+        body = await request.json()
+        raw = body.get("start_date")
+        if not raw:
+            raise HTTPException(status_code=400, detail="start_date 必填 (YYYY-MM-DD)")
+        try:
+            start_date = _date.fromisoformat(str(raw))
+        except ValueError:
+            raise HTTPException(status_code=400, detail="start_date 格式错误 (应为 YYYY-MM-DD)")
+
+        if start_date > _date.today():
+            raise HTTPException(status_code=400, detail="起始日期不能晚于今天")
+
+        repo = request.app.state.repo
+        capset = request.app.state.capabilities
+
+        from app.tickflow.capabilities import Cap
+        if not capset.has(Cap.KLINE_DAILY_BATCH):
+            raise HTTPException(status_code=403, detail="需要 Pro+ 权限 (batch K-line)")
+
+        from app.services.repair_daily import run_repair_daily
+        from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot
+        from app.api.data import invalidate_storage_cache
+
+        job_id, is_new = job_store.create()
+        if not is_new:
+            return {"status": "reused", "job_id": job_id}
+
+        async def task() -> None:
+            if not try_acquire_run_slot():
+                job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试")
+                return
+            loop = asyncio.get_event_loop()
+            qs = getattr(request.app.state, "quote_service", None)
+
+            def progress(stage: str, pct: int, msg: str,
+                         stage_pct: int | None = None, skip_log: bool = False) -> None:
+                job_store.progress(job_id, stage, pct, msg,
+                                   stage_pct=stage_pct, skip_log=skip_log)
+
+            def _run() -> dict:
+                # 修正运行期间暂停实时行情, 防止覆写同一批 parquet 竞态
+                if qs:
+                    with qs.paused():
+                        return run_repair_daily(repo, capset, start_date, on_progress=progress)
+                return run_repair_daily(repo, capset, start_date, on_progress=progress)
+
+            try:
+                job_store.start(job_id)
+                result = await loop.run_in_executor(_long_task_executor, _run)
+                if "error" in result:
+                    job_store.fail(job_id, result["error"])
+                else:
+                    job_store.succeed(job_id, result)
+                invalidate_storage_cache()
+            except Exception as e:
+                logger.exception("repair_daily failed: job_id=%s", job_id)
+                job_store.fail(job_id, str(e))
+                invalidate_storage_cache()
+            finally:
+                release_run_slot()
+
+        asyncio.create_task(task())
+        return {"status": "started", "job_id": job_id}
+    except HTTPException:
+        raise
+    except Exception as e:
+        logger.error("repair_daily error: %s\n%s", e, _tb.format_exc())
+        raise HTTPException(status_code=500, detail=str(e)) from e
+
+
 @router.post("/rebuild_enriched")
 async def rebuild_enriched(request: Request):
     """全量重算 enriched 表 — 不获取任何数据,仅基于已有 kline_daily + adj_factor 重算复权+指标。

backend/app/api/pipeline.py

@@ -44,6 +44,8 @@ async def run_now(request: Request) -> dict:
         if not try_acquire_run_slot():
             job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试")
             return
+        # 管道运行期间暂停实时行情取数, 防止覆写同一批 parquet 竞态
+        qs = getattr(request.app.state, "quote_service", None)
         try:
             job_store.start(job_id)
             loop = asyncio.get_event_loop()
@@ -52,10 +54,13 @@ async def run_now(request: Request) -> dict:
                          skip_log: bool = False) -> None:
                 job_store.progress(job_id, stage, pct, msg, stage_pct=stage_pct, skip_log=skip_log)
 
-            result = await loop.run_in_executor(
-                _long_task_executor,
-                lambda: daily_pipeline.run_now(repo, capset, on_progress=progress),
-            )
+            def _run() -> dict:
+                if qs:
+                    with qs.paused():
+                        return daily_pipeline.run_now(repo, capset, on_progress=progress)
+                return daily_pipeline.run_now(repo, capset, on_progress=progress)
+
+            result = await loop.run_in_executor(_long_task_executor, _run)
             job_store.succeed(job_id, result)
             invalidate_storage_cache()
             repo.refresh_cache()  # 刷新 Polars 缓存

backend/app/api/settings.py

@@ -662,6 +662,9 @@ def update_realtime_quotes(req: RealtimeQuotesPrefs, request: Request) -> dict:
         if qs:
             qs.disable()
         return {"realtime_quotes_enabled": False, "realtime_allowed": False}
+    if req.realtime_quotes_enabled and qs and qs.is_paused():
+        # 管道/数据修正运行期间禁止开启实时行情 — 防止写盘竞态
+        raise HTTPException(status_code=409, detail="数据同步运行中，实时行情已临时暂停，请稍后再开启")
     if req.realtime_quotes_enabled and qs and qs.realtime_mode() == "watchlist" and not preferences.get_realtime_watchlist_symbols():
         preferences.save({"realtime_quotes_enabled": False})
         return {"realtime_quotes_enabled": False, "realtime_allowed": True, "mode": "watchlist", "error": "watchlist_empty"}

backend/app/jobs/daily_pipeline.py

@@ -102,11 +102,15 @@ def run_now(
     repo: KlineRepository,
     capset: CapabilitySet,
     on_progress: ProgressCb | None = None,
+    override_start_date: _date | None = None,
 ) -> dict:
     """立即执行一次盘后管道,支持进度回调。
 
     跳过的 stage **不 emit**,避免前端把"无 capability"的卡片错误标记为 active/done。
     result 里带 skipped_stages 列表供前端展示。
+
+    override_start_date: 传入时强制走 batch 拉取分支,用该日期作为日K/除权/指数的
+        拉取起点(到今天),用于「数据修正/补数据」场景。None 时走原有自动判定逻辑。
     """
     emit = on_progress or _noop
     skipped: list[str] = []
@@ -127,6 +131,7 @@ def run_now(
     emit("resolve_universe", 10, f"标的池规模:{len(universe)} 只")
 
     # Step 1: 日 K 同步
+    #   override_start_date 传入 → 强制 batch 拉取 [override_start_date ~ today] (数据修正)
     #   付费档 + 今天有数据 → 实时行情接口拉一次覆写（1请求全市场）
     #   有历史数据 → batch K-line API 补齐缺口
     #   无任何数据 → batch K-line API 拉首次 1 年
@@ -135,15 +140,36 @@ def run_now(
     today = _date.today()
     today_exists = latest_daily and latest_daily >= today
     new_daily_days = 0
-    # 日K范围拉取的起点(分支3补缺口/分支4首次); 实时增量/跳过时为 None。
+    # 日K范围拉取的起点(分支3补缺口/分支4首次/数据修正); 实时增量/跳过时为 None。
     # 供 Step 1.5 除权因子回溯范围对齐: 范围拉取→用日K范围, 非范围→最近N天兜底。
     daily_range_start: _date | None = None
 
-    # A 股日K拉取开关(默认开);关闭时跳过日K同步,保留已有数据
+    # A 股日K拉取开关(默认开);关闭时跳过日K同步,保留已有数据。
+    # 数据修正(override_start_date)时即使关闭开关也强制拉取 — 修正就是来补数据的。
     pull_a_share = _prefs.get_pipeline_pull_a_share()
-    if not pull_a_share:
+    if not pull_a_share and not override_start_date:
         emit("sync_daily", 45, "已跳过 A 股日K同步(拉取内容未勾选)")
         logger.info("sync_daily: skipped (pipeline_pull_a_share=False)")
+    elif override_start_date:
+        # 数据修正: 强制用传入日期作起点 batch 拉取, 忽略实时行情覆写分支。
+        start_date = override_start_date
+        daily_range_start = start_date
+        emit("sync_daily", 12, f"获取日K [{start_date} ~ {today}]…")
+        logger.info("sync_daily: [%s ~ %s] repair/override", start_date, today)
+
+        def _daily_chunk_progress(cur: int, tot: int) -> None:
+            emit("sync_daily", 12 + int(33 * cur / tot),
+                 f"日K 批次 {cur}/{tot}", stage_pct=int(100 * cur / tot), skip_log=True)
+        written_daily = kline_sync.sync_and_persist_daily_batch(
+            universe, repo, capset,
+            start_date=_dt.combine(start_date, _dt.min.time()),
+            end_date=_dt.combine(today, _dt.min.time()),
+            on_chunk_done=_daily_chunk_progress,
+        )
+        gap_days = (today - start_date).days
+        new_daily_days = gap_days
+        emit("sync_daily", 45, f"日K 完成,覆盖 {gap_days} 天")
+        logger.info("sync_daily: [%s ~ %s] done, %d days", start_date, today, gap_days)
     elif today_exists and capset.has(Cap.QUOTE_POOL) and _prefs.get_daily_data_provider() == "tickflow":
         # 付费档:今天有数据(QuoteService 已落盘)→ 实时行情覆写,确保最新。
         # free/none 档无 quote.pool 能力,即便今天已有数据(如从 expert 降级),
@@ -358,7 +384,11 @@ def run_now(
                     d.name[5:] for d in index_dir.glob("date=*")
                     if d.is_dir() and d.name.startswith("date=")
                 ) if index_dir.exists() else []
-                index_start = _date.fromisoformat(index_dates[-1]) if index_dates else today - _td(days=365)
+                # 数据修正模式下用传入起点; 否则用本地指数最新日期补到今天
+                if override_start_date:
+                    index_start = override_start_date
+                else:
+                    index_start = _date.fromisoformat(index_dates[-1]) if index_dates else today - _td(days=365)
 
                 def _index_chunk(cur: int, tot: int) -> None:
                     emit("sync_index", 88, f"指数日K批次 {cur}/{tot}",
@@ -835,8 +865,14 @@ def start_scheduler(repo: KlineRepository, capset: CapabilitySet) -> AsyncIOSche
         # 旧 capset —— 否则 Key 中途过期/续费后, 调度管道仍按旧档位打端点。
         app_state = _get_app_state()
         capset_live = getattr(app_state, "capabilities", None) or capset
+        # 管道运行期间暂停实时行情取数, 防止覆写同一批 parquet 竞态
+        qs = getattr(app_state, "quote_service", None)
         try:
-            result = run_now(repo, capset_live, on_progress=on_progress)
+            if qs:
+                with qs.paused():
+                    result = run_now(repo, capset_live, on_progress=on_progress)
+            else:
+                result = run_now(repo, capset_live, on_progress=on_progress)
         finally:
             # 即便有阶段软失败(run_now 末尾抛 PipelineStageError), 已落盘的日K/enriched
             # 仍需刷进内存缓存, 否则 live_agg 基准列停留在旧交易日。放 finally 保证部分

backend/app/services/quote_service.py

@@ -27,6 +27,7 @@ import logging
 import threading
 import time
 from concurrent.futures import ThreadPoolExecutor
+from contextlib import contextmanager
 from datetime import date, time as dt_time
 
 import polars as pl
@@ -136,6 +137,10 @@ class QuoteService:
         self._fetch_lock = threading.Lock()
         self._running = False
         self._enabled = False      # 全局开关 (持久化到 preferences)
+        # 暂停态: 盘后管道/数据修正运行期间临时暂停取数, 防止与管道写同一批 parquet 竞态。
+        # 与 _enabled 不同 — pause 不改 preferences、不 stop 线程, 仅让轮询循环跳过取数;
+        # 进程重启后 _paused 归零, 从 preferences 恢复真实开关态, 无"假关闭"副作用。
+        self._paused = False
         self._interval = self.DEFAULT_INTERVAL
         self._thread: threading.Thread | None = None
         self._repo = None          # 延迟注入, 避免循环导入
@@ -210,6 +215,44 @@ class QuoteService:
         self.stop()
         logger.info("行情服务已关闭")
 
+    # ================================================================
+    # 临时暂停 (盘后管道/数据修正期间, 防止写盘竞态)
+    # ================================================================
+
+    def pause(self) -> None:
+        """临时暂停行情轮询取数 (不关闭线程、不改 preferences)。
+
+        用于盘后管道/数据修正运行期间, 防止实时行情覆写管道正在写的 parquet。
+        与 stop() 的区别: 线程继续存活但跳过 _fetch_quotes; preferences 开关态不变,
+        管道结束调用 resume() 即恢复。线程级检查, 即时生效, 无 join 等待。
+        """
+        self._paused = True
+        logger.info("行情轮询已临时暂停 (管道/修正运行中)")
+
+    def resume(self) -> None:
+        """恢复暂停的行情轮询取数 (对应 pause)。"""
+        self._paused = False
+        logger.info("行情轮询已恢复")
+
+    def is_paused(self) -> bool:
+        """是否处于临时暂停态 (管道运行期间)。"""
+        return self._paused
+
+    @contextmanager
+    def paused(self):
+        """上下文管理器: 进入时暂停轮询取数, 退出时(含异常)自动恢复。
+
+        供盘后管道/数据修正复用:
+            with quote_service.paused():
+                run_pipeline(...)
+        无论正常结束还是异常/crash, finally 都会 resume (除非进程直接被 kill)。
+        """
+        self.pause()
+        try:
+            yield
+        finally:
+            self.resume()
+
     def boot_check(self) -> None:
         """启动时检查 preferences，若 enabled 则自动启动。
 
@@ -391,6 +434,7 @@ class QuoteService:
         return {
             "enabled": self._enabled,
             "running": self._running,
+            "paused": self._paused,
             "mode": mode,
             "realtime_allowed": mode != "none",
             "watchlist_symbol_count": len(preferences.get_realtime_watchlist_symbols()),
@@ -420,21 +464,24 @@ class QuoteService:
     def _poll_loop(self) -> None:
         while self._running and self._enabled:
             try:
-                phase = self._market_phase()
-                if self._should_fetch_for_phase(phase):
-                    is_final = phase in {"morning_final", "close_final"}
-                    ok = self._fetch_quotes(final=is_final)
-                    if is_final:
-                        key = self._final_sync_key(phase)
-                        if key and ok:
-                            self._final_sync_done.add(key)
-                            self._final_sync_failed.pop(key, None)
-                            logger.info("%s 最终行情同步完成, 进入休盘态", "午休" if phase == "morning_final" else "收盘")
-                        elif key:
-                            self._final_sync_failed[key] = "fetch_failed"
-                            logger.warning("%s 最终行情同步失败, 将继续重试", "午休" if phase == "morning_final" else "收盘")
-                else:
-                    logger.debug("非轮询阶段(%s), 跳过行情轮询", phase)
+                # 管道/数据修正运行期间临时暂停取数, 防止与管道写同一批 parquet 竞态。
+                # 线程继续存活 + 分片 sleep, resume() 后即时恢复, 无需重启线程。
+                if not self._paused:
+                    phase = self._market_phase()
+                    if self._should_fetch_for_phase(phase):
+                        is_final = phase in {"morning_final", "close_final"}
+                        ok = self._fetch_quotes(final=is_final)
+                        if is_final:
+                            key = self._final_sync_key(phase)
+                            if key and ok:
+                                self._final_sync_done.add(key)
+                                self._final_sync_failed.pop(key, None)
+                                logger.info("%s 最终行情同步完成, 进入休盘态", "午休" if phase == "morning_final" else "收盘")
+                            elif key:
+                                self._final_sync_failed[key] = "fetch_failed"
+                                logger.warning("%s 最终行情同步失败, 将继续重试", "午休" if phase == "morning_final" else "收盘")
+                    else:
+                        logger.debug("非轮询阶段(%s), 跳过行情轮询", phase)
             except Exception as e:  # noqa: BLE001
                 logger.warning("行情轮询异常: %s", e)
 

backend/app/services/repair_daily.py

@@ -0,0 +1,54 @@
+"""修正 / 补全日K数据 — 完全复用盘后管道,只是日期范围由用户传入。
+
+典型场景: 昨天没看盘 / 服务挂了一天,本地日K缺了若干天。
+
+设计原则: 不重复盘后管道的任何逻辑。直接调用 daily_pipeline.run_now(),
+通过 override_start_date 参数把"自动算日期"换成"用户指定起点",
+其余 (维表 / A股日K / 除权因子 / enriched / 指数 / ETF / 错误兜底) 全部原样复用。
+这样修正功能与盘后管道永远保持一致,不会出现遗漏。
+
+落盘是 merge-upsert (按 symbol+date 去重 keep="last"), 新数据天然覆盖旧值,
+不会产生重复,也不需要先删分区。
+"""
+from __future__ import annotations
+
+import logging
+from collections.abc import Callable
+from datetime import date
+
+from app.tickflow.capabilities import CapabilitySet
+from app.tickflow.repository import KlineRepository
+
+logger = logging.getLogger(__name__)
+
+
+def run_repair_daily(
+    repo: KlineRepository,
+    capset: CapabilitySet,
+    start_date: date,
+    end_date: date | None = None,
+    on_progress: Callable | None = None,
+) -> dict:
+    """修正 / 补全数据 — 复用盘后管道,日期范围由用户指定。
+
+    通过 run_now(override_start_date=start_date) 把日K/除权/指数的拉取起点
+    统一设为 start_date (到今天),其余流程与盘后管道完全一致。
+
+    Args:
+        repo:        数据仓库
+        capset:      权限集
+        start_date:  用户选定的起始日期
+        end_date:    保留参数(目前盘后管道固定拉到今天, 此值未使用, 为接口兼容保留)
+        on_progress: 进度回调
+
+    Returns:
+        run_now() 的完整结果 dict。
+    """
+    today = date.today()
+    if start_date > today:
+        return {"error": "起始日期不能晚于今天"}
+
+    logger.info("repair_daily: run pipeline with override_start_date=%s", start_date)
+
+    from app.jobs.daily_pipeline import run_now
+    return run_now(repo, capset, on_progress=on_progress, override_start_date=start_date)

frontend/src/components/Layout.tsx

@@ -344,6 +344,8 @@ export function Layout() {
   const toggleQuote = useToggleRealtimeQuotes()
   const isRunning = quoteStatus?.running ?? false
   const isTrading = quoteStatus?.is_trading_hours ?? false
+  // 管道/数据修正运行期间实时行情被临时暂停 — 此时禁止开启
+  const isPaused = quoteStatus?.paused ?? false
   const tier = tierRank(caps?.label ?? '')
   const isNoneTier = tier < 0
   const isWatchlistMode = tier === 0
@@ -589,12 +591,13 @@ export function Layout() {
               </div>
               <button
                 onClick={() => handleToggle(!realtimeEnabled)}
-                disabled={toggleQuote.isPending}
+                disabled={toggleQuote.isPending || isPaused}
+                title={isPaused ? '数据同步运行中，实时行情已临时暂停' : undefined}
                 className={`relative inline-flex h-4 w-7 items-center rounded-full shrink-0 transition-colors duration-200 ${
                   realtimeEnabled
                     ? 'bg-accent shadow-[0_0_6px_rgba(59,130,246,0.3)]'
                     : 'bg-elevated'
-                } ${toggleQuote.isPending ? 'opacity-50' : 'cursor-pointer'}`}
+                } ${toggleQuote.isPending || isPaused ? 'opacity-50' : 'cursor-pointer'}`}
               >
                 <span className={`inline-block h-3 w-3 rounded-full bg-white shadow-sm transition-transform duration-200 ${
                   realtimeEnabled ? 'translate-x-[14px]' : 'translate-x-0.5'
@@ -618,7 +621,9 @@ export function Layout() {
                   </button>
                 </div>
               )}
-              {isRunning && isTrading ? (
+              {isPaused ? (
+                <div className="text-warning/80">数据同步运行中，实时行情已临时暂停</div>
+              ) : isRunning && isTrading ? (
                 <div className="text-accent">行情运行中</div>
               ) : realtimeEnabled && !isTrading ? (
                 <div className="text-warning/70">非交易时段，将在交易时间自动开启</div>

frontend/src/components/data/RepairDailyPanel.tsx

@@ -0,0 +1,107 @@
+import { useState } from 'react'
+import { useMutation, useQueryClient } from '@tanstack/react-query'
+import { Loader2 } from 'lucide-react'
+import { api } from '@/lib/api'
+import { QK } from '@/lib/queryKeys'
+import { DatePicker } from '@/components/DatePicker'
+
+function pad(n: number) { return String(n).padStart(2, '0') }
+function todayStr() {
+  const d = new Date()
+  return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())}`
+}
+/** 往前推 N 天 */
+function daysAgo(n: number): string {
+  const d = new Date()
+  d.setDate(d.getDate() - n)
+  return `${d.getFullYear()}-${pad(d.getMonth() + 1)}-${pad(d.getDate())}`
+}
+
+export function RepairDailyPanel({ caps, isRunning, latestDate, onStart }: {
+  caps: { label: string; capabilities: Record<string, { rpm: number | null; batch: number | null; subscribe: number | null }> } | undefined
+  isRunning: boolean
+  latestDate: string | null
+  onStart: () => void
+}) {
+  const qc = useQueryClient()
+  const hasBatchCap = !!caps?.capabilities?.['kline.daily.batch']
+
+  // 默认起始日期: 最新数据往前推 30 天 (兼顾补缺口 + 复核近期数据, 成本不高)
+  const [startDate, setStartDate] = useState(daysAgo(30))
+
+  const repair = useMutation({
+    mutationFn: () => api.repairDaily(startDate),
+    onSuccess: () => {
+      onStart()
+      qc.invalidateQueries({ queryKey: QK.pipelineJobs })
+    },
+  })
+
+  const today = todayStr()
+  const canSubmit = hasBatchCap && !isRunning && !repair.isPending && startDate <= today
+  const gapDays = latestDate
+    ? Math.max(0, Math.round((Date.parse(today) - Date.parse(latestDate)) / 86400000))
+    : null
+
+  return (
+    <div className="space-y-3">
+      <div className="rounded-card bg-accent/8 border border-accent/20 p-3 space-y-1.5">
+        <div className="text-xs text-foreground">
+          当数据出现缺口时(漏跑、停服),从这里重拉选定区间到今天的全部数据并重算。
+        </div>
+        <div className="text-[10px] text-muted leading-relaxed">
+          完整复用盘后管道流程 (A股日K · 除权因子 · 指标重算 · 指数),新数据按 (个股, 日期) 覆盖旧值,不会重复,也无需先清除。
+        </div>
+      </div>
+
+      {/* 起始日期 — 用户主要操作,放上面 */}
+      <div className="flex items-center justify-between gap-2">
+        <span className="text-xs text-secondary">起始日期</span>
+        <DatePicker
+          value={startDate}
+          onChange={setStartDate}
+          max={today}
+          align="right"
+          buttonClassName="font-mono text-xs"
+        />
+      </div>
+
+      {/* 本地最新数据 — 参考信息,放下面 */}
+      <div className="flex items-center justify-between gap-2">
+        <span className="text-xs text-muted">本地最新数据</span>
+        <span className="font-mono text-xs text-secondary">
+          {latestDate ?? '—'}
+          {gapDays !== null && gapDays > 0 && (
+            <span className="ml-2 text-warning/90">已落后 {gapDays} 天</span>
+          )}
+        </span>
+      </div>
+
+      <div className="text-[10px] text-muted -mt-1">
+        将重拉 <span className="font-mono text-secondary">{startDate}</span>
+        {' → '}<span className="font-mono text-secondary">{today}</span>(今天) 的 A股日K · 除权 · 指数并重算指标
+      </div>
+
+      <button
+        onClick={() => repair.mutate()}
+        disabled={!canSubmit}
+        className="w-full inline-flex items-center justify-center gap-1.5 px-3 py-1.5 rounded-btn bg-accent/90 text-base text-xs font-medium hover:bg-accent disabled:opacity-40 disabled:pointer-events-none transition-colors duration-150"
+      >
+        {repair.isPending ? (
+          <>
+            <Loader2 className="h-3 w-3 animate-spin" />
+            请求中…
+          </>
+        ) : (
+          <>开始修正</>
+        )}
+      </button>
+
+      {!hasBatchCap && (
+        <span className="block text-[10px] text-warning/80 bg-warning/8 rounded px-1.5 py-px font-medium text-center">
+          需 Pro+ 权限
+        </span>
+      )}
+    </div>
+  )
+}

frontend/src/lib/api.ts

@@ -957,6 +957,7 @@ export const api = {
     request<{
       enabled: boolean
       running: boolean
+      paused?: boolean
       mode?: 'none' | 'watchlist' | 'full_market'
       realtime_allowed?: boolean
       interval_s: number
@@ -1205,6 +1206,11 @@ export const api = {
       method: 'POST',
       body: JSON.stringify({ value, unit }),
     }),
+  repairDaily: (startDate: string) =>
+    request<{ status: string; job_id: string }>('/api/kline/repair_daily', {
+      method: 'POST',
+      body: JSON.stringify({ start_date: startDate }),
+    }),
   extendMinuteHistory: (value: number, unit: 'day' | 'month') =>
     request<{ status: string; job_id: string }>('/api/kline/extend_minute_history', {
       method: 'POST',

frontend/src/pages/Data.tsx

@@ -15,6 +15,7 @@ import {
   SlidersHorizontal,
   AlertTriangle,
   Info,
+  WandSparkles,
 } from 'lucide-react'
 import { Link } from 'react-router-dom'
 import { EndpointTestDialog } from '@/components/EndpointTestDialog'
@@ -39,6 +40,7 @@ import { SectionTitle, HistoryRow } from '@/components/data/SectionTitle'
 import { SettingsModal } from '@/components/data/SettingsModal'
 import { ScheduleEditor } from '@/components/data/ScheduleEditor'
 import { ExtendHistoryPanel } from '@/components/data/ExtendHistoryPanel'
+import { RepairDailyPanel } from '@/components/data/RepairDailyPanel'
 import { EnrichedRebuildPanel } from '@/components/data/EnrichedRebuildPanel'
 import { MinuteSyncConfig } from '@/components/data/MinuteSyncConfig'
 import { PipelineScopeConfig } from '@/components/data/PipelineScopeConfig'
@@ -131,6 +133,7 @@ export function Data() {
   const [schemaTable, setSchemaTable] = useState<string | null>(null)
   const [showEndpointTest, setShowEndpointTest] = useState(false)
   const [showCreateExt, setShowCreateExt] = useState(false)
+  const [showRepair, setShowRepair] = useState(false)
   const [editingExt, setEditingExt] = useState<ExtDataConfig | null>(null)
   const [indexBatchInput, setIndexBatchInput] = useState('100')
 
@@ -557,6 +560,14 @@ export function Data() {
               <CheckSquare className="h-3.5 w-3.5" />
               数据范围
             </button>
+            <button
+              onClick={() => setShowRepair(true)}
+              disabled={!hasData || isRunning}
+              className="inline-flex items-center gap-1 px-2 py-1 rounded-btn text-secondary hover:text-accent hover:bg-accent/8 text-xs transition-colors duration-150 disabled:opacity-40 disabled:pointer-events-none"
+            >
+              <WandSparkles className="h-3.5 w-3.5" />
+              修正数据
+            </button>
             <div className="w-px h-4 bg-border" />
             <div className="flex items-center gap-1.5">
               <button
@@ -931,6 +942,19 @@ export function Data() {
         )}
       </AnimatePresence>
 
+      <AnimatePresence>
+        {showRepair && (
+          <SettingsModal title="日 K · 修正 / 补数据" onClose={() => setShowRepair(false)}>
+            <RepairDailyPanel
+              caps={caps.data}
+              isRunning={!!activeJobId}
+              latestDate={s?.daily?.latest_date ?? null}
+              onStart={() => setShowRepair(false)}
+            />
+          </SettingsModal>
+        )}
+      </AnimatePresence>
+
       <AnimatePresence>
         {openSettings === 'enriched' && (
           <SettingsModal title="Enriched · 计算设置" onClose={() => setOpenSettings(null)}>

frontend/src/pages/settings/Monitoring.tsx

@@ -60,6 +60,8 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } =
   const indicesPinned = prefs?.indices_nav_pinned ?? true
   const isRunning = quoteStatus?.running ?? false
   const isTrading = quoteStatus?.is_trading_hours ?? false
+  // 管道/数据修正运行期间实时行情被临时暂停 — 此时禁止开启
+  const isPaused = quoteStatus?.paused ?? false
   const interval = intervalData?.interval ?? 10
   const minInterval = intervalData?.min_interval ?? 5
   const maxInterval = intervalData?.max_interval ?? 60
@@ -244,9 +246,15 @@ export function SettingsMonitoringPanel({ highlight }: { highlight?: string } =
         <Card icon={Activity} title="行情轮询">
           <ToggleRow
             label="实时行情"
-            desc={isRunning && isTrading ? '运行中' : isRunning ? '运行中 (非交易时段)' : '已关闭'}
+            desc={
+              isPaused ? '数据同步运行中，已临时暂停'
+              : isRunning && isTrading ? '运行中'
+              : isRunning ? '运行中 (非交易时段)'
+              : '已关闭'
+            }
             checked={realtimeEnabled}
             onChange={handleToggleQuote}
+            disabled={isPaused}
           />
 
           <div className="mt-3 pt-3 border-t border-border">
@@ -623,12 +631,14 @@ function ToggleRow({
   checked,
   onChange,
   icon: Icon,
+  disabled,
 }: {
   label: string
   desc: string
   checked: boolean
   onChange: (v: boolean) => void
   icon?: React.ComponentType<{ className?: string }>
+  disabled?: boolean
 }) {
   return (
     <div className="flex items-center justify-between gap-4 py-2">
@@ -640,10 +650,11 @@ function ToggleRow({
         </div>
       </div>
       <button
-        onClick={() => onChange(!checked)}
+        onClick={() => !disabled && onChange(!checked)}
+        disabled={disabled}
         className={`relative inline-flex h-5 w-9 items-center rounded-full shrink-0 transition-colors duration-200 ${
           checked ? 'bg-accent' : 'bg-elevated'
-        }`}
+        } ${disabled ? 'opacity-40 cursor-not-allowed' : 'cursor-pointer'}`}
       >
         <span
           className={`inline-block h-3.5 w-3.5 rounded-full bg-white shadow-sm transition-transform duration-200 ${