如何构建加密货币套利机器人（Python 指南）_加密百科

在金融领域，套利是一种策略，即在一个市场买入证券、商品或货币，然后在另一个市场以更高的价格卖出。两个市场之间的差价减去交易费用即为利润。

对于加密货币而言，大多数交易活动都是全天候在众多交易所进行的。资产的价格由供求经济因素决定，而不同交易所的供求经济因素可能并不一致。跨交易所套利策略可以利用这种差异，从而为交易者提供潜在的获利机会。

在本教程中，我们将讲解如何构建一个基于 Python 的套利机器人，该机器人通过追踪加密货币价格和其他加密货币交易所的有用指标来识别套利机会。设置好机器人后，您就可以使用Python 交易机器人在交易所进行自动化交易。

加密套利机器人 - 套利机器人交易 - Python 编程

我们将使用CoinGecko API中的 Demo 端点来检索加密货币价格数据。这些端点可以免费访问，但需要密钥身份验证。

/exchanges- 获取所有受支持的交易所列表及其相关数据（ID、名称、国家/地区等）的端点。请注意，只有在 CoinGecko 上有活跃交易量的交易所才会被列出。
/exchanges/{id}/tickers- 根据给定交易所 ID 查询特定股票代码的端点。响应包含有用数据，例如上次交易时间、最新价格和交易量。
/exchange_rates- 获取 BTC 与其他货币的汇率列表。
/exchanges/{id}/volume_chart;- 获取带有 unix 时间戳的选定交易所的历史总量数据（以 BTC 为单位）。

💡专业提示：任何人都可以注册 CoinGecko API 演示计划（免费！）并生成 API 密钥，每月上限为 10,000 次调用，每分钟 30 次调用的速率限制。

然而，由于我们的加密货币套利机器人需要全程运行，因此可能会遇到速率限制。在这种情况下，Pro API 密钥可能会有所帮助。

免责声明：任何使用或依赖我们内容的行为均由您自行承担风险和判断。在依赖我们的内容之前，请自行研究、审阅、分析和验证。交易是一项高风险活动，可能导致重大损失，因此，请在做出任何决定之前咨询您的财务顾问。

先决条件

我们将使用 Jupyter Notebook 来创建并运行该机器人。请确保已安装 Python 3 及以下附加软件包：

pip install jupyterlab

pip install pandas

pip install numpy

pip install pytz

要在新笔记本中开始编码，请执行以下命令：

jupyter lab

这会在浏览器窗口中打开一个新标签页。如果你想使用通过 GitHub 仓库共享的笔记本，请先克隆该仓库，然后在 Jupyter 环境中打开它。

1. 设置项目环境和 API 访问

可以按如下所示加载 Python 包：

	import requests as rq
	import json
	import pandas as pd
	pd.set_option('display.precision', 4,
	'display.colheader_justify', 'center')
	import numpy as np
	import warnings
	import pytz
	import datetime
	import time
	from IPython.display import clear_output

请注意，我们已经使用 pd.set_option 为 Pandas 库设置了额外的显示选项。这在之后会很有帮助，因为我们将使用 DataFrames 来可视化所有数据。此外，一旦我们开始运行机器人并希望数据表在 Notebook 中定期刷新，就需要使用 clear_output。

演示 API 密钥可以从本地文件读取。get_response函数会使用use_demo标头，该函数会为我们发出请求。状态码 200 表示请求成功。

	def get_demo_key():
	f = open("/home/vikas/Documents/CG_demo_key.json")
	key_dict = json.load(f)
	return key_dict["key"]

	use_demo = {
	"accept": "application/json",
	"x-cg-demo-api-key" : get_demo_key()
	}

	def get_response(endpoint, headers, params, URL):
	url = "".join((URL, endpoint))
	response = rq.get(url, headers = headers, params = params)
	if response.status_code == 200:
	data = response.json()
	return data
	else:
	print(f"Failed to fetch data, check status code {response.status_code}")

2. 获取所有加密货币交易所列表

要获取所有加密货币交易所的完整列表，请对CoinGecko API 中的/exchanges端点进行 API 调用。以下查询参数将确保我们在单个页面中获取所有结果。

	# Valid values for results per page is between 1-250
	exchange_params = {
	"per_page": 250,
	"page": 1
	}

使用之前定义的函数，我们现在可以发出请求。然后可以将结果转换为 Pandas DataFrame 以便于可视化。

	exchange_list_response = get_response("/exchanges", use_demo, exchange_params, PUB_URL)
	df_ex = pd.DataFrame(exchange_list_response)

以 DataFrame 形式显示的交易所列表

要按交易量列出交易所，可以轻松地根据“trade_volume_24h_btc”列对上述 DataFrame 进行排序。

	df_ex_subset = df_ex[["id", "name", "country", "trade_volume_24h_btc"]]
	df_ex_subset = df_ex_subset.sort_values(by = ["trade_volume_24h_btc"], ascending = False)

