Uniswap v2 reading swaps real-time (minimal)

This is a minimal example code for showing live swaps happening on Uniswap v2 compatible examples.

  • This example runs on a free Polygon JSON-RPC node.

  • It will print out live trade events for all Uniswap v2 compatible exchanges. This includes QuickSwap, Sushi and couple of others. See the full DEX list here.

  • It demonstrates the chain reorganisation detection and event reader API.

  • See the more complete example

To run:

python scripts/uniswap-v2-swaps-live-minimal.py

Example output:

Swap at block:42,549,528 tx:0x2011d03b4f3d80992339eb6303b0b7b86ec77f629ce7f2508344e739c4536cc7
Swap at block:42,549,528 tx:0x67af6d9d28634747d83f14d48bdc3d56421df7b686055c4519850b97e863291d
Block 42,549,527 is 0x83fd3f8dfd6065fcc3406fed9e81b069a45cf0e823fe4863c89a5e9cef49bdc6
Block 42,549,528 is 0xb29ce58ad7267b5c9906eea32aeacf043965a7223a35e0aa3c495dcdf3815eac
Swap at block:42,549,529 tx:0xa9ba7e61c1bdedf53b657419874d528b4164b9e286c3baf162f20f8d1c428b80
Block 42,549,529 is 0x686ebaa6ac7fa5f32aedbdc05e1352f681072acb46c0c158b779afd8e0fce21f
"""Minimal example of a chain reader with chain reorganisation detection.

.. code-block:: shell

    python scripts/read-uniswap-v2-pairs-and-swaps-live.py


import os
import time

from web3 import HTTPProvider, Web3

from eth_defi.abi import get_contract
from eth_defi.chain import install_chain_middleware
from eth_defi.event_reader.filter import Filter
from eth_defi.event_reader.reader import read_events, LogResult
from eth_defi.event_reader.reorganisation_monitor import JSONRPCReorganisationMonitor

def main():
    json_rpc_url = os.environ.get("JSON_RPC_POLYGON", "https://polygon-rpc.com")
    web3 = Web3(HTTPProvider(json_rpc_url))

    # Get contracts
    Pair = get_contract(web3, "sushi/UniswapV2Pair.json")

    filter = Filter.create_filter(address=None, event_types=[Pair.events.Swap])  # Listen events from any smart contract

    reorg_mon = JSONRPCReorganisationMonitor(web3, check_depth=3)

    # Get the headers of last 5 blocks before starting

    processed_events = set()

    latest_block = reorg_mon.get_last_block_live()

    # Keep reading events as they land
    while True:
        chain_reorg_resolution = reorg_mon.update_chain()
        start, end = chain_reorg_resolution.get_read_range()

        if chain_reorg_resolution.reorg_detected:
            print("Chain reorg warning")

        evt: LogResult
        for evt in read_events(
            # How to uniquely identify EVM logs
            key = evt["blockHash"] + evt["transactionHash"] + evt["logIndex"]

            # The reader may cause duplicate events as the chain tip reorganises
            if key not in processed_events:
                print(f"Swap at block:{evt['blockNumber']:,} tx:{evt['transactionHash']}")

        if end != latest_block:
            for block_num in range(latest_block + 1, end + 1):
                block_data = reorg_mon.get_block_by_number(block_num)
                print(f"Block {block_num:,} is {block_data.block_hash}")

            latest_block = end


if __name__ == "__main__":