web3: How to be notified when address performs transaction with contract?

Question

Using python web3, is it possible to listen for when a given target address uses a specific contract?

For context: I want to be able to know every time a wallet purchases/sells tokens via Pancakeswap's router contract (using swapExactEthForTokens or swapExactTokensForEth).

Currently I am using bscscan's api to query for "token transfer events" via one of their APIs. But this is very limited since I am dependent on a third party's API and consequently have to use rate limited repeated http polling to check for a new event every t time interval.

I know something of the sort is possible via websockets and subscriptions, but I have been unable to find an example similar to mine using web3.py v6.

If anybody has any suggestions it would be greatly appreciated!

Answer 1

You can use web3 to contantly listen the transfer event of specific token and match the from or to for the specific wallet. Working with web3 is almost same in all programing language. Eg. in python

         async def log_loop(event_filter, poll_interval):
            while True:
                for Transfer in event_filter.get_new_entries():
                    handle_event(Transfer)
                await asyncio.sleep(poll_interval)
         
       
       
         def main():
            event_filter = contract.events.Transfer.createFilter(fromBlock='latest')
            #block_filter = web3.eth.filter('latest')
            # tx_filter = web3.eth.filter('pending')
            loop = asyncio.get_event_loop()
            try:
                loop.run_until_complete(
                    asyncio.gather(
                        log_loop(event_filter, 2)))
                        # log_loop(block_filter, 2),
                        # log_loop(tx_filter, 2)))
            finally:
                # close loop to free up system resources
                loop.close()