Merge pull request #11 from pactfi/add-zap

feat: Add Zap

examples/zap.py

@@ -0,0 +1,37 @@
+"""This example performs a zap on a pool."""
+
+import algosdk
+from algosdk.v2client.algod import AlgodClient
+
+import pactsdk
+
+private_key = algosdk.mnemonic.to_private_key("<mnemonic>")
+address = algosdk.account.address_from_private_key(private_key)
+
+algod = AlgodClient("<token>", "<url>")
+pact = pactsdk.PactClient(algod)
+
+algo = pact.fetch_asset(0)
+usdc = pact.fetch_asset(31566704)
+pool = pact.fetch_pools_by_assets(algo, usdc)[0]
+
+# Opt-in for usdc.
+opt_in_txn = usdc.prepare_opt_in_tx(address)
+sent_optin_txid = algod.send_transaction(opt_in_txn.sign(private_key))
+print(f"Opt-in transaction {sent_optin_txid}")
+
+# Opt-in for liquidity token.
+plp_opt_in_txn = pool.liquidity_asset.prepare_opt_in_tx(address)
+sent_plp_optin_txid = algod.send_transaction(plp_opt_in_txn.sign(private_key))
+print(f"OptIn transaction {sent_plp_optin_txid}")
+
+# Do a zap.
+zap = pool.prepare_zap(
+    asset=algo,
+    amount=100_000,
+    slippage_pct=2,
+)
+zap_tx_group = zap.prepare_tx_group(address)
+signed_group = zap_tx_group.sign(private_key)
+algod.send_transactions(signed_group)
+print(f"Zap transaction group {zap_tx_group.group_id}")

pactsdk/__init__.py

@@ -7,3 +7,4 @@
 from .pool import Pool, PoolState  # noqa
 from .swap import Swap, SwapEffect  # noqa
 from .transaction_group import TransactionGroup  # noqa
+from .zap import Zap, ZapParams  # noqa

pactsdk/pool.py

@@ -23,6 +23,7 @@
 from .pool_calculator import PoolCalculator
 from .swap import Swap
 from .transaction_group import TransactionGroup
+from .zap import Zap
 
 PoolType = Literal["CONSTANT_PRODUCT", "STABLESWAP"]
 
@@ -531,6 +532,62 @@ def parse_internal_state(self, state: AppInternalState) -> PoolState:
             secondary_asset_price=self.calculator.secondary_asset_price,
         )
 
