CCXT简介

1.什么是ccxt？

在ccxt官方github文档中给出的解释为：ccxt是一个用于加密货币交易和电子商务的JavaScript / Python / PHP / C# / Go库，支持许多比特币/以太坊/山寨币交易市场和商户API。

GitHub项目地址https://github.com/ccxt/ccxt

2.为什么使用ccxt？

统一 API 接口：不同交易所的原生 API 接口差异较大，CCXT 为全球 100 多家交易所提供了统一的 API 接口，开发者只需学习一套方法就能在多个交易所进行交易操作。

多语言支持：CCXT 支持 Python、JavaScript、PHP 等多种编程语言，方便不同技术栈的开发者使用。

丰富的功能：提供了从简单的市场数据获取（如行情、订单簿）到复杂的交易操作（如下单、撤单）等一系列功能，满足不同层次的交易需求。

标准化数据结构：将不同交易所返回的异构数据标准化为统一格式，大大降低了数据处理的复杂度。

自动频率控制：每个交易所对 API 请求频率都有限制（如每秒 / 每分钟允许的请求次数），CCXT 会自动处理这些限制。

自动处理认证：简化了 API 密钥的管理和认证流程，开发者只需设置好密钥，库会自动处理签名和认证相关的操作。

错误处理：统一处理不同交易所可能返回的各种错误类型，提供清晰的错误信息和异常处理机制。

持续更新：随着新交易所的出现和现有交易所 API 的更新，CCXT 社区会及时跟进并添加支持。

ccxt官方使用文档https://docs.ccxt.com/#/

CCXT的安装与使用

如果你能够使用官方文档完成开发，可以参考文章：CCXT 深度解析与实战教程来进行统一化的开发。

本文以python为例，重在解决如何使用各交易所官方文档，通过ccxt完成非统一性的量化开发！

1.安装ccxt

直接使用pip即可安装python版本的 CCXT：

pip install ccxt

2.查看ccxt支持的交易所

可以通过 ccxt.exchanges 属性获取 CCXT 支持的所有交易所，也可以直接去github查看。

import ccxt
import pprint # 用于美化打印

print(f"CCXT 版本: {ccxt.__version__}")
print("支持的交易所:")
pprint.pprint(ccxt.exchanges)
print(f"\n总共支持 {len(ccxt.exchanges)} 家交易所")

量化开发

如果你打算量化开发的交易所不在ccxt支持内，没关系！本质上ccxt就是帮助我们按照各个交易所的API文档要求发送了各种http请求，我们也可以自己使用request库、封装各种函数方法进行量化开发，本质上与使用ccxt库是一致的，相信在你看完以下的量化开发过程以后也可以不使用ccxt开发！

1.实例化交易所

首先我们要创建交易所类并完成初始化，此后调用的所有方法都需要用到一开始创建的交易所类。

如代码中的binance.milliseconds()就是调用实例化的binance交易所的miliseconds方法获取交易所当前时间戳，binance.iso8601()就是将时间戳转换为ISO 8601 格式。

import ccxt

# 实例化 Binance 交易所对象
binance = ccxt.binance({
    'apiKey': '',
    'secret': '',
    'timeout': 10000,
    'enableRateLimit': True
})

# 加载市场数据,完成初始化
markets = binance.load_markets()

print('交易所当前时间:', binance.iso8601(binance.milliseconds()))

本文以币安(binance)为例进行展示代码，如你想使用其他交易所比如欧意(okx)，那么你的代码可以写成：

import ccxt

# 实例化 okx 交易所对象
okx = ccxt.okx({
    'apiKey': '',
    'secret': '',
    'password': '',
    'timeout': 10000,
    'enableRateLimit': True
})

# 加载市场数据,完成初始化
markets = okx.load_markets()

print('交易所当前时间:', okx.iso8601(okx.milliseconds()))

不同交易所要求的验证略有不同，这里okx需要除了api密钥与密码，还要创建时给api起的口令

2.使用交易所类

我们获得了交易所类以后可以看到他有很多方法，那么我们要获取行情数据、下单、撤单要具体使用哪个呢？

首先当然是去看官方说明文档，按方法分类的接口规范、按交易所分类的接口规范......

如果你和我一样完全搞不懂这些，那么就不得不说到一个方法.__dir__()，获取对象的所有可访问属性（包括变量、方法、特殊属性等）

import ccxt

binance = ccxt.binance()

print(binance.__dir__())

保存下来可以看到我们获取到了海量的方法，细心的可以发现这些方法中不少包含post、get这些http请求种类，那么大胆的猜想ccxt是不是将所有接口都按照接口类型（get/post）+url的方式来命名方法呢？

例如我们来看这个binanceAPI文档中获取合约最新价格的接口 GET /fapi/v1/ticker/price

果真如此，我们可以再看看下单接口 POST /fapi/v1/order

至此我们发现了ccxt的另一打开方式：使用隐式API接口！

只需在各个交易所的官方API文档中找到你想使用的接口，按照url来一顿操作组成出来ccxt的隐式API接口即可！传参按照API文档中的要求传入一个字典就好了！

那么开头的问题：你准备开发的交易所ccxt不支持怎么办？

我们只需要按照这个思路，自己封装一个类，类中要正确配置好API密钥验证与频率限制，在类内实现多个方法：使用request发出url请求，正确处理返回信息、抛出异常。

3.示例：查询最新价格

这里以查询U本位合约最新价格为例：

我们找到url来确定使用的隐式API，找到请求参数来正确传参。

import ccxt

# 实例化 Binance 交易所对象
binance = ccxt.binance({
    'apiKey': '',
    'secret': '',
    'timeout': 10000,
    'enableRateLimit': True
})

