feat(portfolio): get_fundamentals_df 批量提速(fields= 按需短路 + ThreadPool 并发)

策略02 _pick_stocks 对 5128 只按 market_cap+eps 排序, 旧实现逐只读 4 表(valuation+
income+balance+financial_abstract)+算 calc_roic, 首仓全 cache miss 卡死。

优化(就地, 策略代码零改动, fields=None 向后兼容):
- fields= 参数: 只读请求字段依赖的源表(策略02 只要 market_cap+eps → 跳 balance/
  financial_abstract/roic, 省一半 parquet 读)
- ThreadPool 并发逐只(>64 只; 本地文件 I/O 非 baostock 网络, 不触不并发铁律)
- _build_fundamental_row 加 need= 守卫读取(lpp + unified 两层)

Mac TDD 4 测试(子集/短路/回归/并发保序)全绿, 124 回归通过。
注: 策略02 要拿满提速需在其 get_fundamentals_df 调用加 fields=[market_cap,eps](策略层, 归策略session)

sanguo_portfolio/providers/local_parquet_provider.py

@@ -564,15 +564,31 @@ class LocalParquetProvider(DataProvider):  # type: ignore[misc]
             df = df.set_index("code", drop=False)
         return df
 
-    def _build_fundamental_row(self, jq_code: str, date_str: str) -> Dict[str, Any]:
+    def _build_fundamental_row(
+        self, jq_code: str, date_str: str,
+        need: Optional[Dict[str, bool]] = None,
+    ) -> Dict[str, Any]:
         fc = jq_to_file_code(jq_code)
         row: Dict[str, Any] = {"code": jq_code}
+        n = need or {}
 
-        val = self._latest_row_before(self._read_valuation(fc), "date", date_str)
+        def want(key: str) -> bool:
+            return n.get(key, True)  # 未传 need({})→全读(向后兼容)
+
+        val = (
+            self._latest_row_before(self._read_valuation(fc), "date", date_str)
+            if want("akshare_val") else None
+        )
         # income/balance 取最新**已披露年报**(NOTICE_DATE<=date, REPORT_TYPE 含"年"):
         # 修复旧按 REPORT_DATE 过滤的前视偏差(用了未披露年报) + 年报口径跨股可比
-        inc = self._latest_published_annual(self._read_quarter("income", fc), date_str)
+        inc = (
-        bal = self._latest_published_annual(self._read_quarter("balance", fc), date_str)
+            self._latest_published_annual(self._read_quarter("income", fc), date_str)
+            if want("income") else None
+        )
+        bal = (
+            self._latest_published_annual(self._read_quarter("balance", fc), date_str)
+            if want("balance") else None
+        )
 
         def g(d: Optional[pd.Series], k: str) -> Optional[float]:
             return _to_float(d.get(k)) if d is not None else None
@@ -626,7 +642,7 @@ class LocalParquetProvider(DataProvider):  # type: ignore[misc]
         )
 
         # gross_profit_margin: 从 financial_abstract 读现成"毛利率"(百分数→小数)
-        fa = self._read_financial_abstract(fc)
+        fa = self._read_financial_abstract(fc) if want("fa") else None
         row["gross_profit_margin"] = _pct_to_decimal(self._latest_indicator(fa, "毛利率"))
         # roic = NOPAT / (归母权益 + 有息负债 - 货币资金)
         # actual_tax_rate akshare 无现成指标, 传 None 让 calc_roic 用 inc_tax/total_profit 兜底

sanguo_portfolio/providers/local_unified_provider.py

@@ -15,6 +15,7 @@ import logging
 import os
 import re
 import sqlite3
+from concurrent.futures import ThreadPoolExecutor
 from datetime import datetime
 from typing import Any, Dict, List, Optional, Union
 
@@ -43,6 +44,10 @@ _EXC_TO_JQ_SUFFIX = {"SSE": "XSHG", "SZSE": "XSHE"}
 _DATE_RE = re.compile(r"^\d{4}-\d{2}-\d{2}$")
 _INTERVAL_RE = re.compile(r"^[A-Za-z0-9_]+$")
 
