Use web3.py to get ERC-20 transfer logs for a specific account.

README.md

This code uses the eth_getLogs method to retrieve logs. Logs are generated when an event is emitted from a smart contract. We look at the transfer event from ERC-20 contracts in this case.

Here is how logs work:

The transfer event in the ERC-20 contract looks like this:

event Transfer(address indexed from, address indexed to, uint value);

The indexed parameters will be part of the topics, while the non-indexed (value in this case) will be part of data. When a transfer happens, the contract emits the event with the following:

• The address that made the transfer
• The address that received the transfer
• The amount transferred

We create a logs filter object in the Python code using the ERC-20 contract address, the range of blocks to get the logs from, and the topics.

The topics essentially define what we want to retrieve.

The first element of the topics is always the event signature, which is the keccak hash of the event name and the kind of parameter. In this case, it is the following:

Transfer(address,address,uint256)

Which becomes:

0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef

The other element of the topic is the first indexed parameter (the from address in this case), and it has to be encoded as a 32 bytes element. That's why all of the zeros are in front.

We can talk more about how to encode the elements if you want, but the TL;DR is that this filter tracks all of the USDT transfers made by this address 0x8c8d7c46219d9205f056f28fee5950ad564d7465.

I added some code to make the response more legible so that I can explain. The response will look like this:

address: 0xdAC17F958D2ee523a2206206994597C13D831ec7
topics: [HexBytes('0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef'), HexBytes('0x0000000000000000000000008c8d7c46219d9205f056f28fee5950ad564d7465'), HexBytes('0x000000000000000000000000c1e076085fdfca8e7faee36be340b7335330a685')]
data: 0x000000000000000000000000000000000000000000000000000000000ec481a6
blockNumber: 16878247
transactionHash: b'\xb1\x03\xfak6\xe5C\xac\x1a\x05\xc9\xeb\xdd\xc8\xceS\xd9\xf0\xe4+J \xa4\xd3\xfc\xcd\xc9\xc0\x8b\xbd\xbd6'
transactionIndex: 149
blockHash: b'\xb43~\xdd\xc7\x81\xfd\xcf\xd9?\x0e\xc6\xd8\x8d\xc6q^\x03\xdc\xb9=\xee\xab\x80\x112x\x08\xf2\xd6\xd1z'
logIndex: 336
removed: False

The data that interest you is in the topics and data field.

As we said, topics tell you the kind of event and indexed parameters. So, in this case, it shows that address 0x8c8d7c46219d9205f056f28fee5950ad564d7465 sent some USDT to address 0xc1e076085fdfca8e7faee36be340b7335330a685. The amount of tokens is emitted by the event is and is not indexed, so it will be in the data field. This follows the same principle; it's a 32-byte element, so you must remove the zeros before the data. In this case, the amount transferred is ec481a6, which equals 247759270. USDT has 6 decimals, so that becomes 247.759270 USDT.

You can also see the log emitted on Etherscan: https://etherscan.io/tx/0xb103fa6b36e543ac1a05c9ebddc8ce53d9f0e42b4a20a4d3fccdc9c08bbdbd36#eventlog

You can edit the erc_20_contract variable to the smart contract you need to track and the second topic for the specific address. The chain will also have all of the previous transfers saved, but I recommend querying a max of 2000/3000 blocks at the time.

blockReceipt.py

# Make the JSON-RPC request and parse the response
response = web3.provider.make_request('eth_getBlockReceipts', [12160863])
receipts = response['result']
#pprint(receipts)

# Extract the transaction hash and log data for each receipt
log_list = []

erc_20_contract = "0x767FE9EDC9E0dF98E07454847909b5E959D7ca0E" # Illuvium (ILV) address

# Extract the log data for the target address
for receipt in receipts:
    log_data = receipt['logs']
    for log in log_data:
        if log['address'] == erc_20_contract.lower():
            log_dict = {}
            log_dict['tx_hash'] = log['transactionHash']
            log_dict['address'] = log['address']
            log_dict['topics'] = log['topics']
            log_dict['data'] = log['data']

            log_list.append(log_dict)

