移除自选和回测功能

Co-Authored-By: Claude <noreply@anthropic.com>

serve/backend/app/api/backtest.py

@@ -1,474 +0,0 @@
-"""回测 API — 信号回测 + 因子回测 + 策略回测。"""
-from __future__ import annotations
-
-import asyncio
-import json
-import queue
-import threading
-from dataclasses import asdict
-from datetime import date, timedelta
-from typing import Literal
-
-from fastapi import APIRouter, HTTPException, Request
-from fastapi.responses import StreamingResponse
-from pydantic import BaseModel, Field
-
-from app.config import settings
-from app.services.backtest import (
-    BacktestConfig,
-    BacktestService,
-    VectorbtUnavailable,
-    is_available,
-)
-
-router = APIRouter(prefix="/api/backtest", tags=["backtest"])
-
-FACTOR_DEFAULT_DAYS = 180
-STRATEGY_DEFAULT_DAYS = 365 * 3
-BACKTEST_MAX_SERVER_DAYS = 186
-FACTOR_MAX_SYMBOLS = 1000
-BACKTEST_SERVER_GUARD_MESSAGE = (
-    "当前服务器内存约 1.8GB，回测区间最多支持 6 个月；"
-    "更长周期容易触发 OOM，建议在 8GB 以上内存环境或本机运行。"
-)
-
-
-def _get_engine(request: Request):
-    """获取或创建 BacktestEngine (单例，PanelCache 跨请求生效)。"""
-    from app.backtest.engine import BacktestEngine
-    engine = getattr(request.app.state, "backtest_engine", None)
-    if engine is None:
-        engine = BacktestEngine(request.app.state.repo)
-        request.app.state.backtest_engine = engine
-    return engine
-
-
-def _resolve_start(req: BaseModel, end: date, default_days: int) -> date:
-    """未传 start 使用默认区间；显式传 null/空值表示全部历史。"""
-    start = getattr(req, "start")
-    if start is not None:
-        return start
-    if "start" in req.model_fields_set:
-        return date(1900, 1, 1)
-    return end - timedelta(days=default_days)
-
-
-def _guard_server_backtest_range(start: date, end: date):
-    if not settings.backtest_range_guard:
-        return
-    days = (end - start).days + 1
-    if days > BACKTEST_MAX_SERVER_DAYS:
-        raise HTTPException(status_code=400, detail=BACKTEST_SERVER_GUARD_MESSAGE)
-
-
-# ================================================================
-# 状态
-# ================================================================
-
-@router.get("/status")
-def status():
-    """前端可用此接口判断回测页是否要灰显。"""
-    return {"available": True}
-
-
-# ================================================================
-# 信号回测 (现有接口，保持不变)
-# ================================================================
-
-class BacktestRequest(BaseModel):
-    symbols: list[str] = Field(..., min_length=1)
-    start: date | None = None
-    end: date | None = None
-    entries: list[str] = []
-    exits: list[str] = []
-    stop_loss_pct: float | None = None
-    max_hold_days: int | None = None
-    fees_pct: float = 0.0002
-    slippage_bps: float = 5
-    matching: Literal["close_t", "open_t+1"] = "close_t"
-
-
-@router.post("/run")
-def run(req: BacktestRequest, request: Request):
-    """信号回测 — 现有接口，向后兼容。"""
-    repo = request.app.state.repo
-    svc = BacktestService(repo)
-    end = req.end or date.today()
-    start = req.start or (end - timedelta(days=365 * 3))
-
-    cfg = BacktestConfig(
-        symbols=req.symbols,
-        start=start,
-        end=end,
-        entries=req.entries,
-        exits=req.exits,
-        stop_loss_pct=req.stop_loss_pct,
-        max_hold_days=req.max_hold_days,
-        fees_pct=req.fees_pct,
-        slippage_bps=req.slippage_bps,
-        matching=req.matching,
-    )
-    try:
-        result = svc.run(cfg)
-    except VectorbtUnavailable as e:
-        raise HTTPException(status_code=503, detail=str(e)) from e
-    return asdict(result)
-
-
-# ================================================================
-# 因子回测
-# ================================================================
-
-class FactorColumnsResponse(BaseModel):
-    columns: list[dict]
-
-
-@router.get("/factor/columns")
-def factor_columns():
-    """返回可用的因子列列表。"""
-    from app.backtest.factor import FACTOR_COLUMNS
-    return {"columns": FACTOR_COLUMNS}
-
-
-class FactorBacktestRequest(BaseModel):
-    factor_name: str
-    symbols: list[str] | None = None
-    start: date | None = None
-    end: date | None = None
-    n_groups: int = 5
-    rebalance: Literal["daily", "weekly", "monthly"] = "monthly"
-    weight: Literal["equal", "factor_weight"] = "equal"
-    fees_pct: float = 0.0002
-    slippage_bps: float = 5.0
-
-
-@router.post("/factor/run")
-def factor_run(req: FactorBacktestRequest, request: Request):
-    """因子回测 — IC/IR 分析 + 分层回测。"""
-    from app.backtest.factor import FactorBacktestService, FactorConfig
-
-    engine = _get_engine(request)
-    svc = FactorBacktestService(engine)
-
-    end = req.end or date.today()
-    start = _resolve_start(req, end, STRATEGY_DEFAULT_DAYS)
-    _guard_server_backtest_range(start, end)
-    symbols = req.symbols if req.symbols else None
-    if symbols is not None and len(symbols) > FACTOR_MAX_SYMBOLS:
-        raise HTTPException(
-            status_code=400,
-            detail=f"指定标的最多支持 {FACTOR_MAX_SYMBOLS} 只，请缩小标的范围。",
-        )
-
-    cfg = FactorConfig(
-        factor_name=req.factor_name,
-        symbols=symbols,
-        start=start,
-        end=end,
-        n_groups=req.n_groups,
-        rebalance=req.rebalance,
-        weight=req.weight,
-        fees_pct=req.fees_pct,
-        slippage_bps=req.slippage_bps,
-    )
-    result = svc.run(cfg)
-    return asdict(result)
-
-
-# ================================================================
-# 策略回测
-# ================================================================
-
-class StrategyBacktestRequest(BaseModel):
-    strategy_id: str
-    symbols: list[str] | None = None
-    start: date | None = None
-    end: date | None = None
-    params: dict | None = None
-    overrides: dict | None = None
-    # matching 向后兼容; 显式传 entry_fill/exit_fill 时以二者为准。
-    matching: Literal["close_t", "open_t+1"] = "open_t+1"
-    entry_fill: Literal["close_t", "open_t+1"] | None = None
-    exit_fill: Literal["close_t", "open_t+1"] | None = None
-    fees_pct: float = 0.0002
-    slippage_bps: float = 5.0
-    max_positions: int = 10
-    max_exposure_pct: float = 1.0
-    initial_capital: float = 1_000_000.0
-    position_sizing: Literal["equal", "score_weight"] = "equal"
-    mode: Literal["position", "full"] = "position"
-    holding_days: int = 5
-
-
-@router.post("/strategy/run")
-def strategy_run(req: StrategyBacktestRequest, request: Request):
-    """策略回测 — 复用 StrategyDef 体系做全周期回测。"""
-    from app.backtest.strategy import StrategyBacktestService, StrategyBacktestConfig
-
-    engine = _get_engine(request)
-    strategy_engine = request.app.state.strategy_engine
-    svc = StrategyBacktestService(engine, strategy_engine)
-
-    end = req.end or date.today()
-    start = _resolve_start(req, end, FACTOR_DEFAULT_DAYS)
-    _guard_server_backtest_range(start, end)
-
-    cfg = StrategyBacktestConfig(
-        strategy_id=req.strategy_id,
-        symbols=req.symbols if req.symbols else None,
-        start=start,
-        end=end,
-        params=req.params,
-        overrides=req.overrides,
-        matching=req.matching,
-        entry_fill=req.entry_fill,
-        exit_fill=req.exit_fill,
-        fees_pct=req.fees_pct,
-        slippage_bps=req.slippage_bps,
-        max_positions=req.max_positions,
-        max_exposure_pct=req.max_exposure_pct,
-        initial_capital=req.initial_capital,
-        position_sizing=req.position_sizing,
-        mode=req.mode,
-        holding_days=req.holding_days,
-    )
-    result = svc.run(cfg)
-    return asdict(result)
-
-
-# ── SSE 流式回测 (实时进度 + 可取消 + 支持重连) ───────────────────
-
-import time
-import hashlib
-
-
-class _BacktestJob:
-    """单个回测任务的状态, 存模块级供重连使用。"""
-    __slots__ = ("key", "cancel_event", "progress", "result", "error", "done", "finish_ts")
-
-    def __init__(self, key: str):
-        self.key = key
-        self.cancel_event = threading.Event()
-        self.progress: list[dict] = []   # 进度历史 (新连接可回放)
-        self.result = None               # 完成后的结果
-        self.error: str | None = None
-        self.done = False
-        self.finish_ts: float = 0.0
-
-
-# 模块级任务表: key -> _BacktestJob
-_running_jobs: dict[str, _BacktestJob] = {}
-_jobs_lock = threading.Lock()
-_JOB_TTL = 300  # 完成后保留 5 分钟
-
-
-def _cleanup_stale_jobs():
-    """清理过期任务 (完成超过 TTL 的)。"""
-    now = time.time()
-    stale = [k for k, j in _running_jobs.items() if j.done and now - j.finish_ts > _JOB_TTL]
-    for k in stale:
-        _running_jobs.pop(k, None)
-
-
-def _make_job_key(
-    strategy_id: str, symbols: str | None, start: str | None, end: str | None,
-    matching: str, entry_fill: str | None, exit_fill: str | None,
-    fees_pct: float, slippage_bps: float,
-    max_positions: int, max_exposure_pct: float, initial_capital: float, position_sizing: str,
-    params: str | None, overrides: str | None,
-    mode: str = "position", holding_days: int = 5,
-) -> 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}"
-    return hashlib.md5(raw.encode()).hexdigest()[:12]
-
-
-@router.get("/strategy/stream")
-async def strategy_stream(
-    request: Request,
-    strategy_id: str,
-    symbols: str | None = None,
-    start: str | None = None,
-    end: str | None = None,
-    matching: str = "open_t+1",
-    entry_fill: str | None = None,
-    exit_fill: str | None = None,
-    fees_pct: float = 0.0002,
-    slippage_bps: float = 5.0,
-    max_positions: int = 10,
-    max_exposure_pct: float = 1.0,
-    initial_capital: float = 1_000_000.0,
-    position_sizing: str = "equal",
-    params: str | None = None,
-    overrides: str | None = None,
-    mode: str = "position",
-    holding_days: int = 5,
-):
-    """SSE 流式策略回测: 实时推送进度, 完成后推送结果, 支持重连 (刷新/切页后恢复)。
-
-    - 相同参数的任务只启动一次, 多次连接订阅同一个任务
-    - 断开连接不会取消任务 (除非显式调用 cancel)
-    - 结果保留 5 分钟供重连
-
-    事件类型:
-      - progress: {day, total, date, equity}
-      - done: {result} (完整回测结果)
-      - error: {message}
-    """
-    from app.backtest.strategy import StrategyBacktestService, StrategyBacktestConfig
-
-    engine = _get_engine(request)
-    strategy_engine = request.app.state.strategy_engine
-    svc = StrategyBacktestService(engine, strategy_engine)
-
-    end_date = date.fromisoformat(end) if end else date.today()
-    if start:
-        start_date = date.fromisoformat(start)
-    else:
-        # 空 start = 全部历史: 用本地最早日K日期, 查不到再回退到默认窗口
-        earliest = request.app.state.repo.earliest_daily_date()
-        start_date = earliest or (end_date - timedelta(days=FACTOR_DEFAULT_DAYS))
-
-    # 服务端范围保护
-    guard_violated = False
-    if settings.backtest_range_guard:
-        days = (end_date - start_date).days + 1
-        if days > BACKTEST_MAX_SERVER_DAYS:
-            guard_violated = True
-
-    job_key = _make_job_key(
-        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,
-    )
-
-    _cleanup_stale_jobs()
-
-    # 获取或创建任务
-    with _jobs_lock:
-        job = _running_jobs.get(job_key)
-        if job is None:
-            job = _BacktestJob(job_key)
-            _running_jobs[job_key] = job
-            is_new = True
-        else:
-            is_new = False
-
-    async def event_generator():
-        # 范围保护: 直接报错
-        if guard_violated:
-            yield f"event: error\ndata: {json.dumps({'message': BACKTEST_SERVER_GUARD_MESSAGE}, ensure_ascii=False)}\n\n"
-            return
-
-        # 如果是新任务, 启动回测线程
-        if is_new and not job.done:
-            cfg = StrategyBacktestConfig(
-                strategy_id=strategy_id,
-                symbols=[s.strip() for s in symbols.split(",") if s.strip()] if symbols else None,
-                start=start_date,
-                end=end_date,
-                params=json.loads(params) if params else None,
-                overrides=json.loads(overrides) if overrides else None,
-                matching=matching,
-                entry_fill=entry_fill,
-                exit_fill=exit_fill,
-                fees_pct=fees_pct,
-                slippage_bps=slippage_bps,
-                max_positions=int(max_positions),
-                max_exposure_pct=float(max_exposure_pct),
-                initial_capital=float(initial_capital),
-                position_sizing=position_sizing,
-                mode=mode,
-                holding_days=int(holding_days),
-            )
-
-            def _run_backtest():
-                try:
-                    result = svc.run(cfg, lambda d: job.progress.append(d), job.cancel_event)
-                    job.result = result
-                    job.done = True
-                    job.finish_ts = time.time()
-                except Exception as e:
-                    job.error = str(e)
-                    job.done = True
-                    job.finish_ts = time.time()
-
-            # 启动后台线程 (不阻塞事件循环)
-            threading.Thread(target=_run_backtest, daemon=True).start()
-
-        # 订阅进度: 用读指针读 job.progress 列表 (多连接互不干扰)
-        cursor = 0
-        tick = 0
-
-        try:
-            while True:
-                # 已完成: 推送最终结果/错误并退出
-                if job.done:
-                    if job.error:
-                        yield f"event: error\ndata: {json.dumps({'message': job.error}, ensure_ascii=False)}\n\n"
-                    elif job.result is not None:
-                        r = job.result
-                        if hasattr(r, "error") and r.error == "cancelled":
-                            yield f"event: error\ndata: {json.dumps({'message': '回测已取消'}, ensure_ascii=False)}\n\n"
-                        elif hasattr(r, "error") and r.error:
-                            yield f"event: error\ndata: {json.dumps({'message': r.error}, ensure_ascii=False)}\n\n"
-                        else:
-                            yield f"event: done\ndata: {json.dumps(asdict(r), ensure_ascii=False, default=str)}\n\n"
-                    return
-
-                # 断开检测: 每 4 轮检查一次 (降低 GIL 抢占频率)
-                tick += 1
-                if tick % 4 == 0 and await request.is_disconnected():
-                    break
-
-                # 推送新进度 (从 cursor 开始读)
-                prog_list = job.progress
-                while cursor < len(prog_list):
-                    msg = prog_list[cursor]
-                    cursor += 1
-                    yield f"event: progress\ndata: {json.dumps(msg, ensure_ascii=False, default=str)}\n\n"
-
-                await asyncio.sleep(0.5)
-
-        except asyncio.CancelledError:
-            raise
-
-    return StreamingResponse(event_generator(), media_type="text/event-stream")
-
-
-@router.post("/strategy/cancel")
-async def strategy_cancel(request: Request):
-    """取消正在运行的回测任务 (前端传 query string, 后端算 job_key)。"""
-    body = await request.json()
-    qs = body.get("qs", "")
-    # 解析 qs 得到参数
-    from urllib.parse import parse_qs
-    p = parse_qs(qs)
-    def _get(key: str, default: str = "") -> str:
-        return p.get(key, [default])[0]
-    job_key = _make_job_key(
-        _get("strategy_id"),
-        _get("symbols") or None,
-        _get("start") or None,
-        _get("end") or None,
-        _get("matching", "open_t+1"),
-        _get("entry_fill") or None,
-        _get("exit_fill") or None,
-        float(_get("fees_pct", "0.0002")),
-        float(_get("slippage_bps", "5")),
-        int(_get("max_positions", "10")),
-        float(_get("max_exposure_pct", "1")),
-        float(_get("initial_capital", "1000000")),
-        _get("position_sizing", "equal"),
-        _get("params") or None,
-        _get("overrides") or None,
-        _get("mode", "position"),
-        int(_get("holding_days", "5")),
-    )
-    job = _running_jobs.get(job_key)
-    if job and not job.done:
-        job.cancel_event.set()
-        return {"ok": True}
-    return {"ok": False, "message": "任务不存在或已完成"}
-

