feat(minute-k): 分钟K同步流式落盘 + 段大小可配 + 回测成交修复 + 卡死超时调整

1. 分钟K流式落盘 (修复1年全市场卡死)
   - sync_minute_batch 加 on_segment 回调,每段拉完立即写盘
   - 抽出 _write_minute_partition 公共函数,消除重复
   - 内存峰值从全量(~25GB)降到单段,避免 OOM

2. 段大小可配 (默认20交易日,范围[5,30])
   - 新增 minute_sync_segment_days 偏好项
   - 「单次获取」与「获取1年」共用此分段设置
   - 前端设置弹窗新增分段大小步进器

3. 回测分钟K精确成交修复 (从未成功跑通过)
   - _resolve_minute_fill 改接受 DataFrame,避免 to_numpy() 退化为
     object 数组后字段名索引抛 IndexError
   - _load_minute_for_fills 用 partition_by 向量化分组替换逐行循环
   - 新增 test_minute_fill.py 回归测试 (6 用例)

4. 卡死 watchdog 超时阈值按任务类型区分
   - 普通任务 600s → 1200s;分钟K长任务 → 1800s
   - create(timeout_s=) 支持按 job 存阈值,reap_stale 读 job 自身值
   - 修复分钟K正常拉取被误杀导致僵尸线程继续写盘的问题

5. UI 优化
   - 设置弹窗按 自动同步/手动获取/清空 三区块分组
   - 「单次获取」改为按分段大小拉一段 (天数=分段设置)
   - 状态文案: 盘后自动同步 → 自动同步已开启/已关闭

backend/app/api/kline.py

@@ -585,7 +585,7 @@ async def sync_minute(request: Request):
     """
     import asyncio
 
-    from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot
+    from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot, LONG_JOB_TIMEOUT_S
     from app.api.data import invalidate_storage_cache
     from app.services.preferences import get_minute_sync_days
     from app.tickflow.capabilities import Cap
@@ -607,7 +607,8 @@ async def sync_minute(request: Request):
     override_days = body.get("days")
     extend_flag = body.get("extend")
 
-    job_id, is_new = job_store.create()
+    # 分钟K全市场同步是长任务(数据量是日K的 ~240 倍),用更宽松的卡死阈值
+    job_id, is_new = job_store.create(timeout_s=LONG_JOB_TIMEOUT_S)
     if not is_new:
         return {"status": "reused", "job_id": job_id}
 

backend/app/api/settings.py

@@ -337,6 +337,8 @@ def _realtime_allowed() -> bool:
 class MinuteSyncPrefs(BaseModel):
     minute_sync_enabled: bool
     minute_sync_days: int = 5
+    # 单段大小(交易日),None 表示不修改现有值。范围 [5, 30],默认 20。
+    minute_sync_segment_days: int | None = None
 
 
 class DataProvidersIn(BaseModel):
@@ -395,6 +397,7 @@ def get_preferences() -> dict:
         "indices_nav_pinned": preferences.get_indices_nav_pinned(),
         "minute_sync_enabled": preferences.get_minute_sync_enabled(),
         "minute_sync_days": preferences.get_minute_sync_days(),
+        "minute_sync_segment_days": preferences.get_minute_sync_segment_days(),
         "daily_data_provider": preferences.get_daily_data_provider(),
         "adj_factor_provider": preferences.get_adj_factor_provider(),
         "minute_data_provider": preferences.get_minute_data_provider(),
@@ -643,16 +646,24 @@ def update_screener_result_columns(req: dict) -> dict:
 
 @router.put("/preferences/minute-sync")
 def update_minute_sync(req: MinuteSyncPrefs) -> dict:
-    """保存分钟 K 同步偏好。"""
+    """保存分钟 K 同步偏好。
+
+    minute_sync_segment_days 为可选:未传(None)时不覆盖现有值,便于开关/天数
+    与段大小各自独立更新。
+    """
     from app.services import preferences
     days = max(1, min(30, req.minute_sync_days))
-    preferences.save({
+    updates: dict = {
         "minute_sync_enabled": req.minute_sync_enabled,
         "minute_sync_days": days,
-    })
+    }
+    if req.minute_sync_segment_days is not None:
+        updates["minute_sync_segment_days"] = max(5, min(30, req.minute_sync_segment_days))
+    preferences.save(updates)
     return {
         "minute_sync_enabled": req.minute_sync_enabled,
         "minute_sync_days": days,
+        "minute_sync_segment_days": preferences.get_minute_sync_segment_days(),
     }
 
 

backend/app/backtest/engine.py

@@ -527,7 +527,7 @@ class BacktestEngine:
                 )
                 for _key, _mdf in _loaded.items():
                     if not _mdf.is_empty():
-                        minute_cache[_key] = _mdf.to_numpy()
+                        minute_cache[_key] = _mdf
 
         def _refill_price(idx: int, side: str, daily_price: float) -> float:
             if not config.minute_fill or not minute_cache:
@@ -839,29 +839,33 @@ class BacktestEngine:
 
     @staticmethod
     def _resolve_minute_fill(
-        minute_rows: np.ndarray,
+        minute_df: pl.DataFrame,
         ref_price: float | None,
         side: str,
     ) -> float | None:
         """用当日分钟K确定精确成交价。
 
         Args:
-            minute_rows: structured numpy array, 字段含 open/high/low/close/volume/amount
+            minute_df: 当日分钟K polars DataFrame, 列含 open/high/low/close/volume/amount
             ref_price: 信号参考线价格 (如 MA5 值); None 表示无参考线
             side: "buy" 或 "sell", 决定穿越方向
 
         Returns:
             精确成交价, 或 None (降级到日K口径)
+
+        注: 原实现用 df.to_numpy() 转 structured array 再按字段名索引, 但当列类型不
+        一致时 (如 datetime 列 + float 列) to_numpy() 退化为 object 二维数组, 字段名
+        索引 arr["open"] 会抛 IndexError。改为直接按列取 Series, 稳定且更快。
         """
-        if minute_rows is None or len(minute_rows) == 0:
+        if minute_df is None or minute_df.is_empty():
             return None
 
-        opens = minute_rows["open"].astype(float)
-        highs = minute_rows["high"].astype(float)
-        lows = minute_rows["low"].astype(float)
-        closes = minute_rows["close"].astype(float)
-        volumes = minute_rows["volume"].astype(float) if "volume" in minute_rows.dtype.names else None
-        amounts = minute_rows["amount"].astype(float) if "amount" in minute_rows.dtype.names else None
+        opens = minute_df["open"].to_numpy().astype(float)
+        highs = minute_df["high"].to_numpy().astype(float)
+        lows = minute_df["low"].to_numpy().astype(float)
+        closes = minute_df["close"].to_numpy().astype(float)
+        volumes = minute_df["volume"].to_numpy().astype(float) if "volume" in minute_df.columns else None
+        amounts = minute_df["amount"].to_numpy().astype(float) if "amount" in minute_df.columns else None
 
         # 有参考线 → 穿越价成交 (逻辑同止损: 找价格穿越参考线的时刻)
         if ref_price is not None and ref_price > 0 and np.isfinite(ref_price):
@@ -914,22 +918,19 @@ class BacktestEngine:
         if df.is_empty():
             return {}
 
+        # 按 (symbol, 日期) 向量化分组, 替代原 iter_rows 逐行 Python 循环。
+        # 原实现对每行做 dict 转换再重建 DataFrame, 回测区间内触发股数多时极慢。
+        df = df.with_columns(
+            pl.col("datetime").dt.strftime("%Y-%m-%d").alias("_d_str")
+        )
         cache: dict = {}
-        for row in df.iter_rows(named=True):
-            dt = row.get("datetime")
-            if dt is None:
+        for sub in df.partition_by(["symbol", "_d_str"], as_dict=False):
+            if sub.is_empty():
                 continue
-            d_str = str(dt)[:10]
-            sym = row["symbol"]
-            key = (sym, d_str)
-            if key not in cache:
-                cache[key] = []
-            cache[key].append(row)
-        # 转 DataFrame per key
-        result: dict = {}
-        for key, rows in cache.items():
-            result[key] = pl.DataFrame(rows)
-        return result
+            sym = sub["symbol"][0]
+            d_str = sub["_d_str"][0]
+            cache[(sym, d_str)] = sub.drop("_d_str")
+        return cache
 
     def simulate_portfolio(
         self,
@@ -1052,7 +1053,7 @@ class BacktestEngine:
                 )
                 for key, mdf in loaded.items():
                     if not mdf.is_empty():
-                        minute_cache[key] = mdf.to_numpy()
+                        minute_cache[key] = mdf
 
         def _refill_price(idx: int, side: str, daily_price: float) -> float:
             """分钟K精确成交价; 无数据则降级为 daily_price。"""

backend/app/services/kline_sync.py

@@ -499,6 +499,34 @@ def _datetime_to_ms(dt: datetime) -> int:
     return int(dt.timestamp() * 1000)
 
 
+def _write_minute_partition(df: pl.DataFrame, minute_dir) -> int:
+    """按 _trade_date 分区落盘分钟 K (读旧→concat→unique→原子写)。返回写入行数。
+
+    抽自原 sync_and_persist_minute 末尾的循环, 供流式落盘 (每段一次) 与一次性迁移共用。
+    """
+    if df.is_empty():
+        return 0
+    df = df.with_columns(pl.col("datetime").dt.date().alias("_trade_date"))
+    written = 0
+    for day_df in df.partition_by("_trade_date"):
+        trade_date = day_df["_trade_date"][0]
+        out = minute_dir / f"date={trade_date}" / "part.parquet"
+        out.parent.mkdir(parents=True, exist_ok=True)
+        if out.exists():
+            existing = pl.read_parquet(out)
+            if "datetime" in existing.columns:
+                existing = existing.filter(pl.col("datetime").is_not_null())
+            day_df = pl.concat([existing, day_df.drop("_trade_date")]).unique(
+                subset=["symbol", "datetime"], keep="last",
+            )
+        else:
+            day_df = day_df.drop("_trade_date")
+        day_df = day_df.sort("symbol", "datetime")
+        _atomic_write_parquet(day_df, out)
+        written += day_df.height
+    return written
+
+
 def sync_minute_batch(
     symbols: list[str],
     start_time: datetime | None = None,
@@ -507,6 +535,8 @@ def sync_minute_batch(
     batch_size: int | None = None,
     rpm: int | None = None,
     on_chunk_done: Callable[[int, int, str], None] | None = None,
+    segment_trading_days: int = 20,
+    on_segment: Callable[[pl.DataFrame], None] | None = None,
 ) -> pl.DataFrame:
     """批量拉取多股分钟 K。
 
