如何通过 Python 获取加密货币的历史实时行情

10

Feb 2025如何通过 Python 获取加密货币的历史实时行情Contents概要说明什么是 ccxt安装 ccxt初始化交易所实时价格历史行情订单簿（市场深度）实时监听总结如果你想参与加密货币市场，获取实时、准确的市场数据就是重要的第一步。

不同于传统金融市场，加密货币市场上的交易所有两类，分别 DEX 和 CEX，即中心化交易所和去中心化交易所。两者的区别这里就不展开了，我们大部分人接触的一般都是中心化交易所。和传统金融市场一样，加密货币市场上的数据，我们也可找专业的数据提供商，但基本都是付费的。

本文将介绍如何通过 Python 从中心化交易所拿到行情数据，包括历史和实时数据，包括直接从 websocket 监控数据，可用来实时监听不对交易对间的价差数据。

概要说明从去中心化交易所拿到交易所数据，可直接通过访问它的 HTTP 和 Websocket 接口，如欧易 OKX 的接口，查看它的官方文档：欧易 API 文档，其他每个交易所也都有这类文档，可查看其支持的 API 接口。

不过市面上的中心化交易所很多，而且如果要获取账户级别的信息，还有经过认证流程。本文为了简化流程，覆盖更多的去中心化交易所，我会使用一个非常流行的 Python 库 ccxt 演示，从而把重点集中在数据的获取方法上。

什么是 ccxtccxt，全称 CryptoCurrency eXchange Trading，这个开源库能让你用统一的代码访问全球上百家数字货币交易所。它支持多个编程语言，如 Javascript、Python 和 PHP。

ccxt，除了可用来获取数据，还可以支持直接下单、订单管理、访问仓位、账户等，用来实现一个自动化交易人。

接下来开始正文吧！

本文将从最简单的价格获取到复杂的订单簿处理，带你一步步深入。

安装 ccxt打开终端，输入如下命令，安装 ccxt 库：

1

pip install ccxt

初始化交易所以币安为例，先创建一个交易所对象：

1

2

3

4

5

6

import ccxt

exchange = ccxt.binance({

'enableRateLimit': True, # 必须开启！防止被交易所封IP

'timeout': 15000 # 超时设为15秒

})

交易所的API都有严格的调用频率限制，enableRateLimit 会启用频率限制能力，防止触发交易所的 API 调用频率。

如果遇到网络问题，也可通过设置 proxies 切换网络环境：

1

2

3

4

5

6

exchange = ccxt.okx({

"proxies": {

"http": "http://127.0.0.1:7890",

"https": "http://127.0.0.1:7890",

}

})

获取交易所上的所有品种：

1

2

markets = exchange.load_markets()

print(list(markets.keys)[:10])

输出如下：

1

['ETH/BTC', 'LTC/BTC', 'BNB/BTC', 'NEO/BTC', 'QTUM/ETH', 'EOS/ETH', 'SNT/ETH', 'BNT/ETH', 'BCC/BTC', 'GAS/BTC']

在开始其他操作前，这一步是必要的，否则你可能会遇到品种不存在的问题。

实时价格首先，示例如何通过 ccxt 获取某个品种当前的价格，主要依赖于 fetch_ticker 函数接口。

示例代码，读取 BTC/USDT 交易对的最新价。

1

2

3

4

5

6

7

8

ticker = exchange.fetch_ticker('BTC/USDT')

print(f"""

实时价格：

- 最新价：{ticker['last']} USDT

- 24小时最高：{ticker['high']}

- 24小时成交量：{ticker['quoteVolume']} USDT

""")

输出：

1

2

3

4

实时价格：

- 最新价：97004.87 USDT

- 24小时最高：97640.02

- 24小时成交量：1743212485.2952676 USDT

ticker 的结构体可查看官方文档 Ticker Structure。

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

