feat: ETF 支持(选股 / 回测 / 监控) (#61)

* feat(screener): 选股引擎支持 ETF

- 12 个内置策略打 asset_types 白名单 + strategy_supports_asset;涨停类
  (连板/断板反包)仅股票,其余 10 个技术类对 ETF 开放
- ScreenerService(repo, asset_type) 分流取数,ETF 复用 kline_etf_enriched,
  跳过股票专用历史缓存与涨停信号;进程级 _history_cache key 含 asset_type
- API /run、/run_preset 透传 asset_type;/strategies 按资产过滤;
  股票专有策略在 ETF 下返回空
- 新增 enriched_dirname(asset_type) 共享 helper;get_enriched_latest_asset
  增 refresh 参数(供轮询线程避免冷缓存同步重算)
- 前端「策略」页加 股票/ETF 切换,ETF 走实时单跑(空日期→用 ETF 自身最新日);
  QK.screenerStrategies 按 asset_type keyed
- 测试:test_screener_etf.py

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(backtest): 回测支持 ETF(个股/因子/策略组合)

- 三条回测路径 + 共用 BacktestEngine 面板加载按 asset_type 路由到
  kline_etf_enriched(复用 enriched_dirname);PanelCache key 隔离资产;
  ETF 跳过股票专用 get_enriched_range 缓存
- 面板 compute_all/名称 JOIN 按 asset_type 取维表(get_instruments_asset),
  修复 ETF 策略回测用错股票维表致名称为空/涨停信号算错
- BacktestConfig/FactorConfig/StrategyBacktestConfig 增 asset_type
- 三个回测 API + SSE stream 透传 asset_type;_make_job_key 纳入 asset_type
  (修复 stream 与 cancel job_key 不对齐致取消失效的回归)
- 前端策略组合页/因子页加 股票/ETF 切换,标的搜索与策略列表跟随资产;
  assetType 持久化
- 测试:test_backtest_etf.py(含 job_key 一致性回归);既有回测测试替身同步

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(monitor): 监控规则支持 ETF

- engine.evaluate(df, asset_type) 按规则 asset_type 分轮评估;quote_service
  增开 ETF 评估轮(用 ETF enriched 快照),股票轮不受影响、不重置其策略结果
- ETF 评估轮独立 try(异常不丢弃已算出的股票告警)+ refresh=False(不在轮询
  线程触发 ETF 冷缓存同步重算)
- ETF 版历史加载器(main.py 注入)+ 按规则 asset_type 选加载器
- _strategy_pools 按 (sid, asset_type) 键,避免同策略股票/ETF 规则互相覆盖
- name_map 仅在有 ETF 规则时补 ETF 维表, setdefault 保股票名优先
- RuleModel/normalize 增 asset_type(默认 stock,持久化往返)
- 前端 RuleEditor 加 股票/ETF 选择,策略列表与标的搜索跟随资产
- 测试:test_monitor_etf.py

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* feat(etf): 前端 API 绑定透传 asset_type + 文档

- api.ts: screener/backtest 绑定加 assetType 参数,MonitorRule 类型加 asset_type
- docs/features.md: 标注选股/回测/监控的 ETF 支持范围与前提

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

* fix(reliability): 管道并发/原子写/能力探测/监控告警多处加固

后端可靠性专项修复(均带回归测试, backend 全套 64 passed):

并发与数据完整性:
- 盘后管道单飞: JobStore.create() 去重纳入 pending∨running, 关闭"两次快速点击"
  并发双跑窗口; 新增 _heavy_run_lock 执行槽挡住 reap 后僵尸线程并发写 parquet
- adj_factor/minute 全部改走原子写(tmp+replace), 消除 kill/断电致 all.parquet 损坏
- 分块拉取失败聚合 WARNING 可见化(不再静默当成功); 复权失败标的会保持旧价已提示

能力探测:
- 周期重探(60min)热更新 app.state.capabilities, 付费 Key 过期/续费无需重启即可见
- 瞬时探测失败(超时/连接/5xx, 按 _is_transient 判定)不降级、保留旧付费档;
  真 401/无权限仍正常降级回落 free-api

监控告警:
- 评估仅在连续竞价(9:30-11:30/13:00-15:00)+ 快照当日新鲜度下进行, 避开集合竞价/
  收盘后陈旧价与节假日误告警
- scope=sector fail-closed(validate 拒绝新建 + _apply_scope 返回空), 修复板块规则
  对全市场刷屏
- 飞书 webhook 加退避重试并移到独立线程池 fire-and-forget, 不再阻塞行情轮询线程

单标的新鲜度: 新增 repo.symbols_lagging() 检测掉队标的并 WARNING + 计入 job 结果

Co-Authored-By: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

---------

Co-authored-by: Claude Opus 4.8 (1M context) <noreply@anthropic.com>

backend/app/api/backtest.py

@@ -86,6 +86,7 @@ class BacktestRequest(BaseModel):
     fees_pct: float = 0.0002
     slippage_bps: float = 5
     matching: Literal["close_t", "open_t+1"] = "close_t"
+    asset_type: str = "stock"
 
 
 @router.post("/run")
@@ -107,6 +108,7 @@ def run(req: BacktestRequest, request: Request):
         fees_pct=req.fees_pct,
         slippage_bps=req.slippage_bps,
         matching=req.matching,
+        asset_type=req.asset_type,
     )
     try:
         result = svc.run(cfg)
@@ -140,6 +142,7 @@ class FactorBacktestRequest(BaseModel):
     weight: Literal["equal", "factor_weight"] = "equal"
     fees_pct: float = 0.0002
     slippage_bps: float = 5.0
+    asset_type: str = "stock"
 
 
 @router.post("/factor/run")
@@ -170,6 +173,7 @@ def factor_run(req: FactorBacktestRequest, request: Request):
         weight=req.weight,
         fees_pct=req.fees_pct,
         slippage_bps=req.slippage_bps,
+        asset_type=req.asset_type,
     )
     result = svc.run(cfg)
     return asdict(result)
@@ -200,6 +204,7 @@ class StrategyBacktestRequest(BaseModel):
     position_sizing: Literal["equal", "score_weight"] = "equal"
     mode: Literal["position", "full"] = "position"
     holding_days: int = 5
+    asset_type: str = "stock"
 
 
 @router.post("/strategy/run")
@@ -235,6 +240,7 @@ def strategy_run(req: StrategyBacktestRequest, request: Request):
         position_sizing=req.position_sizing,
         mode=req.mode,
         holding_days=req.holding_days,
+        asset_type=req.asset_type,
     )
     result = svc.run(cfg)
     return asdict(result)
@@ -282,8 +288,9 @@ def _make_job_key(
     params: str | None, overrides: str | None,
     mode: str = "position", holding_days: int = 5,
     commission_pct: float | None = None, stamp_tax_pct: float | None = None,
+    asset_type: str = "stock",
 ) -> str:
-    raw = f"{strategy_id}|{symbols}|{start}|{end}|{matching}|{entry_fill}|{exit_fill}|{fees_pct}|{slippage_bps}|{max_positions}|{max_exposure_pct}|{initial_capital}|{position_sizing}|{params}|{overrides}|{mode}|{holding_days}|{commission_pct}|{stamp_tax_pct}"
+    raw = f"{strategy_id}|{symbols}|{start}|{end}|{matching}|{entry_fill}|{exit_fill}|{fees_pct}|{slippage_bps}|{max_positions}|{max_exposure_pct}|{initial_capital}|{position_sizing}|{params}|{overrides}|{mode}|{holding_days}|{commission_pct}|{stamp_tax_pct}|{asset_type}"
     return hashlib.md5(raw.encode()).hexdigest()[:12]
 
 
