Skip to content

Instantly share code, notes, and snippets.

@normanlmfung
Created November 16, 2022 05:11
Show Gist options
  • Save normanlmfung/8a49197fdd26d4a5b145f409f116dcc4 to your computer and use it in GitHub Desktop.
Save normanlmfung/8a49197fdd26d4a5b145f409f116dcc4 to your computer and use it in GitHub Desktop.
aptos query transactions under given wallets
import time
import json
import requests
from datetime import datetime
# https://fullnode.mainnet.aptoslabs.com/v1/spec#/operations/get_account_transactions
all_transactions = []
for wallet in wallets:
try:
url = f"https://fullnode.mainnet.aptoslabs.com/v1/accounts/{wallet.address()}/transactions"
method="get"
rsp = requests.request(method, url)
transactions = json.loads(rsp.text)
for transaction in transactions:
ts_sec = int(transaction['timestamp']) / 1_000_000 # Aptos timestamp in micro seconds
dt_utc = datetime.utcfromtimestamp(ts_sec)
# This is for easy grouping/pivot table
year = str(dt_utc.year)
month = f'0{str(dt_utc.month)}' if len(str(dt_utc.month))==1 else str(dt_utc.month)
dt_str = f"{year}-{month}"
payload_function = None
payload_type_args = None
payload_arguments = None
if 'payload' in transaction:
payload_function = transaction['payload']['function'] if 'function' in transaction['payload'] else None
payload_type_args = transaction['payload']['type_arguments'] if 'type_arguments' in transaction['payload'] else None
payload_arguments = transaction['payload']['arguments'] if 'arguments' in transaction['payload'] else None
all_transactions.append(
{
'address' : wallet.address().hex(),
'hash' : transaction['hash'],
'dt_utc' : dt_utc,
'year' : dt_utc.year,
'month' : dt_utc.month,
'year-month' : dt_str,
'day' : dt_utc.day,
'ts_ms' : ts_sec * 1000,
'sender' : transaction['sender'], # Should be same as wallet address!?!
'success' : transaction['success'],
'gas_used_apt' : int(transaction['gas_used']) / 1_000_000, # 82,964 Gas Units, divided by 1mn to get gas in APT
'gas_unit_price' : transaction['gas_unit_price'],
'payload_function' : payload_function,
'payload_type_args' : payload_type_args,
'payload_arguments' : payload_arguments
}
)
except Exception as error:
print(f"Error while processing {wallet.address().hex()}, error: {error}")
time.sleep(2) # This is 2 seconds. If you start reducing this, you may get rate limit error from Aptos.
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment