Aptos Coin Transfer From Tx

aptos_transfer.py

#!/usr/bin/env python3
import logging

logging.basicConfig(
    format="[%(asctime)s] - %(levelname)s - %(message)s", level=logging.INFO
)


WITHDRAW_EVENT = "0x1::coin::WithdrawEvent"
DEPOSIT_EVENT = "0x1::coin::DepositEvent"
COINSTORE_TYPE = "0x1::coin::CoinStore<"


def extract_coin_address(coin):
    if not coin.startswith(COINSTORE_TYPE) or not coin.endswith(">"):
        return None
    coin_address = coin[len(COINSTORE_TYPE) : -1]
    # LP coin is a collection of multi coins:
    # eg:
    # 0x1::coin::CoinStore<
    #   0x61d2c22a6cb7831bee0f48363b0eec92369357aece0d1142062f7d5d85c7bef8::lp_coin::LP<
    #     0x665d06fcd9c94430099f82973f2a5e5f13142e42fa172e72ce14f51a64bd8ad9::coin_mbx::MBX,
    #     0x1::aptos_coin::AptosCoin,
    #     0x163df34fccbf003ce219d3f1d9e70d140b60622cb9dd47599c25fb2f797ba6e::curves::Uncorrelated
    #   >
    # >
    # coin_parts = coin_address.split("::")
    # if len(coin_parts) != 3:
    #     return None
    return coin_address


# Event
# DepositEvent
# {
#     "guid": {
#         "creation_number": "2",
#         "account_address": "0x598dbe189e466ffb544fe1bc36a9663aa93ab9ee87e722420152bc853983b7dd",
#     },
#     "sequence_number": "1",
#     "type": "0x1::coin::DepositEvent",
#     "data": {"amount": "368946795"},
# }
# WithdrawEvent
# {
#     "guid": {
#         "creation_number": "5",
#         "account_address": "0x598dbe189e466ffb544fe1bc36a9663aa93ab9ee87e722420152bc853983b7dd",
#     },
#     "sequence_number": "0",
#     "type": "0x1::coin::WithdrawEvent",
#     "data": {"amount": "33879327"},
# }
def extract_transfer_events(events):
    transfers = []
    for event in events:
        if event["type"] in (DEPOSIT_EVENT, WITHDRAW_EVENT):
            transfers.append(
                {
                    "creation_number": event["guid"]["creation_number"],
                    "type": "Deposit" if event["type"] == DEPOSIT_EVENT else "Withdraw",
                    "amount": int(event["data"]["amount"]),
                    "address": event["guid"]["account_address"],
                }
            )
    return transfers


# Changes
# {
#   "address": "0x598dbe189e466ffb544fe1bc36a9663aa93ab9ee87e722420152bc853983b7dd",
#   "state_key_hash": "0x49b86859e2965b94dcd9b60847f6e80713da9230896ae293c4646f8ef2d7b9a2",
#   "data": {
#     "type": "0x1::coin::CoinStore<0x1::aptos_coin::AptosCoin>",
#     "data": {
#       "coin": { "value": "473927358" },
#       "deposit_events": {
#         "counter": "2",
#         "guid": {
#           "id": {
#             "addr": "0x598dbe189e466ffb544fe1bc36a9663aa93ab9ee87e722420152bc853983b7dd",
#             "creation_num": "2"
#           }
#         }
#       },
#       "frozen": false,
#       "withdraw_events": {
#         "counter": "7",
#         "guid": {
#           "id": {
#             "addr": "0x598dbe189e466ffb544fe1bc36a9663aa93ab9ee87e722420152bc853983b7dd",
#             "creation_num": "3"
#           }
#         }
#       }
#     }
#   },
#   "type": "write_resource"
# }
def extract_coin_changes(changes):
    coin_changes = {}
    for change in changes:
        if change["type"] != "write_resource":
            continue
        address = change["address"]
        coin_address = extract_coin_address(change["data"]["type"])
        if coin_address is None:
            continue
        coin_data = change["data"]["data"]

        deposit = coin_data["deposit_events"]["guid"]["id"]
        if deposit["addr"] != address:
            raise ValueError(
                "deposit_address: {} != change.address: {}".format(
                    deposit["address"], address
                )
            )
        coin_changes[address + "::" + deposit["creation_num"]] = {
            "type": "Deposit",
            "address": address,
            "coin_address": coin_address,
        }

        withdraw = coin_data["withdraw_events"]["guid"]["id"]
        if withdraw["addr"] != address:
            raise ValueError(
                "withdraw_address: {} != change.address: {}".format(
                    withdraw["address"], address
                )
            )
        coin_changes[address + "::" + withdraw["creation_num"]] = {
            "type": "Withdraw",
            "address": address,
            "coin_address": coin_address,
        }

    return coin_changes


def get_transfer_from_tx(tx):
    version = tx["version"]
    txhash = tx["hash"]

    if tx.get("vm_status") != "Executed successfully":
        logging.error(
            "tx: {} is not executed successfully: {}".format(
                version, tx.get("vm_status")
            )
        )
        return None

    if "events" not in tx or "changes" not in tx:
        return None

    try:
        transfers = extract_transfer_events(tx["events"])
    except Exception as e:
        raise ValueError(f"Failed to extract transfer events from tx: {version}, {e}")

    try:
        coin_changes = extract_coin_changes(tx["changes"])
    except Exception as e:
        raise ValueError(f"Failed to extract coin changes from tx: {version}, {e}")
    for idx, transfer in enumerate(transfers):
        creation_number = transfer["address"] + "::" + transfer["creation_number"]
        if creation_number not in coin_changes:
            raise ValueError(
                f"tx: {version} transfer: {transfer} not found in coin_changes: {coin_changes}"
            )
        change = coin_changes[creation_number]
        if transfer["type"] != change["type"]:
            raise ValueError(
                f"tx: {txhash} transfer.type: {transfer} != change.type: {change}"
            )
        if transfer["address"] != change["address"]:
            raise ValueError(
                f"tx: {version} transfer.address: {transfer} != change.address {change}"
            )
        transfers[idx].update(
            {
                "hash": txhash,
                "version": version,
                "coin_address": change["coin_address"],
            }
        )

    return transfers

import requests
tx = requests.get("https://fullnode.mainnet.aptoslabs.com/v1/transactions/by_version/450000043").json()

get_transfer_from_tx(tx)

[{'creation_number': '5',
  'type': 'Withdraw',
  'amount': 33879327,
  'address': '0x598dbe189e466ffb544fe1bc36a9663aa93ab9ee87e722420152bc853983b7dd',
  'hash': '0x7fd7cc3292ba0cad805d0b0f675e2c527a350b869f5a6a1cac883fa27c221c01',
  'version': '450000043',
  'coin_address': '0xf22bede237a07e121b56d91a491eb7bcdfd1f5907926a9e58338f964a01b17fa::asset::USDC'},
 {'creation_number': '2',
  'type': 'Deposit',
  'amount': 368946795,
  'address': '0x598dbe189e466ffb544fe1bc36a9663aa93ab9ee87e722420152bc853983b7dd',
  'hash': '0x7fd7cc3292ba0cad805d0b0f675e2c527a350b869f5a6a1cac883fa27c221c01',
  'version': '450000043',
  'coin_address': '0x1::aptos_coin::AptosCoin'}]