@@ -309,6 +316,7 @@ async def strategy_stream(
     overrides: str | None = None,
     mode: str = "position",
     holding_days: int = 5,
+    asset_type: str = "stock",
 ):
     """SSE 流式策略回测: 实时推送进度, 完成后推送结果, 支持重连 (刷新/切页后恢复)。
 
@@ -349,6 +357,7 @@ async def strategy_stream(
         params, overrides,
         mode, holding_days,
         commission_pct, stamp_tax_pct,
+        asset_type=asset_type,
     )
 
     _cleanup_stale_jobs()
@@ -391,6 +400,7 @@ async def strategy_stream(
                 position_sizing=position_sizing,
                 mode=mode,
                 holding_days=int(holding_days),
+                asset_type=asset_type,
             )
 
             def _run_backtest():
@@ -481,6 +491,7 @@ async def strategy_cancel(request: Request):
         int(_get("holding_days", "5")),
         commission_pct=_get_opt_float("commission_pct"),
         stamp_tax_pct=_get_opt_float("stamp_tax_pct"),
+        asset_type=_get("asset_type", "stock"),
     )
     job = _running_jobs.get(job_key)
     if job and not job.done:

backend/app/api/kline.py

@@ -582,7 +582,7 @@ async def sync_minute(request: Request):
     """手动触发分钟 K 同步(全市场)。返回 pipeline job_id 可轮询进度。"""
     import asyncio
 
-    from app.services.pipeline_jobs import job_store
+    from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot
     from app.api.data import invalidate_storage_cache
     from app.services.preferences import get_minute_sync_days
     from app.tickflow.capabilities import Cap
@@ -594,19 +594,21 @@ async def sync_minute(request: Request):
     if not _minute_allowed(capset):
         raise HTTPException(status_code=403, detail="需要 Pro+ 权限")
 
-    job_id = job_store.create()
-    existing = job_store.get(job_id)
-    if existing and existing["status"] == "running":
+    job_id, is_new = job_store.create()
+    if not is_new:
         return {"status": "reused", "job_id": job_id}
 
     async def task() -> None:
-        job_store.start(job_id)
+        if not try_acquire_run_slot():
+            job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试")
+            return
         loop = asyncio.get_event_loop()
 
         def progress(stage: str, pct: int, msg: str) -> None:
             job_store.progress(job_id, stage, pct, msg)
 
         try:
+            job_store.start(job_id)
             progress("sync_minute", 5, "解析标的池…")
             universe = sorted(set(get_pool("watchlist")) | set(get_pool("CN_Equity_A")))
             # 补充 instruments 全量标的，覆盖北交所、新股等
@@ -637,6 +639,8 @@ async def sync_minute(request: Request):
         except Exception as e:  # noqa: BLE001
             job_store.fail(job_id, str(e))
             invalidate_storage_cache()
+        finally:
+            release_run_slot()
 
     asyncio.create_task(task())
     return {"status": "started", "job_id": job_id}
@@ -668,16 +672,17 @@ async def extend_history(request: Request):
             raise HTTPException(status_code=403, detail="需要 Pro+ 权限 (batch K-line)")
 
         from app.services.extend_history import run_extend_history
-        from app.services.pipeline_jobs import job_store
+        from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot
         from app.api.data import invalidate_storage_cache
 
-        job_id = job_store.create()
-        existing = job_store.get(job_id)
-        if existing and existing["status"] == "running":
+        job_id, is_new = job_store.create()
+        if not is_new:
             return {"status": "reused", "job_id": job_id}
 
         async def task() -> None:
-            job_store.start(job_id)
+            if not try_acquire_run_slot():
+                job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试")
+                return
             loop = asyncio.get_event_loop()
 
             def progress(stage: str, pct: int, msg: str,
@@ -686,6 +691,7 @@ async def extend_history(request: Request):
                                    stage_pct=stage_pct, skip_log=skip_log)
 
             try:
+                job_store.start(job_id)
                 result = await loop.run_in_executor(
                     _long_task_executor,
                     lambda: run_extend_history(repo, capset, value, unit, on_progress=progress),
@@ -699,6 +705,8 @@ async def extend_history(request: Request):
                 logger.exception("extend_history failed: job_id=%s", job_id)
                 job_store.fail(job_id, str(e))
                 invalidate_storage_cache()
+            finally:
+                release_run_slot()
 
         asyncio.create_task(task())
         return {"status": "started", "job_id": job_id}
@@ -719,16 +727,17 @@ async def rebuild_enriched(request: Request):
     try:
         repo = request.app.state.repo
 
-        from app.services.pipeline_jobs import job_store
+        from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot
         from app.api.data import invalidate_storage_cache
 
-        job_id = job_store.create()
-        existing = job_store.get(job_id)
-        if existing and existing["status"] == "running":
+        job_id, is_new = job_store.create()
+        if not is_new:
             return {"status": "reused", "job_id": job_id}
 
         async def task() -> None:
-            job_store.start(job_id)
+            if not try_acquire_run_slot():
+                job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试")
+                return
             loop = asyncio.get_event_loop()
 
             def progress(stage: str, pct: int, msg: str,
@@ -737,6 +746,7 @@ async def rebuild_enriched(request: Request):
                                    stage_pct=stage_pct, skip_log=skip_log)
 
             try:
+                job_store.start(job_id)
                 progress("rebuild_enriched", 10, "全量计算 enriched…")
                 from app.indicators.pipeline import run_pipeline
 
@@ -777,6 +787,8 @@ async def rebuild_enriched(request: Request):
                 logger.exception("rebuild_enriched failed: job_id=%s", job_id)
                 job_store.fail(job_id, str(e))
                 invalidate_storage_cache()
+            finally:
+                release_run_slot()
 
         asyncio.create_task(task())
         return {"status": "started", "job_id": job_id}
@@ -838,16 +850,17 @@ async def extend_minute_history(request: Request):
         if total_days <= 0:
             raise HTTPException(status_code=400, detail="扩展范围无效")
 
-        from app.services.pipeline_jobs import job_store
+        from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot
         from app.api.data import invalidate_storage_cache
 
-        job_id = job_store.create()
-        existing = job_store.get(job_id)
-        if existing and existing["status"] == "running":
+        job_id, is_new = job_store.create()
+        if not is_new:
             return {"status": "reused", "job_id": job_id}
 
         async def task() -> None:
-            job_store.start(job_id)
+            if not try_acquire_run_slot():
+                job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试")
+                return
             loop = asyncio.get_event_loop()
 
             def progress(stage: str, pct: int, msg: str,
@@ -856,6 +869,7 @@ async def extend_minute_history(request: Request):
                                    stage_pct=stage_pct, skip_log=skip_log)
 
             try:
+                job_store.start(job_id)
                 # 获取当前最早日期
                 earliest = repo.earliest_minute_date()
                 if not earliest:
@@ -925,7 +939,8 @@ async def extend_minute_history(request: Request):
                             else:
                                 day_df = day_df.drop("_trade_date")
                             day_df = day_df.sort("symbol", "datetime")
-                            day_df.write_parquet(out)
+                            from app.services.kline_sync import _atomic_write_parquet
+                            _atomic_write_parquet(day_df, out)
                             written += day_df.height
                             day_count += 1
 
@@ -955,6 +970,8 @@ async def extend_minute_history(request: Request):
                 logger.exception("extend_minute_history failed: job_id=%s", job_id)
                 job_store.fail(job_id, str(e))
                 invalidate_storage_cache()
+            finally:
+                release_run_slot()
 
         asyncio.create_task(task())
         return {"status": "started", "job_id": job_id}

backend/app/api/monitor_rules.py

@@ -38,6 +38,7 @@ class RuleModel(BaseModel):
     name: str
     enabled: bool = True
     type: str          # strategy | signal | price | market
+    asset_type: str = "stock"   # stock | etf (etf: strategy 型走 ETF 历史加载器)
     scope: str = "symbols"   # symbols | all | sector
     symbols: list[str] = []
     sector: str | None = None

backend/app/api/pipeline.py

@@ -8,7 +8,7 @@ import logging
 from fastapi import APIRouter, HTTPException, Request
 
 from app.jobs import daily_pipeline
-from app.services.pipeline_jobs import job_store
+from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot
 from app.api.data import invalidate_storage_cache
 
 # 长时间任务专用线程池（隔离于 FastAPI 默认线程池，防止阻塞请求处理）
@@ -33,23 +33,25 @@ async def run_now(request: Request) -> dict:
     # reap_stale 会在 /run 和 /jobs/{id} 轮询端点都调用,保证卡死后能自愈。
     job_store.reap_stale()
 
-    job_id = job_store.create()
-
-    # 如果是复用的 active job,直接返回(不重启)
-    existing = job_store.get(job_id)
-    if existing and existing["status"] == "running":
+    # 单飞: 复用任何活跃 (pending∨running) 任务, is_new=False 时不再调度新任务
+    job_id, is_new = job_store.create()
+    if not is_new:
         return {"job_id": job_id, "reused": True}
 
     # 在 executor 里跑同步任务(pipeline 内部都是阻塞 IO + CPU)
     async def task() -> None:
-        job_store.start(job_id)
-        loop = asyncio.get_event_loop()
-
-        def progress(stage: str, pct: int, msg: str, stage_pct: int | None = None,
-                     skip_log: bool = False) -> None:
-            job_store.progress(job_id, stage, pct, msg, stage_pct=stage_pct, skip_log=skip_log)
-
+        # 重任务执行槽: 防僵尸并发(reap 后线程仍活时新任务不得并行写 parquet)
+        if not try_acquire_run_slot():
+            job_store.fail(job_id, "已有数据任务在运行(或上一次任务卡死未结束),请稍后再试")
+            return
         try:
+            job_store.start(job_id)
+            loop = asyncio.get_event_loop()
+
+            def progress(stage: str, pct: int, msg: str, stage_pct: int | None = None,
+                         skip_log: bool = False) -> None:
+                job_store.progress(job_id, stage, pct, msg, stage_pct=stage_pct, skip_log=skip_log)
+
             result = await loop.run_in_executor(
                 _long_task_executor,
                 lambda: daily_pipeline.run_now(repo, capset, on_progress=progress),
@@ -61,6 +63,8 @@ async def run_now(request: Request) -> dict:
             logger.exception("pipeline failed")
             job_store.fail(job_id, str(e))
             invalidate_storage_cache()
+        finally:
+            release_run_slot()
 
     asyncio.create_task(task())
     return {"job_id": job_id, "reused": False}

backend/app/api/screener.py

@@ -12,7 +12,7 @@ from typing import Any, Optional
 from fastapi import APIRouter, HTTPException, Query, Request
 from pydantic import BaseModel
 
-from app.services.screener import PRESET_STRATEGIES, ScreenerService
+from app.services.screener import PRESET_STRATEGIES, ScreenerService, strategy_supports_asset
 from app.services import strategy_cache
 from app.strategy import config as strategy_config
 
@@ -28,6 +28,7 @@ class CustomRequest(BaseModel):
     pool: Optional[list[str]] = None
     as_of: Optional[date] = None
     ext_columns: Optional[str] = None
+    asset_type: str = "stock"
 
 
 class PresetRequest(BaseModel):
@@ -35,6 +36,7 @@ class PresetRequest(BaseModel):
     pool: Optional[list[str]] = None
     as_of: Optional[date] = None
     ext_columns: Optional[str] = None
+    asset_type: str = "stock"
 
 
 def _safe(result_dict: dict) -> dict:
@@ -174,23 +176,25 @@ def _update_cache_strategy(data_dir, as_of: str, strategy_id: str, safe_data: di
 
 
 @router.get("/strategies")
-def strategies(request: Request):
-    """策略清单（内置 + 自定义 + AI）。"""
+def strategies(request: Request, asset_type: str = Query("stock")):
+    """策略清单（内置 + 自定义 + AI）。按 asset_type 过滤：ETF 仅返回技术类内置策略。"""
     data_dir = request.app.state.repo.store.data_dir
     presets = []
     seen_ids: set[str] = set()
 
     # 内置策略
     for k, v in PRESET_STRATEGIES.items():
+        if not strategy_supports_asset(v, asset_type):
+            continue
         overrides = strategy_config.load_override(data_dir, k)
         name = (overrides.get("name") or v["name"]) if overrides else v["name"]
         desc = (overrides.get("description") or v["description"]) if overrides else v["description"]
         presets.append({"id": k, "name": name, "description": desc, "source": "builtin"})
         seen_ids.add(k)
 
-    # 自定义/AI 策略（不在 PRESET_STRATEGIES 中的）
+    # 自定义/AI 策略（不在 PRESET_STRATEGIES 中的）; 未标注资产类型, 保守仅 stock 返回
     engine = getattr(request.app.state, "strategy_engine", None)
-    if engine:
+    if engine and asset_type == "stock":
         for meta in engine.list_strategies():
             sid = meta["id"]
             if sid not in seen_ids:
@@ -210,7 +214,7 @@ def strategies(request: Request):
 @router.post("/run")
 def run_custom(req: CustomRequest, request: Request):
     repo = request.app.state.repo
-    svc = ScreenerService(repo)
+    svc = ScreenerService(repo, asset_type=req.asset_type)
     as_of = req.as_of or svc.latest_date()
     if not as_of:
         raise HTTPException(status_code=400,
@@ -230,7 +234,7 @@ def run_custom(req: CustomRequest, request: Request):
 @router.post("/run_preset")
 def run_preset(req: PresetRequest, request: Request):
     repo = request.app.state.repo
-    svc = ScreenerService(repo)
+    svc = ScreenerService(repo, asset_type=req.asset_type)
     as_of = req.as_of or svc.latest_date()
     if not as_of:
         raise HTTPException(status_code=400, detail="无可用数据日期")

backend/app/backtest/engine.py

@@ -134,8 +134,9 @@ class PanelCache:
         end: date,
         columns: list[str] | None,
         compute_fn,
+        asset_type: str = "stock",
     ) -> pl.DataFrame:
-        key = self._make_key(symbols, start, end, columns)
+        key = self._make_key(symbols, start, end, columns, asset_type)
         now = time.monotonic()
 
         if key in self._cache:
@@ -145,7 +146,7 @@ class PanelCache:
                 return entry.df
             del self._cache[key]
 
-        df = compute_fn(symbols, start, end, columns)
+        df = compute_fn(symbols, start, end, columns, asset_type)
         self._cache[key] = _CacheEntry(df=df, ts=now)
         if len(self._cache) > self._max_size:
             self._cache.popitem(last=False)
@@ -155,13 +156,13 @@ class PanelCache:
         self._cache.clear()
 
     @staticmethod
-    def _make_key(symbols: list[str] | None, start: date, end: date, columns: list[str] | None) -> str:
+    def _make_key(symbols: list[str] | None, start: date, end: date, columns: list[str] | None, asset_type: str = "stock") -> str:
         if symbols is None:
             h = "all"
         else:
             h = hashlib.md5(",".join(sorted(symbols)).encode()).hexdigest()[:12]
         cols = "all" if columns is None else hashlib.md5(",".join(sorted(columns)).encode()).hexdigest()[:8]
-        return f"{h}:{start}:{end}:{cols}"
+        return f"{asset_type}:{h}:{start}:{end}:{cols}"
 
 
 # ================================================================
@@ -183,9 +184,10 @@ class BacktestEngine:
         start: date,
         end: date,
         columns: list[str] | None = None,
+        asset_type: str = "stock",
     ) -> pl.DataFrame:
-        """加载 enriched 数据面板，带缓存。"""
-        return self._cache.get_or_compute(symbols, start, end, columns, self._load_panel_inner)
+        """加载 enriched 数据面板，带缓存。asset_type='etf' 时读 ETF enriched。"""
+        return self._cache.get_or_compute(symbols, start, end, columns, self._load_panel_inner, asset_type=asset_type)
 
     def _load_panel_inner(
         self,
@@ -193,12 +195,13 @@ class BacktestEngine:
         start: date,
         end: date,
         columns: list[str] | None = None,
+        asset_type: str = "stock",
     ) -> pl.DataFrame:
         t0 = time.perf_counter()
 
-        # 近期区间优先复用 repository 的预计算 enriched 历史缓存，避免重复 scan_parquet + compute_all。
+        # 近期区间优先复用 repository 的预计算 enriched 历史缓存 (仅 stock: 该缓存为股票专用)。
         try:
-            if self.repo is not None and hasattr(self.repo, "get_enriched_range"):
+            if asset_type == "stock" and self.repo is not None and hasattr(self.repo, "get_enriched_range"):
                 cached = self.repo.get_enriched_range(start, end, symbols=symbols, columns=columns)
                 if cached is not None and not cached.is_empty():
                     elapsed = (time.perf_counter() - t0) * 1000
@@ -207,7 +210,8 @@ class BacktestEngine:
         except Exception as e:  # noqa: BLE001
             logger.debug("backtest load panel cache miss: %s", e)
 
-        enriched_glob = str(self.repo.store.data_dir / "kline_daily_enriched" / "**" / "*.parquet")
+        from app.tickflow.repository import enriched_dirname
+        enriched_glob = str(self.repo.store.data_dir / enriched_dirname(asset_type) / "**" / "*.parquet")
 
         try:
             lf = pl.scan_parquet(enriched_glob)
@@ -242,7 +246,9 @@ class BacktestEngine:
             return df
 
         from app.indicators.pipeline import compute_all
-        instruments = self.repo.get_instruments()
+        # 按 asset_type 取维表: ETF 回测须用 ETF 维表, 否则名称 JOIN 失败(全 null)、
+        # 涨停信号算在错误的 instruments 上。
+        instruments = self.repo.get_instruments_asset(asset_type)
         df = compute_all(df, instruments=instruments)
         if not instruments.is_empty() and "name" not in df.columns:
             inst_cols = [c for c in ["symbol", "name"] if c in instruments.columns]

backend/app/backtest/factor.py

@@ -52,6 +52,7 @@ class FactorConfig:
     weight: Literal["equal", "factor_weight"] = "equal"
     fees_pct: float = 0.0002
     slippage_bps: float = 5.0
+    asset_type: str = "stock"
 
 
 @dataclass
@@ -117,6 +118,7 @@ class FactorBacktestService:
             load_start,
             config.end,
             columns=panel_columns,
+            asset_type=config.asset_type,
         )
         if panel.is_empty():
             return _err("无数据，请检查日期范围或先运行盘后管道")

backend/app/backtest/strategy.py

@@ -43,6 +43,7 @@ class StrategyBacktestConfig:
     initial_capital: float = 1_000_000.0
     position_sizing: Literal["equal", "score_weight"] = "equal"
     mode: Literal["position", "full"] = "position"
+    asset_type: str = "stock"
     holding_days: int = 5
 
     def __post_init__(self) -> None:
@@ -149,7 +150,7 @@ class StrategyBacktestService:
             load_end = config.end + timedelta(days=fwd_buffer * 2)  # 日历日放宽, 确保覆盖 N 个交易日
 
         t_load = time.perf_counter()
-        panel = self.engine.load_panel(config.symbols, load_start, load_end)
+        panel = self.engine.load_panel(config.symbols, load_start, load_end, asset_type=config.asset_type)
         timing_ms["load_panel"] = round((time.perf_counter() - t_load) * 1000, 1)
         if panel.is_empty():
             return _err("无数据，请检查日期范围或先运行盘后管道")

backend/app/jobs/daily_pipeline.py

@@ -17,6 +17,7 @@ from pathlib import Path
 import polars as pl
 from apscheduler.schedulers.asyncio import AsyncIOScheduler
 from apscheduler.triggers.cron import CronTrigger
+from apscheduler.triggers.interval import IntervalTrigger
 
 from app.indicators.pipeline import run_pipeline
 from app.config import settings
@@ -180,6 +181,20 @@ def run_now(
         logger.info("sync_daily: [%s ~ %s] done", start_date, today)
     _invalidate("daily")
 
+    # 单标的新鲜度: 全局 max(date) 会被任一有今日数据的标的"拉高", 掩盖停牌/复牌/
+    # 一直拉失败而掉队的个股缺口(全局判据只刷"今天", 永不回补掉队标的的历史缺口)。
+    # 这里检测并**可见化**(WARNING + 计入结果), 让掉队标的不再隐形。
+    # (自动回补暂不做 —— 需带退市判定, 否则对已退市标的每轮空拉浪费 API 额度。)
+    lagging_symbols: list[str] = []
+    if pull_a_share and latest_daily:
+        try:
+            lagging_symbols = repo.symbols_lagging(today, min_gap_days=3)
+            if lagging_symbols:
+                logger.warning("日K新鲜度: %d 只标的落后 >3 日 (停牌/退市/拉取失败; 样例: %s)",
+                               len(lagging_symbols), lagging_symbols[:10])
+        except Exception as e:  # noqa: BLE001
+            logger.warning("laggard detection failed: %s", e)
+
     # Step 1.5: 同步除权因子 — 范围与日K拉取方式对齐
     #   日K范围拉取(补缺口/首次) → 除权用日K范围 [daily_range_start, now]
     #     首次会覆盖整个日K区间内的历史除权事件; 补缺口天然只增量(起点=latest_daily≈昨天)
@@ -462,6 +477,7 @@ def run_now(
         "etf_daily_rows": written_etf_daily,
         "etf_adj_factor_symbols": etf_adj_symbols,
         "minute_rows": written_minute,
+        "lagging_symbols": len(lagging_symbols),
         "skipped_stages": skipped,
     }
 
@@ -543,23 +559,36 @@ def _refresh_instruments_view(repo: KlineRepository) -> None:
 
 
 def _run_tracked(fn, job_label: str) -> None:
-    """调度触发时包装 JobStore 跟踪，确保同步历史有记录。"""
-    from app.services.pipeline_jobs import job_store
+    """调度触发时包装 JobStore 跟踪，确保同步历史有记录。
 
