Introduction

There are two main accounting models used in blockchains: the UTXO model and the account model.

The former is used by Bitcoin and Cardano; the latter is used by Ethereum, Algorand, and most other blockchains.

Let's see if we can use Python simulate the UTXO model inside an Algorand smart contract.

Understanding UTXOs

A UTXO is a bit like a treasure chest:

It has a value, locked by an address, which can be opened by a signature from the private key corresponding to the address.

A transaction is a mapping from one or more inputs to one or more outputs:

Each input has to be unlocked (accompanied by a signature) and spent as a whole.

In the system we'll create, the sum of input values must exactly equal the sum of output values for a transaction to be valid.

The smart contract's account will serve as a kind of internal ledger.

We'll have a way to move money into the system, by converting Algos to UTXOs, and a way to take it back out.

But all other internal transactions will follow the UTXO model.

Representing UTXOs on Algorand

There is no primitive type for UTXOs in the Algorand protocol, but we can create something similar using NFTs.

The locking address and value can be stored directly in the asset's parameters:

Let's start by defining a method in the contract that mints a pure NFT representing a UTXO:

class Utxo(ARC4Contract):
    """A contract that simulates UTXOs on Algorand."""

    @subroutine
    def _mint_utxo(self, lock: Account, value: Bytes) -> Asset:
        """An internal method that mints a UTXO.

        Args:
            lock (Account): The address that locks the UTXO.
            value (Bytes): The value of the UTXO.

        Returns:
            Asset: The UTXO asset.
        """
        return (
            itxn.AssetConfig(
                asset_name="UTXO",
                total=1,
                decimals=0,
                metadata_hash=value + op.bzero(24),
                reserve=lock,
                fee=0,
            )
            .submit()
            .created_asset
        )

The value of the UTXO is stored as the first 8 bytes of the metadata_hash, so we'll add a subroutine to parse the value back out of the field:

@arc4.abimethod
def value(self, utxo: Asset) -> UInt64:
    """Parses the value of a UTXO from its metadata hash.

    Args:
        utxo (Asset): The UTXO asset.

    Returns:
        UInt64: The value of the UTXO.
    """
    return op.extract_uint64(utxo.metadata_hash, 0)

Converting Algos to UTXOs

In order for meaningful money to enter this system, we need to implement a method that allows a user to deposit some Algos and convert them to a UTXO:

@arc4.abimethod
def convert_algo_to_utxo(self, payment: gtxn.PaymentTransaction) -> UInt64:
    """Converts Algos to a UTXO.

    Args:
        payment (gtxn.PaymentTransaction): The payment transaction.

    Returns:
        UInt64: The ID of the UTXO asset created.
    """
    assert (
        payment.receiver == Global.current_application_address
    ), "Payment receiver must be the application address"

    return self._mint_utxo(lock=Txn.sender, value=op.itob(payment.amount)).id

Let's quickly check this on LocalNet:

>>> ptxn = PaymentTxn(
...     sender=account.address,
...     sp=sp,
...     receiver=app.app_address,
...     amt=7_000,
... )
>>> signer = AccountTransactionSigner(account.private_key)
>>> asset_id = app.convert_algo_to_utxo(payment=TransactionWithSigner(ptxn, signer)).return_value
>>> utxo_value = app.value(utxo=asset_id).return_value
>>> print(f"Asset ID: {asset_id}")
Asset ID: 1020
>>> print(f"UTXO Value: {utxo_value}")
UTXO Value: 7000

UTXO Transactions

Now we need a method that validates and executes a transaction.

A user will need to specify a set of input UTXOs, which can be provided as an array of asset IDs:

Inputs: TypeAlias = arc4.DynamicArray[arc4.UInt64]

For the set of output UTXOs, the user needs to provide both the lock (address) and value.

We can use an ARC-4 struct to represent a transaction output:

class TxOut(arc4.Struct, kw_only=True):
    lock: arc4.Address
    value: arc4.UInt64

Which can be provided in an array:

Outputs: TypeAlias = arc4.DynamicArray[TxOut]

The method will begin with a check to ensure there is at least one input and at least one output provided:

@arc4.abimethod
def process_transaction(self, tx_ins: Inputs, tx_outs: Outputs) -> None:
    assert tx_ins, "Must provide at least one input"
    assert tx_outs, "Must provide at least one output"

Next, we loop over the transaction inputs and sum the input amounts:

tx_in_total = UInt64(0)
for tx_in in tx_ins:
    utxo = Asset(tx_in.native)
    assert utxo.creator == Global.current_application_address, "Input must be created by the application"
    assert utxo.reserve == Txn.sender, "Input must be locked by the sender"
    tx_in_value = self.value(utxo)
    self._burn_utxo(utxo)
    tx_in_total += tx_in_value

For each input, we verify that the asset was created by the contract (meaning it is a UTXO NFT), and that the asset's reserve address is the transaction sender.

Since the application call transaction is signed by the sender, this should suffice for checking that each input can be unlocked.

If all the checks pass, we burn each of the input UTXOs to ensure they can never be spent again:

@subroutine
def _burn_utxo(self, utxo: Asset) -> None:
    itxn.AssetConfig(
        config_asset=utxo,
        sender=Global.current_application_address,
        fee=0,
    ).submit()

See this article for more info on destroying assets.

Then we mint a new UTXO NFT for each of the outputs, and sum the total value of the outputs:

tx_out_total = UInt64(0)
for i in urange(tx_outs.length):
    tx_out = tx_outs[i].copy()
    self._mint_utxo(lock=Account(tx_out.lock.bytes), value=tx_out.value.bytes)
    tx_out_total += tx_out.value.native

Iterating over the array of TxOuts directly isn't supported, so we use urange to achieve the same outcome.

Finally, we check that the sum of inputs exactly equals the sum of the outputs:

assert tx_in_total == tx_out_total, "Total input value must equal total output value"

Note that if this check fails, the other changes are effectively rolled back.

Putting it all together:

@arc4.abimethod
def process_transaction(self, tx_ins: Inputs, tx_outs: Outputs) -> None:
    """Validates and processes a UTXO transaction.

    Args:
        tx_ins (Inputs): Array of UTXO asset IDs to spend.
        tx_outs (Outputs): Array of (address, value) tuples to create UTXOs for.
    """
    assert tx_ins, "Must provide at least one input"
    assert tx_outs, "Must provide at least one output"

    tx_in_total = UInt64(0)
    for tx_in in tx_ins:
        utxo = Asset(tx_in.native)
        assert utxo.creator == Global.current_application_address, "Input must be created by the application"
        assert utxo.reserve == Txn.sender, "Input must be locked by the sender"
        tx_in_value = self.value(utxo)
        self._burn_utxo(utxo)
        tx_in_total += tx_in_value

    tx_out_total = UInt64(0)
    for i in urange(tx_outs.length):
        tx_out = tx_outs[i].copy()
        self._mint_utxo(lock=Account(tx_out.lock.bytes), value=tx_out.value.bytes)
        tx_out_total += tx_out.value.native

    assert tx_in_total == tx_out_total, "Total input value must equal total output value"

Converting UTXOs Back to Algos

Finally, let's write a method that allows a user to convert their UTXO value back to Algos:

@arc4.abimethod
def convert_utxo_to_algo(self, utxo: Asset) -> None:
    """Converts a UTXO to Algos.

    Args:
        utxo (Asset): The UTXO asset to convert.
    """
    assert utxo.creator == Global.current_application_address, "Input must be created by the application"
    assert utxo.reserve == Txn.sender, "Input must be locked by the sender"
    itxn.Payment(
        receiver=Txn.sender,
        amount=self.value(utxo),
        fee=0,
    ).submit()
    self._burn_utxo(utxo)

We burn the UTXO to ensure it can never be spent again.

The Complete Contract Code

The code below is also available on GitHub.

from typing import TypeAlias

