3. REST API实战：使用Python requests库调用交易所行情接口、解析JSON响应、错误处理
说实话，REST API 是量化交易里最基础也最常用的接口形式。我刚开始做量化那会儿，第一个任务就是写个脚本拉行情数据。那时候还不懂什么 websocket，全靠 REST 轮询。今天咱们就把这块彻底讲透。
3.1 准备工作：安装 requests 库
Python 自带的 urllib 也能发 HTTP 请求，但用起来太啰嗦。我个人习惯用 requests 库，代码写起来干净利落。
pip install requests嗯，就这一行。如果你用的是 Anaconda 环境，记得先激活虚拟环境再装。
3.2 调用行情接口：一个真实的例子
咱们拿币安的现货行情接口来演示。为什么选币安？因为它的 API 文档写得清楚，而且接口设计很规范，适合当教学案例。
先看一个最简单的请求——查 BTC/USDT 的最新成交价：
import requests
url = "https://api.binance.com/api/v3/ticker/price"
params = {"symbol": "BTCUSDT"}
response = requests.get(url, params=params)
print(response.text)跑一下，你会看到类似这样的输出：
{"symbol": "BTCUSDT", "price": "43250.12000000"}这里有个细节我提一下：参数名 symbol 的值是 BTCUSDT，不是 BTC-USDT 也不是 BTC/USDT。每个交易所的命名规则都不一样，我在项目中遇到过好几次因为符号写错导致拿不到数据的情况。建议你写代码前先翻翻 API 文档，确认一下命名规范。
3.3 解析 JSON 响应
返回的字符串是 JSON 格式，Python 里处理 JSON 有两种方式：
方式一：直接用 response.json()
data = response.json()
print(data["price"]) # 输出: 43250.12000000方式二：先拿文本，再用 json 模块解析
import json
text = response.text
data = json.loads(text)
print(data["price"])我个人推荐第一种，因为 requests 库内部已经帮你处理了编码问题。但如果你需要调试，比如看看原始报文长什么样，那就用第二种。
行情接口返回的数据结构通常是这样的：
| 字段 | 类型 | 说明 |
|---|---|---|
| symbol | string | 交易对名称 |
| price | string | 最新成交价（注意是字符串） |
| time | integer | 更新时间戳（毫秒） |
你可能会问：为什么 price 是字符串而不是浮点数？
原因很简单——精度问题。浮点数在计算机里是有精度损失的，而价格数据必须精确到小数点后8位甚至更多。所以交易所普遍用字符串来传输价格，你自己解析的时候再转成 Decimal 类型做计算。
from decimal import Decimal
price = Decimal(data["price"])
3.4 错误处理：别让程序默默崩溃
写量化程序，最怕的就是程序跑着跑着突然挂了，你还没发现。我见过有人写的脚本，网络一波动就直接抛异常退出，结果错过了行情。所以错误处理一定要做扎实。
常见的错误分三类：
- 网络错误：连不上服务器、DNS 解析失败、超时
- HTTP 错误：4xx（客户端问题）、5xx（服务端问题）
- 数据错误：返回的 JSON 格式不对、字段缺失
来看一个健壮的写法：
import requests
from requests.exceptions import RequestException, Timeout
def fetch_price(symbol):
url = "https://api.binance.com/api/v3/ticker/price"
params = {"symbol": symbol}
try:
# 设置超时，防止卡死
response = requests.get(url, params=params, timeout=5)
# 检查 HTTP 状态码
response.raise_for_status()
# 解析 JSON
data = response.json()
# 检查关键字段是否存在
if "price" not in data:
raise ValueError(f"响应中缺少 price 字段: {data}")
return data["price"]
except Timeout:
print(f"请求超时: {symbol}")
return None
except RequestException as e:
print(f"网络请求失败: {e}")
return None
except ValueError as e:
print(f"数据解析错误: {e}")
return None
# 使用示例
price = fetch_price("BTCUSDT")
if price:
print(f"当前价格: {price}")
else:
print("获取价格失败，稍后重试")这里我用了 raise_for_status()，它会自动检查状态码。如果是 4xx 或 5xx，直接抛异常。你想想看，如果不做这个检查，程序可能拿到一个错误页面还当正常数据去解析，那后果就严重了。
3.5 实战：获取深度数据并解析
光拿个最新价不过瘾，咱们来点更实用的——获取订单簿深度数据。这个在策略里经常用到，比如判断买卖盘的厚度。
def fetch_order_book(symbol, limit=10):
url = "https://api.binance.com/api/v3/depth"
params = {"symbol": symbol, "limit": limit}
try:
response = requests.get(url, params=params, timeout=5)
response.raise_for_status()
data = response.json()
# 解析买单和卖单
bids = data.get("bids", []) # 买单：[[价格, 数量], ...]
asks = data.get("asks", []) # 卖单：[[价格, 数量], ...]
# 计算买卖盘口总深度
bid_volume = sum(float(b[1]) for b in bids)
ask_volume = sum(float(a[1]) for a in asks)
return {
"bids": bids,
"asks": asks,
"bid_volume": bid_volume,
"ask_volume": ask_volume
}
except Exception as e:
print(f"获取深度数据失败: {e}")
return None
# 调用
depth = fetch_order_book("ETHUSDT", limit=5)
if depth:
print(f"买单总深度: {depth['bid_volume']:.2f} ETH")
print(f"卖单总深度: {depth['ask_volume']:.2f} ETH")你看，深度数据里每个订单都是一个列表，第一个元素是价格，第二个是数量。这里又回到刚才说的精度问题——价格和数量都是字符串，记得转成 Decimal 再计算。
3.6 知识体系总览
下面这张图把 REST API 调用的核心流程串起来了，你可以对照着理解：
这张图把整个调用链路画清楚了。你写代码的时候，就按这个流程来，每一步都做检查，基本不会出大问题。
3.7 避坑指南
最后分享几个我踩过的坑：
- 超时设置不能省：默认情况下 requests 会一直等，网络波动时你的程序就卡死了。我习惯设 3-5 秒超时。
- 重试要有退避策略：失败后立即重试往往还是失败。等 1 秒、2 秒、4 秒这样指数退避，效果更好。
- 别把 API Key 写死在代码里：用环境变量或者配置文件管理。我见过有人把 Key 提交到 GitHub 上，结果被机器人扫到，账户被盗刷。
- 日志要打全：请求的 URL、参数、响应状态码、耗时，这些信息都记下来。出问题的时候，日志就是你最好的朋友。
好了，REST API 实战这块就讲到这里。代码不多，但每个细节都值得你亲手敲一遍。下一节咱们会聊 websocket 实时数据流，那个又是另一番天地了。
公众号:蓝海资料掘金营，微信deep3321