Klaytn: Contract Sizzle 100

Klaytn is an EVM-compatible protocol with a few modifications. Check the Klaytn docs for a full break-down on the compatibility: Ethereum Compatibility.

In this tutorial, we'll build a quick Python project called Contract Sizzle 100. Contract Sizzle 100 prints the top 3 hottest contracts on the Klaytn Mainnet (aka Cypress Network) over the past 100 blocks.

The way it works is very simple: the script ingests each new incoming block from the Klaytn Mainnet, extracts all contract interactions, counts them and prints the top3 contracts that had the most interactions.

Prerequisites

Chainstack account to deploy a Klaytn Mainnet node
web3.py

Step-by-step

Get a Klaytn node

Create the script

A few details on the implementation.

How do you actually identify the contract addresses? One easy way is to:

Retrieve off the network each block with all transactions in the block.
Extract the to: address.
Check if the extracted address is a contract by doing an eth_getCode to the address.

This process in Python, however, can be slow, so let's optimize it a bit:

After doing an eth_getCode to an extracted address, cache the results so that if we get this address again, we don't do a eth_getCode to it as we already know this is a contract address.
Multithread the script by using our tutorial Mastering multithreading in Python for Web3 requests: A comprehensive guide.

Here's the final script:

from web3 import Web3
from collections import defaultdict
import heapq
from concurrent.futures import ThreadPoolExecutor

# Connect to a Klaytn node
w3 = Web3(Web3.HTTPProvider('CHAINSTACK_NODE'))

# Data structure to hold the count of interactions per contract
contract_interactions = defaultdict(int)

# Priority queue to maintain top 3 contracts
top_contracts = []

# Cache for storing contract check results
is_contract_cache = {}

def is_contract(address):
    if address not in is_contract_cache:
        code = w3.eth.get_code(address)
        is_contract = code != '0x'
        is_contract_cache[address] = is_contract
    return is_contract_cache[address]

def process_block(block_number):
    print(f"Processing block {block_number}")
    block = w3.eth.get_block(block_number, full_transactions=True)
    for tx in block.transactions:
        if tx.to and is_contract(tx.to):
            contract_interactions[tx.to] += 1

# Main loop to process 100 blocks
def main():
    latest_block = w3.eth.block_number
    print("Starting to process blocks...")
    with ThreadPoolExecutor(max_workers=10) as executor:
        executor.map(process_block, range(latest_block - 100, latest_block))
    
    # Identify top 3 contracts
    for contract, interactions in contract_interactions.items():
        heapq.heappush(top_contracts, (interactions, contract))
        if len(top_contracts) > 3:
            heapq.heappop(top_contracts)

    # Print top 3 contracts
    print("Top contracts:", top_contracts)
    while top_contracts:
        interactions, contract = heapq.heappop(top_contracts)
        print(f'Contract {contract} had {interactions} interactions')

if __name__ == '__main__':
    try:
        main()
    except Exception as e:
        print(f"An error occurred: {e}")

where

CHAINSTACK_NODE — your Klaytn node deployed with Chainstack
max_workers=10 — set to however parallel threads you feel is reasonable. Make sure don't hit the limits: Limits.

Conclusion

This tutorial guided you through creating a basic setup to live-track the hottest contract on the Klaytn Network. There are many fun ways you can use it as basis to build upon — from passing the data to a Twitter bot account to setting up your own alerts or bot flow.