Risk-Free Arbitrage on Algorand

Introduction

Arbitrage is the simultaneous purchase and sale of the same or similar asset in different markets in order to profit from tiny differences in the asset’s listed price.

Source: Arbitrage: How Arbitraging Works in Investing, With Examples

Alice wants to buy an asset on exchange A and sell it on exchange B, where it commands a higher price.

She executes the first trade, but the price on exchange B drops before she is able to sell.

This is called execution risk.

Some blockchains support atomic swaps, where transactions can be bundled such that the entire group is guaranteed to succeed or fail together.

This eliminates the risk of partial execution (buying on exchange A, and not getting filled on exchange B).

But what about the risk of slippage?

Some DEXs may support limit orders or other mechanisms to ensure a trade will not be executed if price moves unfavourably.

But there’s also an easy way to guarantee you never lose money on a trade, by using a simple smart contract.

The Smart Contract

The idea is to construct an atomic transaction group as follows:

Let’s start by defining the application creation method:

class BalanceProtector(ARC4Contract):
    """Fails a group of transactions if the owner's account balance ends lower than it starts."""

    @arc4.abimethod(create="require")
    def new(self, owner: Account) -> None:
        self.owner = owner
        self.asset = Asset(0)
        self.starting_balance = UInt64(0)

The owner will be the only account authorised to call the other contract methods for this application instance.

Next we define a subroutine to check the owner’s balance of an asset:

@subroutine
def balance(self) -> UInt64:
    """Fetches the owner's balance of the asset (ASA or Algo).

    Returns:
        UInt64: The owner's balance.
    """
    return self.asset.balance(self.owner) if self.asset else self.owner.balance

In Algorand Python, an Asset is falsey if its ID is zero. We use that here to represent native Algo.

Any other valid ID represents an Algorand Standard Asset (ASA).

Now let’s define the method to be called first in the transaction group:

@arc4.abimethod
def take_snapshot(self, asset: Asset) -> UInt64:
    """Stores a snapshot of the owner's balance in the contract's global state.

    Args:
        asset (Asset): The asset to snapshot the balance of.

    Returns:
        UInt64: The round the snapshot was taken in.
    """
    assert Txn.sender == self.owner, "Only the owner can call this method."
    assert Txn.group_index == 0, "Transaction must be first in group."
    assert (
        gtxn.ApplicationCallTransaction(Global.group_size - 1).app_id == Global.current_application_id
    ), "Last transaction in group must be a call to this application."

    self.asset = asset
    self.starting_balance = self.balance()
    return Global.round

The three assert statements are to guard the following conditions:

• That only the owner can call this method

• That the call to the take_snapshot method is the first transaction in the group

• That the last transaction in the group is another call to this application (but not to this method)

The method then updates the asset and starting_balance in the application’s global state.

The last method to define is protect, which must be called as the last transaction in the group:

@arc4.abimethod
def protect(self) -> UInt64:
    """Fails if the owner's balance is lower than it was at the snapshot round.

    Returns:
        UInt64: The balance delta.
    """
    assert Txn.sender == self.owner, "Only the owner can call this method."
    assert Txn.group_index == Global.group_size - 1, "Transaction must be last in group."
    assert (
        gtxn.ApplicationCallTransaction(0).app_id == Global.current_application_id
    ), "First transaction in group must be a call to this application."

    # Will cause an error (would result in negative)
    # If the balance is less than the starting balance
    return self.balance() - Txn.fee - self.starting_balance

And that’s all there is to it!

The contract code below is also available on GitHub:

from algopy import (
    Account,
    ARC4Contract,
    Asset,
    Global,
    Txn,
    UInt64,
    arc4,
    gtxn,
    subroutine,
)


class BalanceProtector(ARC4Contract):
    """Fails a group of transactions if the owner's account balance ends lower than it starts."""

    @arc4.abimethod(create="require")
    def new(self, owner: Account) -> None:
        self.owner = owner
        self.asset = Asset(0)
        self.starting_balance = UInt64(0)

    @subroutine
    def balance(self) -> UInt64:
        """Fetches the owner's balance of the asset (ASA or Algo).

        Returns:
            UInt64: The owner's balance.
        """
        return self.asset.balance(self.owner) if self.asset else self.owner.balance

    @arc4.abimethod
    def take_snapshot(self, asset: Asset) -> UInt64:
        """Stores a snapshot of the owner's balance in the contract's global state.

        Args:
            asset (Asset): The asset to snapshot the balance of.

        Returns:
            UInt64: The round the snapshot was taken in.
        """
        assert Txn.sender == self.owner, "Only the owner can call this method."
        assert Txn.group_index == 0, "Transaction must be first in group."
        assert (
            gtxn.ApplicationCallTransaction(Global.group_size - 1).app_id == Global.current_application_id
        ), "Last transaction in group must be a call to this application."

        self.asset = asset
        self.starting_balance = self.balance()
        return Global.round

    @arc4.abimethod
    def protect(self) -> UInt64:
        """Fails if the owner's balance is lower than it was at the snapshot round.

        Returns:
            UInt64: The balance delta.
        """
        assert Txn.sender == self.owner, "Only the owner can call this method."
        assert Txn.group_index == Global.group_size - 1, "Transaction must be last in group."
        assert (
            gtxn.ApplicationCallTransaction(0).app_id == Global.current_application_id
        ), "First transaction in group must be a call to this application."

        # Will cause an error (would result in negative)
        # If the balance is less than the starting balance
        return self.balance() - Txn.fee - self.starting_balance