serve/backend/app/api/data.py

@@ -505,7 +505,7 @@ def _compute_storage(data_dir: Path) -> dict:
         stats[f"{key}_size_mb"] = sz
 
     # total: 再加上其他零散文件(pools, financials, capabilities.json 等)
-    other_dirs = ["pools", "financials", "backtest_results", "screener_results", "ai_cache"]
+    other_dirs = ["pools", "financials", "screener_results", "ai_cache"]
     for name in other_dirs:
         d = data_dir / name
         if d.exists():
@@ -629,7 +629,7 @@ def clear_data(request: Request):
         "kline_daily", "kline_daily_enriched", "kline_index_daily", "kline_index_enriched",
         "kline_etf_daily", "kline_etf_enriched", "kline_etf_minute", "kline_minute",
         "adj_factor", "adj_factor_etf", "instruments", "instruments_index", "instruments_etf", "pools", "financials",
-        "backtest_results", "screener_results", "ai_cache",
+        "screener_results", "ai_cache",
     ):
         d = data_dir / sub
         if d.exists():

serve/backend/app/api/kline.py

@@ -150,7 +150,7 @@ def _attach_ext(resp: dict, repo, symbol: str, ext_columns: Optional[str]) -> di
     """按 ext_columns 规格为单只股票 LEFT JOIN 扩展数据，平铺到 stock_info['ext']。
 
     key 形如 "{config_id}__{field_name}"，与自选列表 enriched 接口保持一致。
-    JOIN 逻辑参考 watchlist.watchlist_enriched；任何 ext 表/字段缺失都静默跳过。
+    JOIN 逻辑；任何 ext 表/字段缺失都静默跳过。
     """
     if not ext_columns or not ext_columns.strip():
         return resp
