Uncovering LP Holder Transaction Actions using web3.py

This analysis examines Ethereum transaction logs to track liquidity provider (LP) behavior in a Uniswap pool, focusing on activities like liquidity changes, fee collection, and swaps over a specific time period.

Disclaimer: The information provided here is for informational purposes only and is not intended to be personal financial, investment, or other advice.

This article builds upon two earlier articles: “Ethereum Transaction Log Analysis: Uncovering LP Holder Actions” and “Ethereum Transaction Log Analysis: Uncovering LP Holder Actions Using GraphQL”. In the first article, I demonstrated how to use web3py to extract liquidity provider (LP) positions from a Uniswap v3 pool by analyzing on-chain transaction logs, with a focus on identifying core LP activities — such as adding or removing liquidity — directly from the Ethereum blockchain. The second article built on that foundation by assigning USD values to LP positions, providing a more intuitive and financially relevant view of the data. In this follow-up, we present LP actions in a format similar to Etherscan’s Transaction Action section, as shown below.

Transaction Action from Etherscan

The above will be further improved by adding Token IDs to the ‘Add,’ ‘Remove,’ and ‘Collect’ events.

Familiarity with fundamental Python is required. Code is available as a Jupyter notebook on GitHub.

Setup

Prerequsites

You need an API key for Etherscan and a provider for the web3py library. They’re both free. This code uses Alchemy’s provider. You can learn more about providers here. You will also need a Coingecko Public API users (Demo plan) free API key for retrieving token prices.

Python Libraries

The required Python libraries are:

• web3.py: a Python library for interacting with Ethereum — required to get LP transactions
• tabulate: to display data in table format
• python-dotenv: reads key-value pairs from a .env file and can set them as environment variables.
• requests: HTTP library
• pandas: DataFrame and other utilities; imported as pd

Import Libraries

from web3 import Web3
from web3.contract.contract import Contract, ContractFunctions, ContractEvents

from eth_typing import ChecksumAddress
from eth_typing import ChainId

from functools import lru_cache
from dataclasses import dataclass

# For enums
from enum import StrEnum

# To read environment property file
import os
from dotenv import load_dotenv
from pathlib import Path

import requests
import json

# Date calculations
from datetime import datetime, timedelta
import time

# Importing Pandas to create DataFrame
import pandas as pd

# For display table
from tabulate import tabulate

Constants

# Token pair
TOKEN0_ADDRESS = '0xA0b86991c6218b36c1d19D4a2e9Eb0cE3606eB48'
TOKEN1_ADDRESS = '0xC02aaA39b223FE8D0A0e5C4F27eAD9083C756Cc2'

# Fee tier (0.05%)
FEE_TIER = 500

# Path to ABIs
ABI_PATH = 'assets/abi'

# Pool Events
class PoolEvent(StrEnum):
    BURN = 'Burn'
    COLLECT = 'Collect'
    MINT = 'Mint'
    SWAP = 'Swap'

# NFT Events
class NFTEvent(StrEnum):
    COLLECT = 'Collect'
    INCREASE_LIQUIDITY = 'IncreaseLiquidity'
    DECREASE_LIQUIDITY = 'DecreaseLiquidity'

# How many days back - approximately 1 year
DAYS = 360

# Page size for ether scan search
PAGE_SIZE = 100

# Maximum pages we are going to search; max number of items return = PAGE_SIZE * MAX_PAGES
MAX_PAGES = 5

# Etherscan endpoint
ETHERSCAN_ENDPOINT = 'https://api.etherscan.io/v2/api'

# Coingecko price API endpoint
COINGECKO_PRICE_ENDPOINT = 'https://api.coingecko.com/api/v3/simple/token_price/ethereum?contract_addresses={address}&vs_currencies=usd'

• TOKEN0, TOKEN1 — addresses of the pool token pair
• FEE_TIER — specifies a unique pool among those with matching token pairs
• PoolEvent, NFTEvent — enum classes
• DAYS — indicates the number of days to look back with search
• PAGE_SIZE — page size for the Etherscan API
• MAX_PAGES — maximum pages; required for the Etherscan API
• ETHERSCAN_ENDPOINT — the endpoint for the Etherscan API
• COINGECKO_PRICE_ENDPOINT — the endpoint for the token price

