搞定人民币兑美元汇率抓取：避开这3个坑，高频面试题不再丢分

刚把网上找来的汇率转换脚本复制到本地，python main.py 一跑，报错 ConnectionError，或者返回的汇率是三天前的旧数据，改半天配置还是不通。这种“复制代码跑不通”的绝望感，在 Python 后端开发中太常见了，尤其是涉及外部数据源如人民币兑美元的实时汇率获取时。很多候选人以为这只是个简单的 API 调用，结果在高频面试题环节被追问异常处理、数据一致性、缓存策略时，张口结舌。今天我们就从零搭建一个稳健的汇率获取服务，不玩虚的，直接上实战项目。

项目目标与场景定义

我们要解决的核心痛点是：如何在一个 Python 项目中，稳定、准确、低延迟地获取人民币兑美元的实时汇率，并能应对网络波动和数据源失效的情况。

很多初学者会直接调用免费的公共 API，比如 open.er-api.com 或者某些财经网站的公开接口。这些接口虽然免费，但存在三个致命问题：

频率限制严格：每秒请求超过一定次数直接封 IP。
数据延迟高：免费接口的数据更新周期通常是 15 分钟甚至 1 小时，对于金融场景来说，这就是“错误数据”。
缺乏 SLA 保障：没有可用性承诺，挂了没人管。

在真实的劳务班组或金融后端项目中，我们通常不会直接依赖单一免费源。我们的目标是构建一个“多源冗余 + 本地缓存 + 故障降级”的汇率服务。这个项目不仅是一个工具，更是一个考察后端工程化能力的绝佳案例。在面试中，如果你能讲清楚如何从“直接调 API”进化到“构建高可用汇率服务”，这就比单纯背八股文要有说服力得多。

目录结构与依赖管理

为了保持项目的可复现性，我们采用标准 Python 项目结构。这里推荐使用 pyproject.toml 来管理依赖，比 requirements.txt 更现代，能锁定版本，避免环境漂移。

currency_service/
├── main.py          # 入口文件，启动 FastAPI 服务
├── core/
│   ├── __init__.py
│   ├── config.py    # 配置管理
│   └── exceptions.py# 自定义异常
├── services/
│   ├── __init__.py
│   ├── fetcher.py   # 数据抓取核心逻辑
│   └── cache.py     # 缓存层
├── utils/
│   └── logger.py    # 日志工具
└── pyproject.toml   # 依赖与项目元数据

在 pyproject.toml 中，我们引入 httpx（异步 HTTP 客户端，比 requests 更适合高并发场景）、fastapi（Web 框架）、pydantic（数据验证）和 redis（缓存客户端）。

[project]
name = "currency-service"
version = "0.1.0"
dependencies = ["fastapi>=0.100.0","httpx>=0.24.0","pydantic>=2.0.0","redis>=5.0.0","loguru>=0.7.0"
]

注意：不要使用 requests。在涉及人民币兑美元这类高频变动的数据时，同步 IO 会成为瓶颈。httpx 支持异步，能显著提升并发处理能力。这一点在高频面试题中经常被问到：“为什么你的服务在高并发下变慢？”答案往往就是 IO 阻塞。

核心代码实现：多源抓取与容错

这是项目的核心。我们定义两个数据源：主源（假设是某商业 API）和备源（免费 API）。代码逻辑遵循“先查缓存 -> 查主源 -> 查备源 -> 返回兜底值”的顺序。

1. 数据模型定义

使用 pydantic 定义汇率数据结构，确保类型安全。

# services/fetcher.py
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Optionalclass ExchangeRate(BaseModel):base: str = "CNY"target: str = "USD"rate: float = Field(..., gt=0, description="1 CNY = ? USD")timestamp: datetimesource: str = Field(..., description="数据源标识")is_fallback: bool = False

2. 异步抓取器实现

这里展示了如何处理网络异常。很多新手代码只写了 try-except 捕获 Exception，这是大忌。我们需要具体捕获 httpx.ConnectError、httpx.TimeoutException 等。

