I'm trying to build a web3.py program to take in a given user's address, and return the amount of ETH traded since they added liquidity to a Uniswap exchange. Uniswap is a smart contract on the Ethereum platform.

The Uniswap contract has a method 'AddLiquidity' that allows users to add funds, and I want to find when a given user did this.

I looked through the documentation for how to do this. I'm accessing the blockchain via an infura node, and the docs seemed to say that on infura you can't use a filter.

On the contract documentation, it seemed like I could do this via the events object, but couldnt't get this to work. Think the eventslog may be important but as I say have struggled to get any of it to do what I was hoping.

Please let me know if you need any further info to help me and thanks!


Smart Contract

pragma solidity 0.5.1;

contract Uniswap {

   event Triggered(address indexed user);

   function addLiquidity() external {
       emit Triggered(msg.sender);


Javascript Event Filtering

var sha3HashOfTheEvent = web3.sha3("event Triggered(address)");
var filterOptions = {
    fromBlock: 0,
    toBlock: 'latest',
filter.watch(function(error, result){
       console.log("The event has been triggered.......")

Hope it will help you.

  • I think the author of the question was asking for a Python solution. Mar 12 '20 at 10:19

These would be steps

  • Take Uniswap ABI and construct web3.Contract objects based on it

  • Scan all AddLiquidity events from the block zero to the last mainnet block

  • Build your own database/list of deposits based on these

Here is some similar code to pick up all Transfer events for tokens:

    def scan_chunk(self, start_block, end_block) -> Set[str]:
        """Populate TokenHolderStatus for certain blocks.
        :return: Set of addresses where balance changes between scans.

        mutated_addresses = set()
        token = self.get_token_contract(self.address)

        # Discriminate between ERC-20 transfer and ERC-667
        # The latter is not used anywhere yet AFAIK
        Transfer = token.events.Transfer("from", "to", "value")
        Issued = token.events.Issued("to", "value")
        block_timestamps = {}
        get_block_timestamp = self.get_block_timestamp

        # Cache block timestamps to reduce some RPC overhead
        # Real solution would be smarter models around block
        def get_block_when(block_num):
            if not block_num in block_timestamps:
                block_timestamps[block_num] = get_block_timestamp(block_num)
            return block_timestamps[block_num]

        for event_type in [Issued, Transfer]:

            # events = event_type.createFilter(fromBlock=start_block, toBlock=end_block).get_all_entries()

            events = getLogs(event_type, fromBlock=start_block, toBlock=end_block)

            # AttributeDict({'args': AttributeDict({'from': '0xDE5bC059aA433D72F25846bdFfe96434b406FA85', 'to': '0x0bdcc26C4B8077374ba9DB82164B77d6885b92a6', 'value': 300000000000000000000}), 'event': 'Transfer', 'logIndex': 0, 'transactionIndex': 0, 'transactionHash': HexBytes('0x973eb270e311c23dd6173a9092c9ad4ee8f3fe24627b43c7ad75dc2dadfcbdf9'), 'address': '0x890042E3d93aC10A426c7ac9e96ED6416B0cC616', 'blockHash': HexBytes('0x779f55173414a7c0df0d9fc0ab3fec461a66ceeee0b4058e495d98830c92abf8'), 'blockNumber': 7})
            for e in events:
                idx = e["logIndex"]  # nteger of the log index position in the block. null when its pending log.
                if idx is None:
                    raise RuntimeError("Somehow tried to scan a pending block")

                if e["event"] == "Issued":
                    # New issuances pop up from empty air - mark this specially in the database.
                    # Also some ERC-20 tokens use Transfer from null address to symbolise issuance.
                    from_ = self.TokenScanStatus.NULL_ADDRESS
                    from_ = e["args"]["from"]

                block_when = get_block_when(e["blockNumber"])

                self.create_deltas(e["blockNumber"], block_when, e["transactionHash"].hex(), idx, from_, e["args"]["to"], e["args"]["value"])
                self.logger.debug("Imported %s, token:%s block:%d from:%s to:%s value:%s", e["event"], self.address, e["blockNumber"], from_, e["args"]["to"], e["args"]["value"])


        return mutated_addresses

You can find the full source code here.

If you know the user address before hand you are also most likely be able to filter based on this address, so that you do not need to scan all events. Each Solidity event can have 4 indexed parameters that allow you to filter those events on Ethereum node level, meaning that internal Ethereum node database allows you to query them. However if there are many events, the Ethereum node might not response to these queries really fast and you need to build your own database of events any case.

Your Answer

By clicking “Post Your Answer”, you agree to our terms of service, privacy policy and cookie policy

Not the answer you're looking for? Browse other questions tagged or ask your own question.