Load environment variables

dotenv_path = Path('.env/uniswap')
load_dotenv(dotenv_path=dotenv_path)

PROVIDER_URL = os.getenv('PROVIDER_URL')
ETHERSCAN_API_KEY = os.getenv('ETHERSCAN_API_KEY')
WALLET_ADDRESS = os.getenv('WALLET_ADDRESS')
COINGECKO_API_KEY = os.getenv('COINGECKO_API_KEY')

The dotenv library loads the above property values from the .env/uniswap file.

• PROVIDER_URL — as stated under Prerequisites, this is the provider from Alchemy.
• ETHERSCAN_API_KEY — holds the Etherscan API key
• WALLET_ADDRESS — LP’s wallet address
• COINGECKO_API_KEY — Coingecko API key

Contracts

To interact with the blockchain for reading data and executing transactions, we utilize contracts. We use an ERC20Contract wrapper class to simplify working with ERC20 standard token contracts. Similarly, wrapper classes are also in place to provide simplified interfaces for the Factory V3, Pool, and NFT Position Manager contracts.

# Token pair contracts
token0 = ERC20Contract(w3=web3, address=TOKEN0_ADDRESS)
token1 = ERC20Contract(w3=web3, address=TOKEN1_ADDRESS)

# Factory contract
factory_contract = FactoryV3Contract(w3=web3)
pool_address = factory_contract.functions.getPool(token0.address, token1.address, FEE_TIER).call()

# Create Pool and NFT manager contracts
pool_contract = PoolV3Contract(w3=web3, address=pool_address)
nftmgr_contract = NFTPositionManagerContract(w3=web3)

The Factory V3 wrapper class is used to obtain a pool address by passing the addresses of the token pair and the relevant fee tier. The notebook contains the source code for these helpful wrapper classes.

Event Signatures

Event signatures are crucial for determining the event name. We compute event signatures for both smart and NFT contracts, saving them in a dictionary. A key-value pair makes up each item; the key is the hexadecimal event signature, and the value is the event name. Check out my article, “*Understanding Ethereum Raw Transactions: Mapping Smart Contract Functions”*, for details on creating event signatures.

The event_signature method is part of the BaseContract, which serves as the parent class for the ‘pool’ and ‘NFT Manager’ contract classes. The ‘key.value’ represents the event name.

# Event signatures for pool
pool_signatures = {pool_contract.event_signature(key.value): key.value for key in [
    PoolEvent.BURN, PoolEvent.MINT, PoolEvent.COLLECT]}

# Event signatures for nft manager
nft_signatures = {nftmgr_contract.event_signature(key.value): key.value for key in [
    NFTEvent.COLLECT, NFTEvent.INCREASE_LIQUIDITY, NFTEvent.DECREASE_LIQUIDITY]}

Event signatures:

Pool:
{'0x0c396cd989a39f4459b5fa1aed6a9a8dcdbc45908acfd67e028cd568da98982c': 'Burn',
 '0x7a53080ba414158be7ec69b987b5fb7d07dee101fe85488f0853ae16239d0bde': 'Mint',
 '0x70935338e69775456a85ddef226c395fb668b63fa0115f5f20610b388e6ca9c0': 'Collect'}

NFT:
{'0x40d0efd1a53d60ecbf40971b9daf7dc90178c3aadc7aab1765632738fa8b8f01': 'Collect',
 '0x3067048beee31b25b2f1681f88dac838c8bba36af25bfb2b7cf7473a5847e35f': 'IncreaseLiquidity',
 '0x26f6a048ee9138f2c0ce266f322cb99228e8d619ae2bff30c67f8dcf9d2377b4': 'DecreaseLiquidity'}

Next, we need to retrieve normal (as stated in the Etherscan API docs) and NFT transactions linked to the wallet.

Normal and NFT transactions for wallet

