feat(portfolio): get_security_info_batch + get_value_metrics_batch 批量接口

策略层提速第三轮(filters 通病 + 策略01):
- get_security_info_batch: 2 条 SQL(symbol IN chunk + GROUP BY sym,exc 拿
  min/max; constituent_unified 拿 name)替 N×2 逐只。filters.py filter_st_stock/
  filter_new_stock 自动探测批量(hasattr + isinstance dict 回退逐只, 向后兼容)。
  三策略 ST/次新过滤通病: 万次查询压成 2 条。
- get_value_metrics_batch: ThreadPool 并发逐只委托 lpp.get_value_metrics
  (多期 ROE/FCF/流动比率逻辑不变, 只并发)。策略01 价值精选提速。
- 接口3 get_ticks_batch 不做: 实证 get_current_tick 无 last_price/paused 字段
  → 回测涨跌停/停牌 filter 恒不过滤(死代码), 批量化无意义; 真问题是回测
  涨跌停检测失效(策略层另修)。

Mac TDD parity 测试全绿(batch==逐只); 93 回归通过。

sanguo_portfolio/filters.py

@@ -29,15 +29,34 @@ def _safe_call(fn: Any, *args, **kwargs) -> Any:
         return None
 
 
+def _batch_security_info(provider: Any, stocks: List[str]) -> Optional[dict]:
+    """若 provider 支持 ``get_security_info_batch`` 则一次性预取(2 条 SQL 替 N×2),
+    否则返回 None → 调用方回退逐只。返回非 dict(MagicMock 等)也回退。"""
+    fn = getattr(provider, "get_security_info_batch", None)
+    if fn is None or not callable(fn):
+        return None
+    try:
+        result = fn(stocks)
+    except Exception as exc:
+        logger.debug("get_security_info_batch 失败, 回退逐只: %s", exc)
+        return None
+    return result if isinstance(result, dict) else None
+
+
 def filter_st_stock(stocks: Iterable[str], provider: Any) -> List[str]:
     """过滤 ST/* /退市股:名字含 'ST' / '*' / '退'。
 
-    provider 用 ``get_security_info(code)`` 拿 ``display_name``。
+    provider 用 ``get_security_info(code)`` 拿 ``display_name``; 支持
+    ``get_security_info_batch`` 时批量预取(filters ST/次新通病提速)。
     取不到名字时**保留**该股(宁错过不误杀,反向错过只是少买)。
     """
+    stocks = list(stocks)
+    infos = _batch_security_info(provider, stocks)
     result: List[str] = []
     for stock in stocks:
-        info = _safe_call(provider.get_security_info, stock)
+        info = infos.get(stock) if infos is not None else _safe_call(
+            provider.get_security_info, stock
+        )
         if not info:
             result.append(stock)
             continue
@@ -100,13 +119,18 @@ def filter_new_stock(
 ) -> List[str]:
     """过滤次新股:上市 < ``days`` 天(默认 375 ≈ 1 年+少量缓冲)。
 
-    provider 用 ``get_security_info(code).start_date`` 拿上市日。
+    provider 用 ``get_security_info(code).start_date`` 拿上市日; 支持
+    ``get_security_info_batch`` 时批量预取(filters ST/次新通病提速)。
     ``today`` 接受 datetime/date/str(YYYY-MM-DD),聚宽风格 ``context.previous_date``。
     """
     today_dt = _coerce_datetime(today)
+    stocks = list(stocks)
+    infos = _batch_security_info(provider, stocks)
     result: List[str] = []
     for stock in stocks:
-        info = _safe_call(provider.get_security_info, stock)
+        info = infos.get(stock) if infos is not None else _safe_call(
+            provider.get_security_info, stock
+        )
         if not info:
             result.append(stock)
             continue

sanguo_portfolio/providers/local_unified_provider.py