-    job_id = job_store.create()
-    job_store.start(job_id)
+    单飞: 若已有活跃(pending∨running)任务(手动同步中), 本次调度直接跳过, 不并发。
+    重任务执行槽: 再挡一层僵尸并发(reap 后线程仍活时不得并行写 parquet)。
+    """
+    from app.services.pipeline_jobs import job_store, release_run_slot, try_acquire_run_slot
+
+    job_id, is_new = job_store.create()
+    if not is_new:
+        logger.info("scheduled %s 跳过: 已有活跃任务在运行 (job_id=%s)", job_label, job_id)
+        return
+    if not try_acquire_run_slot():
+        logger.warning("scheduled %s 跳过: 重任务执行槽被占用(疑似上次任务卡死)", job_label)
+        job_store.fail(job_id, f"scheduled {job_label} skipped: 已有数据任务在运行")
+        return
 
     def progress(stage: str, pct: int, msg: str, stage_pct: int | None = None,
                  skip_log: bool = False) -> None:
         job_store.progress(job_id, stage, pct, msg, stage_pct=stage_pct, skip_log=skip_log)
 
     try:
+        job_store.start(job_id)
         result = fn(on_progress=progress)
         job_store.succeed(job_id, result)
         logger.info("scheduled %s completed: job_id=%s", job_label, job_id)
     except Exception:
         logger.exception("scheduled %s failed: job_id=%s", job_label, job_id)
         job_store.fail(job_id, f"scheduled {job_label} failed")
+    finally:
+        release_run_slot()
 
 
 # ================================================================
@@ -799,7 +828,11 @@ def start_scheduler(repo: KlineRepository, capset: CapabilitySet) -> AsyncIOSche
         # 与手动触发 (/api/pipeline/run) 对齐: 管道落盘后重建 Polars 内存缓存,
         # 否则 live_agg 的昨日连板数等基准列会停留在旧交易日, 次日开盘连板梯队
         # 整体少算一档 (仅手动触发或重启才会刷缓存, cron 调度路径此前漏了这步)。
-        result = run_now(repo, capset, on_progress=on_progress)
+        # 用 app.state 上的**实时** capset(周期重探会热更新它), 而非启动时捕获的
+        # 旧 capset —— 否则 Key 中途过期/续费后, 调度管道仍按旧档位打端点。
+        app_state = _get_app_state()
+        capset_live = getattr(app_state, "capabilities", None) or capset
+        result = run_now(repo, capset_live, on_progress=on_progress)
         repo.refresh_cache()
         return result
 
@@ -831,6 +864,36 @@ def start_scheduler(repo: KlineRepository, capset: CapabilitySet) -> AsyncIOSche
         replace_existing=True,
     )
 
+    # 周期性能力重探: 付费 Key 中途过期/续费无需重启即可被发现。
+    # 只热更新 app.state.capabilities(API 端点、盘后管道 _pipeline_then_refresh 均读它);
+    # 档位变化记 WARNING, 让「Key 失效」在日志/前端可见, 不再静默按旧档位打 403 端点。
+    def _reprobe_capabilities():
+        from app.tickflow.policy import detect_capabilities, tier_label
+        app_state = _get_app_state()
+        if app_state is None:
+            return
+        try:
+            old = getattr(app_state, "capabilities", None)
+            old_n = len(old.all()) if old else -1
+            new_capset = detect_capabilities(force=True)
+            app_state.capabilities = new_capset
+            new_n = len(new_capset.all())
+            if old_n != new_n:
+                logger.warning(
+                    "能力集变化: %d → %d capabilities (档位=%s)。Key 过期/续费或端点波动, "
+                    "已热更新 app.state.capabilities。", old_n, new_n, tier_label(),
+                )
+        except Exception as e:  # noqa: BLE001
+            logger.warning("周期能力重探失败(保留现有能力集): %s", e)
+
+    scheduler.add_job(
+        _reprobe_capabilities,
+        trigger=IntervalTrigger(minutes=60),
+        id="reprobe_capabilities",
+        misfire_grace_time=600,
+        replace_existing=True,
+    )
+
     # 定时复盘 (AI 大盘复盘报告): 工作日到点自动生成并归档。
     # 默认关闭 —— 仅当用户在复盘页开启时才注册 job。
     # 复用 recap_market_once(非流式) + market_recap_reports.save_report(落盘)。

backend/app/main.py

@@ -130,6 +130,7 @@ async def lifespan(app: FastAPI):
     from app.services.screener import ScreenerService
 
     _screener_svc = ScreenerService(repo)
+    _etf_screener_svc = ScreenerService(repo, asset_type="etf")
     strategy_dirs = [
         Path(__file__).resolve().parent / "strategy" / "builtin",
         store.data_dir / "strategies" / "custom",
@@ -153,6 +154,8 @@ async def lifespan(app: FastAPI):
     # 复用 ScreenerService 的历史窗口加载器 (三级缓存, 启动预计算命中 ~0ms),
     # 让声明 filter_history 的策略 (如反包) 也能在实时监控里跑选股 → 盘中触发通知。
     monitor_engine.set_history_loader(_screener_svc._load_enriched_history)
+    # ETF 版历史加载器: asset_type=etf 的 strategy 型规则用 (读 kline_etf_enriched)。
+    monitor_engine.set_history_loader_etf(_etf_screener_svc._load_enriched_history)
 
     # 自动迁移: 把旧 strategy_monitor_ids 同步为 type=strategy 规则 (统一到监控页)
     try:

backend/app/services/backtest.py

@@ -87,6 +87,7 @@ class BacktestConfig:
     matching: Literal["close_t", "open_t+1"] = "close_t"
     rsi_oversold_threshold: float = 30
     rsi_overbought_threshold: float = 70
+    asset_type: str = "stock"
 
 
 @dataclass
@@ -125,13 +126,16 @@ class BacktestService:
         symbols: list[str],
         start: date,
         end: date,
+        asset_type: str = "stock",
     ) -> pd.DataFrame:
         """加载 [date × symbol] 价格面板 — Polars scan_parquet + 即时计算指标。
 
         **全项目唯一从 Polars 转 pandas 的边界**(§7.4 / ADR-19)。
+        asset_type='etf' 时读 ETF enriched。
         """
         try:
-            enriched_glob = str(self.repo.store.data_dir / "kline_daily_enriched" / "**" / "*.parquet")
+            from app.tickflow.repository import enriched_dirname
+            enriched_glob = str(self.repo.store.data_dir / enriched_dirname(asset_type) / "**" / "*.parquet")
             df = (
                 pl.scan_parquet(enriched_glob)
                 .filter(
@@ -203,7 +207,7 @@ class BacktestService:
         vbt = _get_vbt()
         run_id = uuid.uuid4().hex[:10]
 
-        panel = self._load_panel(config.symbols, config.start, config.end)
+        panel = self._load_panel(config.symbols, config.start, config.end, config.asset_type)
         if panel.empty:
             return BacktestResult(
                 run_id=run_id,

backend/app/services/kline_sync.py

@@ -23,6 +23,20 @@ from app.tickflow.repository import KlineRepository
 logger = logging.getLogger(__name__)
 
 
+def _atomic_write_parquet(df: pl.DataFrame, out) -> None:
+    """先写临时文件再原子替换, 避免进程中断留下损坏的 parquet。
+
+    与 repository._atomic_write_parquet 同语义。adj_factor 的 all.parquet 是全市场
+    单文件、每次「读→concat→原地写」, 直接 write_parquet(out) 在进程被 kill
+    (dev.sh 清端口用 kill -9)、reap 超时或断电时会留下半截文件, 之后复权视图
+    scan_parquet 整条链路报错、enriched 全市场重算不出。临时文件后缀 .tmp 不匹配
+    *.parquet glob, 不会被扫描误读。
+    """
+    tmp = out.with_name(out.name + ".tmp")
+    df.write_parquet(tmp)
+    tmp.replace(out)  # 同目录 rename, POSIX/NTFS 均为原子操作
+
+
 # 标准列(无论 SDK 返回什么形状,我们把它规范成这套)
 CANONICAL_DAILY_COLS = [
     "symbol", "date", "open", "high", "low", "close", "volume", "amount",
@@ -77,15 +91,20 @@ def sync_daily_batch(symbols: list[str],
                      rpm: int | None = None,
                      start_time: datetime | None = None,
                      end_time: datetime | None = None,
-                     on_chunk_done: Callable[[int, int], None] | None = None) -> pl.DataFrame:
+                     on_chunk_done: Callable[[int, int], None] | None = None,
+                     failed_out: list[str] | None = None) -> pl.DataFrame:
     """批量拉取多股日 K。
 
     优先使用 start_time / end_time 区间 + count=10000,确保覆盖完整时间段。
     仅传 count 时按条数回溯。
+
+    failed_out: 可选出参。拉取失败的分块标的会追加进该 list, 供上层判定「部分失败」
+                而非静默当成功(某分块断网 → 这些标的本轮未更新, 保持旧数据)。
     """
     tf = get_client()
     out: list[pl.DataFrame] = []
     chunks = chunked(symbols, batch_size)
+    failed_syms: list[str] = []
 
     for i, chunk in enumerate(chunks):
         sleep_between_batches(i, rpm)
@@ -102,7 +121,9 @@ def sync_daily_batch(symbols: list[str],
                 raw = tf.klines.batch(chunk, period="1d", count=count or 250, adjust="none",
                                       as_dataframe=True, show_progress=False)
         except Exception as e:  # noqa: BLE001
-            logger.warning("batch fetch failed for %d symbols: %s", len(chunk), e)
+            logger.warning("batch fetch failed for %d symbols (chunk %d/%d): %s",
+                           len(chunk), i + 1, len(chunks), e)
+            failed_syms.extend(chunk)
             continue
 
         # 兼容两种形态:dict[sym → df] 和扁平 df
@@ -117,6 +138,13 @@ def sync_daily_batch(symbols: list[str],
         if on_chunk_done:
             on_chunk_done(i + 1, len(chunks))
 
+    # 部分失败可见化: 聚合一条 WARNING(而非只有逐块 debug/warning), 并回传出参。
+    if failed_syms:
+        logger.warning("日K批量同步部分失败: %d/%d 标的未获取, 本轮保持旧数据 (样例: %s)",
+                       len(failed_syms), len(symbols), failed_syms[:10])
+        if failed_out is not None:
+            failed_out.extend(failed_syms)
+
     if not out:
         return pl.DataFrame()
     return pl.concat(out, how="diagonal_relaxed")
@@ -328,9 +356,9 @@ def sync_adj_factor(symbols: list[str], repo: KlineRepository,
                 merged = pl.concat([existing, new_data]).unique(
                     subset=["symbol", "trade_date"], keep="last",
                 ).sort(["symbol", "trade_date"])
-                merged.write_parquet(out)
+                _atomic_write_parquet(merged, out)
                 return merged.height - before, affected
-            new_data.sort(["symbol", "trade_date"]).write_parquet(out)
+            _atomic_write_parquet(new_data.sort(["symbol", "trade_date"]), out)
             return new_data.height, affected
         # 自定义源未配置 adj_factor → 回退 TickFlow
 
@@ -355,6 +383,7 @@ def sync_adj_factor(symbols: list[str], repo: KlineRepository,
 
     chunks = chunked(symbols, limit.batch)
     all_dfs: list[pl.DataFrame] = []
+    failed_syms: list[str] = []
 
     for i, chunk in enumerate(chunks):
         sleep_between_batches(i, limit.rpm)
@@ -365,11 +394,18 @@ def sync_adj_factor(symbols: list[str], repo: KlineRepository,
                 all_dfs.append(normalized)
             logger.debug("adj_factor chunk %d/%d: %d symbols", i + 1, len(chunks), len(chunk))
         except Exception as e:  # noqa: BLE001
-            logger.warning("adj_factor chunk %d failed: %s", i + 1, e)
+            logger.warning("adj_factor chunk %d/%d failed: %s", i + 1, len(chunks), e)
+            failed_syms.extend(chunk)
 
         if on_chunk_done:
             on_chunk_done(i + 1, len(chunks))
 
+    # 部分失败可见化: 失败分块的标的不在 affected 里 → enriched 不会重算它们,
+    # 它们会保持**旧的前复权价**直到下次成功同步。聚合一条 WARNING 让其可见。
+    if failed_syms:
+        logger.warning("adj_factor 同步部分失败: %d/%d 标的未获取复权因子, 将保持旧复权价 (样例: %s)",
+                       len(failed_syms), len(symbols), failed_syms[:10])
+
     if not all_dfs:
         return 0, []
 
@@ -388,13 +424,13 @@ def sync_adj_factor(symbols: list[str], repo: KlineRepository,
         merged = pl.concat([existing, new_data]).unique(
             subset=["symbol", "trade_date"], keep="last",
         ).sort(["symbol", "trade_date"])
-        merged.write_parquet(out)
+        _atomic_write_parquet(merged, out)
         added = merged.height - before
         logger.info("adj_factor merged: %d total (+%d new), %d/%d symbols",
                      merged.height, added, new_data.height, len(symbols))
         return added, affected
     else:
-        new_data.sort(["symbol", "trade_date"]).write_parquet(out)
+        _atomic_write_parquet(new_data.sort(["symbol", "trade_date"]), out)
         logger.info("adj_factor synced: %d rows (%d symbols)", new_data.height, len(symbols))
         return new_data.height, affected
 
@@ -646,7 +682,7 @@ def _migrate_symbol_to_date_partition(repo: KlineRepository) -> None:
         out = minute_dir / f"date={trade_date}" / "part.parquet"
         out.parent.mkdir(parents=True, exist_ok=True)
         day_df = day_df.drop("_trade_date").sort("symbol", "datetime")
-        day_df.write_parquet(out)
+        _atomic_write_parquet(day_df, out)
 
     # 删旧目录
     for d in old_dirs:
@@ -734,7 +770,7 @@ def sync_and_persist_minute(
         else:
             day_df = day_df.drop("_trade_date")
         day_df = day_df.sort("symbol", "datetime")
-        day_df.write_parquet(out)
+        _atomic_write_parquet(day_df, out)
         written += day_df.height
 
     # 刷新视图

backend/app/services/pipeline_jobs.py

@@ -96,10 +96,21 @@ class JobStore:
 
     # ===== lifecycle =====
 
-    def create(self) -> str:
+    def create(self) -> tuple[str, bool]:
+        """单飞创建任务。返回 (job_id, is_new)。
+
+        去重条件为 **pending ∨ running**(而非仅 running):`/run` 先 create() 再在
+        后台任务里 start() 置 running,两者之间存在 pending 窗口。旧实现只在 running 时
+        复用,两次快速点击时首个 job 仍是 pending → 第二次绕过去重、另起并发任务、覆盖
+        _active_id,导致两条全市场拉取同时读改写同一 parquet。纳入 pending 后该窗口关闭。
+
+        is_new=False 表示复用了已有活跃任务,调用方**不得**再调度新的后台任务。
+        """
         with self._lock:
-            if self._active_id and self._active_jobs.get(self._active_id, {}).get("status") == "running":
-                return self._active_id
+            if self._active_id:
+                active = self._active_jobs.get(self._active_id)
+                if active and active.get("status") in ("pending", "running"):
+                    return self._active_id, False
 
             job_id = uuid.uuid4().hex[:10]
             self._active_jobs[job_id] = {
@@ -116,7 +127,7 @@ class JobStore:
                 "error": None,
             }
             self._active_id = job_id
-            return job_id
+            return job_id, True
 
     def start(self, job_id: str) -> None:
         with self._lock:
@@ -283,3 +294,32 @@ def _duration_s(j: dict[str, Any]) -> float | None:
 
 # 进程内单例
 job_store = JobStore()
+
+
+# ================================================================
+# 重任务互斥锁 — 防「僵尸并发」
+# ================================================================
+# create() 的单飞去重能挡住 pending/running 窗口内的重复点击, 但挡不住
+# reap_stale 把卡死 job 标记 failed、清掉 _active_id 之后 —— 此时 executor
+# 线程仍在跑(线程无法被中断), 下一次 /run 会视作无活跃任务而另起一条,
+# 与僵尸线程并发读改写同一 parquet。
+#
+# 该锁绑定「实际执行体(协程/线程)」的生命周期而非 job 状态: 每个重任务在真正
+# 开跑前 try_acquire_run_slot(), 结束(含异常)在 finally 里 release_run_slot()。
+# 僵尸任务因卡在 executor await 中始终未 release, 新任务 try_acquire 失败 → 快速
+# 失败而非并发执行。代价: 真卡死时需重启进程才能再次跑重任务(优先保证数据不损坏)。
+_heavy_run_lock = threading.Lock()
+
+
+def try_acquire_run_slot() -> bool:
+    """尝试占用重任务执行槽(非阻塞)。成功返回 True。"""
+    return _heavy_run_lock.acquire(blocking=False)
+
+
+def release_run_slot() -> None:
+    """释放重任务执行槽(允许跨线程释放)。"""
+    try:
+        _heavy_run_lock.release()
+    except RuntimeError:
+        # 未持有(重复释放)—— 幂等忽略
+        pass

backend/app/services/quote_service.py

@@ -26,6 +26,7 @@ from __future__ import annotations
 import logging
 import threading
 import time
+from concurrent.futures import ThreadPoolExecutor
 from datetime import date, time as dt_time
 
 import polars as pl
@@ -34,6 +35,12 @@ from app.market_time import cn_now, cn_today
 
 logger = logging.getLogger(__name__)
 
+# Webhook(飞书等)投递专用线程池 —— 与行情轮询线程隔离。
+# send_feishu 内置重试(最坏 ~3×5s 超时 + 退避), 若在 _poll_loop 上同步投递,
+# webhook 慢/宕机会逐条累加, 拖垮整条实时行情+告警轮询。这里 fire-and-forget,
+# 失败由 webhook_adapter 记 WARNING(可见), 但绝不阻塞热路径。
+_WEBHOOK_EXECUTOR = ThreadPoolExecutor(max_workers=2, thread_name_prefix="feishu-webhook")
+
 
 class QuoteSubscriber:
     """一个 SSE 连接对应一个订阅者: 独立事件 + 独立队列。
@@ -752,12 +759,28 @@ class QuoteService:
     @staticmethod
     def _is_trading_hours() -> bool:
         # 显式北京时间: 容器/服务器本地时区可能是 UTC, 用 naive now() 会整体错开轮询窗口
+        # 注: 这是**轮询**窗口(含 9:15 集合竞价与 15:05 收盘缓冲, 用于盘前预热/收盘捕捉),
+        # 比连续竞价宽。监控告警用更严格的 _is_continuous_trading。
         now = cn_now()
         t = now.time()
         morning = dt_time(9, 15) <= t <= dt_time(11, 35)
         afternoon = dt_time(12, 55) <= t <= dt_time(15, 5)
         return now.weekday() < 5 and (morning or afternoon)
 
