Web3.py: How to Monitor ETH Transfers to an Address

Using Web3.py, we can check every new block for any new Ether transactions to our Ethereum account.

**Disclaimer: After rechecking the script, it appears that it takes roughly 0.17 seconds to check a transaction (on my machine). This would mean that for any block it encounters that exceeds 70 transactions, the script would fail to keep up with the current 12 second Ethereum average block time and consequently miss blocks. A solution for this would be not querying the network for the latest block all the time, but storing the latest checked block number and always querying for the next one.

If you’d like to use subscriptions/websockets instead, read about Subscribing to Pending Ethereum Transactions in Python.**

Setting Up Our Project

First off, we need to have the Web3.py library installed on our machine:

pip install web3

Then we can import the Web3.py library in our code, as well as time so we can query the Ethereum blockchain for a new block every set amount of seconds:

from web3 import Web3
import time

Connecting to Infura

Let’s connect to the Infura JSON-RPC endpoint and define the address for which we want to monitor incoming transactions. Additionally, feel free to replace mainnet with another network such as goerli if you’d like to use a testnet instead.

infura_url = 'https://mainnet.infura.io/v3/<YOUR_PROJECT_ID>'
account = '<YOUR_PUBLIC_ADDRESS>'
web3 = Web3(Web3.HTTPProvider(infura_url))

Make sure to replace <YOUR_PROJECT_ID> with your Infura project ID and <YOUR_PUBLIC_ADDRESS> with the Ethereum account you wish to monitor.

Monitoring New Blocks

We can create a watch() function that will include all of our code that will check for new blocks and its transactions:

def watch():
    while True:
        block = web3.eth.get_block('latest')
        print("Searching in block " + str(block.number))

This will return the latest block number from Infura, after which we can loop through all included transactions:

if block and block.transactions: 
    for transaction in block.transactions: 
        tx_hash = transaction.hex() # the hashes are stored in a hexBytes format
        tx = web3.eth.get_transaction(tx_hash)

Using eth.get_transaction, we retrieve all the details of every transaction included in the new block; we can then check if the recipient matches the account variable we’ve set earlier in our code:

if tx.to != None:
    if tx.to == account:
        print("Transaction found in block {} :".format(block.number))
        print({
            "hash": tx_hash,
            "from": tx["from"],
            "value": web3.fromWei(tx["value"], 'ether')
            })

Finally, outside the if statement, we can add a time.sleep(5) so that our code will wait some time before checking the next latest block, as blocks typically only get mined every 10-20 seconds.

Make sure to call our new watch() function just below the actual method, so our new function will execute upon running our program.

The watch() method in full:

def watch():
    while True:
        block = web3.eth.get_block('latest')
        print("Searching in block " + str(block.number))

        if block and block.transactions: 
            for transaction in block.transactions: 
                tx_hash = transaction.hex() # the hashes are stored in a hexBytes format
                tx = web3.eth.get_transaction(tx_hash)
                if tx.to != None:
                    if tx.to == account:
                        print("Transaction found in block {} :".format(block.number))
                        print({
                            "hash": tx_hash,
                            "from": tx["from"],
                            "value": web3.fromWei(tx["value"], 'ether')
                            })
        time.sleep(5)

watch()

Upon running the script in a terminal, you’ll see that it will print any newly found transactions going to the address we’ve specified:

➜  Python python3 watch.py
Searching in block 15019035
Transaction found in block 15019035 :
{'hash': '0x0f878eb882dfd069c482740df533e0ddef63504d795dcc3c934c3f9a6c159362', 'from': '0x95B564F3B3BaE3f206aa418667bA000AFAFAcc8a', 'value': 0}
Transaction found in block 15019035 :
{'hash': '0xee75ed766e17fef1cae917686b5b73e7c72b2fcf51e1558629b8fe96a7e5a1bd', 'from': '0x9696f59E4d72E237BE84fFD425DCaD154Bf96976', 'value': 0}
Transaction found in block 15019035 :
{'hash': '0x1cddbeff3ac97651f5d7e49e98f4289aeef728e08b05180c75bc04cdc970895d', 'from': '0x21a31Ee1afC51d94C2eFcCAa2092aD1028285549', 'value': 0}
...

Checking for Confirmations

You might often see exchanges wait until a deposit has reached X amount of confirmations before processing the new transaction. The amount of confirmations really comes down to the number of blocks that have passed since the transaction was included on-chain.

Why?

A block’s hash is generated by hashing all of the included transactions + the hash of the previous block (and a nonce). When the next block is mined, the process repeats. If a miner attempted to change the details of a previously mined transaction, every block after that would be invalid, as the hash of every block after would change.

By checking that a transaction has reached X amount of confirmations, the exchange can be confident that this transaction is final and they can process the deposit.

In Code!

We can create a function to determine the amount of confirmations a transaction has, which comes down to getting the latest block number, and subtracting the block number that the transaction was mined in:

def confirmations(tx_hash):
    tx = web3.eth.get_transaction(tx_hash)
    return web3.eth.block_number - tx.blockNumber

Then, we can call it:

print(confirmations("0x0d40d60e118e9e1f61c2baa2252cc5f8b8ed491c885ec35db6fd6cfc8589c1a7"))

Result:

➜  Python python3 watch.py
67341

Complete Code Overview

from web3 import Web3
import time 

infura_url = 'https://mainnet.infura.io/v3/<YOUR_PROJECT_ID>'
account = '<YOUR_PUBLIC_ADDRESS>'
web3 = Web3(Web3.HTTPProvider(infura_url))

def confirmations(tx_hash):
    tx = web3.eth.get_transaction(tx_hash)
    return web3.eth.block_number - tx.blockNumber

def watch():
    while True:
        block = web3.eth.get_block('latest')
        print("Searching in block " + str(block.number))

        if block and block.transactions: 
            for transaction in block.transactions: 
                tx_hash = transaction.hex() # the hashes are stored in a hexBytes format
                tx = web3.eth.get_transaction(tx_hash)
                if tx.to != None:
                    if tx.to == account:
                        print("Transaction found in block {} :".format(block.number))
                        print({
                            "hash": tx_hash,
                            "from": tx["from"],
                            "value": web3.fromWei(tx["value"], 'ether')
                            })

watch()
# print(confirmations("0x0d40d60e118e9e1f61c2baa2252cc5f8b8ed491c885ec35db6fd6cfc8589c1a7"))

Special thanks to @wtzb for helping write this article!

5 Likes