fix: 为 Akshare 历史兜底接口增加超时保护 (#1465)

* fix(issue-1464): [bug]-scheduled-analysis-hangs-when-stoc

* fix(review-feedback-1465): Avoid leaking daemon threads on timed-out Akshare calls

* fix(review-feedback-1465): fix can reintroduce hangs/crashes in the exact scheduled pipeline it

* fix(review-feedback-1465): Use a safe start method for timeout subprocesses

* fix(review-feedback-1465): Call freeze support before spawning timeout workers

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",
             )
 
             # 标准化腾讯数据列名

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
 

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())

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

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()})