构建 Python 库时,目标是将复杂的,繁重的过程 (原始 API 调用) 转化为简单,优雅的单行本. 汇率数据API 提供主要货币对的实时宏观经济指标,为量交易者和分析师提供了金矿.
采用了"Raw API"来强迫开发人员重复代码进行身份验证,错误检查和URL构建. 干燥 (不要重复自己) 根据这个原则,我开始在上面构建一个专门的Python库,创建一个用户友好的包装. 本文将介绍该包装的核心组件,包括同步和异步客户端,适当的异常处理和实用功能.
1. 项目脚手架和处理认证
一个好的库从一个直观的入口点开始. 我的目标是把HTTP请求变成一个干净的Python方法调用, client.get("aud", "inflation")现在我们要做什么?
客户端构造商
没有什么. Client 类包含了基础URL和API键.FXMacroData API具有一个独特的功能: 美元数据是公开的,但其他货币需要API密钥. 制造商会提前处理这个要求.
# client.py or async_client.py
from typing import Optional
from .exceptions import FXMacroDataError
class Client:
BASE_URL = "https://fxmacrodata.com/api/v1/announcements"
def __init__(self, api_key: Optional[str] = None):
"""
Synchronous FXMacroData Client.
api_key: Required for non-USD currencies. USD is public.
"""
self.api_key = api_key
2.核心逻辑:同步客户端 (Client) 没有
没有什么. 时间同步 Client 使用了流行 requests 图书馆. 主要逻辑在于 get 方法,它动态地构建URL并执行API密钥要求.
没有什么. get 方法:动态URL构建和关键检查
# client.py
def get_indicator(
self,
currency: str,
indicator: str,
start_date: Optional[str] = None,
end_date: Optional[str] = None,
) -> dict:
currency = currency.lower()
url = f"{self.BASE_URL}/{currency}/{indicator}"
headers = {}
if currency != "usd":
if not self.api_key:
# Custom exception is crucial for user-friendly errors
raise FXMacroDataError(f"API key required for {currency.upper()} endpoints.")
headers["X-API-Key"] = self.api_key
params = {}
# ... params and API call logic ...
强大的错误处理自定义异常
强大的库必须优雅地处理故障.我创建了一个自定义例外, FXMacroDataError捕获网络问题和非200状态码,返回一个清晰的,可操作的信息.
# exceptions.py
class FXMacroDataError(Exception):
"""Custom exception for FXMacroData client errors."""
pass
核心请求逻辑与错误包装:
# client.py (continued)
try:
response = requests.get(url, headers=headers, params=params)
except Exception as e:
raise FXMacroDataError(f"Request failed: {e}")
if response.status_code != 200:
# Raise a clear error if the API returns a problem
raise FXMacroDataError(f"API Error ({response.status_code}): {response.text}")
return response.json()
3.高级功能:异步客户端 (AsyncClient) 没有
对于自动交易机器人或高流量仪表板, 异步编程 对于表现至关重要. AsyncClient 使用 aiohttp 库为不阻塞I/O.
异步会话管理
我已经实现了异步语境管理器 (__aenter__ 现在我 __aexit__确保 aiohttp.ClientSession 防止资源泄漏.
# async_client.py
import aiohttp
# ... imports ...
class AsyncClient:
# ... init ...
async def __aenter__(self) -> "AsyncClient":
self.session = aiohttp.ClientSession()
return self
async def __aexit__(self, exc_type, exc_val, exc_tb) -> None:
if self.session:
await self.session.close()
self.session = None
这允许同时执行,总时间是最大延迟,而不是总数:
import asyncio
from fxmacrodata import AsyncClient
async def main():
async with AsyncClient(api_key="YOUR_KEY") as client:
# Concurrent calls are now trivial
data_aud = client.get_indicator("aud", "inflation")
data_eur = client.get_indicator("eur", "gdp")
aud_data, eur_data = await asyncio.gather(data_aud, data_eur)
# ...
4. 实用性:清理数据
数据消费者期望时间序列排序的数据,但API并不总是保证. 一个小的实用功能确保输出始终是干净的时间序数数据,检查一个 'date' 没有 'release_date' 关键是
# utils.py
def sort_by_date(data_list):
"""Sorts a list of indicator data dictionaries by 'date' or 'release_date'."""
return sorted(data_list, key=lambda x: x.get('date') or x.get('release_date'))
打造这个包装,巩固了我对 面向对象的设计对于果的关键性 同步与异步 网络,以及一个伟大的重要 开发人员的经验 通过自定义例外.你可以在 基特库现在我们要做什么?
最后一步是包装库并发布到PyPI. 如果你正在构建一个工具来集成实时FX宏数据,或者只是想要一个模式来创建自己的包装,
罗伯 @ FXMacroData
