关键词：Python、Polygon API、历史行情、K线、加密货币、股票、期货、外汇、RESTClient、pandas
自2019 年以来，Polygon 以 实时、低延迟、全栈覆盖 成为量化交易与数据科学圈最受欢迎的 市场数据 API 提供商。本文手把手教你 纯 Python 方式 连接 Polygon，5 分钟完成环境配置、代码封装，并轻松批量获取加密、股票、期货、外汇的 分钟级历史 K 线，做到 开箱即用。
1. 快速安装：三步打包 Polygon
# 1. 创建虚拟环境
python -m venv polygon_env
source polygon_env/bin/activate # Windows 用 polygon_env\Scripts\activate
# 2. 安装 Polygon 官方客户端
pip install -U polygon-api-client
# 3. 准备 API Key
# 将 Key 保存到 local_settings.py，便于安全 gitignore
# local_settings.py 内容：
# polygon = {"api_key": "YOUR_API_KEY_HERE"}👉 获取 Polygon API 官方文档与实时行情预览 无需注册即可查看示例代码。
2. 拓展 RESTClient：自带重试、超时控制
Polygon SDK 的 RESTClient 使用简单，但生产环境需要 重试策略 和 网络超时。我们继承并额外嵌入 Retry Adapter 以实现 429 与 5xx 场景的无感重试。
from datetime import date, datetime
import pandas as pd
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
from polygon import RESTClient
from local_settings import polygon as settings
ALLOWED_MARKETS = ['crypto', 'stocks', 'fx', 'futures'] # 支持市场列表
class RobustPolygon(RESTClient):
"""带智能重试的 Polygon 客户端"""
def __init__(self, auth_key: str = settings['api_key'], timeout: int = 10):
super().__init__(auth_key, connect_timeout=timeout)
retry = Retry(
total=10,
backoff_factor=0.5,
status_forcelist=[429, 500, 502, 503, 504],
raise_on_status=False,
)
adapter = HTTPAdapter(max_retries=retry)
self._session.mount('https://', adapter)3. 轻松抓取资产列表
交易系统的第一步，是建立资产清单。Polygon 的 reference_tickers_v3 能一次性返回所有股票、加密、外汇、期货的 代号、名称、市场状态。以下封装自动去重并精简字段：
import warnings, itertools, math
class RobustPolygon(RobustPolygon):
def get_tickers(self, market: str) -> pd.DataFrame:
if market not in ALLOWED_MARKETS:
raise ValueError(f"market 必须是 {ALLOWED_MARKETS} 之一。")
resp = self.reference_tickers_v3(market=market)
df_list = []
while True:
if not hasattr(resp, 'results'):
break
df_list.extend(resp.results)
if not hasattr(resp, 'next_url'):
break
resp = self.reference_tickers_v3(next_url=resp.next_url)
df = pd.DataFrame(df_list)
if market == 'crypto': # 仅保留 USD 计价
df = df[df['currency_symbol'] == 'USD']
df['name'] = df['base_currency_name']
df = df[['ticker', 'name', 'market', 'active']].drop_duplicates('ticker')
return df.reset_index(drop=True)
# 实战：拉取加密货币列表
cli = RobustPolygon()
crypto_df = cli.get_tickers('crypto')
print(crypto_df.head(10))返回片段：
ticker name market active
0 X:1INCHUSD 1inch crypto True
1 X:AAVEUSD Aave crypto True
15 X:BTCUSD Bitcoin crypto True
26 X:ETHUSD Ethereum crypto True👉 点击预览 BTC 实时分钟级成交量与价格衍化曲线 直接命中交集行情。
4. 精准下载 K 线：一张代码刷全量
Polygon 聚合接口 crypto_aggregates、stocks_aggregates 支持 日、分钟、秒级 自定义粒度。以下通用封装解决三个痛点：
- 分页自动循环：单次 50,000 条也是 秒级循环。
- 去冗余断点续传：按最后时间戳续传，避免重复拉取。
- 数据结构标准化：统一命名、统一时区、缺少列补空。
import time
class RobustPolygon(RobustPolygon):
def get_bars(self,
market: str,
ticker: str,
multiplier: int = 1,
timespan: str = 'minute',
frm: str | date = None,
to: str | date = None) -> pd.DataFrame:
# 检查参数
if market not in ALLOWED_MARKETS:
raise ValueError(f"market 必须是 {ALLOWED_MARKETS} 之一。")
if not ticker:
raise ValueError("ticker 不能为空。")
frm = frm or date(2000, 1, 1)
to = to or date.today()
# 统一日期格式字符串
start_str = frm.strftime('%Y-%m-%d') if isinstance(frm, date) else frm
end_str = to.strftime('%Y-%m-%d') if isinstance(to, date) else to
# 选择对应市场聚合端点
if market == 'crypto':
api_fn = self.crypto_aggregates
elif market == 'stocks':
api_fn = self.stocks_aggregates
else:
raise NotImplementedError("当前仅支持 crypto 与 stocks。")
bars, last_ts = [], 0
while True:
resp = api_fn(ticker, multiplier, timespan, start_str, end_str, limit=50000, sort='asc')
if not hasattr(resp, 'results') or not resp.results:
break
bars.extend(resp.results)
last_ts = resp.results[-1]['t']
start_str = datetime.utcfromtimestamp(last_ts / 1000).strftime('%Y-%m-%d')
if len(resp.results) < 50000:
break
# 轻微限制速率，避免 429
time.sleep(0.05)
if not bars:
return pd.DataFrame(columns=['date', 'open', 'high', 'low', 'close', 'volume'])
df = pd.DataFrame(bars)
df['date'] = pd.to_datetime(df['t'], unit='ms')
# 标准化列名
rename_map = {'o': 'open', 'h': 'high', 'l': 'low', 'c': 'close', 'v': 'volume'}
df = df.rename(columns=rename_map)
return df[['date', 'open', 'high', 'low', 'close', 'volume']].reset_index(drop=True)
# 案例：抓取 2021-01-01 至 2021-12-31 BTC 1 分钟 K 线
btc_df = cli.get_bars('crypto', 'X:BTCUSD', timespan='minute', frm='2021-01-01', to='2021-12-31')
print(btc_df.shape, btc_df.date.min(), btc_df.date.max())
# -> (524160, 2021-01-01 00:00:00, 2021-12-31 23:59:00)5. FAQ｜90% 新手疑问一站式解答
Q1：免费额度够用吗？
免费层每日命中 5 次聚合请求，仅能抓 2 年加密分钟级数据。付费层约 0.0001 美元 / 请求，百美元即可吃透全部加密历史，性价比极高。
Q2：为何提示 429？
你已触发速率限制。RobustPolygon 自带 Retry 已经优化，但仍需注意 单次 50k 上限。切记循环中 sleep。
Q3：秒级数据时间戳精度是多少？
毫秒级，但在股票时间 9:30:00.001 与 9:30:00.002 即有两条。可再参数 timestamp_round='second' 自行压缩。
Q4：证券交易所 与 加密交易所** 的粒度差异？
Polygon 加密端默认 UTC；美股端默认为美东（EST/EDT）；调用 df['date'] = df['date'].dt.tz_localize('UTC') 后自行转换即可。
Q5：如何批量更新？
封装 check_last_record → 断点续传，再配 date.today() 每日 cron。把 DataFrame 写至 Parquet，避免重复 CSV。
Q6：如何接入 Qlib、backtrader？
Polygon 直接输出 pandas DataFrame，存 numpy.save 或 zarr 即可。一套代码适配任何 机器学习框架。
6. 实战加速：10 行代码跑回测
将 分钟数据 与 高频能级标记 转存为 OHLC 文件，送入 backtrader 回测仅需 10 行。
df = cli.get_bars('crypto', 'X:ETHUSD', 5, 'minute', '2023-01-01', '2023-03-31')
df.to_csv('ETHE_min5.csv', index=False)
# backtrader 示例省略 …7. 结语：把 Polygon 当作 “数据高速公路”
本文仅用两个扩展函数就把 Polygon 100% 实用场景 打通：
- 资产列表 → 数据字典
- K 线循环 → 回测引擎
不要犹豫，现在就用 3 分钟跑通第一条策略。
👉 立刻体验零门槛行情订阅，马上用布林带脚本抓下一轮 Alpha。