GTokenTool全网最好的代币发行工具平台
当前位置：首页 >> 加密百科 >> 使用 Python 进行算法交易：如何在 DEX 上执行交易

使用 Python 进行算法交易：如何在 DEX 上执行交易

admin 加密百科 15

与 Coinbase 或币安等中心化交易所 (CEX) 相比，在 Uniswap 等去中心化交易所 (DEX) 上交易有着根本的不同。DEX 上的用户连接他们的加密钱包，并直接与智能合约交互，以便在特定网络上兑换不同的代币。最广泛使用的 DEX 基于自动做市商 (AMM) 模型，用户可以通过资金池获取流动性。代币兑换在资金池中进行，并根据智能合约的规定收取费用。因此，同一交易对可能存在多个高流动性资金池，有时只是费用不同。


尽管 DEX 近年来颇受欢迎，但它们通常缺乏 CEX 那样用户友好的监控界面。用户还需要使用自己的钱包手动连接到 DEX，这增加了额外的复杂性，尤其是在执行多笔交易时。因此，使用 DEX API 来自动化交易的监控和执行非常有用。


在本教程中，我们将学习如何利用CoinGecko API中的链上 DEX 端点来监控不同矿池的交易活动。此外，我们还将使用 Python 监控加密钱包的余额，并在 Uniswap 等 DEX 上执行兑换交易。

让我们开始吧！

使用 Python 进行算法交易 - 在 DEX 上执行和自动化加密货币交易

先决条件

我们将使用 Jupyter Notebook 作为主要开发环境。此外，还需要安装以下软件包：

pip install uniswap

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

jupyter lab

这会在浏览器中打开一个新标签页。如果您想使用通过GitHub 仓库共享的笔记本，请先克隆该仓库，然后在 Jupyter 环境中打开它。请务必将 API 密钥文件的路径替换为您自己的路径。

💡 专业提示：请勿将 API 密钥直接插入笔记本。这是一种不安全的做法，并且如果笔记本通过公共 GitHub 存储库共享，则存在密钥泄露的风险。

您还需要一个支持 ERC-20（以太坊）代币的加密钱包。MetaMask 易于设置，可作为浏览器扩展程序使用，因此推荐使用。稍后我们将使用 MetaMask 内部生成的账户地址和主密钥。要查看主密钥，请点击查看账户摘要时显示的三个点（右上角）。选择“账户详情”，然后点击“显示主密钥”。系统将提示您输入钱包密码，然后按住“显示”按钮。

如何在 MetaMask 钱包上显示主密钥？

一旦主键可见，就将其复制到文件并以以下 JSON 格式在本地保存：

{ “primary_key” : “XXYYZZ”}

💡 专业提示：钱包账户的主密钥赋予了与该地址关联的所有资金的完全访问权限，因此，您在处理它时应格外小心。请勿与任何人共享主密钥，也不要将 JSON 文件上传到代码库或云备份。如果您想测试本教程中的代码，最好创建一个新的钱包账户/地址，并向其中转入少量加密货币。

最后，我们需要一个 RPC 端点来与以太坊区块链进行交互。我们可以使用 Ankr 提供的免费 RPC 端点：https://rpc.ankr.com/eth

虽然此端点有速率限制，但对于我们的演示用例来说已经足够了。如果您需要更高的速率限制，请考虑切换到他们的付费计划。另一个选择是使用Infura的免费计划，该计划提供单个 API 密钥和丰厚的每日积分。

CoinGecko API 链上端点

我们将使用以下链上 DEX 端点，所有CoinGecko API 付费计划均可访问这些端点：

我在上一篇文章中概述了如何设置项目环境和安全 API 访问。

如何监控趋势DEX池

由于 DEX 上的交易是基于加密货币池提供的流动性执行的，因此能够跟踪不同池中的代币活动对于制定明智的交易策略至关重要。现在，我们将深入探讨一些使用 CoinGecko API 的链上 DEX 数据 并结合 Python 的示例。

让我们首先编写一个便利函数，它将帮助我们发出 API 请求。

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}")
接下来，我们将使用以下辅助函数，根据所需的 API 端点生成正确的 URL。
def get_url(url_type,
        network,
        dex = "",
        pool_address = "",
        token_address = ""):
url_dict = {
    "trending_pools": f"/onchain/networks/{network}/trending_pools",
    "top_pools": f"/onchain/networks/{network}/pools",
    "top_pools_dex": f"/onchain/networks/{network}/dexes/{dex}/pools",
    "specific_pool_dex": f"/onchain/networks/{network}/pools/{pool_address}",
    "top_pools_add": f"/onchain/networks/{network}/tokens/{token_address}/pools"
}
return url_dict[url_type]
虽然支持多种网络，但我们主要关注“ETH”。根据调用的 API，其他输入参数可以传递给此函数。

