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

Step-by-step

Get a Klaytn node

Log in to your Chainstack account and deploy a node.

Create the script

A few details on the implementation.

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

  1. Retrieve off the network each block with all transactions in the block.
  2. Extract the to: address.
  3. 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:

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.

About the author

Ake

๐Ÿ› ๏ธ Developer Experience Director @ Chainstack
๐Ÿ’ธ Talk to me all things Web3 infrastructure and I'll save you the costs
Ake | Warpcast Ake | GitHub Ake | Twitter Ake | LinkedIN