From f416a17b6d8e922c5d31d5e9480f61b5dcc098f2 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Tue, 28 Jul 2026 23:31:13 +0800 Subject: [PATCH] =?UTF-8?q?feat(portfolio):=20get=5Ffundamentals=5Fdf=20?= =?UTF-8?q?=E6=89=B9=E9=87=8F=E6=8F=90=E9=80=9F(fields=3D=20=E6=8C=89?= =?UTF-8?q?=E9=9C=80=E7=9F=AD=E8=B7=AF=20+=20ThreadPool=20=E5=B9=B6?= =?UTF-8?q?=E5=8F=91)?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 策略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) --- .../providers/local_parquet_provider.py | 26 ++++-- .../providers/local_unified_provider.py | 81 +++++++++++++++++-- .../portfolio/test_local_unified_provider.py | 72 +++++++++++++++++ 3 files changed, 167 insertions(+), 12 deletions(-) diff --git a/sanguo_portfolio/providers/local_parquet_provider.py b/sanguo_portfolio/providers/local_parquet_provider.py index 9ce8b0d..877921a 100644 --- a/sanguo_portfolio/providers/local_parquet_provider.py +++ b/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) - bal = self._latest_published_annual(self._read_quarter("balance", fc), date_str) + inc = ( + 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 兜底 diff --git a/sanguo_portfolio/providers/local_unified_provider.py b/sanguo_portfolio/providers/local_unified_provider.py index 2b76a06..ce6a88f 100644 --- a/sanguo_portfolio/providers/local_unified_provider.py +++ b/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]] = [ - self._build_fundamental_row(s, date_str) for s in stocks - ] + need = self._fields_to_need(fields) if fields else None + + 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 diff --git a/tests/portfolio/test_local_unified_provider.py b/tests/portfolio/test_local_unified_provider.py index 628f032..a945210 100644 --- a/tests/portfolio/test_local_unified_provider.py +++ b/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):