Script for Uniswap Universal Router Command Statistics

ur_command_statistics.py

import asyncio
import os
import time

from web3 import AsyncWeb3, AsyncHTTPProvider

from uniswap_universal_router_decoder import RouterCodec


block_start = 18473543
block_end = 18687850

ur_address = AsyncWeb3.to_checksum_address("0x3fC91A3afd70395Cd496C647d5a6CC9D4B2b7FAD")

async_w3_instances = (    
    AsyncWeb3(AsyncHTTPProvider(os.environ["PROVIDER_1"])),
    AsyncWeb3(AsyncHTTPProvider(os.environ["PROVIDER_2"])),
    AsyncWeb3(AsyncHTTPProvider(os.environ["PROVIDER_3"])),
)

counts = {"block": 0, "trx": 0}
total_blocks = block_end - block_start
codec = RouterCodec()
results = {}
tasks = set()
trx_exceptions = []


def w3_gen():
    while True:
        i = 0
        while i < len(async_w3_instances):
            yield async_w3_instances[i]
            i += 1


w3 = w3_gen()


def process_transaction(trx):
    try:
        if len(trx.input) > 2:
            counts["trx"] += 1
            decoded_trx_input = codec.decode.function_input(trx.input)
            for c in decoded_trx_input[1]["commands"]:
                results[c] = results.get(c, 0) + 1
    except Exception:
        trx_exceptions.append(trx.hash.hex())


async def process_block(block_number):
    counts["block"] += 1
    if counts["block"] % 100 == 0:
        print("block: ", block_number, f"{counts['block']}/{total_blocks}")
        print("Partial results: ", results)
    block = await next(w3).eth.get_block(block_number, True)
    for trx in block.transactions:
        if trx.to == ur_address:
            process_transaction(trx)


async def main():
    loop = asyncio.get_event_loop()
    for block_number in range(block_start, block_end + 1):
        task = loop.create_task(process_block(block_number))
        tasks.add(task)
        task.add_done_callback(tasks.discard)
        await asyncio.sleep(0.025)

    while len(tasks) > 0:
        print(f"Waiting for {len(tasks)} tasks")
        await asyncio.sleep(0.2)


if __name__ == "__main__":
    start = time.perf_counter()
    asyncio.run(main())
    print(f"Results: {results}")
    end = time.perf_counter()
    print("Total time:", end - start)
    print("Processed Transactions:", counts["trx"])
    print(f"Transactions in error ({len(trx_exceptions)}):", trx_exceptions)

See post here