Let’s compile the contract:

algokit compile py smart_contracts/arbitrage/contract.py
algokit generate client smart_contracts/arbitrage/BalanceProtector.arc32.json -o smart_contracts/artifacts/arbitrage/client.py

And write some tests :)

Writing Tests

To simulate a profitable trade, we can transfer money from a secondary account to the contract owner:

def test_profitable_trade_succeeds(app_client: BalanceProtectorClient, account: Account, algod_client: AlgodClient):
    """Simulates a profitable trade by transferring funds from another account."""

    balance = lambda a: algod_client.account_info(a.address)["amount"]
    balance_before = balance(account)

    # Get or create a new account for testing
    counterparty = get_account(algod_client, "counterparty")

    ensure_funded(
        algod_client,
        EnsureBalanceParameters(
            account_to_fund=counterparty,
            min_spending_balance_micro_algos=10_000,
        ),
    )

    txn = PaymentTxn(
        sender=counterparty.address,
        receiver=account.address,
        amt=10_000,
        sp=algod_client.suggested_params(),
    )
    tws = TransactionWithSigner(txn, AccountTransactionSigner(counterparty.private_key))

    atc = app_client.compose().take_snapshot(asset=0).build()
    atc = atc.add_transaction(tws)
    atc = app_client.compose(atc).protect().build()

    result = app_client.compose(atc).execute()

    balance_after = balance(account)

    assert balance_after >= balance_before, "Balance should be protected"
    assert result.confirmed_round > 0, "Transaction should be confirmed"

Simulating a negative balance delta is much easier.

We can call the take_snapshot and protect methods in a group, without making any trades in between.

The transaction fees alone will cause the owner’s balance to end lower than it started, so the contract should throw an error to prevent the loss:

def test_unprofitable_trade_fails(app_client: BalanceProtectorClient):
    """To simulate a losing trade, call the contract methods without making any trades in between."""
    atc = app_client.compose().take_snapshot(asset=0).protect()

    with pytest.raises(LogicError) as e:
        atc.execute()
    assert "would result negative" in str(e.value)

Please only use these tests as a starting point. They are far from comprehensive :)

The test code below is also available on GitHub:

import pytest
from algokit_utils import (
    Account,
    EnsureBalanceParameters,
    ensure_funded,
    get_account,
)
from algokit_utils.config import config
from algokit_utils.logic_error import LogicError
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.arbitrage.client import BalanceProtectorClient


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

    client = BalanceProtectorClient(
        algod_client,
        creator=account,
        indexer_client=indexer_client,
    )
    client.create_new(owner=account.address)
    return client


def test_profitable_trade_succeeds(app_client: BalanceProtectorClient, account: Account, algod_client: AlgodClient):
    """Simulates a profitable trade by transferring funds from another account."""

    balance = lambda a: algod_client.account_info(a.address)["amount"]
    balance_before = balance(account)

    # Get or create a new account for testing
    counterparty = get_account(algod_client, "counterparty")

    ensure_funded(
        algod_client,
        EnsureBalanceParameters(
            account_to_fund=counterparty,
            min_spending_balance_micro_algos=10_000,
        ),
    )

    txn = PaymentTxn(
        sender=counterparty.address,
        receiver=account.address,
        amt=10_000,
        sp=algod_client.suggested_params(),
    )
    tws = TransactionWithSigner(txn, AccountTransactionSigner(counterparty.private_key))

    atc = app_client.compose().take_snapshot(asset=0).build()
    atc = atc.add_transaction(tws)
    atc = app_client.compose(atc).protect().build()

    result = app_client.compose(atc).execute()

    balance_after = balance(account)

    assert balance_after >= balance_before, "Balance should be protected"
    assert result.confirmed_round > 0, "Transaction should be confirmed"


def test_unprofitable_trade_fails(app_client: BalanceProtectorClient):
    """To simulate a losing trade, call the contract methods without making any trades in between."""
    atc = app_client.compose().take_snapshot(asset=0).protect()

    with pytest.raises(LogicError) as e:
        atc.execute()
    assert "would result negative" in str(e.value)

Happy coding!