从 CoinGecko API 收到的响应将通过以下函数解析为 pandas DataFrame 的列：

def collect_response(list_response):
response_all = []
for response in list_response["data"]:
    all_attributes = response["attributes"]
    daily_tx = all_attributes["transactions"]["h24"]
    rel = response["relationships"]
    temp_dict = dict(
        pair = all_attributes["name"],
        dex = rel["dex"]["data"]["id"],
        add = all_attributes["address"],
        fdv_usd = all_attributes["fdv_usd"],
        market_cap_usd = all_attributes["market_cap_usd"],
        daily_volume = all_attributes["volume_usd"]["h24"],
        daily_price_change = all_attributes["price_change_percentage"]["h24"],
        daily_buys = daily_tx["buys"],
        daily_sells = daily_tx["sells"],
        daily_buyers = daily_tx["buyers"],
        daily_sellers = daily_tx["sellers"]
    )
    response_all.append(temp_dict)
return response_all
现在可以将上述所有内容组合起来，列出趋势池，并按所需列进行排序。
让我们按照“daily_volume”的降序检查趋势池。

热门矿池列表

如何追踪顶级 DEX 池

与前面的示例类似，我们也可以跟踪特定网络的顶级池。

def get_top_pools_network(network, sort_by_col):
target_url = get_url("top_pools", network)
toppool_list_response = get_response(target_url,
                                    use_pro,
                                    "",
                                    PRO_URL)
toppool_all = collect_response(toppool_list_response)
return pd.DataFrame(toppool_all).sort_values(by = [f"{sort_by_col}"],
                                            ascending = False)
让我们按照美元市值的降序排列它们。

顶级泳池列表

如上例所示，Uniswap 似乎是一个热门的去中心化交易所 (DEX)。对于同一个代币对，不同的池子可能根据其收取的费用而有所不同，费用通常在对名称旁边以百分比表示。查看 Uniswap 的顶级池子会很方便。值得庆幸的是，CoinGecko API 提供了一个端点！我们将在下一个示例中使用此端点，其中可以将特定的 DEX 作为输入参数提供。

def get_top_pools_dex(network, dex, sort_by_col):
target_url = get_url("top_pools_dex", network, dex)
toppool_list_response = get_response(target_url,
                                    use_pro,
                                    "",
                                    PRO_URL)
toppool_all = collect_response(toppool_list_response)
return pd.DataFrame(toppool_all).sort_values(by = [f"{sort_by_col}"],
                                            ascending = False)
现在我们可以轻松比较“uniswap_v2”和“uniswap_v3”之间的一些顶级池（按市值递减排列）。

Uniswap v2 上的顶级池

顶级矿池 Uniswap v3

如何获取特定池地址的数据

有时，调查特定矿池以确定与其日常活动相关的更有用的指标会很有用。使用矿池地址作为附加输入，可以使用以下函数来获得进一步的洞察：

def collect_pool_response(list_response):
response = list_response["data"]
all_attributes = response["attributes"]
daily_tx = all_attributes["transactions"]["h24"]
rel = response["relationships"]
response_dict = dict(
    pair = all_attributes["name"],
    dex = rel["dex"]["data"]["id"],
    add = all_attributes["address"],
    fdv_usd = all_attributes["fdv_usd"],
    market_cap_usd = all_attributes["market_cap_usd"],
    daily_volume = all_attributes["volume_usd"]["h24"],
    daily_price_change =  
all_attributes["price_change_percentage"]["h24"],
    daily_buys = daily_tx["buys"],
    daily_sells = daily_tx["sells"],
    daily_buyers = daily_tx["buyers"],
    daily_sellers = daily_tx["sellers"]
)
return response_dict
def get_pool_data(network, dex, pool_address):
target_url = get_url("specific_pool_dex", network, dex, pool_address)
pool_list_response = get_response(target_url,
                                  use_pro,
                                  "",
                                  PRO_URL)
pool_all = collect_pool_response(pool_list_response)
return pool_all

特定地址的池数据

如何监控特定代币的价格

到目前为止，我们已经研究并比较了不同池子的各种汇总指标。然而，我们还需要比较代币价格数据。这可以帮助用户开发和测试跨不同池子的套利策略。 