pprint(log_list)

logs.py

from web3 import Web3

node_url="YOUR_NODE"

# Create the node connection
web3 = Web3(Web3.HTTPProvider(node_url))

latest_block = web3.eth.blockNumber

# Address of the smart contract to track (USDT in this example)
erc_20_contract = "0xdAC17F958D2ee523a2206206994597C13D831ec7"
checksum_address = web3.toChecksumAddress(erc_20_contract)

# The topics define which event to track; the ERC-20 transfer in this case
topics = [
    '0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef', # event signature
    '0x0000000000000000000000008c8d7c46219d9205f056f28fee5950ad564d7465'  # address to track
    ]

# the longs to filter
filter = {
    'fromBlock': latest_block - 100,
    'toBlock': 'latest',
    'address': checksum_address,
    'topics': topics
}

raw_logs = web3.eth.get_logs(filter)

for log in raw_logs:
    # Loop over the key-value pairs of the AttributeDict object
    for key, value in log.items():
        print(f"{key}: {value}")

logs_date.py

from web3 import Web3
from datetime import datetime
import json
import requests
from pprint import pprint

node_url = "YOUR_NODE" # Erigon
API_KEY = 'API_KEY' # Etherscan API key

# Create the node connection
web3 = Web3(Web3.HTTPProvider(node_url))

# Verify if the connection is successful
if web3.isConnected():
    print("-" * 50)
    print("Connection Successful")
    print("-" * 50)
else:
    print("Connection Failed")

def get_unix_timestamp(date_string):
    # Convert date string to datetime object
    date_obj = datetime.strptime(date_string, '%Y-%m-%d %H:%M:%S')

    # Convert datetime object to Unix timestamp
    unix_timestamp = int(date_obj.timestamp())

    # Return Unix timestamp
    return unix_timestamp


def get_block_number(timestamp, api_key):
    url = 'https://api.etherscan.io/api'
    payload = {
        'module': 'block',
        'action': 'getblocknobytime',
        'timestamp': timestamp,
        'closest': 'before',
        'apikey': api_key
    }

    response = requests.get(url, params=payload)
    content = response.content

    response_dict = json.loads(content)
    block_number = response_dict['result']

    return block_number

# Function to encode a hex string as an Ethereum event topic
def encode_topic(value):
    add_padding = value[2:].zfill(64) 
    encoded_topic = '0x' + add_padding
    return encoded_topic

# Function to decode an encoded Ethereum event topic back into its original hex string
def decode_topic(encoded_value):
    decoded_value = '0x' + encoded_value[2:].lstrip('0')
    return decoded_value

# Function to convert a hex string to a decimal integer
def to_decimal(hex_string):
    bytes_value = Web3.toBytes(hexstr=hex_string)
    decimal_value = int.from_bytes(bytes_value, byteorder='big')
    return decimal_value 

# Function to convert a number with decimals to an integer value
# Note: this function is specifically for the USDT token, which has 6 decimal places
def convert_to_usdt(number_with_decimals):
    number_without_decimals = number_with_decimals // (10**6)
    return number_without_decimals

# Convert Wei to Ether
def wei_to_eth(wei):
    eth_value = Web3.fromWei(wei, 'ether')
    return eth_value

