数字货币交易接口实现（含源代码）

本系列将打造一套数字货币量化交易系统，跟上一个专栏有很多相似之处 七天打造一套量化交易系统 。今天主要分享数字货币交易主流的接口平台：币安和OKX的对接方式，以及如何获取全市场合约，如何获取数字货币实时行情数据。

在此基础上，结合前面的文章 七天打造一套量化交易系统：Day3-回测系统的选择、搭建及改造 ，后续我也会将数字货币历史行情数据整合进去，从而能够支持数字货币交易策略的回测。

使用币安交易接口

步骤1：注册API密钥

1. 登录到你的币安账户。
2. 进入API管理页面。
3. 创建一个新的API密钥，并确保你记录下API密钥和秘密密钥。这些密钥将用于身份验证。

步骤2：安装所需库

Python用户可以使用binance库与币安API进行交互。你可以使用pip安装这个库：

pip install python-binance

步骤3：使用API进行交易

以下是一些基本的操作示例，包括如何获取市场数据、查看账户信息和执行交易。

获取市场数据

from binance.client import Clientapi_key = 'your_api_key'
api_secret = 'your_api_secret'client = Client(api_key, api_secret)# 获取市场价格
prices = client.get_all_tickers()
print(prices)# 获取订单簿深度
depth = client.get_order_book(symbol='BTCUSDT')
print(depth)

查看账户信息

# 获取账户信息
account = client.get_account()
print(account)# 获取账户余额
balances = client.get_asset_balance(asset='BTC')
print(balances)

执行交易

# 市价买入
order = client.order_market_buy(symbol='BTCUSDT',quantity=0.001
)
print(order)# 限价卖出
order = client.order_limit_sell(symbol='BTCUSDT',quantity=0.001,price='50000'
)
print(order)

错误处理

使用API时，可能会遇到各种错误和异常。你可以使用try和except来捕获这些错误：

from binance.exceptions import BinanceAPIException, BinanceOrderExceptiontry:# 例如，尝试市价买入order = client.order_market_buy(symbol='BTCUSDT',quantity=0.001)print(order)
except BinanceAPIException as e:# 捕获API错误print(e)
except BinanceOrderException as e:# 捕获订单错误print(e)

安全提示

1. 不要将API密钥暴露在公共代码库中。使用环境变量或配置文件来存储密钥。
2. 设置API密钥的权限，只允许必要的权限，例如读取市场数据和执行交易。
3. 定期更换API密钥，以确保安全性。

使用OKX交易接口

步骤1：注册API密钥

1. 登录到你的OKX账户。
2. 进入API管理页面。
3. 创建一个新的API密钥，并确保你记录下API密钥、秘密密钥和Passphrase。这些密钥将用于身份验证。

步骤2：安装所需库

Python用户可以使用requests库或更专门的库如okex与OKX API进行交互。你可以使用pip安装这些库：

pip install requests

步骤3：使用API进行交易

以下是一些基本的操作示例，包括如何获取市场数据、查看账户信息和执行交易。

获取市场数据

import requests
import json# 替换为你的API密钥、秘密密钥和Passphrase
api_key = 'your_api_key'
api_secret = 'your_api_secret'
passphrase = 'your_passphrase'# OKX API基础URL
base_url = 'https://www.okx.com'def get_ticker(symbol):endpoint = f'/api/v5/market/ticker?instId={symbol}'url = base_url + endpointresponse = requests.get(url)return response.json()# 获取BTC/USDT市场价格
ticker = get_ticker('BTC-USDT')
print(json.dumps(ticker, indent=4))

查看账户信息

import time
import hmac
import hashlib
import base64def sign(message, secret):mac = hmac.new(bytes(secret, encoding='utf8'), bytes(message, encoding='utf-8'), digestmod='sha256')return base64.b64encode(mac.digest()).decode()def get_timestamp():return str(int(time.time()))def get_headers(api_key, api_secret, passphrase, method, endpoint):timestamp = get_timestamp()message = timestamp + method + endpointsignature = sign(message, api_secret)headers = {'OK-ACCESS-KEY': api_key,'OK-ACCESS-SIGN': signature,'OK-ACCESS-TIMESTAMP': timestamp,'OK-ACCESS-PASSPHRASE': passphrase,'Content-Type': 'application/json'}return headers# 获取账户余额
def get_account_balance():endpoint = '/api/v5/account/balance'url = base_url + endpointheaders = get_headers(api_key, api_secret, passphrase, 'GET', endpoint)response = requests.get(url, headers=headers)return response.json()balance = get_account_balance()
print(json.dumps(balance, indent=4))

