From f9e7b778c48c914af4e0c69dd301e6907d7ea5f9 Mon Sep 17 00:00:00 2001 From: SynapShift <127070675+SynapShift@users.noreply.github.com> Date: Sat, 19 Sep 2026 23:09:08 +0800 Subject: [PATCH] fix(market): bound AkShare market stats calls (#2345) --- data_provider/akshare_fetcher.py | 32 ++++- docs/CHANGELOG.md | 1 + tests/test_akshare_market_stats_timeout.py | 134 +++++++++++++++++++++ 3 files changed, 165 insertions(+), 2 deletions(-) create mode 100644 tests/test_akshare_market_stats_timeout.py diff --git a/data_provider/akshare_fetcher.py b/data_provider/akshare_fetcher.py index 4b3faa5a8..148413f30 100644 --- a/data_provider/akshare_fetcher.py +++ b/data_provider/akshare_fetcher.py @@ -64,6 +64,7 @@ logger = logging.getLogger(__name__) SINA_REALTIME_ENDPOINT = "hq.sinajs.cn/list" TENCENT_REALTIME_ENDPOINT = "qt.gtimg.cn/q" _AKSHARE_HISTORY_CALL_TIMEOUT = 30.0 +_AKSHARE_MARKET_STATS_CALL_TIMEOUT = 30.0 _AKSHARE_TIMEOUT_PROCESS_JOIN_GRACE = 1.0 _AKSHARE_TIMEOUT_PROCESS_START_METHOD = "spawn" @@ -428,6 +429,7 @@ class AkshareFetcher(BaseFetcher): self.sleep_max = sleep_max self._last_request_time: Optional[float] = None self._history_call_timeout = _AKSHARE_HISTORY_CALL_TIMEOUT + self._market_stats_call_timeout = _AKSHARE_MARKET_STATS_CALL_TIMEOUT # 东财补丁开启才执行打补丁操作 if get_config().enable_eastmoney_patch: eastmoney_patch() @@ -2013,7 +2015,11 @@ class AkshareFetcher(BaseFetcher): "[MarketStats] component=market_stats provider=AkshareFetcher " "api=ak.stock_zh_a_spot_em action=request_start" ) - df = ak.stock_zh_a_spot_em() + df = _akshare_call_with_timeout( + ak.stock_zh_a_spot_em, + timeout=self._market_stats_call_timeout, + call_name="ak.stock_zh_a_spot_em", + ) elapsed = time.monotonic() - started_at logger.info( "[MarketStats] component=market_stats provider=AkshareFetcher " @@ -2026,6 +2032,15 @@ class AkshareFetcher(BaseFetcher): "[MarketStats] component=market_stats provider=AkshareFetcher " "api=ak.stock_zh_a_spot_em action=parse status=empty" ) + except TimeoutError as e: + elapsed = time.monotonic() - started_at + logger.warning( + "[MarketStats] component=market_stats provider=AkshareFetcher " + "api=ak.stock_zh_a_spot_em action=request_timeout elapsed=%.2fs " + "error=%s fallback=ak.stock_zh_a_spot", + elapsed, + e, + ) except Exception as e: logger.warning( "[MarketStats] component=market_stats provider=AkshareFetcher " @@ -2043,7 +2058,11 @@ class AkshareFetcher(BaseFetcher): "[MarketStats] component=market_stats provider=AkshareFetcher " "api=ak.stock_zh_a_spot action=request_start" ) - df = ak.stock_zh_a_spot() + df = _akshare_call_with_timeout( + ak.stock_zh_a_spot, + timeout=self._market_stats_call_timeout, + call_name="ak.stock_zh_a_spot", + ) elapsed = time.monotonic() - started_at logger.info( "[MarketStats] component=market_stats provider=AkshareFetcher " @@ -2056,6 +2075,15 @@ class AkshareFetcher(BaseFetcher): "[MarketStats] component=market_stats provider=AkshareFetcher " "api=ak.stock_zh_a_spot action=parse status=empty" ) + except TimeoutError as e: + elapsed = time.monotonic() - started_at + logger.error( + "[MarketStats] component=market_stats provider=AkshareFetcher " + "api=ak.stock_zh_a_spot action=request_timeout elapsed=%.2fs " + "error=%s fallback=none", + elapsed, + e, + ) except Exception as e: logger.error( "[MarketStats] component=market_stats provider=AkshareFetcher " diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 0141e06cc..dd56d26f0 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -13,6 +13,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). - [测试] 修复股票名称解析冷启动超时并发测试的同步竞态：在放行后台抓取前确认两个等待者均已结束并返回空结果，避免 Docker 发布门禁偶发失败。 - [文档] 将仓库内所有 SerpApi 链接统一更新为新的赞助转化追踪地址。 - [修复] 智能导入兼容带 UTF-8 BOM 的 CSV 与剪贴板文本，避免 `code` 表头被误当成数据并丢失有效股票代码。 +- [修复] 为 AkShare 大盘涨跌统计的东财与新浪降级调用增加可强制终止的子进程超时，避免外部接口无响应时长期占用分析线程（Fixes #2340）。 diff --git a/tests/test_akshare_market_stats_timeout.py b/tests/test_akshare_market_stats_timeout.py new file mode 100644 index 000000000..c1d417a06 --- /dev/null +++ b/tests/test_akshare_market_stats_timeout.py @@ -0,0 +1,134 @@ +# -*- coding: utf-8 -*- +"""Regression tests for AkShare market statistics timeout handling.""" + +import sys +from types import SimpleNamespace + +import pandas as pd + +from tests.litellm_stub import ensure_litellm_stub + +ensure_litellm_stub() + +from data_provider.akshare_fetcher import AkshareFetcher + + +def _market_frame() -> pd.DataFrame: + return pd.DataFrame( + { + "代码": ["600000", "000001"], + "名称": ["浦发银行", "平安银行"], + "最新价": [11.0, 9.0], + "昨收": [10.0, 10.0], + "成交额": [100_000_000, 200_000_000], + } + ) + + +def test_market_stats_eastmoney_call_uses_timeout_wrapper(monkeypatch) -> None: + calls = [] + eastmoney_api = object() + sina_api = object() + + def fake_call(func, *args, timeout=None, call_name="", **kwargs): + calls.append((func, timeout, call_name)) + return _market_frame() + + monkeypatch.setitem( + sys.modules, + "akshare", + SimpleNamespace( + stock_zh_a_spot_em=eastmoney_api, + stock_zh_a_spot=sina_api, + ), + ) + monkeypatch.setattr( + "data_provider.akshare_fetcher._akshare_call_with_timeout", + fake_call, + ) + + fetcher = AkshareFetcher(sleep_min=0, sleep_max=0) + fetcher._market_stats_call_timeout = 7 + + stats = fetcher.get_market_stats() + + assert calls == [(eastmoney_api, 7, "ak.stock_zh_a_spot_em")] + assert stats == { + "up_count": 1, + "down_count": 1, + "flat_count": 0, + "limit_up_count": 1, + "limit_down_count": 1, + "total_amount": 3.0, + } + + +def test_market_stats_falls_back_to_sina_after_eastmoney_timeout( + monkeypatch, +) -> None: + calls = [] + eastmoney_api = object() + sina_api = object() + + def fake_call(func, *args, timeout=None, call_name="", **kwargs): + calls.append((func, timeout, call_name)) + if func is eastmoney_api: + raise TimeoutError("eastmoney timeout") + return _market_frame() + + monkeypatch.setitem( + sys.modules, + "akshare", + SimpleNamespace( + stock_zh_a_spot_em=eastmoney_api, + stock_zh_a_spot=sina_api, + ), + ) + monkeypatch.setattr( + "data_provider.akshare_fetcher._akshare_call_with_timeout", + fake_call, + ) + + fetcher = AkshareFetcher(sleep_min=0, sleep_max=0) + fetcher._market_stats_call_timeout = 9 + + stats = fetcher.get_market_stats() + + assert calls == [ + (eastmoney_api, 9, "ak.stock_zh_a_spot_em"), + (sina_api, 9, "ak.stock_zh_a_spot"), + ] + assert stats is not None + assert stats["total_amount"] == 3.0 + + +def test_market_stats_returns_none_after_both_calls_timeout(monkeypatch) -> None: + calls = [] + eastmoney_api = object() + sina_api = object() + + def fake_call(func, *args, timeout=None, call_name="", **kwargs): + calls.append((func, timeout, call_name)) + raise TimeoutError(f"{call_name} timeout") + + monkeypatch.setitem( + sys.modules, + "akshare", + SimpleNamespace( + stock_zh_a_spot_em=eastmoney_api, + stock_zh_a_spot=sina_api, + ), + ) + monkeypatch.setattr( + "data_provider.akshare_fetcher._akshare_call_with_timeout", + fake_call, + ) + + fetcher = AkshareFetcher(sleep_min=0, sleep_max=0) + fetcher._market_stats_call_timeout = 5 + + assert fetcher.get_market_stats() is None + assert calls == [ + (eastmoney_api, 5, "ak.stock_zh_a_spot_em"), + (sina_api, 5, "ak.stock_zh_a_spot"), + ]