From d2cd8fa94511aa8eb369f672f2b66d44a30aff6b Mon Sep 17 00:00:00 2001 From: claude_dev Date: Wed, 29 Jul 2026 08:23:21 +0800 Subject: [PATCH] =?UTF-8?q?feat(portfolio):=20get=5Fsecurity=5Finfo=5Fbatc?= =?UTF-8?q?h=20+=20get=5Fvalue=5Fmetrics=5Fbatch=20=E6=89=B9=E9=87=8F?= =?UTF-8?q?=E6=8E=A5=E5=8F=A3?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 策略层提速第三轮(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 | 32 ++++++- .../providers/local_unified_provider.py | 92 +++++++++++++++++++ .../portfolio/test_local_unified_provider.py | 77 ++++++++++++++++ 3 files changed, 197 insertions(+), 4 deletions(-) diff --git a/sanguo_portfolio/filters.py b/sanguo_portfolio/filters.py index 6a69bd4..fcb425e 100644 --- a/sanguo_portfolio/filters.py +++ b/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 diff --git a/sanguo_portfolio/providers/local_unified_provider.py b/sanguo_portfolio/providers/local_unified_provider.py index ce6a88f..65b2f91 100644 --- a/sanguo_portfolio/providers/local_unified_provider.py +++ b/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) diff --git a/tests/portfolio/test_local_unified_provider.py b/tests/portfolio/test_local_unified_provider.py index a945210..e6a495c 100644 --- a/tests/portfolio/test_local_unified_provider.py +++ b/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):