代币通常分为基础代币和报价代币。以 ETH/USDT 池为例，ETH为基础代币，USDT为报价代币。我们想知道 ETH 到 USDT 的兑换价格，该价格反映在下面的“base_token_price_quote_token”字段中：

def collect_response_token(list_response):
response_all = []
for response in list_response["data"]:
    all_attributes = response["attributes"]
    daily_tx = all_attributes["transactions"]["h24"]
    rel = response["relationships"]
    temp_dict = dict(
        pair = all_attributes["name"],
        dex = rel["dex"]["data"]["id"],
        add = all_attributes["address"],
        base_token_price_quote_token =
all_attributes["base_token_price_quote_token"],
        fdv_usd = all_attributes["fdv_usd"],
        market_cap_usd = all_attributes["market_cap_usd"],
        daily_volume = all_attributes["volume_usd"]["h24"],
        daily_price_change =
all_attributes["price_change_percentage"]["h24"],
        daily_buys = daily_tx["buys"],
        daily_sells = daily_tx["sells"],
        daily_buyers = daily_tx["buyers"],
        daily_sellers = daily_tx["sellers"]
    )
    response_all.append(temp_dict)
return response_all
本文的其余部分还将使用代币合约地址，该地址唯一地代表了代币在其原生区块链上的位置。您可以通过访问 DEX 的池子摘要页面找到该地址。例如，我们可以在 Uniswap 池子摘要页面的左下角（“链接”面板）找到 USDT 代币合约地址，如下所示：

USDT代币合约地址

使用代币合约地址作为输入，我们可以使用以下函数来获取给定网络和代币地址的顶级池列表。

def get_top_pools_token(network,
                    token_address,
                    sort_by_col):
target_url = get_url("top_pools_add",
                    network,
                    "",
                    "",
                    token_address)
toppool_list_response = get_response(target_url,
                                    use_pro,
                                    "",
                                    PRO_URL)
toppool_all = collect_response_token(toppool_list_response)
return pd.DataFrame(toppool_all).sort_values(by = [f"{sort_by_col}"],
ascending = False)
请注意，Uniswap 使用 WETH（Wrapped Ethereum），这是一种用于与 dApp 交互的 ETH 代币化形式。因此，交易对名称包含 WETH。我们将过滤上述函数的输出，仅列出名称中带有“WETH /”的行。此外，这些行可以按“base_token_price_quote_token”进行排序。

usdt = "0xdAC17F958D2ee523a2206206994597C13D831ec7"

df_token = get_top_pools_token("eth", usdt, "fdv_usd")

(
    df_token[df_token["pair"]
    .str.contains("WETH /")]
    .sort_values(by = ["base_token_price_quote_token"], ascending = True)
)

以上列表清晰地概述了套利机会——不仅涵盖不同 DEX 之间的套利机会，也涵盖同一 DEX 的不同版本（Uniswap v2、v3 等）之间的套利机会。接下来，我们将学习如何使用 Python 在 Uniswap 上执行掉期交易。为此，我们将以突出显示的池子（红色）为目标，卖出一些 ETH 并获得 USDT。然后，我们将以价格较低的池子（蓝色）为目标，以较低的 USDT 价格回购相同数量的 ETH。

算法交易：如何在 Uniswap 上兑换代币

我们将使用Uniswap Python 库与 Uniswap 协议进行交互，该库提供了各种便捷的函数来查询价格和执行掉期交易。这非常方便，因为许多 web3 实现都已从最终用户抽象出来。此外，它还可以用来监控用户钱包地址的余额。

如本文前面所述，要使用钱包地址进行交易，需要提供关联的主键。您可以将其保存在本地文件中，然后按如下所示读取：

# Get wallet private key
def get_private_key():
    f = open("/home/vikas/Documents/MetaMask_private_key.json")
    key_dict = json.load(f)
    return key_dict["private_key"]

接下来，需要设置 Uniswap 类：

from uniswap import Uniswap
# Use None for address and private_key when not doing a transaction
address = "0xAf418C54351BA8a0Aa15Ba4A5C99C46C122B3DBC"
private_key = get_private_key()
version = 3
provider = "https://rpc.ankr.com/eth"
uniswap = Uniswap(address = address,
              private_key = private_key,
              version = version,
              provider = provider)
还需要相关代币的合约地址（如前所述）。

# Token contract address
# https://support.uniswap.org/hc/en-us/articles/26757826138637-What-is-a-token-contract-address
eth = "0x0000000000000000000000000000000000000000"
usdt = "0xdAC17F958D2ee523a2206206994597C13D831ec7"

