diff --git a/data_provider/akshare_fetcher.py b/data_provider/akshare_fetcher.py index 97b53da3c..05b5247e6 100644 --- a/data_provider/akshare_fetcher.py +++ b/data_provider/akshare_fetcher.py @@ -24,6 +24,7 @@ AkshareFetcher - 主数据源 (Priority 1) """ import logging +import multiprocessing import os import random import time @@ -60,6 +61,9 @@ 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_TIMEOUT_PROCESS_JOIN_GRACE = 1.0 +_AKSHARE_TIMEOUT_PROCESS_START_METHOD = "spawn" # User-Agent 池，用于随机轮换 @@ -301,6 +305,72 @@ def _build_realtime_failure_message( ) +def _akshare_call_with_timeout( + func, + *args, + timeout: Optional[float] = None, + call_name: str = "akshare", + **kwargs, +): + """Run an akshare call with a bounded wait time.""" + wait_seconds = _AKSHARE_HISTORY_CALL_TIMEOUT if timeout is None else float(timeout) + + multiprocessing.freeze_support() + ctx = multiprocessing.get_context(_AKSHARE_TIMEOUT_PROCESS_START_METHOD) + parent_conn, child_conn = ctx.Pipe(duplex=False) + process = ctx.Process( + target=_akshare_timeout_worker, + args=(child_conn, func, args, kwargs), + name=f"akshare-{call_name}", + daemon=True, + ) + + process.start() + child_conn.close() + + try: + if not parent_conn.poll(wait_seconds): + _terminate_akshare_process(process) + raise TimeoutError(f"{call_name} 调用超过 {wait_seconds:g}s，已放弃等待") + + try: + ok, value = parent_conn.recv() + except EOFError as exc: + raise RuntimeError(f"{call_name} 调用进程未返回结果") from exc + finally: + parent_conn.close() + process.join(_AKSHARE_TIMEOUT_PROCESS_JOIN_GRACE) + _terminate_akshare_process(process) + + if ok: + return value + raise value + + +def _akshare_timeout_worker(conn, func, args, kwargs) -> None: + try: + conn.send((True, func(*args, **kwargs))) + except BaseException as exc: + try: + conn.send((False, exc)) + except BaseException: + try: + conn.send((False, RuntimeError(f"{type(exc).__name__}: {exc}"))) + except BaseException: + pass + finally: + conn.close() + + +def _terminate_akshare_process(process) -> None: + if process.is_alive(): + process.terminate() + process.join(_AKSHARE_TIMEOUT_PROCESS_JOIN_GRACE) + if process.is_alive(): + process.kill() + process.join(_AKSHARE_TIMEOUT_PROCESS_JOIN_GRACE) + + class AkshareFetcher(BaseFetcher): """ Akshare 数据源实现 @@ -328,6 +398,7 @@ class AkshareFetcher(BaseFetcher): self.sleep_min = sleep_min self.sleep_max = sleep_max self._last_request_time: Optional[float] = None + self._history_call_timeout = _AKSHARE_HISTORY_CALL_TIMEOUT # 东财补丁开启才执行打补丁操作 if get_config().enable_eastmoney_patch: eastmoney_patch() @@ -495,11 +566,14 @@ class AkshareFetcher(BaseFetcher): self._enforce_rate_limit() try: - df = ak.stock_zh_a_daily( + df = _akshare_call_with_timeout( + ak.stock_zh_a_daily, symbol=symbol, start_date=start_date.replace('-', ''), end_date=end_date.replace('-', ''), - adjust="qfq" + adjust="qfq", + timeout=self._history_call_timeout, + call_name="ak.stock_zh_a_daily", ) # 标准化新浪数据列名 @@ -541,11 +615,14 @@ class AkshareFetcher(BaseFetcher): self._enforce_rate_limit() try: - df = ak.stock_zh_a_hist_tx( + df = _akshare_call_with_timeout( + ak.stock_zh_a_hist_tx, symbol=symbol, start_date=start_date.replace('-', ''), end_date=end_date.replace('-', ''), - adjust="qfq" + adjust="qfq", + timeout=self._history_call_timeout, + call_name="ak.stock_zh_a_hist_tx", ) # 标准化腾讯数据列名 diff --git a/docs/CHANGELOG.md b/docs/CHANGELOG.md index 878bcd747..a7a51d3ff 100644 --- a/docs/CHANGELOG.md +++ b/docs/CHANGELOG.md @@ -45,6 +45,7 @@ and this project adheres to [Semantic Versioning](https://semver.org/). - [改进] P2-min：LLM Prompt 注入市场阶段上下文。 - [修复] 问股 single-agent 新增 provider-aware trace 分轨，跨轮保留 DeepSeek V4 thinking + tool-call 的 `reasoning_content` 与工具协议材料。 - [新功能] 股票自动补全索引默认支持从 GitHub main 远程刷新并缓存到本地，Web/CLI 分析入口失败时自动降级到内置索引，降低摘帽和更名后旧简称污染分析的概率。 +- [修复] 为 Akshare 新浪/腾讯 A 股历史兜底接口增加调用级超时，并补齐 Tushare `605xxx` 沪市代码路由回归测试，避免定时分析因数据源无响应而挂起。 ## [3.18.0] - 2026-05-21 diff --git a/main.py b/main.py index 4ac3b96eb..e16614c1d 100644 --- a/main.py +++ b/main.py @@ -23,6 +23,7 @@ A股自选股智能分析系统 - 主调度程序 """ from __future__ import annotations +import multiprocessing import os from pathlib import Path from typing import Any, Callable, Dict, List, Optional, Tuple @@ -1013,4 +1014,5 @@ def main() -> int: if __name__ == "__main__": + multiprocessing.freeze_support() sys.exit(main()) diff --git a/tests/test_akshare_history_timeout.py b/tests/test_akshare_history_timeout.py new file mode 100644 index 000000000..8d2c7e5fe --- /dev/null +++ b/tests/test_akshare_history_timeout.py @@ -0,0 +1,207 @@ +# -*- coding: utf-8 -*- +"""Regression tests for Akshare historical fallback timeout handling.""" + +import multiprocessing +import sys +import time +from types import SimpleNamespace + +import pandas as pd +import pytest + +from tests.litellm_stub import ensure_litellm_stub + +ensure_litellm_stub() + +from data_provider.akshare_fetcher import AkshareFetcher, _akshare_call_with_timeout + + +def _sleep_for(seconds: float) -> None: + time.sleep(seconds) + + +def _return_value(value): + return value + + +def test_akshare_call_with_timeout_uses_spawn_context(monkeypatch) -> None: + requested_methods = [] + call_order = [] + + class FakeConnection: + def __init__(self, messages): + self.messages = messages + + def send(self, value): + self.messages.append(value) + + def poll(self, timeout): + return bool(self.messages) + + def recv(self): + if not self.messages: + raise EOFError + return self.messages.pop(0) + + def close(self): + pass + + class FakeProcess: + def __init__(self, target, args, name, daemon): + self.target = target + self.args = args + self.name = name + self.daemon = daemon + + def start(self): + self.target(*self.args) + + def join(self, timeout=None): + pass + + def is_alive(self): + return False + + def terminate(self): + pass + + def kill(self): + pass + + class FakeContext: + def Pipe(self, duplex=False): + messages = [] + return FakeConnection(messages), FakeConnection(messages) + + Process = FakeProcess + + def fake_get_context(method=None): + call_order.append("get_context") + requested_methods.append(method) + return FakeContext() + + def fake_freeze_support(): + call_order.append("freeze_support") + + monkeypatch.setattr( + "data_provider.akshare_fetcher.multiprocessing.get_context", + fake_get_context, + ) + monkeypatch.setattr( + "data_provider.akshare_fetcher.multiprocessing.freeze_support", + fake_freeze_support, + ) + + result = _akshare_call_with_timeout( + _return_value, + "ok", + timeout=1, + call_name="unit-default-context", + ) + + assert result == "ok" + assert requested_methods == ["spawn"] + assert call_order == ["freeze_support", "get_context"] + + +def test_akshare_call_with_timeout_returns_promptly() -> None: + started = time.monotonic() + + with pytest.raises(TimeoutError, match="unit-hang"): + _akshare_call_with_timeout( + _sleep_for, + 0.2, + timeout=0.01, + call_name="unit-hang", + ) + + assert time.monotonic() - started < 0.5 + + +def test_akshare_call_with_timeout_reaps_timed_out_worker_process() -> None: + call_name = "unit-hang-reap" + + with pytest.raises(TimeoutError, match=call_name): + _akshare_call_with_timeout( + _sleep_for, + 5, + timeout=0.01, + call_name=call_name, + ) + + leaked = [ + process + for process in multiprocessing.active_children() + if process.name == f"akshare-{call_name}" + ] + assert leaked == [] + + +@pytest.mark.parametrize( + ("method_name", "api_name", "call_name"), + [ + ("_fetch_stock_data_sina", "stock_zh_a_daily", "ak.stock_zh_a_daily"), + ("_fetch_stock_data_tx", "stock_zh_a_hist_tx", "ak.stock_zh_a_hist_tx"), + ], +) +def test_sina_and_tencent_history_calls_use_timeout_wrapper( + monkeypatch, + method_name: str, + api_name: str, + call_name: str, +) -> None: + captured = {} + + def fake_call(func, *args, timeout=None, call_name="", **kwargs): + captured["func"] = func + captured["timeout"] = timeout + captured["call_name"] = call_name + captured["kwargs"] = kwargs + return pd.DataFrame( + { + "date": ["2026-05-25"], + "open": [10.0], + "high": [10.5], + "low": [9.8], + "close": [10.2], + "volume": [1000], + "amount": [20000], + } + ) + + fake_api_func = object() + fake_akshare = SimpleNamespace(**{api_name: fake_api_func}) + monkeypatch.setitem(sys.modules, "akshare", fake_akshare) + monkeypatch.setattr("data_provider.akshare_fetcher._akshare_call_with_timeout", fake_call) + + fetcher = AkshareFetcher(sleep_min=0, sleep_max=0) + fetcher._history_call_timeout = 7 + + method = getattr(fetcher, method_name) + df = method("605218", "2026-05-01", "2026-05-25") + + assert captured["func"] is fake_api_func + assert captured["timeout"] == 7 + assert captured["call_name"] == call_name + assert captured["kwargs"]["symbol"] == "sh605218" + assert captured["kwargs"]["start_date"] == "20260501" + assert captured["kwargs"]["end_date"] == "20260525" + assert captured["kwargs"]["adjust"] == "qfq" + assert list(df.columns)[:7] == ["日期", "开盘", "最高", "最低", "收盘", "成交量", "成交额"] + + +def test_stock_data_falls_back_after_sina_timeout(monkeypatch) -> None: + fetcher = AkshareFetcher(sleep_min=0, sleep_max=0) + tx_df = pd.DataFrame({"日期": ["2026-05-25"], "收盘": [10.2]}) + + monkeypatch.setattr(fetcher, "_fetch_stock_data_em", lambda *args: pd.DataFrame()) + monkeypatch.setattr( + fetcher, + "_fetch_stock_data_sina", + lambda *args: (_ for _ in ()).throw(TimeoutError("sina timeout")), + ) + monkeypatch.setattr(fetcher, "_fetch_stock_data_tx", lambda *args: tx_df) + + result = fetcher._fetch_stock_data("605218", "2026-05-01", "2026-05-25") + + assert result is tx_df diff --git a/tests/test_tushare_fetcher_followups.py b/tests/test_tushare_fetcher_followups.py index 104e0f2f2..f40089ed6 100644 --- a/tests/test_tushare_fetcher_followups.py +++ b/tests/test_tushare_fetcher_followups.py @@ -159,6 +159,7 @@ class TestTushareFetcherFollowUps(unittest.TestCase): self.assertEqual(fetcher._convert_stock_code("SZ000001"), "000001.SZ") self.assertEqual(fetcher._convert_stock_code("SH600519"), "600519.SH") + self.assertEqual(fetcher._convert_stock_code("605218"), "605218.SH") self.assertEqual(fetcher._convert_stock_code("600519.SS"), "600519.SH") @patch.dict(sys.modules, {"tushare": MagicMock()})