# Get the start block
start_block = calculate_start_block(chainid=ChainId.ETH)

# Normal transactions
trans = get_transactions(chainid=ChainId.ETH, action='txlist', start_block=start_block)
# Filter out non contract and error transactions, only interested in hash
normal_list = [x['hash'] for x in trans if is_contract_address(x['to']) and x['isError']=='0']

# NFT transactions
trans = get_transactions(chainid=ChainId.ETH, action='tokennfttx', start_block=start_block)
nft_list = [x['hash'] for x in trans]

# Merge two lists, remove any duplicates
txlist = list(set(normal_list + nft_list))

The get_transactions utility method fetches specified transaction types from the Etherscan API, beginning its search at the start_block returned by the calculate_start_block utility method. After retrieval, this list is filtered to exclude transactions that resulted in errors or were sent to non-contract addresses. This filtering step utilizes another utility method, is_contract_address, which specifically checks if an address belongs to a smart contract. The code for calculate_start_block, get_transactions and is_contract_address can be found in the notebook.

Similarly, we also save only the hash for NFT transactions. Finally, we merge the lists using a set operation, thus removing any duplicates.

The txlist may contain transactions unrelated to our pool. To address this, we will apply additional filtering to include only transactions associated with our pool address and whose logs contain at least one of the pool’s signature methods. Here is the method to do filtering:

def is_tx_in_scope(tran_hash:str):
    """ Return the transaction hash if it is in scope or else None is returned
    Parameters:
    tran_hash : str
        transaction hash

    Returns:
    str
        Transaction hash if in scope or None
    """
    # Get transaction receipt
    for log in web3.eth.get_transaction_receipt(tran_hash).logs:
        # Check topic 0 is in our interested signatures
        if pool_contract.address == log.address and web3.to_hex(log['topics'][0]) in pool_signatures:
            return tran_hash
    return None

And we apply the map operator to each transaction as shown below:

# Only include transactions with logs with pool contracts address and topics with pool signatures
txlogs = set(map(is_tx_in_scope, txlist))
# Remove any None values
txlogs = [x for x in txlogs if x is not None] 

Filter out None values, which are returned when a transaction falls outside the scope, to produce a list of transactions that have interacted with the pool and include logs of events relevant to us, such as ‘Burn’, ‘Mint’, or ‘Collect’ events.

Display Transaction Actions

Getting Transactions

The method outlined below generates a list of transaction actions for display. While the method is lengthy, its primary function is to invoke the utility method when a transaction includes ‘Burn’, ‘Mint’, ‘Collect’, or ‘Swap’ events. Each event action is concatenated using the system’s newline character, with the transaction and timestamp appended to form a complete item before adding it to the list of actions to be returned.

def get_transaction_actions(transactions:set) -> list:
    burn_signatures = [key for key, val in pool_signatures.items() if val == 'Burn']
    burn_signatures.append([key for key, val in nft_signatures.items() if val == 'DecreaseLiquidity'][0])

    mint_signatures = [key for key, val in pool_signatures.items() if val == 'Mint']
    mint_signatures.append([key for key, val in nft_signatures.items() if val == 'IncreaseLiquidity'][0])

    collect_signatures = [key for key, val in pool_signatures.items() if val == 'Collect']
    collect_signatures.append([key for key, val in nft_signatures.items() if val == 'Collect'][0])

    swap_signature = pool_contract.event_signature('Swap')

    # List of transaction actions to return
    action_list = []
    for tx in transactions:
        # A list contains action details
        action = []

        receipt = web3.eth.get_transaction_receipt(tx)

        # A list to collect results from various events
        messages = []

        # Is it a Burn event?
        logs = [x for x in receipt.logs if web3.to_hex(x['topics'][0]) in burn_signatures]
        # Has to be at least 2, Burn and DecreaseLiquidity
        if len(logs) > 1:
            # Handle burn events
            messages.append(handle_events(logs=logs))

        # Is it a Collect event?
        logs = [x for x in receipt.logs if web3.to_hex(x['topics'][0]) in collect_signatures]
        # Has to be at least 2, Collect (pool) and Collect (NFT)
        if len(logs) > 1:
            # Handle collect events
            messages.append(handle_events(logs=logs))

        # Is it a Mint event?
        logs = [x for x in receipt.logs if web3.to_hex(x['topics'][0]) in mint_signatures]
        # Has to be at least 2, Mint and IncreaseLiquidity
        if len(logs) > 1:
            # Handle mint events
            messages.append(handle_events(logs=logs))

        # Is it a Swap event?
        logs = [x for x in receipt.logs if web3.to_hex(x['topics'][0]) == swap_signature]
        if logs:
            # Handle swap events
            messages.append(handle_events(logs=logs, swap_signature=swap_signature))

        # Messages may contain multiple lists; flatten them
        messages = [item for x in messages for item in x]
        # Remove any None values
        messages = [x for x in messages if x is not None]
        if messages:
            action.append(tx)
            # Calculate the timestamp of the transaction
            rec = web3.eth.get_transaction(transaction_hash=tx)
            action.append(web3.eth.get_block(block_identifier=rec.blockNumber).timestamp)
            # Join each message by new line char
            action.append(f'{os.linesep}'.join(messages))
            action_list.append(action)
    return action_list