函数 get_price_output 返回兑换 X 笔 USDT 所需的 ETH 数量（以 wei 为单位）。注意，由于 1 ETH = 10¹⁸ wei，我们将输出结果除以 ETH，将结果转换回 ETH。在以下示例中，我们想知道兑换 3 笔 USDT 需要兑换多少 ETH。参数采用代币的最小单位，因此 USDT 输入为 3 x 10⁶。Fee = 100 将确保使用手续费为 0.01% 的池子。对于 Uniswap v3，建议先研究各种池子，然后选择一个在低手续费和高流动性之间取得最佳平衡的池子。

uniswap.get_price_output(eth, usdt, 3 * 10**6, fee = 100) / (10**18)

Get swap price for ETH-USDT pair

💡 专业提示：强烈建议预览掉期价格输出，如上所示。这有助于验证代币的输入金额是否符合预期，从而防止意外订单。

常见代币的小数：

  • ETH、DAI、UNI、BAT、LINK 使用 18 位小数

  • WBTC 使用 8 位小数

  • USDC、USDT 使用 6 位小数

总是可以在 Etherscan 上查找小数位数。

💡 专业提示： Uniswap Python 库中的函数使用常见的十进制格式，这意味着代币输入或输出采用最小单位。 

为了执行上述交换，我们调用 make_trade_output 函数，并使用与上述相同的参数集。

uniswap.make_trade_output(eth, usdt, 3 * 10**6, fee = 100)

要检查钱包余额，请使用以下功能：

检查 MetaMask 钱包余额

以上数据与下图MetaMask钱包上看到的一致。

MetaMask 钱包网页视图

此外，我们可以确认已收到预期数量的 USDT 代币（= 3）。该交易也将出现在目标池中的 Uniswap 网页界面上。

Uniswap 上线交易

在 Etherscan 上查找钱包地址，我们可以看到更多关于这笔交易的详细信息。需要注意的是，Uniswap 使用 WETH。然而，我们最初的钱包里是 ETH。因此，兑换 USDT 的流程首先需要先转账到 WETH。

Etherscan 上列出的交易详情

接下来，我们将尝试使用 USDT 回购相同数量的 ETH（约 0.001135）。这次，我们的目标池为蓝色高亮显示的池子。

顶级 ETH-USDT 池

请注意，此池子基于 Uniswap 协议 v2 版本，因此我们需要更新 Uniswap 类定义中的版本号 (= 2)。现在我们可以检查此交换需要多少 USDT。输入的 ETH 金额（以 wei 为单位）需要为整数。由于这是 v2 版本上唯一的池子，因此此处未使用费用参数。

USDT-ETH 掉期价格

与之前的互换相比，回购相同数量的 ETH 所需的 USDT 有所减少。这是意料之中的，因为该池中的 ETH 价格（以 USDT 计价）较低。然而，需要注意的是，由于这些互换是链上交易，因此也会消耗 Gas。Gas 价格会随着网络拥堵情况而变化。因此，任何潜在的收益都需要通过考虑交易成本的变化来谨慎抵消。

为了执行交换，我们应用与之前相同的函数。

执行 USDT-ETH 兑换示例

输出是交易哈希，可以在Etherscan上验证。“交易操作”部分提供了简要概述。

Etherscan 上的交易操作摘要

向下滚动可以找到有关交换路线的更多详细信息：

在 Etherscan 上交换路由详情

该交易也出现在Uniswap 的池摘要页面上。

Uniswap 上线交易

结论

在本文中，我们学习了如何结合使用 CoinGecko API 的链上 DEX 数据和 Python 来监控各种 DEX 池的交易活动。以 Uniswap 为例，我们演示了如何以编程方式在不同池中兑换代币，从而寻找潜在的套利机会。此外，我们还在 Uniswap 和 Etherscan 上验证了交易，以确保兑换路径符合我们的预期。


这里提供的脚本可以帮助交易者简化和自动化他们的交易策略，从钱包余额跟踪开始，扩展到在 DEX 上执行代币交换。

作者：GTokenTool一键发币平台

交流群：https://t.me/+Kz4u3xoDpFo3ZWY1

Solana工具推荐

solana市值机器  solana批量转账  solana添加流动性 Solana多对多转账  Solana租金回收 批量转账 市值机器人Uniswap添加流动性 STON.fi生成流动性 Solana燃烧流动性池 Solana创建流动性 PUMP发币 Raydium V2创建流动性 Raydium V3添加流动性 Meteora swap加池子 Orca创建流动性 生成流动性代币 uniswapv2添加流动性 uniswap v3添加流动性

同类推荐
网站分类
最近发表