Automated Trade Execution on Uniswap v3 in Python — We buy and sell a Token again …

10 min readAug 27, 2023

Welcome back ladies and gents.

Before we can do the cool stuff like building automated trading strategies for our bots, we need to cover some more basics like the trade execution.

In the last tutorial, we learned, how to buy and sell a token on Uniswap v2. Today we’re going to do the same, but on Uniswap v3 this time.

You will learn:

How to execute a swap on the Uniswap v3 Router02 to buy and sell USDT
How to wrap and unwrap Ether to WETH
How to call functions in a Solidity smart contract expecting a struct as parameter

Probably over 90% of the crypto trading volume out there is on Uniswap v2 / v3 clones, so if we can handle both of them, we’re good to go to the more advanced topics.

I will only explain the new v3 things. For the basics on how to execute a transaction, check out the last tutorial.

If you don’t know yet about the major differences between v2 and v3 like concentrated liquidity, check this official announcement from the Uniswap team and start reading a bit from there: https://blog.uniswap.org/uniswap-v3 . They had some really nice ideas, which we will use in later strategies.

Hint: For the full working code, check out the Github repository provided at the end of the article.

On Uniswap v3 you have three choices to execute a swap:

Swap with the pool contract: Details about the interface are here — https://docs.uniswap.org/contracts/v3/reference/core/UniswapV3Pool — we don’t need that for now and it’s less flexible compared to the router.
Use the SwapRouter02: This is what we’re going to do. It’s the easiest of those three and nearly the same as our last tutorial using the v2 router. You can find the details of the Swap Router here — https://docs.uniswap.org/contracts/v3/reference/periphery/SwapRouter .
Use the new Universal Router: Basically it is a wrapper around the other router contracts and can also handle NFTs. It has a more complicated command structure which is quite hard to debug, but it allows you to chain multiple commands with a single transaction and save on gas that way. We will use it in a later tutorial, because it’s more complicated, we dont care about NFTs anyway and you have to learn about some other new concepts like Permit2. So think its better to start with the v3 router first. You find the details here — https://docs.uniswap.org/contracts/universal-router/technical-reference .

Ok, let’s go.

First we do the basics like importing our modules and defining some basic variables like we did last time. We are working on a local Ganache fork of Ethereum Mainet again.

The only new variable is total_gas_used* which is basically a counter, where we track how much gas we spend. This becomes interesting when we compare the SwapRouter02 with the new universal router.

from web3 import Account, Web3
from abi import UNISWAP_V3_ROUTER2_ABI, WETH9_ABI, MIN_ERC20_ABI
import eth_abi.packed

private_key = "0x5d9d3c897ad4f2b8b51906185607f79672d7fec086a6fb6afc2de423c017330c"

chain_id = 1337
rpc_endpoint = "http://127.0.0.1:8545"

web3 = Web3(Web3.HTTPProvider(rpc_endpoint))
account = Account.from_key(private_key)

total_gas_used_buy = 0
amount_in = 1 * 10**18

weth_address = "0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2"
usdt_address = "0xdAC17F958D2ee523a2206206994597C13D831ec7"
swap_router02_address = "0x68b3465833fb72A70ecDF485E0e4C7bD8665Fc45"

# load contracts
swap_router_contract = web3.eth.contract(address=swap_router02_address, abi=UNISWAP_V3_ROUTER2_ABI)
weth_contract = web3.eth.contract(address=weth_address, abi=WETH9_ABI)
usdt_contract = web3.eth.contract(address=usdt_address, abi=MIN_ERC20_ABI)

An important note here: We have two versions of the Router. You can check all the deployment addresses here: https://docs.uniswap.org/contracts/v3/reference/deployments

During this tutorial we are using the SwapRouter02! So make sure to import the right ABI!

Now we want to swap some ETH for USDT. Therefore the SwapRouter02 offers four swap functions:

exactInputSingle: You would use this, if you know how many token you want to spend and there is a direct swap path available for that token pair. So if you want to buy USDT for 1 Ether and there is a USDT/WETH pool, it makes sense to use this.
exactInput: If you know how many token you want to spend and want to swap via intermediary tokens, you would use the exactInput. So if you want to swap your Ether to USDT and your USDT then to another token in a single transaction.
exactOutputSingle: Basically the same like the input variant, the only difference is that this time you know how many token you want to buy. So it makes sense, when you want to buy 1.000 USDT for any amount of Ether.
exactOutput: Should be clear by now.

