tech

MEV套利机器人开发记录

本文记录我使用纯 Python 编写多 DEX 套利机器人的过程。套利策略就像打地鼠游戏一样。

一键发币： SUI | SOL | BNB | ETH | BASE | ARB | OP | POLYGON | AVAX | FTM | OK

作为一名 MEV 搜索者就像成为一名哲学家。你的日子将由 80% 的思考、10% 的构建和剩下的 10% 的梦想 组成。过去两周对我来说就是这样。

我刚开始研究 MEV 一个半月时，意识到如果继续这样下去，我可以花一辈子学习 MEV 而从未真正构建出一个实际的机器人。这个主题是如此庞大，我越深入研究，就越深入，却发现没有尽头。

因此，我决定暂停我的学习，开始做我最初想要做的事情：构建一个真正的 MEV 机器人。

我发现套利策略与经典的“打地鼠”游戏非常相似，所以我决定以这款游戏的名字来命名我的机器人。

表面上看，套利策略非常容易理解，因为它归根结底就是低价买入，高价卖出。

这些机会会随机出现在市场的各个角落，人们不断监控这些事件，迅速行动以从价差中获利。因此，这本质上是一个由世界各地的参与者玩的游戏。

然而，很容易被表面的简单性所误导，因为其内部机制相当复杂，在执行策略之前需要有一个深思熟虑的计划。

今天，我会涵盖你开始 MEV 套利所需的一切知识。你会学到如何提前规划，并了解顶级搜索者正在做什么来改进他们的机器人。我用来向你解释这个项目的机器人可能不是最好的，但它用的是 Python，易于跟随。我在构建 MEV 机器人原型时更重视简洁性和可读性。

我开源了这个项目，以便像我一样从事个人项目的人都能从中受益。你可以查看GitHub 仓库链接（这是我的 DEX 套利策略的“examples”分支）。

该项目仍在积极开发中，所以事情会很快变化，但我会在博客上记录大部分发现和进展。并且我不会对这个特定分支进行更改，以便这篇文章中的示例在未来仍然可以正常工作。

1、为什么我要开源这个？

这个项目包含了 MEV 机器人的所有基本组件，包括：

循环路径查找器
异步事件流
Gas 估算器
在线交换模拟器
可以在多个 DEX 上交易的智能合约
使用 Flashbots 执行订单
使用 InfluxDB 监控价差
使用 Telegram 发送警报

（最后两个并不是必需的…… 😚”）*

然而，目前这个项目更像是为 MEV 搜索者提供一个基础模板，需要在这里和那里进行一些优化才能使其变得有利可图。

此外，Whack-A-Mole V1 当前仅支持单链上的 DEX 套利，而在三种不同类型的套利中：

单链、多 DEX 套利
多链、多 DEX 套利
CEX-DEX 套利

这是最容易实现的，也是利润最低的。

在文章末尾，我会谈谈如何优化这个机器人，以及这个系统如何扩展到多链套利和 CEX-DEX 套利。

注意：这个机器人实际上是为了让大多数人开始时使用的平均 MEV 机器人而进行了优化不足。

在合约层面，它故意使用 Solidity 而不使用 Yul 来演示如何在以后的文章中通过优化 gas 成本来提高机器人的盈利能力。
在 gas 成本层面，它只是简单地使用 Blocknative 的 gas 估算服务来设置 gas 成本。我们其实可以进一步分析这个领域的竞争对手并尝试理解他们的 gas 定价策略。
在模拟层面，它通过智能合约进行在线模拟，由于使用了像 Python 这样高级语言，增加了延迟。模拟引擎需要离线运行以加快速度，同时使用更低级的语言。
在执行层面，它只将其私人交易发送到 Flashbots，这是众多区块提议者之一。这不是理想的，因为 Flashbots 现在大约只有五分之一的机会将它们的捆绑/交易添加到新区块中。这是一个保证 20% 的成功率。不太好。
在资本效率层面，它还没有使用闪电贷，因此无法充分利用市场上的机会，并且因此落后于那些赚取更多利润并贿赂更多区块建设者的竞争对手。（我们会更深入探讨我们的交易/捆绑如何更有效地添加）

当上述问题得到妥善处理后，这个机器人将轻松扩展到夹击和其他策略，如 JIT 流动性提供等，因为 所有的 MEV 实质上共享相同的代码库。

1、为什么选择 MEV 套利？

当我们提取 MEV 时，可以选择多种策略。这些机会以 1. 套利、2. 前置交易、3. 后置交易、和/或 4. 夹击 的形式存在。

在这几种广为人知的 MEV Alpha 中，我选择了套利，原因如下：

去中心化协议现在变得更加 MEV 意识，并努力通过减少对用户产生负面影响的 MEV 机会来回馈用户。因此，前置交易和后置交易/夹击将变得越来越难以提取。

你可以去看看 UniswapX，作为这一趋势的指示器：

此外，随着越来越多的竞争者涌入争夺同样的 Alpha，这些机会变得越来越难以提取。下面就是当很多人盯着同一个交易提取 MEV 时会发生的情况。

有多个 MEV 搜索者都想使用那个单一交易，因为它能给他们带来最大的利润。然而，Jared（MEV 搜索者 #1）想用那个交易来构建他的三明治捆绑包，而 MEV 搜索者 #2 想用它来进行 Uniswap V3 JIT（即时）机器人。还有几个搜索者试图用其他形式的 MEV 策略来利用它。

当这种情况发生时，谁支付最多的 gas，并以哪种策略获得最多的利润，谁就能使用那个交易来构建自己的捆绑包。

