fix: 扩展数据拉取修复与连板梯队queryKey修复

2026-09-12 16:44:15 +08:00 · 2026-07-01 23:02:37 +08:00
parent 5a9fa9b255
commit e7615fb9e6
6 changed files with 426 additions and 153 deletions
@@ -575,6 +575,13 @@ def configure_pull(request: Request, config_id: str, body: PullConfigReq):
    # 刷新调度器
    pull_scheduler.refresh(_data_dir(request))

+    # 关闭定时拉取时清理残留的 next_run, 避免前端展示一个永不执行的"下次"
+    if not config.pull.enabled:
+        cleared = store.get(config_id)
+        if cleared and cleared.pull and cleared.pull.next_run:
+            cleared.pull.next_run = None
+            store.upsert(cleared)
+
    return {"status": "ok", "pull": config.pull.to_dict()}


@@ -630,8 +637,25 @@ async def run_pull(request: Request, config_id: str):
    try:
        n, d = await fetch_and_ingest(config, _data_dir(request))
        _refresh_views(request)
+        # 写回执行状态, 让前端"上次执行"面板立即反映
+        updated = store.get(config_id)
+        if updated and updated.pull:
+            from datetime import datetime, timezone
+            updated.pull.last_run = datetime.now(timezone.utc).isoformat()
+            updated.pull.last_status = "success"
+            updated.pull.last_message = f"{n} rows @ {d}"
+            updated.pull.last_rows = n
+            store.upsert(updated)
        return {"status": "ok", "rows": n, "date": d}
    except Exception as e:
+        # 失败也写回状态, 记录错误信息
+        failed = store.get(config_id)
+        if failed and failed.pull:
+            from datetime import datetime, timezone
+            failed.pull.last_run = datetime.now(timezone.utc).isoformat()
+            failed.pull.last_status = "error"
+            failed.pull.last_message = str(e)[:200]
+            store.upsert(failed)
        raise HTTPException(400, f"拉取失败: {e}") from e


@@ -38,6 +38,7 @@ class PullConfig:
        "url", "method", "headers", "body", "response_path",
        "field_map", "schedule_minutes", "enabled",
        "last_run", "last_status", "last_message", "last_rows",
+        "next_run",
    )

    def __init__(
@@ -54,6 +55,7 @@ class PullConfig:
        last_status: str | None = None,
        last_message: str | None = None,
        last_rows: int | None = None,
+        next_run: str | None = None,
    ) -> None:
        self.url = url
        self.method = method              # GET | POST
@@ -67,6 +69,7 @@ class PullConfig:
        self.last_status = last_status      # "success" | "error"
        self.last_message = last_message
        self.last_rows = last_rows
+        self.next_run = next_run            # 下次预计运行 (ISO, 调度器写入)

    def to_dict(self) -> dict:
        return {
@@ -82,6 +85,7 @@ class PullConfig:
            "last_status": self.last_status,
            "last_message": self.last_message,
            "last_rows": self.last_rows,
+            "next_run": self.next_run,
        }

    @classmethod
@@ -101,6 +105,7 @@ class PullConfig:
            last_status=d.get("last_status"),
            last_message=d.get("last_message"),
            last_rows=d.get("last_rows"),
+            next_run=d.get("next_run"),
        )


