Flashbots — Project Improvements: Stateful LP Fetcher, WETH Cycle Arbitrage Helper, Balance Updater, Local Gas Estimation, Relay Retry, V3 Multicall Decoder, Unified Submitter, Bundle Recorder
Flashbots — Project Improvements: Stateful LP Fetcher, WETH Cycle Arbitrage Helper, Balance Updater, Local Gas Estimation, Relay Retry, V3 Multicall Decoder, Unified Submitter, Bundle Recorder
More Features For the UniswapV2 Flashbots Framework
I regularly make modifications and test new features for my bots, and I wrote the first project improvement roll-up post with added features for the UniswapV2 Flashbots Framework a month after publishing the initial bot code.
If you’ve been using that for a while, you’ve likely seen a lot of MEV opportunities, submitted a lot of bundles, and seen a ton of console text flying around. I hope you’ve landed some arbs!
There are several new features I’ve added to make running this bot framework easier.
Stateful LP Fetcher
The first is an updated LP fetcher that will find the last-known pool at startup, and start from there instead of fetching the full list of LPs every time. Big time savings, especially on popular DEX like UniswapV2.
This uses JSON as the data storage format, and is set to use Sushiswap and UniswapV2 in this example.
BUGFIX 2022-11-09: added a missing .abi attribute to line LP_ABI = Contract(factory.allPairs(0))
ethereum_lp_fetcher_json.py
from brownie import network, Contract
import sys
import os
import json
import web3
BROWNIE_NETWORK = "mainnet-local"
os.environ["ETHERSCAN_TOKEN"] = "[CHANGE ME]"
try:
network.connect(BROWNIE_NETWORK)
except:
sys.exit("Could not connect!")
exchanges = [
{
"name": "SushiSwap",
"filename": "ethereum_sushiswap_lps.json",
"factory_address": "0xC0AEe478e3658e2610c5F7A4A2E1777cE9e4f2Ac",
},
{
"name": "Uniswap V2",
"filename": "ethereum_uniswapv2_lps.json",
"factory_address": "0x5C69bEe701ef814a2B6a3EDD4B1652CB9cc5aA6f",
},
]
w3 = web3.Web3(web3.WebsocketProvider())
for name, factory_address, filename in [
(
exchange["name"],
exchange["factory_address"],
exchange["filename"],
)
for exchange in exchanges
]:
print(f"DEX: {name}")
try:
with open(filename) as file:
lp_data = json.load(file)
except FileNotFoundError:
lp_data = []
if lp_data:
print(f"Found previous LP data: {len(lp_data)} pools")
previous_pool_count = len(lp_data)
try:
factory = Contract(factory_address)
except:
try:
factory = Contract.from_explorer(factory_address)
except:
factory = None
finally:
if factory is None:
sys.exit("FACTORY COULD NOT BE LOADED")
# Retrieve ABI for typical LP deployed by this factory
try:
LP_ABI = Contract(factory.allPairs(0)).abi
except:
try:
LP_ABI = Contract.from_explorer(factory.allPairs(0)).abi
except:
LP_ABI = None
finally:
if LP_ABI is None:
sys.exit("LP ABI COULD NOT BE LOADED")
# count the number of pairs tracked by the factory
pool_count = int(factory.allPairsLength())
if pool_count == previous_pool_count:
print("No new pools\n")
continue
print(f"Found {pool_count - len(lp_data)} new pools")
# retrieve pool addresses found from the factory
print("• Fetching LP addresses and token data")
for pool_id in range(
len(lp_data),
pool_count,
1,
):
lp_dict = {}
lp_address = factory.allPairs(pool_id)
w3_pool = w3.eth.contract(
address=lp_address,
abi=LP_ABI,
)
token0 = w3_pool.functions.token0().call()
token1 = w3_pool.functions.token1().call()
lp_dict["pool_id"] = pool_id
lp_dict["pool_address"] = lp_address
lp_dict["token0"] = token0
lp_dict["token1"] = token1
lp_data.append(lp_dict)
print("• Saving pool data to JSON\n")
with open(filename, "w") as file:
json.dump(lp_data, file)WETH Cycle Arbitrage Helper
We have spent a lot of time learning how to do flash borrow arbitrage, but many everyday arbs are on the order of a few dollars at a time, and these can be snagged only if you have an efficient method that minimizes gas use.
The most efficient arbitrage takes a starting balance of some commonly-held wrapped token and swaps across two pools with shared tokens. You see these all the time on the Flashbots Explorer. Swap WETH into pool A, swap the resulting token into pool B, keep the profit. Simple!
I have built an arb helper for just this case called LpSwapWithFuture, which you can find HERE or by pulling/cloning the degenbot code base.
It is a simple arb helper that supports overriding reserves for mempool TX, so if you’re already working with other helpers you’ll feel comfortable with this one.
It relies on your arb contract holding a balance of WETH, so if you have no money don’t bother with this one. You can chase most of the “peanut sized” cycle arbs with 1-2 WETH, so don’t get discouraged.
Cycle arb helpers can be built from the existing 2pool and triangle arb JSON paths you already have. It’s a bit hacky but works well until I can build a more sophisticated parser that really makes this robust.
Bot WETH Balance Updater
Since we are going to be chasing cycle arbs, which are limited by the WETH balance of our contract, we need a way to track our WETH balance and update the arb helpers whenever it increases (usually from a successful arb, but a direct transfer is valid, too).
This simple function runs on every new block, checking for an update, and when found, updates the cycle arb helpers with the new balance.
async def track_balance():
weth_balance = 0
last_block = newest_block
while True:
await asyncio.sleep(0)
# check for balance updates once per block
if last_block == newest_block:
continue
try:
balance = weth.balanceOf(arb_contract.address)
except Exception as e:
print(f"track_balance: {e}")
else:
if balance != weth_balance:
print()
print(f"Updated balance: {balance/(10**18):.3f} WETH")
print()
weth_balance = balance
for arb in degenbot_cycle_arbs.values():
arb.max_input = weth_balance
finally:
last_block = newest_blockLocal Gas Estimation
If you’re running a local Ethereum node (you should be), it is trivial to test potential arbitrage transactions for estimated gas consumption. We can test through the Flashbots relay as well, but this is much slower even on a relatively fast broadband connection. Much better to use our local node wherever possible, and to maintain a record of that gas use to avoid unnecessary re-tests.
This function allows us to simulate a basic cycle arb locally, and return the status of that simulation. If the simulation fails too many times, the arb can be dropped and blacklisted.
async def test_onchain_arb_gas(
arb_dict: dict,
arb_id=None,
):
global arb_simulations
# get a pointer to the arb helper
arb_helper = degenbot_cycle_arbs.get(arb_id)
if VERBOSE_TIMING:
start = time.monotonic()
print("starting test_onchain_arb")
tx_params = {
"from": bot_account.address,
"chainId": brownie.chain.id,
"nonce": bot_account.nonce,
}
swap_amount = arb_dict.get("swap_amount")
# generate payloads for these steps: transfer, swap*
# *multiple swap payloads may be needed
try:
# transfer the input token to the first swap pool
transfer_payload = [
# address
input_token.address,
# bytes calldata
w3.keccak(text="transfer(address,uint256)")[0:4]
+ eth_abi.encode(
[
"address",
"uint256",
],
[
arb_dict.get("swap_pool_addresses")[0],
swap_amount,
],
),
# msg.value
0,
]
except Exception as e:
print(f"transfer_payload: {e}")
print(arb_dict)
print(f"id: {arb_id}")
return
try:
swap_payloads = []
swap_steps = len(arb_dict.get("swap_pool_addresses"))
for i, swap_pool_address in enumerate(
arb_dict.get("swap_pool_addresses")
):
# use the next swap pool address as the destination unless
# this is the last swap
swap_destination = (
arb_contract.address
if i == swap_steps - 1
else arb_dict.get("swap_pool_addresses")[i + 1]
)
swap_payloads.append(
[
# address
swap_pool_address,
# bytes calldata
w3.keccak(
text="swap(uint256,uint256,address,bytes)"
)[0:4]
+ eth_abi.encode(
[
"uint256",
"uint256",
"address",
"bytes",
],
[
*arb_dict.get("swap_pool_amounts")[i],
swap_destination,
b"",
],
),
# msg.value
0,
]
)
except Exception as e:
print(f"swap_payload: {e}")
print(arb_dict)
print(f"id: {arb_id}")
return
success, gas_estimate = test_gas(
arb_helper,
[transfer_payload, *swap_payloads],
tx_params,
arb_id=arb_helper.id,
)
if not success:
return
else:
arb_helper.gas_estimate = gas_estimate
if VERBOSE_TIMING:
print(
f"test_onchain_arb completed in {time.monotonic() - start:0.4f}s"
)This function relies on there being an actual opportunity to test, so there is a corresponding function that will continually look for cycle arb helpers that have both (1) no previous gas estimate set and (2) some positive expected profit, even if trivially small.
This function works hand-in-hand with the gas estimator, and in fact calls it directly.
async def activate_arbs():
# pre-update all arbs, yielding to the event loop frequently to allow the websocket to receive new blocks and keep pool states up to date
for arb_helper in degenbot_cycle_arbs.values():
await asyncio.sleep(0)
arb_helper.update_reserves()
while True:
await asyncio.sleep(0)
if (
not status_new_blocks
or not status_sync_events
or not status_pools_updated
):
continue
# generator expression to identify all potential arbs that have not been simulated for gas
arbs_to_process = (
arb_helper
for arb_helper in degenbot_cycle_arbs.values()
if not arb_helper.gas_estimate
if arb_helper.best.get("profit_amount")
)
try:
# get the first match only
arb_helper = next(arbs_to_process)
except StopIteration:
continue
# yield to the event loop, then check if the arb is still valid
try:
await asyncio.sleep(0)
arb_helper.update_reserves()
except Exception as e:
print(f"estimate_arbs: {e}")
print(type(e))
else:
if not arb_helper.best.get("profit_amount"):
continue
if VERBOSE_ACTIVATION:
print(f"Gas test for arb: {arb_helper} ...")
await test_onchain_arb_gas(
arb_dict=arb_helper.best,
arb_id=arb_helper.id,
)Relay Retry
The frustrating this about the Flashbots Relay is sometimes it can’t be reached. If you’ve found a really hot arb but the Relay throws some trivial error (connection reset, for example) and you miss it, you’re gonna feel real bad.
I’ve made some improvements to the simulator and bundle submission code blocks that will attempt to retry the relay up to 5 times (adjustable) whenever an error is encountered. It’s not sophisticated enough to stop on failed simulations, but this catches the easy “false negatives” when the relay goes AWOL for a bit.
V3 Multicall Decoder
In the UniswapV3 Mempool Watcher lesson, we developed a simple transaction watcher that observed and decoded multicall transactions. As luck would have it, some of these multicall transactions are swaps through V2 pools that are being sent through the V3 front-end.
Other than the added difficulty of deciphering these multicall transactions, they can be handled as before and backrun to your heart’s content.
The V3 multicall section is a bit primitive (for now), and only processes the first V2 transaction it finds. Some users submit two or more V2 swaps via multicall, and it will not catch these until I make the handling more sophisticated.
Unified Submitter
I have also collapsed the relay execution into a single function that can assemble a bundle with an arb transaction and an optional backrun or frontrun from the mempool. This simplifies the logic and eliminates the need for separate functions to handle the various types.
Bundle Recorder
If you’re going to be submitting bundles, wouldn’t you like to see what they were?
I have written a bundle recorder function that maintains a JSON dict of all successfully submitted bundles. This is very useful if you want to inspect a bundle later, because there is no central database to query for the bundles you’ve already sent.
The JSON dict is keyed by bundleHash, and the information inside each entry includes the block it targeted, the calldata for every transaction in the bundle, the arb identifier, and the time it was submitted.
Bundles targeting a mempool TX will include both the mempool TX and the arb TX, and it will store multiple block targets, if you chose to do so.
What to Expect
This version operates a bit differently from the earlier version. It is built to handle a lot of simultaneous arbs, and it’s very “chatty” with the default settings. Feel free to adjust the various VERBOSE_ options to reduce the visual spam.
Initially, it will look like the bot isn’t doing much as it updates all of the pools. However once it does, it will start testing arbs for gas estimates, processing mempool transactions, and really chugging along. You’ll also see A LOT of these messages:
Source Code
Please report bugs to me directly or in the Discord. Happy searching!
BUGFIX 2022-11-09:
Commented out a status check in
watch_pending_transactionsthat caused it to skip processing transactions during pool updates.Removed a +1 in the send_bundle and record_bundle steps, which was incrementing the target blocks “off by one”
ethereum_flashbots_cycle.py