+# get_fundamentals_df 并发阈值: 超过则 ThreadPool 并发逐只(本地文件 I/O, 安全)
+# 小列表(all_weather 多为单只/[stock])走顺序, 避线程池开销; 策略 02 全市场(5128)走并发
+_FUND_POOL_THRESHOLD = 64
+
 
 def _safe_date_literal(s: str) -> str:
     """``"2022-01-01"`` → ``"'2022-01-01'"`` (SQL 安全字面量, 用于 UNION ALL 注入)。
@@ -537,10 +542,48 @@ class LocalUnifiedProvider(DataProvider):  # type: ignore[misc]
             self._val_bs_cache[year] = pd.DataFrame()
             return pd.DataFrame()
 
+    # fields → 源表依赖(短路: 未请求字段其源表不读, 省 parquet 读 + calc_roic)
+    # akshare_val: static/valuation(市值 + akshare pe/pb fallback)
+    # bs_val:      valuation_baostock(baostock 覆盖 pe/pb/ps/pcf)
+    # income:      static/income(eps/yoy/net_profit→roe/roa/roic/margin)
+    # balance:     static/balance(total_liability/权益/资产→roe/roa/roic)
+    # fa:          static/financial_abstract(gross_profit_margin)
+    _FIELDS_AKSHARE_VAL = frozenset({
+        "market_cap", "circulating_market_cap",
+        "pe_ratio", "pb_ratio", "ps_ratio", "pcf_ratio",
+    })
+    _FIELDS_BS_VAL = frozenset({"pe_ratio", "pb_ratio", "ps_ratio", "pcf_ratio"})
+    _FIELDS_INCOME = frozenset({
+        "eps", "inc_revenue_year_on_year", "inc_operation_profit_year_on_year",
+        "inc_total_revenue_year_on_year", "net_profit_margin", "roe", "roa", "roic",
+    })
+    _FIELDS_BALANCE = frozenset({
+        "total_liability", "total_sheet_owner_equities", "retained_profit",
+        "roe", "roa", "roic",
+    })
+    _FIELDS_FA = frozenset({"gross_profit_margin"})
+
+    @staticmethod
+    def _fields_to_need(fields: Optional[List[str]]) -> Dict[str, bool]:
+        """fields 列表 → 各源表是否需读。``fields=None`` 时调用方不调本方法(走全读)。"""
+        if not fields:
+            return {"akshare_val": True, "bs_val": True, "income": True,
+                    "balance": True, "fa": True}
+        fs = set(fields)
+        cls = LocalUnifiedProvider
+        return {
+            "akshare_val": bool(fs & cls._FIELDS_AKSHARE_VAL),
+            "bs_val": bool(fs & cls._FIELDS_BS_VAL),
+            "income": bool(fs & cls._FIELDS_INCOME),
+            "balance": bool(fs & cls._FIELDS_BALANCE),
+            "fa": bool(fs & cls._FIELDS_FA),
+        }
+
     def get_fundamentals_df(
         self,
         stocks: List[str],
         date: Optional[Union[str, datetime]] = None,
+        fields: Optional[List[str]] = None,
     ) -> pd.DataFrame:
         """合并多股 fundamentals, 列对齐 ``_FUNDAMENTAL_COLUMNS``。
 
@@ -550,17 +593,35 @@ class LocalUnifiedProvider(DataProvider):  # type: ignore[misc]
           (baostock valuation 无市值列)
         - 三表(eps/yoy/total_liability/...) ← ``static/{income,balance}`` akshare
           (委托 LocalParquetProvider 读, DRY)