@@ -380,7 +380,7 @@ async def sync_minute(request: Request):
 
         try:
             progress("sync_minute", 5, "解析标的池…")
-            universe = sorted(set(get_pool("watchlist")) | set(get_pool("CN_Equity_A")))
+            universe = sorted(set(get_pool("CN_Equity_A")))
             # 补充 instruments 全量标的，覆盖北交所、新股等
             inst_path = repo.store.data_dir / "instruments" / "instruments.parquet"
             if inst_path.exists():

serve/backend/app/api/settings.py

@@ -332,7 +332,6 @@ def get_preferences() -> dict:
         "instruments_schedule": preferences.get_instruments_schedule(),
         "enriched_batch_size": preferences.get_enriched_batch_size(),
         "index_daily_batch_size": preferences.get_index_daily_batch_size(),
-        "watchlist_columns": preferences.get_watchlist_columns(),
         "screener_result_columns": preferences.get_screener_result_columns(),
         "sidebar_index_symbols": preferences.get_sidebar_index_symbols(),
         "nav_order": preferences.get_nav_order(),
@@ -343,14 +342,6 @@ def get_preferences() -> dict:
     }
 
 
-@router.get("/preferences/watchlist-columns")
-def get_watchlist_columns() -> dict:
-    """返回自选列表列配置。"""
-    from app.services import preferences
-    cols = preferences.get_watchlist_columns()
-    return {"columns": cols}
-
-
 class NavOrderIn(BaseModel):
     nav_order: list[str]
 