总而言之，并不是所有人都能使用相同的交易。交易是稀缺的，更糟糕的是，随着协议变得更加 MEV 意识，这些机会正在慢慢消失。

然而，有一种 MEV 可以使社区更加健康，并且随着更多加密交易所的引入，其规模也会扩大。套利通常会让终端用户和 MEV 搜索者都受益。

根据 Eigenphi 的数据，套利似乎在 MEV 领域占据了最大的市场份额。

这是可以预期的，因为像夹击这样的 MEV 随着 UI/UX 改进来解决这些负面外部性将会逐渐消失，而越来越多的新交易所，无论是链上还是链下，都会出现并导致价格偏差。

价格差异是由流动性差异引起的，除非所有交易所拥有相同的流动性和相同数量的用户，否则这些交易所之间的价格永远不会完全同步。而这正是我们套利者所希望的。

2、套利机会是否依然存在？

让我介绍一下 Whacker。他是一名搜索者，他会谈论套利策略。以下是我认为的套利的样子。它就像打地鼠游戏。

那么，套利机会是否依然存在？

这是一个大问题。 这是任何对该领域感兴趣的人在开始深入挖掘之前应该问的第一个问题。我们应该始终了解市场规模，然后再进行任何严肃的业务。

当然，套利机会依然存在，因为加密市场充满了低效，仅仅通过查看 Eigenphi 的仪表板就可以发现：在过去的一周里，仅在以太坊上就通过套利提取了近 120 万美元的 MEV。

好消息是，这些机会无处不在：DEX 到 DEX，CEX 到 DEX，CEX 到 CEX。这几乎是无穷无尽的。

我还建议你阅读Frontier Research的这个研究。它将揭示在CEX-DEX套利和仅DEX套利之间，潜在的差距有多大，以及为什么会出现这种情况。

CEX-DEX套利收益更高的简要原因是它涉及更冒险的操作。单独进行DEX套利的风险并不大，因为你的套利可以在原子操作中完成——意味着它要么在一个交易中发生，要么不发生，所以你最多只会损失Gas费用和可能的情绪。

另一方面，在CEX-DEX套利中，我们需要担心的是确保交易按正确的顺序执行，并以正确的价格和数量成交。例如，如果在Uniswap上的一笔交易完成了，但在同一时间段内，Binance上的卖出价格（出价）发生了变化，我们可能会被剔除，只剩下Uniswap的库存。为了应对这种情况，我们需要在另一个交易所寻找次优的出价并发送一个市场卖单，或者如果这样更便宜的话，直接抛售Uniswap的库存。

我将在后续的文章中深入探讨执行CEX-DEX套利的困难，因此目前让我们回到之前讨论的内容。

所以我们应该问的问题不应该是否还有套利机会存在——因为它们确实存在——而是作为初学者的MEV搜索者能否在与已经进入市场的专业交易员竞争时提取相同的alpha值。

要回答这个问题，我们必须一起构建一个简单但可行的机器人，并开始与玩家竞争。除此之外没有其他方法可以证明这一点。

3、我们应该在哪里寻找？

让我们首先回顾一下当我们尝试运行套利策略时，应该在哪里寻找：

单一链，多个DEX，n跳交换
多个链上的多个DEX
多个CEX
多个链和多个CEX

我最终会逐一介绍所有这些选项，但今天我打算先从使用单一链环境并在多个DEX上交易开始。

3.1 单一链，多个DEX，n跳交换

这种情况将是：

我们在单一区块链网络上进行交易：以太坊
我们在多个DEX上进行交易：Uniswap V2, Uniswap V3, Sushiswap V3等
我们执行n跳交换：单跳交换、2跳交换、3跳交换等

让我展示一些非常有趣的东西。就在写这篇文章的时候，这刚刚发生。

以下是我在Uniswap V3和Sushiswap V3上收集的以太坊ETH/USDT数据。我已经创建了多个2跳交换路径，从USDT（输入代币）开始，最终输出为ETH（输出代币）。所以像：

USDT -> ETH,
USDT -> BTC -> ETH,
USDT -> USDC -> ETH

这样的路径都有效。

在过去3小时内，Uniswap和Sushiswap之间的ETH/USDT价格差异在扣除手续费后显示为0.22%和0.60%。（注意，这里不是Gas成本或滑点成本，只是手续费！这是重要的）

计算价差的公式如下：

Spread (%) = (Price 1 * (1 — fee) / Price 2) — 1 * 100

如果你看下面的图表，它是一个时间序列图表，显示Price 1相对于Price 2的溢价。我已经用红色标记了两个时刻，当Price 1比Price 2高出超过0.2%。

现在，看看几秒钟之后发生了什么：

价差消失了... 😯

你刚刚看到了其他套利者的行动。

但是为什么0.22%的价差没有像0.60%的价差那样快消失？你必须注意到上面的价差公式还没有考虑到Gas成本或滑点成本，所以0.22%的价差不足以覆盖执行套利的Gas成本，但0.60%的价差可以，因此后者的机会更具竞争力。

所以我们亲眼看到在单一链、多个DEX环境中存在套利机会。我们也看到有其他竞争对手也在做同样的事情。

这些机会往往会在波动的市场条件下出现。我查看了Binance过去三个小时的ETH/USDT价格。

你可以看到红框中的价格走势，立刻就会明白为什么那些套利机会出现了。

再看另一张图：

这是我过去12小时一直在监测的价差。在这段时间内，有超过0.4%的价格峰值和多次≥0.2%的价格峰值。

🛑 请注意，价差计算不包括Gas成本。 每个搜索者使用的智能合约集不同，Gas成本也不同，因此每个搜索者看到的边缘也会有所不同。这是非常重要的，所以我反复强调这一点。

