Kaia (ex. 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
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:
- 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 aeth_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.
About the author
Updated about 2 months ago