import httpx
import asyncio
from loguru import logger
from core.config import settings
from services.cache import CacheServiceclass RateFetcher:def __init__(self):# 设置超时，避免请求挂起self.client = httpx.AsyncClient(timeout=httpx.Timeout(5.0))self.cache = CacheService()async def fetch_usd_rate(self) -> ExchangeRate:# 1. 检查缓存，TTL 设为 60 秒cached_data = await self.cache.get("cny_usd_rate")if cached_data:logger.info("Hit cache for CNY/USD")return ExchangeRate(**cached_data)# 2. 尝试主数据源try:rate_data = await self._fetch_from_primary()if rate_data:await self.cache.set("cny_usd_rate", rate_data.model_dump(), ttl=60)return rate_dataexcept Exception as e:logger.error(f"Primary source failed: {e}")# 3. 尝试备用数据源try:rate_data = await self._fetch_from_backup()if rate_data:# 备用源数据可靠性稍低，缩短缓存时间await self.cache.set("cny_usd_rate", rate_data.model_dump(), ttl=10)rate_data.is_fallback = Truereturn rate_dataexcept Exception as e:logger.error(f"Backup source failed: {e}")# 4. 兜底策略：返回一个保守的静态值，并标记logger.critical("All sources failed, using fallback value")return ExchangeRate(base="CNY",target="USD",rate=0.138,  # 静态兜底值，需定期人工更新timestamp=datetime.utcnow(),source="FALLBACK_STATIC",is_fallback=True)async def _fetch_from_primary(self) -> Optional[ExchangeRate]:# 模拟主源 API 调用url = settings.PRIMARY_API_URLheaders = {"Authorization": f"Bearer {settings.PRIMARY_API_KEY}"}async with self.client.get(url, headers=headers) as response:response.raise_for_status()data = response.json()return ExchangeRate(base="CNY",target="USD",rate=float(data["rates"]["USD"]),timestamp=datetime.fromisoformat(data["date"]),source="PRIMARY")async def _fetch_from_backup(self) -> Optional[ExchangeRate]:# 模拟备用免费 APIurl = "https://open.er-api.com/v6/latest/CNY"async with self.client.get(url) as response:response.raise_for_status()data = response.json()return ExchangeRate(base="CNY",target="USD",rate=float(data["rates"]["USD"]),timestamp=datetime.utcnow(),source="BACKUP")

逐行解析关键点：

response.raise_for_status()：这行代码至关重要。如果 HTTP 状态码是 400 以上，它会抛出异常，让我们进入 except 块。很多代码忽略这一步，导致拿到 500 错误的 JSON 还在解析，最终报 KeyError。
ttl=60 vs ttl=10：主源数据可信，缓存久一点；备源数据可能滞后，缓存短一点。这种细粒度的控制，是区分“玩具项目”和“生产项目”的分水岭。
兜底值：在金融系统中，永远不要抛异常给用户。如果所有数据源都挂了，返回一个旧的、标记为 is_fallback=True 的值，让前端提示用户“数据可能延迟”，这比直接报错 500 体验好得多。这一点在 Stack Overflow 上的高赞回答中也被反复提及：优雅降级优于故障停止。

运行与测试：验证稳定性

代码写完了，怎么证明它靠谱？不能只跑一次成功就完事。我们需要模拟网络故障。

1. 单元测试

使用 pytest 和 respx（httpx 的 mock 库）来模拟 API 响应。

import pytest
import respx
from services.fetcher import RateFetcher
from datetime import datetime@pytest.mark.asyncio
@respx.mock
async def test_fetch_primary_success():respx.get("https://api.primary.com/rates").mock(return_value=respx.Response(status_code=200,json={"rates": {"USD": 0.139}, "date": "2023-10-27"}))fetcher = RateFetcher()result = await fetcher.fetch_usd_rate()assert result.source == "PRIMARY"assert result.rate == 0.139assert result.is_fallback is False@pytest.mark.asyncio
@respx.mock
async def test_fetch_fallback_on_error():# 模拟主源和备源都超时respx.get("https://api.primary.com/rates").mock(side_effect=Exception("Timeout"))respx.get("https://open.er-api.com/v6/latest/CNY").mock(side_effect=Exception("Timeout"))fetcher = RateFetcher()result = await fetcher.fetch_usd_rate()assert result.source == "FALLBACK_STATIC"assert result.is_fallback is True

2. 压力测试

使用 locust 进行简单压测，观察在 100 并发下，CPU 和内存的使用情况。重点观察 httpx 连接池是否耗尽。如果报错 PoolTimeout，说明连接池大小配置不当，需要调整 httpx.AsyncClient 的 limits 参数。

在测试过程中，你可能会发现 datetime.utcnow() 在不同时区环境下行为不一致。这是一个常见的坑。建议使用 datetime.now(timezone.utc) 代替，这是 Python 3.11+ 推荐的做法，也是 PEP 498 中强调的时区安全意识。

优化扩展：从可用到高性能

当基础功能跑通后，如何进一步提升？

连接池优化：httpx 默认连接池较小。在高并发场景下，建议显式配置 limits=httpx.Limits(max_keepalive_connections=50, max_connections=100)。
熔断机制：如果主源连续失败 5 次，暂时停止调用主源 1 分钟，直接走备源。这可以防止雪崩效应。可以使用 pybreaker 库实现。
数据校验：汇率是有范围的，比如 CNY/USD 不可能大于 1 或小于 0.01。在 pydantic 模型中加入 Field(ge=0.01, le=1.0) 进行硬校验，防止脏数据污染缓存。
监控告警：集成 Prometheus，暴露 /metrics 端点，监控 rate_fetch_duration_seconds 和 rate_fetch_errors_total。当错误率超过 1% 时，触发 Slack 告警。