+    @staticmethod
+    def _is_continuous_trading() -> bool:
+        """A股连续竞价时段(北京时间): 9:30-11:30 / 13:00-15:00, 仅工作日。
+
+        比 _is_trading_hours 严格: 排除 9:15-9:30 集合竞价(指示价, 非成交价)、
+        午间与 15:00 后收盘缓冲。监控评估只在此窗口进行, 不对竞价/收盘后的陈旧价告警。
+        (节假日由 _evaluate_monitors 里的「快照日期=当日」新鲜度判据兜底, 无需交易日历。)
+        """
+        now = cn_now()
+        t = now.time()
+        morning = dt_time(9, 30) <= t <= dt_time(11, 30)
+        afternoon = dt_time(13, 0) <= t <= dt_time(15, 0)
+        return now.weekday() < 5 and (morning or afternoon)
+
     @staticmethod
     def _save_enabled(enabled: bool) -> None:
         from app.services import preferences
@@ -770,10 +793,21 @@ class QuoteService:
     def _evaluate_monitors(self, daily_df: pl.DataFrame, quote_extra: pl.DataFrame | None) -> None:
         """行情更新后评估统一监控规则引擎,并刷新策略结果缓存。"""
         try:
+            # 仅在「交易日 + 连续竞价时段」评估监控 —— 避开集合竞价指示价、盘前/收盘后
+            # 缓冲。轮询窗口(_is_trading_hours)更宽是为盘前预热/收盘捕捉, 但告警不应
+            # 基于这些非连续竞价价格。
+            if not self._is_continuous_trading():
+                return
             # 获取 enriched 数据 (刚算好的)
             enriched_today, enriched_date = self.get_enriched_today()
             if enriched_today.is_empty():
                 return
+            # 快照日期必须是北京当日: 节假日或数据未刷新时 enriched_date 会落后于当日,
+            # 说明市场未在交易 → 跳过。无需维护 A股交易日历即可挡住节假日与陈旧价告警。
+            if enriched_date != cn_today():
+                logger.debug("监控评估跳过: enriched 快照日期 %s 非当日 %s (节假日/数据未刷新)",
+                             enriched_date, cn_today())
+                return
 
             all_alerts: list[dict] = []
             rule_events: list[dict] = []
@@ -783,22 +817,44 @@ class QuoteService:
             if self._app_state:
                 engine = getattr(self._app_state, "monitor_engine", None)
                 if engine and engine.rule_count > 0:
-                    # 预构建 symbol → name 映射 (enriched 已 drop name 列, 引擎触发时回填用)
+                    # 预构建 symbol → name 映射 (enriched 已 drop name 列, 引擎触发时回填用)。
+                    # 含股票 + ETF 维表, 保证 ETF 监控告警也能回填名称。
                     try:
+                        name_map: dict[str, str] = {}
                         inst_df = self._app_state.repo.get_instruments()
                         if not inst_df.is_empty() and "symbol" in inst_df.columns and "name" in inst_df.columns:
-                            engine.set_name_map({
-                                row["symbol"]: row["name"]
-                                for row in inst_df.select(["symbol", "name"]).iter_rows(named=True)
-                                if row.get("name")
-                            })
+                            for row in inst_df.select(["symbol", "name"]).iter_rows(named=True):
+                                if row.get("name"):
+                                    name_map[row["symbol"]] = row["name"]
+                        # 仅当存在 ETF 规则时补 ETF 维表 (股票名优先, setdefault 不覆盖股票)
+                        if engine.has_asset_rules("etf"):
+                            etf_inst = self._app_state.repo.get_etf_instruments()
+                            if not etf_inst.is_empty() and "symbol" in etf_inst.columns and "name" in etf_inst.columns:
+                                for row in etf_inst.select(["symbol", "name"]).iter_rows(named=True):
+                                    if row.get("name"):
+                                        name_map.setdefault(row["symbol"], row["name"])
+                        if name_map:
+                            engine.set_name_map(name_map)
                     except Exception as e:  # noqa: BLE001
                         logger.debug("name_map 构建失败 (不影响监控): %s", e)
                     # 连板梯队封单监控: 有 ladder 规则时, 从 depth_service 注入封单量到 enriched
                     eval_df = enriched_today
                     if engine.has_rule_type("ladder"):
                         eval_df = self._inject_sealed_vol(enriched_today, enriched_date)
-                    rule_events = engine.evaluate(eval_df)
+                    rule_events = engine.evaluate(eval_df, asset_type="stock")
+                    # ETF 规则轮: 股票快照不含 ETF, 用 ETF enriched 快照单独评估。
+                    # 独立 try —— ETF 轮任何异常都不得丢弃本轮已算出的股票告警。
+                    # refresh=False —— 不在轮询线程上触发 ETF 冷缓存的同步重算 (缓存由 ETF 实时
+                    # flush 焐热; 未焐热说明无 ETF 实时数据, 跳过本轮 ETF 评估)。
+                    if engine.has_asset_rules("etf") and self._repo is not None:
+                        try:
+                            etf_enriched, _ = self._repo.get_enriched_latest_asset("etf", refresh=False)
+                            if not etf_enriched.is_empty():
+                                rule_events = rule_events + engine.evaluate(
+                                    etf_enriched, asset_type="etf", reset_strategy_results=False,
+                                )
+                        except Exception as e:  # noqa: BLE001
+                            logger.warning("ETF 监控评估失败 (不影响股票告警): %s", e)
                     if rule_events:
                         # 落盘到 alerts.jsonl
                         try:
@@ -911,8 +967,7 @@ class QuoteService:
                 "price": "价格", "market": "异动",
             }
             rules = engine.rules if engine is not None else {}
-            pushed_feishu = 0
-            pushed_wecom = 0
+            enqueued = 0
             for ev in rule_events:
                 rule = rules.get(ev.get("rule_id"))
                 if not rule or not rule.get("webhook_enabled"):
@@ -924,16 +979,19 @@ class QuoteService:
                 message = ev.get("message") or ""
                 title = f"TickFlow · {source_label}"
                 body = f"{symbol} {name} {message}".strip() if symbol else (message or name)
-                if feishu_url and webhook_adapter.send_feishu(feishu_url, title, body, feishu_secret):
-                    pushed_feishu += 1
-                if wecom_url and webhook_adapter.send_wecom(wecom_url, title, body):
-                    pushed_wecom += 1
-            if pushed_feishu:
-                logger.info("飞书 Webhook 推送: %d 条", pushed_feishu)
-            if pushed_wecom:
-                logger.info("企业微信 Webhook 推送: %d 条", pushed_wecom)
+                # 提交到独立线程池, 不阻塞行情轮询线程 (webhook 慢/重试不拖累实时行情+告警)。
+                # 飞书 + 企业微信双通道; 应用内 alerts.jsonl 记录与 SSE 已在前面完成, 不依赖
+                # webhook 成败, 失败由 webhook_adapter 记 WARNING(可见)。
+                if feishu_url:
+                    _WEBHOOK_EXECUTOR.submit(webhook_adapter.send_feishu, feishu_url, title, body, feishu_secret)
+                    enqueued += 1
+                if wecom_url:
+                    _WEBHOOK_EXECUTOR.submit(webhook_adapter.send_wecom, wecom_url, title, body)
+                    enqueued += 1
+            if enqueued:
+                logger.info("Webhook 已提交 %d 条 (异步投递, 飞书+企业微信, 失败记 WARNING)", enqueued)
         except Exception as e:  # noqa: BLE001
-            logger.debug("Webhook 推送异常 (不影响告警主流程): %s", e)
+            logger.warning("Webhook 提交异常 (不影响告警主流程): %s", e)
 
     def _maybe_send_system_notifications(self, all_alerts: list[dict]) -> None:
         """把告警转发到操作系统通知中心 (由 preferences 开关控制)。

backend/app/services/screener.py

@@ -19,7 +19,7 @@ from app.tickflow.repository import KlineRepository
 logger = logging.getLogger(__name__)
 
 # ── 进程级历史数据缓存 (避免 run_all 每次重新扫描 parquet + 计算指标) ──
-_history_cache: dict[tuple[date, int], tuple[float, pl.DataFrame]] = {}
+_history_cache: dict[tuple[str, date, int], tuple[float, pl.DataFrame]] = {}
 _HISTORY_CACHE_TTL = 120.0  # 秒
 
 
@@ -36,6 +36,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "momentum_60d",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock", "etf"],
     },
     "ma_golden_cross": {
         "name": "MA 金叉",
@@ -48,6 +49,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "momentum_20d",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock", "etf"],
     },
     "macd_golden": {
         "name": "MACD 金叉放量",
@@ -59,6 +61,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "momentum_60d",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock", "etf"],
     },
     "volume_price_surge": {
         "name": "量价齐升",
@@ -71,6 +74,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "vol_ratio_5d",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock", "etf"],
     },
     "low_volatility_leader": {
         "name": "低波动龙头",
@@ -83,6 +87,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "momentum_60d",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock", "etf"],
     },
     "broken_board_recovery": {
         "name": "断板反包",
@@ -95,6 +100,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "change_pct",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock"],
     },
     "oversold_bounce": {
         "name": "超跌反弹",
@@ -107,6 +113,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "rsi_14",
         "descending": False,
         "limit": 100,
+        "asset_types": ["stock", "etf"],
     },
     "boll_breakout": {
         "name": "布林突破",
@@ -118,6 +125,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "vol_ratio_5d",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock", "etf"],
     },
     "bullish_alignment": {
         "name": "均线多头",
@@ -131,6 +139,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "momentum_60d",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock", "etf"],
     },
     "consecutive_limit_ups": {
         "name": "连板股",
@@ -142,6 +151,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "consecutive_limit_ups",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock"],
     },
     "pullback_to_support": {
         "name": "缩量回踩",
@@ -156,6 +166,7 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "momentum_60d",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock", "etf"],
     },
     "n_day_low_reversal": {
         "name": "新低反转",
@@ -168,10 +179,16 @@ PRESET_STRATEGIES: dict[str, dict] = {
         "order_by": "change_pct",
         "descending": True,
         "limit": 100,
+        "asset_types": ["stock", "etf"],
     },
 }
 
 
+def strategy_supports_asset(strat: dict, asset_type: str) -> bool:
+    """策略是否支持该资产类型。默认仅 stock（未标注 asset_types 的自定义/AI 策略保守视为股票专用）。"""
+    return asset_type in strat.get("asset_types", ["stock"])
+
+
 @dataclass
 class ScreenerResult:
     as_of: date
@@ -182,8 +199,11 @@ class ScreenerResult:
 
 
 class ScreenerService:
-    def __init__(self, repo: KlineRepository) -> None:
+    def __init__(self, repo: KlineRepository, asset_type: str = "stock") -> None:
         self.repo = repo
+        self.asset_type = asset_type
+        from app.tickflow.repository import enriched_dirname
+        self._enriched_dirname = enriched_dirname(asset_type)
 
     @staticmethod
     def clear_history_cache() -> None:
@@ -200,33 +220,34 @@ class ScreenerService:
         对于最新日, 优先使用内存缓存 (已包含完整指标)。
         """
         # 优先使用 repo 最新日缓存
-        cache, cache_date = self.repo.get_enriched_latest()
+        cache, cache_date = self.repo.get_enriched_latest_asset(self.asset_type)
         if cache is not None and not cache.is_empty() and cache_date == target_date:
             df = cache
             # JOIN instruments
-            df_i = self.repo.get_instruments()
+            df_i = self.repo.get_instruments_asset(self.asset_type)
             if not df_i.is_empty():
                 inst_cols = [c for c in ["symbol", "name", "total_shares", "float_shares"] if c in df_i.columns]
                 if "name" not in df.columns:
                     df = df.join(df_i.select(inst_cols), on="symbol", how="left")
             return df
 
-        # 尝试从 repo 级预计算历史缓存中提取目标日期
-        cached_hist = self.repo.get_enriched_history(target_date, 1)
-        if cached_hist is not None and not cached_hist.is_empty() and "date" in cached_hist.columns:
-            df = cached_hist.filter(pl.col("date") == target_date)
-            if not df.is_empty():
-                logger.debug("_load_enriched_for_date: repo history cache for %s", target_date)
-                # JOIN instruments
-                df_i = self.repo.get_instruments()
-                if not df_i.is_empty():
-                    inst_cols = [c for c in ["symbol", "name", "total_shares", "float_shares"] if c in df_i.columns]
-                    if "name" not in df.columns:
-                        df = df.join(df_i.select(inst_cols), on="symbol", how="left")
-                return df
+        # 尝试从 repo 级预计算历史缓存中提取目标日期 (仅 stock: 该缓存为股票专用)
+        if self.asset_type == "stock":
+            cached_hist = self.repo.get_enriched_history(target_date, 1)
+            if cached_hist is not None and not cached_hist.is_empty() and "date" in cached_hist.columns:
+                df = cached_hist.filter(pl.col("date") == target_date)
+                if not df.is_empty():
+                    logger.debug("_load_enriched_for_date: repo history cache for %s", target_date)
+                    # JOIN instruments
+                    df_i = self.repo.get_instruments_asset(self.asset_type)
+                    if not df_i.is_empty():
+                        inst_cols = [c for c in ["symbol", "name", "total_shares", "float_shares"] if c in df_i.columns]
+                        if "name" not in df.columns:
+                            df = df.join(df_i.select(inst_cols), on="symbol", how="left")
+                    return df
 
         # 历史日期: 从 parquet 读取 14 列, 即时计算指标 (慢路径)
-        enriched_dir = self.repo.store.data_dir / "kline_daily_enriched"
+        enriched_dir = self.repo.store.data_dir / self._enriched_dirname
         ds = target_date.isoformat()
         target_parquet = enriched_dir / f"date={ds}" / "part.parquet"
 
@@ -254,7 +275,7 @@ class ScreenerService:
         from app.indicators.pipeline import compute_indicators, compute_signals, compute_limit_signals
 
         # 加载 warmup 历史 (目标日期前 ~120 天)
-        enriched_dir = self.repo.store.data_dir / "kline_daily_enriched"
+        enriched_dir = self.repo.store.data_dir / self._enriched_dirname
         start = target_date - timedelta(days=150)
         read_cols = ["symbol", "date", "open", "high", "low", "close", "volume",
                      "amount", "raw_close", "raw_high", "raw_low"]
@@ -281,9 +302,9 @@ class ScreenerService:
         df_full = compute_indicators(df_hist)
         df_full = compute_signals(df_full)
 
-        # 计算涨跌停信号 (需要 instruments)
-        instruments = self.repo.get_instruments()
-        if instruments is not None and not instruments.is_empty():
+        # 计算涨跌停信号 (需要 instruments; 涨停为股票专有, ETF 跳过)
+        instruments = self.repo.get_instruments_asset(self.asset_type)
+        if self.asset_type == "stock" and instruments is not None and not instruments.is_empty():
             df_full = compute_limit_signals(df_full, instruments)
 
         # 只保留目标日期