@@ -636,6 +636,34 @@ class LocalUnifiedProvider(DataProvider):  # type: ignore[misc]
         """
         return self._get_lpp_helper().get_value_metrics(stock, date)
 
+    def get_value_metrics_batch(
+        self,
+        stocks: List[str],
+        date: Union[str, datetime],
+    ) -> Dict[str, Optional[Dict[str, Any]]]:
+        """批量 get_value_metrics(ThreadPool 并发逐只; 策略01 价值精选提速)。
+
+        逐只多期逻辑(ROE/FCF/流动比率/yoy, NOTICE_DATE<=date 前视过滤)委托
+        LocalParquetProvider.get_value_metrics 不变; 只把 N 次调用并发化。
+        多期聚合不宜向量化(300 只够用, 更深优化 YAGNI)。返回 {stock: metrics_or_None}。
+        """
+        if not stocks:
+            return {}
+        lpp = self._get_lpp_helper()
+
+        def _one(jq_code: str):
+            try:
+                return jq_code, lpp.get_value_metrics(jq_code, date)
+            except Exception as exc:
+                logger.debug("get_value_metrics(%s) 失败: %s", jq_code, exc)
+                return jq_code, None
+
+        if len(stocks) <= _FUND_POOL_THRESHOLD:
+            return dict(_one(s) for s in stocks)
+        workers = min(8, os.cpu_count() or 4)
+        with ThreadPoolExecutor(max_workers=workers) as ex:
+            return dict(ex.map(_one, stocks))
+
     def _build_fundamental_row(
         self, jq_code: str, date_str: str,
         need: Optional[Dict[str, bool]] = None,
@@ -746,6 +774,70 @@ class LocalUnifiedProvider(DataProvider):  # type: ignore[misc]
             "type": "stock",
         }
 
+    def get_security_info_batch(
+        self,
+        securities: List[str],
+        date: Optional[Union[str, datetime]] = None,
+    ) -> Dict[str, Dict[str, Any]]:
+        """批量 get_security_info: **2 条 SQL 替 N×2 逐只查询**(filters ST/次新通病提速)。
+
+        filters.filter_st_stock / filter_new_stock 逐只调 get_security_info, 每只 2 条
+        SQL(dbbardata min/max + constituent_unified name); 03 每日 10 行业×几百只、02/01
+        调仓千次 → 万次查询。本方法一次查全:
+        - dbbardata: ``symbol IN (...) GROUP BY symbol, exchange`` 拿 min/max(走索引)
+        - constituent_unified: ``code IN (...)`` 拿 code_name
+        返回 {jq_code: {code, display_name, name, start_date, end_date, type}}, 与逐只逐字一致。
+        """
+        if not securities:
+            return {}
+        pairs = [(s, jq_to_dbbardata(s)) for s in securities]
+        syms = list({sym for sym, _ in (p[1] for p in pairs)})
+        conn = self._connect()
+
+        # 1. dbbardata min/max per (symbol, exchange); chunk 防 >999 参数
+        minmax: Dict[tuple, tuple] = {}
+        for i in range(0, len(syms), 400):
+            chunk = syms[i:i + 400]
+            ph = ",".join("?" * len(chunk))
+            for r in conn.execute(
+                f"SELECT symbol, exchange, MIN(datetime), MAX(datetime) FROM dbbardata "
+                f"WHERE interval='d' AND symbol IN ({ph}) GROUP BY symbol, exchange",
+                chunk,
+            ):
+                minmax[(r[0], r[1])] = (
+                    r[2][:10] if r[2] else None,
+                    r[3][:10] if r[3] else None,
+                )
+
+        # 2. constituent_unified code_name(碰撞股名; 缺则回退 jq_code)
+        names: Dict[str, str] = {}
+        for i in range(0, len(syms), 400):
+            chunk = syms[i:i + 400]
+            ph = ",".join("?" * len(chunk))
+            try:
+                for r in conn.execute(
+                    f"SELECT code, code_name FROM constituent_unified WHERE code IN ({ph})",
+                    chunk,
+                ):
+                    if r[1]:
+                        names[r[0]] = str(r[1])
+            except sqlite3.Error:
+                pass
+
+        out: Dict[str, Dict[str, Any]] = {}
+        for jq_code, (sym, exc) in pairs:
+            start_dt, end_dt = minmax.get((sym, exc), (None, None))
+            name = names.get(sym, jq_code)
+            out[jq_code] = {
+                "code": jq_code,
+                "display_name": name,
+                "name": name,
+                "start_date": start_dt,
+                "end_date": end_dt,
+                "type": "stock",
+            }
+        return out
+
     def get_current_tick(self, security: str) -> Optional[Dict[str, Any]]:
         """dbbardata 最近 close + 高低涨停 ±10%(简化,ST/创业/科创精确规则 v2)。"""
         sym, exc = jq_to_dbbardata(security)

tests/portfolio/test_local_unified_provider.py

@@ -546,6 +546,83 @@ class TestGetFundamentalsFields:
         assert pd.isna(df.loc["000001.XSHE", "eps"])  # 缺失股票 eps NaN
 
 
+# ======================== Task 3c: get_security_info_batch / get_value_metrics_batch ========================
+def _make_security_info_fixture(tmp_path):
+    """dbbardata(3 只日线, 含碰撞 SSE/SZSE) + constituent_unified(含 1 ST) 供 batch 测试。"""
+    db = tmp_path / "s.db"
+    c = sqlite3.connect(str(db))
+    c.execute(
+        "CREATE TABLE dbbardata(symbol TEXT, exchange TEXT, datetime TEXT, "
+        "interval TEXT, close_price REAL)"
+    )
+    c.executemany("INSERT INTO dbbardata VALUES(?,?,?,?,?)", [
+        ("600519", "SSE", "2024-06-18 00:00:00", "d", 1500.0),
+        ("600519", "SSE", "2024-06-20 00:00:00", "d", 1510.0),
+        ("000001", "SZSE", "2023-01-03 00:00:00", "d", 12.0),
+        ("000001", "SZSE", "2024-06-20 00:00:00", "d", 11.0),
+        ("000002", "SZSE", "2024-06-19 00:00:00", "d", 8.0),
+    ])
+    c.execute("CREATE TABLE constituent_unified(code TEXT, code_name TEXT)")
+    c.executemany("INSERT INTO constituent_unified VALUES(?,?)", [
+        ("600519", "贵州茅台"), ("000001", "平安银行"), ("000002", "*ST某某"),
+    ])
+    c.commit()
+    c.close()
+    return db
+
+
+class TestGetSecurityInfoBatch:
+    """get_security_info_batch: 2 条 SQL 替 N×2 逐只(filters ST/次新通病)。"""
+
+    def test_batch_matches_per_stock(self, tmp_path):
+        db = _make_security_info_fixture(tmp_path)
+        p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
+        codes = ["600519.XSHG", "000001.XSHE", "000002.XSHE", "999999.XSHG"]
+        batch = p.get_security_info_batch(codes)
+        assert set(batch.keys()) == set(codes)
+        for code in codes:  # 核心回归: batch == 逐只
+            assert batch[code] == p.get_security_info(code), f"mismatch {code}"
+        assert batch["600519.XSHG"]["start_date"] == "2024-06-18"
+        assert batch["600519.XSHG"]["end_date"] == "2024-06-20"
+        assert "*ST" in batch["000002.XSHE"]["display_name"]  # 名字从 constituent_unified
+        assert batch["999999.XSHG"]["start_date"] is None  # 缺数据
+
+    def test_batch_empty(self, tmp_path):
+        db = _make_security_info_fixture(tmp_path)
+        p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
+        assert p.get_security_info_batch([]) == {}
+
+
+class TestGetValueMetricsBatch:
+    """get_value_metrics_batch: ThreadPool 并发逐只(策略01 价值精选), 与逐只一致。"""
+
+    @staticmethod
+    def _norm(d):
+        """NaN 容错规范化(NaN!=NaN 会让 dict==False; 用 'NaN' 占位)。"""
+        if d is None:
+            return None
+        out = {}
+        for k, v in d.items():
+            if isinstance(v, list):
+                out[k] = ["NaN" if (isinstance(x, float) and x != x) else x for x in v]
+            elif isinstance(v, float) and v != v:
+                out[k] = "NaN"
+            else:
+                out[k] = v
+        return out
+
+    def test_batch_matches_per_stock(self, tmp_path):
+        db = _make_fundamentals_fixture(tmp_path)
+        p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
+        codes = ["600519.XSHG", "999999.XSHG"]
+        batch = p.get_value_metrics_batch(codes, date="2024-09-30")
+        assert set(batch.keys()) == set(codes)
+        for code in codes:  # 核心回归: batch == 逐只(多期 dict, NaN 容错)
+            assert self._norm(batch[code]) == self._norm(
+                p.get_value_metrics(code, date="2024-09-30")
+            ), code
+
+
 # ======================== Task 4: 辅助方法 ========================
 class TestAuxMethods:
     def test_get_trade_days_from_dbbardata(self, unified_provider):