from algopy import Account, ARC4Contract, Asset, Bytes, Global, Txn, UInt64, arc4, gtxn, itxn, op, subroutine, urange


class TxOut(arc4.Struct, kw_only=True):
    lock: arc4.Address
    value: arc4.UInt64


Inputs: TypeAlias = arc4.DynamicArray[arc4.UInt64]
Outputs: TypeAlias = arc4.DynamicArray[TxOut]


class Utxo(ARC4Contract):
    """A contract that simulates UTXOs on Algorand."""

    @subroutine
    def _mint_utxo(self, lock: Account, value: Bytes) -> Asset:
        """An internal method that mints a UTXO.

        Args:
            lock (Account): The address that locks the UTXO.
            value (Bytes): The value of the UTXO.

        Returns:
            Asset: The UTXO asset.
        """
        return (
            itxn.AssetConfig(
                asset_name="UTXO",
                total=1,
                decimals=0,
                metadata_hash=value + op.bzero(24),
                reserve=lock,
                fee=0,
            )
            .submit()
            .created_asset
        )

    @subroutine
    def _burn_utxo(self, utxo: Asset) -> None:
        """An internal method that burns a UTXO.

        Args:
            utxo (Asset): The UTXO asset to burn.
        """
        itxn.AssetConfig(
            config_asset=utxo,
            sender=Global.current_application_address,
            fee=0,
        ).submit()

    @arc4.abimethod
    def convert_algo_to_utxo(self, payment: gtxn.PaymentTransaction) -> UInt64:
        """Converts Algos to a UTXO.

        Args:
            payment (gtxn.PaymentTransaction): The payment transaction.

        Returns:
            UInt64: The ID of the UTXO asset created.
        """
        assert (
            payment.receiver == Global.current_application_address
        ), "Payment receiver must be the application address"

        return self._mint_utxo(lock=Txn.sender, value=op.itob(payment.amount)).id

    @arc4.abimethod
    def convert_utxo_to_algo(self, utxo: Asset) -> None:
        """Converts a UTXO to Algos.

        Args:
            utxo (Asset): The UTXO asset to convert.
        """
        assert utxo.creator == Global.current_application_address, "Input must be created by the application"
        assert utxo.reserve == Txn.sender, "Input must be locked by the sender"
        itxn.Payment(
            receiver=Txn.sender,
            amount=self.value(utxo),
            fee=0,
        ).submit()
        self._burn_utxo(utxo)

    @arc4.abimethod
    def value(self, utxo: Asset) -> UInt64:
        """Parses the value of a UTXO from its metadata hash.

        Args:
            utxo (Asset): The UTXO asset.

        Returns:
            UInt64: The value of the UTXO.
        """
        return op.extract_uint64(utxo.metadata_hash, 0)

    @arc4.abimethod
    def process_transaction(self, tx_ins: Inputs, tx_outs: Outputs) -> None:
        """Validates and processes a UTXO transaction.

        Args:
            tx_ins (Inputs): Array of UTXO asset IDs to spend.
            tx_outs (Outputs): Array of (address, value) tuples to create UTXOs for.
        """
        assert tx_ins, "Must provide at least one input"
        assert tx_outs, "Must provide at least one output"

        tx_in_total = UInt64(0)
        for tx_in in tx_ins:
            utxo = Asset(tx_in.native)
            assert utxo.creator == Global.current_application_address, "Input must be created by the application"
            assert utxo.reserve == Txn.sender, "Input must be locked by the sender"
            tx_in_value = self.value(utxo)
            self._burn_utxo(utxo)
            tx_in_total += tx_in_value

        tx_out_total = UInt64(0)
        for i in urange(tx_outs.length):
            tx_out = tx_outs[i].copy()
            self._mint_utxo(lock=Account(tx_out.lock.bytes), value=tx_out.value.bytes)
            tx_out_total += tx_out.value.native

        assert tx_in_total == tx_out_total, "Total input value must equal total output value"

Writing Tests

The tests below are also available on GitHub.