{{

'symbol': string 类型，表示市场的交易对（例如：'BTC/USD', 'ETH/BTC' 等），

'info': 原始未修改的交易所 API 回复信息（未经解析的原始数据），

'timestamp': integer 类型，64 位 Unix 时间戳，单位为毫秒，自 1970 年 1 月 1 日起，

'datetime': ISO8601 格式的日期时间字符串（带毫秒），

'high': float 类型，市场的最高价格，

'low': float 类型，市场的最低价格，

'bid': float 类型，当前最佳买入价，

'bidVolume': float 类型，当前最佳买入量（可能缺失或未定义），

'ask': float 类型，当前最佳卖出价，

'askVolume': float 类型，当前最佳卖出量（可能缺失或未定义），

'vwap': float 类型，加权平均价格，

'open': float 类型，开盘价，

'close': float 类型，最后交易价格（当前周期的收盘价），

'last': float 类型，等同于 `close`，为方便重复表示，

'previousClose': float 类型，上一个周期的收盘价，

'change': float 类型，绝对变化，`last - open`，

'percentage': float 类型，相对变化，`(change/open) * 100`，

'average': float 类型，平均价格，`(last + open) / 2`，

'baseVolume': float 类型，过去 24 小时内交易的基础货币的交易量，

'quoteVolume': float 类型，过去 24 小时内交易的报价货币的交易量，

}

简言之，fetch_ticker 返回的是某个交易对的当前最新的市场数据。

历史行情除了最新行情，历史行情对于交易也非常有帮助，如我们用它来计算指标，识别模式，产生交易信号。

如下是获取 BTC/USDT 的日线数据的代码：

1

2

3

4

5

6

7

# 获取最近3天的日线数据

ohlcvs = exchange.fetch_ohlcv(

symbol='BTC/USDT',

timeframe='1d', # 可选：1m, 5m, 1h, 1d 等

limit=5 # 获取条数

)

print(ohlcvs)

参数设置获取 BTC/USDT 的历史行情，时间周期为 1d，数量为 5。

输出：

1

2

3

4

5

[[1739138760000, 94897.63, 95052.0, 94884.44, 95047.09, 17.96677],

[1739138820000, 95047.1, 95128.75, 95047.1, 95118.36, 12.27411],

[1739138880000, 95118.36, 95135.99, 95051.65, 95122.51, 10.40116],

[1739138940000, 95122.52, 95178.69, 95104.42, 95178.57, 7.81859],

[1739139000000, 95178.57, 95246.13, 95160.0, 95246.13, 48.00226]]

列表中的每个元素依然是一个列表，从左到右分别是时间戳、开盘价、最高价、最低价、收盘价和交易量。

可以将其转为更加易读的 pandas 格式，如下所示；

1

2

data = pd.DataFrame(ohlcvs, columns=["timestamp_ms", "open", "high", "low", "close", "volume"])

print(data)

输出：

1

2

3

4

5

6

timestamp_ms open high low close volume

0 1739139000000 95178.57 95246.13 95160.00 95246.13 48.00226

1 1739139060000 95246.12 95436.12 95230.02 95429.63 24.59766

2 1739139120000 95429.64 95429.64 95282.44 95371.00 44.81550

3 1739139180000 95371.00 95443.28 95371.00 95433.12 12.77712

4 1739139240000 95433.12 95527.79 95433.12 95436.70 15.98278

如果想设计一个高频策略，可能还要拿到分钟级别数据，很多交易所都支持拿到分钟级别数据，甚至是全量的分钟数据。

假设，获取 BTC/USDT 的 1 分钟数据，只需修改 fetch_ohlcv 的 timeframe 参数即可。

示例代码:

1

2

# 获取最近20根15分钟K线

ohlcvs = exchange.fetch_ohlcv('BTC/USDT', '1m', limit=20)

订单簿（市场深度）除了行情数据，如果你拿到订单薄数据，ccxt 也可以做到，只要调用 fetch_order_book 即可。

一个简单示例：

1

2

3

4

5

order_book = exchange.fetch_order_book('BTC/USDT', limit=100)

best_bid = order_book['bids'][0][0] # 买一价

best_ask = order_book['asks'][0][0] # 卖一价

print(f"买一价：{best_bid}, 卖一价：{best_ask}, 价差：{best_ask - best_bid}")

输出：

1

买一价：97141.35, 卖一价：97141.36, 价差：0.00999999999476131

其中，fetch_order_book 可用于指定获取深度，像 binance 的接口默认是 100，而最多可以获取到 5000 的订单薄深度。

实时监听前面介绍完的几种数据接口，都是我们主动发起请求，对于一些实时性要求非常高的场景，轮询延迟大，丢掉一些交易机会，且可能触发交易所的频率限制。ccxt 还提供了一个 pro 版本 ccxt.pro，可直接通过 WebSocket 实时监听。

前面的接口，我们简单修改下，就可以实现实时监听的效果。

1

2

3

4

5

6

7

8

9

10

11

12

13

14

15

16

17

18

19

20

21

22

23

24

25

import asyncio

import ccxt.pro as ccxtpro

exchange = ccxtpro.binance({

'enableRateLimit': True, # 必须开启！防止被交易所封IP

'timeout': 15000, # 超时设为15秒

})

async def watch_order_book():

while True:

order_book = await exchange.watch_order_book('BTC/USDT')

best_ask = order_book["asks"][0][0]

best_bid = order_book["bids"][0][0]

print(f"order_book, 价差:{best_ask-best_bid}")

async def watch_ticker():

while True:

ticker = await exchange.watch_ticker('BTC/USDT')

print(f"ticker, 最新价: {ticker['last']}")

async def main():

await asyncio.gather(watch_order_book(), watch_ticker())

if __name__ == "__main__":

asyncio.run(main())

输出：

1

2

3

4

5

6

7

8

9

10

order_book, 价差:0.00999999999476131

order_book, 价差:0.00999999999476131

order_book, 价差:0.00999999999476131

order_book, 价差:0.00999999999476131

order_book, 价差:0.00999999999476131

order_book, 价差:0.00999999999476131

order_book, 价差:0.00999999999476131

order_book, 价差:0.00999999999476131

ticker, 最新价: 97228.01

...

现在就能实时监控，不断输出新的数据。

总结现在，就已经掌握了如何利用 ccxt 获取市场行情，接下来可以：

监控价差：实时计算不同交易对或交易所间的套利机会；对接交易：用ccxt的 create_order 方法实现自动交易；最后，希望本文对你有所帮助，感谢阅读。

本文来源于 POLOXUE's BLOG，地址： 如何通过 Python 获取加密货币的历史实时行情