我们将在下一节中查看在考虑Gas成本后机会是否仍然存在。简短的答案是不存在，因为需要考虑Gas成本和滑点成本。而且滑点成本惊人地高，很快你就会发现。但仍然有一些方法可以绕过这个限制。

4、Whack-A-Mole的工作原理

看到Whack-A-Mole的实际操作将是理解它可以做什么和正在尝试做什么的最可靠方式。

MEV机器人有许多移动部件，代码的正确组织至关重要。通过一起完成以下任务，我希望向你展示MEV机器人是如何构建的。这基本上适用于所有MEV，因此它将为目前在MEV领域摸索的人提供指导。

4.1 任务

我们将尝试为ETH/USDT生成n跳交换路径，其中你希望使用USDT购买ETH。我们将比较多个ETH/USDT路径的价格 （使用USDT购买ETH的路径）与目标价差，例如+0.4%。

当价格差高于我们的目标价差时，我们将使用OnlineSimulator检查模拟交换是否真的能产生利润，同时考虑Gas成本和滑点成本。

此外，我们将计算实现目标利润金额所需的资本量。一旦找到有利可图的套利循环，我们将通过Flashbots发送一个包含单个私人交易的捆绑包。

最后，我们将努力了解如何改进我们的系统以变得更有利可图。这需要对Flashbots拍卖系统的工作原理有深刻的理解，并计算出我们应该优化我们的智能合约以与其他玩家竞争的程度。

让我们立即开始吧 😎

4.2 项目设置

让我们从GitHub克隆Whack-A-Mole：

git clone https://github.com/solidquant/whack-a-mole.git  
git checkout examples/strategy/dex_arb_base

确保切换到分支examples/strategy/dex_arb_base。

我在这个项目中使用的是Python 3.10，但它应该适用于所有3.x版本。接下来，你应该为项目创建一个虚拟环境，然后安装依赖项。假设你已经这样做，请运行：

pip install -r requirements.txt

可能会出现web3和websockets之间的冲突，任何遇到错误的人都可以简单地从requirements.txt文件中删除“websockets”，然后手动安装“websockets”库并使用最新版本11.0.3，策略应该可以正常运行。

要检查项目，请转到main.py：

import asyncio  
  
from strategies.dex_arb_base import main  
  
  
if __name__ == '__main__':  
    asyncio.run(main())

这里没有什么复杂的，因为我希望将来为其他策略添加更多模板，所以大部分逻辑将在我们的strategies文件夹中。

4.3 模块 #1: 数据

我不会深入细节，但系统会将策略中使用的代币和池映射为整数。因此，来自addresses/ethereum.py的信息如下：

EXCHANGE = 'ethereum'

TOKENS = {
    'ETH': ['0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2', 18],
    'USDT': ['0xdAC17F958D2ee523a2206206994597C13D831ec7', 6],
    'USDC': ['0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48', 6],
    'BTC': ['0x2260FAC5E5542a773Aa44fBCfeDf7C193bc2C599', 8],
    'DAI': ['0x6B175474E89094C44Da98b954EedeAC495271d0F', 18],
    'PEPE': ['0x6982508145454Ce325dDbE47a25d4ec3d2311933', 18],
}

columns = ['chain', 'exchange', 'version', 'name', 'address', 'fee', 'token0', 'token1']

POOLS = [
    ['uniswap', 3, 'ETH/USDT', '0x11b815efB8f581194ae79006d24E0d814B7697F6', 500, 'ETH', 'USDT'],
    ['uniswap', 3, 'USDC/USDT', '0x3416cF6C708Da44DB2624D63ea0AAef7113527C6', 100, 'USDC', 'USDT'],
    ['uniswap', 3, 'BTC/ETH', '0x4585FE77225b41b697C938B018E2Ac67Ac5a20c0', 500, 'BTC', 'ETH'],
    ['uniswap', 3, 'DAI/USDC', '0x5777d92f208679DB4b9778590Fa3CAB3aC9e2168', 100, 'DAI', 'USDC'],
    ['uniswap', 3, 'PEPE/ETH', '0x11950d141EcB863F01007AdD7D1A342041227b58', 3000, 'PEPE', 'ETH'],

    ['uniswap', 2, 'ETH/USDT', '0x0d4a11d5EEaaC28EC3F61d100daF4d40471f1852', 3000, 'ETH', 'USDT'],
    ['uniswap', 2, 'USDC/ETH', '0xB4e16d0168e52d35CaCD2c6185b44281Ec28C9Dc', 3000, 'USDC', 'ETH'],
    ['uniswap', 2, 'DAI/USDC', '0xAE461cA67B15dc8dc81CE7615e0320dA1A9aB8D5', 3000, 'DAI', 'USDC'],
    ['uniswap', 2, 'PEPE/ETH', '0xA43fe16908251ee70EF74718545e4FE6C5cCEc9f', 3000, 'PEPE', 'ETH'],

    ['sushiswap', 3, 'ETH/USDT', '0x72c2178E082feDB13246877B5aA42ebcE1b72218', 500, 'ETH', 'USDT'],
    ['sushiswap', 3, 'USDC/ETH', '0x35644Fb61aFBc458bf92B15AdD6ABc1996Be5014', 500, 'USDC', 'ETH'],
    ['sushiswap', 3, 'BTC/ETH', '0x801CCFae9d2C77893B545E8D0E4637C055CD26cB', 500, 'BTC', 'ETH'],
    ['sushiswap', 3, 'USDC/USDT', '0xfA6e8E97ecECDC36302eCA534f63439b1E79487B', 100, 'USDC', 'USDT'],
    ['sushiswap', 3, 'DAI/USDC', '0x31ac258B911Af9a0d2669ebDFC4e39D92e96b772', 100, 'DAI', 'USDC'],

    ['sushiswap', 2, 'BTC/ETH', '0xCEfF51756c56CeFFCA006cD410B03FFC46dd3a58', 3000, 'BTC', 'ETH'],
    ['sushiswap', 2, 'ETH/USDT', '0x06da0fd433C1A5d7a4faa01111c044910A184553', 3000, 'ETH', 'USDT'],
    ['sushiswap', 2, 'USDC/ETH', '0x397FF1542f962076d0BFE58eA045FfA2d347ACa0', 3000, 'USDC', 'ETH'],
    ['sushiswap', 2, 'DAI/ETH', '0xC3D03e4F041Fd4cD388c549Ee2A29a9E5075882f', 3000, 'DAI', 'ETH'],
]