执行交易

def place_order(symbol, side, size, price=None):endpoint = '/api/v5/trade/order'url = base_url + endpointtimestamp = get_timestamp()order_data = {"instId": symbol,"tdMode": "cash", # "cash" for spot trading"side": side,"ordType": "limit" if price else "market","sz": size}if price:order_data["px"] = pricebody = json.dumps(order_data)message = timestamp + 'POST' + endpoint + bodysignature = sign(message, api_secret)headers = {'OK-ACCESS-KEY': api_key,'OK-ACCESS-SIGN': signature,'OK-ACCESS-TIMESTAMP': timestamp,'OK-ACCESS-PASSPHRASE': passphrase,'Content-Type': 'application/json'}response = requests.post(url, headers=headers, data=body)return response.json()# 市价买入0.001 BTC
order_response = place_order('BTC-USDT', 'buy', '0.001')
print(json.dumps(order_response, indent=4))

错误处理

使用API时，可能会遇到各种错误和异常。你可以使用try和except来捕获这些错误：

try:# 例如，尝试获取账户余额balance = get_account_balance()print(json.dumps(balance, indent=4))
except requests.exceptions.RequestException as e:# 捕获HTTP错误print(e)
except Exception as e:# 捕获其他错误print(e)

安全提示

1. 不要将API密钥暴露在公共代码库中。使用环境变量或配置文件来存储密钥。
2. 设置API密钥的权限，只允许必要的权限，例如读取市场数据和执行交易。
3. 定期更换API密钥，以确保安全性。

获取数字货币全市场合约

要获取全市场的数字货币合约数据，可以使用加密货币交易所的API，例如 Binance、CoinGecko、CoinMarketCap 等。以下是使用 ccxt 库来从 Binance 获取所有交易对（合约）的示例代码。

步骤1

首先，你需要安装 ccxt 库（如果尚未安装）：

pip install ccxt

步骤2

使用以下 Python 代码来获取 Binance 交易所的所有交易对（合约）：

import ccxtdef get_binance_markets():exchange = ccxt.binance()markets = exchange.load_markets()return markets# 获取 Binance 所有市场的合约信息
markets = get_binance_markets()# 打印出部分合约信息
for symbol, market in list(markets.items())[:10]:  # 仅打印前10个合约信息print(f"Symbol: {symbol}")print(f"Market Info: {market}")

代码说明

在这段代码中：

1. ccxt.binance() 创建一个 Binance 交易所的实例。
2. exchange.load_markets() 加载所有市场数据，包括所有交易对（合约）。
3. markets 是一个字典，其中键是交易对符号（例如 ‘BTC/USDT’），值是市场信息的字典。

你可以根据需要调整代码来获取更多合约数据或使用其他交易所的API。记得查阅 ccxt 和交易所 API 的文档，以了解如何使用它们获取具体数据和处理限制。

获取数字货币实时行情数据

import ccxt
import timedef get_binance_ticker(symbol):exchange = ccxt.binance()ticker = exchange.fetch_ticker(symbol)return tickerdef main():symbol = 'BTC/USDT'  # 你可以根据需要修改交易对while True:ticker = get_binance_ticker(symbol)print(f"Symbol: {ticker['symbol']}")print(f"Last Price: {ticker['last']}")print(f"Bid Price: {ticker['bid']}")print(f"Ask Price: {ticker['ask']}")print(f"24h High: {ticker['high']}")print(f"24h Low: {ticker['low']}")print(f"24h Volume: {ticker['baseVolume']}")print("-----")time.sleep(10)  # 每10秒更新一次if __name__ == "__main__":main()

代码说明

1. ccxt.binance()：创建一个 Binance 交易所实例。
2. exchange.fetch_ticker(symbol)：获取指定交易对的最新行情数据。symbol 是交易对的名称，例如 ‘BTC/USDT’。
3. ticker 是一个包含行情数据的字典，‘last’ 是最新价格，‘bid’ 是当前的买入价格，‘ask’ 是当前的卖出价格，‘high’ 和 ‘low’ 是24小时内的最高和最低价格，‘baseVolume’ 是24小时内的成交量。
4. time.sleep(10)：每10秒钟更新一次数据。你可以根据需要调整时间间隔。

这种方法可以帮助你获取实时行情数据并在终端中显示。如果需要获取多个交易对的数据，只需修改 symbol 变量的值，或者在 main 函数中使用循环遍历多个交易对。