@@ -303,23 +324,24 @@ class ScreenerService:
         优先从 repo 内存缓存获取 (启动时已预计算), 命中时 0ms。
         缓存 miss 时走 scan_parquet + compute_indicators 慢路径。
         """
-        # 优先级 1: repo 级预计算缓存 (启动时 _refresh_enriched 已计算完整历史)
+        # 优先级 1: repo 级预计算缓存 (启动时 _refresh_enriched 已计算完整历史; 仅 stock)
         t0 = time.perf_counter()
-        cached = self.repo.get_enriched_history(target_date, lookback_days)
-        if cached is not None and not cached.is_empty():
-            # JOIN instruments (repo 缓存不含 name 等列)
-            instruments = self.repo.get_instruments()
-            if instruments is not None and not instruments.is_empty() and "name" not in cached.columns:
-                inst_cols = [c for c in ["symbol", "name", "total_shares", "float_shares"]
-                             if c in instruments.columns]
-                cached = cached.join(instruments.select(inst_cols), on="symbol", how="left")
-            elapsed = (time.perf_counter() - t0) * 1000
-            logger.info("_load_enriched_history(%s, %d): repo cache hit, %.1fms, %d rows",
-                        target_date, lookback_days, elapsed, len(cached))
-            return cached
+        if self.asset_type == "stock":
+            cached = self.repo.get_enriched_history(target_date, lookback_days)
+            if cached is not None and not cached.is_empty():
+                # JOIN instruments (repo 缓存不含 name 等列)
+                instruments = self.repo.get_instruments_asset(self.asset_type)
+                if instruments is not None and not instruments.is_empty() and "name" not in cached.columns:
+                    inst_cols = [c for c in ["symbol", "name", "total_shares", "float_shares"]
+                                 if c in instruments.columns]
+                    cached = cached.join(instruments.select(inst_cols), on="symbol", how="left")
+                elapsed = (time.perf_counter() - t0) * 1000
+                logger.info("_load_enriched_history(%s, %d): repo cache hit, %.1fms, %d rows",
+                            target_date, lookback_days, elapsed, len(cached))
+                return cached
 
         # 优先级 2: 进程级 history_cache (之前的 TTL 缓存)
-        cache_key = (target_date, lookback_days)
+        cache_key = (self.asset_type, target_date, lookback_days)
         now = time.monotonic()
         ttl_cached = _history_cache.get(cache_key)
         if ttl_cached is not None:
@@ -337,7 +359,7 @@ class ScreenerService:
         warmup = 60
         start = target_date - timedelta(days=min((lookback_days + warmup) * 2, 180))
 
-        enriched_dir = self.repo.store.data_dir / "kline_daily_enriched"
+        enriched_dir = self.repo.store.data_dir / self._enriched_dirname
         read_cols = ["symbol", "date", "open", "high", "low", "close", "volume",
                      "amount", "raw_close", "raw_high", "raw_low"]
 
@@ -359,8 +381,8 @@ class ScreenerService:
         df_full = compute_indicators(df_hist)
         df_full = compute_signals(df_full)
 
-        instruments = self.repo.get_instruments()
-        if instruments is not None and not instruments.is_empty():
+        instruments = self.repo.get_instruments_asset(self.asset_type)
+        if self.asset_type == "stock" and instruments is not None and not instruments.is_empty():
             df_full = compute_limit_signals(df_full, instruments)
 
         if instruments is not None and not instruments.is_empty():
@@ -463,6 +485,10 @@ class ScreenerService:
         if not strat:
             raise ValueError(f"unknown strategy: {strategy_id}")
 
+        # 资产兼容拦截: 该策略不支持当前资产类型时直接返回空 (避免命中 ETF 不存在的列)
+        if not strategy_supports_asset(strat, self.asset_type):
+            return ScreenerResult(as_of=as_of, strategy=strategy_id)
+
         if precomputed is not None and not precomputed.is_empty():
             df = precomputed
         else:
@@ -577,6 +603,9 @@ class ScreenerService:
         return df
 
     def latest_date(self) -> date | None:
+        if self.asset_type != "stock":
+            _, d = self.repo.get_enriched_latest_asset(self.asset_type)
+            return d
         d = self.repo.enriched_latest_date()
         if d:
             return d

backend/app/services/webhook_adapter.py

@@ -63,40 +63,57 @@ def _truncate_card(text: str) -> str:
     return text[:_CARD_MAX_LEN] + ("…" if len(text) > _CARD_MAX_LEN else "")
 
 
+_FEISHU_MAX_ATTEMPTS = 3
+
+
 def _post_feishu(webhook_url: str, payload: dict, secret: str) -> bool:
-    """发送一次飞书 webhook 请求并判定成败 (供 text / card 共用)。
+    """发送飞书 webhook 请求并判定成败 (供 text / card 共用)。
 
-    成功响应: HTTP 200 且业务 code=0 (或非 JSON 的 200)。失败静默返回 False。
+    成功响应: HTTP 200 且业务 code=0 (或非 JSON/非 dict 的 200)。
+
+    瞬时失败 (网络/超时/HTTP 5xx) 会**带退避重试** —— 告警冷却在事件生成时即打戳,
+    一次瞬时 5xx/timeout 若不重试, 该告警会被冷却窗口(默认 1h)压掉, 离屏用户彻底
+    收不到推送。永久失败 (4xx / 业务 code≠0, 如签名错、URL 失效) 不重试。最终失败
+    记 WARNING (而非之前的 debug), 保证「推送丢了」在日志里可见。
     """
-    try:
-        import httpx
+    import httpx
 
-        # 启用签名校验时, 请求体须带 timestamp + sign (秒级时间戳)
-        if secret:
-            timestamp = str(int(time.time()))
-            payload["timestamp"] = timestamp
-            payload["sign"] = _gen_sign(timestamp, secret)
+    last_err = ""
+    for attempt in range(1, _FEISHU_MAX_ATTEMPTS + 1):
+        try:
+            # 启用签名校验时, 请求体须带 timestamp + sign (每次重试都重算, 防时间戳过期)
+            if secret:
+                timestamp = str(int(time.time()))
+                payload["timestamp"] = timestamp
+                payload["sign"] = _gen_sign(timestamp, secret)
 
-        resp = httpx.post(webhook_url, json=payload, timeout=5.0)
-        # 飞书成功响应: {"code":0,"msg":"success"} (或 StatusCode 200 + Extra)
-        if resp.status_code == 200:
-            try:
-                data = resp.json()
-                # code=0 表示飞书业务侧成功; 部分版本无 code 字段则按 msg 判断
+            resp = httpx.post(webhook_url, json=payload, timeout=5.0)
+            if resp.status_code == 200:
+                try:
+                    data = resp.json()
+                except ValueError:
+                    return True  # 非 JSON 的 200, 视为成功
                 if isinstance(data, dict):
                     code = data.get("code", data.get("StatusCode", 0))
                     if code == 0:
                         return True
-                    logger.debug("飞书推送业务失败: %s", data)
+                    # 业务失败(签名错/格式错等): 重试无益, 直接失败
+                    logger.warning("飞书推送业务失败(不重试): %s", data)
                     return False
-            except ValueError:
-                # 非 JSON 响应但 HTTP 200, 视为成功
-                return True
-        logger.debug("飞书推送 HTTP %s: %s", resp.status_code, resp.text[:200])
-        return False
-    except Exception as e:  # noqa: BLE001
-        logger.debug("飞书 Webhook 推送失败: %s", e)
-        return False
+                return True  # 200 且 JSON 非 dict, 视为成功
+            # 4xx 客户端错误(URL 失效等): 不重试; 5xx: 落入重试
+            last_err = f"HTTP {resp.status_code}: {resp.text[:200]}"
+            if resp.status_code < 500:
+                logger.warning("飞书推送失败(不重试, 客户端错误): %s", last_err)
+                return False
+        except Exception as e:  # noqa: BLE001 — 网络/超时, 可重试
+            last_err = str(e)
+
+        if attempt < _FEISHU_MAX_ATTEMPTS:
+            time.sleep(min(2 ** (attempt - 1), 3))  # 退避: 1s, 2s
+
+    logger.warning("飞书 Webhook 推送最终失败(已重试 %d 次): %s", _FEISHU_MAX_ATTEMPTS, last_err)
+    return False
 
 
 def send_feishu(webhook_url: str, title: str, body: str, secret: str = "") -> bool:

backend/app/strategy/monitor.py

@@ -319,6 +319,8 @@ class MonitorRuleEngine:
         # 用于声明 filter_history 的策略 (如反包), 实时监控时拼历史窗口 + 今日行情跑选股。
         # 为 None 时, filter_history 策略仍会被跳过 (保持旧行为, 不破坏无历史场景)。
         self._history_loader: Callable[[_dt.date, int], "pl.DataFrame"] | None = None
+        # ETF 版历史窗口加载器 (asset_type=etf 的规则用)。为 None 时 ETF filter_history 策略跳过。
+        self._history_loader_etf: Callable[[_dt.date, int], "pl.DataFrame"] | None = None
         # 本轮 evaluate() 产出的策略选股结果: strategy_id → {rows, total, as_of}
         # 供策略页实时回显复用 (/api/screener/cached 端点直接读取此内存结果), 避免重跑
         self._latest_strategy_results: dict[str, dict] = {}
@@ -340,6 +342,20 @@ class MonitorRuleEngine:
         """
         self._history_loader = fn
 
+    def set_history_loader_etf(self, fn) -> None:
+        """注入 ETF 版历史窗口加载器 (asset_type=etf 的 strategy 型规则用)。
+
+        签名同 set_history_loader; 复用 ScreenerService(asset_type='etf')._load_enriched_history。
+        为 None 时 ETF filter_history 策略退回到跳过逻辑。
+        """
+        self._history_loader_etf = fn
+
+    def _history_loader_for(self, rule: dict):
+        """按规则的 asset_type 选历史加载器。etf → ETF 加载器, 否则股票加载器。"""
+        if rule.get("asset_type") == "etf":
+            return self._history_loader_etf
+        return self._history_loader
+
     def set_name_map(self, name_map: dict[str, str]) -> None:
         """注入 symbol → 股票名 映射, 用于在告警事件里回填 name 字段。
 
@@ -404,11 +420,27 @@ class MonitorRuleEngine:
         )
 
     # ── 评估 ───────────────────────────────────────────
-    def evaluate(self, df: pl.DataFrame) -> list[dict]:
-        """行情更新后评估所有规则。
+    def has_asset_rules(self, asset_type: str) -> bool:
+        """是否存在指定资产类型的 (已启用) 规则。供 quote_service 判断是否需要 ETF 评估轮。"""
+        if not self._rules:
+            return False
+        return any(
+            r.get("enabled", True) and r.get("asset_type", "stock") == asset_type
+            for r in list(self._rules.values())
+        )
+
+    def evaluate(self, df: pl.DataFrame, asset_type: str = "stock",
+                 reset_strategy_results: bool = True) -> list[dict]:
+        """行情更新后评估规则。
+
+        按 asset_type 只评估匹配资产类型的规则; ETF 规则应传 ETF enriched 快照。
+        股票/ETF 分两轮评估时, 仅股票轮重置 _latest_strategy_results (它供股票策略页
+        /cached 回显; ETF 策略页走实时单跑, 不依赖它)。
 
         Args:
-            df: 实时 enriched 数据 (~5500行, 含 signal_/csg_/指标列)
+            df: 实时 enriched 数据 (含 signal_/csg_/指标列)
+            asset_type: 只评估该资产类型的规则 (默认 stock, 向后兼容)
+            reset_strategy_results: 是否重置策略结果缓存 (多轮评估时仅首轮 True)
         Returns:
             触发的 AlertEvent dict 列表 (含 ts/rule_id/source/type/symbol/...)
         """
@@ -418,11 +450,14 @@ class MonitorRuleEngine:
         now = time.time()
         events: list[dict] = []
         # 每轮重置: 只保留本次 evaluate 产出的策略结果
-        self._latest_strategy_results = {}
+        if reset_strategy_results:
+            self._latest_strategy_results = {}
 
         # list() 快照: 本方法跑在行情轮询线程, API 线程同时 add/remove 规则
         # 会触发 "dictionary changed size during iteration", 整轮告警丢失
         for rule_id, rule in list(self._rules.items()):
+            if rule.get("asset_type", "stock") != asset_type:
+                continue
             try:
                 events.extend(self._evaluate_rule(df, rule, now))
             except Exception as e:
@@ -525,9 +560,13 @@ class MonitorRuleEngine:
                 return df.head(0)
             return df.filter(pl.col("symbol").is_in(syms))
         if scope == "sector":
-            # sector 过滤: 需 df 含板块列 (后续接入 ext_data JOIN)
-            # 当前先返回全量, sector 精确过滤第二步完善
-            return df
+            # sector 过滤需 df 含板块列 (后续接入 ext_data JOIN)。在 JOIN 落地前
+            # fail-closed 返回空 —— 绝不退化为「全市场」误触发 (旧行为 return df 会让
+            # 一条板块规则对全市场每只命中都告警)。新建 sector 规则已在 validate 拦截,
+            # 此处兜底任何历史遗留的 sector 规则。
+            logger.warning("scope=sector 规则 %s 暂不支持(板块 JOIN 未实现), 本轮跳过",
+                           rule.get("id"))
+            return df.head(0)
         return df
 
     def _match_strategy(
@@ -544,6 +583,8 @@ class MonitorRuleEngine:
         sid = rule.get("strategy_id")
         if not sid:
             return []
+        at = rule.get("asset_type", "stock")
+        pool_key = (sid, at)
         try:
             s = self._strategy_engine.get(sid)
         except Exception:
@@ -568,13 +609,15 @@ class MonitorRuleEngine:
             "overrides": overrides,
         }
         if s.filter_history_fn:
-            if self._history_loader is None:
-                logger.debug("策略 %s 需要历史数据但未注入 history_loader, 跳过实时监控", sid)
+            history_loader = self._history_loader_for(rule)
+            if history_loader is None:
+                logger.debug("策略 %s 需要历史数据但未注入 history_loader (asset_type=%s), 跳过实时监控",
+                             sid, rule.get("asset_type", "stock"))
                 return []
             try:
                 today = cn_today()
                 lookback = max(1, getattr(s, "lookback_days", 30))
-                hist_df = self._history_loader(today, lookback)
+                hist_df = history_loader(today, lookback)
                 if hist_df is None or hist_df.is_empty():
                     logger.debug("策略 %s 历史数据为空, 跳过本轮实时监控", sid)
                     return []
@@ -602,26 +645,28 @@ class MonitorRuleEngine:
 
         # 记录本轮完整选股结果 (供策略页实时回显: /cached 端点直接读取, 不落盘)。
         # 与下面的 diff 事件无关 — 无论是否产生 new_entry/dropped, 结果都该可用于回显。
-        try:
-            import math
-            self._latest_strategy_results[sid] = {
-                "total": result.total,
-                "as_of": str(cn_today()),
-                "rows": [
-                    {k: (None if isinstance(v, float) and not math.isfinite(v) else v)
-                     for k, v in row.items()}
-                    for row in result.rows
-                ],
-            }
-        except Exception:  # noqa: BLE001
-            pass
+        # 策略结果缓存仅用于股票策略页 /cached 回显; ETF 策略页走实时单跑, 不写入。
+        if at == "stock":
+            try:
+                import math
+                self._latest_strategy_results[sid] = {
+                    "total": result.total,
+                    "as_of": str(cn_today()),
+                    "rows": [
+                        {k: (None if isinstance(v, float) and not math.isfinite(v) else v)
+                         for k, v in row.items()}
+                        for row in result.rows
+                    ],
+                }
+            except Exception:  # noqa: BLE001
+                pass
 
         current_pool: set[str] = {r["symbol"] for r in result.rows}
-        prev_pool = self._strategy_pools.get(sid)
+        prev_pool = self._strategy_pools.get(pool_key)
 
         # 首次运行: 仅记录当前选股池, 不产生事件
         if prev_pool is None:
-            self._strategy_pools[sid] = current_pool
+            self._strategy_pools[pool_key] = current_pool
             return []
 
         new_entries = current_pool - prev_pool
@@ -632,7 +677,7 @@ class MonitorRuleEngine:
             return []
 
         # 更新存储
-        self._strategy_pools[sid] = current_pool
+        self._strategy_pools[pool_key] = current_pool
 
         sname = s.meta.get("name", "") or s.meta.get("id", sid)
 

backend/app/strategy/monitor_rules.py

@@ -154,6 +154,11 @@ def validate(rule: dict) -> None:
         syms = rule.get("symbols")
         if not isinstance(syms, list) or len(syms) == 0:
             raise ValueError("scope=symbols 时 symbols 不能为空")
+    # sector 作用域的板块 JOIN 尚未实现: _apply_scope 目前会退化为「全市场」,
+    # 一条本意针对某板块的规则会对全市场每只命中都触发(告警风暴)。在板块 JOIN
+    # 落地前, 拒绝创建 sector 规则(fail-closed), 避免用户建出会刷屏的规则。
+    if rule.get("scope") == "sector":
+        raise ValueError("scope=sector 暂未支持(板块 JOIN 未实现),请改用 scope=symbols 指定标的或 scope=all")
 
     # 其余枚举
     if rule.get("severity", "info") not in SEVERITIES:
@@ -167,6 +172,7 @@ def normalize(rule: dict) -> dict:
     """补全默认字段,返回规范化后的规则 (不校验)。"""
     r = dict(rule)
     r.setdefault("enabled", True)
+    r.setdefault("asset_type", "stock")
     r.setdefault("scope", "symbols")
     r.setdefault("symbols", [])
     r.setdefault("sector", None)

backend/app/tickflow/policy.py

@@ -106,7 +106,7 @@ def _call_with_retry(fn, attempts: int = 3, backoff: float = 0.6) -> None:
     raise last_exc
 
 
-def _probe_real(tiers: dict) -> tuple[CapabilitySet, list[str]]:
+def _probe_real(tiers: dict) -> tuple[CapabilitySet, list[str], set[Cap]]:
     """逐 capability 试探。需要 API key。
 
     **关键**:探测始终在付费端点(api.tickflow.org)上进行,用 key 鉴权验证有效性。