The handle_event method constructs the appropriate transaction action event string, including the Token ID when applicable. It also filters out events where both amount values are zero. Here is the code for the handle_event method:

def handle_events(logs:list, swap_signature:str=None) -> list:
    # List of responses
    resp_list = []
    # Saves a pool event for NFT event to access it
    stack = []
    for log in logs:
        key = web3.to_hex(log['topics'][0])
        if key in pool_signatures:
            event_name = pool_signatures[key]
            # Process log using pool contract
            processed_log = pool_contract.events[event_name]().process_log(log)
            # Check for a log event without any transaction value for both tokens
            if (processed_log['args']['amount0'] == 0) and (processed_log['args']['amount1'] == 0):
                # Continue with the next log event
                continue
            # Add it to the stack to retrieve when we arrive at the NFT event
            stack.append(event_action(event_name=event_name, processed_log=processed_log))
        elif key in nft_signatures:
            event_name = nft_signatures[key]
            # Process log using nft contract
            processed_log = nftmgr_contract.events[event_name]().process_log(log)
            # Check for a log event without any transaction value for both tokens
            if (processed_log['args']['amount0'] == 0) and (processed_log['args']['amount1'] == 0):
                # Contune with the next log event
                continue
            else:
                # Token id + the message from the pool contract
                token_id = processed_log['args']['tokenId']
                resp_list.append(f'{stack.pop()} Token ID {token_id}')
        elif key == swap_signature:
            event_name = 'Swap'
            processed_log = pool_contract.events[event_name]().process_log(log)
            # There is no token associated with Swap. Just add this to the response
            resp_list.append(event_action(event_name=event_name, processed_log=processed_log))
        else:
            # Ignore as this event as it is not in our signatures
            pass
    return resp_list

As shown below, the handle_event method gets the transaction action string using the event_action method.