@@ -514,8 +544,15 @@ def sync_minute_batch(
     count 仅作为 fallback 保留。
     on_chunk_done(current, total) 每个 chunk 完成后回调。
 
-    TickFlow count 上限 10000 根/股, 1 天 240 根 → 单次最多约 41 天。
-    当区间超过 35 天时自动按月 (30 天) 分段拉取, 拼接结果。
+    segment_trading_days: 单段大小 (交易日), 控制每次 SDK 请求覆盖的天数。
+        TickFlow count 上限 10000 根/股, 1 天 240 根 → 物理上限 ~41 交易日;
+        默认 20 (4800 根, 安全余量足), 范围建议 [5, 30]。
+        段越小: 单次内存峰值越低 (适合小内存机器), 但总批数↑ → 限速 sleep↑ → 更慢。
+        段越大: 速度越快, 内存峰值越高。
+    on_segment: 每个时间段拉完后回调 (传入该段拼接后的 DataFrame)。
+        传入时走「流式落盘」: 段内结果累积到 seg_out, 段末 concat 后回调并清空,
+        不进入全局 out → 内存峰值从「全量」降到「单段」。适用于 sync_and_persist_minute。
+        不传时 (如 get_minute_batch 的实时补拉) 保持原契约: 累积进 out 末尾一次性返回。
     """
     # 自定义数据源分流: minute provider
     provider_name = preferences.get_minute_data_provider()
@@ -531,8 +568,9 @@ def sync_minute_batch(
     tf = get_client()
 
     # TickFlow count 上限 10000 根/股, 1 天 240 根 → 单次最多约 41 个交易日。
-    # 按 41 交易日 (≈57 自然日) 分段: ≤41 交易日的区间只产生 1 段 (单次拉满)。
-    SEG_CHUNK = timedelta(days=57)  # 41 交易日 × 7/5 ≈ 57 自然日
+    # 按 segment_trading_days 交易日分段 (交易日→自然日 ×7/5 换算, 含节假日余量)。
+    seg_calendar_days = max(1, int(segment_trading_days * 7 / 5))
+    SEG_CHUNK = timedelta(days=seg_calendar_days)
     time_segments: list[tuple[datetime, datetime]] = []
     if start_time and end_time:
         seg_start = start_time
@@ -545,7 +583,10 @@ def sync_minute_batch(
 
     total_steps = len(time_segments) * len(chunked(symbols, batch_size))
     step = 0
+    # 全局累积 (仅 on_segment=None 时使用, 末尾一次性 concat 返回)
     out: list[pl.DataFrame] = []
+    # 段内累积: 每段拉完即 flush, 避免全量攒内存 (OOM 根因)
+    seg_out: list[pl.DataFrame] = []
 
     for seg_idx, (seg_start, seg_end) in enumerate(time_segments):
         # 当前的日期段描述 (供进度展示)
@@ -580,13 +621,21 @@ def sync_minute_batch(
                 for sym, sub in raw.items():
                     if sub is None or len(sub) == 0:
                         continue
-                    out.append(_normalize_minute(sub, default_symbol=sym))
+                    seg_out.append(_normalize_minute(sub, default_symbol=sym))
             elif raw is not None and len(raw) > 0:
-                out.append(_normalize_minute(raw))
+                seg_out.append(_normalize_minute(raw))
 
             if on_chunk_done:
                 on_chunk_done(step, total_steps, seg_label)
 
+        # 段末 flush: 流式落盘回调 或 并入全局 out
+        if seg_out:
+            if on_segment:
+                on_segment(pl.concat(seg_out, how="diagonal_relaxed"))
+            else:
+                out.extend(seg_out)
+            seg_out = []
+
     if not out:
         return pl.DataFrame()
     return pl.concat(out, how="diagonal_relaxed")
@@ -777,9 +826,8 @@ def sync_and_persist_minute(
     if extend_backward:
         # 向前扩展模式: 从本地最早数据往前补, 叠加已有数据避免缺口。
         earliest_dt = _earliest_minute_datetime(repo)
-        # 按交易日换算自然日 (7/5 系数)
-        # ≤41 交易日: 不加余量, 确保落在单段 (57 自然日) 内 → 单次拉满
-        # >41 交易日: +10 天余量覆盖节假日
+        # 按交易日换算自然日 (7/5 系数)。>41 交易日时 +10 天余量覆盖节假日。
+        # (分段由 sync_minute_batch 的 segment_trading_days 控制, 与此处的区间天数独立。)
         calendar_days = int(days * 7 / 5) + (10 if days > 41 else 0)
         if earliest_dt:
             end_time = earliest_dt
@@ -805,33 +853,26 @@ def sync_and_persist_minute(
         default_rpm_when_unset=False,
     )
 
-    df = sync_minute_batch(symbols, start_time=start_time, end_time=end_time,
-                           batch_size=limit.batch, rpm=limit.rpm,
-                           on_chunk_done=on_chunk_done)
-    if df.is_empty():
-        return 0
+    # 流式落盘: 每段拉完立即写盘, 内存峰值 = 单段 (而非全量)。
+    # 全量攒内存曾导致 1 年全市场分钟 K OOM 卡死 (3 亿行 / 数十 GB)。
+    minute_dir = repo.store.data_dir / "kline_minute"
+    written_box = [0]  # list 闭包, 绕过 Python 闭包外层赋值
 
-    # 按日期分区写: data/kline_minute/date={YYYY-MM-DD}/part.parquet
-    df = df.with_columns(
-        pl.col("datetime").dt.date().alias("_trade_date")
+    def _persist(seg_df: pl.DataFrame) -> None:
+        written_box[0] += _write_minute_partition(seg_df, minute_dir)
+
+    segment_days = preferences.get_minute_sync_segment_days()
+    sync_minute_batch(
+        symbols, start_time=start_time, end_time=end_time,
+        batch_size=limit.batch, rpm=limit.rpm,
+        on_chunk_done=on_chunk_done,
+        segment_trading_days=segment_days,
+        on_segment=_persist,
     )
-    written = 0
-    for day_df in df.partition_by("_trade_date"):
-        trade_date = day_df["_trade_date"][0]
-        out = repo.store.data_dir / "kline_minute" / f"date={trade_date}" / "part.parquet"
-        out.parent.mkdir(parents=True, exist_ok=True)
-        if out.exists():
-            existing = pl.read_parquet(out)
-            if "datetime" in existing.columns:
-                existing = existing.filter(pl.col("datetime").is_not_null())
-            day_df = pl.concat([existing, day_df.drop("_trade_date")]).unique(
-                subset=["symbol", "datetime"], keep="last",
-            )
-        else:
-            day_df = day_df.drop("_trade_date")
-        day_df = day_df.sort("symbol", "datetime")
-        _atomic_write_parquet(day_df, out)
-        written += day_df.height
+
+    if written_box[0] == 0:
+        return 0
+    written = written_box[0]
 
     # 刷新视图
     try:

backend/app/services/pipeline_jobs.py

@@ -26,7 +26,16 @@ JobStatus = Literal["pending", "running", "succeeded", "failed"]
 # 运行超过此秒数视为卡死(reload 后孤儿 task / 网络读无限阻塞等)。
 # 由 reap_stale() 在 /run 和 /jobs/{id} 轮询端点检查 — 保证卡死后能自愈,
 # 无需用户再次点击「同步」。
-STALE_JOB_TIMEOUT_S = 600
+#
+# 超时阈值按任务类型区分:
+#   - 普通任务(日K管道/扩展/修正/重算): 1200s (20 分钟)
+#   - 长任务(分钟K全市场同步,数据量是日K的 ~240 倍): 1800s (30 分钟)
+# 分钟K即使流式落盘后仍可能跑十几到数十分钟(限速 sleep 是主因),
+# 用 600s 会误杀正常任务并留下写盘僵尸线程。
+DEFAULT_JOB_TIMEOUT_S = 1200
+LONG_JOB_TIMEOUT_S = 1800
+# 向后兼容: 旧调用方引用 STALE_JOB_TIMEOUT_S
+STALE_JOB_TIMEOUT_S = DEFAULT_JOB_TIMEOUT_S
 
 
 def _default_store_dir() -> Path:
@@ -96,7 +105,7 @@ class JobStore:
 
     # ===== lifecycle =====
 
-    def create(self) -> tuple[str, bool]:
+    def create(self, timeout_s: int = DEFAULT_JOB_TIMEOUT_S) -> tuple[str, bool]:
         """单飞创建任务。返回 (job_id, is_new)。
 
         去重条件为 **pending ∨ running**(而非仅 running):`/run` 先 create() 再在
@@ -105,6 +114,9 @@ class JobStore:
         _active_id,导致两条全市场拉取同时读改写同一 parquet。纳入 pending 后该窗口关闭。
 
         is_new=False 表示复用了已有活跃任务,调用方**不得**再调度新的后台任务。
+
+        timeout_s: reap_stale 判定卡死的阈值。普通任务默认 1200s;
+            分钟K全市场同步等长任务传 LONG_JOB_TIMEOUT_S (1800s)。
         """
         with self._lock:
             if self._active_id:
@@ -125,6 +137,7 @@ class JobStore:
                 "duration_s": None,
                 "result": None,
                 "error": None,
+                "timeout_s": timeout_s,
             }
             self._active_id = job_id
             return job_id, True
@@ -224,12 +237,16 @@ class JobStore:
     def active_id(self) -> str | None:
         return self._active_id
 
-    def reap_stale(self, timeout_s: int = STALE_JOB_TIMEOUT_S) -> None:
-        """回收运行超过 timeout_s 的卡死 running job(标记为 failed)。
+    def reap_stale(self, timeout_s: int | None = None) -> None:
+        """回收运行超过阈值(卡死)的 running job(标记为 failed)。
 
         在 /run 和 /jobs/{id} 轮询端点都会调用 — 保证卡死后任意轮询都能自愈,
         无需用户再次手动触发同步。reload 后的孤儿 task(内存里已无 job 记录)
         不在此处理:它们没有 active_id,只能靠 executor 线程自然结束或进程重启。
+
+        timeout_s: 显式覆盖。None 时用 job 自身 create() 时存的 timeout_s,
+        缺失则回退 DEFAULT_JOB_TIMEOUT_S。分钟K长任务在 create 时存了更大阈值,
+        不被普通任务的 1200s 误杀。
         """
         with self._lock:
             jid = self._active_id
@@ -241,6 +258,8 @@ class JobStore:
             started = j.get("started_at")
             if not started:
                 return
+            # 优先用显式传入, 其次 job 自身阈值, 最后默认值
+            effective_timeout = timeout_s if timeout_s is not None else j.get("timeout_s", DEFAULT_JOB_TIMEOUT_S)
         # 时间计算放到锁外(避免 datetime 解析持锁)。
         # started_at 形如 "2026-07-04T12:00:00Z"(start() 用 datetime.utcnow 存)。
         # 两端都用 timezone-aware UTC 比较,避免 naive/aware 混用导致 TypeError。
@@ -249,9 +268,9 @@ class JobStore:
             elapsed = (datetime.now(start_dt.tzinfo) - start_dt).total_seconds()
         except Exception:  # noqa: BLE001
             return
-        if elapsed > timeout_s:
-            logger.warning("reap_stale: 强制取消卡死 job %s (已运行 %.0fs)",
-                           jid, elapsed)
+        if elapsed > effective_timeout:
+            logger.warning("reap_stale: 强制取消卡死 job %s (已运行 %.0fs, 阈值 %ss)",
+                           jid, elapsed, effective_timeout)
             self.fail(jid, f"超时自动取消 (运行 {int(elapsed)}s, 疑似卡死)")
             # 强制释放重任务锁: 卡死的线程无法被中断, 锁永远不会自然释放。
             # job 已标记 failed, 即使僵尸线程后续写入 parquet, 下次拉取会覆盖, 安全。

backend/app/services/preferences.py

@@ -99,6 +99,16 @@ def get_minute_sync_days() -> int:
     return max(1, min(30, load().get("minute_sync_days", 5)))
 
 
+def get_minute_sync_segment_days() -> int:
+    """分钟 K 拉取的单段大小(交易日)。默认 20,范围 [5, 30]。
+
+    每段拉完后立即落盘(流式),避免全量攒内存导致 OOM。
+    段越小内存峰值越低但总耗时越长(限速 sleep 随段数线性增加);
+    物理上限 ~41 交易日(TickFlow 单次 10000 根 / 一天 241 根 ≈ 41 天),max=30 留出余量。
+    """
+    return max(5, min(30, load().get("minute_sync_segment_days", 20)))
+
+
 # ===== 数据源选择 (默认 TickFlow；第一阶段仅日K切换入口) =====
 
 _ALLOWED_DATA_PROVIDERS = {"tickflow"}

backend/tests/backtest/test_minute_fill.py

@@ -0,0 +1,104 @@
+"""分钟K精确成交 (_resolve_minute_fill / _load_minute_for_fills) 回归测试。
+
+背景: 原实现用 df.to_numpy() 转 structured array 再按字段名索引 (arr["open"])。
+当 DataFrame 含 datetime 列 + float 列时, to_numpy() 退化为 dtype=object 的二维
+数组, 字段名索引抛 IndexError: "only integers, slices... are valid indices"。
+开启 minute_fill 的回测从未成功跑通过。此测试锁定该 bug 不再复发。
+"""
+from __future__ import annotations
+
+from datetime import date, datetime
+
+import polars as pl
+
+from app.backtest.engine import BacktestEngine
+
+
+def _sample_minute_df(symbol: str = "000001.SZ") -> pl.DataFrame:
+    """构造一份带 datetime 列 + float 列的分钟K (复现 to_numpy 退化的场景)。"""
+    base = datetime(2024, 1, 2, 9, 31)
+    return pl.DataFrame({
+        "symbol": [symbol] * 4,
+        "datetime": [base.replace(hour=h, minute=m) for h, m in
+                     [(9, 31), (10, 0), (14, 0), (14, 57)]],
+        "open": [10.0, 10.5, 10.8, 10.6],
+        "high": [10.6, 10.7, 10.9, 10.7],
+        "low": [9.9, 10.4, 10.7, 10.5],
+        "close": [10.2, 10.6, 10.85, 10.65],
+        "volume": [100, 200, 150, 120],
+        "amount": [1020.0, 2120.0, 1627.0, 1278.0],
+    })
+
+
+def test_resolve_minute_fill_with_mixed_columns_no_index_error():
+    """混合列类型 (datetime + float) 不再抛 IndexError。
+
+    这是原 bug 的精确复现点: 旧实现 _resolve_minute_fill 接收 ndarray,
+    arr["open"] 在 object 数组上会炸。现在接受 DataFrame, 按列取值。
+    """
+    mdf = _sample_minute_df()
+    # 三种分支都应正常返回, 不抛 IndexError
+    assert BacktestEngine._resolve_minute_fill(mdf, ref_price=10.5, side="buy") is not None
+    assert BacktestEngine._resolve_minute_fill(mdf, ref_price=10.5, side="sell") is not None
+    # 无参考线 → VWAP 分支
+    vwap = BacktestEngine._resolve_minute_fill(mdf, ref_price=None, side="buy")
+    assert vwap is not None and vwap > 0
+
+
+def test_resolve_minute_fill_buy_cross_above_ref():
+    """买入: 价格涨破参考线 → 开盘已高于则按开盘。"""
+    mdf = _sample_minute_df()
+    # ref=9.5, 开盘 10.0 已高于 → 按开盘
+    assert BacktestEngine._resolve_minute_fill(mdf, 9.5, "buy") == 10.0
+
+
+def test_resolve_minute_fill_sell_cross_below_ref():
+    """卖出: 价格跌破参考线 → 开盘已低于则按开盘。"""
+    mdf = _sample_minute_df()
+    # ref=10.5, 开盘 10.0 已低于 → 按开盘
+    assert BacktestEngine._resolve_minute_fill(mdf, 10.5, "sell") == 10.0
+
+
+def test_resolve_minute_fill_vwap():
+    """无参考线 → VWAP = 总成交额 / 总成交量。"""
+    mdf = _sample_minute_df()
+    total_amt = 1020.0 + 2120.0 + 1627.0 + 1278.0
+    total_vol = 100 + 200 + 150 + 120
+    expected = total_amt / total_vol
+    assert BacktestEngine._resolve_minute_fill(mdf, None, "buy") == expected
+
+
+def test_resolve_minute_fill_empty_returns_none():
+    """空 DataFrame → None (降级到日K口径)。"""
+    assert BacktestEngine._resolve_minute_fill(pl.DataFrame(), None, "buy") is None
+
+
+class _FakeRepo:
+    """最小 repo 桩: get_minute_range 直接返回预构造的混合列 DataFrame。"""
+
+    def __init__(self, df: pl.DataFrame) -> None:
+        self._df = df
+
+    def get_minute_range(self, symbols, start, end, asset_type="stock") -> pl.DataFrame:  # noqa: ANN001
+        return self._df
+
+
+def test_load_minute_for_fills_returns_dataframe_dict():
+    """_load_minute_for_fills 返回 {(symbol, date_str): DataFrame}, 而非 ndarray。
+
+    锁定: cache 值类型必须是 pl.DataFrame (旧实现返回的对象后续被 .to_numpy()
+    退化成 object 数组触发 bug)。
+    """
+    df = _sample_minute_df()
+    repo = _FakeRepo(df)
+    result = BacktestEngine._load_minute_for_fills(
+        repo, ["000001.SZ"], {"2024-01-02"}, "stock",
+    )
+    assert ("000001.SZ", "2024-01-02") in result
+    val = result[("000001.SZ", "2024-01-02")]
+    # 关键断言: 返回的是 DataFrame, 可直接喂给 _resolve_minute_fill
+    assert isinstance(val, pl.DataFrame)
+    assert not val.is_empty()
+    # 端到端: load → resolve 不抛异常
+    price = BacktestEngine._resolve_minute_fill(val, None, "buy")
+    assert price is not None and price > 0

frontend/src/components/data/MinuteSyncConfig.tsx

@@ -11,17 +11,20 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string;
     queryFn: api.preferences,
   })
   const update = useMutation({
-    mutationFn: ({ enabled, days }: { enabled: boolean; days: number }) =>
-      api.updateMinuteSync(enabled, days),
+    mutationFn: ({ enabled, days, segmentDays }: { enabled: boolean; days: number; segmentDays?: number }) =>
+      api.updateMinuteSync(enabled, days, segmentDays),
     onSuccess: () => qc.invalidateQueries({ queryKey: QK.preferences }),
   })
 
   const hasMinuteCap = !!caps?.capabilities?.['kline.minute.batch']
   const enabled = prefs.data?.minute_sync_enabled ?? false
   const days = prefs.data?.minute_sync_days ?? 5