@@ -375,15 +366,6 @@ def update_nav_hidden(req: NavHiddenIn) -> dict:
     return {"nav_hidden": saved}
 
 
-@router.put("/preferences/watchlist-columns")
-def update_watchlist_columns(req: dict) -> dict:
-    """保存自选列表列配置。"""
-    from app.services import preferences
-    columns = req.get("columns", [])
-    saved = preferences.set_watchlist_columns(columns)
-    return {"columns": saved}
-
-
 @router.get("/preferences/screener-result-columns")
 def get_screener_result_columns() -> dict:
     """返回策略结果列表列配置。"""

serve/backend/app/api/watchlist.py

@@ -1,222 +0,0 @@
-"""自选股 API。"""
-from __future__ import annotations
-
-import logging
-import math
-import time
-from datetime import date
-
-import polars as pl
-from fastapi import APIRouter, Query, Request
-from pydantic import BaseModel
-
-from app.services import watchlist
-
-logger = logging.getLogger(__name__)
-
-router = APIRouter(prefix="/api/watchlist", tags=["watchlist"])
-
-
-class AddRequest(BaseModel):
-    symbol: str
-    note: str = ""
-
-
-class BatchAddRequest(BaseModel):
-    symbols: list[str]
-    note: str = ""
-
-
-def _with_names(rows: list[dict], request: Request) -> list[dict]:
-    if not rows:
-        return rows
-    try:
-        df_i = request.app.state.repo.get_instruments()
-        if df_i.is_empty() or "symbol" not in df_i.columns or "name" not in df_i.columns:
-            return rows
-        name_by_symbol = dict(df_i.select(["symbol", "name"]).iter_rows())
-        return [{**row, "name": name_by_symbol.get(row.get("symbol"))} for row in rows]
-    except Exception as e:  # noqa: BLE001
-        logger.debug("attach watchlist names failed: %s", e)
-        return rows
-
-
-@router.get("")
-def list_all(request: Request):
-    return {"symbols": _with_names(watchlist.list_symbols(), request)}
-
-
-@router.post("")
-def add_one(req: AddRequest, request: Request):
-    rows = watchlist.add(req.symbol, req.note)
-    return {"symbols": _with_names(rows, request)}
-
-
-@router.post("/batch")
-def add_batch(req: BatchAddRequest, request: Request):
-    for sym in req.symbols:
-        watchlist.add(sym, req.note)
-    return {"symbols": _with_names(watchlist.list_symbols(), request), "added": len(req.symbols)}
-
-
-@router.post("/{symbol}/top")
-def move_one_to_top(symbol: str, request: Request):
-    rows = watchlist.move_to_top(symbol)
-    return {"symbols": _with_names(rows, request)}
-
-
-@router.delete("/{symbol}")
-def remove_one(symbol: str, request: Request):
-    rows = watchlist.remove(symbol)
-    return {"symbols": _with_names(rows, request)}
-
-
-@router.delete("")
-def clear_all():
-    """清空自选列表。"""
-    count = watchlist.clear()
-    return {"removed": count}
-
-
-# 自选页需要的列
-_WATCHLIST_COLS = [
-    "symbol", "close", "change_pct", "change_amount", "amount",
-    "turnover_rate",
-    "amplitude", "annual_vol_20d",
-    "vol_ratio_5d",
-    "ma5", "ma10", "ma20", "ma60",
-    "vol_ma5", "vol_ma10",
-    "high_60d", "low_60d",
-    "rsi_6", "rsi_14", "rsi_24",
-    "macd_dif", "macd_dea", "macd_hist",
-    "kdj_k", "kdj_d", "kdj_j",
-    "boll_upper", "boll_lower",
-    "atr_14",
-    "momentum_5d", "momentum_10d", "momentum_20d", "momentum_30d", "momentum_60d",
-    "consecutive_limit_ups", "consecutive_limit_downs",
-    "signal_limit_up", "signal_limit_down", "signal_volume_surge",
-    "signal_ma_golden_5_20", "signal_macd_golden", "signal_n_day_high",
-    "signal_boll_breakout_upper", "signal_ma20_breakout",
-    "signal_ma_dead_5_20", "signal_macd_dead", "signal_n_day_low",
-    "signal_boll_breakdown_lower", "signal_ma20_breakdown",
-]
-
-
-@router.get("/enriched")
-def watchlist_enriched(
-    request: Request,
-    ext_columns: str | None = Query(None, description="逗号分隔的 ext 列: config_id.field_name"),
-):
-    """自选股 enriched 数据 — 直接从 enriched 最新日读取, 无即时计算。
-
-    ext_columns 参数示例: "industry_rating.score,fund_flow.net_inflow"
-    会动态 LEFT JOIN 对应的 ext_{config_id} DuckDB view。
-    """
-    t0 = time.perf_counter()
-
-    repo = request.app.state.repo
-    symbols = [r["symbol"] for r in watchlist.list_symbols()]
-    if not symbols:
-        return {"rows": [], "as_of": None, "elapsed_ms": 0}
-
-    df_e, cache_date = repo.get_enriched_latest()
-    if df_e.is_empty():
-        return {"rows": [], "as_of": None, "elapsed_ms": 0}
-
-    # 按 symbol 过滤
-    df = df_e.filter(pl.col("symbol").is_in(symbols))
-    if df.is_empty():
-        return {"rows": [], "as_of": str(cache_date) if cache_date else None, "elapsed_ms": 0}
-
-    # JOIN instruments 取 name + float_shares
-    df_i = repo.get_instruments()
-    if not df_i.is_empty() and "name" in df_i.columns:
-        inst_cols = [c for c in ["symbol", "name", "float_shares"] if c in df_i.columns]
-        df = df.join(df_i.select(inst_cols), on="symbol", how="left")
-
-    # 选择内置需要的列
-    keep = [c for c in _WATCHLIST_COLS + ["name", "float_shares"] if c in df.columns]
-    df = df.select(keep)
-
-    # 动态 JOIN 扩展数据表
-    ext_specs = _parse_ext_columns(ext_columns) if ext_columns else []
-    if ext_specs:
-        db = repo.store.db
-        data_dir = repo.store.data_dir
-        from app.services.ext_data import ExtConfigStore
-        from app.api.ext_data import _read_ext_dataframe
-
-        ext_store = ExtConfigStore(data_dir)
-        configs = {c.id: c for c in ext_store.load_all()}
-
-        for config_id, field_name in ext_specs:
-            view_name = f"ext_{config_id}"
-            ext_col_name = f"{config_id}__{field_name}"
-            try:
-                # 扩展时序数据必须只取最新分区；否则一个 symbol 会按历史分区数被 JOIN 放大。
-                cfg = configs.get(config_id)
-                if cfg:
-                    ext_df, _ = _read_ext_dataframe(cfg, data_dir)
-                else:
-                    ext_df = pl.from_arrow(db.query(
-                        f"SELECT symbol, \"{field_name}\" FROM {view_name}"
-                    ).arrow())
-                if not ext_df.is_empty() and "symbol" in ext_df.columns:
-                    ext_df = (
-                        ext_df
-                        .select(["symbol", field_name])
-                        .unique(subset=["symbol"], keep="last")
-                        .rename({field_name: ext_col_name})
-                    )
-                    df = df.join(ext_df.select(["symbol", ext_col_name]), on="symbol", how="left")
-            except Exception:
-                # view 不存在或字段不存在，尝试直接读 parquet
-                cfg = configs.get(config_id)
-                if cfg:
-                    try:
-                        ext_df, _ = _read_ext_dataframe(cfg, data_dir)
-                        if not ext_df.is_empty() and "symbol" in ext_df.columns and field_name in ext_df.columns:
-                            ext_df = (
-                                ext_df
-                                .select(["symbol", field_name])
-                                .unique(subset=["symbol"], keep="last")
-                                .rename({field_name: ext_col_name})
-                            )
-                            df = df.join(ext_df, on="symbol", how="left")
-                    except Exception as e2:
-                        logger.debug("ext join fallback failed for %s.%s: %s", config_id, field_name, e2)
-
-    # sanitize NaN / Inf
-    float_cols = [c for c in df.columns if df[c].dtype.is_float()]
-    if float_cols:
-        df = df.with_columns([
-            pl.when(pl.col(c).is_nan() | pl.col(c).is_infinite())
-              .then(None)
-              .otherwise(pl.col(c))
-              .alias(c)
-            for c in float_cols
-        ])
-
-    # 按自选添加顺序（新加的在前）重排行
-    order_map = {s: i for i, s in enumerate(symbols)}
-    df = df.with_columns(pl.col("symbol").map_elements(lambda s: order_map.get(s, len(symbols)), return_dtype=pl.Int32).alias("_sort_order"))
-    df = df.sort("_sort_order").drop("_sort_order")
-
-    rows = df.to_dicts()
-    elapsed = (time.perf_counter() - t0) * 1000
-    return {"rows": rows, "as_of": str(cache_date) if cache_date else None, "elapsed_ms": elapsed}
-
-
-def _parse_ext_columns(ext_columns: str) -> list[tuple[str, str]]:
-    """解析 'config_id1.field1,config_id2.field2' 为 [(config_id, field_name), ...]"""
-    result = []
-    for part in ext_columns.split(","):
-        part = part.strip()
-        if "." not in part:
-            continue
-        config_id, field_name = part.split(".", 1)
-        config_id = config_id.strip()
-        field_name = field_name.strip()
-        if config_id and field_name:
-            result.append((config_id, field_name))
-    return result