feat(realtime): 自选实时分流与数据落盘可靠性

- quote_service 实时通道重构与自选分流, 旧 watchlist 批量测试相应移除
- repository.replace_with_retry 原子写重试; pipeline_jobs 任务记录落盘
- provider 适配增强; 分时图组件与预览入口调整

backend/app/services/pipeline_jobs.py

@@ -2,8 +2,12 @@
 
 设计:
   - job_store/ 文件夹,每个 job 一个 {id}.json,最多保留 max_jobs 个文件
-  - running/pending 状态的 job 仅存内存(高频读写)
-  - succeeded/failed 后写入独立文件并从内存释放
+  - create()/start() 即落盘 pending/running 快照(进度只更新内存) ——
+    进程意外死亡(uvicorn --reload 热重载 / 被 kill)时记录不蒸发
+  - succeeded/failed 后写入终态并从内存释放
+  - 实例化时扫描磁盘,把上个进程遗留的 pending/running 孤儿记录补标为
+    failed(中断);finished_at 取文件 mtime(最后已知存活时刻),不用下次
+    开机时间虚增时长
   - 列表查询 = 内存中的活跃 job + 磁盘文件扫描,按时间排序
   - 单个查询 = 内存优先,没有则读磁盘
   - 创建新 job 前检查文件数量,>= max_jobs 时删除最老的文件
@@ -15,7 +19,7 @@ import logging
 import os
 import threading
 import uuid
-from datetime import datetime
+from datetime import UTC, datetime
 from pathlib import Path
 from typing import Any, Literal
 
@@ -105,11 +109,12 @@ class JobStore:
         self._active_id: str | None = None
         self._lock = threading.Lock()
         self._store_dir.mkdir(parents=True, exist_ok=True)
+        self._reap_orphans()
 
     # ===== persistence =====
 
     def _write_file(self, job: dict[str, Any]) -> None:
-        """将终态 job 写入独立 JSON 文件。"""
+        """将 job 快照写入独立 JSON 文件(create/start/终态均落盘)。"""
         path = self._store_dir / f"{job['id']}.json"
         try:
             path.write_text(
@@ -154,6 +159,42 @@ class JobStore:
         jobs.sort(key=lambda j: j.get("started_at") or "", reverse=True)
         return jobs
 
+    def _reap_orphans(self) -> None:
+        """启动补录: 把上个进程遗留的 pending/running 记录标为中断。
+
+        单例在进程启动时实例化,此时磁盘上的 pending/running 必然来自已死
+        亡的进程(uvicorn --reload 热重载 / 被 kill)—— 不补录则这些记录
+        永远停留在「运行中」, 同步历史里既看不到结果也看不到失败, 即
+        「数据在、记录丢」。finished_at 取文件 mtime(最后一次落盘 = 最后
+        已知存活时刻), 避免拿下次开机时间虚增 duration。
+        """
+        for f in self._store_dir.glob("*.json"):
+            try:
+                j = json.loads(f.read_text("utf-8"))
+            except Exception:
+                continue
+            orig_status = j.get("status")
+            if not j.get("id") or orig_status not in ("pending", "running"):
+                continue
+            j["status"] = "failed"
+            j["error"] = "后端重启,任务中断(启动时补录)"
+            try:
+                mtime = datetime.fromtimestamp(f.stat().st_mtime, tz=UTC)
+                end = mtime.strftime("%Y-%m-%dT%H:%M:%SZ")
+            except Exception:
+                end = datetime.utcnow().isoformat(timespec="seconds") + "Z"
+            # mtime 早于 started_at(时钟回拨等)时夹住, 避免 duration 为负
+            if j.get("started_at") and end < j["started_at"]:
+                end = j["started_at"]
+            j["finished_at"] = end
+            j["duration_s"] = _duration_s(j)
+            logger.warning(
+                "job_store: 补录中断任务 %s (上个进程遗留 %s 记录)",
+                j["id"],
+                orig_status,
+            )
+            self._write_file(j)
+
     # ===== lifecycle =====
 
     def create(
@@ -189,7 +230,7 @@ class JobStore:
                     return self._active_id, False
 
             job_id = uuid.uuid4().hex[:10]
-            self._active_jobs[job_id] = {
+            job = {
                 "id": job_id,
                 "status": "pending",
                 "stage": "init",
@@ -204,7 +245,11 @@ class JobStore:
                 "error": None,
                 "timeout_s": timeout_s,
             }
+            self._active_jobs[job_id] = job
             self._active_id = job_id
+            # pending 即落盘: 进程在 start() 前死亡时记录也不丢
+            self._delete_oldest()
+            self._write_file(job)
         _register_cancel_flag(job_id)
         return job_id, True
 
@@ -218,6 +263,8 @@ class JobStore:
             # 心跳基准初始化为启动时刻: start() 到首次 progress() 之间的
             # 初始化阶段(解析标的池等)同样计入停滞计时。
             j["last_progress_at"] = j["started_at"]
+            # running 快照落盘: 进程死亡后由下次启动的 _reap_orphans 补录
+            self._write_file(j)
 
     def succeed(self, job_id: str, result: Any) -> None:
         with self._lock:

backend/app/services/quote_service.py

@@ -425,15 +425,18 @@ class QuoteService:
 
     @classmethod
     def realtime_mode(cls) -> str:
-        """当前实时行情模式: none / watchlist / full_market。"""
+        """当前实时行情模式: none / full_market。
+
+        TickFlow 免费档不再提供"自选前 5 只"降级实时(自定义源 fuyao 的全市场
+        快照已全面覆盖且免费); TickFlow 免费档 = 无实时, 接入自定义实时源
+        (如 fuyao)或升级 TickFlow 后恢复全市场模式。
+        """
         from app.services import preferences
         if preferences.get_realtime_data_provider() != "tickflow":
             return "full_market"
         tier = cls._current_tier()
-        if tier == "none":
+        if tier in ("none", "free"):
             return "none"
-        if tier == "free":
-            return "watchlist"
         return "full_market"
 
     @classmethod
@@ -501,7 +504,6 @@ class QuoteService:
 
     def status(self) -> dict:
         """返回行情服务状态。"""
-        from app.services import preferences
         age = (time.perf_counter() - self._fetch_time) * 1000 if self._fetch_time else -1
         mode = self.realtime_mode()
         phase = self._market_phase()
@@ -514,7 +516,6 @@ class QuoteService:
             "paused": self._paused,
             "mode": mode,
             "realtime_allowed": mode != "none",
-            "watchlist_symbol_count": len(preferences.get_realtime_watchlist_symbols()),
             "interval_s": self._interval,
             "symbol_count": self._symbol_count,
             "index_symbol_count": self._index_symbol_count,
@@ -568,15 +569,12 @@ class QuoteService:
                 waited += 0.5
 
     def _fetch_quotes(self, *, final: bool = False) -> bool:
-        """按当前档位拉取行情。加锁串行化 (后台轮询 vs 手动 refresh)。返回本轮是否成功更新。"""
+        """拉取行情。加锁串行化 (后台轮询 vs 手动 refresh)。返回本轮是否成功更新。"""
         with self._fetch_lock:
             before = self._fetched_at
             if final:
                 logger.info("最终行情同步开始")
-            if self.realtime_mode() == "watchlist":
-                self._fetch_watchlist_quotes()
-            else:
-                self._fetch_full_market_quotes()
+            self._fetch_full_market_quotes()
             return self._fetched_at > before
 
     def _fetch_full_market_quotes(self) -> None:
@@ -772,150 +770,11 @@ class QuoteService:
         # ---- 策略监控 + 告警评估 ----
         self._evaluate_monitors(daily_df, quote_extra)
 
-    def _fetch_watchlist_quotes(self) -> None:
-        """Free 档自选股实时: 按 capability batch 上限分批拉取。"""
-        from app.services import preferences
-        from app.tickflow.client import get_paid_realtime_client
-        from app.tickflow.capabilities import Cap
-        from app.tickflow.policy import detect_capabilities
-        from app.tickflow.rate_limits import chunked, resolve_limit, sleep_between_batches
-
-        symbols = preferences.get_realtime_watchlist_symbols()
-        # 指数监控规则标的并入轮询 (与股票共享 batch 额度)
-        engine = getattr(self._app_state, "monitor_engine", None) if self._app_state else None
-        if engine:
-            for _r in list(engine.rules.values()):
-                if _r.get("enabled", True) and _r.get("asset_type") == "index" and _r.get("scope") == "symbols":
-                    for _s in _r.get("symbols", []):
-                        if _s and _s not in symbols:
-                            symbols.append(_s)
-        if not symbols:
-            logger.info("自选实时未配置标的, 跳过行情拉取")
-            return
-
-        tf = get_paid_realtime_client()
-        if tf is None:
-            logger.warning("自选实时拉取失败:未配置付费服务器 API Key")
-            return
-
-        # 按 capability batch 上限分批: 股票+指数共享额度, 超过上限会导致整轮失败
-        capset = detect_capabilities()
-        lim = resolve_limit(capset, Cap.QUOTE_BY_SYMBOL, default_batch=5)
-        batches = chunked(symbols, lim.batch)
-
-        t0 = time.perf_counter()
-        now_ts = time.perf_counter()
-        resp = []
-        for i, batch in enumerate(batches):
-            sleep_between_batches(i, lim.rpm)
-            try:
-                resp.extend(tf.quotes.get(symbols=batch) or [])
-            except Exception as e:  # noqa: BLE001
-                logger.warning("自选实时批次 %d/%d 拉取失败: %s", i + 1, len(batches), e)
-
-        if not resp:
-            logger.warning("自选实时行情数据为空")
-            return
-
-        records = []
-        for q in resp:
-            ext = q.get("ext") or {}
-            last_price = q.get("last_price")
-            prev_close = q.get("prev_close")
-            change_amount = ext.get("change_amount")
-            change_pct = ext.get("change_pct")
-            if change_amount is None and last_price is not None and prev_close is not None:
-                change_amount = float(last_price) - float(prev_close)
-            if change_pct is None and change_amount is not None and prev_close not in (None, 0):
-                # 小数制, 与 ext.change_pct / enriched 口径一致 (不乘 100)
-                change_pct = float(change_amount) / float(prev_close)
-            records.append({
-                "symbol": q.get("symbol"),
-                "name": q.get("name") or ext.get("name"),
-                "last_price": last_price,
-                "prev_close": prev_close,
-                "open": q.get("open"),
-                "high": q.get("high"),
-                "low": q.get("low"),
-                "volume": q.get("volume"),
-                "amount": q.get("amount"),
-                "change_pct": change_pct,
-                "change_amount": change_amount,
-                "amplitude": ext.get("amplitude"),
-                "turnover_rate": ext.get("turnover_rate"),
-                "timestamp": q.get("timestamp"),
-                "session": q.get("session"),
-            })
-
-        index_set = self._repo.get_index_symbol_set() if self._repo else set()
-        etf_set = self._repo.get_etf_symbol_set() if self._repo else set()
-        index_records, etf_records, stock_records = self._split_records_by_asset(records, index_set, etf_set)
-
-        fetch_ms = (time.perf_counter() - t0) * 1000
-        fetched_at = time.time() * 1000
-        with self._lock:
-            self._fetch_time = now_ts
-            self._fetch_ms = fetch_ms
-            self._fetched_at = fetched_at
-            self._symbol_count = len(stock_records)
-            self._index_symbol_count = len(index_records)
-            self._etf_symbol_count = len(etf_records)
-            self._index_quotes_cache = self._build_index_quotes(index_records) if index_records else None
-
-        _persist_last_fetch(fetched_at)
-        logger.info("自选实时刷新: %d 只股票, %d 只ETF, %d 只指数, 耗时 %.0fms",
-                    len(stock_records), len(etf_records), len(index_records), fetch_ms)
-
-        daily_df = self._build_daily(stock_records)
-        quote_extra = self._build_quote_extra(stock_records)
-        if not daily_df.is_empty() and self._repo:
-            try:
-                self._repo.merge_live_daily_asset("stock", daily_df)
-            except Exception as e:  # noqa: BLE001
-                logger.warning("自选实时日K写盘失败: %s", e)
-            self._flush_live_enriched(daily_df, quote_extra, asset_type="stock", merge=True)
-
-        # ETF/指数进自选前5时按各自资产落盘, 不污染股票表
-        etf_daily_df = self._build_daily(etf_records)
-        if not etf_daily_df.is_empty() and self._repo:
-            try:
-                self._repo.merge_live_daily_asset("etf", etf_daily_df)
-            except Exception as e:  # noqa: BLE001
-                logger.warning("自选实时 ETF 日K写盘失败: %s", e)
-            self._flush_live_enriched(etf_daily_df, self._build_quote_extra(etf_records), asset_type="etf", merge=True)
-        index_daily_df = self._build_daily(index_records)
-        if not index_daily_df.is_empty() and self._repo:
-            try:
-                self._repo.merge_live_daily_asset("index", index_daily_df)
-            except Exception as e:  # noqa: BLE001
-                logger.warning("自选实时指数日K写盘失败: %s", e)
-            self._flush_live_enriched(index_daily_df, self._build_quote_extra(index_records), asset_type="index", merge=True)
-
-        self._broadcast_quote_updated()
-        self._evaluate_monitors(daily_df, quote_extra)
-
     # ================================================================
     # 工具
     # ================================================================
 
     @staticmethod
-    def _split_records_by_asset(
-        records: list[dict], index_set: set[str], etf_set: set[str],
-    ) -> tuple[list[dict], list[dict], list[dict]]:
-        """把行情 records 按资产拆成 (index, etf, stock)。判定顺序与 resolve_asset_type 一致: 先 ETF 后指数。"""
-        index_records: list[dict] = []
-        etf_records: list[dict] = []
-        stock_records: list[dict] = []
-        for r in records:
-            sym = r.get("symbol")
-            if sym in etf_set:
-                etf_records.append(r)
-            elif sym in index_set:
-                index_records.append(r)
-            else:
-                stock_records.append(r)
-        return index_records, etf_records, stock_records
-
     @staticmethod
     def _build_daily(records: list[dict]) -> pl.DataFrame:
         """将 API records 转为日K格式 DataFrame (OHLCV + quote_ts, 写 kline_daily 用)。"""

backend/app/tickflow/repository.py

@@ -38,6 +38,33 @@ from app.parquet import scan_enriched_parquet
 logger = logging.getLogger(__name__)
 
 
+def replace_with_retry(src: Path, dst: Path, *, attempts: int = 10, delay_s: float = 0.5) -> None:
+    """os.replace 的 Windows 读锁重试版。
+
+    分区 parquet 的读端 (polars scan_parquet / DuckDB read_parquet 视图) 在扫描进行
+    期间持有句柄; Windows 不允许替换"仍被读端打开"的目标文件 (PermissionError,
+    WinError 5), Linux 的 inode 交换语义则无此限制。读端扫描通常亚秒级完成,
+    短退避重试即可穿过并发读窗口; attempts 次仍被占用则原样抛出, 由上层记录失败。
+    """
+    last: PermissionError | None = None
+    for i in range(attempts):
+        try:
+            src.replace(dst)
+            if i:
+                logger.info("parquet replace succeeded after %d blocked attempt(s): %s", i, dst)
+            return
+        except PermissionError as e:
+            last = e
+            if i == 0:
+                logger.warning(
+                    "parquet replace blocked by concurrent reader, retrying (total <= %.1fs): %s",
+                    attempts * delay_s, dst,
+                )
+            if i < attempts - 1:
+                time.sleep(delay_s)
+    raise last  # type: ignore[misc]  # attempts >= 1 时 last 必已赋值
+
+
 def enriched_dirname(asset_type: str) -> str:
     """asset_type → enriched parquet 目录名。ETF 走独立目录, 其余(stock)用日K enriched。"""
     return "kline_etf_enriched" if asset_type == "etf" else "kline_daily_enriched"
@@ -1813,13 +1840,15 @@ class KlineRepository:
     # ================================================================
 
     def latest_minute_date(self, symbol: str, asset_type: str = "stock") -> date | None:
+        # 注意: 必须走 execute_one (cursor+close)。直连 self.db.execute(...).fetchone()
+        # 的未消费结果集会把首个分区 parquet 的句柄钉在共享连接上, Windows 下阻塞
+        # 同步写入的 os.replace → 个股分时"补齐数据"500。
         table = "kline_etf_minute" if asset_type == "etf" else "kline_minute"
         try:
-            with self._lock:
-                row = self.db.execute(
-                    f"SELECT max(CAST(datetime AS DATE)) FROM {table} WHERE symbol = ?",
-                    [symbol],
-                ).fetchone()
+            row = self.execute_one(
+                f"SELECT max(CAST(datetime AS DATE)) FROM {table} WHERE symbol = ?",
+                [symbol],
+            )
             if row and row[0]:
                 return row[0] if isinstance(row[0], date) else date.fromisoformat(str(row[0]))
         except duckdb.CatalogException:
@@ -1829,10 +1858,9 @@ class KlineRepository:
     def latest_minute_date_global(self) -> date | None:
         """全市场最近分钟K日期 (不分 symbol)。用于非交易日回退到上一交易日。"""
         try:
-            with self._lock:
-                row = self.db.execute(
-                    "SELECT max(CAST(datetime AS DATE)) FROM kline_minute",
-                ).fetchone()
+            row = self.execute_one(
+                "SELECT max(CAST(datetime AS DATE)) FROM kline_minute",
+            )
             if row and row[0]:
                 return row[0] if isinstance(row[0], date) else date.fromisoformat(str(row[0]))
         except Exception:  # noqa: BLE001
@@ -1841,10 +1869,9 @@ class KlineRepository:
     def earliest_daily_date(self) -> date | None:
         """本地日K数据的最早日期。"""
         try:
-            with self._lock:
-                res = self.db.execute(
-                    "SELECT min(date) FROM kline_daily",
-                ).fetchone()
+            res = self.execute_one(
+                "SELECT min(date) FROM kline_daily",
+            )
             if res and res[0]:
                 d = res[0]
                 return d if isinstance(d, date) else date.fromisoformat(str(d))
@@ -1855,10 +1882,9 @@ class KlineRepository:
     def earliest_minute_date(self) -> date | None:
         """本地分钟K数据的最早日期。"""
         try:
-            with self._lock:
-                res = self.db.execute(
-                    "SELECT min(CAST(datetime AS DATE)) FROM kline_minute",
-                ).fetchone()
+            res = self.execute_one(
+                "SELECT min(CAST(datetime AS DATE)) FROM kline_minute",
+            )
             if res and res[0]:
                 d = res[0]
                 return d if isinstance(d, date) else date.fromisoformat(str(d))
@@ -1892,10 +1918,9 @@ class KlineRepository:
     def latest_daily_date(self) -> date | None:
         """本地日K数据的最新日期。"""
         try:
-            with self._lock:
-                res = self.db.execute(
-                    "SELECT max(date) FROM kline_daily",
-                ).fetchone()
+            res = self.execute_one(
+                "SELECT max(date) FROM kline_daily",
+            )
             if res and res[0]:
                 d = res[0]
                 return d if isinstance(d, date) else date.fromisoformat(str(d))
@@ -1947,10 +1972,9 @@ class KlineRepository:
 
     def _latest_enriched_date_duckdb(self) -> date | None:
         try:
-            with self._lock:
-                res = self.db.execute(
-                    "SELECT max(date) FROM kline_enriched",
-                ).fetchone()
+            res = self.execute_one(
+                "SELECT max(date) FROM kline_enriched",
+            )
             if res and res[0]:
                 d = res[0]
                 return d if isinstance(d, date) else date.fromisoformat(str(d))
@@ -2118,10 +2142,11 @@ class KlineRepository:
         直接 write_parquet(out) 在进程被 kill (dev.sh 清端口用 kill -9)
         或断电时会留下半截文件, 之后 scan_parquet glob 整条链路报错。
         临时文件后缀 .tmp 不匹配 *.parquet glob, 不会被扫描误读。
+        Windows 下目标正被并发读取时由 replace_with_retry 短退避穿过。
         """
         tmp = out.with_name(out.name + ".tmp")
         df.write_parquet(tmp)
-        tmp.replace(out)  # 同目录 rename, POSIX/NTFS 均为原子操作
+        replace_with_retry(tmp, out)
 
     def _write_daily_partition(self, df: pl.DataFrame, table: str) -> None:
         """按 date 分区写入 parquet，每个日期一个文件，支持 merge-upsert。"""

backend/tests/test_atomic_write_retry.py

@@ -0,0 +1,161 @@
+"""Windows 读锁竞态下 parquet 原子替换的重试测试。
+
+根因: polars scan_parquet / DuckDB read_parquet 扫描进行中持有分区句柄,
+Windows os.replace 替换"仍被打开"的目标文件抛 PermissionError (WinError 5);
+Linux 的 inode 交换语义无此限制。表现为个股分时"补齐数据"500。
+
+修复: replace_with_retry 短退避重试穿过读窗口; 永久占用则原样抛出。
+两处 _atomic_write_parquet (repository / kline_sync) 均接入。
+
+另含 DuckDB 句柄泄漏回归: latest_minute_date 等曾用 self.db.execute(...)
+.fetchone() 直连共享连接, 未消费结果集把首个分区句柄钉死在连接上,
+导致同步 os.replace 永久被拒 (修为 execute_one cursor+close)。
+"""
+from __future__ import annotations
+
+import os
+import sys
+from datetime import date, datetime
+
+import polars as pl
+import pytest
+
+from app.services import kline_sync
+from app.tickflow import repository
+
+try:
+    import psutil
+
+    _PSUTIL = True
+except ImportError:  # pragma: no cover
+    _PSUTIL = False
+
+
+def _minute_frame() -> pl.DataFrame:
+    return pl.DataFrame({
+        "symbol": ["600519.SH"],
+        "datetime": [datetime(2026, 1, 15, 9, 30)],
+        "open": [10.0], "high": [10.5], "low": [9.5], "close": [10.2],
+        "volume": [100.0], "amount": [1020.0],
+    })
+
+
+def _flaky_replace(monkeypatch, fail_times: int) -> dict:
+    """os.replace 前 fail_times 次 raise PermissionError, 之后正常执行。"""
+    real_replace = os.replace
+    state = {"calls": 0}
+
+    def _flaky(src, dst):
+        state["calls"] += 1
+        if state["calls"] <= fail_times:
+            raise PermissionError(5, "拒绝访问。")
+        return real_replace(src, dst)
+
+    monkeypatch.setattr(os, "replace", _flaky)
+    return state
+
+
+# ---------- replace_with_retry 本体 ----------
+
+def test_retry_succeeds_after_transient_blocks(tmp_path, monkeypatch):
+    out = tmp_path / "part.parquet"
+    out.write_bytes(b"old")
+    src = tmp_path / "part.parquet.tmp"
+    src.write_bytes(b"new")
+    state = _flaky_replace(monkeypatch, fail_times=2)
+
+    repository.replace_with_retry(src, out, attempts=5, delay_s=0)
+
+    assert out.read_bytes() == b"new"
+    assert state["calls"] == 3
+    assert not src.exists()
+
+
+def test_retry_exhausted_raises_last_error(tmp_path, monkeypatch):
+    monkeypatch.setattr(os, "replace", lambda s, d: (_ for _ in ()).throw(PermissionError(5, "拒绝访问。")))
+    src = tmp_path / "a.tmp"
+    src.write_bytes(b"x")
+
+    with pytest.raises(PermissionError, match="拒绝访问"):
+        repository.replace_with_retry(src, tmp_path / "a.parquet", attempts=3, delay_s=0)
+    assert src.exists()  # 未被消费, 目标未生成
+
+
+def test_retry_no_block_single_attempt(tmp_path, monkeypatch):
+    out = tmp_path / "part.parquet"
+    out.write_bytes(b"old")
+    src = tmp_path / "part.parquet.tmp"
+    src.write_bytes(b"new")
+    state = _flaky_replace(monkeypatch, fail_times=0)
+
+    repository.replace_with_retry(src, out, attempts=5, delay_s=0)
+
+    assert state["calls"] == 1
+    assert out.read_bytes() == b"new"
+
+
+# ---------- 两处 _atomic_write_parquet 接入 ----------
+
+def test_kline_sync_atomic_write_survives_transient_lock(tmp_path, monkeypatch):
+    state = _flaky_replace(monkeypatch, fail_times=1)
+    out = tmp_path / "date=2026-01-15" / "part.parquet"
+    out.parent.mkdir(parents=True)
+
+    kline_sync._atomic_write_parquet(_minute_frame(), out)
+
+    assert out.exists()
+    assert state["calls"] == 2
+    assert pl.read_parquet(out).height == 1
+
+
+def test_repository_atomic_write_survives_transient_lock(tmp_path, monkeypatch):
+    state = _flaky_replace(monkeypatch, fail_times=1)
+    out = tmp_path / "kline_minute" / "date=2026-01-15" / "part.parquet"
+    out.parent.mkdir(parents=True)
+
+    repository.KlineRepository._atomic_write_parquet(_minute_frame(), out)
+
+    assert out.exists()
+    assert state["calls"] == 2
+
+
+def test_write_minute_partition_survives_reader_race(tmp_path, monkeypatch):
+    """集成: _write_minute_partition 读旧→concat→写新全程有读锁竞态仍完成。"""
+    state = _flaky_replace(monkeypatch, fail_times=2)
+    # 预置旧分区 (读改写路径)
+    old_dir = tmp_path / "date=2026-01-15"
+    old_dir.mkdir(parents=True)
+    _minute_frame().write_parquet(old_dir / "part.parquet")
+
+    written = kline_sync._write_minute_partition(_minute_frame(), tmp_path)
+
+    assert written == 1
+    assert state["calls"] >= 3  # 至少经历了重试
+
+
+# ---------- DuckDB 句柄泄漏回归 (Windows 实测语义) ----------
+
+@pytest.mark.skipif(sys.platform != "win32" or not _PSUTIL, reason="Windows 句柄语义 + psutil")
+def test_minute_date_queries_do_not_pin_partition_handles(tmp_path):
+    """latest_minute_date 等查询后不得残留分区句柄。
+
+    旧实现 self.db.execute(...).fetchone() 的未消费结果集经 DuckDB buffer
+    manager 钉住首个分区句柄, 后续同步 os.replace 永久 PermissionError。
+    """
+    from app.tickflow.repository import DataStore, KlineRepository
+
+    minute_dir = tmp_path / "kline_minute"
+    kline_sync._write_minute_partition(
+        _minute_frame(), minute_dir)  # date=2026-01-15
+    repo = KlineRepository(DataStore(data_dir=tmp_path))
+
+    assert repo.latest_minute_date("600519.SH") == date(2026, 1, 15)
+    assert repo.latest_minute_date_global() == date(2026, 1, 15)
+    assert repo.earliest_minute_date() == date(2026, 1, 15)
+
+    me = psutil.Process()
+    held = [f.path for f in me.open_files() if "kline_minute" in f.path]
+    assert held == []
+
+    # 钉住场景的端到端后果: 查询后重写同一分区必须成功 (旧实现在此 PermissionError)
+    assert kline_sync._write_minute_partition(_minute_frame(), minute_dir) == 1

backend/tests/test_job_record_persistence.py

@@ -0,0 +1,125 @@
+"""回归测试: job 记录跨进程死亡持久化(「数据在、记录丢」补丁)。
+
+背景(用户反馈): 全市场同步 12:11~12:42 成功结束后 0.7s, uvicorn --reload
+检测到代码变更杀死 worker, 恰好落在管道完成与 job_store.succeed() 落盘之间
+—— 数据已写盘但同步历史无任何记录。旧实现 pending/running 仅存内存、终态才
+落盘, 存在整段丢失窗口。
+
+修复后契约:
+  - create()/start() 即落盘 pending/running 快照;
+  - 下次进程启动(= 新 JobStore 实例, 同目录)把遗留的 pending/running
+    孤儿记录补标为 failed(中断), finished_at 取文件 mtime;
+  - 终态记录不受补录影响; 终态写入覆盖 running 快照(同一文件)。
+均为纯逻辑, 不触网。
+"""
+from __future__ import annotations
+
+import json
+
+from app.services.pipeline_jobs import JobStore
+
+
+def _read_disk(d, jid: str) -> dict:
+    return json.loads((d / f"{jid}.json").read_text("utf-8"))
+
+
+# ── 创建/启动即落盘 ──────────────────────────────────────────────────────
+
+def test_create_writes_pending_snapshot_to_disk(tmp_path):
+    d = tmp_path / "jobs"
+    store = JobStore(store_dir=d)
+    jid, _ = store.create(timeout_s=60)
+
+    disk = _read_disk(d, jid)
+    assert disk["status"] == "pending"
+    assert disk["stage"] == "init"
+
+
+def test_start_updates_disk_snapshot_to_running(tmp_path):
+    d = tmp_path / "jobs"
+    store = JobStore(store_dir=d)
+    jid, _ = store.create(timeout_s=60)
+    store.start(jid)
+
+    disk = _read_disk(d, jid)
+    assert disk["status"] == "running"
+    assert disk["started_at"] is not None
+
+
+# ── 进程死亡 → 下次启动补录 ──────────────────────────────────────────────
+
+def test_orphan_running_record_is_reaped_on_next_boot(tmp_path):
+    """核心场景: 进程死在 running(甚至工作已做完但未终态), 记录必须可见。"""
+    d = tmp_path / "jobs"
+    dead = JobStore(store_dir=d)
+    jid, _ = dead.create(timeout_s=60)
+    dead.start(jid)
+    dead.progress(jid, "sync", 50, "halfway")  # 进度只更新内存
+
+    # 新进程 = 同目录新实例(内存为空, 只有磁盘)
+    revived = JobStore(store_dir=d)
+    j = revived.get(jid)
+    assert j is not None
+    assert j["status"] == "failed"
+    assert "中断" in j["error"]
+    assert j["finished_at"] is not None
+    # finished_at 基于文件 mtime(≈ start 时刻), 时长不得虚增为负或巨大
+    assert j["duration_s"] is not None
+    assert 0 <= j["duration_s"] <= 60
+    # 同步历史列表可见
+    assert any(x["id"] == jid for x in revived.list_recent())
+
+
+def test_orphan_pending_record_is_reaped(tmp_path):
+    """进程死在 create() 与 start() 之间: 记录同样可见, 时长为 None。"""
+    d = tmp_path / "jobs"
+    dead = JobStore(store_dir=d)
+    jid, _ = dead.create(timeout_s=60)
+    # 未 start 即死亡
+
+    revived = JobStore(store_dir=d)
+    j = revived.get(jid)
+    assert j["status"] == "failed"
+    assert j["duration_s"] is None
+
+
+def test_reap_does_not_touch_terminal_records(tmp_path):
+    d = tmp_path / "jobs"
+    store = JobStore(store_dir=d)
+    jid, _ = store.create(timeout_s=60)
+    store.start(jid)
+    store.succeed(jid, {"daily_rows": 100})
+
+    revived = JobStore(store_dir=d)
+    j = revived.get(jid)
+    assert j["status"] == "succeeded"
+    assert j["result"] == {"daily_rows": 100}
+
+
+def test_reap_allows_new_job_after_dead_orphan(tmp_path):
+    """补录后旧 job 已 failed: 新进程 create() 不被死孤儿阻塞(单飞只看内存)。"""
+    d = tmp_path / "jobs"
+    dead = JobStore(store_dir=d)
+    old_jid, _ = dead.create(timeout_s=60)
+    dead.start(old_jid)
+
+    revived = JobStore(store_dir=d)
+    new_jid, is_new = revived.create(timeout_s=60)
+    assert is_new is True
+    assert new_jid != old_jid
+
+
+# ── 终态覆盖快照 ─────────────────────────────────────────────────────────
+
+def test_terminal_write_replaces_running_snapshot(tmp_path):
+    d = tmp_path / "jobs"
+    store = JobStore(store_dir=d)
+    jid, _ = store.create(timeout_s=60)
+    store.start(jid)
+    store.fail(jid, "boom")
+
+    files = list(d.glob("*.json"))
+    assert len(files) == 1
+    disk = _read_disk(d, jid)
+    assert disk["status"] == "failed"
+    assert disk["error"] == "boom"

backend/tests/test_realtime_mode.py

@@ -0,0 +1,31 @@
+"""回归测试: 实时行情模式判定 — 免费档不再提供自选实时降级。
+
+watchlist(自选前 5 只)模式已于 2026-08 移除: 自定义实时源(如 fuyao)的
+全市场快照免费且更优, TickFlow 免费档不再保留降级通路。锁定判定结果,
+防止该通路被无意恢复。
+"""
+from app.services.quote_service import QuoteService
+
+
+def test_custom_realtime_source_is_full_market(monkeypatch):
+    """自定义实时源(如 fuyao)无视 TickFlow 档位, 恒为全市场。"""
+    from app.services import preferences
+    monkeypatch.setattr(preferences, "get_realtime_data_provider", lambda: "fuyao")
+    monkeypatch.setattr(QuoteService, "_current_tier", lambda: "free")
+    assert QuoteService.realtime_mode() == "full_market"
+
+
+def test_tickflow_free_has_no_realtime(monkeypatch):
+    """TickFlow 免费档 = 无实时(不再降级为自选模式)。"""
+    from app.services import preferences
+    monkeypatch.setattr(preferences, "get_realtime_data_provider", lambda: "tickflow")
+    monkeypatch.setattr(QuoteService, "_current_tier", lambda: "free")
+    assert QuoteService.realtime_mode() == "none"
+    assert QuoteService.is_realtime_allowed() is False
+
+
+def test_tickflow_paid_is_full_market(monkeypatch):
+    from app.services import preferences
+    monkeypatch.setattr(preferences, "get_realtime_data_provider", lambda: "tickflow")
+    monkeypatch.setattr(QuoteService, "_current_tier", lambda: "pro")
+    assert QuoteService.realtime_mode() == "full_market"

backend/tests/test_watchlist_batch.py

@@ -1,123 +0,0 @@
-"""回归测试: Free 档自选实时 symbols 超过 capability batch 上限时分批请求 (PR #46 问题 4)。"""
-from contextlib import ExitStack
-from unittest.mock import MagicMock, patch
-
-import polars as pl
-
-from app.services.quote_service import QuoteService
-from app.tickflow.capabilities import Cap, CapabilityLimits, CapabilitySet
-
-
-def _make_svc(engine_rules: dict) -> QuoteService:
-    """创建最小可用的 QuoteService 实例 (跳过 __init__)。"""
-    svc = QuoteService.__new__(QuoteService)
-    svc._app_state = MagicMock()
-    svc._repo = MagicMock()
-    svc._lock = MagicMock()
-
-    engine = MagicMock()
-    engine.rules = engine_rules
-    svc._app_state.monitor_engine = engine
-    svc._app_state.repo = svc._repo
-
-    svc._repo.get_index_symbol_set.return_value = {"000001.SH"}
-    svc._repo.get_etf_symbol_set.return_value = set()
-    return svc
-
-
-def _run_fetch(svc, tf, watchlist: list[str], capset: CapabilitySet):
-    """在完整 patch 环境下执行 _fetch_watchlist_quotes。"""
-    with ExitStack() as stack:
-        stack.enter_context(patch(
-            "app.services.preferences.get_realtime_watchlist_symbols",
-            return_value=watchlist,
-        ))
-        stack.enter_context(patch(
-            "app.tickflow.client.get_paid_realtime_client", return_value=tf,
-        ))
-        stack.enter_context(patch(
-            "app.tickflow.policy.detect_capabilities", return_value=capset,
-        ))
-        stack.enter_context(patch("app.tickflow.rate_limits.sleep_between_batches"))
-        # patch 分批之后的下游处理
-        stack.enter_context(patch.object(
-            QuoteService, "_build_daily", return_value=pl.DataFrame(),
-        ))
-        stack.enter_context(patch.object(
-            QuoteService, "_build_quote_extra", return_value=pl.DataFrame(),
-        ))
-        stack.enter_context(patch.object(
-            QuoteService, "_build_index_quotes", return_value=pl.DataFrame(),
-        ))
-        stack.enter_context(patch.object(QuoteService, "_broadcast_quote_updated"))
-        stack.enter_context(patch.object(QuoteService, "_evaluate_monitors"))
-        stack.enter_context(patch("app.services.quote_service._persist_last_fetch"))
-        svc._fetch_watchlist_quotes()
-
-
-def test_watchlist_batch_respects_capability_limit():
-    """6 symbols / batch 5 → 分 2 批请求, 不整轮失败。"""
-    engine_rules = {
-        "r_idx": {"enabled": True, "asset_type": "index", "scope": "symbols",
-                  "symbols": ["000001.SH"]},
-    }
-    svc = _make_svc(engine_rules)
-
-    tf = MagicMock()
-    tf.quotes.get.return_value = [
-        {"symbol": "600000.SH", "last_price": 10.0, "prev_close": 9.9, "ext": {}},
-    ]
-    capset = CapabilitySet({Cap.QUOTE_BY_SYMBOL: CapabilityLimits(batch=5, rpm=60)})
-
-    _run_fetch(svc, tf,
-               ["600000.SH", "600001.SH", "600002.SH", "600003.SH", "600004.SH"],
-               capset)
-
-    # 5 股票 + 1 指数 = 6 symbols, batch 5 → 2 批
-    assert tf.quotes.get.call_count == 2
-    first_batch = tf.quotes.get.call_args_list[0][1]["symbols"]
-    second_batch = tf.quotes.get.call_args_list[1][1]["symbols"]
-    assert len(first_batch) == 5
-    assert len(second_batch) == 1
-    assert "000001.SH" in second_batch
-
-
-def test_watchlist_batch_partial_failure_keeps_other_batches():
-    """某一批拉取失败不影响其他批次 (已有股票实时刷新不丢失)。"""
-    engine_rules = {
-        "r_idx": {"enabled": True, "asset_type": "index", "scope": "symbols",
-                  "symbols": ["000001.SH"]},
-    }
-    svc = _make_svc(engine_rules)
-
-    tf = MagicMock()
-    # 第一批 (股票) 成功, 第二批 (指数) 失败
-    tf.quotes.get.side_effect = [
-        [{"symbol": "600000.SH", "last_price": 10.0, "prev_close": 9.9, "ext": {}}],
-        ConnectionError("timeout"),
-    ]
-    capset = CapabilitySet({Cap.QUOTE_BY_SYMBOL: CapabilityLimits(batch=5, rpm=60)})
-
-    _run_fetch(svc, tf,
-               ["600000.SH", "600001.SH", "600002.SH", "600003.SH", "600004.SH"],
-               capset)
-
-    # 两批都被尝试 (第二批失败不阻断)
-    assert tf.quotes.get.call_count == 2
-
-
-def test_watchlist_no_index_rules_no_extra_symbols():
-    """无指数监控规则时, symbols 不追加指数标的。"""
-    svc = _make_svc({})  # 无规则
-
-    tf = MagicMock()
-    tf.quotes.get.return_value = [
-        {"symbol": "600000.SH", "last_price": 10.0, "prev_close": 9.9, "ext": {}},
-    ]
-    capset = CapabilitySet({Cap.QUOTE_BY_SYMBOL: CapabilityLimits(batch=5, rpm=60)})
-
-    _run_fetch(svc, tf, ["600000.SH", "600001.SH"], capset)
-
-    # 2 symbols / batch 5 → 1 批
-    assert tf.quotes.get.call_count == 1
-    assert tf.quotes.get.call_args_list[0][1]["symbols"] == ["600000.SH", "600001.SH"]

backend/tests/test_watchlist_realtime_split.py

@@ -1,19 +0,0 @@
-"""Free 档自选实时资产分流测试。"""
-from app.services.quote_service import QuoteService
-
-
-def test_split_records_by_asset():
-    records = [
-        {"symbol": "600000.SH"}, {"symbol": "510300.SH"}, {"symbol": "000001.SH"},
-    ]
-    index, etf, stock = QuoteService._split_records_by_asset(
-        records, {"000001.SH"}, {"510300.SH"},
-    )
-    assert [r["symbol"] for r in index] == ["000001.SH"]
-    assert [r["symbol"] for r in etf] == ["510300.SH"]
-    assert [r["symbol"] for r in stock] == ["600000.SH"]
-    # etf 优先于 index (与 resolve_asset_type 判定顺序一致)
-    index2, etf2, stock2 = QuoteService._split_records_by_asset(
-        [{"symbol": "X"}], {"X"}, {"X"},
-    )
-    assert etf2 and not index2 and not stock2

frontend/src/components/EChartsIntraday.tsx

@@ -72,6 +72,9 @@ function buildOption(data: MinuteKlineRow[], prevClose: number | undefined, avgP
   const volumes = new Array(FULL_DAY_TIMES.length).fill(null) as (any | null)[]
 
   const volNeutral = 'rgba(161,161,170,0.5)'
+  // 量柱着色基准: 前一分钟 close; 第一根用昨收。
+  // 不用 row.open —  历史日无真实分钟 open(为 null), close-vs-open 会全偏。
+  let prevRef: number | null = prevClose ?? null
   for (let i = 0; i < data.length; i++) {
     const timeKey = formatMinuteTime(data[i].datetime)
     const idx = timeIndexMap.get(timeKey)
@@ -83,9 +86,16 @@ function buildOption(data: MinuteKlineRow[], prevClose: number | undefined, avgP
       volumes[idx] = {
         value: data[i].volume,
         itemStyle: {
-          color: data[i].close > data[i].open ? THEME.volUp : data[i].close < data[i].open ? THEME.volDown : volNeutral,
+          color: prevRef == null
+            ? volNeutral
+            : data[i].close > prevRef
+              ? THEME.volUp
+              : data[i].close < prevRef
+                ? THEME.volDown
+                : volNeutral,
         },
       }
+      prevRef = data[i].close
     }
   }
 
@@ -570,7 +580,7 @@ export function EChartsIntraday({
             <>
               {date && <span className="text-muted">{date}</span>}
               <span className="text-muted">开</span>
-              <span style={{ color: priceClr }}>{d.open.toFixed(2)}</span>
+              <span style={{ color: priceClr }}>{d.open != null ? d.open.toFixed(2) : '—'}</span>
               <span className="text-muted">高</span>
               <span style={{ color: priceClr }}>{d.high.toFixed(2)}</span>
               <span className="text-muted">低</span>

frontend/src/components/EChartsMultiDayIntraday.tsx

@@ -74,6 +74,9 @@ function buildModel(sessions: MinuteKlineSession[]) {
 
     const dayValues: (number | null)[] = []
     const dayAverages: (number | null)[] = []
+    // 量柱着色基准: 前一分钟 close; 当日第一根用 session 昨收。
+    // 不用 row.open —  历史日无真实分钟 open(为 null), close-vs-open 会全偏。
+    let prevRef: number | null = session.prev_close
     for (const time of FULL_DAY_TIMES) {
       const point = rowsByTime.get(time)
       const index = categories.length
@@ -91,13 +94,16 @@ function buildModel(sessions: MinuteKlineSession[]) {
       volumeData.push({
         value: row.volume,
         itemStyle: {
-          color: row.close > row.open
-            ? COLORS.volumeUp
-            : row.close < row.open
-              ? COLORS.volumeDown
-              : COLORS.volumeFlat,
+          color: prevRef == null
+            ? COLORS.volumeFlat
+            : row.close > prevRef
+              ? COLORS.volumeUp
+              : row.close < prevRef
+                ? COLORS.volumeDown
+                : COLORS.volumeFlat,
         },
       })
+      prevRef = row.close
       priceValues.push(row.low, row.high, average)
       pointByIndex.set(index, {
         date: session.date,
@@ -408,7 +414,7 @@ export function EChartsMultiDayIntraday({
           {info ? (
             <>
               <span className="text-muted">{info.date} {formatMinuteTime(info.row.datetime)}</span>
-              <span className="text-muted">开</span><span style={{ color: infoColor }}>{info.row.open.toFixed(2)}</span>
+              <span className="text-muted">开</span><span style={{ color: infoColor }}>{info.row.open != null ? info.row.open.toFixed(2) : '—'}</span>
               <span className="text-muted">高</span><span style={{ color: infoColor }}>{info.row.high.toFixed(2)}</span>
               <span className="text-muted">低</span><span style={{ color: infoColor }}>{info.row.low.toFixed(2)}</span>
               <span className="text-muted">收</span><span className="font-semibold" style={{ color: infoColor }}>{info.row.close.toFixed(2)}</span>

frontend/src/components/Layout.tsx

@@ -575,10 +575,6 @@ export function Layout() {
         toast('当前数据源无实时行情能力, 请先配置数据源', 'error')
         return
       }
-      if (fresh.mode === 'watchlist' && (prefs?.realtime_watchlist_symbols?.length ?? 0) === 0) {
-        navigate('/watchlist')
-        return
-      }
     }
     await toggleQuote.mutateAsync(enabled)
     // 仅在交易时段立即获取一次行情

frontend/src/components/StockPreviewDialog.tsx

@@ -50,11 +50,11 @@ interface PriceAlertDraft {
 }
 const INTRADAY_DAY_OPTIONS = [1, 5, 10, 20] as const
 
-function loadIntradayDays(): number {
+function loadIntradayDays(): number | null {
   const saved = storage.stockPreviewIntradayDays.get(10)
   return INTRADAY_DAY_OPTIONS.includes(saved as typeof INTRADAY_DAY_OPTIONS[number])
     ? saved
-    : 10
+    : null
 }
 
 function boardTag(symbol: string): { label: string; color: string } | null {
@@ -81,7 +81,7 @@ function fmtAbnormalCalcTime(asofSec: number): string {
 
 export function StockPreviewDialog({ symbol, name, onClose, triggerInfo }: Props) {
   const [view, setView] = useState<PreviewView>('daily')
-  const [intradayDays, setIntradayDays] = useState(loadIntradayDays)
+  const [intradayDays, setIntradayDays] = useState<number | null>(loadIntradayDays)
   const [dateRange, setDateRange] = useState(getDefaultRange)
   const [showMonitorEditor, setShowMonitorEditor] = useState(false)
   const [priceAlertDraft, setPriceAlertDraft] = useState<PriceAlertDraft | null>(null)
@@ -167,6 +167,21 @@ export function StockPreviewDialog({ symbol, name, onClose, triggerInfo }: Props
   const { data: prefs } = usePreferences()
   const intradayRefetchMs = (prefs?.minute_intraday_refresh_interval ?? 6) * 1000
 
+  // 分时档位按分钟源历史深度收窄: 浅源(如 =5日)只显示可行档位、默认 5日;
+  // 深源(tickflow/未声明)全档位、默认 20日。用户已保存的可行选择优先保留。
+  const minuteHistoryDays = prefs?.minute_history_days ?? null
+  const dayOptions = useMemo<number[]>(
+    () => INTRADAY_DAY_OPTIONS.filter(d => minuteHistoryDays == null || d <= minuteHistoryDays),
+    [minuteHistoryDays],
+  )
+  const defaultIntradayDays = minuteHistoryDays != null && minuteHistoryDays < 20 ? 5 : 20
+  const effectiveIntradayDays = intradayDays ?? defaultIntradayDays
+  useEffect(() => {
+    if (!dayOptions.includes(effectiveIntradayDays)) {
+      setIntradayDays(defaultIntradayDays)
+    }
+  }, [dayOptions, effectiveIntradayDays, defaultIntradayDays])
+
   const handleRefresh = () => {
     if (!symbol) return
     if (view === 'daily') {
@@ -270,14 +285,14 @@ export function StockPreviewDialog({ symbol, name, onClose, triggerInfo }: Props
                 ) : (
                   <div className="flex items-center gap-1">
                     <div className="inline-flex shrink-0 items-center rounded border border-border bg-elevated p-0.5" aria-label="分时周期">
-                      {INTRADAY_DAY_OPTIONS.map(days => (
+                      {dayOptions.map(days => (
                         <button
                           key={days}
                           type="button"
-                          aria-pressed={intradayDays === days}
+                          aria-pressed={effectiveIntradayDays === days}
                           onClick={() => selectIntradayDays(days)}
                           className={`h-5 rounded px-1.5 font-mono text-[10px] transition-colors ${
-                            intradayDays === days
+                            effectiveIntradayDays === days
                               ? 'bg-accent/20 text-accent'
                               : 'text-muted hover:text-secondary'
                           }`}
@@ -486,7 +501,7 @@ export function StockPreviewDialog({ symbol, name, onClose, triggerInfo }: Props
                 />
                 <StockMultiDayIntradayChart
                   symbol={symbol}
-                  days={intradayDays}
+                  days={effectiveIntradayDays}
                   height={480}
                   refetchIntervalMs={intradayRefetchMs}
                   priceLines={monitorPriceLines}

frontend/src/components/stock-table/MiniIntraday.tsx

@@ -36,23 +36,23 @@ export function MiniIntraday({ rows, prevClose, changePct, width = 100, height =
   const n = rows.length
 
   // 涨跌着色: 优先用 changePct (后端 enriched 字段, 最可靠);
-  // 其次用 prevClose vs lastClose; 最后回退到第一根 open
+  // 其次用 prevClose vs lastClose; 最后回退到首根 open(可能为 null, 再退首根 close)
   const lastClose = rows[n - 1].close
-  const firstOpen = rows[0].open
+  const firstRef = rows[0].open ?? rows[0].close
   const isUp = changePct != null
     ? changePct >= 0
     : prevClose != null && prevClose > 0
       ? lastClose >= prevClose
-      : lastClose >= firstOpen
+      : lastClose >= firstRef
   const color = isUp ? BULL : BEAR
 
   // 昨收基准线: 优先用 prevClose; 其次用 changePct 反算 (close/(1+changePct));
-  // 最后回退到第一根 open
+  // 最后回退到首根 open(为 null 时退首根 close)
   const baseline = (prevClose != null && prevClose > 0)
     ? prevClose
     : (changePct != null && changePct !== 0)
       ? lastClose / (1 + changePct)
-      : firstOpen
+      : firstRef
 
   // 价格区间: close + 昨收 + 均线 全部纳入, 确保都在可视范围
   let hi = -Infinity, lo = Infinity

frontend/src/lib/intraday-chart.ts

@@ -1,10 +1,10 @@
 import type { MinuteKlineRow } from '@/lib/api'
 
+/** 从 datetime 串取 HH:MM。契约: 分钟K datetime 已在后端入口统一为北京墙钟, 前端不做时区换算。 */
 export function formatMinuteTime(datetime: string): string {
   const match = datetime.match(/(\d{2}):(\d{2})/)
   if (!match) return datetime.slice(11, 16)
-  const hour = (parseInt(match[1]) + 8) % 24
-  return `${String(hour).padStart(2, '0')}:${match[2]}`
+  return `${match[1]}:${match[2]}`
 }
 
 export function computeIntradayAverage(data: MinuteKlineRow[]): number[] {