做量化回测和盯盘工具的朋友都知道策略模型能不能跑出结果关键不在花里胡哨的指标而在数据链路是否稳。加密货币历史数据和实时行情接口这两样东西看起来免费入口到处都是真正要稳定拿到手却有不少坑。我这两年帮自己和朋友做过好几个看板和量化小项目从免费接口选型、批量拉取到实时推送都折腾了一遍。这篇文章就把整条链路拆开讲告诉你哪些免费接口值得用、为什么值得用、每一步的坑在哪里。适合正在做回测、计划做实时告警、或者想给图表工具接行情源的同学参考。1. 免费行情接口的整体选型与思路1.1 先区分历史数据与实时行情的不同要求很多人在选接口的时候上来就问“哪个平台免费”其实更该问的是“我需要哪类数据”。历史数据和实时行情的接口策略完全不同选错方向后面全是麻烦。历史数据指的是已经发生的K线、收盘价、成交量、逐笔成交等主要喂给回测引擎做回放也用来画走势图。历史数据接口看重三件事深度、连续性、限频容忍度。所谓深度就是能不能拿到很久以前的数据最好是从项目上线第一根K线开始连续性是指K线之间不能有空缺否则回测结果会出现脏点限频容忍度则是你批量拉取时能不能承受大量请求。实时行情指 ticker、最新成交价、订单簿深度这类“正在发生”的数据。它看重延迟、推送频率、连接稳定性。用 REST 接口轮询能做到秒级用 WebSocket 能做到毫秒级。如果你做的是逐笔监控或者分钟级策略选了轮询方案还没开始写策略就已经输了。所以在动手之前先把自己的使用场景写清楚做日频回测做分钟级策略做秒级价格告警还是只做一个5个币种的简单看板场景不同接口选型和数据落地方式都不一样。1.2 主流免费行情接口横向对比下面是我自己实际用过或者调研过的免费接口整理。为了避免写成一篇随时过期的文档我在表后附上“以官方最新文档为准”的原则但这个阵列基本能帮你快速形成判断。数据源类型是否需要Key免费额度特点历史数据深度实时推送适合场景CoinGecko聚合行情免费Key有限频每分钟请求有限适合轻量调用常见币种多年价格粒度有限无强实时推送币种榜单、市值查询、简单趋势图CoinCap聚合行情无需Key公开限频适合中小规模有历史价格曲线接口提供公共WebSocket价格推送原型验证、中小看板、价格机器人主流交易所Public API交易所原生一般无需Key权重限频比聚合平台更宽松从项目上线到当前全量K线WebSocket流稳定回测取数、自建实时行情服务Alpha Vantage聚合行情需要Key每日次数有限日/周/月线无良好实时推送少量币种的长期趋势分析为什么我更推崇交易所原生接口因为它直接把最完整的K线字段给你时间戳、开高低收、成交量、成交笔数、主动买量一应俱全而且免费额度通常按权重计算只要你会控制节奏实际拉取量可以非常大。聚合平台胜在“多币种统一”但历史深度和实时推送往往只是“够用”而不是“好用”。提醒一下免费接口的额度条款经常调整你在参考任何人文章时都不要把“每分钟能请求多少次”当成永久事实。我自己的习惯是接到项目后先去看官方文档的 Rate Limit 部分再写一个最小测试脚本实测一遍。1.3 为什么不建议自己写爬虫抓网页我完全能理解为什么有人想用爬虫去抓行情网站的页面数据因为视觉上最直观好像什么字段都有。但实际上这是性价比最低的方案原因有三第一网页结构说改就改。昨天还在的 class 今天换个名字你的定位器就彻底失效第二反爬手段多。有些网站会做限流、浏览器指纹检测、验证码你要花大量时间在“对抗”上而不是在数据业务上第三网页拿到的数据往往是格式化后的字符串缺少可靠的时间戳和原始成交属性做回测时需要反复清洗精度还不够。正规接口返回的是结构化 JSON字段有明确含义时间戳是标准的毫秒数这才是工程上该走的路径。接口要付费的情况确实有但免费入口并不少关键是把通用方法学会然后按需替换数据源。2. 核心概念与接口原理2.1 读懂K线历史数据的返回结构免费接口再怎么五花八门历史K线的核心字段都大同小异。下面是一段典型的交易所K线返回数组我加了注释[ 1507024800000, 0.01630000, 0.05000000, 0.01630000, 0.04000000, 1000.00000000, 1507024860000, 1000.00000000, 10, 0.02000000, 0.01000000, 0.00000000 ]字段含义从左到右分别是K线开盘时间毫秒时间戳、开盘价、最高价、最低价、收盘价、成交量、K线收盘时间、成交额、成交笔数、主动买入量、主动买入额、忽略字段。有两点值得注意。一是价格和成交量通常是字符串不是数字。这类字段如果直接做算术运算很容易出现类型错误正确做法是先转成 Decimal 或者 float 再计算。二是时间戳是毫秒级 Unix 时间戳它本身不包含时区信息。严格来说任何时区下表示的都是同一个瞬间只是你把它显示成“北京时间”还是“UTC时间”的区别。回测入库时我建议把 openTime 直接存成 INT 类型同时再存一个可读的时间字段比如 UTC 的 ISO 格式。这样做既能保证排序效率也方便人眼排查问题。2.2 限频权重与请求合并免费接口大部分不是“按次数限频”而是“按权重限频”。每个接口的耗时程度不同权重也不同。比如拉取1000根K线可能消耗权重2而查询账户信息可能消耗权重20。这样设计是为了防止有人用高频操作阻塞服务。正确姿势是“少请求、多取数”。比如拉历史K线时一次请求尽量拉满允许的最大 limit宁可循环几次也不要拆成几百个小请求。如果你要订阅多个交易对能用一条 WebSocket 连接批量订阅就不要为每个交易对建一条连接。这里还要提一个常见误区有些人写代码时每次循环都发一次 HTTP 请求然后 sleep 0.1 秒看起来“很温柔”但实际权重消耗非常多而且容易被限频。更聪明的做法是把你的数据按时间分片每个分片尽量大到接近接口上限再配合缓存机制把真实的请求量降下来。2.3 时间缝隙与未收盘K线历史K线接口的一个隐形坑是最后一根K线经常“还没定型”。比如你拉取 1 分钟K线当前时间的这根K线会在 1 分钟结束前被反复更新开盘价确定了但收盘价还在变化。如果你在这个过程里把它当成最终历史数据入库回测就会用到未来函数。处理办法有两个层面。第一在你做实时增量更新时识别到同一个 openTime 的K线再次推送不要插入新记录而是更新旧记录。第二做回测取数时宁可选择“时间上已经完整收盘”的K线例如拉历史数据时把 endTime 设为当前时间减去一个K线周期拒绝最后一根不完整K线。我在实际项目里的做法是数据库里给K线表加一个状态标志已经收盘的历史K线标记为 final实时流推过来的未收盘K线标记为 pending。回测引擎只读取 final看板则把 pending 也显示出来。3. 实操用 Python 拉取免费历史数据并落地3.1 环境准备本文代码以 Python 为基础你只需要安装 requests 和 pandas。如果你做轻量落地可以只用标准库但 pandas 处理K线数组确实方便得多。python -m venv venv source venv/bin/activate # Windows 下用 venv\Scripts\activate pip install requests pandas建议你在项目目录下建一个 fetch_history.py把拉取函数封装成独立模块方便后续接入不同的数据源。因为不同平台的域名、字段名会有差异模块化之后替换数据源时只需要改一个 adapter。3.2 拉取一个时间窗口的K线下面这段代码用公共行情接口拉取某一交易对的历史K线。以某交易所公开文档中的标准端点为示例你使用之前务必替换为当前可访问的接口域名。import requests import pandas as pd BASE_URL https://api.example.com/api/v3/klines def fetch_klines(symbolBTCUSDT, interval1h, start_msNone, end_msNone, limit1000): params { symbol: symbol, interval: interval, limit: limit, } if start_ms: params[startTime] start_ms if end_ms: params[endTime] end_ms resp requests.get(BASE_URL, paramsparams, timeout10) resp.raise_for_status() data resp.json() df pd.DataFrame(data, columns[ open_time, open, high, low, close, volume, close_time, quote_volume, trades, buy_base_volume, buy_quote_volume, ignore ]) # 把字符串价格转成浮点 for col in [open, high, low, close, volume, quote_volume]: df[col] df[col].astype(float) return df这里有个细节返回的 JSON 是二维数组用 pandas 直接指定列名最方便。但要注意数组里的空值或者字段顺序变化某些平台会在后续版本增加字段。你可以在解析前先判断列数如果小于预期就做兼容处理。请求超时建议设置 10 秒以上大网络环境下偶发抖动很正常超时后可以重试。不要用默认超时否则程序可能卡死很久。3.3 循环分片拉取全量历史数据并校验连续性单个请求最多返回 1000 根K线要拉一整年甚至更早的数据就必须按时间窗口循环。这里的关键逻辑是下一次请求的 startTime 本次最后一根K线的 open_time interval 对应的毫秒数。某些平台支持按 endTime 往前翻但用 startTime 向后翻最通用。def fetch_history_until(symbol, interval, start_ms, end_msNone): interval_ms { 1m: 60_000, 5m: 300_000, 15m: 900_000, 1h: 3_600_000, 4h: 14_400_000, 1d: 86_400_000, } step interval_ms.get(interval, 60_000) all_frames [] cursor start_ms while True: if end_ms and cursor end_ms: break df fetch_klines(symbol, interval, start_mscursor, limit1000) if df.empty: break all_frames.append(df) last_open_time df[open_time].iloc[-1] cursor last_open_time step # 止损保护 if cursor - start_ms 5_000 * step: break result pd.concat(all_frames, ignore_indexTrue) result result.drop_duplicates(subsetopen_time) return result.sort_values(open_time).reset_index(dropTrue)你可能会问为什么最后要做 drop_duplicates因为并发重试或者接口异常返回时重叠时间区域很可能拉回相同K线去重是写进数据链路里最便宜的一道保险。连续性校验同样重要。拿到完整数据后你要检查相邻两根K线的 open_time 差值是否严格等于 interval。如果出现空洞就需要回到接口单独补那一小段区间而不是整个重拉。def check_gaps(df, interval_ms): diff df[open_time].diff().dropna() gaps diff[diff ! interval_ms] return gaps3.4 落盘CSV 还是 SQLite历史数据落地常见两种方案CSV 和 SQLite。CSV 的好处是文件小、可读性好、Excel 直接打开缺点是增量更新麻烦每次都要全量重写。SQLite 的好处是支持 SQL 去重、条件更新相对轻松适合长期维护的数据仓库。我的建议是如果只是临时拉一份数据跑回测CSV 够用如果这个数据要每天定时更新、还要边拉边增量那就用 SQLite。建表语句可以参考下面这样CREATE TABLE IF NOT EXISTS klines ( symbol TEXT NOT NULL, interval TEXT NOT NULL, open_time INTEGER NOT NULL, open REAL, high REAL, low REAL, close REAL, volume REAL, close_time INTEGER, PRIMARY KEY (symbol, interval, open_time) );用(symbol, interval, open_time)做联合主键天然就能防止重复。写入时用INSERT OR REPLACE实时行情推过来的未收盘K线也能直接覆盖不会产生脏数据。这个设计是我在用了一段时间 CSV 之后才改的强烈推荐。4. 实时行情接口WebSocket 订阅与增量更新4.1 REST 轮询和 WebSocket 怎么选实时行情的获取有两个方向REST 轮询和 WebSocket 长连接。REST 轮询很好理解每隔几秒请求一次最新价格。优点是实现简单缺点是延迟高、请求密度大、容易触发限频。适合那些“慢半拍也无所谓”的场景比如每5分钟刷新一次某个币的价格。WebSocket 则是客户端和服务端建立一条长连接服务端有新数据时主动推给你。它的延迟可以到毫秒级而且不会因为订阅多个交易对而线性增加请求次数。适合实时告警、逐笔滚动、盘口刷新等场景。对比项REST 轮询WebSocket 推送延迟秒级毫秒级请求消耗高低实现难度低中多交易对扩展每个交易对都要请求一次一条连接订阅多个符号断线恢复无需恢复需要自己处理重连如果你需要实时K线画图建议用 WebSocket 订阅K线流来做增量再用历史数据接口拉一次全量K线做底图。这样既有历史深度又有实时速度。4.2 一个可以直接运行的 WebSocket 示例下面用两个不同层级的例子。第一个例子是最简单的无 Key 公共订阅适合快速验证第二个例子是交易所K线流的典型订阅方法适合正经项目。先看无 Key 的公共价格流。CoinCap 官方文档提供了这种订阅方式只要你网络能访问它代码几乎不用配置from websocket import create_connection ws create_connection(wss://ws.coincap.io/prices?assetsbitcoin,ethereum) for _ in range(5): message ws.recv() print(message) ws.close()返回的消息是类似这样的 JSON{bitcoin: 66812.123, ethereum: 3420.456}这种形式的接口适合做简单价格展示但毛病也很明显它不区分交易对和K线周期只给你一个“现价”做不了细致的历史分时。所以真正做K线级实时行情还是要看交易所的标准订阅协议。交易所 WebSocket 的通用逻辑通常是建立连接后先发送订阅指令再持续接收服务端推送。可以参考下面这段框架接口域名以你选择的交易平台官方文档为准import json from websocket import WebSocketApp def on_message(ws, message): data json.loads(message) # 不同的流字段结构不同这里以常见的K线事件为例 k data.get(k) if k: print( fopenTime{k.get(t)} close{k.get(c)} ) def on_error(ws, error): print(error:, error) ws WebSocketApp( wss://stream.example.com/ws/btcusdtkline_1m, on_messageon_message, on_erroron_error, ) ws.run_forever()这里的核心动作是run_forever()。它会阻塞当前线程不断接收数据。如果你在回调里做数据库写入要注意写入速度赶不上推送速度的情况必要时用队列缓冲让消费者线程慢慢落库。4.3 心跳保活与断线重连WebSocket 使用中最常见的故障是“连接还挂着但已经收不到数据了”。很多服务端会在一段时间内没有收到任何客户端消息时主动断开或者网络设备把空闲连接清理掉。解决办法是定时发送心跳消息。交易所类连接一般支持发送一个简单的 PING 帧或者 JSON 格式的 ping 消息。如果没有标准心跳指令一个最保底的做法是记录last_message_time用一个后台线程检查如果距离上次收到消息超过阈值就主动重连。下面是一个加了心跳守护的框架import threading import time from websocket import WebSocketApp class KlineStream: def __init__(self, url): self.url url self.last_msg time.time() self.ws None def on_message(self, ws, message): self.last_msg time.time() # 处理数据... def on_open(self, ws): def send_ping(): while True: time.sleep(20) try: ws.send(ping) except Exception: break t threading.Thread(targetsend_ping, daemonTrue) t.start() def run(self): self.ws WebSocketApp( self.url, on_messageself.on_message, on_openself.on_open, ) while True: self.ws.run_forever() print(连接断开5秒后重连...) time.sleep(5) self.last_msg time.time()重连逻辑里有个细节直接调用run_forever()在断线后不会自动退出所以用while True包裹并在外面重新赋值WebSocketApp。定时线程也要注意防止重复启动否则每次重连都会多一个线程发心跳。4.4 实时数据如何与历史数据拼接使用实时K线流最忌讳的一件事是“闭着眼插入”。因为一根K线在它所在周期内会被推送很多次收盘价一直在变化而你库里已经有了开盘时拉到的一条记录。不做更新只做插入很快就会产生大量重复行。正确做法是把实时事件里的 openTime 作为主键和正在更新的K线进行匹配。如果数据库里没有这根K线就插入如果已经有就覆盖最新价格和成交量。这个操作在 SQLite 里就是INSERT OR REPLACE在内存里就是 update 一个 dict。我自己在写盘时还会加一个“是否已收盘”判断当消息里的 K线 closeTime 小于当前时间说明这根K线已经定型就不再接受后续更新了。这样可以避免在数据落地后还看到同一根K线的数值反复跳变。5. 常见问题与排查技巧实录5.1 接口返回 429 或 418被限频了怎么办429 是请求太频繁418 表示你已经触发了更严格的限制需要等待一段时间。这两个状态码几乎是免费接口用户最容易遇到的。遇到 429 不能直接“不休息地重试”那样只会更糟。正确做法是记录返回头里的限频信息或者按固定退避时间重试。下面是一个最简单的指数退避实现import time from requests import RequestException def get_with_retry(session, url, params, retries5): wait 1 for attempt in range(retries): resp session.get(url, paramsparams, timeout10) if resp.status_code 429: time.sleep(wait) wait * 2 continue resp.raise_for_status() return resp.json() raise RequestException(retry exceeded)退避的倍数一般取 2 或者 3。如果你同时跑多个线程限频概率会成倍上升。更优雅的做法是把所有请求收敛到同一个调度器里做一个全局的“最低间隔”控制。5.2 历史数据中间缺了一段出现数据空洞常见原因有两个一是分页游标参数算错了比如把 endTime 当成毫秒还是秒搞混二是请求被限频后某一段没有成功返回但代码没有检测到就直接拼接了。排查的步骤我整理成下面这条路径先对拼接好的 DataFrame 做 open_time diff 检测确认空洞范围。用空洞前后两根K线的时间戳作为新的startTime和endTime单独拉一次。如果单独拉取依然缺失检查该交易对在那个时间段是否真的开盘。某些币种上市时间并不早那不是数据空洞而是根本不存在数据。我曾经遇到过一只新上线的代币拿它半年前的数据时一直缺一段后来才发现它上市时间比我的起始时间晚前面那一整段自然不会有数据。这个错误非常典型。5.3 时间戳看起总差 8 小时毫秒时间戳本身没有时区概念但你把它转成字符串时如果不指定时区很多本地化方法会把 UTC 时间当作本地时间显示。如果你的系统时区是东八区那显示出来的时间就会比真实时间快 8 小时。正确的入库策略是数据库里只存 UTC 时间或者直接存毫秒整数展示层要做本地化转换时再统一加时区。不要在清洗数据的过程中就把时间格式化成“YYYY-MM-DD HH:mm:ss”因为一旦你失去了毫秒精度后续做区间查询会非常痛苦。5.4 WebSocket 连上之后却没有数据这种问题通常不是“没连上”而是“订阅没成功”。很多平台的 WebSocket 需要你连接后先发一条订阅指令服务端收到后才会开始推数据。如果你只是建立了连接然后傻等当然收不到任何消息。排查时可以分三看一看返回的消息里有没有订阅回执比如result: null, id: 1二看订阅符号是否准确比如BTCUSDT和btcusdt在不同平台可能大小写规则不同三看是否订阅到了正确频道K线、成交、ticker是三个完全不同的流。顺带说一个常见误操作订阅多个交易对时要看平台是否支持streams...参数不支持的话只能一条连接里发一个数组订阅发错格式一样会没数据。5.5 接口字段改动导致程序崩溃这类问题是免费接口最让人头疼的。某个平台某天在数组末尾加了一个字段你的硬编码取数逻辑就可能越界或者把字符串价格送到数学函数里当场抛异常。建议所有解析层都做一个“宽容读取”。也就是优先使用字典的.get()不要用data[field]直接访问。如果字段缺失至少不要让整个程序挂掉而是记录一条 warning 并跳过这一条。样例代码如下def safe_float(value, default0.0): try: return float(value) except (TypeError, ValueError): return default这种“防御性编程”在对接第三方免费接口时不是可选项而是必须项。因为免费接口没有严苛的维护承诺字段变化只在一夜之间。5.6 免费接口的使用边界与合规提示最后这部分我想认真提醒一句免费接口不是公有领域数据它有使用条款。不同平台的免费额度条款会写清楚你可以用它做什么、不能做什么。我个人的原则是个人研究、学习、内部看板用途完全没问题公开转售、商业化分发原始行情、批量打包数据提供给第三方基本都不在免费条款允许范围内。同时加密货币相关业务本身受不同国家和地区的法规约束。请确保你的使用行为符合你所在地区的法律法规以及服务商条款。本文所有内容仅用于技术研究不构成任何投资建议也不鼓励任何违规交易行为。就我个人实际跑下来的体会来说历史数据用交易所 REST 接口批量落盘 SQLite实时行情用 WebSocket 做增量更新两者用(symbol, interval, open_time)这个联合主键串起来是一套性价比很高、维护成本也很低的组合。真正值得花时间的不是反复换接口而是把数据校验、去重、断线重连这套底座磨扎实。你把这套链路跑顺之后再去接任何新平台剩下的其实只有字段映射和限频适配。 
长河君 · 2026/10/8 14:59:03 · 阅读 1.2k 免费加密货币行情接口实战：历史数据抓取与实时推送方案
做量化回测和盯盘工具的朋友都知道策略模型能不能跑出结果关键不在花里胡哨的指标而在数据链路是否稳。加密货币历史数据和实时行情接口这两样东西看起来免费入口到处都是真正要稳定拿到手却有不少坑。我这两年帮自己和朋友做过好几个看板和量化小项目从免费接口选型、批量拉取到实时推送都折腾了一遍。这篇文章就把整条链路拆开讲告诉你哪些免费接口值得用、为什么值得用、每一步的坑在哪里。适合正在做回测、计划做实时告警、或者想给图表工具接行情源的同学参考。1. 免费行情接口的整体选型与思路1.1 先区分历史数据与实时行情的不同要求很多人在选接口的时候上来就问“哪个平台免费”其实更该问的是“我需要哪类数据”。历史数据和实时行情的接口策略完全不同选错方向后面全是麻烦。历史数据指的是已经发生的K线、收盘价、成交量、逐笔成交等主要喂给回测引擎做回放也用来画走势图。历史数据接口看重三件事深度、连续性、限频容忍度。所谓深度就是能不能拿到很久以前的数据最好是从项目上线第一根K线开始连续性是指K线之间不能有空缺否则回测结果会出现脏点限频容忍度则是你批量拉取时能不能承受大量请求。实时行情指 ticker、最新成交价、订单簿深度这类“正在发生”的数据。它看重延迟、推送频率、连接稳定性。用 REST 接口轮询能做到秒级用 WebSocket 能做到毫秒级。如果你做的是逐笔监控或者分钟级策略选了轮询方案还没开始写策略就已经输了。所以在动手之前先把自己的使用场景写清楚做日频回测做分钟级策略做秒级价格告警还是只做一个5个币种的简单看板场景不同接口选型和数据落地方式都不一样。1.2 主流免费行情接口横向对比下面是我自己实际用过或者调研过的免费接口整理。为了避免写成一篇随时过期的文档我在表后附上“以官方最新文档为准”的原则但这个阵列基本能帮你快速形成判断。数据源类型是否需要Key免费额度特点历史数据深度实时推送适合场景CoinGecko聚合行情免费Key有限频每分钟请求有限适合轻量调用常见币种多年价格粒度有限无强实时推送币种榜单、市值查询、简单趋势图CoinCap聚合行情无需Key公开限频适合中小规模有历史价格曲线接口提供公共WebSocket价格推送原型验证、中小看板、价格机器人主流交易所Public API交易所原生一般无需Key权重限频比聚合平台更宽松从项目上线到当前全量K线WebSocket流稳定回测取数、自建实时行情服务Alpha Vantage聚合行情需要Key每日次数有限日/周/月线无良好实时推送少量币种的长期趋势分析为什么我更推崇交易所原生接口因为它直接把最完整的K线字段给你时间戳、开高低收、成交量、成交笔数、主动买量一应俱全而且免费额度通常按权重计算只要你会控制节奏实际拉取量可以非常大。聚合平台胜在“多币种统一”但历史深度和实时推送往往只是“够用”而不是“好用”。提醒一下免费接口的额度条款经常调整你在参考任何人文章时都不要把“每分钟能请求多少次”当成永久事实。我自己的习惯是接到项目后先去看官方文档的 Rate Limit 部分再写一个最小测试脚本实测一遍。1.3 为什么不建议自己写爬虫抓网页我完全能理解为什么有人想用爬虫去抓行情网站的页面数据因为视觉上最直观好像什么字段都有。但实际上这是性价比最低的方案原因有三第一网页结构说改就改。昨天还在的 class 今天换个名字你的定位器就彻底失效第二反爬手段多。有些网站会做限流、浏览器指纹检测、验证码你要花大量时间在“对抗”上而不是在数据业务上第三网页拿到的数据往往是格式化后的字符串缺少可靠的时间戳和原始成交属性做回测时需要反复清洗精度还不够。正规接口返回的是结构化 JSON字段有明确含义时间戳是标准的毫秒数这才是工程上该走的路径。接口要付费的情况确实有但免费入口并不少关键是把通用方法学会然后按需替换数据源。2. 核心概念与接口原理2.1 读懂K线历史数据的返回结构免费接口再怎么五花八门历史K线的核心字段都大同小异。下面是一段典型的交易所K线返回数组我加了注释[ 1507024800000, 0.01630000, 0.05000000, 0.01630000, 0.04000000, 1000.00000000, 1507024860000, 1000.00000000, 10, 0.02000000, 0.01000000, 0.00000000 ]字段含义从左到右分别是K线开盘时间毫秒时间戳、开盘价、最高价、最低价、收盘价、成交量、K线收盘时间、成交额、成交笔数、主动买入量、主动买入额、忽略字段。有两点值得注意。一是价格和成交量通常是字符串不是数字。这类字段如果直接做算术运算很容易出现类型错误正确做法是先转成 Decimal 或者 float 再计算。二是时间戳是毫秒级 Unix 时间戳它本身不包含时区信息。严格来说任何时区下表示的都是同一个瞬间只是你把它显示成“北京时间”还是“UTC时间”的区别。回测入库时我建议把 openTime 直接存成 INT 类型同时再存一个可读的时间字段比如 UTC 的 ISO 格式。这样做既能保证排序效率也方便人眼排查问题。2.2 限频权重与请求合并免费接口大部分不是“按次数限频”而是“按权重限频”。每个接口的耗时程度不同权重也不同。比如拉取1000根K线可能消耗权重2而查询账户信息可能消耗权重20。这样设计是为了防止有人用高频操作阻塞服务。正确姿势是“少请求、多取数”。比如拉历史K线时一次请求尽量拉满允许的最大 limit宁可循环几次也不要拆成几百个小请求。如果你要订阅多个交易对能用一条 WebSocket 连接批量订阅就不要为每个交易对建一条连接。这里还要提一个常见误区有些人写代码时每次循环都发一次 HTTP 请求然后 sleep 0.1 秒看起来“很温柔”但实际权重消耗非常多而且容易被限频。更聪明的做法是把你的数据按时间分片每个分片尽量大到接近接口上限再配合缓存机制把真实的请求量降下来。2.3 时间缝隙与未收盘K线历史K线接口的一个隐形坑是最后一根K线经常“还没定型”。比如你拉取 1 分钟K线当前时间的这根K线会在 1 分钟结束前被反复更新开盘价确定了但收盘价还在变化。如果你在这个过程里把它当成最终历史数据入库回测就会用到未来函数。处理办法有两个层面。第一在你做实时增量更新时识别到同一个 openTime 的K线再次推送不要插入新记录而是更新旧记录。第二做回测取数时宁可选择“时间上已经完整收盘”的K线例如拉历史数据时把 endTime 设为当前时间减去一个K线周期拒绝最后一根不完整K线。我在实际项目里的做法是数据库里给K线表加一个状态标志已经收盘的历史K线标记为 final实时流推过来的未收盘K线标记为 pending。回测引擎只读取 final看板则把 pending 也显示出来。3. 实操用 Python 拉取免费历史数据并落地3.1 环境准备本文代码以 Python 为基础你只需要安装 requests 和 pandas。如果你做轻量落地可以只用标准库但 pandas 处理K线数组确实方便得多。python -m venv venv source venv/bin/activate # Windows 下用 venv\Scripts\activate pip install requests pandas建议你在项目目录下建一个 fetch_history.py把拉取函数封装成独立模块方便后续接入不同的数据源。因为不同平台的域名、字段名会有差异模块化之后替换数据源时只需要改一个 adapter。3.2 拉取一个时间窗口的K线下面这段代码用公共行情接口拉取某一交易对的历史K线。以某交易所公开文档中的标准端点为示例你使用之前务必替换为当前可访问的接口域名。import requests import pandas as pd BASE_URL https://api.example.com/api/v3/klines def fetch_klines(symbolBTCUSDT, interval1h, start_msNone, end_msNone, limit1000): params { symbol: symbol, interval: interval, limit: limit, } if start_ms: params[startTime] start_ms if end_ms: params[endTime] end_ms resp requests.get(BASE_URL, paramsparams, timeout10) resp.raise_for_status() data resp.json() df pd.DataFrame(data, columns[ open_time, open, high, low, close, volume, close_time, quote_volume, trades, buy_base_volume, buy_quote_volume, ignore ]) # 把字符串价格转成浮点 for col in [open, high, low, close, volume, quote_volume]: df[col] df[col].astype(float) return df这里有个细节返回的 JSON 是二维数组用 pandas 直接指定列名最方便。但要注意数组里的空值或者字段顺序变化某些平台会在后续版本增加字段。你可以在解析前先判断列数如果小于预期就做兼容处理。请求超时建议设置 10 秒以上大网络环境下偶发抖动很正常超时后可以重试。不要用默认超时否则程序可能卡死很久。3.3 循环分片拉取全量历史数据并校验连续性单个请求最多返回 1000 根K线要拉一整年甚至更早的数据就必须按时间窗口循环。这里的关键逻辑是下一次请求的 startTime 本次最后一根K线的 open_time interval 对应的毫秒数。某些平台支持按 endTime 往前翻但用 startTime 向后翻最通用。def fetch_history_until(symbol, interval, start_ms, end_msNone): interval_ms { 1m: 60_000, 5m: 300_000, 15m: 900_000, 1h: 3_600_000, 4h: 14_400_000, 1d: 86_400_000, } step interval_ms.get(interval, 60_000) all_frames [] cursor start_ms while True: if end_ms and cursor end_ms: break df fetch_klines(symbol, interval, start_mscursor, limit1000) if df.empty: break all_frames.append(df) last_open_time df[open_time].iloc[-1] cursor last_open_time step # 止损保护 if cursor - start_ms 5_000 * step: break result pd.concat(all_frames, ignore_indexTrue) result result.drop_duplicates(subsetopen_time) return result.sort_values(open_time).reset_index(dropTrue)你可能会问为什么最后要做 drop_duplicates因为并发重试或者接口异常返回时重叠时间区域很可能拉回相同K线去重是写进数据链路里最便宜的一道保险。连续性校验同样重要。拿到完整数据后你要检查相邻两根K线的 open_time 差值是否严格等于 interval。如果出现空洞就需要回到接口单独补那一小段区间而不是整个重拉。def check_gaps(df, interval_ms): diff df[open_time].diff().dropna() gaps diff[diff ! interval_ms] return gaps3.4 落盘CSV 还是 SQLite历史数据落地常见两种方案CSV 和 SQLite。CSV 的好处是文件小、可读性好、Excel 直接打开缺点是增量更新麻烦每次都要全量重写。SQLite 的好处是支持 SQL 去重、条件更新相对轻松适合长期维护的数据仓库。我的建议是如果只是临时拉一份数据跑回测CSV 够用如果这个数据要每天定时更新、还要边拉边增量那就用 SQLite。建表语句可以参考下面这样CREATE TABLE IF NOT EXISTS klines ( symbol TEXT NOT NULL, interval TEXT NOT NULL, open_time INTEGER NOT NULL, open REAL, high REAL, low REAL, close REAL, volume REAL, close_time INTEGER, PRIMARY KEY (symbol, interval, open_time) );用(symbol, interval, open_time)做联合主键天然就能防止重复。写入时用INSERT OR REPLACE实时行情推过来的未收盘K线也能直接覆盖不会产生脏数据。这个设计是我在用了一段时间 CSV 之后才改的强烈推荐。4. 实时行情接口WebSocket 订阅与增量更新4.1 REST 轮询和 WebSocket 怎么选实时行情的获取有两个方向REST 轮询和 WebSocket 长连接。REST 轮询很好理解每隔几秒请求一次最新价格。优点是实现简单缺点是延迟高、请求密度大、容易触发限频。适合那些“慢半拍也无所谓”的场景比如每5分钟刷新一次某个币的价格。WebSocket 则是客户端和服务端建立一条长连接服务端有新数据时主动推给你。它的延迟可以到毫秒级而且不会因为订阅多个交易对而线性增加请求次数。适合实时告警、逐笔滚动、盘口刷新等场景。对比项REST 轮询WebSocket 推送延迟秒级毫秒级请求消耗高低实现难度低中多交易对扩展每个交易对都要请求一次一条连接订阅多个符号断线恢复无需恢复需要自己处理重连如果你需要实时K线画图建议用 WebSocket 订阅K线流来做增量再用历史数据接口拉一次全量K线做底图。这样既有历史深度又有实时速度。4.2 一个可以直接运行的 WebSocket 示例下面用两个不同层级的例子。第一个例子是最简单的无 Key 公共订阅适合快速验证第二个例子是交易所K线流的典型订阅方法适合正经项目。先看无 Key 的公共价格流。CoinCap 官方文档提供了这种订阅方式只要你网络能访问它代码几乎不用配置from websocket import create_connection ws create_connection(wss://ws.coincap.io/prices?assetsbitcoin,ethereum) for _ in range(5): message ws.recv() print(message) ws.close()返回的消息是类似这样的 JSON{bitcoin: 66812.123, ethereum: 3420.456}这种形式的接口适合做简单价格展示但毛病也很明显它不区分交易对和K线周期只给你一个“现价”做不了细致的历史分时。所以真正做K线级实时行情还是要看交易所的标准订阅协议。交易所 WebSocket 的通用逻辑通常是建立连接后先发送订阅指令再持续接收服务端推送。可以参考下面这段框架接口域名以你选择的交易平台官方文档为准import json from websocket import WebSocketApp def on_message(ws, message): data json.loads(message) # 不同的流字段结构不同这里以常见的K线事件为例 k data.get(k) if k: print( fopenTime{k.get(t)} close{k.get(c)} ) def on_error(ws, error): print(error:, error) ws WebSocketApp( wss://stream.example.com/ws/btcusdtkline_1m, on_messageon_message, on_erroron_error, ) ws.run_forever()这里的核心动作是run_forever()。它会阻塞当前线程不断接收数据。如果你在回调里做数据库写入要注意写入速度赶不上推送速度的情况必要时用队列缓冲让消费者线程慢慢落库。4.3 心跳保活与断线重连WebSocket 使用中最常见的故障是“连接还挂着但已经收不到数据了”。很多服务端会在一段时间内没有收到任何客户端消息时主动断开或者网络设备把空闲连接清理掉。解决办法是定时发送心跳消息。交易所类连接一般支持发送一个简单的 PING 帧或者 JSON 格式的 ping 消息。如果没有标准心跳指令一个最保底的做法是记录last_message_time用一个后台线程检查如果距离上次收到消息超过阈值就主动重连。下面是一个加了心跳守护的框架import threading import time from websocket import WebSocketApp class KlineStream: def __init__(self, url): self.url url self.last_msg time.time() self.ws None def on_message(self, ws, message): self.last_msg time.time() # 处理数据... def on_open(self, ws): def send_ping(): while True: time.sleep(20) try: ws.send(ping) except Exception: break t threading.Thread(targetsend_ping, daemonTrue) t.start() def run(self): self.ws WebSocketApp( self.url, on_messageself.on_message, on_openself.on_open, ) while True: self.ws.run_forever() print(连接断开5秒后重连...) time.sleep(5) self.last_msg time.time()重连逻辑里有个细节直接调用run_forever()在断线后不会自动退出所以用while True包裹并在外面重新赋值WebSocketApp。定时线程也要注意防止重复启动否则每次重连都会多一个线程发心跳。4.4 实时数据如何与历史数据拼接使用实时K线流最忌讳的一件事是“闭着眼插入”。因为一根K线在它所在周期内会被推送很多次收盘价一直在变化而你库里已经有了开盘时拉到的一条记录。不做更新只做插入很快就会产生大量重复行。正确做法是把实时事件里的 openTime 作为主键和正在更新的K线进行匹配。如果数据库里没有这根K线就插入如果已经有就覆盖最新价格和成交量。这个操作在 SQLite 里就是INSERT OR REPLACE在内存里就是 update 一个 dict。我自己在写盘时还会加一个“是否已收盘”判断当消息里的 K线 closeTime 小于当前时间说明这根K线已经定型就不再接受后续更新了。这样可以避免在数据落地后还看到同一根K线的数值反复跳变。5. 常见问题与排查技巧实录5.1 接口返回 429 或 418被限频了怎么办429 是请求太频繁418 表示你已经触发了更严格的限制需要等待一段时间。这两个状态码几乎是免费接口用户最容易遇到的。遇到 429 不能直接“不休息地重试”那样只会更糟。正确做法是记录返回头里的限频信息或者按固定退避时间重试。下面是一个最简单的指数退避实现import time from requests import RequestException def get_with_retry(session, url, params, retries5): wait 1 for attempt in range(retries): resp session.get(url, paramsparams, timeout10) if resp.status_code 429: time.sleep(wait) wait * 2 continue resp.raise_for_status() return resp.json() raise RequestException(retry exceeded)退避的倍数一般取 2 或者 3。如果你同时跑多个线程限频概率会成倍上升。更优雅的做法是把所有请求收敛到同一个调度器里做一个全局的“最低间隔”控制。5.2 历史数据中间缺了一段出现数据空洞常见原因有两个一是分页游标参数算错了比如把 endTime 当成毫秒还是秒搞混二是请求被限频后某一段没有成功返回但代码没有检测到就直接拼接了。排查的步骤我整理成下面这条路径先对拼接好的 DataFrame 做 open_time diff 检测确认空洞范围。用空洞前后两根K线的时间戳作为新的startTime和endTime单独拉一次。如果单独拉取依然缺失检查该交易对在那个时间段是否真的开盘。某些币种上市时间并不早那不是数据空洞而是根本不存在数据。我曾经遇到过一只新上线的代币拿它半年前的数据时一直缺一段后来才发现它上市时间比我的起始时间晚前面那一整段自然不会有数据。这个错误非常典型。5.3 时间戳看起总差 8 小时毫秒时间戳本身没有时区概念但你把它转成字符串时如果不指定时区很多本地化方法会把 UTC 时间当作本地时间显示。如果你的系统时区是东八区那显示出来的时间就会比真实时间快 8 小时。正确的入库策略是数据库里只存 UTC 时间或者直接存毫秒整数展示层要做本地化转换时再统一加时区。不要在清洗数据的过程中就把时间格式化成“YYYY-MM-DD HH:mm:ss”因为一旦你失去了毫秒精度后续做区间查询会非常痛苦。5.4 WebSocket 连上之后却没有数据这种问题通常不是“没连上”而是“订阅没成功”。很多平台的 WebSocket 需要你连接后先发一条订阅指令服务端收到后才会开始推数据。如果你只是建立了连接然后傻等当然收不到任何消息。排查时可以分三看一看返回的消息里有没有订阅回执比如result: null, id: 1二看订阅符号是否准确比如BTCUSDT和btcusdt在不同平台可能大小写规则不同三看是否订阅到了正确频道K线、成交、ticker是三个完全不同的流。顺带说一个常见误操作订阅多个交易对时要看平台是否支持streams...参数不支持的话只能一条连接里发一个数组订阅发错格式一样会没数据。5.5 接口字段改动导致程序崩溃这类问题是免费接口最让人头疼的。某个平台某天在数组末尾加了一个字段你的硬编码取数逻辑就可能越界或者把字符串价格送到数学函数里当场抛异常。建议所有解析层都做一个“宽容读取”。也就是优先使用字典的.get()不要用data[field]直接访问。如果字段缺失至少不要让整个程序挂掉而是记录一条 warning 并跳过这一条。样例代码如下def safe_float(value, default0.0): try: return float(value) except (TypeError, ValueError): return default这种“防御性编程”在对接第三方免费接口时不是可选项而是必须项。因为免费接口没有严苛的维护承诺字段变化只在一夜之间。5.6 免费接口的使用边界与合规提示最后这部分我想认真提醒一句免费接口不是公有领域数据它有使用条款。不同平台的免费额度条款会写清楚你可以用它做什么、不能做什么。我个人的原则是个人研究、学习、内部看板用途完全没问题公开转售、商业化分发原始行情、批量打包数据提供给第三方基本都不在免费条款允许范围内。同时加密货币相关业务本身受不同国家和地区的法规约束。请确保你的使用行为符合你所在地区的法律法规以及服务商条款。本文所有内容仅用于技术研究不构成任何投资建议也不鼓励任何违规交易行为。就我个人实际跑下来的体会来说历史数据用交易所 REST 接口批量落盘 SQLite实时行情用 WebSocket 做增量更新两者用(symbol, interval, open_time)这个联合主键串起来是一套性价比很高、维护成本也很低的组合。真正值得花时间的不是反复换接口而是把数据校验、去重、断线重连这套底座磨扎实。你把这套链路跑顺之后再去接任何新平台剩下的其实只有字段映射和限频适配。