用 Python 监测币安上币价波动并进行套利

我将向您介绍的机器人能够分析 Binance 上所有代币的价格变化，并在波动最大的代币上进行交易。除此之外，此币安交易算法还将根据您指定的止损和止盈跟踪所有买入和卖出的代币。

要求

•币安账户•GitHub 帐户•币安测试网和主网API密钥•一些python库•一些Python知识

开始

首先，如果你没有币安Binance 帐户，或者想创建一个单独的帐户来运行机器人，请继续创建一个。您可以使用我的推荐链接获得 20% 的交易费用返现，在电脑中将下面的地址复制到浏览器，或者用手机长按扫描下方二维码。创建 Binance 帐户后，我们将生成一些 API 密钥。我们将从主网密钥开始，然后是测试网密钥。

币安Binance返现20%注册链接：

https://www.binancezh.sh/zh-CN/register?ref=CS346NXT

币安Binance返现20%注册邀请码：CS346NXT

主网密钥

要获取您的主网密钥，请浏览您的帐户，然后单击 API 管理。

在 API 管理页面，为您的 API 密钥添加标签，然后单击创建 API。

生成您的密钥和安全密码后，请确保保存这些，因为您只会看到一次密码。如果您尚未保存密钥，则需要重新生成密钥。

现在让我们为其授予在其上运行机器人所需的权限。我们只要做现货交易，所以我们只需要选择允许读取和允许现货及杠杆交易。

测试网密钥

前往 testnet.binance.vision 以创建一个测试网帐户。

使用 GitHub 登录并使用您的 GitHub 帐户进行身份验证。如果您没有 GitHub 帐户，请直接前往 GitHub 并先创建一个。使用 GitHub 登录后，单击 Generate HMAC_SHA256 Key。

为您的密钥提供描述，然后单击生成。

出于安全原因，保存显示的 API 密钥和安全密码很重要，这是您唯一能够看到它们的时间。如果您还没有保存它们，只需重新生成密钥。

编写币安交易机器人

有了币安账户和密钥，现在到了最激动人心的部分——用 Python 编写我们的交易机器人！让我们首先明确定义这个机器人的相关参数：

定义参数

•机器人将监听币安上所有代币的价格变化•默认情况下，我们只选择 USDT 交易对•我们不选择ETF类的交易对（如 BTCDOWNUSDT）和法币交易对•机器人会检查所有代币在过去 5 分钟内上涨是否超过 3%•机器人将在币安上购买 100 USDT 波动最大的货币品种•机器人将以 6% 的利润或 3% 的止损出售

这些都很容易配置，我鼓励您使用这些参数。

开始进行编码

导入模块

我们要做的第一件事是导入脚本所需的模块。python-binance 需要使用 pip install python-binance 命令安装。

# 用于环境变量import os# 币安 API 和 websockets 需要from binance.client importClient# 用于日期from datetime import datetime, timedeltaimport time# 用于重复执行代码from itertools import count# 用于存储交易和出售资产import json

配置币安客户端

我们现在需要在我们的脚本中存储 Binance API 密钥。为了能够轻松地在测试网和主网之间切换，有一个简短的条件语句将根据 TESTNET 变量中的值向币安客户端提供正确密钥。如果您想使用测试网，请将其设置为 True，或者（风险自担）将其设置为 False 以使用主网。使用主网，您将使用账户中的真实资金进行交易，请在此处特别注意。

# 在测试网和主网之间切换# 将此设置为 False 将使用真实资金，使用风险自负# 在下面定义您的 API 密钥以便切换工作


    
TESTNET = True# 获取 TEST 和 MAINNET 的币安密钥和安全密码# 下面的键是使用 `os.getenv` 从环境变量中提取的# 只需删除它并使用以下格式：api_key_test = '你的API密钥' 来代替api_key_test = os.getenv('binance_api_stalkbot_testnet')api_secret_test = os.getenv('binance_secret_stalkbot_testnet')api_key_live = os.getenv('binance_api_stalkbot_live')api_secret_live = os.getenv('binance_secret_stalkbot_live')# 与客户端进行身份验证if TESTNET:    client = Client(api_key_test, api_secret_test)# API URL 需要在库中手动更改才能在 TESTNET 上工作    client.API_URL = 'https://testnet.binance.vision/api'else:    client = Client(api_key_live, api_secret_live)

这里要注意的另一件事是 API 密钥的语法。在上面的代码中，我使用 os.getenv 方法来调用我存储在我的机器上的 API 密钥。如果您不打算公开代码，则不必这样做，因此只需将密钥的语法重新编写为 api_key_test = “YOUR_API_KEY”

定义用户输入

是时候定义我们的用户输入——或者说我们的币安交易机器人将进行交易的参数。您可以修改这些以更改算法的参数以测试不同的策略，默认情况下我们使用以下内容：