3. 获取加密货币交易所代码

对于每个交易所，都有多个加密货币代码（或交易对）的数据可用。我们希望筛选列表，找到我们感兴趣的代码的最新交易。向相关端点发出请求后，我们可以循环遍历响应，找到相关的基础货币和目标货币。如果未找到匹配项，则返回空字符串。这种情况可能发生，因为并非所有交易所都提供所有交易对。

	def get_trade_exchange(id, base_curr, target_curr):

	exchange_ticker_response = get_response(f"/exchanges/{id}/tickers",
	use_demo,
	{},
	PUB_URL)

	found_match = ""

	for ticker in exchange_ticker_response["tickers"]:
	if ticker["base"] == base_curr and ticker["target"] == target_curr:
	found_match = ticker
	break

	if found_match == "":
	warnings.warn(f"No data found for {base_curr}-{target_curr} pair in {id}")

	return found_match

以交易所“gdax”（Coinbase）的交易对（或代码）ETH-USD 为例，如下所示：

Coinbase 上的 ETH-USD 代码

4. 将时间戳转换为本地时区（Python）

如上所示，返回的数据带有不同时区的时间戳。在监控机器人活动时，将其转换为我们自己的时区会很有帮助。使用 pytz 库可以轻松实现，例如“Europe/Amsterdam”的时区转换。

	def convert_to_local_tz(old_ts):

	new_tz = pytz.timezone("Europe/Amsterdam")
	old_tz = pytz.timezone("UTC")

	format = "%Y-%m-%dT%H:%M:%S+00:00"
	datetime_obj = datetime.datetime.strptime(old_ts, format)

	localized_ts = old_tz.localize(datetime_obj)
	new_ts = localized_ts.astimezone(new_tz)

	return new_ts

5. 获取多个交易所的加密货币行情数据

现在我们已经知道如何获取单个交易所的股票行情数据，我们可以扩展相同的逻辑来收集多个交易所的数据。为了专注于特定市场，我们还可以添加针对国家/地区的过滤器。从股票行情响应中，我们收集最新交易价格、最新交易量、价差和交易时间（转换为当地时区）。如果交易对未在指定交易所上线，则会显示警告。

	def get_trade_exchange_per_country(country,
	base_curr,
	target_curr):

	df_all = df_ex_subset[(df_ex_subset["country"] == country)]

	exchanges_list = df_all["id"]
	ex_all = []

	for exchange_id in exchanges_list:
	found_match = get_trade_exchange(exchange_id, base_curr, target_curr)
	if found_match == "":
	continue
	else:
	temp_dict = dict(
	exchange = exchange_id,
	last_price = found_match["last"],
	last_vol = found_match["volume"],
	spread = found_match["bid_ask_spread_percentage"],
	trade_time = convert_to_local_tz(found_match["last_traded_at"])
	)
	ex_all.append(temp_dict)

	return pd.DataFrame(ex_all)

买卖价差百分比是指卖方对某项资产的最低报价与潜在买方的最高出价之间的差额。价差越低，表明该资产在交易所的流动性和交易量越高。相反，价差越高通常表明流动性越低。因此，该指标可用于判断是否应考虑在某个交易所进行套利交易。

6. 获取多种货币的比特币汇率

各个端点的数据（例如，交易所交易量）以BTC为单位报告。对于我们的机器人来说，额外确定特定代码的交易量占总交易量的百分比将会很有趣，这可以让我们进一步了解特定交易所的流动性。

为了将 BTC 转换为不同的目标货币，我们可以通过 CoinGecko API 获取汇率，如下所示：

	def get_exchange_rate(base_curr):

	# This returns current BTC to base_curr exchange rate
	exchange_rate_response = get_response(f"/exchange_rates",
	use_demo,
	{},
	PUB_URL)
	rate = ""
	try:
	rate = exchange_rate_response["rates"][base_curr.lower()]["value"]
	except KeyError as ke:
	print("Currency not found in the exchange rate API response:", ke)

	return rate

7. 获取历史交易量数据

