simple script to export payment activity for a given wallet address

export_account_activity.py

import requests
import pandas as pd


ACCOUNT = "B58_WALLET_ADDRESS"

# UTC timestamps are given in ISO 8601 format
MIN_TIME = "2022-02-01"
MAX_TIME = "2022-03-01"


def get_transaction_by_hash(hash: str) -> dict:
    url = f"https://api.helium.io/v1/transactions/{hash}"
    return requests.get(url).json()["data"]


def get_payments_for_account(address: str, min_time: str, max_time: str) -> pd.DataFrame:
    # first endpoint fetches the transaction hashes
    # https://docs.helium.com/api/blockchain/accounts#roles-for-account
    url = f"https://api.helium.io/v1/accounts/{address}/roles?filter_types=payment_v1,payment_v2&min_time={min_time}&max_time={max_time}"
    txns = []
    while True:
        res = requests.get(url).json()
        txns += res["data"]
        try:
            url = f"https://api.helium.io/v1/accounts/{address}/roles?filter_types=payment_v1,payment_v2&min_time={min_time}&max_time={max_time}&cursor={res['cursor']}"
        except KeyError:
            break

    # second endpoint fetches the transaction data for that hash and parses accordingly
    # https://docs.helium.com/api/blockchain/transactions#transaction-for-hash
    # `payment_v1` type is single payee
    # `payment_v2` type is multi-payee
    payments = []
    for txn in txns:
        txn_data = get_transaction_by_hash(txn["hash"])
        if txn["type"] == "payment_v2":
            for payment in txn_data["payments"]:
                payments.append({
                    "role": txn["role"],
                    "type": txn_data["type"],
                    "time": txn_data["time"],
                    "payee": payment["payee"],
                    "amount_hnt": payment["amount"] / 1e8,
                    "payer": txn_data["payer"],
                    "nonce": txn_data["nonce"],
                    "height": txn_data["height"],
                    "hash": txn_data["hash"],
                    "fee": txn_data["fee"]
                })
        elif txn["type"] == "payment_v1":
            payments.append({
                "role": txn["role"],
                "type": txn_data["type"],
                "time": txn_data["time"],
                "payee": txn_data["payee"],
                "amount_hnt": txn_data["amount"] / 1e8,
                "payer": txn_data["payer"],
                "nonce": txn_data["nonce"],
                "height": txn_data["height"],
                "hash": txn_data["hash"],
                "fee": txn_data["fee"]
            })
        else:
            raise ValueError(f"Unexpected transaction type: {txn['type']}")

    # return a dataframe
    return pd.DataFrame(payments)


txns = get_payments_for_account(ACCOUNT, MIN_TIME, MAX_TIME)

# export as a CSV
txns.to_csv(f"payment_activity_{MIN_TIME}_{MAX_TIME}.csv")