# 选择与代币配对的对象PAIR_WITH = 'USDT'# 定义每笔交易的大小，默认为USDTQUANTITY = 100# 要排除的交易对列表# 默认情况下，我们排除了最受欢迎的法定货币对# 和一些保证金关键字，因为我们只在 SPOT 帐户上工作FIATS = ['EURUSDT', 'GBPUSDT', 'JPYUSDT', 'USDUSDT', 'DOWN', 'UP']# 计算与当前价格差异的时间量（分钟级）TIME_DIFFERENCE = 5# 第一次和第二次价格检查之间的差异（%），默认设置为 10 分钟。CHANGE_IN_PRICE = 3# 定义何时出售没有盈利的代币（以%为单位）STOP_LOSS = 3# 定义何时在盈利的代币上获利（以%为单位）TAKE_PROFIT = 6

•PAIR_WITH 定义了与每个加密货币配对的代币（或法币）。本文只测试了这个 USDT，因为大多数代币都与它配对。
•QUANTITY 代表交易量或者交易规模，默认为 USDT。例如，如果您将 PAIR_WITH 更改为 BNB，请格外小心 QUANTITY。
•FIATS 是我排除的法定货币和保证金符号列表。此处添加的任何内容都将被排除在代币输出之外，并且不会进行交易。
•TIME_DIFFERENCE 默认情况下，我们会检查过去 5 分钟内币安上每个代币的价格差异，您可以更改此值以获得不同的结果。这也决定了每次代码迭代执行的频率。
•CHANGE_IN_PRICE 机器人决定购买代币的门槛。默认情况下，如果代币在过去 5 分钟内波动超过 3%，我们认为这是一个强烈的买入信号。
•STOP LOSS 和 TAKE PROFIT 以百分比定义如何出售购买的代币。

生成币安机器人组合

下一步是检查机器人是否已经进行了任何交易，如果是，则生成机器人投资组合。默认情况下，机器人会将每笔交易保存在与脚本相同的目录中的 json 文件中，以便我们可以跟踪交易并在达到 TP 或 SL 时卖出。

# 机器人自开始以来购买的代币coins_bought = {}# 保存的coins_bought 文件的路径coins_bought_file_path = 'coins_bought.json'# 为 testnet 和 live 使用单独的文件if TESTNET:    coins_bought_file_path = 'testnet_'+ coins_bought_file_path# 如果保存的coins_bought json文件存在则加载它


    
if os.path.isfile(coins_bought_file_path):with open(coins_bought_file_path) as file:        coins_bought = json.load(file)

获取币安上列出的每个代币的当前价格

是时候让我们的算法读取每枚代币的价格了。 get_price() 函数将返回符合我们标准的每个代币的价格。

def get_price():'''Return the current price for all coins on binance'''    initial_price = {}    prices = client.get_all_tickers()for coin in prices:# 仅返回 USDT 对并排除保证金符号，如 BTCDOWN USDTif PAIR_WITH in coin['symbol'] and all(item notin coin['symbol'] for item in FIATS):            initial_price[coin['symbol']] = { 'price': coin['price'], 'time': datetime.now()}return initial_price

等待再次获取价格

下一个函数将根据 TIME_DIFFERENCE 变量中定义的时间等待，并返回任何移动超过 CHANGE_IN_PRICE 的代币——默认为 3%。

def wait_for_price():'''calls the initial price and ensures the correct amount of time has passed    before reading the current price again'''    volatile_coins = {}    initial_price = get_price()while initial_price['BNBUSDT']['time'] > datetime.now() - timedelta(minutes=TIME_DIFFERENCE):print(f'not enough time has passed yet...')#让我们在这里等待，直到时间过去......        time.sleep(60*TIME_DIFFERENCE)else:        last_price = get_price()# 计算第一个和最后一个价格读数之间的差异for coin in initial_price:            threshold_check = (float


    
(initial_price[coin]['price']) - float(last_price[coin]['price'])) / float(last_price[coin]['price']) * 100# 每个收益高于我们的 `CHANGE_IN_PRICE` 的代币都被添加到 `volatile_coins` 字典中if threshold_check > CHANGE_IN_PRICE:                volatile_coins[coin] = threshold_check                volatile_coins[coin] = round(volatile_coins[coin], 3)print(f'{coin} has gained {volatile_coins[coin]}% in the last {TIME_DIFFERENCE} minutes, calculating volume in {PAIR_WITH}')if len(volatile_coins) < 1:print(f'No coins moved more than {CHANGE_IN_PRICE}% in the last {TIME_DIFFERENCE} minute(s)')return volatile_coins, len(volatile_coins), last_price

计算从USDT转换为每个代币的数量

下一步是将我们 100USDT（默认）的 QUANTITY 转换为我们将要购买的每个代币的相应数量。因为 Binance 对交易量的格式有点特殊，我们的交易机器人需要知道每个代币的精度。例如 BTC 支持 6 个小数点的精度，而 XRP 只支持一个。所以如果我们要购买 XRP，我们需要确保数量格式正确。尝试购买 1.000 XRP 会失败，而 1.0 会被执行。

def convert_volume():'''Converts the volume given in QUANTITY from USDT to the each coin's volume'''    volatile_coins, number_of_coins, last_price = wait_for_price()


    
    lot_size = {}    volume = {}for coin in volatile_coins:# 为每个硬币找到正确的精度# 例如 BTC 的最大精度是 6 个小数点# 而 XRP 只有 1个try:            info = client.get_symbol_info(coin)            step_size = info['filters'][2]['stepSize']            lot_size[coin] = step_size.index('1') - 1if lot_size[coin] < 0:                lot_size[coin] = 0except:pass# 从 QUANTITY 以 USDT 计算代币的数量（默认）        volume[coin] = float(QUANTITY / float(last_price[coin]['price']))# 用正确的精度定义数量if coin notin


    
 lot_size:            volume[coin] = float('{:.1f}'.format(volume[coin]))else:            volume[coin] = float('{:.{}f}'.format(volume[coin], lot_size[coin]))return volume, last_price

进行交易

现在是时候让我们的 Binance 交易机器人为我们进行一些交易了。

def trade():'''Place Buy market orders for each volatile coin found'''    volume, last_price = convert_volume()    orders = {}for coin in volume:# 仅在代币没有活跃交易时才购买if coin notin coins_bought or coins_bought[coin] == None:print(f' preparing to buy {volume[coin]} {coin}')if TESTNET :


    
# 在推送实际订单之前创建测试订单                test_order = client.create_test_order(symbol=coin, side='BUY', type='MARKET', quantity=volume[coin])# 如果测试订单没有引发异常，尝试创建一个真正的订单try:                buy_limit = client.create_order(                    symbol=coin,                    side='BUY',                    type='MARKET',                    quantity=volume[coin])# 此处错误处理以防变量无法放置exceptExceptionas e:print(e)# 如果仓位已被放置并返回订单信息，则运行 else 块else:                orders[coin] = client.get_all_orders(symbol=coin, limit=1)else:print(f'Signal detected, but there is already an active trade on {coin}')return orders, last_price, volume

更新投资组合

下一步是通过将每笔交易的详细信息保存到我们在每次迭代开始时检查的 json 文件来更新我们的投资组合。默认情况下，我们为每个代币保存符号、订单 ID、时间戳、购买价格和数量。

您可以通过 client.get_all_orders(symbol=coin, limit=1) 的格式查看更多信息，并添加您认为必要的其他信息。

def update_porfolio(orders, last_price, volume):'''add every coin bought to our portfolio for tracking/selling later'''for coin in orders:        coins_bought[coin] = {'symbol': orders[coin][0]['symbol'],'orderid': orders[coin][0]['orderId'],'timestamp': orders[coin][0]['time'],'bought_at': last_price[coin]['price'],'volume': volume[coin]}# 将代币保存在同一目录下的json文件中with open(coins_bought_file_path, 'w') 


    
as file:            json.dump(coins_bought, file, indent=4)

执行卖单

这个函数（也是最后一个！）检查我们在我们的机器人投资组合中拥有的任何代币是否应该因为达到 SL 或 TP 而被出售。请注意我们使用 Spot 帐户的代币，因此我们只能出售我们拥有的代币。如果硬币下跌并且我们不拥有它，我们不能下卖单。但我很想知道币安交易机器人在期货账户上的表现如何。

def sell_coins():'''sell coins that have reached the STOP LOSS or TAKE PROFIT thershold'''    last_price = get_price()for coin in coins_bought:# 定义止损和止盈        TP = float(coins_bought[coin]['bought_at']) + (float(coins_bought[coin]['bought_at']) * TAKE_PROFIT) / 100        SL = float(coins_bought[coin]['bought_at']) - (float(coins_bought[coin]['bought_at']) * STOP_LOSS) / 100# 检查价格是否高于止盈或低于止损iffloat(last_price[coin]['price']) > TP orfloat(last_price[coin]['price']) < SL:print(f


    
"TP or SL reached, selling {coins_bought[coin]['volume']} {coin}...")if TESTNET :# 在推送实际订单之前创建测试订单                test_order = client.create_test_order(symbol=coin, side='SELL', type='MARKET', quantity=coins_bought[coin]['volume'])# 如果测试订单没有引发异常，尝试创建一个真正的订单try:                sell_coins_limit = client.create_order(                    symbol=coin,                    side='SELL',                    type='MARKET',                    quantity=coins_bought[coin]['volume'])# 此处错误处理以防变量无法放置exceptExceptionas e:print(e)# 如果仓位已放置，则运行 else 块并更新购买的代币 json 文件else:                coins_bought[coin] = Nonewith open(coins_bought_file_path, 


    
'w') as file:                    json.dump(coins_bought, file, indent=4)else:print(f'TP or SL not yet reached, not selling {coin} for now...')

把所有代码放在一起

我们代码中的最后一段将所有函数组合在一起，并使代码继续执行，直到您停止它为止。

if __name__ == '__main__':print('Press Ctrl-Q to stop the script')for i in count():        orders, last_price, volume = trade()        update_porfolio(orders, last_price, volume)        sell_coins()

就是这样。您拥有一个功能齐全的 Binance 交易机器人，它可以检测 Binance 上波动最大的代币，并根据每个代币的表现进行买卖。