3. 数据源接入基础
在金融智能体的工程实践中，数据源接入是决定智能体“能否看到真实世界”的第一道关口。无论是行情数据、财报公告、宏观经济指标，还是舆情与研报，最终都要通过 API 或文件接口进入智能体的检索增强链路。本节聚焦接入层最核心的五个问题：请求如何发出、身份如何证明、频率如何控制、数据如何解析、异常如何兜底。
一、API 接入原理：一次请求的完整生命周期
金融数据 API 大多基于 HTTP/HTTPS 协议，采用 RESTful 风格。一次典型调用包含四个阶段：构造请求、建立连接、服务端处理、返回响应。理解这条链路，才能在出错时快速定位是网络层、认证层还是业务层的问题。
- 请求构造：确定 URL、方法（GET/POST）、请求头（Header）、查询参数（Query）或请求体（Body）。
- 连接与传输：DNS 解析、TLS 握手、发送请求，金融场景通常强制 HTTPS。
- 服务端处理：网关鉴权、限流、路由到业务服务、查询数据库或行情源。
- 响应返回：状态码、响应头（含限流信息）、响应体（JSON/XML/CSV）。
下面是一个使用 Python requests 调用行情接口的最小示例，注意请求头中携带了认证信息：
import requests
url = "https://api.example-finance.com/v1/quote"
headers = {
"Authorization": "Bearer eyJhbGciOiJIUzI1NiIs...",
"Accept": "application/json"
}
params = {"symbol": "600519.SH", "fields": "open,close,volume"}
resp = requests.get(url, headers=headers, params=params, timeout=5)
print(resp.status_code, resp.headers.get("X-RateLimit-Remaining"))
二、认证方式：从 到 OAuth 2.0
金融数据通常具有商业价值，认证是接入的前提。常见方式有以下几类，选择时需权衡安全性与实现成本。
| 认证方式 | 典型场景 | 优点 | 风险点 |
|---|---|---|---|
| （Header/Query） | 公开行情、内部服务 | 实现简单 | 易泄露，需配合 HTTPS |
| Basic Auth | 传统数据商接口 | 兼容性好 | Base64 非加密，必须走 TLS |
| Bearer / JWT | 多数现代金融 API | 可携带过期与权限声明 | 需处理刷新与时钟偏移 |
| OAuth 2.0 | 第三方授权、券商开放平台 | 权限粒度细、可撤销 | 流程复杂，需管理 生命周期 |
| 签名（HMAC/RSA） | 交易所、支付类接口 | 防篡改、防重放 | 实现与调试成本高 |
三、请求频率限制：限流模型与应对策略
数据提供方为了保护服务稳定，通常会对调用频率设限。常见限流模型包括：固定窗口、滑动窗口、令牌桶与漏桶。响应头中常出现 X-RateLimit-Limit、X-RateLimit-Remaining、Retry-After 等字段，是判断是否触发限流的关键信号。
应对限流的工程手段主要有四种：
- 客户端节流：在请求间加入固定间隔，如每秒不超过 5 次。
- 指数退避重试：遇到 429 时按 1s、2s、4s 递增等待，避免雪崩。
- 批量与缓存：合并多标的请求，对不常变的数据做本地缓存。
- 多密钥轮换：在合规前提下分散配额，但需注意服务商条款。
import time, random
def call_with_backoff(fn, max_retry=5):
for i in range(max_retry):
resp = fn()
if resp.status_code == 429:
wait = (2 ** i) + random.uniform(0, 0.5)
time.sleep(wait)
continue
return resp
raise RuntimeError("rate limit exceeded after retries")
四、数据格式解析：JSON、XML 与 CSV
金融数据源的返回格式差异很大。JSON 是现代 API 的主流，结构灵活；XML 常见于老牌数据商与监管报送；CSV 多用于批量下载与历史数据。解析时要关注编码、缺失值、时间格式与数值精度。
- JSON：使用
json.loads或resp.json()，注意嵌套层级与 null 值。 - XML：使用
xml.etree.ElementTree或lxml，注意命名空间。 - CSV：使用
pandas.read_csv，指定encoding与dtype防止精度丢失。
import pandas as pd
# CSV 解析：指定编码与数值类型
df = pd.read_csv("history.csv", encoding="utf-8",
dtype={"symbol": str, "close": float},
parse_dates=["trade_date"])
print(df.head())
五、错误处理与复盘：构建健壮的接入层
接入层的错误可分为四类：网络错误（超时、连接失败）、认证错误（401/403）、限流错误（429）、业务错误（参数非法、标的不存在）。一个健壮的接入层应当对每类错误有明确的处理策略，并记录足够的上下文用于复盘。
下面这张图展示了从请求发出到结果落库的完整判断路径，包括认证、限流、解析与重试分支，可作为接入层设计的参考蓝图。
复盘时建议关注三个指标：成功率、平均延迟、重试占比。若某数据源成功率骤降，优先检查密钥是否过期、配额是否耗尽、接口是否变更字段。常见误区包括：把 429 当成普通失败直接丢弃、忽略响应头中的限流信息、解析 CSV 时不指定编码导致中文乱码、以及对时间字段不做时区统一，最终在检索增强阶段造成数据错配。