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.
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.
from web3 import Web3from collections import defaultdictimport heapqfrom concurrent.futures import ThreadPoolExecutor# Connect to a Klaytn nodew3 = Web3(Web3.HTTPProvider('CHAINSTACK_NODE'))# Data structure to hold the count of interactions per contractcontract_interactions = defaultdict(int)# Priority queue to maintain top 3 contractstop_contracts = []# Cache for storing contract check resultsis_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 blocksdef 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.
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.