# 加载市场数据,完成初始化
markets = binance.load_markets()

symbol = 'BTCUSDT'
while True:
    ticker_data = binance.fapipublic_get_ticker_price({'symbol': symbol})
    print(ticker_data)

获取的频率应逼近理论最快速度50ms，若达不到是网络问题，需更换网络。

4.示例：下单

下单接口属于私有接口，需要API验证，在实例化交易所对象时要正确配置API信息，此外申请API密钥时大多交易所要求配置白名单，无法使用VPN，配置静态IP网络延迟太高。

建议直接购买交易所未限制地区的服务器，在linux服务器上运行代码既可以解决IP白名单问题，又可以使用tmux复用终端保证24h运行量化脚本。

依旧在官方API文档中找到接口url，确定所需传参。

import ccxt

# 实例化 Binance 交易所对象
binance = ccxt.binance({
    'apiKey': '',  # 替换为你申请的apikey
    'secret': '',  # 替换为你申请的secret
    'timeout': 10000,
    'enableRateLimit': True
})

# 加载市场数据,完成初始化
markets = binance.load_markets()

symbol = 'BTCUSDT'
order_params = {
    'symbol': symbol,
    'side': 'BUY',
    'type': 'MARKET',
    'quantity': 0.002
}
order_res = binance.fapiPrivatePostOrder(params=order_params)

此代码是在单向持仓模式下的传参，双向持仓模式传参略有不同，详细请看binanceAPI文档

代码须正确配置api密钥、在白名单ip服务器上运行，账户需保证金充足、持仓模式单向，运行脚本账户会以市价成交一个0.002的btc多仓

5.重试装饰器

ccxt库在遇到任何错误时会立马抛出异常，导致代码崩溃中止。我们当然不希望出现这种情况，因此需要引入重试装饰器让代码正确处理异常：

1.代码中需要请求服务器资源的地方（下单、获取行情信息、获取账户信息等），在自定义方法中调用，下面代码示例就是将获取行情数据方法（exchange.fapipublic_get_ticker_price）另在自定义方法（fetch_ticker_price）中调用：

import ccxt

# 查询合约指定币对价格
def fetch_ticker_price(exchange, para_symbol):
    return exchange.fapipublic_get_ticker_price({'symbol': para_symbol})

# 实例化 Binance 交易所对象
binance = ccxt.binance({
    'apiKey': '',
    'secret': '',
    'timeout': 10000,
    'enableRateLimit': True
})

# 加载市场数据,完成初始化
markets = binance.load_markets()

symbol = 'BTCUSDT'
while True:
    ticker_data = fetch_ticker_price(binance, symbol)
    print(ticker_data)

2.定义重试装饰器，将重试装饰器集成进自定义方法。当被装饰的方法执行失败时，会按照重试装饰器中预设的重试策略进行重试：

import ccxt
import time
import json

# 定义重试装饰器
def retry(max_retries=3, delay=5):
    def decorator(func):
        def wrapper(*args, **kwargs):
            for attempt in range(max_retries):
                try:
                    return func(*args, **kwargs)
                # 交易所请求超时重试，先要捕获子类
                except ccxt.RequestTimeout as e:
                    wait_time = delay * (attempt + 1)
                    print(f"请求超时，重试第 {attempt + 1}/{max_retries} 次，等待 {wait_time} 秒...")
                    time.sleep(wait_time)
                except ccxt.NetworkError as e:
                    wait_time = delay * (attempt + 1)
                    print(f"网络错误，重试第 {attempt + 1}/{max_retries} 次，等待 {wait_time} 秒...")
                    time.sleep(wait_time)
                except ccxt.BaseError as e:
                    # 尝试解析错误代码
                    error_code = None
                    try:
                        raw_message = e.args[0]  # e.args 是一个 tuple
                        if isinstance(raw_message, str) and '{' in raw_message:
                            error_data = json.loads(raw_message.split('binance')[-1].strip())
                            error_code = error_data.get('code')
                            # 输出解析出的错误码,是int
                            # print(error_code)
                    except Exception as parse_err:
                        print(f"错误码解析失败: {parse_err}")

                    # 根据不同错误码处理错误
                    if error_code == -1008:
                        wait_time = delay * (attempt + 1)
                        print(f"服务器过载（code -1008），重试第 {attempt + 1}/{max_retries} 次，等待 {wait_time} 秒...")
                        time.sleep(wait_time)
                    else:
                        print(f"操作失败：{e}")
                        raise

        return wrapper
    return decorator


# 查询合约指定币对价格，使用重试装饰器避免崩溃
@retry(max_retries=100, delay=2)  # 最大重试100次，初始延迟2秒
def fetch_ticker_price(exchange, para_symbol):
    return exchange.fapipublic_get_ticker_price({'symbol': para_symbol})

# 实例化 Binance 交易所对象
binance = ccxt.binance({
    'apiKey': '',
    'secret': '',
    'timeout': 10000,
    'enableRateLimit': True
})

# 加载市场数据,完成初始化
markets = binance.load_markets()

symbol = 'BTCUSDT'
while True:
    ticker_data = fetch_ticker_price(binance, symbol)
    print(ticker_data)

使用重试装饰器处理崩溃问题并非一蹴而就，代码可能总是在意想不到的情况下崩溃，每次崩溃以后查看异常抛出信息，增加重试装饰器处理新错误的逻辑，方能使代码越来越健全。

具体错误可以直接阅读官方API文档中关于错误码的介绍部分，直接穷尽处理所有错误代码情况（巨大工作量），可一步到位！

总结

看完本文你应该已经可以通过查阅各交易所官方文档来优雅的使用CCXT库编写一个高效稳定的量化脚本来帮你操作账户了。祝你开发顺利，早日成为加密大佬！