POOLS = [dict(zip(columns, [EXCHANGE] + pool)) for pool in POOLS]

这可以映射为：

from data.dex import DEX

from configs import RPC_ENDPOINTS, TOKENS, POOLS


dex = DEX(rpc_endpoints=RPC_ENDPOINTS,
          tokens=TOKENS,
          pools=POOLS,
          trading_symbols=['ETH/USDT'],
          max_swap_number=2)

print('Chain ID: ', dex.chain_to_id)
print('Exchange ID: ', dex.exchange_to_id)
print('Token ID: ', dex.token_to_id)

"""
Output:

Chain ID:  {'ethereum': 0}
Exchange ID:  {'sushiswap': 0, 'uniswap': 1}
Token ID:  {'BTC': 0, 'DAI': 1, 'ETH': 2, 'PEPE': 3, 'USDC': 4, 'USDT': 5}
"""

所有代码可以从这里获取。

使用在 DEX 类实例中创建的 ID 映射器，它接下来将创建一个 6 维 Numpy 数组，其定义如下（在 data/dex.py 中）：

"""
storage_array
: 6-dimensional array that stores storage values from pool contracts
"""
self.storage_array = np.zeros((
    len(self.chains_list),      # chains
    len(self.exchanges_list),   # exchanges
    len(tokens_list),           # token in
    len(tokens_list),           # token out
    2,                          # uniswap variant version: 2, 3
    8                           # decimals0, decimals1, reserve0, reserve1, sqrtPriceX96,
                                # fee, token0_is_input, pool_index
))

该数组将区块链上的存储值存储到 8 个字段中，分别为：1. decimals0、2. decimals1、3. reserve0、4. reserve1、5. sqrtPriceX96、6. fee、7. token0_is_input 和 8. pool_index。

这非常方便，因为例如，如果我们想获取以太坊上 Uniswap V3 ETH-USDT 池的存储值，我们可以使用如下数据结构轻松检索该数据：

idx_1 = dex.get_index(chain='ethereum',
                      exchange='uniswap',
                      token0='ETH',
                      token1='USDT',
                      version=3)

idx_2 = dex.get_index(chain='ethereum',
                      exchange='uniswap',
                      token0='USDT',
                      token1='ETH',
                      version=3)

idx_1_values = dex.storage_array[idx_1]
idx_2_values = dex.storage_array[idx_2]

print(idx_1, idx_1_values)
print(idx_2, idx_2_values)

"""
Output:

(0, 1, 2, 5, 1) [1.80000000e+01 6.00000000e+00 0.00000000e+00 0.00000000e+00
 3.46077663e+24 5.00000000e-04 1.00000000e+00 0.00000000e+00]

(0, 1, 5, 2, 1) [1.80000000e+01 6.00000000e+00 0.00000000e+00 0.00000000e+00
 3.46077663e+24 5.00000000e-04 0.00000000e+00 0.00000000e+00]
"""

使用 ETH 或 USDT 都应该返回相同的池数据，事实也确实如此，因为它在 storage_array 的两个空间中保存了相同的数据。此外，为了简单起见，我为每个池设置了一个费用级别。我只需选择流动性/交易量最高的池即可。

我们希望根据实时事件更新存储数据，因此我们使用 DexStream 类来实现这一点。

打开 data/dex_streams.py，我们可以看到名为 start_streams 的函数：

def start_streams(self):
    streams = []

    for chain in self.dex.chains_list:
        block_stream = reconnecting_websocket_loop(
            partial(self.stream_new_blocks, chain),
            tag=f'{chain.upper()}_Blocks'
        )
        streams.append(block_stream)

    for chain in self.dex.chains_list:
        v2_stream = reconnecting_websocket_loop(
            partial(self.stream_uniswap_v2_events, chain),
            tag=f'{chain.upper()}_V2'
        )
        v3_stream = reconnecting_websocket_loop(
            partial(self.stream_uniswap_v3_events, chain),
            tag=f'{chain.upper()}_V3'
        )
        streams.extend([asyncio.ensure_future(f) for f in [v2_stream, v3_stream]])

    loop = asyncio.get_event_loop()
    loop.run_until_complete(asyncio.wait(streams))

Whack-A-Mole 支持多链，因此我们循环遍历所有链，并启动一个异步 websocket 流来检索以下数据：

新标头、
Uniswap V2 同步事件、
Uniswap V3 交换事件。

运行此操作非常简单：

import asyncio
import aioprocessing
from multiprocessing import Process

from data.dex import DEX
from data.dex_streams import DexStream

from configs import (
    RPC_ENDPOINTS,
    WS_ENDPOINTS,
    TOKENS,
    POOLS,
)

