结构化异常体系

概述

参考 AkShare exceptions.py 分层设计，本项目实现了结构化异常体系，使上层业务可按异常类型精确决定重试/降级/告警策略。

实现文件

文件	职责
`backend/app/core/exceptions.py`	异常层次 + `should_fallback()` 决策
`backend/app/services/data_sources/retry_policy.py`	`with_retry` 装饰器 + 状态码可重试判定
`backend/app/services/market_data_service/infra.py`	`is_transient_error()` 瞬态兜底判定 + 东财熔断器

异常层次

异常类	HTTP	含义	处理策略
`NetworkError`（`ExternalServiceError` 的兼容子类）	502	网络连接/超时/上游服务异常	可重试（指数退避）
`RateLimitError`	429	429/403 限流或 IP 风控	应降级（切换数据源），403 不重试
`DataParsingError`	502	JSON格式错误/字段缺失	不重试（数据问题）
`SourceUnavailableError`	502	熔断器开启/服务下线	直接降级
`InvalidParameterError`	422	股票代码格式错误	不重试（参数问题）

旧名是兼容子类而非类型别名（NetworkError(ExternalServiceError)、 SourceUnavailableError(DataSourceError)、DataParsingError(DataSourceError)、 InvalidParameterError(ValidationError)，见 core/exceptions.py）：语义上是「except 新名 能接住旧名抛出」，但 except 旧名 接不住新名，因此新代码请直接用新名。所有异常携带 source 字段（tdx/tencent/eastmoney/sina/cninfo 等）。

降级决策工具

# core/exceptions.py
_FALLBACK_ERRORS = (RateLimitError, SourceUnavailableError, DataSourceError)

def should_fallback(exc: Exception) -> bool:
    """判断异常是否应触发数据源降级。"""
    return isinstance(exc, _FALLBACK_ERRORS)

重试与否的判定不在 exceptions.py，而由重试策略层按 HTTP 状态码/异常类型决定：

# services/data_sources/retry_policy.py
RETRYABLE_STATUS = {429, 500, 502, 503, 504}       # 瞬态，退避后可能恢复
NON_RETRYABLE_STATUS = {401, 403, 404, 501}        # 永久性，重试无益
#   403: 东财风控信号（重试反而加重封禁）
#   501: 腾讯接口下线/不支持的代码（2026-08-24 web.ifzq.gtimg.cn 全量 501）

@with_retry(max_attempts=3, backoff_base=1.0, backoff_factor=2.0, max_backoff=30.0)
def fetch_data(code: str) -> dict:
    """wait = min(backoff_base * backoff_factor**attempt, max_backoff)"""
    ...

def is_rate_limited(error: Exception) -> bool:
    """判断错误是否为限流信号（403 东财风控 / 429 限流）。"""

with_retry 的判定顺序：先看 HTTP 状态码（urllib HTTPError 继承 OSError，若先按异常类型判会把 4xx 永久性错误误判为可重试），再看异常类型，最后按错误文本关键词（connection reset / timeout / broken pipe 等）匹配。

使用示例

from ..core.exceptions import (
    NetworkError, RateLimitError, DataParsingError, should_fallback,
)

try:
    data = await fetch_from_source()
except RateLimitError as e:
    # 403/429 → 触发降级，不重试
    logger.warning(f"限流: {e}")
    if should_fallback(e):
        return await fallback_source()
except NetworkError as e:
    # 网络超时 → 可重试
    if attempt < max_retries:
        await asyncio.sleep(2 ** attempt)
        continue
except DataParsingError as e:
    # 数据格式错误 → 不重试，记录日志
    logger.error(f"解析失败: {e}")
    return None

瞬态错误判定

market_data_service/infra.py 的 is_transient_error() 作为兜底判断（未分类异常）：

def is_transient_error(exc: Exception) -> bool:
    """判断是否为瞬态错误（网络/超时），值得重试。"""
    # TimeoutError / socket.error / OSError
    # httpx.TimeoutException / ConnectError / NetworkError
    # 异常类名含 Timeout/ConnectionError/RemoteProtocolError 等

services/data_sources/http_manager.py::_is_transient_error() 是另一处独立实现（借鉴 yfinance data.py），服务于通用 HTTP 管理器的重试。

实际集成：`em_fetch_with_retry`

东财 HTTP 请求函数（market_data_service/providers_eastmoney.py）已全面使用结构化异常：

async def em_fetch_with_retry(url, params, retries=3, timeout=8.0) -> dict:
    # 0. push2 镜像池全部冷却 → 直接抛 NetworkError 快速失败（不再白等 3 次退避）
    if em_all_cooling(host):
        raise NetworkError("东财主机全部冷却，快速降级", source="eastmoney", url=url[:120])
    for attempt in range(retries):
        resp = await client.get(url, params=params, timeout=timeout)
        if resp.status_code == 403:
            em_breaker.record_failure()          # 熔断器记失败
            em_record_host_failure(host)         # 风控信号：同步共享冷却
            raise RateLimitError("403 Forbidden (东财风控)", source="eastmoney", status_code=403)
        if resp.status_code == 429:
            last_err = RateLimitError("429 Too Many Requests", ...)
            await asyncio.sleep(2 ** attempt * 1.5 + random.uniform(0.2, 0.8))
            continue                             # 429 → 退避重试
        resp.raise_for_status()
        return resp.json()
        # httpx 瞬态异常 → NetworkError，退避 2**attempt * 0.5s 后重试
        # httpx.HTTPStatusError（4xx/5xx）→ NetworkError，break 不重试
# 重试耗尽：未分类异常统一包装为 NetworkError

降级决策集成

主降级流程按异常类型精确处理：

if em_breaker.can_request():
    try:
        result = await em_fetch_with_retry(...)
        em_breaker.record_success()
    except Exception:
        em_breaker.record_failure()   # 任何异常都记失败，冷却期内 can_request() 返回 False
else:
    # 熔断开启 → 跳过东财，直接走备用源（新浪/TDX）
    ...

设计原则

提示

精确分类：不同异常类型对应不同处理策略，避免一刀切
安全默认：未知异常默认不重试，防止无限循环
source 标识：每个异常携带 source 字段（自由字符串，非枚举；常见 tdx/tencent/eastmoney/sina/cninfo/ths/sse/szse）
与熔断器联动：403/429 与瞬态耗尽均记 em_breaker 失败，连续 EM_BREAKER_FAILURE_THRESHOLD（默认 4）次进入 EM_BREAKER_COOLDOWN_SEC（默认 120s）冷却

概述​

实现文件​

异常层次​

降级决策工具​

使用示例​

瞬态错误判定​

实际集成：em_fetch_with_retry​

降级决策集成​

设计原则​

相关文档​

概述