Analyze ERC-20 token data on Polygon Mainnet with AMB Access

Introduction

In this tutorial, you will learn how to analyze the stablecoin, Tether (USDT), on Polygon Mainnet using Amazon Managed Blockchain (AMB) Access and web3.py. It covers extracting historical transfer events, loading them into a Pandas DataFrame, performing calculations, and using Seaborn to gains insights by visualizing data trends.

The visualizations that you create will be similar to the following:

Prerequisites

• Installation of Python
• A text editor (such as Visual Studio Code)
• Installation of the AWS Command Line Interface (CLI).
  • Run the command aws configure to set the variables for your IAM User’s Access Key ID, Secret Access Key, and Region.
    • Ensure that this User has the appropriate IAM permissions for AMB Access Polygon. For this tutorial, you can use the AmazonManagedBlockchainFullAccess managed policy.

Step 1: Setting up the project

Create a new directory for your project and change into it by running the following command:

mkdir erc20Analysis && cd erc20Analysis

Run the following command to install the necessary dependencies:

pip install seaborn web3 python-dotenv boto3

Step 2: AMB Access Polygon Endpoint Configuration

The following script automates the following tasks:

• Retrieval or creation of an AMB (Amazon Managed Blockchain) Polygon Mainnet Accessor token.
  • The script first checks for any existing Polygon Mainnet Accessor tokens in your AWS account. If an available token is found, it will be reused. Otherwise, a new Accessor token will be created.
• Configuration of an AMB Access Polygon Mainnet endpoint with your Accessor token.
  • Your configured endpoint will allow you to make JSON-RPC calls to Polygon Mainnet with web3.py.

Warning: Never share or expose your Accessor token publicly. It is important to note that this code should not be used in production environments, as it poses a security risk.

Create a new file (e.g init.py) in the root of your directory and copy the following code:

import boto3

network = "POLYGON_MAINNET"

def get_existing_accessor(client):
    try:
        accessors = client.list_accessors(NetworkType=network)['Accessors']
        return next((client.get_accessor(AccessorId=acc['Id'])['Accessor'] for acc in accessors if acc['Status'] == 'AVAILABLE'), None)
    except Exception as error:
        print('Error retrieving existing Accessor:', error)
        raise

def get_or_create_accessor():
    client = boto3.client('managedblockchain')
    existing_accessor = get_existing_accessor(client)

    if existing_accessor:
        print('Using existing Accessor token.')
        return existing_accessor['BillingToken']
    else:
        print('Creating a new Accessor token.')
    try:
        response = client.create_accessor(AccessorType='BILLING_TOKEN', NetworkType = network)
        return response['BillingToken']
    except Exception as error:
        print('Error creating Accessor token:', error)
        
def main():
    print('Creating or retrieving AMB Access Polygon Accessor token...')
    billingToken = get_or_create_accessor()
    access_endpoint = f"https://mainnet.polygon.managedblockchain.us-east-1.amazonaws.com?billingtoken={billingToken}"
    data_to_save = f"AMB_ACCESS_POLYGON_MAINNET_ENDPOINT={access_endpoint}"
    with open('.env', 'w') as file:
        file.write(data_to_save)
    print('Accessor token created or retrieved. Details saved to .env file.')
    
if __name__ == '__main__':
    main()

Execute the script by running:

python init.py

Step 3: Write the data analysis script

The script establishes a connection to AMB Access Polygon and retrieves USDT transfer events from the most recent 1,000 blocks. It processes this data using a Pandas DataFrame and leverages the Seaborn library to create insightful visualizations. Additionally, a series of calculations are performed and the results are displayed in the terminal.

Create a new file (e.g analyze.py) and copy the following code:

import os, json, warnings, seaborn as sns, matplotlib.pyplot as plt
from web3 import Web3
from dotenv import load_dotenv
load_dotenv()

w3 = Web3(Web3.HTTPProvider(os.getenv('AMB_ACCESS_POLYGON_MAINNET_ENDPOINT')))
contract = w3.eth.contract(address='0xc2132D05D31c914a87C6611C10748AEb04B58e8F', abi=json.loads('[{"anonymous":false,"inputs":[{"indexed":true,"internalType":"address","name":"from","type":"address"},{"indexed":true,"internalType":"address","name":"to","type":"address"},{"indexed":false,"internalType":"uint256","name":"value","type":"uint256"}],"name":"Transfer","type":"event"},{"inputs":[],"name":"decimals","outputs":[{"internalType":"uint8","name":"","type":"uint8"}],"stateMutability":"view","type":"function"}]'))
token_decimals = contract.functions.decimals().call()

def fetch_transfer_events(from_block, to_block):
    try:
        return [{"from": e['args']['from'], "to": e['args']['to'], "blockNumber": e['blockNumber'], "value": e['args']['value'] / 10**token_decimals} for e in contract.events.Transfer().get_logs(fromBlock=from_block, toBlock=to_block)]
    except Exception as e:
        print(f"An error occurred: {e}")
        return []

latest_block, block_range, block_chunk_size = w3.eth.block_number, 1000, 10
print(f"Fetching Transfer events from block {latest_block-block_range:,} to {latest_block:,}\n")
data_list = [event for block in range(latest_block - block_range, latest_block, block_chunk_size) for event in fetch_transfer_events(block, min(block + block_chunk_size - 1, latest_block))]

warnings.filterwarnings("ignore", "\nPyarrow", DeprecationWarning) #Filter Pyarrow dependency warning message
import pandas as pd
df = pd.DataFrame(data_list)

print("Top 3 Block Numbers by Count of Transfer Events:\n", df['blockNumber'].value_counts().head(3), "\n")
print("Top 3 Block Numbers by Value Transferred:\n", df.groupby('blockNumber')['value'].sum().nlargest(3), "\n")
print(f"Total USDT Value Transferred:\n ${df['value'].sum():,.2f}\n")

fig, axs = plt.subplots(1, 2, figsize=(14, 7))
ticks = range(int(df['blockNumber'].min()), int(df['blockNumber'].max()), 100)

sns.histplot(df, x="blockNumber", bins=20, kde=True, ax=axs[0]).set(title="Distribution of Transfer Events", xlabel="Block Number", ylabel="Number of Transfer Events")
axs[0].set_xticks(ticks); axs[0].set_xticklabels([str(label) for label in ticks], rotation=45)
sns.histplot(data=df.groupby('blockNumber')['value'].sum().reset_index(), x='blockNumber', weights='value', bins=20, ax=axs[1]).set(title='Distribution of USDT Value Transferred', xlabel='Block Number', ylabel='Value (USDT)')
axs[1].set_xticks(ticks); axs[1].set_xticklabels([str(label) for label in ticks], rotation=45)

plt.tight_layout()
plt.show()

Execute the script by running:

python analyze.py

Conclusion

You have utilized AMB Access Polygon to fetch transfer event data for the ERC-20 token, Tether (USDT). You then performed analyses on the retrieved data by utilizing the Pandas and Seaborn libraries. Remember to prioritize security, especially with sensitive data such as your Amazon Managed Blockchain Accessor token.

Please leave a comment below if you have any questions. For further information on AMB Access, you can refer to the documentation.