# Settings
chain = 'ethereum'
rpc_endpoints = {chain: RPC_ENDPOINTS[chain]}
ws_endpoints = {chain: WS_ENDPOINTS[chain]}
tokens = {chain: TOKENS[chain]}
pools = [pool for pool in POOLS if pool['chain'] == chain]
trading_symbols = ['ETH/USDT']


def dex_stream_process(publisher: aioprocessing.AioQueue):
    dex = DEX(rpc_endpoints=rpc_endpoints,
              tokens=tokens,
              pools=pools,
              trading_symbols=trading_symbols,
              max_swap_number=2)

    dex_stream = DexStream(dex=dex,
                           ws_endpoints=ws_endpoints,
                           publisher=publisher)
    dex_stream.start_streams()


async def strategy(subscriber: aioprocessing.AioQueue):
    while True:
        data = await subscriber.coro_get()
        print(data)


if __name__ == '__main__':
    # Starting DexStream
    queue = aioprocessing.AioQueue()

    # Start a process of DEX streams
    p = Process(target=dex_stream_process, args=(queue,))
    p.start()

    asyncio.run(strategy(queue))

运行结果如下：

注意：要运行此程序，您需要拥有 Blocknative 的 API 令牌！这对于 Gas 估算至关重要。Blocknative 有一个如下所示的仪表板，您可以通过 API 调用访问此处的数据：

4.4 模块 #2：路径查找器

要理解通过 DEX 数据流检索到的实时数据，如下所示：

{
  'source': 'dex',
  'type': 'event',
  'block': 17726053,
  'path': [[[0, 1, 5, 2, 1], [0, 0, 0, 0, 0]], [[0, 1, 5, 2, 0], [0, 0, 0, 0, 0]], [[0, 0, 5, 2, 1], [0, 0, 0, 0, 0]], [[0, 0, 5, 2, 0], [0, 0, 0, 0, 0]], [[0, 1, 5, 4, 1], [0, 1, 4, 2, 0]], [[0, 1, 5, 4, 1], [0, 0, 4, 2, 1]], [[0, 1, 5, 4, 1], [0, 0, 4, 2, 0]], [[0, 0, 5, 4, 1], [0, 1, 4, 2, 0]], [[0, 0, 5, 4, 1], [0, 0, 4, 2, 1]], [[0, 0, 5, 4, 1], [0, 0, 4, 2, 0]]],
  'pool_indexes': [[0], [5], [9], [15], [1, 6], [1, 10], [1, 16], [12, 6], [12, 10], [12, 16]],
  'symbol': 'ETH/USDT',
  'tag': ['ethereum-0', 'ethereum-1', 'ethereum-2', 'ethereum-3', 'ethereum-4', 'ethereum-5', 'ethereum-6', 'ethereum-7', 'ethereum-8', 'ethereum-9'],
  'price': [1908.0519148221722, 1910.8749837806993, 1911.117066215452, 1911.0445021506505, 1908.5838300538105, 1910.425470198242, 1911.5290103508617, 1908.4971959551506, 1910.3387525041794, 1911.4422425651003],
  'fee': [0.0004999999999999449, 0.0030000000000000027, 0.0004999999999999449, 0.0030000000000000027, 0.0030997000000000385, 0.0005999499999999047, 0.0030997000000000385, 0.0030997000000000385, 0.0005999499999999047, 0.0030997000000000385]
}

我必须向你展示一下 DEX 类中的路径查找器的功能。这可以通过多种方式实现，所以没有绝对的正确或错误，但我先来演示一下我是如何生成路径的。

如果你仔细看看“path”中的第一个路径：

[[0, 1, 5, 2, 1], [0, 0, 0, 0, 0]]

你会发现它是一个包含两个 List 元素的 List：List[List[int]]。这是因为我指定了要生成最多 2 跳路径。

我们来看看第一个元素：[0, 1, 5, 2, 1]。第二个元素全为 0。这是一个 1 跳路径，以 Token #5 作为 token_in，Token #2 作为 token_out。也就是 USDT 和 ETH。

正如你所见，这是我们之前看到的 DEX.storage_array 的索引。现在查看“pool_indexes”、“tag”、“price”和“fee”中的数据字段。

“pool_indexes”的第一个元素是 [0]。这意味着第一条路径使用来自第 0 个索引的池数据，即：

['uniswap', 3, 'ETH/USDT', '0x11b815efB8f581194ae79006d24E0d814B7697F6', 500, 'ETH', 'USDT']

标签、价格和费用相同。每个字段的第一个元素将指向匹配的路径。因此，对于我们查看的第一条路径，它是：

报价为：1908.0519148221722 USDT/ETH
标签为：“ethereum-0”，表示以太坊中的第 0 条路径（这对于核心逻辑来说并非必要，但出于调试目的而添加）
费用为：0.0004999999999999449（即 0.05%，但由于它表示为浮点数，因此不准确）

如何找到用于套利的循环路径：

利用上面的路径，我们现在可以生成用于套利策略的循环路径。

让我们从上面随机选取两条路径：

路径 #1：[[0, 1, 5, 2, 1], [0, 0, 0, 0, 0]]
路径 #2：[[0, 1, 5, 4, 1], [0, 1, 4, 2, 0]]

记住，这里的索引代表：

(chain, exchange, token_in, token_out, version)

我们只看一下 token_in 和 token_out。创建 DEX 类实例后，我们尝试打印出 dex.chain_to_id、dex.exchange_to_id 和 dex.token_to_id 字典，如下所示：

