Python classes to aid rapid development of Uniswap (V2, V3, V4), Curve V1, Solidly V2, Balancer V2, and Aave V3 integrations on EVM-compatible blockchains.
- Overview
- Installation
- Quick Start
- Supported Protocols
- Examples
- CLI Reference
- Configuration
- Rust Extension
- Documentation
- Contributing
- License
- Donation
Degenbot is a set of Python classes that abstract many of the implementation details of Uniswap liquidity pools and their underlying ERC-20 tokens. It uses web3.py for communication with an EVM blockchain through the standard JSON-RPC interface.
These classes serve as building blocks for the lessons published by BowTiedDevil on Degen Code.
- Python 3.12+
pip,uv, or similar package management tool
pip install degenbotgit clone https://github.com/BowTiedDevil/degenbot.git
cd degenbot
uv sync # or: pip install -e .import web3
import degenbot
# Connect to an Ethereum RPC endpoint
w3 = web3.Web3(web3.HTTPProvider("https://eth-mainnet.example.com"))
# Verify connection
assert w3.is_connected()
# Create a Uniswap V3 pool helper from an address
pool = degenbot.UniswapV3Pool("0x8ad599c3a0ff1de082011efddc58f1908eb6e6d8")
# Inspect pool state
print(f"Pool: {pool.name}")
print(f"Token 0: {pool.token0.symbol}")
print(f"Token 1: {pool.token1.symbol}")
print(f"Liquidity: {pool.liquidity}")
# Calculate swap outputs
amount_out = pool.calculate_tokens_out_from_tokens_in(
token_in=pool.token0,
token_in_quantity=10**18, # 1 token (18 decimals)
)
print(f"Output: {amount_out}")| Protocol | Versions | Chains |
|---|---|---|
| Uniswap | V2, V3, V4 | Ethereum, Base |
| Aerodrome | V2, V3 | Base |
| PancakeSwap | V2, V3 | Ethereum, Base |
| SushiSwap | V2, V3 | Ethereum, Base |
| Curve | V1 | Ethereum |
| Solidly | V2 | Ethereum, Base |
| Balancer | V2 | Ethereum |
| Camelot | V2 | Arbitrum |
| Protocol | Features |
|---|---|
| Aave V3 | Supply, Borrow, Withdraw, Repay, Liquidation, E-Mode, GHO |
| Feature | Description |
|---|---|
| Chainlink Price Feeds | Oracle price data |
| Anvil Forking | Local forked blockchain for testing |
The following snippets assume a connected Web3 instance with a working provider on Ethereum mainnet (chain ID #1), and the classes imported under the degenbot namespace.
# Create `UniswapV2Pool` object from on-chain data at the given address and
# current chain height
>>> lp = degenbot.UniswapV2Pool('0xBb2b8038a1640196FbE3e38816F3e67Cba72D940')
• WBTC (Wrapped BTC)
• WETH (Wrapped Ether)
• Token 0: WBTC - Reserves: 10732489743
• Token 1: WETH - Reserves: 2056834999904002274711
# Inspect the tokens held by the pool
>>> lp.token0
Erc20Token(
address=0x2260FAC5E5542a773Aa44fBCfeDf7C193bc2C599,
symbol='WBTC',
name='Wrapped BTC',
decimals=8
)
>>> lp.token1
Erc20Token(
address=0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2,
symbol='WETH',
name='Wrapped Ether',
decimals=18
)
>>> lp.fee_token0
Fraction(3, 1000)
>>> lp.fee_token1
Fraction(3, 1000)
# Predict the input and output values for swaps through the pool, accounting for fees
>>> lp.calculate_tokens_out_from_tokens_in(
token_in=lp.token1,
token_in_quantity=1*10**18
)
5199789
>>> lp.calculate_tokens_in_from_tokens_out(
token_out=lp.token0,
token_out_quantity=5199789
)
999999992817074189
# Update the current reserves from the live blockchain
>>> lp.auto_update(silent=False)
[WBTC-WETH (V2, 0.30%)]
WBTC: 10732455184
WETH: 2056841643098872755548
>>> lp.reserves_token0
10732455184
>>> lp.reserves_token1
2056841643098872755548>>> lp = degenbot.UniswapV3Pool('0xCBCdF9626bC03E24f779434178A73a0B4bad62eD')
WBTC-WETH (V3, 0.30%)
• Token 0: WBTC
• Token 1: WETH
• Fee: 3000
• Liquidity: 544425151051415575
• SqrtPrice: 34048891009198980752047510166697902
• Tick: 259432
# Calculate inputs and outputs
>>> lp.calculate_tokens_out_from_tokens_in(
token_in=lp.token1,
token_in_quantity=1*10**18
)
5398169
>>> lp.calculate_tokens_in_from_tokens_out(
token_out=lp.token0,
token_out_quantity=5398169
)
999999871563434214
# Inspect the known liquidity positions
>>> lp.tick_bitmap
{
0: UniswapV3BitmapAtWord(
bitmap=1,
block=18517665
),
16: UniswapV3BitmapAtWord(
bitmap=115792089237316195423570985008687907853268655437644779123584680198630541352072,
block=18517670
)
}
# The V3 liquidity pool helper is optimized for fast instantiation, and will lazy-load
# liquidity data for positions outside of the current range as needed.
>>> lp.tick_data
{
0: UniswapV3LiquidityAtTick(
liquidityNet=10943161472679,
liquidityGross=10943161472679,
block=18517665
),
261060: UniswapV3LiquidityAtTick(
liquidityNet=-910396189679465,
liquidityGross=910396189679465,
block=18517670
),
261000: UniswapV3LiquidityAtTick(
liquidityNet=-3774266260841234,
liquidityGross=3774266260841234,
block=18517670
),
...
}Uniswap V4 introduces hooks and a new pool manager architecture. The UniswapV4Pool class provides access to V4 pools with support for the new features.
>>> lp = degenbot.UniswapV4Pool(
... pool_id='0x96d4b53a38337a5733179751781178a2613306063c511b78cd02684739288c0a',
... pool_manager_address='0x498581fF718922c3f8e6A244956aF099B2652b2b',
... state_view_address='0xA3c0c9b65baD0b08107Aa264b0f3dB444b867A71',
... tokens=[
... '0x0000000000000000000000000000000000000000',
... '0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913',
... ],
... fee=500,
... tick_spacing=10
... )
ETH-USDC (UniswapV4Pool, id=0x96d4b53a38337a5733179751781178a2613306063c511b78cd02684739288c0a)
• ID: 0x96d4b53a38337a5733179751781178a2613306063c511b78cd02684739288c0a
• Token 0: ETH
• Token 1: USDC
• Liquidity: 60429069420043934
• SqrtPrice: 4220772448119892035402666
• Tick: -196812
# Calculate output for a 1 ETH swap
>>> lp.calculate_tokens_out_from_tokens_in(
... token_in=lp.token0,
... token_in_quantity=1*10**18,
... )
2834164215
# Inspect active hooks
>>> lp.active_hooks
frozenset()
# Get pool key information
>>> lp.pool_key
UniswapV4PoolKey(
currency0='0x0000000000000000000000000000000000000000',
currency1='0x833589fCD6eDb6E08f4c7C32D4f71b54bdA02913',
fee=500,
tick_spacing=10,
hooks='0x0000000000000000000000000000000000000000'
)The AnvilFork class is used to launch a fork with anvil from the Foundry toolkit. The object provides a w3 attribute, connected to an IPC socket, which can be used to communicate with the fork like a typical RPC.
>>> fork = degenbot.AnvilFork(fork_url='http://localhost:8545')
>>> fork.w3.eth.chain_id
1
>>> fork.w3.eth.block_number
22675736
# The `AnvilFork` instance also exposes HTTP and WS endpoints that can be used to make a
# separate connection from a remote machine.
>>> import web3
>>> _w3 = web3.Web3(web3.HTTPProvider(fork.http))
>>> _w3.is_connected()
True
>>> _w3 = web3.Web3(web3.LegacyWebSocketProvider(fork.ws_url))
>>> _w3.is_connected()
True
# The fork can be reset to a different endpoint, which defaults to the latest block.
>>> fork.reset(fork_url='http://localhost:8544')
>>> fork.w3.eth.chain_id
8453
# The fork can also be reset with a specified block number or a transaction hash.
>>> fork.reset(fork_url='http://localhost:8545', block_number=22_675_800)
>>> fork.w3.eth.chain_id
1
>>> fork.w3.eth.block_number
22675800
>>> fork.reset(fork_url='http://localhost:8545', block_number=22_675_800)
>>> fork.w3.eth.chain_id
1
>>> fork.w3.eth.block_number
22675800
# The fork can also be reset to an imaginary block after a specific transaction
# hash. See the [Anvil reference](https://getfoundry.sh/anvil/reference/) for the
# associated `--fork-transaction-hash` option.
>>> fork.reset(
fork_url='http://localhost:8545',
transaction_hash='0xc16e63e693a2748559c0fd653ade195be426472dddc5bfa3fcc769c4c88c249c'
)
>>> fork.w3.eth.block_number
22675814
# Blocks can be manually mined
>>> fork.mine()
>>> fork.w3.eth.block_number
22675815
# Byte code can be set for an arbitrary address.
>>> fork.set_code(
address='0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045',
bytecode=bytes.fromhex('45')
)
>>> fork.w3.eth.get_code('0xd8dA6BF26964aF9D7eEd9e03E53415D37aA96045')
HexBytes('0x45')The Anvil client offers many options. The most common ones are exposed by constructor options to AnvilFork.
Users wanting fine-grained control over all client options may pass them through the anvil_opts argument, which takes a list of strings. These will be passed directly to the client after all of the managed options.
# Launch with the Optimism feature set, which enables special transaction types.
>>> fork = degenbot.AnvilFork(
fork_url='http://localhost:8544',
anvil_opts=['--optimism']
)
# Launch with a non-default hardfork, which may be necessary for accurate simulation on a
# historical block.
>>> fork = degenbot.AnvilFork(
fork_url='http://localhost:8545',
fork_block=12_980_000,
anvil_opts=['--hardfork=london']
)
# Launch with a non-default transaction pool ordering scheme
>>> fork = degenbot.AnvilFork(
fork_url='http://localhost:8545',
anvil_opts=['--order=fifo']
)
# Launch with certain debugging features enabled
>>> fork = degenbot.AnvilFork(
fork_url='http://localhost:8545',
anvil_opts=[
'--disable-block-gas-limit',
'--disable-code-size-limit',
'--disable-min-priority-fee',
]
)Several classes are provided to simplify the calculation of optimal arbitrage amounts for a given sequence of pools.
>>> v2_lp = degenbot.UniswapV2Pool('0xBb2b8038a1640196FbE3e38816F3e67Cba72D940')
• WBTC (Wrapped BTC)
• WETH (Wrapped Ether)
• Token 0: WBTC - Reserves: 6390612659
• Token 1: WETH - Reserves: 2534027291379197003140
>>> v3_lp = degenbot.UniswapV3Pool('0xCBCdF9626bC03E24f779434178A73a0B4bad62eD')
WBTC-WETH (UniswapV3Pool, 0.30%)
• Address: 0xCBCdF9626bC03E24f779434178A73a0B4bad62eD
• Token 0: WBTC
• Token 1: WETH
• Fee: 3000
• Liquidity: 261799575241796322
• SqrtPrice: 49883600179466982678044042954714957
• Tick: 267070
• State Block (Initial): 22676748
>>> weth = v2_lp.token1
>>> arb = degenbot.UniswapLpCycle(
id="test",
input_token=weth,
swap_pools=[v2_lp, v3_lp]
)
# The minimum rate of exchange for a profitable arbitrage is 1.0. The pool states at
# a given block are likely to be less, so override the minimum for illustration.
# The `ArbitrageCalculationResult` must be encoded as a properly-formed transaction
# by the user and broadcast to the network to secure the opportunity.
>>> arb.calculate(min_rate_of_exchange=0.8)
ArbitrageCalculationResult(
id='test',
input_token=Erc20Token(
address=0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2,
symbol='WETH',
name='Wrapped Ether',
decimals=18
),
profit_token=Erc20Token(
address=0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2,
symbol='WETH',
name='Wrapped Ether',
decimals=18
),
input_amount=69600394635598,
profit_amount=-623178922742,
swap_amounts=(
UniswapV2PoolSwapAmounts(
pool='0xBb2b8038a1640196FbE3e38816F3e67Cba72D940',
amounts_in=(0, 69600394635598),
amounts_out=(175, 0),
recipient=None
),
UniswapV3PoolSwapAmounts(
pool='0xCBCdF9626bC03E24f779434178A73a0B4bad62eD',
amount_specified=175,
zero_for_one=True,
sqrt_price_limit_x96=4295128740,
recipient=None
)
),
state_block=22676748
)Chainlink price feeds provide reliable oracle data for various assets. The ChainlinkPriceContract class simplifies access to these feeds.
# Load the price feed for ETH/USD
>>> price_feed = degenbot.ChainlinkPriceContract(
... '0x5f4eC3Df9cbd43714FE2740f5E3616155c5b8419'
... )
>>> price_feed.price
2836.68731709
# Check the decimals used by the price feed
>>> price_feed.decimals
8
# Call an arbitrary function `latestRoundData` on the underlying contract
>>> price_feed.w3_contract.functions.latestRoundData().call()
[
129127208515966883788,
283668731709,
1766031970,
1766031983,
129127208515966883788
]Degenbot provides a command-line interface for managing blockchain data and pool state.
The CLI is installed automatically with the package:
pip install degenbot
degenbot --help# Back up the database
degenbot database backup
# Reset database (creates fresh schema)
degenbot database reset
# Upgrade database schema to latest version
degenbot database upgrade [--force]
# Compact database to reclaim space
degenbot database compact# Update pool metadata and liquidity positions for all active exchanges
degenbot pool update [--chunk SIZE] [--to-block BLOCK]
# Activate an exchange for tracking
degenbot exchange activate base_uniswap_v3
# Deactivate an exchange
degenbot exchange deactivate base_uniswap_v3Supported exchanges:
- Base:
base_aerodrome_v2,base_aerodrome_v3,base_pancakeswap_v2,base_pancakeswap_v3,base_sushiswap_v2,base_sushiswap_v3,base_swapbased_v2,base_uniswap_v2,base_uniswap_v3,base_uniswap_v4 - Ethereum:
ethereum_pancakeswap_v2,ethereum_pancakeswap_v3,ethereum_sushiswap_v2,ethereum_sushiswap_v3,ethereum_uniswap_v2,ethereum_uniswap_v3,ethereum_uniswap_v4
# Update Aave V3 positions for all active markets
degenbot aave update [--chunk SIZE] [--to-block BLOCK]
# Activate an Aave market
degenbot aave activate ethereum_aave_v3
# Deactivate an Aave market
degenbot aave deactivate ethereum_aave_v3Commands accepting --to-block support the following formats:
| Format | Example | Description |
|---|---|---|
latest |
latest |
Latest block |
latest:-N |
latest:-64 |
N blocks before latest (default) |
safe:+N |
safe:128 |
N blocks after safe block |
| Number | 18900000 |
Specific block number |
| Variable | Values | Description |
|---|---|---|
DEGENBOT_DEBUG |
1, true, yes |
Enable debug-level logging output |
DEGENBOT_DEBUG=1 python my_script.pyDegenbot uses a TOML configuration file located at ~/.config/degenbot/config.toml:
[rpc]
# Chain ID to RPC endpoint mapping
1 = "https://eth-mainnet.example.com"
8453 = "https://base-mainnet.example.com"
[database]
# SQLite database path (optional, defaults to platform-specific location)
path = "/path/to/degenbot.db"Experimental: The Rust extension is not activated by default. To enable it in a future release, set the environment variable
DEGENBOT_EXPERIMENTAL_RUST=1. The Python API documented below is subject to change.
Degenbot includes a high-performance Rust extension module (_rs) that provides optimized implementations of performance-critical operations. The extension is built automatically during installation using maturin.
| Crate | Purpose |
|---|---|
| alloy | Ethereum primitives (Address, U256, B256), RPC types, keccak256 |
| pyo3 | Python bindings with abi3-py312 for Python 3.12+ support |
| tokio | Multi-threaded async runtime for concurrent RPC calls |
| parking_lot | High-performance RwLock for thread-safe caching |
| num-bigint | Arbitrary precision integers for Python int interop |
| thiserror | Derivative error types |
Uniswap V3 tick-to-price conversions:
from degenbot._rs import get_sqrt_ratio_at_tick, get_tick_at_sqrt_ratio
# Convert tick to sqrt price (X96 format)
sqrt_price = get_sqrt_ratio_at_tick(253320) # Returns: 56736275128821120...
# Convert sqrt price back to tick
tick = get_tick_at_sqrt_ratio(56736275128821120) # Returns: 253320High-performance ABI decoding for contract data:
from degenbot._rs import decode, decode_single
# Decode multiple values
types = ["address", "uint256", "uint256"]
data = bytes.fromhex("...")
values = decode(types, data) # Returns list of decoded values
# Decode a single value
address = decode_single("address", bytes.fromhex("..."))EIP-55 checksummed address conversion:
from degenbot._rs import to_checksum_address
checksummed = to_checksum_address("0xdeadbeefdeadbeefdeadbeefdeadbeefdeadbeef")
# Returns: "0xDeaDbeefdEAdbeefdEadbEEFdeadbeEFdEaDbeeF"The extension includes synchronous and async Ethereum RPC providers:
from degenbot._rs import AlloyProvider, Contract
# Create provider with connection pooling
provider = AlloyProvider(
rpc_url="https://eth-mainnet.example.com",
max_connections=10,
timeout=30.0,
)
# Query blockchain
block_number = provider.get_block_number()
chain_id = provider.get_chain_id()
logs = provider.get_logs(from_block=1000000, to_block=1000100, addresses=["0x..."])
# Contract interaction
contract = Contract("0x...", provider_url="https://...")
result = contract.call("balanceOf(address)", ["0x..."])
provider.close()A high-performance bytes type with pre-computed hex representation:
from degenbot._rs import FastHexBytes
# Create from various inputs
data = FastHexBytes("0xdeadbeef")
data = FastHexBytes(b"\xde\xad\xbe\xef")
# Zero-cost hex conversion (pre-computed)
hex_str = data.hex() # "0xdeadbeef"
# Direct bytes access
raw = data.raw # b"\xde\xad\xbe\xef"| Operation | Pure Python | Rust Extension |
|---|---|---|
| Tick math | ~50μs | ~0.1μs |
| ABI decode (10 values) | ~200μs | ~5μs |
| Address checksum | ~10μs | ~0.5μs |
| Log query (1000 logs) | ~100ms | ~20ms |
The extension is pre-built in published packages. For source builds:
- Rust 1.70+ (stable toolchain)
- maturin (installed automatically with
uv sync)
# Build the extension
cargo build --release --features extension-module --manifest-path rust/Cargo.toml
# Or use the justfile
just dev # Build and install Python extensionAdditional documentation is available in the docs/ directory:
- Aave V3: Comprehensive control flow diagrams and amount transformations for Aave operations
- Arbitrage: Multi-pool cycle testing documentation
- CLI: Detailed CLI command reference
- Configuration: Configuration options
Contributions are welcome! Please submit issues and pull requests to the GitHub repository.
git clone https://github.com/BowTiedDevil/degenbot.git
cd degenbot
uv sync
# Run tests
uv run pytest
# Run linting
uv run ruff check
uv run mypyThis code is published under a permissive MIT license. See LICENSE for details.
If you find this code valuable, please fund continuing development by donating to 0xADAf500b965545C8A766CD9Cdeb3BF3FBef073e5 on any EVM compatible chain.