def main():
    
    # This section takes the contract address and encodes the topics
    # Get the logs based on ERC-20 address and timestamp

    # Address of the smart contract to track (UNI in this example)
    erc_20_contract = "0x1f9840a85d5aF5bf1D1762F925BDADdC4201F984"
    checksum_address = web3.toChecksumAddress(erc_20_contract)

    # Encode a wallet address. This in case you want to track the transfers only from this address.
    address_to_track = '0x95b564f3b3bae3f206aa418667ba000afafacc8a'
    encoded_address = encode_topic(address_to_track)

    # The topics define which event to track; the ERC-20 transfer in this case
    logs_topics = [
        '0xddf252ad1be2c89b69c2b068fc378daa952ba7f163c4a11628f55a4df523b3ef',  # event signature
        # encoded_address,  # In case you want to only get transfers from this address
    ]

    
    result = []

    days = 3   # How many days in the past to get
    end_day = 22 # The limit of the search. In this case today

    while days > 0:
        print('Getting logs...')
        end_date = f"2023-03-{end_day} 00:00:00"
        end_timestamp = get_unix_timestamp(end_date)
        end_block = get_block_number(end_timestamp, API_KEY)
        #print(f'End block: {end_block}')

        start_date = f"2023-03-{end_day - 1} 00:00:00"
        timestamp = get_unix_timestamp(start_date)
        start_block = get_block_number(timestamp, API_KEY)
        #print(f'Start block: {start_block}')        

        filter = {
            'fromBlock': start_block,
            'toBlock': end_block,
            'address': checksum_address,
            'topics': logs_topics
        }

        raw_logs = web3.eth.get_logs(filter)

        for log in raw_logs:
            extracted_data = {key: log[key] for key in ['address', 'topics', 'data', 'transactionHash'] if key in log}

            address = extracted_data.get('address')
            topics = extracted_data.get('topics')
            data = extracted_data.get('data')
            transaction_hash = extracted_data.get('transactionHash').hex()

            if topics:
                topics_str = [topic.hex() for topic in topics]

            from_address = decode_topic(topics_str[1])
            to_address = decode_topic(topics_str[2])
            value_transferred = decode_topic(data)
            value_hex = to_decimal(value_transferred)
            value_converted = wei_to_eth(value_hex)

            log_entry = {
                'from_address': from_address,
                'to_address': to_address,
                'value_converted': float(value_converted),
                'transaction_hash': transaction_hash
            }

            result.append(log_entry)
            

        end_day -= 1
        days -= 1
        print(f'Retrieved logs between block {start_block} and block {end_block}')


    # Print the result list containing dictionaries
    pprint(result)

if __name__ == "__main__":
    main()

utils.py

from web3 import Web3

# Function to encode a hex string as an Ethereum event topic
def encode_topic(value):
    add_padding = value[2:].zfill(64) 
    encoded_topic = '0x' + add_padding
    return encoded_topic

# Function to decode an encoded Ethereum event topic back into its original hex string
def decode_topic(encoded_value):
    decoded_value = '0x' + encoded_value[2:].lstrip('0')
    return decoded_value

# Function to convert a hex string to a decimal integer
def to_decimal(hex_string):
    bytes_value = Web3.toBytes(hexstr=hex_string)
    decimal_value = int.from_bytes(bytes_value, byteorder='big')
    return decimal_value 

# Function to convert a number with decimals to an integer value
# Note: this function is specifically for the USDT token, which has 6 decimal places
def convert_to_usdt(number_with_decimals):
    number_without_decimals = number_with_decimals // (10**6)
    return number_without_decimals

# Convert Wei to Ether
def wei_to_eth(wei):
    eth_value = Web3.fromWei(wei, 'ether')
    return eth_value
    
# Encode topics
value = '0xCbd8bfd328c77A2781F58f5fBA237670Efdd7201'
encoded_topic = encode_topic(value)
print(encoded_topic)  # '0x000000000000000000000000cbd8bfd328c77a2781f58f5fba237670efdd7201'

# Decode address topic
encoded_value = '0x000000000000000000000000feb036b40993e80834c3e150f1528e6766cebfbe'
decoded_address = decode_topic(encoded_value)
print(decoded_address)  

# Decode data topic
encoded_value = '0x00000000000000000000000000000000000000000000000abc145cd856da0000'
decoded_value = decode_topic(encoded_value)
print(decoded_value)  

# Convert value to decimal
decimal_amount = to_decimal(decoded_value)
eth_value = wei_to_eth(decimal_amount)
print(eth_value)  

# Convert if using USDT
usdt_logs = 1000000000 # Value converted from data
usdt = convert_to_usdt(usdt_logs)
print(usdt)