+
+        性能(2026-07-28 批量优化, 解锁策略 02 全市场 5128 只选股):
+        - ``fields=`` 按需短路: 只读请求字段依赖的源表(策略 02 只要 market_cap+eps
+          → 跳 balance/financial_abstract/roic, 省一半 parquet 读 + 跳 calc_roic)。
+        - ``len(stocks) > _FUND_POOL_THRESHOLD`` 时 ThreadPool 并发逐只读(本地文件 I/O,
+          非 baostock 网络 → 并发安全, 不触"baostock 不并发"铁律)。
+        ``fields=None`` 全列(向后兼容, all_weather/small_cap 现有调用零改动)。
         """
         from .local_parquet_provider import _FUNDAMENTAL_COLUMNS, jq_to_file_code
         if not stocks:
             return pd.DataFrame(columns=_FUNDAMENTAL_COLUMNS)
         date_str = self._to_date_str(date) or datetime.now().strftime("%Y-%m-%d")
-        rows: List[Dict[str, Any]] = [
+        need = self._fields_to_need(fields) if fields else None
-            self._build_fundamental_row(s, date_str) for s in stocks
+
-        ]
+        def _one(jq_code: str) -> Dict[str, Any]:
+            return self._build_fundamental_row(jq_code, date_str, need)
+
+        if len(stocks) <= _FUND_POOL_THRESHOLD:
+            rows: List[Dict[str, Any]] = [_one(s) for s in stocks]
+        else:
+            workers = min(8, os.cpu_count() or 4)
+            with ThreadPoolExecutor(max_workers=workers) as ex:
+                rows = list(ex.map(_one, stocks))
         df = pd.DataFrame(rows, columns=_FUNDAMENTAL_COLUMNS)
         if "code" in df.columns:
             df = df.set_index("code", drop=False)
+        if fields:
+            keep = ["code"] + [f for f in fields if f in df.columns]
+            df = df[keep]
         return df
 
     def get_value_metrics(
@@ -577,8 +638,12 @@ class LocalUnifiedProvider(DataProvider):  # type: ignore[misc]
 
     def _build_fundamental_row(
         self, jq_code: str, date_str: str,
+        need: Optional[Dict[str, bool]] = None,
     ) -> Dict[str, Any]:
-        """单股 fundamentals 行: baostock 估值覆盖 akshare pe/pb, 市值+三表用 LocalParquetProvider。"""
+        """单股 fundamentals 行: baostock 估值覆盖 akshare pe/pb, 市值+三表用 LocalParquetProvider。
+
+        need: 源表短路字典(None=全读); 由 get_fundamentals_df(fields=) 构造。
+        """
         from .local_parquet_provider import (
             jq_to_file_code, _to_float, _or_nan, _pct_to_decimal,
         )
@@ -587,15 +652,17 @@ class LocalUnifiedProvider(DataProvider):  # type: ignore[misc]
         sym, _ = jq_to_dbbardata(jq_code)
         fc = jq_to_file_code(jq_code)
 
-        # 先委托 LocalParquetProvider 取整行(akshare 全套)
+        # 先委托 LocalParquetProvider 取整行(akshare 全套, 按 need 短路读哪些表)
         lpp = self._get_lpp_helper()
-        ak_row = lpp._build_fundamental_row(jq_code, date_str)
+        ak_row = lpp._build_fundamental_row(jq_code, date_str, need)
 
         # 复制 akshare 行(市值/eps/三表/yoy/...), 然后用 baostock 覆盖 pe/pb/ps/pcf
         row: Dict[str, Any] = dict(ak_row)
         row["code"] = jq_code
 
-        # baostock 估值覆盖 pe/pb/ps/pcf
+        # baostock 估值覆盖 pe/pb/ps/pcf(按 need['bs_val'] 短路; 未请求则跳)
+        if need is not None and not need.get("bs_val", True):
+            return row
         year = int(date_str[:4])
         vbs = self._read_valuation_baostock(year)
         vrow = None

tests/portfolio/test_local_unified_provider.py

@@ -474,6 +474,78 @@ class TestGetFundamentals:
         assert df.empty
 
 
+# ======================== Task 3b: get_fundamentals_df fields= 按需短路 + 并发 ========================
+class TestGetFundamentalsFields:
+    """fields= 只读必需表(跳 balance/financial_abstract/roic) + ThreadPool 并发,零 VPS 回归。
+
+    动机:策略 02 _pick_stocks 对 5128 只按 market_cap+eps 排序,旧实现逐只读 4 表 + 算 roic
+    → 首仓卡死。fields=['market_cap','eps'] 只读 valuation+income,并发逐只。
+    """
+
+    def test_fields_subset_columns_and_values(self, tmp_path):
+        db = _make_fundamentals_fixture(tmp_path)
+        p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
+        df = p.get_fundamentals_df(
+            ["600519.XSHG"], date="2024-09-30", fields=["market_cap", "eps"]
+        )
+        # 只返 code + 请求列
+        assert list(df.columns) == ["code", "market_cap", "eps"]
+        assert abs(df.loc["600519.XSHG", "market_cap"] - 2e4) < 1
+        assert abs(df.loc["600519.XSHG", "eps"] - 41.0) < 1e-6
+
+    def test_fields_skips_unneeded_tables(self, tmp_path, monkeypatch):
+        # fields=['market_cap','eps'] 只需 valuation(akshare)+income → 不读 balance/financial_abstract
+        from sanguo_portfolio.providers import local_parquet_provider as lpp_mod
+
+        db = _make_fundamentals_fixture(tmp_path)
+        p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
+        orig_q = lpp_mod.LocalParquetProvider._read_quarter
+        orig_fa = lpp_mod.LocalParquetProvider._read_financial_abstract
+        tables_read: List[str] = []
+        fa_called = {"v": False}
+
+        def spy_q(self, table, file_code):  # noqa: ANN001
+            tables_read.append(table)
+            return orig_q(self, table, file_code)
+
+        def spy_fa(self, file_code):  # noqa: ANN001
+            fa_called["v"] = True
+            return orig_fa(self, file_code)
+
+        monkeypatch.setattr(lpp_mod.LocalParquetProvider, "_read_quarter", spy_q)
+        monkeypatch.setattr(lpp_mod.LocalParquetProvider, "_read_financial_abstract", spy_fa)
+
+        df = p.get_fundamentals_df(
+            ["600519.XSHG"], date="2024-09-30", fields=["market_cap", "eps"]
+        )
+        assert "income" in tables_read
+        assert "balance" not in tables_read
+        assert fa_called["v"] is False
+        assert abs(df.loc["600519.XSHG", "eps"] - 41.0) < 1e-6
+
+    def test_fields_none_backward_compat(self, tmp_path):
+        # fields=None 仍返全部 _FUNDAMENTAL_COLUMNS(回归: roe/roa/roic/gross_profit_margin 仍在)
+        db = _make_fundamentals_fixture(tmp_path)
+        p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
+        df = p.get_fundamentals_df(["600519.XSHG"], date="2024-09-30")
+        for col in ("roe", "roa", "roic", "gross_profit_margin", "total_liability"):
+            assert col in df.columns
+
+    def test_threadpool_preserves_values_and_order(self, tmp_path, monkeypatch):
+        # >阈值触发 ThreadPool: 70 只(1 有数据 + 69 缺失)→ 值正确 + 顺序保持
+        import sanguo_portfolio.providers.local_unified_provider as up
+
+        monkeypatch.setattr(up, "_FUND_POOL_THRESHOLD", 1)
+        db = _make_fundamentals_fixture(tmp_path)
+        p = LocalUnifiedProvider({"db_path": str(db), "data_dir": str(tmp_path)})
+        stocks = ["600519.XSHG"] + [f"00000{i}.XSHE" for i in range(1, 70)]
+        df = p.get_fundamentals_df(stocks, date="2024-09-30")
+        assert len(df) == 70
+        assert df.index[0] == "600519.XSHG"  # ex.map 保序
+        assert abs(df.loc["600519.XSHG", "eps"] - 41.0) < 1e-6
+        assert pd.isna(df.loc["000001.XSHE", "eps"])  # 缺失股票 eps NaN
+
+
 # ======================== Task 4: 辅助方法 ========================
 class TestAuxMethods:
     def test_get_trade_days_from_dbbardata(self, unified_provider):