Swap assets on Tinyman v2 DEX from command line

tinyman_v2_swap.py

#!/usr/bin/env python3
from dotenv import dotenv_values
from algosdk import mnemonic, account
from algosdk.v2client import algod
from tinyman.v2.client import TinymanV2MainnetClient
from pprint import pprint
from urllib.parse import quote_plus

# Load user address
env_vars = dotenv_values(".env")
private_key = mnemonic.to_private_key(env_vars['mnemonic'])
account = {
    'address': account.address_from_private_key(private_key),
    'private_key': private_key
}

# Connect to algod node
algod_token = ''
algod_address = 'https://mainnet-api.algonode.cloud'
algod_client = algod.AlgodClient(algod_token, algod_address)
client = TinymanV2MainnetClient(algod_client=algod_client, user_address=account['address'])
# By default all subsequent operations are on behalf of user_address

# Get user balances
account['info'] = client.algod.account_info(client.user_address)
account['balances'] = {asset['asset-id']: asset['amount'] for asset in account['info']['assets']}
account['balances'][0] = account['info']['amount']     # special treatment for ALGO asset

# fetch user assets
account['assets'] = [client.fetch_asset(asset_id) for asset_id in sorted(account['balances'])]
# enumerated assets list for the selection prompt
print('Account:', account['address'])
print(', '.join(['{}-{}'.format(i, a.unit_name) for i, a in enumerate(account['assets'])]))
# Fetch two assets of interest using their seq number in account['assets'] list
from_ask = input('From: ').lower()
asset_from = account['assets'][int(from_ask)]
print(f'Selected {asset_from}')

to_ask = input('To: ').lower()
asset_to = account['assets'][int(to_ask)]
print(f'Selected {asset_to}')

# ask for swap amount using full balance as default value
amount_full = asset_from(account['balances'][asset_from.id])  # AssetAmount type with user balance
amount_ask = input(f'Amount [{amount_full}]: ').strip()
if amount_ask:
    amount_from = int(float(amount_ask) * 10**asset_from.decimals)
else:
    amount_from = amount_full.amount

slippage_ask = input('slippage % [0.1]: ').strip()
slippage = float(slippage_ask) / 100.0 if slippage_ask else 0.001

# Fetch the pool we will work with
pool = client.fetch_pool(asset_from, asset_to)

# Get a quote for a swap of {asset_from} to {asset_to} and print swap info
quote = pool.fetch_fixed_input_swap_quote(amount_in=asset_from(amount_from), slippage=slippage)
print('#'*70)
print(f'Swap {quote.swap_type} from {quote.amount_in} to {quote.amount_out}')
print('#'*70)
print(f'Price (worst case): {quote.price_with_slippage:.6f} {asset_to.unit_name} per {asset_from.unit_name}')
print(f'Price: {quote.price:.6f} {asset_to.unit_name} per {asset_from.unit_name}')
print(f'InvPrice: {1/quote.price:.6f} {asset_from.unit_name} per {asset_to.unit_name}')
print(f'Swap Fee: {quote.swap_fees}')
print(f'Slippage: {100*quote.slippage:.2f} %')
print(f'Price Impact: {100*quote.price_impact:.2f} %')
print('#'*70)
print(f'Output amount: {quote.amount_out}')
print(f'Minimum received: {quote.amount_out_with_slippage}')
print('#'*70)

proceed_ask = input('Proceed (y/N)?: ').lower()
do_swap = True if proceed_ask in ['y', 'yes'] else False
if do_swap and quote.price_impact > 0.05:
    print(f'Aborting swap! Price impact is too high: {100*quote.price_impact:.2f} %')
    do_swap = False

if do_swap:
    print(f'Swapping {quote.amount_in} to {quote.amount_out_with_slippage} ...')
    # Prepare a transaction group
    txn_group = pool.prepare_swap_transactions_from_quote(quote=quote)
    # Sign the group with our key
    txn_group.sign_with_private_key(account['address'], account['private_key'])
    # Submit transactions to the network and wait for confirmation
    txn_info = client.submit(txn_group, wait=True)

    print("Transaction Info")
    pprint(txn_info)
    print(f"groupid: {quote_plus(txn_group.id)}")