diff --git a/sanguo_portfolio/providers/sanguo_fundamentals.py b/sanguo_portfolio/providers/sanguo_fundamentals.py index 057a4a1..f495ddb 100644 --- a/sanguo_portfolio/providers/sanguo_fundamentals.py +++ b/sanguo_portfolio/providers/sanguo_fundamentals.py @@ -57,6 +57,12 @@ except ImportError as _e: # Mac dev 环境可能未装,允许模块加载 from .local_unified_provider import LocalUnifiedProvider +# unified 本地 parquet 可供的 fundamentals 列(2026-08-26 整步本地化白名单): +# fields 给定且全落此集合 → get_fundamentals_df 直接委托 unified, 跳过 xt。 +# 白名单外列(roic/pe_ratio/... xt 口径更全)仍走老路径, 不做半本地半 xt 混合。 +_LOCAL_FUND_FIELDS = frozenset( + {"market_cap", "circulating_market_cap", "eps"}) + # 聚宽 valuation/indicator/balance 列名 → 我们合并 DataFrame 的列名 # (统一用聚宽列名,方便策略层直接 pandas 筛选) JQ_COLUMN_ALIASES: Dict[str, str] = { @@ -454,9 +460,24 @@ class SanguoMiniQmtProvider(MiniQMTProvider): # type: ignore[misc] if not stocks: return pd.DataFrame(columns=list(JQ_COLUMN_ALIASES.values())) - xt = self._ensure_xtdata() date_str = _to_date_str(date) + # Step 0: fields 全落 unified 本地可供列 → 整步本地化(2026-08-26 根治 + # small_cap 选股 2/3 18min)。xt 路径 = download_financial_data(全池)卡满 + # 120s 超时被弃 + get_financial_data 逐股读缓存 ≈0.3s × 3226 只 ≈ 16min + # (08-26 live_20 [delay=+1084s] 实锤); unified parquet 批量秒级, 且 eps + # 与回测同源(LPP 年报 BASIC_EPS)——live 原走 miniQMT 最新一期, 双轨本就 + # 漂移, 全本地化后 live/回测同口径。本地异常回退 xt 老路径(fail-open)。 + if fields and set(fields) <= _LOCAL_FUND_FIELDS: + try: + return self._unified.get_fundamentals_df( + stocks, date=(date_str or datetime.now().strftime("%Y-%m-%d")), + fields=fields) + except Exception as exc: # noqa: BLE001 - 本地失败回退 xt 路径 + logger.warning("本地 fundamentals 快路失败, 回退 xt 路径: %s", exc) + + xt = self._ensure_xtdata() + # Step 1: 拉财务数据(miniQMT 返回 dict[stock_code] -> dict[table] -> DataFrame) # 成分股用 jq code,xtdata 要 QMT 风格("600519.SH"),通过 _normalize_security_code qmt_stocks = [self._normalize_security_code(s) for s in stocks] diff --git a/tests/portfolio/test_provider.py b/tests/portfolio/test_provider.py index c40463a..8d6bd5e 100644 --- a/tests/portfolio/test_provider.py +++ b/tests/portfolio/test_provider.py @@ -165,6 +165,102 @@ class TestFundamentalsMarketCapLocalization: assert kwargs["fields"] == ["market_cap", "circulating_market_cap"] +class TestFundamentalsFullLocalShortcut: + """2026-08-26 根治: fields 全落本地可供列 → 整步委托 unified, 跳过 xt。 + + 实锤: live_20 选股 2/3 fundamentals 335s→1081s(18min)才下单, 根因 = + xt.download_financial_data(3226) 卡满 120s 超时 + get_financial_data 逐股 + ≈0.3s × 全池 ≈ 16min; 而策略只消费 market_cap(排序)+eps(>0 过滤), unified + parquet 两列都能秒级供, 且 eps 与回测同源(LPP 年报 BASIC_EPS)——live 原走 + miniQMT 最新一期, 双轨本就漂移, 全本地化后 live/回测同口径。""" + + @staticmethod + def _unified(rows=None, side_effect=None): + m = MagicMock() + if side_effect is not None: + m.get_fundamentals_df.side_effect = side_effect + else: + m.get_fundamentals_df.return_value = pd.DataFrame(rows or []) + return m + + def test_all_local_fields_delegate_and_never_touch_xt(self, mock_xtquant): + """fields=["market_cap","eps"](small_cap 实际形态) → 直接返 unified 结果, + xt 全链路(get_financial_data/_ensure_xtdata)一次都不碰。""" + provider = SanguoMiniQmtProvider({"mode": "backtest", "auto_download": False}) + provider._unified = self._unified([ + {"code": "600519.XSHG", "market_cap": 3500.0, "eps": 1.2}, + ]) + df = provider.get_fundamentals_df( + ["600519.XSHG"], date="2024-09-30", fields=["market_cap", "eps"]) + assert abs(float(df.iloc[0]["market_cap"]) - 3500.0) < 1e-6 + mock_xtquant["xtdata"].get_financial_data.assert_not_called() + kwargs = provider._unified.get_fundamentals_df.call_args.kwargs + assert kwargs["fields"] == ["market_cap", "eps"] + assert kwargs["date"] == "2024-09-30" + + def test_fast_path_skips_ensure_xtdata(self, mock_xtquant): + """快路连 _ensure_xtdata 都不进(开盘不 import/初始化 xtdata, 也不留 + download 超时残留线程)。""" + provider = SanguoMiniQmtProvider({"mode": "backtest", "auto_download": False}) + provider._unified = self._unified([ + {"code": "600519.XSHG", "market_cap": 3500.0, "eps": 1.2}, + ]) + provider._ensure_xtdata = MagicMock( + side_effect=AssertionError("快路不得碰 xt")) + df = provider.get_fundamentals_df( + ["600519.XSHG"], date="2024-09-30", fields=["market_cap", "eps"]) + assert len(df) == 1 + + def test_circulating_market_cap_also_whitelisted(self, mock_xtquant): + provider = SanguoMiniQmtProvider({"mode": "backtest", "auto_download": False}) + provider._unified = self._unified([{"code": "600519.XSHG", + "circulating_market_cap": 9.9}]) + df = provider.get_fundamentals_df( + ["600519.XSHG"], date="2024-09-30", fields=["circulating_market_cap"]) + assert "circulating_market_cap" in df.columns + mock_xtquant["xtdata"].get_financial_data.assert_not_called() + + def test_unified_failure_falls_back_to_xt_path(self, mock_xtquant): + """本地异常 → 回退 xt 老路径(fail-open, 开盘永不因本地读挂死)。""" + provider = SanguoMiniQmtProvider({"mode": "backtest", "auto_download": False}) + provider._unified = self._unified(side_effect=RuntimeError("parquet io")) + df = provider.get_fundamentals_df( + ["600519.XSHG"], date="2024-09-30", fields=["market_cap", "eps"]) + assert not df.empty + mock_xtquant["xtdata"].get_financial_data.assert_called_once() + assert 19000 < float(df.iloc[0]["market_cap"]) < 22000 # Capital 原路径 + + def test_mixed_fields_stay_on_xt_path(self, mock_xtquant): + """fields 含白名单外列(如 roic) → 整组不走快路(不做半本地半 xt 混合)。""" + provider = SanguoMiniQmtProvider({"mode": "backtest", "auto_download": False}) + provider._unified = self._unified() + provider.get_fundamentals_df( + ["600519.XSHG"], date="2024-09-30", + fields=["market_cap", "eps", "roic"]) + mock_xtquant["xtdata"].get_financial_data.assert_called_once() + # 快路(整组委托)没触发; Step 3.5 的市值覆盖照旧 + assert provider._unified.get_fundamentals_df.call_count == 1 + assert provider._unified.get_fundamentals_df.call_args.kwargs["fields"] == \ + ["market_cap", "circulating_market_cap"] + + def test_fields_none_stays_on_xt_path(self, mock_xtquant): + """fields=None(全列, all_weather 小池形态) → 老路径, 回归护栏。""" + provider = SanguoMiniQmtProvider({"mode": "backtest", "auto_download": False}) + provider._unified = self._unified() + provider.get_fundamentals_df(["600519.XSHG"], date="2024-09-30") + mock_xtquant["xtdata"].get_financial_data.assert_called_once() + + def test_none_date_passes_today(self, mock_xtquant): + """live date=None(9:30 选股) → 快路委托收到今天, 与 Step 3.5 同口径。""" + from datetime import datetime as _dt + provider = SanguoMiniQmtProvider({"mode": "backtest", "auto_download": False}) + provider._unified = self._unified([{"code": "600519.XSHG", "eps": 1.0}]) + provider.get_fundamentals_df( + ["600519.XSHG"], date=None, fields=["market_cap", "eps"]) + kwargs = provider._unified.get_fundamentals_df.call_args.kwargs + assert kwargs["date"] == _dt.now().strftime("%Y-%m-%d") + + class TestGetFundamentalsQueryDictMode: def test_dict_with_stocks_returns_dataframe(self, mock_xtquant): provider = SanguoMiniQmtProvider({"mode": "backtest", "auto_download": False}) @@ -342,12 +438,26 @@ class TestGetIndexStocksDelegation: class TestGetFundamentalsDfFields: """get_fundamentals_df 加 fields 契约(对齐 unified:keep = code + 请求列)。""" - def test_fields_filters_columns(self, mock_xtquant): + def test_fields_filters_columns_local_shortcut(self, mock_xtquant): + """fields 全落白名单(market_cap+eps=small_cap 实际形态) → unified 快路, + 列契约同口径: keep = code + 请求列。""" provider = SanguoMiniQmtProvider({"mode": "backtest", "auto_download": False}) + provider._unified = MagicMock() + provider._unified.get_fundamentals_df.return_value = pd.DataFrame( + [{"code": "600519.XSHG", "market_cap": 3500.0, "eps": 1.2}]) df = provider.get_fundamentals_df( ["600519.XSHG"], date="2024-09-30", fields=["market_cap", "eps"], ) assert list(df.columns) == ["code", "market_cap", "eps"] + + def test_fields_filters_columns_xt_path(self, mock_xtquant): + """白名单外列混入(roic) → xt 老路径, 列过滤契约不变。""" + provider = SanguoMiniQmtProvider({"mode": "backtest", "auto_download": False}) + df = provider.get_fundamentals_df( + ["600519.XSHG"], date="2024-09-30", + fields=["market_cap", "eps", "roic"], + ) + assert list(df.columns) == ["code", "market_cap", "eps", "roic"] assert 19000 < float(df.iloc[0]["market_cap"]) < 22000 def test_fields_none_keeps_all_columns(self, mock_xtquant):