You can find the details and the call parameters here: https://docs.uniswap.org/contracts/v3/reference/periphery/SwapRouter

Important: The documentation still refers to the old SwapRouter and not SwapRouter02. The 02 version has a little bit different ABI like the missing deadline and you sometimes get really hard to debug revert errors when doing mistakes here. We talk about the details later, but it took me a few days to figure all this out. So make sure to follow along exactly with my code / ABI / address and calls.

The last time we could use swapExectETHforTokens to swap ETH for our tokens, but as you just saw there is no direct function to swap ETH anymore. Other functions, like the ones to handle tax tokens are also removed!

So we need to wrap our ETH first to WETH. Wrapping means, that we lock our native Ether token in a wrapper smart contract and get a “tokenized” version of it adhering the ERC20 standard.

To do so we can just call the deposit function of the WETH9 contract ( https://etherscan.io/address/0xc02aaa39b223fe8d0a0e5c4f27ead9083c756cc2#code )

Hint: If you ask yourself, why it is WETH9 … here you go: It uses Kelvin versioning. So its start at a higher number and when it gets to zero, it is basically frozen and no further changes are allowed anymore.

# wrap eth
tx = weth_contract.functions.deposit().build_transaction({
        'chainId': web3.eth.chain_id,
        'gas': 2000000,
        "maxPriorityFeePerGas": web3.eth.max_priority_fee,
        "maxFeePerGas": 100 * 10**9,
        'nonce': web3.eth.get_transaction_count(account.address),
        'value': amount_in, # wrap 1 eth
})

signed_transaction = web3.eth.account.sign_transaction(tx, private_key)
tx_hash = web3.eth.send_raw_transaction(signed_transaction.rawTransaction)

tx_receipt = web3.eth.wait_for_transaction_receipt(tx_hash)
print(f"tx hash: {Web3.to_hex(tx_hash)}")
total_gas_used_buy += tx_receipt["gasUsed"]

weth_balance = weth_contract.functions.balanceOf(account.address).call()
print(f"weth balance: {weth_balance / 10**18}")

Nothing special to see here, just call the deposit() function and send the amount you want to wrap as value. As in our previous tutorial, we are using arbitrary high gas values for now and will optimize that later.

Now we need to allow the SwapRouter02 to spend our WETH. So we have to send an approve again.

# now approve the router to spend our weth
approve_tx = weth_contract.functions.approve(swap_router02_address, 2**256-1).build_transaction({
    'gas': 500_000,  # Adjust the gas limit as needed
    "maxPriorityFeePerGas": web3.eth.max_priority_fee,
    "maxFeePerGas": 100 * 10**9,
    "nonce": web3.eth.get_transaction_count(account.address),
})   

raw_transaction = web3.eth.account.sign_transaction(approve_tx, account.key).rawTransaction
tx_hash = web3.eth.send_raw_transaction(raw_transaction)
tx_receipt = web3.eth.wait_for_transaction_receipt(tx_hash)
print(f"tx hash: {Web3.to_hex(tx_hash)}")

if tx_receipt["status"] == 1:
    print(f"approve transaction send for unlimited amount")

total_gas_used_buy += tx_receipt["gasUsed"]

Again everything should be familiar by now. We approve an unlimited amount, you might want to fix that later in production.

Now we can finally prepare the swap transaction. We will use the exactInput function of the swap router. As we just want to swap from WETH to USDT you could also use exactInputSingle but using the exactInput gives you some more flexibility when adapting this code and i can show you the only interesting new stuff here: the path variable and how to pass structs as function parameters.

The call arguments are explained here: https://docs.uniswap.org/contracts/v3/reference/periphery/interfaces/ISwapRouter

Now lets start with the path which was quite tough to figure out.

The path is an ordered list of token addresses along the swap route and between each of the addresses is the fee of the pool.

Example:

You want to swap from WETH to USDT.

The path for the 0.05% pool is:

[weth_address,500,usdt_address]

If its a 1% fee pool, you would replace the 500 with 10.000.

Hint: The Uniswap v3 Factory contract allows to create pools with 0.05 percent (500), 0.3 percent (3000) and 1% (10.000). You need to figure out on your own, using popular tools like Dextools or Dexscreener which pool you want to trade. (When writing our Bot later, we will develop a smarter way for this)

If you want to swap to another token on a 1% pool you would use the following path:

[weth_address,500,usdt_address,10_000,another_token_address]

And if you want to sell a token later, you just inverse the path.

Ok, now we got the path as a python list but we need to convert it to a bytes object first.

When reading the Uniswap Code you see the exact format for the path is:

20 bytes address + 3 bytes fee + 20 bytes address […]

So we need to do some conversion magic here.

Using just Python and Web3 we could do it like this:

path = bytes.fromhex(Web3.to_checksum_address(weth_address)[2:]) + int.to_bytes(500, 3, "big") + bytes.fromhex(Web3.to_checksum_address(usdt_address)[2:])

That looks a bit scary but all we do is, we convert the addresses to hex checksum addresses, remove the leading 0x and add the pool fee as 3 bytes big endian integer in bytes and again the last address without leading 0x as bytes.

That would work, but there is also a quite handy Python module called eth_abi we in wise foresight already imported at the top of our script.

Eth_abi helps you to encode Python data types into the formats any arbitrary ABI expects:

path = eth_abi.packed.encode_packed(['address','uint24','address'], [weth_address,500,usdt_address])

So you just take the input parameters from the ABI as the first parameter and your Python data structure as the second. I guess that’s easier to read and that way from now on you can call any smart contract function expecting structs or complicated byte strings as parameters.

Hint: If you ever get some strange revert errors, check your path first. In general this is quite hard to debug but if you follow along my code exactly it should work.

After we managed to construct the path, we no can prepare our swap transaction call.

Instead of passing each argument separate they are passed as a struct in Solidity. This helps to save on gas fees because it may decrease the overall size of the passed arguments as smaller integers for example can be packed more closely together in memory.

So the Solidity struct the exactInput function expects as parameter looks like this:

  struct ExactInputParams {
        bytes path;
        address recipient;
        uint256 deadline; //<---- Removed in SwapRouter02!
        uint256 amountIn;
        uint256 amountOutMinimum;
    }

I added that comment for you, because the official documentation seems not to talk about that.

To archive the same in Python you can just use a Tuple like this.

path = eth_abi.packed.encode_packed(['address','uint24','address'], [weth_address,500,usdt_address])


tx_params = (
    path, 
    account.address, 
    amount_in, # amount in
    0 #min amount out
)

As you can see we pass the path, our receiving address and the amount we want to spend and the minimum amount we want to receive. In this case we spend 1 WETH for any amount of USDT.

Hint: In production code, make sure to set an appropriate min amount out. We will have a look at this later in another tutorial.

And finally we can do some swapping.

swap_buy_tx = swap_router_contract.functions.exactInput(tx_params).build_transaction(
        {
            'from': account.address,
            'gas': 500_000,
            "maxPriorityFeePerGas": web3.eth.max_priority_fee,
            "maxFeePerGas": 100 * 10**9,
            'nonce': web3.eth.get_transaction_count(account.address),
        })

raw_transaction = web3.eth.account.sign_transaction(swap_buy_tx, account.key).rawTransaction
print(f"raw transaction: {raw_transaction}")
tx_hash = web3.eth.send_raw_transaction(raw_transaction)
tx_receipt = web3.eth.wait_for_transaction_receipt(tx_hash)
print(f"tx hash: {Web3.to_hex(tx_hash)}")
total_gas_used_buy += tx_receipt["gasUsed"]

usdt_balance = usdt_contract.functions.balanceOf(account.address).call()
print(f"usdt balance: {usdt_balance / 10**6}")

Nothing special here, we just send our transaction, omit proper error checking (feel free to add that), increase our gas counter and print some balances.

That was the hard part already. But to complete this I will also sell some USDT again and unwrap the WETH.

Before we can sell the token, we need to approve the router again to spend our USDT tokens.

total_gas_used_sell = 0
# now approve the router to spend our usdt
approve_tx = usdt_contract.functions.approve(swap_router02_address, 2**256-1).build_transaction({
    'gas': 500_000,  # Adjust the gas limit as needed
    "maxPriorityFeePerGas": web3.eth.max_priority_fee,
    "maxFeePerGas": 100 * 10**9,
    "nonce": web3.eth.get_transaction_count(account.address),
})   

raw_transaction = web3.eth.account.sign_transaction(approve_tx, account.key).rawTransaction
tx_hash = web3.eth.send_raw_transaction(raw_transaction)
tx_receipt = web3.eth.wait_for_transaction_receipt(tx_hash)
print(f"tx hash: {Web3.to_hex(tx_hash)}")


if tx_receipt["status"] == 1:
    print(f"approve transaction send for unlimited amount")
total_gas_used_sell += tx_receipt["gasUsed"]

An important thing to note here is that USDT has a somehow strange implementation of the approval. Before you can change it, you have to reset it to zero. This might cause some frustrations during testing, but shouldn’t happen when setting the unlimited amount. This is not the case with other tokens.

Also USDT has just 6 decimals … just a little reminder in case you want to do some math with this.

Now we just have to reverse the Path from above to sell our USDT again:

path = eth_abi.packed.encode_packed(['address','uint24','address'], [usdt_address,500,weth_address])

And finally we can swap our USDT for WETH by sending the transaction.

tx_params = (
    path, 
    account.address, 
    usdt_balance, # amount in
    0 #min amount out
)

swap_sell_tx = swap_router_contract.functions.exactInput(tx_params).build_transaction(
        {
            'from': account.address,
            'gas': 500_000,
            "maxPriorityFeePerGas": web3.eth.max_priority_fee,
            "maxFeePerGas": 100 * 10**9,
            'nonce': web3.eth.get_transaction_count(account.address),
        })

raw_transaction = web3.eth.account.sign_transaction(swap_sell_tx, account.key).rawTransaction
print(f"raw transaction: {raw_transaction}")
tx_hash = web3.eth.send_raw_transaction(raw_transaction)
tx_receipt = web3.eth.wait_for_transaction_receipt(tx_hash)
print(f"tx hash: {Web3.to_hex(tx_hash)}")
total_gas_used_sell += tx_receipt["gasUsed"]

usdt_balance = usdt_contract.functions.balanceOf(account.address).call()
print(f"usdt balance: {usdt_balance / 10**6}")
weth_balance = weth_contract.functions.balanceOf(account.address).call()
print(f"weth balance: {weth_balance / 10**18}")

And for the sake of completeness, we unwrap our WETH again to ETH and print some balances.

To do so, we just call the withdraw function of the WETH9 contract with the amount we got from the swap.

# now unwrap our weth again to good old eth
tx = weth_contract.functions.withdraw(weth_balance).build_transaction({
        'chainId': web3.eth.chain_id,
        'gas': 2000000,
        "maxPriorityFeePerGas": web3.eth.max_priority_fee,
        "maxFeePerGas": 100 * 10**9,
        'nonce': web3.eth.get_transaction_count(account.address),
})

signed_transaction = web3.eth.account.sign_transaction(tx, private_key)
tx_hash = web3.eth.send_raw_transaction(signed_transaction.rawTransaction)

tx_receipt = web3.eth.wait_for_transaction_receipt(tx_hash)
print(f"tx hash: {Web3.to_hex(tx_hash)}")
total_gas_used_sell += tx_receipt["gasUsed"]

weth_balance = weth_contract.functions.balanceOf(account.address).call()
print(f"weth balance: {weth_balance / 10**18}")

print(f"total gas used for wrap + aprove + buy: {total_gas_used_buy}")
print(f"total gas used for approve + sell + unwrap: {total_gas_used_sell}")
print(f"total gas used for everything: {total_gas_used_buy + total_gas_used_sell}")

And that’s it.

You can find the complete source code of the tutorial and the required ABI file again in my Github here:

https://github.com/crjameson/python-defi-tutorials/blob/main/uniswapv3/

I spare myself the final last words, this already feels quite long. So see you around next time with some other interesting Python and DeFi stuff.

Thank you for reading and if you would like to read more and become a defi crypto python trading ninja expert, subscribe here on medium, substack or twitter.

Automated Trade Execution on Uniswap v3 in Python — We buy and sell a Token again …

Written by crjameson