@@ -126,6 +126,9 @@ def _probe_real(tiers: dict) -> tuple[CapabilitySet, list[str]]:
     tf = TickFlow(api_key=key, base_url=probe_base)
     available: dict[Cap, CapabilityLimits] = {}
     log: list[str] = []
+    # 重试耗尽仍失败的瞬时错误(非明确无权限)对应的 cap。供上层判定: 若「分水岭」
+    # cap(单只日K/复权因子)是瞬时失败, 不要据此把付费用户降级为 free/none。
+    transient_failed: set[Cap] = set()
 
     def try_call(cap: Cap, fn, default_limits: dict[str, Any]) -> None:
         try:
@@ -148,9 +151,16 @@ def _probe_real(tiers: dict) -> tuple[CapabilitySet, list[str]]:
             )
             if is_perm_denied:
                 log.append(f"✗ {cap}(无权限)")
+            elif _is_transient(e):
+                # 仅**真瞬时**错误(超时/连接/5xx/429, 由 _is_transient 判定)才标记为疑似 —
+                # 与探测重试用同一判据。否则一个消息未命中权限关键词的确定性失败
+                # (如 401/"authentication failed"/"key expired")会被误当瞬时, 让降级
+                # 保护(保留旧付费档)反而掩盖真实的 Key 失效, 永不回落到 free-api。
+                transient_failed.add(cap)
+                log.append(f"? {cap} (瞬时: {cls}: {e})")
             else:
-                # 重试耗尽仍失败的瞬时错误 — 标记为疑似,而非直接判定"无此能力"
-                log.append(f"? {cap} ({cls}: {e})")
+                # 非权限关键词、也非瞬时 → 视为该能力确实不可用(不保留、不重试保护)
+                log.append(f"✗ {cap}({cls}: {e})")
 
     # 用各档默认上限作为占位(无 X-RateLimit-* 头时)
     # 取所有档的并集,逐 cap 试探
@@ -243,7 +253,25 @@ def _probe_real(tiers: dict) -> tuple[CapabilitySet, list[str]]:
         )
         log.append("✓ websocket (inferred from expert tier)")
 
-    return CapabilitySet(available), log
+    return CapabilitySet(available), log, transient_failed
+
+
+def _load_cached_capset(cache_path: Path) -> CapabilitySet | None:
+    """读取上次持久化的 capset(schema 匹配时)。供瞬时失败时保留旧档位用。
+
+    此时尚未 _persist 本次探测结果, 缓存文件仍是上一次的值。schema 不匹配则返回 None
+    (旧结构不可靠, 不作为保留依据)。
+    """
+    try:
+        if not cache_path.exists():
+            return None
+        with cache_path.open(encoding="utf-8") as f:
+            cached = json.load(f)
+        if cached.get("schema_version") != _CACHE_SCHEMA_VERSION:
+            return None
+        return _capset_from_json(cached)
+    except Exception:  # noqa: BLE001
+        return None
 
 
 def detect_capabilities(force: bool = False) -> CapabilitySet:
@@ -267,9 +295,26 @@ def detect_capabilities(force: bool = False) -> CapabilitySet:
 
     # 有 API key — 真实探测
     try:
-        capset, probe_log = _probe_real(tiers)
+        capset, probe_log, transient_failed = _probe_real(tiers)
         # 判定档位:无效 key → none,免费 key → free,付费 → starter/pro/expert
         classified = _classify_tier(capset, tiers)
+
+        # 瞬时探测失败不得触发降级: 分水岭 cap(单只日K / 复权因子)本次是瞬时失败
+        # (非明确无权限), 且此前缓存过付费档(有复权因子)时, 保留旧缓存档位、不持久化
+        # 降级。否则一次网络抖动就把付费用户误降为 free/none, 直到强制重探才恢复。
+        prev_capset = _load_cached_capset(cache_path)
+        prev_was_paid = prev_capset is not None and Cap.ADJ_FACTOR in prev_capset.all()
+        transient_downgrade = (
+            (classified.is_invalid and Cap.KLINE_DAILY_BY_SYMBOL in transient_failed)
+            or (classified.is_free and Cap.ADJ_FACTOR in transient_failed)
+        )
+        if prev_was_paid and transient_downgrade:
+            logger.warning(
+                "能力探测分水岭瞬时失败(非无权限): %s; 保留上次缓存档位, 不降级",
+                sorted(str(c) for c in transient_failed),
+            )
+            return prev_capset
+
         if classified.is_invalid:
             # 无效 key(连单只日K都拿不到):归 none 档,标记要求清除 key
             capset = _tier_to_capset(tiers["none"])

backend/app/tickflow/repository.py

@@ -28,6 +28,11 @@ from app.config import settings
 logger = logging.getLogger(__name__)
 
 
+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"
+
+
 class DataStore:
     """唯一的存储入口 — 进程启动时创建。"""
 
@@ -906,12 +911,17 @@ class KlineRepository:
             return pl.DataFrame(), self._enriched_cache_date
         return self._enriched_cache, self._enriched_cache_date
 
-    def get_enriched_latest_asset(self, asset_type: str) -> tuple[pl.DataFrame, date | None]:
-        """按资产类型返回最新 enriched 缓存。stock 保持旧缓存语义。"""
+    def get_enriched_latest_asset(self, asset_type: str, refresh: bool = True) -> tuple[pl.DataFrame, date | None]:
+        """按资产类型返回最新 enriched 缓存。stock 保持旧缓存语义。
+
+        refresh=False: 缓存冷时不触发同步 _refresh_etf_enriched(300 天 scan+compute)。
+        供行情轮询线程使用 —— 避免在热路径上做重活阻塞股票行情/告警;缓存由 ETF 实时
+        flush 焐热, 未焐热(无 ETF 实时数据)时返回空表, 本轮跳过 ETF 评估即可。
+        """
         if asset_type == "stock":
             return self.get_enriched_latest()
         if asset_type == "etf":
-            if self._etf_enriched_cache is None:
+            if self._etf_enriched_cache is None and refresh:
                 self._refresh_etf_enriched()
             if self._etf_enriched_cache is None:
                 return pl.DataFrame(), self._etf_enriched_cache_date
@@ -1478,6 +1488,25 @@ class KlineRepository:
             return None
         return None
 
+    def symbols_lagging(self, reference_date: date, min_gap_days: int = 3) -> list[str]:
+        """返回日K覆盖落后的标的: 其最新 bar 早于 reference_date - min_gap_days。
+
+        全局 max(date) 只要有一只票有今日数据就成立, 会掩盖停牌/复牌/一直拉失败而
+        掉队的个股缺口。此方法按 symbol 聚合最新日期, 找出掉队者。只读, 不改数据。
+        """
+        from datetime import timedelta
+        try:
+            cutoff = reference_date - timedelta(days=min_gap_days)
+            with self._lock:
+                rows = self.db.execute(
+                    "SELECT symbol, max(date) AS mx FROM kline_daily "
+                    "GROUP BY symbol HAVING max(date) < ? ORDER BY mx",
+                    [cutoff],
+                ).fetchall()
+            return [r[0] for r in rows if r and r[0]]
+        except Exception:
+            return []
+
     def _latest_enriched_date_duckdb(self) -> date | None:
         try:
             with self._lock:

backend/tests/backtest/test_engine_portfolio.py

@@ -219,7 +219,8 @@ def test_trailing_take_profit_exits_after_activation():
     assert len(result.trades) == 1
     trade = result.trades[0]
     assert trade.exit_reason == "trailing_take_profit"
-    assert trade.exit_price == 11.7
+    # 纯峰值口径 (跟随 upstream): 触发线 = 峰值价 × (1 - 回撤%) = 12 × 0.97 = 11.64
+    assert trade.exit_price == 11.64
 
 
 def test_score_filter_uses_signal_day_score_range():

backend/tests/backtest/test_strategy_backtest_correctness.py

@@ -53,8 +53,9 @@ class _EngineStub:
         self.sim_panel: pl.DataFrame | None = None
         self.sim_entries: pl.Series | None = None
 
-    def load_panel(self, symbols, start: date, end: date) -> pl.DataFrame:
+    def load_panel(self, symbols, start: date, end: date, columns=None, asset_type: str = "stock") -> pl.DataFrame:
         self.load_args = (symbols, start, end)
+        self.load_asset_type = asset_type
         return self.panel
 
     def simulate_portfolio(self, panel, entries, exits, config, progress_cb=None, cancel_event=None) -> SimResult:
@@ -135,7 +136,7 @@ def test_full_mode_executes_every_candidate_with_strategy_rules():
     ]).sort(["symbol", "date"])
 
     engine = BacktestEngine(repo=None)  # type: ignore[arg-type]