Chain ID:  {'ethereum': 0}
Exchange ID:  {'sushiswap': 0, 'uniswap': 1}
Token ID:  {'BTC': 0, 'DAI': 1, 'ETH': 2, 'PEPE': 3, 'USDC': 4, 'USDT': 5}

因此，第一条路径是一条单跳路径：

Uniswap V3 USDT → 以太坊上的 ETH

第二条路径是一条双跳路径：

(Uniswap V3 USDT → USDC) → (Uniswap V2 USDC → ETH)

比较这两条路径的价格，相当于观察一条三角套利路径。

买入：Uniswap V3 USDT → ETH
卖出：(Uniswap V2 ETH → USDC) → (Uniswap V3 USDC → USDT)

4.5 优势计算与策略评估

数据部分讲完了，我们终于可以进入我们系统中更有趣的部分了。思考部分……😚

就像我之前说的，MEV 的核心在于思考，尤其是思考你的优势和他人的竞争优势。

让我们来思考一下套利策略中的优势（edge）是什么，尤其是在 MEV 套利中。为此，我们将运用一些简单的数学知识——主要是基础代数。

什么是优势？

优势就是利润减去成本。这个大家都懂，对吧？所以，我们将深入探讨每个因素。

先从利润说起：

抱歉，字写得很糟糕……

利润的定义如上所述。它是：

卖出价格 * 入金金额 * (1 - 卖出滑点)
减去
买入价格 * 入金金额 * (1 - 买入滑点)

可以简化为：

这是我们输入的代币数量减去滑点成本，再乘以价差后的结果。

成本定义如下：

成本为 (gas_price * gas_used) + (buying_fee + sell_fee)。其中 gas 价格为以下值：

min(max_fee_per_gas, base_fee + max_priority_fee_per_gas)

更多 gas 成本信息，请点击此处。

如果我们将利润和成本公式加在一起，我们得到：

这非常重要，因为它将是我们评估策略的指标。这基本上意味着：

为了在套利游戏中获利，我们的 amount_in 应该大于成本/利润比。

现在，让我们稍微理解一下。

这个简单的公式实际上告诉我们很多关于 MEV 游戏的信息。它还指出了我们应该如何优化系统以获得更好的性能。

Gas 价格、Gas 使用量和手续费水平直接影响我们的成本，因此支付更多的 Gas 费用、手续费或使用更多的 Gas 会导致我们成本增加。
价差越大，我们的利润就越大，而且随着公式中分母的增大，我们需要的 amount_in 资金就会减少。
你的起始资金越多，就越容易超越成本/利润比。
然而，更多的 amount_in 意味着更高的滑点成本，所以这个游戏也不仅仅是关于资本的。

现在给你一些关于如何优化这些方面的建议：

Gas 价格：分析你的竞争对手，看看他们是否在争夺同样的优势，只需多支付一点 Gas 就能击败他们。
Gas 使用：使用汇编/Yul 或像 Huff 这样的低级语言来优化你的智能合约，你将能够降低 Gas 消耗。
价差：监控更多的 DEX 和 CEX，以获得比竞争对手更优的买入/卖出价格。
滑点：通过使用流动性强的交易所/资金池来降低滑点成本。总锁定价值/流动性越高，你的交易对市场的影响就越小。

4.6 模块 #3：模拟器

我们可以通过运行模拟器来测试我们的优势。对于当前版本，我大量借鉴了之前的文章。我稍微调整了一下 Solidity 代码，以适应打地鼠游戏的需求。Solidity 代码可以在这里查看。

使用此合约非常简单。首先，我们确保设置环境变量 ETHEREUM_SIMULATOR_ADDRESS 的值。

ETHEREUM_SIMULATOR_ADDRESS=<CONTRACT_ADDRESS>

在运行任何模拟之前，让我快速运行 dex_arb_base.py 程序来捕获我能找到的所有边缘。运行代码后，我看到：

[2023-07-19 21:28:49.645293] Update took: 0.0191 secs. SUS3ETHUSDT/UNI3ETHUSDT: 0.15%

Uniswap V3 ETH/USDT 和 Sushiswap V3 ETH/USDT 之间的价差为 0.15%。我会快速运行一个模拟，看看我们能否盈利。让我们来看看模拟所需的信息：

{
  'key': 'SUS3ETHUSDT/UNI3ETHUSDT',
  'max_buy_sell_price': [1906.0595172354178, 1911.117066215452],
  'block': 17727252,
  'cancel_at': 17727253,
  'buy_path': [[0, 1, 5, 2, 1], [0, 0, 0, 0, 0]],
  'sell_path': [[0, 0, 5, 2, 1], [0, 0, 0, 0, 0]],
  'buy_pools': [0],
  'sell_pools': [9],
  'estimated_gas_used': 200000,
  'order_processing': False
}

我们看到买入价为 1906，卖出价为 1911，扣除掉期费后，价差约为 0.15%。

前往这里。

这里有一个如何使用 OnlineSimulator 的示例代码：