def event_action(event_name:str, processed_log:dict) -> str:
    """ Return the action event for an event
    Parameters:
    event_name : str
        event name
    processed_log : dict
        dictionary of processed logs

    Returns:
    str
        An action event as a string
    """
    token_pair = get_token_pair_info(processed_log=processed_log)

    amount0 = token_pair['token0']['amount']
    amount1 = token_pair['token1']['amount']

    symbol0 = token_pair['token0']['symbol']
    symbol1 = token_pair['token1']['symbol']

    price0 = token_pair['token0']['price']
    price1 = token_pair['token1']['price']

    if event_name == 'Burn':
        return(
            f'Remove {amount0:,} (${round((amount0 * price0),2):,.2f}) '
            f'{symbol0} and {amount1:,} (${round((amount1 * price1),2):,.2f}) {symbol1} Liquidity from Uniswap v3')
    elif event_name == 'Collect':
        return(
            f'Collect {amount0:,} (${round((amount0 * price0),2):,.2f}) '
            f'{symbol0} and {amount1:,} (${round((amount1 * price1),2):,.2f}) {symbol1} from Uniswap v3')        
    elif event_name == 'Mint':
        return(
        f'Add {amount0:,} (${round((amount0 * price0),2):,.2f}) '
        f'{symbol0} and {amount1:,} (${round((amount1 * price1),2):,.2f}) {symbol1} Liquidity to Uniswap v3')
    else:            
        if amount0 < 0:
            rate1 = (amount0 * -1) / amount1
            return(
                f'Swapped {amount1:,} (${round((amount1 * price1),2):,.2f}) '
                f'{symbol1} [@ {rate1} {symbol1} per {symbol0}] for {(amount0 * -1)} '
                f'(${round(((amount0 * -1) * price0),2):,.2f}) {symbol0} on Uniswap v3')
        else:
            rate0 = amount1 / (amount0 * -1)
            return(
                f'Swapped {amount0:,} (${round((amount0 * price0),2):,.2f}) '
                f'{symbol0} [@ {rate0} {symbol0} per {symbol1}] for {(amount1 * -1)} '
                f'(${round(((amount1 * -1) * price1),2):,.2f}) {symbol1} on Uniswap v3')

The get_token_pair_info method gathers token data, primarily obtaining some information directly from token contracts and other details from processed logs. It uses Coingecko pricing API for token prices. The notebook contains the code for the get_token_pair_info method and the Coingecko API call. Note that we’re caching prices in the Coingecko method using the lru_cache decorator. This directive might need removing to get real-time prices.

Display

We first convert the data into a DataFrame to sort the transactions by timestamp and then format the timestamps into a human-readable format. Next, we shorten the transaction hashes to make room for displaying transaction events more clearly. Finally, we add a title to the table that incorporates the first and last four characters of the wallet address, along with the pool tokens and the associated fee. The following code block performs these operations:

actions = get_transaction_actions(transactions=txlogs)
# Create DF from list of actions
df = pd.DataFrame(data=actions, columns=['tx', 'timestamp', 'details'])
# Sort by timestamp
df = df.sort_values(by=['timestamp'])
# Convert unix timestamp
df['timestamp'] = df['timestamp'].apply(lambda x: time.strftime('%Y-%m-%d %H:%M:%S', time.gmtime(x)))
# Truncate tx for formatting
df['tx'] = df['tx'].apply(lambda x: '{first_part} ... {last_part}'.format(first_part=x[:6], last_part=x[-6:]))

# Header
print('LP events for the wallet {first_part} ... {last_part} - (Uniswap Pool {symbol0}/{symbol1} {fee}%)'.\
    format(first_part=WALLET_ADDRESS[:4], last_part=WALLET_ADDRESS[-4:], symbol0=token0.functions.symbol.call(),
           symbol1=token1.functions.symbol.call(), fee=FEE_TIER/10000))
# Crate a tabulate table
table = tabulate(df.values.tolist(), headers=['Tx Hash', 'Time', 'Details'], tablefmt="grid",
                 maxcolwidths=[20, 10, None])
print(table)

First few transaction action events for the wallet

As shown above, we have also successfully included the Token ID for each relevant event action.

Conclusion

The first article demonstrated how to extract and analyze LP positions from Uniswap v3 using web3py, focusing on key activities like adding or removing liquidity. The second article enhanced the analysis by calculating the US Dollar value of LP positions, improving comprehension of their financial significance. This article presented LP transaction actions in a format similar to Etherscan’s Transaction Action section, including Token IDs for better user comprehension and easier interpretation.

If you found this guide helpful, please consider leaving a comment or giving it applause. Your support directly contributes to the development of more detailed and insightful educational resources for the Web3 community.

Thank you!

References

• Ethereum Transaction Log Analysis: Uncovering LP Holder Actions
• Ethereum Transaction Log Analysis: Uncovering LP Holder Actions Using GraphQL
• Understanding Ethereum Raw Transactions: Mapping Smart Contract Functions
• The Uniswap v3 Smart Contracts