Python异步编程实践：使用Binance API构建实时交易数据流

新闻中心 NEWS CENTER

您当前位置：首页 > 新闻中心 > 网络学院

Python异步编程实践：使用Binance API构建实时交易数据流

2025-12-01

浏览次数：次

返回列表

python异步编程实践：使用binance api构建实时交易数据流

本教程详细介绍了如何利用Python的asyncio库和python-binance客户端，构建一个高效、实时的币安（Binance）交易数据流应用程序。文章将涵盖环境设置、异步编程核心概念、代码实现，并特别讨论了在Jupyter Notebook与独立Python脚本文件之间运行异步代码的差异及注意事项，旨在帮助开发者顺利部署实时数据监控系统。

1. 环境准备

在开始构建实时数据流应用之前，我们需要确保Python环境已正确配置。推荐使用虚拟环境来管理项目依赖，以避免潜在的包冲突。

创建并激活虚拟环境：

python -m venv venv_binance_stream
# Windows
.\venv_binance_stream\Scripts\activate
# macOS/Linux
source venv_binance_stream/bin/activate

安装所需库： 本教程主要依赖python-binance和nest_asyncio。nest_asyncio在某些交互式环境中（如Jupyter）运行异步代码时非常有用。
```
pip install python-binance==1.0.19 nest_asyncio==1.5.8
```
请注意，这里指定了与成功案例中相同的版本，以确保兼容性。

2. 核心概念解析

理解以下几个核心概念对于构建异步数据流至关重要：

asyncio: Python用于编写并发代码的标准库，通过事件循环（event loop）实现协程（coroutine）的调度。async和await关键字是其核心，分别用于定义协程和暂停协程执行以等待I/O操作完成。
python-binance: 币安API的Python封装库，提供了REST API和WebSocket API的便捷访问接口。其中，BinanceSocketManager是处理WebSocket连接的关键组件。
nest_asyncio: 一个允许在已运行的事件循环中运行新的asyncio事件循环的库。这在Jupyter Notebook这类环境中非常有用，因为Jupyter自身可能已经有一个事件循环在运行，直接调用asyncio.run()会导致错误。通过nest_asyncio.apply()可以解决这个问题。

3. 实现实时交易数据流

我们将构建一个脚本，连接到币安的WebSocket API，实时接收BTC/USDT的交易数据。

Seede AI

AI 驱动的设计工具

713 查看详情 Seede AI

import asyncio
import nest_asyncio
import sys
import logging

from binance.client import Client
from binance import BinanceSocketManager

# 配置日志
logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(levelname)s - %(message)s')

def run_binance_stream():
    """
    设置并运行币安实时交易数据流。
    """
    # 在Jupyter等环境中，如果已存在事件循环，需要应用nest_asyncio
    # 对于独立的.py脚本，通常不是必需的，但保留可确保兼容性
    if sys.platform == 'win32': # Windows系统可能需要特殊处理
        asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())

    nest_asyncio.apply()

    async def main():
        """
        异步主函数，负责连接WebSocket并接收数据。
        """
        # 初始化Binance客户端。此处无需API Key，因为WebSocket公共数据不需要认证。
        client = Client()

        # 初始化Binance Socket管理器
        bsm = BinanceSocketManager(client)

        # 订阅BTCUSDT交易数据流
        socket = bsm.trade_socket('BTCUSDT')

        logging.info("正在连接到Binance WebSocket...")
        try:
            async with socket as ts:
                while True:
                    try:
                        logging.info('等待数据...')
                        msg = await ts.recv() # 异步等待接收数据
                        logging.info(msg) # 打印接收到的交易消息
                    except asyncio.CancelledError:
                        logging.info("数据流已取消。")
                        break
                    except Exception as e:
                        logging.error(f"接收数据时发生错误: {e}")
                        await asyncio.sleep(5) # 发生错误后等待一段时间重试
        finally:
            # 确保客户端连接被关闭，释放资源
            await client.close_connection()
            logging.info("Binance客户端连接已关闭。")

    # 运行异步主函数
    # asyncio.run() 会自动创建并管理事件循环
    try:
        asyncio.run(main())
    except KeyboardInterrupt:
        logging.info("程序被用户中断。")
    except RuntimeError as e:
        logging.error(f"运行时错误: {e}. 可能是事件循环问题。")
    finally:
        logging.info("数据流程序已退出。")