if __name__ == '__main__':
    import os
    from dotenv import load_dotenv

    from configs import RPC_ENDPOINTS
    from addresses.ethereum import TOKENS, POOLS, SIMULATION_HANDLERS

    load_dotenv(override=True)

    ETHEREUM_SIMULATOR_ADDRESS = os.getenv('ETHEREUM_SIMULATOR_ADDRESS')

    chain = 'ethereum'

    rpc_endpoints = {chain: RPC_ENDPOINTS[chain]}
    tokens = {chain: TOKENS}
    pools = [pool for pool in POOLS if pool['chain'] == chain]
    contracts = {chain: ETHEREUM_SIMULATOR_ADDRESS}
    handlers = {chain: SIMULATION_HANDLERS}

    sim = OnlineSimulator(rpc_endpoints, tokens, pools, contracts, handlers)

    """
    ETH/USDT
    - Buy: USDT -> ETH
    - Sell: ETH -> USDT

    Buy, sell should work like CEXs
    """
    for i in range(100, 1000, 100):
        amount_in = i * 10 ** 6
        print('==========')
        print('Amount in: ', amount_in)

        buy_path = [[0, 1, 5, 2, 1], [0, 0, 0, 0, 0]]
        sell_path = [[0, 0, 5, 2, 1], [0, 0, 0, 0, 0]]

        buy_pools = [0]
        sell_pools = [9]

        params = sim.make_params(amount_in, buy_path, sell_path, buy_pools, sell_pools)

        for param in params:
            print(param)
        """
        SUS3ETHUSDT/UNI3ETHUSDT
        
        - Buy: UNI3ETHUSDT
        - Sell: SUS3ETHUSDT
        
        Output:
        
        {'protocol': 1, 'handler': '0x61fFE014bA17989E743c5F6cB21bF9697530B21e', 'tokenIn': '0xdAC17F958D2ee523a2206206994597C13D831ec7', 'tokenOut': '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2', 'fee': 500, 'amount': 100000000}
        {'protocol': 1, 'handler': '0x64e8802FE490fa7cc61d3463958199161Bb608A7', 'tokenIn': '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2', 'tokenOut': '0xdAC17F958D2ee523a2206206994597C13D831ec7', 'fee': 500, 'amount': 0}
        """

        simulated_amount_out = sim.simulate(chain, params)
        print(f'Simulated amount out: {simulated_amount_out / 10 ** 6} USDT')

        simulated_profit_in_usdt = (simulated_amount_out - amount_in) / 10 ** 6
        print(f'Simulated profit: {simulated_profit_in_usdt} USDT')

我运行一个循环，使用 100、200、300 一直到 900 的金额，看看交换这些金额是否会给我们带来利润。上面的输出是：

==========
Amount in:  100000000
Simulated amount out: 100.149096 USDT
Simulated profit: 0.149096 USDT
==========
Amount in:  200000000
Simulated amount out: 200.258044 USDT
Simulated profit: 0.258044 USDT
==========
Amount in:  300000000
Simulated amount out: 300.326868 USDT
Simulated profit: 0.326868 USDT
==========
Amount in:  400000000
Simulated amount out: 400.355592 USDT
Simulated profit: 0.355592 USDT
==========
Amount in:  500000000
Simulated amount out: 500.34424 USDT
Simulated profit: 0.34424 USDT
==========
Amount in:  600000000
Simulated amount out: 600.292837 USDT
Simulated profit: 0.292837 USDT
==========
Amount in:  700000000
Simulated amount out: 700.201405 USDT
Simulated profit: 0.201405 USDT
==========
Amount in:  800000000
Simulated amount out: 800.06997 USDT
Simulated profit: 0.06997 USDT
==========
Amount in:  900000000
Simulated amount out: 899.898555 USDT
Simulated profit: -0.101445 USDT

仔细查看输出。你会发现，随着我们投入更多 USDT 代币，利润金额一开始会增加。然而，在 400 USDT 之后，我们的利润开始缩水。最终，在我们兑换 900 USDT 后，我们发现我们出现了第一次亏损。

遗憾的是，这些资金池的流动性并不强，即使几百 USDT 也能对市场造成重大影响。

此外，在两个连续的资金池中进行兑换，我的合约估计要花费 20 万个 Gas。我打开 Blocknative，快速查看了 base_fee 和 max_priority_fee 的值。它接近 20。如果我用 40 来抬高我的 Gas 成本：

40 gwei * 200,000 gas = 0.008 以太币（=接近 15 美元）

这远不及你在两个池子之间进行套利所能获得的最大利润，最高利润是 0.35 USDT。

使用上一节中的公式，我们可以计算出在给定 Gas 价格、Gas 使用量和价差的情况下，盈利所需的最小 amount_in 金额。为了估算，我简单地使用：

最小输入量 > (gas_price * gas_used) / 价格差

成本：40 gwei * 200,000 gas * 1911.117 = 15.288936 USDT

利润：每输入 1 ETH，1911.117 - 1906.06 = 5.06 USDT

因此，在这种设置下，我们需要输入 3.02 ETH（= 5756 USDT）才能实现盈亏平衡。

如果我们尝试模拟兑换 5756 USDT 的结果，输出如下：

Simulated amount out: 5687.705938 USDT
Simulated profit: -68.294062 USDT

粗略计算显示，滑点接近 1%。

由于网络成本，使用在线模拟器实际上非常低效。通过 Infura、Alchemy 等节点服务进行单次调用大约需要 0.3 到 1.0 秒，而使用本地机器上的私有节点则要快近 10 到 100 倍。

然而，即使这种方法也不是最佳方案。模拟应该离线进行，如果使用 Numpy，所有计算将在 0.0001 秒内完成——如果使用 C、C++ 或 Rust，时间可能更短。

我目前正在将 Uniswap V2/V3 离线模拟器添加到我的代码库中，完成后会与大家分享结果。

4.7 模块 #4：执行

接下来是打地鼠的最后一块基石——执行组件。

要理解区块链世界中订单执行的工作原理，你需要对 Flashbots 有深入的了解。下面我分享了另一张 Flashbots 工作原理的草图：

目前，您可以通过两种方式在区块链上提交交易：通过公共内存池或由 Flashbots、Blocknative、Beaverbuild、Rsync-builder 等区块构建器运营的私有中继。

构建器状态可从这个链接查看。

构建者会选择通过其私有中继器发送的交易和捆绑包，并尝试使其捆绑包尽可能盈利。具体方法是扫描所有收到的交易/捆绑包，并在其机器上运行模拟后，挑选出贿赂金额最高的交易/捆绑包。

这些交易或捆绑包的具体选择方式可从 Flashbots 的发布日志中查看。他们分享了一个简单的公式来对交易进行排序：

这个公式乍一看相当吓人，但核心概念却相当简单。

只需关注 delta coinbase 部分和 gT（交易消耗的 gas）总和的分母即可。这基本上意味着 coinbase（贿赂）金额越大，该交易被选中的可能性就越大。此外，消耗的 gas 总量越小，意味着构建者更喜欢消耗较少 gas 的交易。这是因为以太坊对一个区块内交易消耗的 gas 总量有限制。

如果一个区块内使用了更多 Gas，下一个区块的基础 Gas 费用就会增加，反之亦然。

此外，我们还应该注意，即使我们向区块创建者支付了巨额的优先费用，我们最终也可能无法被添加到新的区块中。这是因为，正如一个搜索者可以赢得一笔交易一样，每个区块也可能有一个创建者获胜。这在 Etherscan 上很容易看到：

我记录了过去几个添加到以太坊的区块，正如你所见，许多区块建造者都在争夺区块的加入。在最近提交的14个区块中，Flashbots只赢了3次，而且从未连续获胜。成功率接近20%，与下表中的数字相近：

为确保您的打包文件能够添加到区块链，您需要注意以下两点：

设置足够高的 Gas 价格水平（基础费用、最高费用、最高优先级费用）；
将打包文件发送给多个构建器，最好发送给所有排名前五的构建器，以使成功率达到 90%。

Whack-A-Mole 目前仅向 Flashbots 发送打包文件，但将在下一个版本更新中快速添加对多构建器提交的支持。

要在 Whack-A-Mole 中使用打包文件提交功能，我们使用 DexOrder 类：

async def send_bundle(self,
                      w3: Web3,
                      bundle: List[Dict[str, Any]],
                      retry: int,
                      block_number: int = None) -> list:

    flashbots: Flashbots = w3.flashbots

    left_retries = retry

    if not block_number:
        block_number = w3.eth.block_number

    receipts = []

    while left_retries > 0:
        print(f'Sending bundles at: #{block_number}')
        try:
            flashbots.simulate(bundle, block_number)
        except Exception as e:
            print('Simulation error', e)
            break

        replacement_uuid = str(uuid4())
        response: FlashbotsBundleResponse = flashbots.send_bundle(
            bundle,
            target_block_number=block_number + 1,
            opts={'replacementUuid': replacement_uuid},
        )

        while w3.eth.block_number < response.target_block_number:
            await asyncio.sleep(1)

        try:
            receipts = list(
                map(lambda tx: w3.eth.get_transaction_receipt(tx['hash']), response.bundle)
            )
            print(f'\nBundle was mined in block {receipts[0].blockNumber}\a')
            break
        except TransactionNotFound:
            print(f'Bundle not found in block {block_number + 1}')
            flashbots.cancel_bundles(replacement_uuid)
            left_retries -= 1
            block_number += 1

    return receipts

当前项目状态不需要发送多笔交易，因此 dex_order.py 中提供的示例将单个签名交易作为捆绑包发送给 Flashbots。

执行多笔交易的智能合约是通过 WhackAMoleBotV1 合约实现的。

这标志着本节的结束。我不确定是否还有人在读这篇文章，但对于那些想知道构建这样一个机器人需要多长时间的人来说，我花了整整两周的时间才完成我的原型——也就是现在的版本。

此外，在此之前，我花了一个多月的时间研究MEV。所以，人们上手所需的总时间至少需要两个月左右。这还只是模板构建部分。进行优化以达到盈利状态将是一个新的水平。

5、未来的优化

这是我们今天的最后一部分。在最后一部分中，我想简要讨论一下Whack-A-Mole在未来版本更新中的改进。

目前仅能处理单链DEX套利的Whack-A-Mole现在将进入未知领域，寻找来自以太坊生态系统之外的更多“鼹鼠”来打击。

我将很快添加对顶级交易所（如Binance、Bybit和OKX）的CEX支持。还将有CEX……基于多订单簿系统的聚合器，它通过将多个交易所的订单簿数据组合成一个单一的订单簿结构。

通过多订单簿系统，我们可以获取多个CEX（中心化交易所）的最佳买入价和卖出价，并比较CEX与DEX（去中心化交易所）之间的价差，从而同时在多个订单场所进行交易。

如果这一切听起来有些陌生，请再稍作停留，看看下一篇是如何完成CEX-DEX套利的！😃 并不会太难。

在此基础上，我会逐步优化本文中介绍的Whack-A-Mole的各个组件，提高我们系统的竞争力。我将从我认为最重要的部分开始：

提高延迟性：

构建离线交换模拟器（Uniswap V2、V3变体）

提高盈利能力+增加区块添加概率：

更好地优化交换金额
使用闪电贷来提高资本效率
向包括Flashbots在内的多个区块提议者发送私人交易或捆绑交易
使用汇编语言实现更节省Gas的WhackAMoleBot合约
分析竞争对手的Gas策略

原文连接：How I built my first DEX arbitrage bot: Introducing Whack-A-Mole

DefiPlot翻译整理，转载请标明出处

免责声明：本站资源仅用于学习目的，也不应被视为投资建议，读者在采取任何行动之前应自行研究并对自己的决定承担全部责任。