import algokit_utils
import pytest
from algokit_utils import (
    Account,
    TransactionParameters,
    TransferParameters,
    transfer,
)
from algokit_utils.config import config
from algosdk.atomic_transaction_composer import (
    AccountTransactionSigner,
    TransactionWithSigner,
)
from algosdk.transaction import PaymentTxn
from algosdk.v2client.algod import AlgodClient
from algosdk.v2client.indexer import IndexerClient

from smart_contracts.artifacts.utxo.client import UtxoClient


@pytest.fixture(scope="session")
def app_client(account: Account, algod_client: AlgodClient, indexer_client: IndexerClient) -> UtxoClient:
    config.configure(
        debug=True,
        # trace_all=True,
    )

    client = UtxoClient(
        algod_client,
        creator=account,
        indexer_client=indexer_client,
    )

    client.deploy(
        on_schema_break=algokit_utils.OnSchemaBreak.AppendApp,
        on_update=algokit_utils.OnUpdate.AppendApp,
    )

    transfer(
        algod_client,
        TransferParameters(
            from_account=account,
            to_address=client.app_address,
            micro_algos=1_000_000,
        ),
    )

    return client


def test_convert_algo_to_utxo(account: Account, app_client: UtxoClient) -> None:
    """Tests the convert_algo_to_utxo() method."""

    sp = app_client.algod_client.suggested_params()
    sp.fee = 2_000
    sp.flat_fee = True

    ptxn = PaymentTxn(
        sender=account.address,
        sp=sp,
        receiver=app_client.app_address,
        amt=7_000,
    )
    signer = AccountTransactionSigner(account.private_key)
    asset_id = app_client.convert_algo_to_utxo(payment=TransactionWithSigner(ptxn, signer)).return_value
    utxo_value = app_client.value(utxo=asset_id).return_value

    assert isinstance(asset_id, int) and asset_id > 0, "Asset creation failed"
    assert utxo_value == 7_000, "Incorrect UTXO value"


def test_process_transaction(account: Account, app_client: UtxoClient) -> None:
    """Tests the process_transaction() method."""

    sp = app_client.algod_client.suggested_params()
    sp.fee = 2_000
    sp.flat_fee = True

    def create_utxo(amount: int) -> int:
        ptxn = PaymentTxn(
            sender=account.address,
            sp=sp,
            receiver=app_client.app_address,
            amt=amount,
        )
        signer = AccountTransactionSigner(account.private_key)
        return app_client.convert_algo_to_utxo(
            payment=TransactionWithSigner(ptxn, signer),
            transaction_parameters=TransactionParameters(suggested_params=sp),
        ).return_value

    asset_1 = create_utxo(10_000)
    asset_2 = create_utxo(20_000)

    sp = app_client.algod_client.suggested_params()
    sp.fee = 5_000
    sp.flat_fee = True

    app_client.process_transaction(
        tx_ins=[asset_1, asset_2],
        tx_outs=[(account.address, 25_000), (account.address, 5_000)],
        transaction_parameters=TransactionParameters(suggested_params=sp, foreign_assets=[asset_1, asset_2]),
    )


def test_convert_utxo_to_algo(account: Account, app_client: UtxoClient) -> None:
    """Tests the convert_utxo_to_algo() method."""
    sp = app_client.algod_client.suggested_params()
    sp.fee = 3_000
    sp.flat_fee = True

    ptxn = PaymentTxn(
        sender=account.address,
        sp=sp,
        receiver=app_client.app_address,
        amt=100_000,
    )
    signer = AccountTransactionSigner(account.private_key)
    asset_id = app_client.convert_algo_to_utxo(payment=TransactionWithSigner(ptxn, signer)).return_value

    balance_before = app_client.algod_client.account_info(account.address)["amount"]

    app_client.convert_utxo_to_algo(utxo=asset_id, transaction_parameters=TransactionParameters(suggested_params=sp))

    balance_after = app_client.algod_client.account_info(account.address)["amount"]

    assert balance_after == balance_before + 100_000 - 3_000, "Incorrect balance after converting back to Algos"