From 2197feac335becb0eaef2468629468779dc99187 Mon Sep 17 00:00:00 2001 From: claude_dev Date: Fri, 21 Aug 2026 22:58:57 +0800 Subject: [PATCH] =?UTF-8?q?fix(strategy):=20=E5=AE=9E=E7=9B=98=E9=80=89?= =?UTF-8?q?=E8=82=A1=E6=8C=82=E6=AD=BB=E6=A0=B9=E6=B2=BB=E2=80=94=E2=80=94?= =?UTF-8?q?=E8=AF=81=E5=88=B8=E4=BF=A1=E6=81=AF=E6=89=B9=E9=87=8F=E5=A7=94?= =?UTF-8?q?=E6=89=98=E6=9C=AC=E5=9C=B0=E5=BA=93=E6=9B=BF600=E6=AC=A1xtdata?= =?UTF-8?q?=20RPC=20+=20=E9=80=89=E8=82=A1=E9=98=B6=E6=AE=B5=E8=80=97?= =?UTF-8?q?=E6=97=B6=E6=97=A5=E5=BF=97=20[vps]?= MIME-Version: 1.0 Content-Type: text/plain; charset=UTF-8 Content-Transfer-Encoding: 8bit 2026-08-21 实锤(live_19/shadow_60 同卡):momentum 9:30:06 buy_sign=True 后 47min 零输出被重启;small_cap 同型(fundamentals 超时120s后沉默)。channel_test 正常(不走走filters重取数)。value_selection/all_weather 月度未到期=零成交正常。 根因(VPS离场时段探针27s跑完全路径反推):filters.filter_st/filter_new 逐只 get_security_info,每只2次xtdata RPC(_detect_instrument_type+get_instrument_detail) 且无超时——momentum 10指数×30只×2过滤=600次RPC,开盘时段QMT忙时47min+不归; provider无get_security_info_batch→filters._batch_security_info回退逐只(风暴路径)。 修法: - SanguoMiniQmtProvider.get_security_info_batch:委托内部LocalUnifiedProvider (2条本地SQL替N×2次RPC,与78d35ba get_closes_panel/get_constituent_ex同款 委托模式);失败/非dict返None→filters自动回退逐只(不崩策略);全部策略 (aw/mom/sc/vs含_ex)经共享filters即时受益,影子/回测同速 - momentum/_ex+small_cap/_ex 选股段加阶段耗时日志(池预取/宽表/过滤3段, 下个9:30若仍有卡点日志直接钉死位置) - 遗留已报issue:SanguoMiniQmtProvider无get_limit_status_batch→实盘涨跌停/ 停牌filter静默失效(涨停股照买),归数据session按本地公式+单次批量tick根治 测试:+5(委托/失败回退None/非dict回退/filters消费批量零逐只RPC×2,__new__ 绕过Mac无bullet_trade的构造raise);portfolio 380绿(13失败全=Mac缺bullet_trade 环境性,NAS CI过) [vps] --- .../providers/sanguo_fundamentals.py | 24 ++++++ .../strategies/momentum_timing.py | 10 +++ .../strategies/momentum_timing_ex.py | 7 ++ sanguo_portfolio/strategies/small_cap.py | 10 +++ sanguo_portfolio/strategies/small_cap_ex.py | 10 +++ .../portfolio/test_provider_secinfo_batch.py | 79 +++++++++++++++++++ 6 files changed, 140 insertions(+) create mode 100644 tests/portfolio/test_provider_secinfo_batch.py diff --git a/sanguo_portfolio/providers/sanguo_fundamentals.py b/sanguo_portfolio/providers/sanguo_fundamentals.py index 8b2f8a1..2c12cd6 100644 --- a/sanguo_portfolio/providers/sanguo_fundamentals.py +++ b/sanguo_portfolio/providers/sanguo_fundamentals.py @@ -162,6 +162,30 @@ class SanguoMiniQmtProvider(MiniQMTProvider): # type: ignore[misc] """TET 别名:与 ``get_index_stocks`` 同一实现(_ex 策略副本调用名)。""" return self.get_index_stocks(index, date) + def get_security_info_batch( + self, + securities: List[str], + date: Optional[Union[str, datetime]] = None, + ) -> Optional[Dict[str, Dict[str, Any]]]: + """批量证券信息:委托本地库 2 条 SQL,替 N×2 次逐只 xtdata RPC。 + + 2026-08-21 实盘选股挂死根因:filters.filter_st/filter_new 逐只 + ``get_security_info``,每次 2 个 xtdata RPC(_detect_instrument_type + + get_instrument_detail)且无超时——开盘时段 QMT 忙,momentum 600 次 RPC + 47min+ 不归(live_19/shadow_60 同卡,离场时段同路径全程 27s)。 + filters._batch_security_info 探测到本方法即一次批量预取。 + + 委托失败/返非 dict → 返回 None,调用方(filters)自动回退逐只旧路径。 + ⚠️ display_name 取 constituent_unified 历史名,极端滞后(近期刚戴帽 + ST)可能漏过滤——沿用 filters「宁错过不误杀」口径,可接受。 + """ + try: + out = self._unified.get_security_info_batch(securities, date) + except Exception as exc: # noqa: BLE001 - 委托失败回退逐只,不崩策略 + logger.warning("get_security_info_batch 委托本地库失败,回退逐只: %s", exc) + return None + return out if isinstance(out, dict) else None + # ------------------------ 主入口 ------------------------ def get_fundamentals( self, diff --git a/sanguo_portfolio/strategies/momentum_timing.py b/sanguo_portfolio/strategies/momentum_timing.py index 83db53f..0f2405f 100644 --- a/sanguo_portfolio/strategies/momentum_timing.py +++ b/sanguo_portfolio/strategies/momentum_timing.py @@ -24,6 +24,7 @@ from __future__ import annotations import datetime import logging +import time from dataclasses import dataclass, field from typing import Any, List, Optional @@ -168,10 +169,17 @@ class MomentumTimingStrategy: # 不吞异常退化成"空目标→全清仓"(2026-08-19 假熊市同型事故) try: # 牛市:当日预取十行业成份股池 close 宽表(层2向量化,每日 1 次批量 SQL) + # 阶段耗时日志(2026-08-21 实盘选股挂死定位用:9:30 buy_sign 后 47min + # 无下文,靠分段日志下个开盘直接钉死卡点) + _t0 = time.monotonic() union_stocks: List[str] = [] for each_index in cfg.index_list: union_stocks.extend(self._stock_pool_cached(each_index, cur_date)) + logger.info("[%s] 选股1/3 池预取完成: %d 只 %.1fs", + cur_date, len(union_stocks), time.monotonic() - _t0) self._ensure_day_panel(union_stocks, cur_date) + logger.info("[%s] 选股2/3 close宽表预取完成 %.1fs", + cur_date, time.monotonic() - _t0) # 取强舍弱(每行业 RPS top_k 并集) → 候选池 candidates = self._find_stock_pool(cfg.index_list, cur_date, pre_date) @@ -199,6 +207,8 @@ class MomentumTimingStrategy: stocks, self.provider, status_map=status_map, ) stocks = _dedup(stocks) + logger.info("[%s] 选股3/3 过滤完成: %d 只候选 %.1fs", + cur_date, len(stocks), time.monotonic() - _t0) except Exception as exc: logger.warning( "[%s] 选股数据失败,跳过当日调仓(持仓不动,不清仓): %s", cur_date, exc, diff --git a/sanguo_portfolio/strategies/momentum_timing_ex.py b/sanguo_portfolio/strategies/momentum_timing_ex.py index 40b1da1..86308fe 100644 --- a/sanguo_portfolio/strategies/momentum_timing_ex.py +++ b/sanguo_portfolio/strategies/momentum_timing_ex.py @@ -27,6 +27,7 @@ from __future__ import annotations import datetime import logging +import time from dataclasses import dataclass, field from typing import Any, List, Optional @@ -163,8 +164,12 @@ class MomentumTimingExStrategy: # 2~5) 牛市选股:任一取数失败 → 跳过当日调仓(持仓不动), # 不吞异常退化成"空目标→全清仓"(2026-08-19 假熊市同型事故) try: + # 阶段耗时日志(2026-08-21 实盘选股挂死定位用,与主文件同款) + _t0 = time.monotonic() # 牛市:取强舍弱(每行业 RPS top_k 并集) → 候选池 candidates = self._find_stock_pool(cfg.index_list, cur_date, pre_date) + logger.info("[%s] 选股1/2 候选池(RPS)完成: %d 只 %.1fs", + cur_date, len(candidates), time.monotonic() - _t0) # 均线动量过滤(close > MA_short > MA_long) stocks = self._select_stocks(candidates, cur_date) @@ -189,6 +194,8 @@ class MomentumTimingExStrategy: stocks, self.provider, status_map=status_map, ) stocks = _dedup(stocks) + logger.info("[%s] 选股2/2 过滤完成: %d 只候选 %.1fs", + cur_date, len(stocks), time.monotonic() - _t0) except Exception as exc: logger.warning( "[%s] 选股数据失败,跳过当日调仓(持仓不动,不清仓): %s", cur_date, exc, diff --git a/sanguo_portfolio/strategies/small_cap.py b/sanguo_portfolio/strategies/small_cap.py index 30b9389..475aabb 100644 --- a/sanguo_portfolio/strategies/small_cap.py +++ b/sanguo_portfolio/strategies/small_cap.py @@ -30,6 +30,7 @@ from __future__ import annotations import datetime import logging +import time from dataclasses import dataclass from typing import Any, List, Optional @@ -198,8 +199,12 @@ class SmallCapStrategy: logger.warning("pick_stocks: previous_date 为 None,跳过本次选股") return None + # 阶段耗时日志(2026-08-21 实盘选股挂死定位用) + _t0 = time.monotonic() # 1) 全市场候选池(universe 成份股) candidates = self._stock_pool(cfg.universe, previous_date) + logger.info("[%s] 选股1/3 候选池完成: %d 只 %.1fs", + previous_date, len(candidates or []), time.monotonic() - _t0) if candidates is None: return None if not candidates: @@ -240,6 +245,9 @@ class SmallCapStrategy: if not top_candidates: return [] + logger.info("[%s] 选股2/3 fundamentals 完成: top %d 只 %.1fs", + previous_date, len(top_candidates), time.monotonic() - _t0) + # 5) 过滤次新股(原策略上市 > 120 天) top_candidates = filters.filter_new_stock( top_candidates, self.provider, previous_date, cfg.new_stock_days, @@ -274,6 +282,8 @@ class SmallCapStrategy: # 8) 取前 buy_stock_count out = list(scored.index)[: cfg.buy_stock_count] + logger.info("[%s] 选股3/3 动量评分完成: %d 只 %.1fs", + previous_date, len(out), time.monotonic() - _t0) return out # =================== 动量评分 =================== diff --git a/sanguo_portfolio/strategies/small_cap_ex.py b/sanguo_portfolio/strategies/small_cap_ex.py index 60ee4a2..a76d704 100644 --- a/sanguo_portfolio/strategies/small_cap_ex.py +++ b/sanguo_portfolio/strategies/small_cap_ex.py @@ -33,6 +33,7 @@ from __future__ import annotations import datetime import logging +import time from dataclasses import dataclass from typing import Any, List, Optional @@ -201,8 +202,12 @@ class SmallCapExStrategy: logger.warning("pick_stocks: previous_date 为 None,跳过本次选股") return None + # 阶段耗时日志(2026-08-21 实盘选股挂死定位用) + _t0 = time.monotonic() # 1) 全市场候选池(universe 成份股) candidates = self._stock_pool(cfg.universe, previous_date) + logger.info("[%s] 选股1/3 候选池完成: %d 只 %.1fs", + previous_date, len(candidates or []), time.monotonic() - _t0) if candidates is None: return None if not candidates: @@ -243,6 +248,9 @@ class SmallCapExStrategy: if not top_candidates: return [] + logger.info("[%s] 选股2/3 fundamentals 完成: top %d 只 %.1fs", + previous_date, len(top_candidates), time.monotonic() - _t0) + # 5) 过滤次新股(原策略上市 > 120 天) top_candidates = filters.filter_new_stock( top_candidates, self.provider, previous_date, cfg.new_stock_days, @@ -277,6 +285,8 @@ class SmallCapExStrategy: # 8) 取前 buy_stock_count out = list(scored.index)[: cfg.buy_stock_count] + logger.info("[%s] 选股3/3 动量评分完成: %d 只 %.1fs", + previous_date, len(out), time.monotonic() - _t0) return out # =================== 动量评分 =================== diff --git a/tests/portfolio/test_provider_secinfo_batch.py b/tests/portfolio/test_provider_secinfo_batch.py new file mode 100644 index 0000000..d45b4ab --- /dev/null +++ b/tests/portfolio/test_provider_secinfo_batch.py @@ -0,0 +1,79 @@ +"""SanguoMiniQmtProvider.get_security_info_batch 委托测试(2026-08-21 挂死根治)。 + +背景:实盘 momentum/small_cap 9:30 选股挂死——filters.filter_st/filter_new 逐只 +get_security_info,每只 2 次 xtdata RPC 无超时,开盘时段 600 次 RPC 47min+ 不归。 +本方法委托内部 LocalUnifiedProvider 批量(2 条本地 SQL),filters 探测到即走批量。 + +Mac 无 bullet_trade:类定义处 MiniQMTProvider=object 兜底,可直接实例化; +_unified 用 MagicMock 替换,只测委托语义(不连本地库)。 +""" +from __future__ import annotations + +from unittest.mock import MagicMock + +from sanguo_portfolio import SanguoMiniQmtProvider +from sanguo_portfolio import filters + + +def _provider(unified: MagicMock) -> SanguoMiniQmtProvider: + # __new__ 绕过 __init__:Mac 无 bullet_trade 时构造直接 raise + # (RuntimeError);被测方法只用 self._unified,委托语义不依赖基类初始化。 + p = object.__new__(SanguoMiniQmtProvider) + p._unified = unified + return p + + +class TestGetSecurityInfoBatch: + def test_delegates_to_unified(self): + unified = MagicMock(name="unified") + expected = { + "600519.XSHG": {"display_name": "贵州茅台", "start_date": "2001-08-27"}, + "600036.XSHG": {"display_name": "招商银行", "start_date": "2002-04-09"}, + } + unified.get_security_info_batch.return_value = expected + p = _provider(unified) + out = p.get_security_info_batch(["600519.XSHG", "600036.XSHG"]) + assert out == expected + unified.get_security_info_batch.assert_called_once_with( + ["600519.XSHG", "600036.XSHG"], None) + + def test_unified_failure_returns_none_not_raise(self): + """委托失败 → None(不崩策略),filters 自动回退逐只旧路径。""" + unified = MagicMock(name="unified") + unified.get_security_info_batch.side_effect = RuntimeError("db locked") + p = _provider(unified) + assert p.get_security_info_batch(["600519.XSHG"]) is None + + def test_non_dict_result_returns_none(self): + unified = MagicMock(name="unified") + unified.get_security_info_batch.return_value = ["not", "a", "dict"] + p = _provider(unified) + assert p.get_security_info_batch(["600519.XSHG"]) is None + + +class TestFiltersConsumeBatch: + def test_filter_st_uses_batch_and_skips_per_stock_rpc(self): + """filters 探测到批量方法 → 一次预取,绝不再逐只 get_security_info + (即挂死根因的 RPC 风暴路径)。""" + provider = MagicMock(name="provider") + provider.get_security_info_batch.return_value = { + "600519.XSHG": {"display_name": "贵州茅台", "start_date": "2001-08-27"}, + "600000.XSHG": {"display_name": "ST浦发", "start_date": "1999-11-10"}, + } + out = filters.filter_st_stock(["600519.XSHG", "600000.XSHG"], provider) + assert out == ["600519.XSHG"] # ST 被滤掉 + provider.get_security_info_batch.assert_called_once() + provider.get_security_info.assert_not_called() # 零逐只 RPC + + def test_filter_new_stock_uses_batch(self): + provider = MagicMock(name="provider") + provider.get_security_info_batch.return_value = { + "600519.XSHG": {"display_name": "贵州茅台", + "start_date": "2001-08-27"}, + "301999.XSHE": {"display_name": "次新股", + "start_date": "2026-08-01"}, + } + out = filters.filter_new_stock( + ["600519.XSHG", "301999.XSHE"], provider, "2026-08-21", days=375) + assert out == ["600519.XSHG"] # 上市 20 天的次新被滤掉 + provider.get_security_info.assert_not_called()