Python interface to HumbleSwap pool state

humble_pool_quote.py

#!/usr/bin/env python3
import sys
import struct
from base64 import b64decode
from algosdk.v2client.algod import AlgodClient
from tinyman.assets import Asset, AssetAmount

class HumblePool:
    """Liquidity pool in the HumbleSwap AMM DEX."""
    def __init__(self, client: AlgodClient, app_id: int, N2NN: bool) -> None:
        """Constructor method for :class:`HumblePool`

        :param client: a :class:`AlgodClient` object
        :param app_id: pool app_id
        :param N2NN: NetworkAsset<->NonNetworkAsset pool type flag, True for ALGO<->OTHER pools
        """
        self.client = client
        self.app_id = app_id
        self.is_algo_pool = N2NN

        app_info = self.client.application_info(self.app_id)
        app_state = app_info['params']['global-state']
        state_data = {b64decode(x['key']).hex(): b64decode(x['value']['bytes']) for x in app_state}
        self.state_data = state_data
        b0 = state_data['00']
        if self.is_algo_pool:
            tokenB_id, = struct.unpack_from('!Q', b0, 32)
            self.asset1 = Asset(id=0).fetch(self.client)
            self.asset2 = Asset(id=tokenB_id).fetch(self.client)
        else:
            tokenA_id, tokenB_id = struct.unpack_from('!QQ', b0, 32)
            self.asset1 = Asset(id=tokenA_id).fetch(self.client)
            self.asset2 = Asset(id=tokenB_id).fetch(self.client)
        self.refresh()

    def refresh(self):
        app_info = self.client.application_info(self.app_id)
        app_state = app_info['params']['global-state']
        state_data = {b64decode(x['key']).hex(): b64decode(x['value']['bytes']) for x in app_state}
        b0 = state_data['00']
        if self.is_algo_pool:
            balA, balB = struct.unpack_from('!QQ', b0, offset=32+57)
        else:
            balA, balB = struct.unpack_from('!QQ', b0, offset=32+65)
        self.asset1_reserves = balA
        self.asset2_reserves = balB

    def fetch_fixed_input_swap_quote(self, amount_in: AssetAmount):
        asset_in, asset_in_amount = amount_in.asset, amount_in.amount
        self.refresh()
        if asset_in == self.asset1:
            asset_out = self.asset2
            input_supply = self.asset1_reserves
            output_supply = self.asset2_reserves
        else:
            asset_out = self.asset1
            input_supply = self.asset2_reserves
            output_supply = self.asset1_reserves

        if not input_supply or not output_supply:
            raise Exception('Pool has no liquidity!')

        # k = input_supply * output_supply
        # ignoring fees, k must remain constant
        # (input_supply + asset_in) * (output_supply - amount_out) = k
        k = input_supply * output_supply
        asset_in_amount_minus_fee = (asset_in_amount * 997) / 1000
        swap_fees = asset_in_amount - asset_in_amount_minus_fee
        asset_out_amount = output_supply - (k / (input_supply + asset_in_amount_minus_fee))

        amount_out = AssetAmount(asset_out, int(asset_out_amount))
        return amount_out


def main():
    trade = float(sys.argv[1]) if len(sys.argv) > 1 else 1000
    client = AlgodClient("", "https://mainnet-api.algonode.cloud")
    pool_id = 777628254  # ALGO-USDC pool
    pool = HumblePool(client, app_id=pool_id, N2NN=True)
    print(f'Pool id: {pool.app_id}')
    print(f'Asset1 reserves {pool.asset1}: {pool.asset1_reserves}')
    print(f'Asset2 reserves {pool.asset2}: {pool.asset2_reserves}')
    print(f'Price: {pool.asset2_reserves / pool.asset1_reserves:.6f}')
    print(f'InvPrice: {pool.asset1_reserves / pool.asset2_reserves:.6f}')
    amount1 = pool.asset1(trade * 10**pool.asset1.decimals)
    amount2 = pool.asset2(trade * 10**pool.asset2.decimals)
    hs_out1 = pool.fetch_fixed_input_swap_quote(amount1)
    hs_out2 = pool.fetch_fixed_input_swap_quote(amount2)
    print(f'quote for {amount1}: {hs_out1}')
    print(f'quote for {amount2}: {hs_out2}')


if __name__ == '__main__':
    main()