@@ -407,8 +412,10 @@ def write_ext_parquet(
                existing = pl.read_parquet(out_path)
                key = "symbol" if "symbol" in df.columns else df.columns[0]
                df = pl.concat([existing, df]).unique(subset=[key], keep="last")
-            except Exception:
-                pass
+            except Exception as e:
+                # schema 不一致 (列不同) 时 concat 失败 → 直接用新 df 覆盖。
+                # 记日志而非静默吞掉, 便于排查"数据结构错乱"类问题。
+                logger.warning("扩展表 %s 合并去重失败, 将覆盖写入: %s", config.id, e)
    else:
        # 时序: timeseries/ 下按日期分区
        out_dir = cfg_dir / "timeseries" / f"date={snap}"
@@ -421,8 +428,8 @@ def write_ext_parquet(
                existing = pl.read_parquet(out_path)
                key = "symbol" if "symbol" in df.columns else df.columns[0]
                df = pl.concat([existing, df]).unique(subset=[key], keep="last")
-            except Exception:
-                pass
+            except Exception as e:
+                logger.warning("扩展表 %s 合并去重失败, 将覆盖写入: %s", config.id, e)

    df = cast_df_to_schema(df, config.fields)
    df.write_parquet(out_path)
@@ -75,6 +75,19 @@ def _apply_field_map(rows: list[dict], field_map: dict[str, str]) -> list[dict]:
 # 拉取执行
 # ---------------------------------------------------------------------------

+def _apply_preset_flatten(config_id: str, rows: list[dict]) -> list[dict]:
+    """对内置预设 (概念/行业) 应用结构转换, 与 fetch_preset 保持一致。
+
+    延迟导入避免与 ext_presets 形成循环依赖。
+    非预设 id 原样返回。
+    """
+    if config_id not in ("ext_gn_ths", "ext_hy_ths"):
+        return rows
+    from app.services.ext_presets import _flatten_concept_rows, _flatten_industry_rows
+    flatten = _flatten_concept_rows if config_id == "ext_gn_ths" else _flatten_industry_rows
+    return flatten(rows)
+
+
 async def fetch_and_ingest(
    config: ExtConfig,
    data_dir,
@@ -111,6 +124,12 @@ async def fetch_and_ingest(
    if not rows:
        raise ValueError("提取到的行数为 0")

+    # 内置预设 (概念/行业): 应用结构转换, 让产出 schema 与分析页一致。
+    # 否则 raw 接口列 (concepts/industries 数组、name) 会直接覆盖正确的 part.parquet,
+    # 导致分析页因找不到维度字段 (所属概念/所属同花顺行业) 而"数据消失"。
+    # 见 ext_presets._flatten_* —— 手动拉取 / 定时拉取都必须走同一套转换。
+    rows = _apply_preset_flatten(config.id, rows)
+
    # 字段映射
    rows = _apply_field_map(rows, pull.field_map)

@@ -129,21 +148,46 @@ async def fetch_and_ingest(
 # ---------------------------------------------------------------------------

 class PullScheduler:
-    """后台调度器：为每个启用了 pull 的 ExtConfig 维护定时任务。"""
+    """后台调度器：为每个启用了 pull 的 ExtConfig 维护定时任务。
+
+    线程安全说明:
+      refresh()/stop() 可能从主事件循环 (lifespan startup) 或同步路由的
+      worker 线程 (configure_pull 是 def 而非 async def, FastAPI 丢进线程池)
+      调用。worker 线程里没有 running loop, 直接 asyncio.create_task 会抛
+      "no running event loop"。因此对 task 的增删一律通过
+      call_soon_threadsafe 提交到主循环执行 —— 同一套代码两种调用场景都安全。
+    """

    def __init__(self) -> None:
        self._tasks: dict[str, asyncio.Task] = {}
        self._running = False
-        self._lock = threading.Lock()
+        self._loop: asyncio.AbstractEventLoop | None = None

    def start(self, data_dir) -> None:
-        """启动调度（在 lifespan startup 调用）。"""
+        """启动调度（在 lifespan startup 调用，主事件循环内）。"""
        self._running = True
        self._data_dir = data_dir
+        try:
+            self._loop = asyncio.get_running_loop()
+        except RuntimeError:
+            self._loop = None
        logger.info("PullScheduler started")

+    def _submit(self, fn, *args) -> None:
+        """把一个 callable 提交到主事件循环执行 (线程安全)。
+
+        startup 在主循环内调用时 fn 立即排队; worker 线程调用时跨线程排队。
+        两者都通过 call_soon_threadsafe, 保证 _tasks 字典的读写只在主循环里发生。
+        """
+        loop = self._loop
+        if loop is None or loop.is_closed():
+            raise RuntimeError(
+                "PullScheduler: 事件循环不可用 (start() 未在事件循环中调用?)"
+            )
+        loop.call_soon_threadsafe(fn, *args)
+
    def stop(self) -> None:
-        """停止所有任务。"""
+        """停止所有任务 (从 shutdown 调用)。"""
        self._running = False
        for task in self._tasks.values():
            task.cancel()
@@ -151,47 +195,61 @@ class PullScheduler:
        logger.info("PullScheduler stopped")

    def refresh(self, data_dir) -> None:
-        """重新加载配置，更新调度任务（增/删/改）。"""
+        """重新加载配置，更新调度任务（增/删/改）。线程安全。"""
        self._data_dir = data_dir
        store = ExtConfigStore(data_dir)
        configs = store.load_all()

        active_ids: set[str] = set()
+        new_configs: list[ExtConfig] = []

        for config in configs:
            if not config.pull or not config.pull.enabled or not config.pull.url:
                continue
            active_ids.add(config.id)
            if config.id not in self._tasks:
-                # 新增调度
-                task = asyncio.create_task(self._run_loop(config))
-                self._tasks[config.id] = task
-                logger.info("PullScheduler: scheduled %s (every %d min)", config.id, config.pull.schedule_minutes)
+                new_configs.append(config)

-        # 移除不再活跃的
-        for cid in list(self._tasks):
-            if cid not in active_ids:
-                self._tasks[cid].cancel()
-                del self._tasks[cid]
-                logger.info("PullScheduler: removed %s", cid)
+        # 需要移除的 id (快照当前 task 字典的键, 避免遍历时改字典)
+        remove_ids = [cid for cid in list(self._tasks) if cid not in active_ids]
+
+        # 所有对 _tasks 的修改都提交到主循环里执行, 保证线程安全
+        def _apply() -> None:
+            for config in new_configs:
+                if config.id not in self._tasks:  # 二次校验, 防重复
+                    self._tasks[config.id] = self._loop.create_task(
+                        self._run_loop(config)
+                    )
+                    logger.info(
+                        "PullScheduler: scheduled %s (every %d min)",
+                        config.id, config.pull.schedule_minutes,
+                    )
+            for cid in remove_ids:
+                task = self._tasks.pop(cid, None)
+                if task is not None:
+                    task.cancel()
+                    logger.info("PullScheduler: removed %s", cid)
+
+        self._submit(_apply)

    async def _run_loop(self, config: ExtConfig) -> None:
-        """单个配置的定时拉取循环。"""
+        """单个配置的定时拉取循环。
+
+        策略: 启用后立即执行一次, 之后按 interval 循环。
+        每次循环重读最新配置 (fresh), interval 取自 fresh.pull.schedule_minutes,
+        这样用户中途修改间隔也能立即生效 (无需重启)。
+        """
        try:
            while self._running:
-                pull = config.pull
-                if not pull:
-                    break
-                interval = max(pull.schedule_minutes * 60, 60)  # 至少 60s
-                await asyncio.sleep(interval)
-                if not self._running:
+                # 每轮重读最新配置 — 用户可能修改了 url / interval / enabled
+                store = ExtConfigStore(self._data_dir)
+                fresh = store.get(config.id)
+                if not fresh or not fresh.pull or not fresh.pull.enabled:
                    break
+                pull = fresh.pull
+
+                # 先执行一次 (启用即拉取, 让用户立刻看到生效)
                try:
-                    # 重新加载最新配置（用户可能中途修改）
-                    store = ExtConfigStore(self._data_dir)
-                    fresh = store.get(config.id)
-                    if not fresh or not fresh.pull or not fresh.pull.enabled:
-                        break
                    n, d = await fetch_and_ingest(fresh, self._data_dir)
                    fresh.pull.last_run = datetime.now(timezone.utc).isoformat()
                    fresh.pull.last_status = "success"
@@ -200,14 +258,79 @@ class PullScheduler:
                    store.upsert(fresh)
                    logger.info("PullScheduler: %s success, %d rows", config.id, n)
                except Exception as e:
-                    store = ExtConfigStore(self._data_dir)
-                    fresh = store.get(config.id)
-                    if fresh and fresh.pull:
-                        fresh.pull.last_run = datetime.now(timezone.utc).isoformat()
-                        fresh.pull.last_status = "error"
-                        fresh.pull.last_message = str(e)[:200]
-                        store.upsert(fresh)
+                    fresh2 = store.get(config.id)
+                    if fresh2 and fresh2.pull:
+                        fresh2.pull.last_run = datetime.now(timezone.utc).isoformat()
+                        fresh2.pull.last_status = "error"
+                        fresh2.pull.last_message = str(e)[:200]
+                        store.upsert(fresh2)
                    logger.warning("PullScheduler: %s error: %s", config.id, e)
+
+                # 间隔取自最新配置 (每次重新读取, 修复改间隔不生效)
+                interval = max(pull.schedule_minutes * 60, 60)  # 至少 60s
+                # 预告下次运行时间, 供前端展示
+                next_dt = datetime.now(timezone.utc).timestamp() + interval
+                latest = store.get(config.id)
+                if latest and latest.pull:
+                    latest.pull.next_run = datetime.fromtimestamp(
+                        next_dt, tz=timezone.utc
+                    ).isoformat()
+                    store.upsert(latest)
+
+                await asyncio.sleep(interval)
+                if not self._running:
+                    break
+        except asyncio.CancelledError:
+            pass
+
+    async def _run_loop(self, config: ExtConfig) -> None:
+        """单个配置的定时拉取循环。
+
+        策略: 启用后立即执行一次, 之后按 interval 循环。
+        每次循环重读最新配置 (fresh), interval 取自 fresh.pull.schedule_minutes,
+        这样用户中途修改间隔也能立即生效 (无需重启)。
+        """
+        try:
+            while self._running:
+                # 每轮重读最新配置 — 用户可能修改了 url / interval / enabled
+                store = ExtConfigStore(self._data_dir)
+                fresh = store.get(config.id)
+                if not fresh or not fresh.pull or not fresh.pull.enabled:
+                    break
+                pull = fresh.pull
+
+                # 先执行一次 (启用即拉取, 让用户立刻看到生效)
+                try:
+                    n, d = await fetch_and_ingest(fresh, self._data_dir)
+                    fresh.pull.last_run = datetime.now(timezone.utc).isoformat()
+                    fresh.pull.last_status = "success"
+                    fresh.pull.last_message = f"{n} rows @ {d}"
+                    fresh.pull.last_rows = n
+                    store.upsert(fresh)
+                    logger.info("PullScheduler: %s success, %d rows", config.id, n)
+                except Exception as e:
+                    fresh2 = store.get(config.id)
+                    if fresh2 and fresh2.pull:
+                        fresh2.pull.last_run = datetime.now(timezone.utc).isoformat()
+                        fresh2.pull.last_status = "error"
+                        fresh2.pull.last_message = str(e)[:200]
+                        store.upsert(fresh2)
+                    logger.warning("PullScheduler: %s error: %s", config.id, e)
+
+                # 间隔取自最新配置 (每次重新读取, 修复改间隔不生效)
+                interval = max(pull.schedule_minutes * 60, 60)  # 至少 60s
+                # 预告下次运行时间, 供前端展示
+                next_dt = datetime.now(timezone.utc).timestamp() + interval
+                latest = store.get(config.id)
+                if latest and latest.pull:
+                    latest.pull.next_run = datetime.fromtimestamp(
+                        next_dt, tz=timezone.utc
+                    ).isoformat()
+                    store.upsert(latest)
+
+                await asyncio.sleep(interval)
+                if not self._running:
+                    break
        except asyncio.CancelledError:
            pass

@@ -1,6 +1,7 @@
 import { useState } from 'react'
-import { Loader2, Search, RefreshCw, Check } from 'lucide-react'
+import { Loader2, Search, Check, Clock, Zap, Settings2, AlertCircle, CheckCircle2, Calendar } from 'lucide-react'
 import { api, type ExtDataConfig } from '@/lib/api'
+import { toast } from '@/components/Toast'

 export function ExtDataPullPanel({ config, onSaved }: {
  config: ExtDataConfig
@@ -22,157 +23,284 @@ export function ExtDataPullPanel({ config, onSaved }: {
  const [saving, setSaving] = useState(false)
  const [testing, setTesting] = useState(false)
  const [running, setRunning] = useState(false)
+  const [runResult, setRunResult] = useState<{ rows: number; date: string } | null>(null)
  const [testResult, setTestResult] = useState<{ total_rows: number; preview: Record<string, unknown>[]; has_symbol: boolean } | null>(null)
  const [error, setError] = useState('')

-  const handleSave = () => {
-    let headers: Record<string, string> | undefined
-    if (headerStr.trim()) {
-      try { headers = JSON.parse(headerStr) }
-      catch { setError('Headers 不是有效 JSON'); return }
-    }
-    let field_map: Record<string, string> | undefined
-    if (fieldMapStr.trim()) {
-      try { field_map = JSON.parse(fieldMapStr) }
-      catch { setError('字段映射不是有效 JSON'); return }
-    }
-    setSaving(true); setError('')
-    api.extDataPullConfig(config.id, {
+  // 解析 JSON 输入, 失败时设置 error 并返回 null
+  const parseJson = (str: string, label: string): Record<string, string> | undefined | null => {
+    if (!str.trim()) return undefined
+    try { return JSON.parse(str) }
+    catch { setError(`${label} 不是有效 JSON`); return null }
+  }
+
+  // 构建保存 payload (复用当前编辑态), enabledOverride 用于开关自动保存
+  const buildPayload = (enabledOverride?: boolean) => {
+    const headers = parseJson(headerStr, 'Headers')
+    if (headers === null) return null
+    const field_map = parseJson(fieldMapStr, '字段映射')
+    if (field_map === null) return null
+    return {
      url, method, headers, body: body || undefined,
      response_path: responsePath, field_map,
-      schedule_minutes: schedule, enabled,
-    }).then(() => onSaved())
+      schedule_minutes: schedule, enabled: enabledOverride ?? enabled,
+    }
+  }
+
+  const handleSave = (silent = false) => {
+    const payload = buildPayload()
+    if (!payload) return
+    setSaving(true); setError('')
+    api.extDataPullConfig(config.id, payload)
+      .then(() => {
+        onSaved()
+        if (!silent) toast('配置已保存', 'success')
+      })
      .catch(e => setError(e.message || '保存失败'))
      .finally(() => setSaving(false))
  }

  const handleTest = () => {
    setTesting(true); setError(''); setTestResult(null)
-    let headers: Record<string, string> | undefined
-    if (headerStr.trim()) {
-      try { headers = JSON.parse(headerStr) }
-      catch { setError('Headers 不是有效 JSON'); setTesting(false); return }
-    }
-    let field_map: Record<string, string> | undefined
-    if (fieldMapStr.trim()) {
-      try { field_map = JSON.parse(fieldMapStr) }
-      catch { setError('字段映射不是有效 JSON'); setTesting(false); return }
-    }
-    api.extDataPullConfig(config.id, {
-      url, method, headers, body: body || undefined,
-      response_path: responsePath, field_map,
-      schedule_minutes: schedule, enabled,
-    }).then(() => api.extDataPullTest(config.id))
+    const payload = buildPayload()
+    if (!payload) { setTesting(false); return }
+    api.extDataPullConfig(config.id, payload)
+      .then(() => api.extDataPullTest(config.id))
      .then(r => { setTestResult(r); onSaved() })
      .catch(e => setError(e.message || '测试失败'))
      .finally(() => setTesting(false))
  }

  const handleRun = () => {
-    setRunning(true); setError('')
+    setRunning(true); setError(''); setRunResult(null)
    api.extDataPullRun(config.id)
-      .then(() => onSaved())
+      .then(r => {
+        setRunResult({ rows: r.rows, date: r.date })
+        onSaved()
+        toast(`拉取成功 · ${r.rows} 行`, 'success')
+      })
      .catch(e => setError(e.message || '执行失败'))
      .finally(() => setRunning(false))
  }

+  // 开关 toggle: 自动保存全量配置 (切换 enabled), 后端 refresh 后立即首次拉取
+  const [toggling, setToggling] = useState(false)
+  const handleToggle = (next: boolean) => {
+    if (toggling) return
+    if (next && !url.trim()) {
+      toast('请先填写拉取 URL', 'error')
+      return
+    }
+    const payload = buildPayload(next)
+    if (!payload) return
+    setToggling(true); setError(''); setEnabled(next)
+    api.extDataPullConfig(config.id, payload)
+      .then(() => {
+        onSaved()
+        toast(next ? '定时拉取已启用 · 立即执行首次拉取' : '定时拉取已关闭', 'success')
+      })
+      .catch(e => {
+        setEnabled(!next)  // 回滚
+        setError(e.message || '切换失败')
+      })
+      .finally(() => setToggling(false))
+  }
+
+  // 格式化时间显示
+  const fmtTime = (iso: string | null | undefined) => {
+    if (!iso) return null
+    const d = new Date(iso)
+    if (isNaN(d.getTime())) return null
+    const mm = String(d.getMonth() + 1).padStart(2, '0')
+    const dd = String(d.getDate()).padStart(2, '0')
+    const hh = String(d.getHours()).padStart(2, '0')
+    const mi = String(d.getMinutes()).padStart(2, '0')
+    return `${mm}-${dd} ${hh}:${mi}`
+  }
+
  return (
-    <div className="space-y-2.5">
-      <div className="flex gap-1.5">
-        <select
-          value={method} onChange={e => setMethod(e.target.value)}
-          className="shrink-0 rounded-md border border-border bg-elevated px-2 py-1.5 text-[11px] text-foreground"
-        >
-          <option value="GET">GET</option>
-          <option value="POST">POST</option>
-        </select>
-        <input
-          value={url} onChange={e => setUrl(e.target.value)}
-          placeholder="https://api.example.com/data"
-          className="flex-1 min-w-0 rounded-md border border-border bg-elevated px-2.5 py-1.5 text-[11px] font-mono text-foreground placeholder:text-muted/50"
-        />
-      </div>
+    <div className="space-y-3">
+      {/* ===== 分区 ①: 请求配置 ===== */}
+      <div className="space-y-2">
+        <div className="flex items-center gap-1.5 text-[11px] font-medium text-secondary">
+          <Settings2 className="h-3 w-3 text-muted" />
+          <span>请求配置</span>
+        </div>

-      <div>
-        <div className="text-[10px] text-muted mb-0.5">Headers (JSON，可选)</div>
-        <textarea
-          value={headerStr} onChange={e => setHeaderStr(e.target.value)}
-          placeholder='{"Authorization": "Bearer xxx"}'
-          rows={2}
-          className="w-full rounded-md border border-border bg-elevated px-2.5 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40 resize-none"
-        />
-      </div>
+        <div className="flex gap-1.5">
+          <select
+            value={method} onChange={e => setMethod(e.target.value)}
+            className="shrink-0 rounded-btn border border-border bg-elevated px-2 py-1.5 text-[11px] text-foreground"
+          >
+            <option value="GET">GET</option>
+            <option value="POST">POST</option>
+          </select>
+          <input
+            value={url} onChange={e => setUrl(e.target.value)}
+            placeholder="https://api.example.com/data"
+            className="flex-1 min-w-0 rounded-btn border border-border bg-elevated px-2.5 py-1.5 text-[11px] font-mono text-foreground placeholder:text-muted/50"
+          />
+        </div>

-      {method === 'POST' && (
        <div>
-          <div className="text-[10px] text-muted mb-0.5">请求体 (JSON，可选)</div>
+          <div className="text-[10px] text-muted mb-1">Headers (JSON，可选)</div>
          <textarea
-            value={body} onChange={e => setBody(e.target.value)}
-            placeholder='{"page": 1}'
+            value={headerStr} onChange={e => setHeaderStr(e.target.value)}
+            placeholder='{"Authorization": "Bearer xxx"}'
            rows={2}
-            className="w-full rounded-md border border-border bg-elevated px-2.5 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40 resize-none"
+            className="w-full rounded-btn border border-border bg-elevated px-2.5 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40 resize-none"
          />
        </div>
-      )}

-      <div className="grid grid-cols-2 gap-2">
-        <div>
-          <div className="text-[10px] text-muted mb-0.5">响应数据路径</div>
-          <input
-            value={responsePath} onChange={e => setResponsePath(e.target.value)}
-            placeholder="data.list"
-            className="w-full rounded-md border border-border bg-elevated px-2 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40"
-          />
+        {method === 'POST' && (
+          <div>
+            <div className="text-[10px] text-muted mb-1">请求体 (JSON，可选)</div>
+            <textarea
+              value={body} onChange={e => setBody(e.target.value)}
+              placeholder='{"page": 1}'
+              rows={2}
+              className="w-full rounded-btn border border-border bg-elevated px-2.5 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40 resize-none"
+            />
+          </div>
+        )}
+
+        <div className="grid grid-cols-2 gap-2">
+          <div>
+            <div className="text-[10px] text-muted mb-1">响应数据路径</div>
+            <input
+              value={responsePath} onChange={e => setResponsePath(e.target.value)}
+              placeholder="data.list"
+              className="w-full rounded-btn border border-border bg-elevated px-2 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40"
+            />
+          </div>
+          <div>
+            <div className="text-[10px] text-muted mb-1">调度间隔 (分钟)</div>
+            <input
+              type="number" min={1} value={schedule} onChange={e => setSchedule(Number(e.target.value))}
+              className="w-full rounded-btn border border-border bg-elevated px-2 py-1.5 text-[10px] font-mono text-foreground"
+            />
+          </div>
        </div>
+
        <div>
-          <div className="text-[10px] text-muted mb-0.5">调度间隔 (分钟)</div>
-          <input
-            type="number" min={1} value={schedule} onChange={e => setSchedule(Number(e.target.value))}
-            className="w-full rounded-md border border-border bg-elevated px-2 py-1.5 text-[10px] font-mono text-foreground"
+          <div className="text-[10px] text-muted mb-1">字段映射 (外部名 → 内部名，JSON，可选)</div>
+          <textarea
+            value={fieldMapStr} onChange={e => setFieldMapStr(e.target.value)}
+            placeholder='{"code": "symbol", "val": "score"}'
+            rows={2}
+            className="w-full rounded-btn border border-border bg-elevated px-2.5 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40 resize-none"
          />
        </div>
      </div>

-      <div>
-        <div className="text-[10px] text-muted mb-0.5">字段映射 (外部名 → 内部名，JSON，可选)</div>
-        <textarea
-          value={fieldMapStr} onChange={e => setFieldMapStr(e.target.value)}
-          placeholder='{"code": "symbol", "val": "score"}'
-          rows={2}
-          className="w-full rounded-md border border-border bg-elevated px-2.5 py-1.5 text-[10px] font-mono text-foreground placeholder:text-muted/40 resize-none"
-        />
+      {/* ===== 分区 ②: 定时拉取状态 ===== */}
+      <div className="rounded-card border border-border/60 bg-elevated/30 p-2.5 space-y-2">
+        <div className="flex items-center justify-between">
+          <div className="flex items-center gap-1.5 text-[11px] font-medium text-secondary">
+            <Clock className="h-3 w-3 text-muted" />
+            <span>定时拉取</span>
+          </div>
+          {/* 自定义 Toggle 开关 */}
+          <button
+            type="button"
+            role="switch"
+            aria-checked={enabled}
+            disabled={toggling}
+            onClick={() => handleToggle(!enabled)}
+            className={`relative inline-flex h-4 w-7 shrink-0 items-center rounded-full transition-colors duration-200 disabled:opacity-50 ${
+              enabled ? 'bg-accent' : 'bg-border'
+            }`}
+          >
+            <span
+              className={`inline-block h-3 w-3 transform rounded-full bg-white shadow transition-transform duration-200 ${
+                enabled ? 'translate-x-3.5' : 'translate-x-0.5'
+              }`}
+            />
+          </button>
+        </div>
+
+        {/* 状态文案 */}
+        <div className="text-[10px] leading-relaxed">
+          {enabled ? (
+            <div className="space-y-0.5">
+              <div className="flex items-center gap-1 text-accent">
+                <span className="h-1 w-1 rounded-full bg-accent animate-pulse" />
+                <span>已启用 · 每 {schedule} 分钟</span>
+              </div>
+              {pull?.next_run && fmtTime(pull.next_run) && (
+                <div className="flex items-center gap-1 text-muted">
+                  <Calendar className="h-2.5 w-2.5" />
+                  <span>下次：{fmtTime(pull.next_run)}</span>
+                </div>
+              )}
+            </div>
+          ) : (
+            <div className="text-muted">未启用 · 仅手动执行</div>
+          )}
+        </div>
+
+        {/* 上次执行结果 */}
+        {pull?.last_run && (
+          <div className="flex items-start gap-1.5 pt-1.5 border-t border-border/40">
+            {pull.last_status === 'success' ? (
+              <CheckCircle2 className="h-3 w-3 text-emerald-500 shrink-0 mt-px" />
+            ) : (
+              <AlertCircle className="h-3 w-3 text-danger shrink-0 mt-px" />
+            )}
+            <div className="min-w-0 flex-1">
+              <div className={`text-[10px] font-medium ${pull.last_status === 'success' ? 'text-emerald-500' : 'text-danger'}`}>
+                {pull.last_message || (pull.last_status === 'success' ? '成功' : '失败')}
+              </div>
+              {fmtTime(pull.last_run) && (
+                <div className="text-[9px] text-muted">{fmtTime(pull.last_run)}</div>
+              )}
+            </div>
+          </div>
+        )}
      </div>

-      <div className="flex items-center justify-between">
-        <label className="inline-flex items-center gap-1.5 cursor-pointer">
-          <input
-            type="checkbox" checked={enabled} onChange={e => setEnabled(e.target.checked)}
-            className="rounded border-border accent-accent"
-          />
-          <span className="text-[10px] text-secondary">启用定时拉取</span>
-        </label>
-        <div className="flex items-center gap-1.5">
+      {/* ===== 分区 ③: 操作按钮 ===== */}
+      <div className="space-y-2">
+        <div className="grid grid-cols-2 gap-2">
          <button
            onClick={handleTest}
            disabled={testing || !url}
-            className="inline-flex items-center gap-1 px-2 py-1 rounded-btn border border-border bg-elevated text-[10px] text-foreground hover:bg-border/30 disabled:opacity-40 transition-colors"
+            className="inline-flex items-center justify-center gap-1 px-2 py-2 rounded-btn border border-border bg-elevated text-xs text-foreground hover:bg-border/30 disabled:opacity-40 transition-colors"
          >
-            {testing ? <Loader2 className="h-3 w-3 animate-spin" /> : <Search className="h-3 w-3" />}
+            {testing ? <Loader2 className="h-3.5 w-3.5 animate-spin" /> : <Search className="h-3.5 w-3.5" />}
            测试
          </button>
          <button
            onClick={handleRun}
            disabled={running || !url}
-            className="inline-flex items-center gap-1 px-2 py-1 rounded-btn bg-accent/90 text-base text-[10px] font-medium hover:bg-accent disabled:opacity-40 transition-colors"
+            className="inline-flex items-center justify-center gap-1 px-2 py-2 rounded-btn bg-accent/90 text-base text-xs font-medium hover:bg-accent disabled:opacity-40 transition-colors"
          >
-            {running ? <Loader2 className="h-3 w-3 animate-spin" /> : <RefreshCw className="h-3 w-3" />}
+            {running ? <Loader2 className="h-3.5 w-3.5 animate-spin" /> : <Zap className="h-3.5 w-3.5" />}
            立即执行
          </button>
        </div>
+        <button
+          onClick={() => handleSave(false)}
+          disabled={saving || !url}
+          className="w-full inline-flex items-center justify-center gap-1 py-2 rounded-btn bg-accent/90 text-base text-xs font-medium hover:bg-accent disabled:opacity-40 transition-colors"
+        >
+          {saving ? <Loader2 className="h-3.5 w-3.5 animate-spin" /> : <Check className="h-3.5 w-3.5" />}
+          保存配置
+        </button>
      </div>

+      {/* ===== 结果展示 ===== */}
+      {runResult && (
+        <div className="rounded-card border border-emerald-500/30 bg-emerald-500/[0.06] p-2.5 flex items-center justify-between text-[10px]">
+          <span className="text-emerald-500 font-medium flex items-center gap-1">
+            <CheckCircle2 className="h-3 w-3" />拉取成功
+          </span>
+          <span className="text-secondary">{runResult.rows} 行 · {runResult.date}</span>
+        </div>
+      )}
+
      {testResult && (
-        <div className="rounded-md border border-accent/30 bg-accent/[0.04] p-2.5 space-y-1.5">
+        <div className="rounded-card border border-accent/30 bg-accent/[0.04] p-2.5 space-y-1.5">
          <div className="flex items-center justify-between text-[10px]">
            <span className="text-accent font-medium">测试成功</span>
            <span className="text-secondary">{testResult.total_rows} 行</span>
@@ -188,25 +316,11 @@ export function ExtDataPullPanel({ config, onSaved }: {
        </div>
      )}

-      {pull?.last_run && (
-        <div className="flex items-center justify-between text-[10px] border-t border-border/50 pt-2">
-          <span className="text-muted">上次执行</span>
-          <span className={pull.last_status === 'success' ? 'text-green-500' : 'text-danger'}>
-            {pull.last_message}
-          </span>
+      {error && (
+        <div className="text-[10px] text-danger text-center bg-danger/[0.06] rounded-btn py-1.5">
+          {error}
        </div>
      )}
-
-      <button
-        onClick={handleSave}
-        disabled={saving || !url}
-        className="w-full inline-flex items-center justify-center gap-1 py-1.5 rounded-btn bg-accent/90 text-base text-xs font-medium hover:bg-accent disabled:opacity-40 transition-colors"
-      >
-        {saving ? <Loader2 className="h-3 w-3 animate-spin" /> : <Check className="h-3 w-3" />}
-        保存配置
-      </button>
-
-      {error && <div className="text-[10px] text-danger text-center">{error}</div>}
    </div>
  )
 }
@@ -1800,6 +1800,7 @@ export interface PullConfig {
  last_status?: string | null
  last_message?: string | null
  last_rows?: number | null
+  next_run?: string | null
 }

 export interface ExtDataConfig {
@@ -1153,7 +1153,11 @@ export function LimitUpLadder() {
  const extColumnsParam = useMemo(() => buildExtColumnsParam(extFields), [extFields])

  const { data, isLoading, refetch, isFetching } = useQuery({
-    queryKey: [QK.limitLadder(asOf || undefined), extColumnsParam, direction],
+    // key 必须扁平: ['limit-ladder', asOf, extColumns, direction]。
+    // QK.limitLadder() 返回数组, 不能整块塞进 key —— 那样第 0 元素会变成嵌套数组
+    // ['limit-ladder', asOf], 与四处 invalidate 用的字符串前缀 ['limit-ladder'] 类型
+    // 不匹配(partialMatchKey 因 typeof 不同而失配), 导致修正/SSE 推送后页面不刷新。
+    queryKey: [...QK.limitLadder(asOf || undefined), extColumnsParam, direction],
    queryFn: () => api.limitLadder(asOf || undefined, extColumnsParam, direction),
    staleTime: 5 * 60_000,
  })