+    def prepare_zap(self, asset: Asset, amount: int, slippage_pct: float) -> Zap:
+        """Creates a new zap instance for getting all required data for performing a zap.
+
+        Args:
+            asset: The asset to zap.
+            amount: Amount used for the zap.
+            slippage_pct: The maximum allowed slippage in percents e.g. `10` is 10%. The swap will fail if the slippage will be higher.
+
+        Returns:
+            A new zap object.
+        """
+        return Zap(
+            self,
+            asset=asset,
+            amount=amount,
+            slippage_pct=slippage_pct,
+        )
+
+    def prepare_zap_tx_group(self, zap: Zap, address: str) -> TransactionGroup:
+        """Prepares a transaction group that when executed will perform a Zap on the pool.
+
+        Args:
+            zap: The zap for which to generate transactions.
+            address: The address that is performing the Zap.
+
+        Returns:
+            Transaction group that when executed will perform a Zap on the pool.
+        """
+        suggested_params = self.algod.suggested_params()
+        txs = self.build_zap_txs(zap, address, suggested_params)
+        return TransactionGroup(txs)
+
+    def build_zap_txs(
+        self, zap: Zap, address: str, suggested_params: transaction.SuggestedParams
+    ) -> list[transaction.Transaction]:
+        """Builds the transactions to perform a Zap on the pool as per the options passed in. Zap allows to add liquidity to the pool by providing only one asset.
+
+        This function will generate swap Txs to get a proper amount of the second asset and then generate add liquidity Txs with both of those assets.
+        See :py:meth:`pactsdk.pool.Pool.buildSwapTxs` and :py:meth:`pactsdk.pool.Pool.buildAddLiquidityTxs` for more details.
+
+        This feature is supposed to work with constant product pools only. Stableswaps can accept one asset to add liquidity by default.
+
+        Args:
+            zap: The zap for which to generate transactions.
+            address: The address that is performing the Zap.
+            suggested_params: Algorand suggested parameters for transactions.
+
+        Returns:
+            List of transactions to perform the Zap.
+        """
+        swap_txs = self.build_swap_txs(zap.swap, address, suggested_params)
+        add_liq_txs = self.build_add_liquidity_txs(
+            address, zap.liquidity_addition, suggested_params
+        )
+        return [*swap_txs, *add_liq_txs]
+
     def _make_deposit_tx(
         self,
         asset: Asset,

pactsdk/pool_calculator.py

@@ -205,7 +205,7 @@ def _gross_amount_received_to_amount_deposited(
         asset: Asset,
         int_gross_amount_received: int,
     ) -> int:
-        A, B = self._get_liquidities(asset)
+        A, B = self.get_liquidities(asset)
         return self.swap_calculator.get_swap_amount_deposited(
             A,
             B,
@@ -217,14 +217,22 @@ def _amount_deposited_to_gross_amount_received(
         asset: Asset,
         amount_deposited: int,
     ) -> int:
-        A, B = self._get_liquidities(asset)
+        A, B = self.get_liquidities(asset)
         return self.swap_calculator.get_swap_gross_amount_received(
             A,
             B,
             amount_deposited,
         )
 
-    def _get_liquidities(self, asset: Asset) -> tuple[int, int]:
+    def get_liquidities(self, asset: Asset) -> tuple[int, int]:
+        """Returns the array of liquidities from the pool, sorting them by setting provided asset as primary.
+
+        Args:
+            asset: The asset that is supposed to be the primary one.
+
+        Returns:
+            Total liquidities of assets.
+        """
         A, B = [self.primary_asset_amount, self.secondary_asset_amount]
         if asset != self.pool.primary_asset:
             A, B = B, A

pactsdk/zap.py

@@ -0,0 +1,155 @@
+"""Set of utility classes for managing and performing zaps.
+"""
+import math
+from dataclasses import dataclass, field
+from typing import TYPE_CHECKING
+
+from .add_liquidity import LiquidityAddition
+from .asset import Asset
+from .swap import Swap
+from .transaction_group import TransactionGroup
+
+if TYPE_CHECKING:
+    from .pool import Pool
+
+
+FEE_PRECISION = 10**4
+
+
+@dataclass
+class ZapParams:
+    """All amounts that should be used in swap and add liquidity transactions."""
+
+    swap_deposited: int
+    primary_add_liq: int
+    secondary_add_liq: int
+
+
+def get_constant_product_zap_params(
+    liq_a: int,
+    liq_b: int,
+    zap_amount: int,
+    fee_bps: int,
+    pact_fee_bps: int,
+) -> ZapParams:
+    swap_deposited = get_swap_amount_deposited_from_zapping(
+        zap_amount, liq_a, fee_bps, pact_fee_bps
+    )
+    primary_add_liq = zap_amount - swap_deposited
+    secondary_add_liq = get_secondary_added_liquidity_from_zapping(
+        swap_deposited, liq_a, liq_b, fee_bps
+    )
+    return ZapParams(swap_deposited, primary_add_liq, secondary_add_liq)
+
+
+def get_swap_amount_deposited_from_zapping(
+    zap_amount: int,
+    total_amount: int,
+    fee_bps: int,
+    pact_fee_bps: int,
+) -> int:
+    pool_fee = fee_bps - pact_fee_bps
+    a = (-FEE_PRECISION - pool_fee + fee_bps) // FEE_PRECISION
+    b = (
+        -2 * total_amount * FEE_PRECISION
+        + zap_amount * pool_fee
+        + total_amount * fee_bps
+    ) // FEE_PRECISION
+    c = total_amount * zap_amount
+
+    delta = b * b - 4 * a * c
+    if b < 0:
+        result = (-b - math.isqrt(delta)) // (2 * a)
+    else:
+        result = (2 * c) // (-b + math.isqrt(delta))
+    return result
+
+
+def get_secondary_added_liquidity_from_zapping(
+    swap_deposited: int,
+    total_primary: int,
+    total_secondary: int,
+    fee_bps,
+) -> int:
+    return (swap_deposited * total_secondary * (FEE_PRECISION - fee_bps)) // (
+        (total_primary + swap_deposited) * FEE_PRECISION
+    )
+
+
+@dataclass
+class Zap:
+    """Zap class represents a zap trade on a particular pool, which allows to exchange single asset for PLP token.
+
+    Zap performs a swap to get second asset from the pool and then adds liquidity using both of those assets. Users may be left with some leftovers due to rounding and slippage settings.
+
+    Zaps are meant only for Constant Product pools; For Stableswaps, adding only one asset works out of the box.
+
+    Typically, users don't have to manually instantiate this class. Use :py:meth:`pactsdk.pool.Pool.prepare_zap` instead.
+    """
+
+    pool: "Pool"
+    """The pool the zap is going to be performed in."""
+
+    asset: Asset
+    """The asset that will be used in zap."""
+
+    amount: int
+    """Amount to be used in zap."""
+
+    slippage_pct: float
+    """The maximum amount of slippage allowed in performing the swap."""
+
+    swap: Swap = field(init=False)
+    """The swap object that will be executed during the zap."""
+
+    liquidity_addition: LiquidityAddition = field(init=False)
+    """Liquidity Addition object that will be executed during the zap."""
+
+    params: ZapParams = field(init=False)
+    """All amounts used in swap and add liquidity transactions."""
+
+    def __post_init__(self):
+        if not self.pool.is_asset_in_the_pool(self.asset):
+            raise AssertionError("Provided asset was not found in the pool.")
+
+        if self.pool.pool_type == "STABLESWAP":
+            raise AssertionError("Zap can only be made on constant product pools.")
+
+        self.params = self.get_zap_params()
+        self.swap = self.pool.prepare_swap(
+            self.asset, self.params.swap_deposited, self.slippage_pct
+        )
+        self.liquidity_addition = LiquidityAddition(
+            pool=self.pool,
+            primary_asset_amount=self.params.primary_add_liq,
+            secondary_asset_amount=self.params.secondary_add_liq,
+        )
+
+    def prepare_tx_group(self, address: str) -> TransactionGroup:
+        """Creates the transactions needed to perform zap and returns them as a transaction group ready to be signed and committed.
+
+        Args:
+            address: The account that will be performing the zap.
+
+        Returns:
+            A transaction group that when executed will perform the zap.
+        """
+        return self.pool.prepare_zap_tx_group(self, address)
+
+    def _is_asset_primary(self):
+        return self.asset.index == self.pool.primary_asset.index
+
+    def get_zap_params(self) -> ZapParams:
+        A, B = self.pool.calculator.get_liquidities(self.asset)
+        if A == 0 or B == 0:
+            raise ValueError("Cannot create a Zap on empty pool.")
+
+        params = get_constant_product_zap_params(
+            A, B, self.amount, self.pool.params.fee_bps, self.pool.params.pact_fee_bps
+        )
+        if not self._is_asset_primary():
+            temp = params.primary_add_liq
+            params.primary_add_liq = params.secondary_add_liq
+            params.secondary_add_liq = temp
+
+        return params