利用给定时间段的历史交易量数据，我们可以使用 7 天窗口期的简单移动平均线来确定最新交易量。然后，可以使用上一节确定的汇率将该交易量（默认为 BTC）转换为我们感兴趣的货币。一旦我们知道了总交易量（所有股票代码的总和），就可以轻松确定我们股票代码交易量的百分比，稍后我们将展示这一点。

	def get_vol_exchange(id, days, base_curr):

	vol_params = {"days": days}

	exchange_vol_response = get_response(f"/exchanges/{id}/volume_chart",
	use_demo,
	vol_params,
	PUB_URL)

	time, volume = [], []

	# Get exchange rate when base_curr is not BTC
	ex_rate = 1.0
	if base_curr != "BTC":
	ex_rate = get_exchange_rate(base_curr)

	# Give a warning when exchange rate is not found
	if ex_rate == "":
	print(f"Unable to find exchange rate for {base_curr}, vol will be reported in BTC")
	ex_rate = 1.0

	for i in range(len(exchange_vol_response)):
	# Convert to seconds
	s = exchange_vol_response[i][0] / 1000
	time.append(datetime.datetime.fromtimestamp(s).strftime('%Y-%m-%d'))

	# Default unit for volume is BTC
	volume.append(float(exchange_vol_response[i][1]) * ex_rate)

	df_vol = pd.DataFrame(list(zip(time, volume)), columns = ["date", "volume"])

	# Calculate SMA for a specific window
	df_vol["volume_SMA"] = df_vol["volume"].rolling(7).mean()

	return df_vol.sort_values(by = ["date"], ascending = False).reset_index(drop = True)

Kraken上 30 天交易量的示例（为简洁起见仅显示 15 行）如下所示：

8. 汇总并显示加密货币交易所的交易

在运行机器人之前，重要的是要思考如何随时间汇总数据。本质上，我们的机器人会定期获取最新的交易数据。对于某些交易所，在此期间可能没有新的交易发生；而对于其他交易所，可能会执行许多交易。因此，删除重复数据至关重要。然后，可以使用唯一值的数量来确定交易数量。此外，为了构建可靠的套利策略，收集一段时间内的统计数据也很有帮助。因此，我们将对交易所 ID 执行分组操作，并计算所有相关列的平均值。

此外，还将添加一个包含总交易量百分比的新列（如前几节所示）。

	def display_agg_per_exchange(df_ex_all, base_curr):

	# Group data and calculate statistics per exchange
	df_agg = (
	df_ex_all.groupby("exchange").agg
	(
	trade_time_min = ("trade_time", 'min'),
	trade_time_latest = ("trade_time", 'max'),
	last_price_mean = ("last_price", 'mean'),
	last_vol_mean = ("last_vol", 'mean'),
	spread_mean = ("spread", 'mean'),
	num_trades = ("last_price", 'count')
	)
	)

	# Get time interval over which statistics have been calculated
	df_agg["trade_time_duration"] = df_agg["trade_time_latest"] - df_agg["trade_time_min"]

	# Reset columns so that we can access exchanges below
	df_agg = df_agg.reset_index()

	# Calculate % of total volume for all exchanges
	last_vol_pert = []
	for i, row in df_agg.iterrows():
	try:
	df_vol = get_vol_exchange(row["exchange"], 30, base_curr)
	current_vol = df_vol["volume_SMA"][0]
	vol_pert = (row["last_vol_mean"] / current_vol) * 100
	last_vol_pert.append(vol_pert)
	except:
	last_vol_pert.append("")
	continue

	# Add % of total volume column
	df_agg["last_vol_pert"] = last_vol_pert

	# Remove redundant column
	df_agg = df_agg.drop(columns = ["trade_time_min"])

	# Round all float values
	# (seems to be overwritten by style below)
	df_agg = df_agg.round({"last_price_mean": 2,
	"last_vol_mean": 2,
	"spread_mean": 2
	})

	display(df_agg.style.apply(highlight_max_min,
	color = 'green',
	subset = "last_price_mean")
	)

	return None

我们将使用以下功能进一步突出显示具有最高价和最低价的交易所，以便更好地了解情况：

	def highlight_max_min(x, color):

	return np.where((x == np.nanmax(x.to_numpy())) \|
	(x == np.nanmin(x.to_numpy())),
	f"color: {color};",
	None)

9. 运行加密货币交易套利机器人

我们的机器人需要持续监控多个交易所的最新交易。因此，我们将执行一个包含 while 语句的单元。这将使代码持续运行，直到用户停止。为了在更新之间引入一分钟的延迟，我们将使用 sleep 语句。由于 API 本身在演示计划中每分钟都会刷新，因此无需设置更短的延迟。

💡专业提示：升级到付费计划可解锁 30 秒的数据新鲜度（免费演示计划为 1 分钟）。

	def run_bot(country,
	base_curr,
	target_curr):

	df_ex_all = get_trade_exchange_per_country(country, base_curr, target_curr)

	# Collect data every minute
	while True:
	time.sleep(60)
	df_new = get_trade_exchange_per_country(country, base_curr, target_curr)

	# Merge to existing DataFrame
	df_ex_all = pd.concat([df_ex_all, df_new])

	# Remove duplicate rows based on all columns
	df_ex_all = df_ex_all.drop_duplicates()

	# Clear previous display once new one is available
	clear_output(wait = True)
	display_agg_per_exchange(df_ex_all, base_curr)

	return None