if __name__ == '__main__':
    run_binance_stream()

代码解析：

import asyncio, nest_asyncio, sys, logging: 导入所需的库。sys用于平台检测，以应对Windows平台可能存在的asyncio事件循环策略问题。logging用于记录程序运行状态。
logging.basicConfig(...): 配置日志系统，使程序运行时的信息和错误能够被记录下来。
Client(): 初始化python-binance客户端。对于公共数据流（如交易数据），通常不需要API Key和Secret。
BinanceSocketManager(client): 创建WebSocket管理器实例。
bsm.trade_socket('BTCUSDT'): 订阅指定交易对（BTCUSDT）的交易数据流。trade_socket返回一个异步上下文管理器。
async with socket as ts:: 使用异步上下文管理器，确保WebSocket连接的正确建立和关闭。
while True: msg = await ts.recv(): 循环异步等待并接收来自WebSocket的数据。await是关键，它允许程序在等待数据时释放控制权给事件循环，从而执行其他任务（如果有的话）。
if __name__ == '__main__': run_binance_stream(): 这是Python脚本的标准入口点。确保run_binance_stream函数只在脚本作为主程序运行时被调用。
错误处理: 增加了try...except块来捕获asyncio.CancelledError（例如，当事件循环被关闭时）和一般异常，提高了程序的健壮性。KeyboardInterrupt用于优雅地处理用户通过Ctrl+C中断程序。
Windows平台事件循环策略: 在Windows系统上，asyncio默认的事件循环策略可能导致一些兼容性问题。asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())可以解决此问题。
await client.close_connection(): 在finally块中确保关闭Binance客户端的连接，这是良好的资源管理实践。

4. 从Jupyter Notebook到独立Python脚本的考量

原始问题中提到代码在Jupyter Notebook中运行正常，但在.py文件中却不工作。这通常是由于异步环境的差异造成的：

Jupyter Notebook环境: Jupyter内核本身可能已经运行了一个asyncio事件循环。在这种情况下，直接调用asyncio.run()会尝试启动一个新的事件循环，导致RuntimeError: Event loop is already running。nest_asyncio.apply()的作用就是允许asyncio.run()在一个已经运行的事件循环中嵌套运行一个新的事件循环，从而解决这个问题。
独立Python脚本 (.py文件) 环境: 当你直接运行一个.py文件时，通常没有预先存在的事件循环。asyncio.run(main())会负责创建一个新的事件循环，运行main()协程，并在main()完成后关闭事件循环。因此，在独立的.py脚本中，nest_asyncio.apply()通常不是必需的，但保留它通常不会造成问题，并能提高代码在不同执行上下文中的兼容性。

为什么原始脚本在.py中可能不工作？

尽管代码在理论上应该在.py文件中正常运行，但出现问题可能的原因包括：

环境差异: 尽管声称Python和库版本相同，但devcontainer的配置、网络代理、防火墙规则或系统级别的事件循环策略可能存在细微差异，影响了WebSocket连接。
事件循环策略: 尤其是在Windows系统上，asyncio的默认事件循环策略可能导致问题。在代码中添加if sys.platform == 'win32': asyncio.set_event_loop_policy(asyncio.WindowsSelectorEventLoopPolicy())可以解决这一问题。
网络连接问题: WebSocket连接可能因为网络不稳定、代理设置不正确或防火墙限制而无法建立或维持。ts.recv()会无限期等待，导致程序看似“卡住”。
程序退出方式: 如果没有适当的错误处理或中断机制，程序在遇到网络问题时可能不会给出明确的反馈。

通过上述提供的完整代码，我们包含了更健壮的错误处理和对Windows平台事件循环策略的适应，这有助于提高在不同环境下的兼容性和稳定性。