-    engine.load_panel = lambda symbols, s, e: panel  # type: ignore[method-assign]
+    engine.load_panel = lambda symbols, s, e, columns=None, asset_type="stock": panel  # type: ignore[method-assign]
     strategy = _strategy(
         filter_fn=lambda df, params: pl.col("date") == start,
         max_hold_days=1,

backend/tests/test_backtest_etf.py

@@ -0,0 +1,91 @@
+import types
+from datetime import date
+
+import polars as pl
+
+from app.services.backtest import BacktestConfig
+from app.backtest.engine import BacktestEngine, PanelCache
+from app.backtest.factor import FactorConfig
+from app.backtest.strategy import StrategyBacktestConfig
+
+
+def test_configs_default_to_stock():
+    assert BacktestConfig(symbols=[], start=date(2026, 1, 1), end=date(2026, 1, 2)).asset_type == "stock"
+    assert FactorConfig(factor_name="x", symbols=None, start=date(2026, 1, 1), end=date(2026, 1, 2)).asset_type == "stock"
+    assert StrategyBacktestConfig(strategy_id="x", symbols=None, start=date(2026, 1, 1), end=date(2026, 1, 2)).asset_type == "stock"
+
+
+def test_panel_cache_key_isolates_asset_type():
+    args = (["510300"], date(2026, 1, 1), date(2026, 1, 2), None)
+    k_stock = PanelCache._make_key(*args, "stock")
+    k_etf = PanelCache._make_key(*args, "etf")
+    assert k_stock != k_etf
+    assert k_etf.startswith("etf:")
+    assert k_stock.startswith("stock:")
+
+
+def test_engine_loads_from_etf_dir(monkeypatch, tmp_path):
+    """asset_type='etf' 时, load_panel 应扫 ETF enriched 目录, 不走 stock 缓存。"""
+    captured = {}
+
+    def fake_scan(path, *a, **k):
+        captured["path"] = str(path)
+        return pl.LazyFrame({
+            "symbol": pl.Series("symbol", [], dtype=pl.Utf8),
+            "date": pl.Series("date", [], dtype=pl.Date),
+            "open": pl.Series("open", [], dtype=pl.Float64),
+            "high": pl.Series("high", [], dtype=pl.Float64),
+            "low": pl.Series("low", [], dtype=pl.Float64),
+            "close": pl.Series("close", [], dtype=pl.Float64),
+            "volume": pl.Series("volume", [], dtype=pl.Float64),
+        })
+
+    monkeypatch.setattr("app.backtest.engine.pl.scan_parquet", fake_scan)
+
+    # get_enriched_range 返回 None: 即便被调也不命中缓存; etf 分支本就不该调它
+    repo = types.SimpleNamespace(
+        store=types.SimpleNamespace(data_dir=tmp_path),
+        get_enriched_range=lambda *a, **k: None,
+    )
+    eng = BacktestEngine(repo)
+    eng._load_panel_inner(["510300"], date(2026, 1, 1), date(2026, 1, 2), None, "etf")
+    assert "kline_etf_enriched" in captured["path"]
+
+
+def test_engine_stock_uses_daily_enriched_dir(monkeypatch, tmp_path):
+    captured = {}
+
+    def fake_scan(path, *a, **k):
+        captured["path"] = str(path)
+        return pl.LazyFrame({
+            "symbol": pl.Series("symbol", [], dtype=pl.Utf8),
+            "date": pl.Series("date", [], dtype=pl.Date),
+            "open": pl.Series("open", [], dtype=pl.Float64),
+            "high": pl.Series("high", [], dtype=pl.Float64),
+            "low": pl.Series("low", [], dtype=pl.Float64),
+            "close": pl.Series("close", [], dtype=pl.Float64),
+            "volume": pl.Series("volume", [], dtype=pl.Float64),
+        })
+
+    monkeypatch.setattr("app.backtest.engine.pl.scan_parquet", fake_scan)
+    repo = types.SimpleNamespace(
+        store=types.SimpleNamespace(data_dir=tmp_path),
+        get_enriched_range=lambda *a, **k: None,
+    )
+    eng = BacktestEngine(repo)
+    eng._load_panel_inner(["600519"], date(2026, 1, 1), date(2026, 1, 2), None, "stock")
+    assert "kline_daily_enriched" in captured["path"]
+
+
+def test_job_key_includes_asset_type_and_is_consistent():
+    """stream 与 cancel 必须用同一 job_key: asset_type 进 key 且相同入参产出相同 key。"""
+    from app.api.backtest import _make_job_key
+
+    args = ("s1", None, None, None, "open_t+1", None, None,
+            0.0002, 5.0, 10, 1.0, 1_000_000.0, "equal", None, None,
+            "position", 5, None, None)
+    k_stock = _make_job_key(*args, asset_type="stock")
+    k_etf = _make_job_key(*args, asset_type="etf")
+    assert k_stock != k_etf
+    # 相同参数(含 asset_type)必须产出相同 key —— stream 端与 cancel 端对齐的前提
+    assert _make_job_key(*args, asset_type="etf") == k_etf

backend/tests/test_monitor_etf.py

@@ -0,0 +1,86 @@
+from app.strategy.monitor import MonitorRuleEngine
+from app.strategy import monitor_rules
+
+
+def test_history_loader_selection_by_asset_type():
+    eng = MonitorRuleEngine()
+
+    def stock_loader(d, l):
+        return "STOCK"
+
+    def etf_loader(d, l):
+        return "ETF"
+
+    eng.set_history_loader(stock_loader)
+    eng.set_history_loader_etf(etf_loader)
+
+    assert eng._history_loader_for({"asset_type": "etf"}) is etf_loader
+    assert eng._history_loader_for({"asset_type": "stock"}) is stock_loader
+    # 未标注 asset_type 的旧规则默认走股票加载器
+    assert eng._history_loader_for({}) is stock_loader
+
+
+def test_etf_loader_defaults_none():
+    eng = MonitorRuleEngine()
+    assert eng._history_loader_for({"asset_type": "etf"}) is None
+
+
+def test_rule_model_defaults_stock():
+    from app.api.monitor_rules import RuleModel
+
+    r = RuleModel(id="x", name="n", type="price")
+    assert r.asset_type == "stock"
+
+
+def test_normalize_preserves_and_defaults_asset_type():
+    assert monitor_rules.normalize({"id": "a", "type": "price"})["asset_type"] == "stock"
+    assert monitor_rules.normalize({"id": "a", "type": "signal", "asset_type": "etf"})["asset_type"] == "etf"
+
+
+def _signal_rule(rid, asset_type, sym):
+    return {
+        "id": rid, "name": rid, "type": "signal", "asset_type": asset_type,
+        "scope": "symbols", "symbols": [sym], "logic": "and",
+        "conditions": [{"field": "rsi_14", "op": "<", "value": 100}],
+        "cooldown_seconds": 0, "enabled": True,
+    }
+
+
+def _etf_df():
+    import polars as pl
+    return pl.DataFrame({
+        "symbol": ["510300"],
+        "close": [4.0],
+        "change_pct": [0.01],
+        "rsi_14": [40.0],
+    })
+
+
+def test_evaluate_asset_type_filters_rules():
+    """evaluate(asset_type=etf) 只评估 ETF 规则; 股票规则被过滤。"""
+    eng = MonitorRuleEngine()
+    eng.set_rules([_signal_rule("r_etf", "etf", "510300"),
+                   _signal_rule("r_stock", "stock", "510300")])
+    df = _etf_df()
+
+    etf_events = eng.evaluate(df, asset_type="etf")
+    assert any(e["rule_id"] == "r_etf" for e in etf_events)
+    assert all(e["rule_id"] != "r_stock" for e in etf_events)
+
+    stock_events = eng.evaluate(df, asset_type="stock", reset_strategy_results=False)
+    assert all(e["rule_id"] != "r_etf" for e in stock_events)
+
+
+def test_has_asset_rules():
+    eng = MonitorRuleEngine()
+    eng.set_rules([_signal_rule("r_etf", "etf", "510300")])
+    assert eng.has_asset_rules("etf") is True
+    assert eng.has_asset_rules("stock") is False
+
+
+def test_evaluate_default_asset_type_is_stock():
+    """不传 asset_type 时默认只评估股票规则 (向后兼容旧调用)。"""
+    eng = MonitorRuleEngine()
+    eng.set_rules([_signal_rule("r_etf", "etf", "510300")])
+    # 默认 asset_type=stock → ETF 规则不评估
+    assert eng.evaluate(_etf_df()) == []

backend/tests/test_pipeline_and_monitor_fixes.py

@@ -0,0 +1,99 @@
+"""回归测试: 本轮修复的几处高风险行为(并发单飞 / 重任务槽 / sector fail-closed)。
+
+均为纯逻辑, 不触网, 不依赖真实数据源。
+"""
+from __future__ import annotations
+
+import polars as pl
+import pytest
+
+from app.services import pipeline_jobs
+from app.services.pipeline_jobs import JobStore
+from app.strategy import monitor_rules
+from app.strategy.monitor import MonitorRuleEngine
+
+
+# ── JobStore 单飞 ────────────────────────────────────────────────────────
+
+def test_create_singleflight_dedupes_pending_window(tmp_path):
+    """两次快速 create() 在 pending 窗口内应复用同一 job(is_new=False)。"""
+    store = JobStore(store_dir=tmp_path / "jobs")
+
+    jid1, new1 = store.create()
+    assert new1 is True
+
+    # 尚未 start(), job 仍是 pending —— 旧实现会在此另起新 job(并发双跑根因)
+    jid2, new2 = store.create()
+    assert jid2 == jid1
+    assert new2 is False
+
+    # start() 后仍复用同一活跃 job
+    store.start(jid1)
+    jid3, new3 = store.create()
+    assert jid3 == jid1
+    assert new3 is False
+
+
+def test_create_new_after_terminal(tmp_path):
+    """job 终态(succeed/fail)后, create() 应给出新 job。"""
+    store = JobStore(store_dir=tmp_path / "jobs")
+    jid1, _ = store.create()
+    store.start(jid1)
+    store.succeed(jid1, {"ok": True})
+
+    jid2, new2 = store.create()
+    assert jid2 != jid1
+    assert new2 is True
+
+
+def test_run_slot_is_exclusive():
+    """重任务执行槽同一时刻只允许一个持有者(防僵尸并发)。"""
+    assert pipeline_jobs.try_acquire_run_slot() is True
+    try:
+        # 已被占用, 第二次获取失败
+        assert pipeline_jobs.try_acquire_run_slot() is False
+    finally:
+        pipeline_jobs.release_run_slot()
+    # 释放后可再次获取
+    assert pipeline_jobs.try_acquire_run_slot() is True
+    pipeline_jobs.release_run_slot()
+    # 重复释放幂等, 不抛
+    pipeline_jobs.release_run_slot()
+
+
+# ── 监控 sector fail-closed ──────────────────────────────────────────────
+
+def _base_price_rule(scope: str) -> dict:
+    return {
+        "id": "r_test",
+        "name": "t",
+        "type": "price",
+        "conditions": [{"field": "close", "op": ">", "value": 10}],
+        "logic": "and",
+        "scope": scope,
+    }
+
+
+def test_validate_rejects_sector_scope():
+    with pytest.raises(ValueError):
+        monitor_rules.validate(_base_price_rule("sector"))
+
+
+def test_validate_accepts_symbols_scope():
+    rule = _base_price_rule("symbols")
+    rule["symbols"] = ["600000.SH"]
+    monitor_rules.validate(rule)  # 不应抛
+
+
+def test_apply_scope_sector_fails_closed():
+    """历史遗留 sector 规则在评估时应返回空(绝不退化为全市场)。"""
+    df = pl.DataFrame({"symbol": ["600000.SH", "000001.SZ"], "close": [10.0, 20.0]})
+    out = MonitorRuleEngine._apply_scope(df, {"id": "r_old", "scope": "sector"})
+    assert out.is_empty()
+
+    # 对照: scope=all 返回全量, symbols 过滤子集
+    assert MonitorRuleEngine._apply_scope(df, {"scope": "all"}).height == 2
+    picked = MonitorRuleEngine._apply_scope(
+        df, {"scope": "symbols", "symbols": ["600000.SH"]}
+    )
+    assert picked.height == 1

backend/tests/test_screener_etf.py

@@ -0,0 +1,119 @@
+from app.services.screener import (
+    PRESET_STRATEGIES,
+    strategy_supports_asset,
+)
+
+
+def test_all_presets_have_asset_types():
+    for sid, strat in PRESET_STRATEGIES.items():
+        assert "asset_types" in strat, f"{sid} 缺 asset_types"
+        assert "stock" in strat["asset_types"], f"{sid} 必须支持 stock"
+
+
+def test_limit_up_strategies_are_stock_only():
+    for sid in ("broken_board_recovery", "consecutive_limit_ups"):
+        assert PRESET_STRATEGIES[sid]["asset_types"] == ["stock"]
+
+
+def test_pure_technical_strategies_support_etf():
+    for sid in (
+        "trend_breakout", "ma_golden_cross", "macd_golden",
+        "volume_price_surge", "low_volatility_leader", "oversold_bounce",
+        "boll_breakout", "bullish_alignment", "pullback_to_support",
+        "n_day_low_reversal",
+    ):
+        assert "etf" in PRESET_STRATEGIES[sid]["asset_types"], sid
+
+
+def test_strategy_supports_asset_defaults_to_stock():
+    assert strategy_supports_asset({}, "stock") is True
+    assert strategy_supports_asset({}, "etf") is False
+    assert strategy_supports_asset({"asset_types": ["stock", "etf"]}, "etf") is True
+
+
+import types
+from datetime import date
+
+import polars as pl
+
+from app.services.screener import ScreenerService
+
+
+class _FakeRepo:
+    """最小 repo 桩：只实现 screener 用到的 _asset 取数接口。"""
+
+    def __init__(self, data_dir, enriched=None, instruments=None, latest=None):
+        self.store = types.SimpleNamespace(data_dir=data_dir)
+        self._enriched = enriched if enriched is not None else pl.DataFrame()
+        self._instruments = instruments if instruments is not None else pl.DataFrame()
+        self._latest = latest
+
+    def get_enriched_latest_asset(self, asset_type):
+        return self._enriched, self._latest
+
+    def get_instruments_asset(self, asset_type):
+        return self._instruments
+
+    def get_enriched_history(self, target_date, lookback_days):
+        return None  # stock 缓存；ETF 分支不应调用它
+
+
+def test_service_defaults_to_stock_dir(tmp_path):
+    svc = ScreenerService(_FakeRepo(tmp_path))
+    assert svc.asset_type == "stock"
+    assert svc._enriched_dirname == "kline_daily_enriched"
+
+
+def test_service_etf_uses_etf_dir(tmp_path):
+    svc = ScreenerService(_FakeRepo(tmp_path), asset_type="etf")
+    assert svc.asset_type == "etf"
+    assert svc._enriched_dirname == "kline_etf_enriched"
+
+
+def test_etf_run_preset_empty_data_degrades(tmp_path):
+    """ETF enriched 为空时，run_preset 返回空结果而非抛错。"""
+    svc = ScreenerService(_FakeRepo(tmp_path), asset_type="etf")
+    result = svc.run_preset("trend_breakout", as_of=date(2026, 1, 2))
+    assert result.total == 0
+    assert result.rows == []
+
+
+def test_etf_run_preset_filters_rows(tmp_path):
+    """给一份含技术列的 ETF enriched，趋势突破策略能选出命中行。"""
+    enriched = pl.DataFrame({
+        "symbol": ["510300", "159915"],
+        "name": ["沪深300ETF", "创业板ETF"],
+        "date": [date(2026, 1, 2), date(2026, 1, 2)],
+        "close": [4.0, 2.0],
+        "open": [3.9, 2.1],
+        "ma60": [3.5, 2.5],
+        "signal_n_day_high": [True, False],
+        "vol_ratio_5d": [2.5, 0.5],
+        "momentum_60d": [0.2, -0.1],
+    })
+    repo = _FakeRepo(tmp_path, enriched=enriched, latest=date(2026, 1, 2))
+    svc = ScreenerService(repo, asset_type="etf")
+    result = svc.run_preset("trend_breakout", as_of=date(2026, 1, 2))
+    assert result.total == 1
+    assert result.rows[0]["symbol"] == "510300"
+
+
+def test_strategies_filtered_for_etf():
+    etf_ids = [sid for sid, s in PRESET_STRATEGIES.items()
+               if strategy_supports_asset(s, "etf")]
+    assert "trend_breakout" in etf_ids
+    assert "consecutive_limit_ups" not in etf_ids
+    assert len(etf_ids) == 10
+
+
+def test_run_preset_stock_only_strategy_on_etf_returns_empty(tmp_path):
+    """对 ETF 跑股票专有策略（连板）应返回空结果，而非误命中或抛错。"""
+    enriched = pl.DataFrame({
+        "symbol": ["510300"],
+        "date": [date(2026, 1, 2)],
+        "close": [4.0],
+    })
+    repo = _FakeRepo(tmp_path, enriched=enriched, latest=date(2026, 1, 2))
+    svc = ScreenerService(repo, asset_type="etf")
+    result = svc.run_preset("consecutive_limit_ups", as_of=date(2026, 1, 2))
+    assert result.total == 0

docs/features.md

@@ -18,6 +18,8 @@
 
 全 A 股一次扫表,Polars 毫秒级返回。选股页点策略卡片即可扫描,结果支持导出。
 
+**ETF 支持**:选股页顶部可切换 `股票 / ETF`。ETF 复用已算好的 `kline_etf_enriched` 技术指标,仅开放**技术类内置策略**(趋势/量价/反转/波动);依赖涨停信号的策略(连板股、断板反包)为股票专有,ETF 模式下不显示。需先在数据页开启 ETF 拉取(`pipeline_pull_etf`)并跑一次盘后管道。
+
 扩展策略的三种方式见 [strategy.md → 扩展策略](./strategy.md#扩展策略的三种方式)。
 
 ---
@@ -52,6 +54,8 @@
 
 输出净值曲线 · 夏普 · 最大回撤 · 胜率 · 交易明细。SSE 流式进度支持切页重连,不会丢失回测任务。
 
+**ETF 支持**:三种模式的后端与 API 均支持 `asset_type=etf`,回测面板改从 `kline_etf_enriched` 读取(单次回测为单一资产类型,不混合股票与 ETF)。策略组合与因子回测页均有 `股票 / ETF` 切换,ETF 模式下策略列表与标的搜索跟随资产。需先开启 ETF 拉取并跑盘后管道。
+
 ---
 
 ## 📡 监控中心(Monitor)
@@ -65,6 +69,8 @@
 | 价格涨跌监控 | 涨跌幅 / 价格突破阈值 |
 | 全市场异动 | 全市场异动(如快速拉升/跌停) |
 
+**ETF 支持**:规则可选资产类型 `股票 / ETF`。监控引擎按规则 `asset_type` 分轮评估——ETF 规则用 ETF enriched 快照评估(`engine.evaluate(..., asset_type="etf")`),策略型规则走 ETF 历史加载器(读 `kline_etf_enriched`)。盘中触发需开启 ETF 实时行情(`realtime_pull_etf`),使 ETF 报价进入 enriched 快照。
+
 **特性:**
 
 - 多条件 AND/OR + 冷却期去重 + 严重级别(info / warn / critical)

frontend/src/components/monitor/RuleEditor.tsx

@@ -27,6 +27,7 @@ const emptyRule = (preset?: Partial<MonitorRule>): MonitorRule => ({
   name: '',
   enabled: true,
   type: 'signal',
+  asset_type: 'stock',
   scope: 'symbols',
   symbols: [],
   sector: null,
@@ -43,7 +44,6 @@ const emptyRule = (preset?: Partial<MonitorRule>): MonitorRule => ({
 export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
   const qc = useQueryClient()
   const options = useQuery({ queryKey: QK.monitorRuleOptions, queryFn: api.monitorRuleOptions })
-  const strategies = useQuery({ queryKey: QK.screenerStrategies, queryFn: api.screenerStrategies })
   const { data: prefs } = usePreferences()
   const feishuConfigured = !!(prefs?.feishu_webhook_url)
   const [editing] = useState(!!rule)
@@ -54,11 +54,19 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
       ? { ...rule, conditions: rule.conditions.map(c => ({ ...c })) }
       : { ...emptyRule(preset), webhook_enabled: preset?.webhook_enabled ?? !!(prefs?.webhook_enabled_default) },
   )
+  const assetType = draft.asset_type ?? 'stock'
+  // 策略列表跟随资产类型: ETF 只列技术类策略。
+  const strategies = useQuery({
+    queryKey: QK.screenerStrategies(assetType),
+    queryFn: () => api.screenerStrategies(assetType),
+  })
   const [error, setError] = useState('')
   const [symbolQuery, setSymbolQuery] = useState('')
+  // ETF 规则时标的搜索一并搜出 ETF。
+  const symbolAssetTypes = assetType === 'etf' ? 'stock,etf' : 'stock'
   const symbolSearch = useQuery({
-    queryKey: QK.instrumentSearch(symbolQuery),
-    queryFn: () => api.instrumentSearch(symbolQuery, 20),
+    queryKey: QK.instrumentSearch(symbolQuery, symbolAssetTypes),
+    queryFn: () => api.instrumentSearch(symbolQuery, 20, symbolAssetTypes),
     enabled: symbolQuery.length > 0,
   })
 
@@ -210,6 +218,26 @@ export function RuleEditor({ rule, preset, simple, onClose, onSaved }: Props) {
         </button>
       </div>
 
+      {/* 资产类型: 股票 / ETF (个股极简模式不显示) */}
+      {!simple && (
+        <div className="space-y-1.5">
+          <span className="text-[11px] text-muted">资产类型</span>
+          <div className="inline-flex h-9 rounded-btn border border-border overflow-hidden">
+            {(['stock', 'etf'] as const).map(t => (
+              <button
+                key={t}
+                type="button"
+                onClick={() => setDraft(d => ({ ...d, asset_type: t, strategy_id: null, symbols: [] }))}
+                className={`h-full px-4 text-xs font-medium transition-colors cursor-pointer
+                  ${assetType === t ? 'bg-accent/10 text-accent' : 'text-muted hover:text-foreground'}`}
+              >
+                {t === 'stock' ? '股票' : 'ETF'}
+              </button>
+            ))}
+          </div>
+        </div>
+      )}
+
       {/* 描述 (可选) + 类型 */}
       <div className="grid grid-cols-1 md:grid-cols-3 gap-3">
         <label className="md:col-span-2 space-y-1.5">

frontend/src/lib/api.ts

@@ -428,6 +428,7 @@ export interface MonitorRule {
   name: string
   enabled: boolean
   type: 'strategy' | 'signal' | 'price' | 'market' | 'ladder'
+  asset_type?: 'stock' | 'etf'
   scope: 'symbols' | 'all' | 'sector'
   symbols: string[]
   sector?: string | null
@@ -1213,16 +1214,17 @@ export const api = {
         : '/api/watchlist/enriched',
     ),
 
-  screenerStrategies: () => request<{ presets: ScreenerStrategy[]; load_errors?: StrategyLoadError[] }>('/api/screener/strategies'),
-  screenerRunPreset: (strategy_id: string, pool?: string[], asOf?: string, extColumns?: string) =>
+  screenerStrategies: (assetType: 'stock' | 'etf' = 'stock') =>
+    request<{ presets: ScreenerStrategy[]; load_errors?: StrategyLoadError[] }>(`/api/screener/strategies?asset_type=${assetType}`),
+  screenerRunPreset: (strategy_id: string, pool?: string[], asOf?: string, extColumns?: string, assetType: 'stock' | 'etf' = 'stock') =>
     request<ScreenerResult>('/api/screener/run_preset', {
       method: 'POST',
-      body: JSON.stringify({ strategy_id, pool, as_of: asOf ?? null, ext_columns: extColumns || null }),
+      body: JSON.stringify({ strategy_id, pool, as_of: asOf ?? null, ext_columns: extColumns || null, asset_type: assetType }),
     }),
-  screenerRunCustom: (conditions: string[], orderBy?: string, limit = 30, pool?: string[], extColumns?: string) =>
+  screenerRunCustom: (conditions: string[], orderBy?: string, limit = 30, pool?: string[], extColumns?: string, assetType: 'stock' | 'etf' = 'stock') =>
     request<ScreenerResult>('/api/screener/run', {
       method: 'POST',
-      body: JSON.stringify({ conditions, order_by: orderBy, limit, pool, ext_columns: extColumns || null }),
+      body: JSON.stringify({ conditions, order_by: orderBy, limit, pool, ext_columns: extColumns || null, asset_type: assetType }),
     }),
   screenerRunAll: (asOf?: string, strategyIds?: string[], extColumns?: string) =>
     request<{ as_of: string | null; results: Record<string, { total: number; as_of: string; rows: any[] }> }>(
@@ -1264,6 +1266,7 @@ export const api = {
     stop_loss_pct?: number
     max_hold_days?: number
     matching?: 'close_t' | 'open_t+1'
+    asset_type?: 'stock' | 'etf'
   }) =>
     request<BacktestResult>('/api/backtest/run', {
       method: 'POST',
@@ -1283,6 +1286,7 @@ export const api = {
     weight?: 'equal' | 'factor_weight'
     fees_pct?: number
     slippage_bps?: number
+    asset_type?: 'stock' | 'etf'
   }) =>
     request<FactorBacktestResult>('/api/backtest/factor/run', {
       method: 'POST',
@@ -1306,6 +1310,7 @@ export const api = {
     max_positions?: number
     initial_capital?: number
     position_sizing?: 'equal' | 'score_weight'
+    asset_type?: 'stock' | 'etf'
   }) =>
     request<StrategyBacktestResult>('/api/backtest/strategy/run', {
       method: 'POST',

frontend/src/lib/backtestTask.ts

@@ -138,6 +138,7 @@ export function startBacktest(params: {
   overrides?: Record<string, any> | null
   mode?: 'position' | 'full'
   holding_days?: number
+  asset_type?: 'stock' | 'etf'
 }): void {
   // 取消之前的任务状态
   if (eventSource) {
@@ -169,6 +170,7 @@ export function startBacktest(params: {
     overrides: params.overrides ? JSON.stringify(params.overrides) : undefined,
     mode: params.mode,
     holding_days: params.holding_days,
+    asset_type: params.asset_type,
   })
 
   // 存 reconnect 信息 (刷新后用)

frontend/src/lib/queryKeys.ts

@@ -33,7 +33,7 @@ export const QK = {
 
   // Screener
   screener:             ['screener'] as const,
-  screenerStrategies:   ['screener-strategies'] as const,
+  screenerStrategies:   (assetType: string = 'stock') => ['screener-strategies', assetType] as const,
   screenerCached:       (ext?: string) => ['screener-cached', ext] as const,
   screenerKlineBatch:   (symbols: string) => ['screener-kline-batch', symbols] as const,
   marketSnapshot:       ['market-snapshot'] as const,

frontend/src/lib/storage.ts

@@ -88,6 +88,7 @@ export const storage = {
   strategyBacktestLast: kv<{
     selectedStrategy: string | null
     symbols: string
+    assetType?: 'stock' | 'etf'
     start: string
     end: string
     matching: 'close_t' | 'open_t+1'

frontend/src/pages/Screener.tsx

@@ -33,6 +33,7 @@ import {
 } from '@/lib/screener-columns'
 
 export function Screener() {
+  const [assetType, setAssetType] = useState<'stock' | 'etf'>('stock')
   const [activeStrategy, setActiveStrategy] = useState<string | null>(null)
   const [result, setResult] = useState<ScreenerResult | null>(null)
   const [asOf, setAsOf] = useState<string>('')
@@ -106,8 +107,8 @@ export function Screener() {
   const screenerAutoRun = prefs?.screener_auto_run ?? true
 
   const strategies = useQuery({
-    queryKey: QK.screenerStrategies,
-    queryFn: api.screenerStrategies,
+    queryKey: QK.screenerStrategies(assetType),
+    queryFn: () => api.screenerStrategies(assetType),
   })
 
   // 策略结果缓存 — 文件读取，SSE invalidation 自动刷新
@@ -341,6 +342,8 @@ export function Screener() {
   // asOf 确定后 + 策略列表就绪 + 策略池非空 → 自动跑一次 (受系统设置开关控制)
   // 缓存命中时秒加载; 未命中时, 仅当 screener_auto_run 开启才自动触发 runAll
   useEffect(() => {
+    // ETF 模式无股票盘后缓存/ runAll, 单策略走实时单跑, 不触发 runAll
+    if (assetType !== 'stock') return
     if (!asOf || !strategies.data?.presets?.length || runAll.isPending || visiblePool.length === 0) return
     const runKey = `${asOf}|${visiblePool.join(',')}|${extColumnsParam}`
     if (runAllDateRef.current === runKey) return
@@ -363,7 +366,7 @@ export function Screener() {
 
   const run = useMutation({
     mutationFn: ({ id, date }: { id: string; date: string }) =>
-      api.screenerRunPreset(id, undefined, date || undefined, extColumnsParam || undefined),
+      api.screenerRunPreset(id, undefined, date || undefined, extColumnsParam || undefined, assetType),
     onSuccess: (data, vars) => {
       setResult(data)
       // 同步更新卡片上的命中数
@@ -379,6 +382,12 @@ export function Screener() {
     handleStrategySwitch(s.id)
     setActiveStrategy(s.id)
     setShowAll(false)
+    // ETF 模式: 无股票盘后缓存, 始终实时单跑。
+    // 传空日期让后端用 ETF 自己的最新交易日 (asOf 跟随的是股票 enriched, 两者可能不同日)。
+    if (assetType !== 'stock') {
+      run.mutate({ id: s.id, date: '' })
+      return
+    }
     // 优先从 effectiveResults (缓存 + runAll) 取数据
     const r = effectiveResults?.[s.id]
     if (r && r.as_of === asOf) {
@@ -441,7 +450,7 @@ export function Screener() {
   const reloadStrategies = useMutation({
     mutationFn: api.strategyReload,
     onSuccess: () => {
-      qc.invalidateQueries({ queryKey: QK.screenerStrategies })
+      qc.invalidateQueries({ queryKey: ['screener-strategies'] })
       if (asOf) runAll.mutate({ date: asOf })
     },
   })
@@ -509,6 +518,22 @@ export function Screener() {
         subtitle="基于本地 enriched 表 · 毫秒级 SQL"
         right={
           <div className="flex items-center gap-2">
+            {/* 资产类型切换: 股票 / ETF */}
+            <div className="flex items-center h-7 rounded-btn border border-border overflow-hidden">
+              {(['stock', 'etf'] as const).map(t => (
+                <button
+                  key={t}
+                  onClick={() => { setAssetType(t); setActiveStrategy(null); setResult(null); setShowAll(false) }}
+                  className={`h-full px-2.5 text-xs font-medium transition-colors cursor-pointer
+                    ${assetType === t
+                      ? 'bg-accent/10 text-accent'
+                      : 'text-muted hover:text-secondary hover:bg-elevated'
+                    }`}
+                >
+                  {t === 'stock' ? '股票' : 'ETF'}
+                </button>
+              ))}
+            </div>
             {/* 重新运行策略：重载策略文件并重跑全部策略，更新命中个股 */}
             <button
               onClick={() => reloadStrategies.mutate()}
@@ -851,7 +876,7 @@ export function Screener() {
             const rules = storage.strategyRules.get({})
             delete rules[settingsStrategyId]; storage.strategyRules.set(rules)
             setStrategyLimits(prev => { const next = {...prev}; delete next[settingsStrategyId]; return next })
-            qc.invalidateQueries({ queryKey: QK.screenerStrategies })
+            qc.invalidateQueries({ queryKey: ['screener-strategies'] })
           }
         }}
       />
@@ -874,7 +899,7 @@ export function Screener() {
         onClose={() => setShowBuilder(false)}
         mode={builderMode}
         onSavedId={async id => {
-          const data = await qc.fetchQuery({ queryKey: QK.screenerStrategies, queryFn: api.screenerStrategies })
+          const data = await qc.fetchQuery({ queryKey: QK.screenerStrategies('stock'), queryFn: () => api.screenerStrategies('stock') })
           if (!data.presets.some(s => s.id === id)) {
             throw new Error(`策略 ${id} 已保存但未加载，请检查策略代码`)
           }

frontend/src/pages/backtest/FactorBacktest.tsx

@@ -83,6 +83,7 @@ function LoadingPanel({ symbolsText }: { symbolsText: string }) {
 export function FactorBacktest() {
   const [factorName, setFactorName] = useState('momentum_20d')
   const [symbols, setSymbols] = useState('')
+  const [assetType, setAssetType] = useState<'stock' | 'etf'>('stock')
   const [start, setStart] = useState(THREE_MONTHS_AGO)
   const [end, setEnd] = useState(TODAY)
   const [nGroups, setNGroups] = useState(5)
@@ -114,6 +115,7 @@ export function FactorBacktest() {
     mutationFn: () =>
       api.factorRun({
         factor_name: factorName,
+        asset_type: assetType,
         symbols: symbols ? symbols.split(',').map(s => s.trim()).filter(Boolean) : null,
         start: start || null,
         end: end || undefined,
@@ -194,8 +196,22 @@ export function FactorBacktest() {
         </div>
 
         <div>
+          <label className="text-xs font-medium text-secondary block mb-1.5">资产类型</label>
+          <div className="inline-flex h-8 rounded-btn border border-border overflow-hidden mb-2">
+            {(['stock', 'etf'] as const).map(t => (
+              <button
+                key={t}
+                type="button"
+                onClick={() => { setAssetType(t); setSymbols('') }}
+                className={`h-full px-3 text-xs font-medium transition-colors cursor-pointer
+                  ${assetType === t ? 'bg-accent/10 text-accent' : 'text-muted hover:text-foreground'}`}
+              >
+                {t === 'stock' ? '股票' : 'ETF'}
+              </button>
+            ))}
+          </div>
           <label className="text-xs font-medium text-secondary block mb-1.5">
-            标的(逗号分隔，留空=全市场)
+            标的(逗号分隔，留空=全市场{assetType === 'etf' ? ' ETF' : ''})
           </label>
           <input
             type="text"

frontend/src/pages/backtest/StrategyBacktest.tsx

@@ -563,15 +563,16 @@ function StrategyParamInput({ param, value, onChange }: {
   )
 }
 
-function StockPoolPicker({ value, onChange }: { value: string; onChange: (value: string) => void }) {
+function StockPoolPicker({ value, onChange, assetType = 'stock' }: { value: string; onChange: (value: string) => void; assetType?: 'stock' | 'etf' }) {
   const symbols = useMemo(() => value.split(',').map(s => s.trim()).filter(Boolean), [value])
   const [query, setQuery] = useState('')
   const [open, setOpen] = useState(false)
   const [symbolNames, setSymbolNames] = useState<Record<string, string>>({})
   const ref = useRef<HTMLDivElement>(null)
+  const searchAssetTypes = assetType === 'etf' ? 'stock,etf' : 'stock'
   const search = useQuery({
-    queryKey: QK.instrumentSearch(query),
-    queryFn: () => api.instrumentSearch(query),
+    queryKey: QK.instrumentSearch(query, searchAssetTypes),
+    queryFn: () => api.instrumentSearch(query, 20, searchAssetTypes),
     enabled: query.trim().length > 0,
     staleTime: 30_000,
   })
@@ -710,6 +711,7 @@ export function StrategyBacktest() {
   const [selectedStrategy, setSelectedStrategy] = useState<string | null>(saved?.selectedStrategy ?? null)
   const [strategyGroup, setStrategyGroup] = useState<StrategyGroup>('all')
   const [symbols, setSymbols] = useState(saved?.symbols ?? '')
+  const [assetType, setAssetType] = useState<'stock' | 'etf'>(saved?.assetType ?? 'stock')
   const [start, setStart] = useState(saved?.start ?? THREE_MONTHS_AGO)
   const [end, setEnd] = useState(saved?.end ?? TODAY)
   // 成交口径: 建仓/清仓可独立配置。向后兼容老 matching (派生为 entry=exit=matching)。
@@ -749,8 +751,8 @@ export function StrategyBacktest() {
   const loadedStrategyRef = useRef<string | null>(null)
 
   const strategies = useQuery({
-    queryKey: QK.screenerStrategies,
-    queryFn: api.screenerStrategies,
+    queryKey: QK.screenerStrategies(assetType),
+    queryFn: () => api.screenerStrategies(assetType),
   })
 
   const strategyList = useMemo(() => strategies.data?.presets ?? [], [strategies.data])
@@ -818,6 +820,7 @@ export function StrategyBacktest() {
       storage.strategyBacktestLast.set({
         selectedStrategy,
         symbols,
+        assetType,
         start,
         end,
         matching,
@@ -843,6 +846,7 @@ export function StrategyBacktest() {
     if (!selectedStrategy) return
     startBacktest({
       strategy_id: selectedStrategy,
+      asset_type: assetType,
       symbols: symbols ? symbols.split(',').map(s => s.trim()).filter(Boolean) : null,
       start: start || null,
       end: end || undefined,
@@ -2009,7 +2013,24 @@ export function StrategyBacktest() {
 
               {settingsTab === 'range' && (
                 <ConfigSection title="回测范围">
-                  <StockPoolPicker value={symbols} onChange={setSymbols} />
+                  <div className="flex items-center gap-2">
+                    <span className="text-[11px] text-muted">资产类型</span>
+                    <div className="inline-flex h-8 rounded-btn border border-border overflow-hidden">
+                      {(['stock', 'etf'] as const).map(t => (
+                        <button
+                          key={t}
+                          type="button"
+                          onClick={() => { setAssetType(t); setSelectedStrategy(null); setSymbols('') }}
+                          className={`h-full px-3 text-xs font-medium transition-colors cursor-pointer
+                            ${assetType === t ? 'bg-accent/10 text-accent' : 'text-muted hover:text-foreground'}`}
+                        >
+                          {t === 'stock' ? '股票' : 'ETF'}
+                        </button>
+                      ))}
+                    </div>
+                    <span className="text-[11px] text-muted/70">ETF 仅技术类策略,读 ETF enriched</span>
+                  </div>
+                  <StockPoolPicker value={symbols} onChange={setSymbols} assetType={assetType} />
                   <div className="text-[11px] leading-5 text-muted">默认全市场回测，由基础过滤、策略条件和买卖触发器筛选；需要单票调试或自选池回测时再限定股票池。</div>
                 </ConfigSection>
               )}