+  const segmentDays = prefs.data?.minute_sync_segment_days ?? 20
   const [localDays, setLocalDays] = useState(days)
+  const [localSegment, setLocalSegment] = useState(segmentDays)
 
   useEffect(() => { setLocalDays(days) }, [days])
+  useEffect(() => { setLocalSegment(segmentDays) }, [segmentDays])
 
   const handleToggle = () => {
     if (!hasMinuteCap) return
@@ -34,6 +37,13 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string;
     update.mutate({ enabled, days: clamped })
   }
 
+  // 段大小: 交易日/段, 步进 ±5, 范围 [5, 30]。越小越省内存但越慢。
+  const setSegment = (v: number) => {
+    const clamped = Math.max(5, Math.min(30, v))
+    setLocalSegment(clamped)
+    update.mutate({ enabled, days: localDays, segmentDays: clamped })
+  }
+
   // 清空分钟K数据 (二次确认)
   const [confirmClear, setConfirmClear] = useState(false)
   const clearMutation = useMutation({
@@ -48,8 +58,8 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string;
   const [fetchingMode, setFetchingMode] = useState<'' | '40d' | '1y'>('')
   const handleFetch = (mode: '40d' | '1y') => {
     if (!hasMinuteCap) return
-    // 两个按钮都用向前扩展模式: 从本地最早数据往前补, 叠加避免缺口
-    const fetchDays = mode === '40d' ? 40 : 365
+    // 单次获取 = 按「分段大小」拉一段 (向前扩展); 1年 = 拉365天按分段切多段
+    const fetchDays = mode === '40d' ? localSegment : 365
     setFetchingMode(mode)
     api.syncMinute(fetchDays, true).then((res) => {
       qc.invalidateQueries({ queryKey: QK.pipelineJobs })
@@ -61,7 +71,8 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string;
 
   return (
     <div className="px-4 pb-4 pt-3 border-t border-accent/20 space-y-3">
-      {/* 第 1 行: 自动同步开关 + 天数 */}
+      {/* 区块 A: 自动同步 (盘后定时拉取的偏好设置) */}
+      <div className="space-y-2.5">
       <div className="flex items-center justify-between">
         <div className="flex items-center gap-2.5">
           <button
@@ -78,7 +89,7 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string;
             />
           </button>
           <span className="text-xs text-foreground font-medium">
-            {enabled ? '盘后自动同步' : '已关闭'}
+            自动同步{enabled ? '已开启' : '已关闭'}
           </span>
         </div>
         <div className="flex items-center gap-2">
@@ -104,8 +115,45 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string;
         </div>
       </div>
 
-      {/* 第 2 行: 两个手动获取按钮 (40天快速 / 1年分段) */}
-      <div className="pt-2 border-t border-border grid grid-cols-2 gap-2">
+      {/* 分段大小: 控制单次 SDK 请求覆盖的交易日数。每段拉完即落盘,避免全量攒内存 OOM。
+          「往前获取」与「获取 1 年」共用此设置 (两者都经过 sync_and_persist_minute)。 */}
+      <div className="flex items-center justify-between">
+        <div className="flex items-center gap-1.5">
+          <span className="text-xs text-foreground font-medium">分段大小</span>
+          <span className="text-[10px] text-muted px-1 py-px rounded bg-warning/8 text-warning/80">内存优化</span>
+        </div>
+        <div className="flex items-center gap-2">
+          <div className="flex items-center">
+            <button
+              onClick={() => setSegment(localSegment - 5)}
+              disabled={!hasMinuteCap || localSegment <= 5}
+              className="h-6 w-6 flex items-center justify-center rounded-l-btn bg-elevated border border-border text-secondary hover:bg-border/50 disabled:opacity-30 transition-colors text-xs"
+            >−</button>
+            <div className="h-6 w-8 flex items-center justify-center border-y border-border text-[11px] font-mono tabular-nums bg-base text-foreground">
+              {localSegment}
+            </div>
+            <button
+              onClick={() => setSegment(localSegment + 5)}
+              disabled={!hasMinuteCap || localSegment >= 30}
+              className="h-6 w-6 flex items-center justify-center rounded-r-btn bg-elevated border border-border text-secondary hover:bg-border/50 disabled:opacity-30 transition-colors text-xs"
+            >+</button>
+          </div>
+          <span className="text-[10px] text-muted">交易日/段</span>
+        </div>
+      </div>
+      <div className="text-[10px] text-muted leading-relaxed -mt-1">
+        每段拉完即写盘,避免内存堆积。越小越省内存但越慢,默认 20 平衡。
+      </div>
+      </div>
+
+      {/* 区块 B: 手动获取 (一次性操作, 独立于上方自动同步开关) */}
+      <div className="pt-3 border-t border-border space-y-2">
+        <div className="flex items-center gap-1.5">
+          <Download className="h-3 w-3 text-secondary" />
+          <span className="text-[11px] text-secondary font-medium">手动获取</span>
+          <span className="text-[10px] text-muted">不受自动同步开关影响</span>
+        </div>
+        <div className="grid grid-cols-2 gap-2">
         <button
           onClick={() => handleFetch('40d')}
           disabled={!hasMinuteCap || fetchingMode !== ''}
@@ -114,7 +162,7 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string;
           {fetchingMode === '40d' ? (
             <><Loader2 className="h-3.5 w-3.5 animate-spin" /><span>获取中…</span></>
           ) : (
-            <><Download className="h-3.5 w-3.5" /><span>往前获取 (单次拉满)</span></>
+            <><Download className="h-3.5 w-3.5" /><span>单次获取 {localSegment} 天</span></>
           )}
         </button>
         <button
@@ -128,9 +176,14 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string;
             <><Calendar className="h-3.5 w-3.5" /><span>获取最近 1 年</span><span className="text-[9px] opacity-70">分段拉取</span></>
           )}
         </button>
+        </div>
+        <div className="text-[10px] text-muted leading-relaxed">
+          A股标的 · 前复权价格 · 从本地最早数据向前叠加 ·{' '}
+          均按上方「分段大小」分段拉取、每段即落盘
+        </div>
       </div>
 
-      {/* 第 3 行: 清空 */}
+      {/* 区块 C: 清空 (危险操作, 独立分隔) */}
       <button
         onClick={() => setConfirmClear(true)}
         disabled={clearMutation.isPending}
@@ -141,13 +194,6 @@ export function MinuteSyncConfig({ caps, onJobStart }: { caps: { label: string;
         清空分钟K数据
       </button>
 
-      {/* 说明 */}
-      <div className="text-[10px] text-muted leading-relaxed">
-        A股标的 · 前复权价格 · 均从本地最早数据向前叠加 ·{' '}
-        <span className="text-accent">单次拉满</span>约 40 个交易日,{' '}
-        <span className="text-amber-400">1 年</span>按月分段 (速度较慢)
-      </div>
-
       {/* 清空确认弹窗 */}
       {confirmClear && (
         <div className="fixed inset-0 z-50 flex items-center justify-center">

frontend/src/lib/api.ts

@@ -802,6 +802,7 @@ export interface Preferences {
   indices_nav_pinned: boolean
   minute_sync_enabled: boolean
   minute_sync_days: number
+  minute_sync_segment_days: number
   daily_data_provider?: string
   adj_factor_provider?: string
   minute_data_provider?: string
@@ -942,10 +943,14 @@ export const api = {
       '/api/settings/preferences/data-providers',
       { method: 'PUT', body: JSON.stringify(cfg) },
     ),
-  updateMinuteSync: (enabled: boolean, days: number) =>
+  updateMinuteSync: (enabled: boolean, days: number, segmentDays?: number) =>
     request<Preferences>('/api/settings/preferences/minute-sync', {
       method: 'PUT',
-      body: JSON.stringify({ minute_sync_enabled: enabled, minute_sync_days: days }),
+      body: JSON.stringify({
+        minute_sync_enabled: enabled,
+        minute_sync_days: days,
+        ...(segmentDays != null ? { minute_sync_segment_days: segmentDays } : {}),
+      }),
     }),
   updatePipelinePullTypes: (cfg: Partial<Pick<Preferences, 'pipeline_pull_a_share' | 'pipeline_pull_etf' | 'pipeline_pull_index'>>) =>
     request<{