Axo

Axo Order Book Module.

AxoCreateResponse

Bases: BaseModel

Source code in src/charli3_dendrite/dexs/ob/axo.py

class AxoCreateResponse(BaseModel):
    policy_script: str
    strat_id: str
    token_name: str
    datum: str
    algo_addr: str
    nft_metadata: dict

    @field_validator("nft_metadata", mode="before")
    @classmethod
    def validate_metadata(cls, v: str) -> dict:
        return json.loads(v)

    @field_validator("policy_script")
    @classmethod
    def strip_policy(cls, v: str) -> str:
        """Trim the extra bytes cbor tag off it."""
        return v[6:]

strip_policy(v: str) -> str classmethod

Trim the extra bytes cbor tag off it.

Source code in src/charli3_dendrite/dexs/ob/axo.py

@field_validator("policy_script")
@classmethod
def strip_policy(cls, v: str) -> str:
    """Trim the extra bytes cbor tag off it."""
    return v[6:]

AxoOBMarketState

Bases: AbstractOrderBookState

Source code in src/charli3_dendrite/dexs/ob/axo.py

class AxoOBMarketState(AbstractOrderBookState):
    fee: int = 10
    spot: float = 1
    plutus_v2: bool = True
    inactive: bool = False
    _stake_address: ClassVar[Address] = Address.decode(
        "addr1z92l7rnra7sxjn5qv5fzc4fwsrrm29mgkleqj9a0y46j5lrryf9mtf9layje8u7u7wmap6alr28l90ry5t9nlyldjjsse4mxc9",
    )
    _client: ClassVar[AxoAPIClient] = AxoAPIClient()
    _reference_utxo: ClassVar[UTxO | None] = None
    _deposit: Assets = Assets(lovelace=8000000)

    @classmethod
    def dex(cls) -> str:
        return "Axo"

    @classmethod
    def order_selector(self) -> list[str]:
        """Order selection information."""
        addresses = [
            "addr1z92l7rnra7sxjn5qv5fzc4fwsrrm29mgkleqj9a0y46j5lrryf9mtf9layje8u7u7wmap6alr28l90ry5t9nlyldjjsse4mxc9",
        ]
        return addresses

    @classmethod
    def pool_selector(self) -> PoolSelector:
        """Pool selection information."""
        return []

    @property
    def swap_forward(self) -> bool:
        return False

    @classmethod
    def default_script_class(cls):
        return PlutusV2Script

    @classmethod
    def reference_utxo(cls) -> UTxO | None:
        if cls._reference_utxo is None:
            script_reference = get_backend().get_script_from_address(cls._stake_address)

            script = cls.default_script_class()(bytes.fromhex(script_reference.script))

            cls._reference_utxo = UTxO(
                input=TransactionInput(
                    transaction_id=TransactionId(
                        bytes.fromhex(
                            script_reference.tx_hash,
                        ),
                    ),
                    index=script_reference.tx_index,
                ),
                output=TransactionOutput(
                    address=Address.decode(script_reference.address),
                    amount=asset_to_value(script_reference.assets),
                    script=script,
                ),
            )
        return cls._reference_utxo

    @classmethod
    def _process_ob(
        self,
        ob: AxoOBResponse,
    ) -> tuple[list[OrderBookOrder], list[OrderBookOrder]]:
        prices = get_token_prices(assets=[ob.left, ob.right])
        left = ob.left if ob.left != "" else "lovelace"
        if prices[0].policy_id + prices[0].policy_name == left:
            token_a_decimals = prices[0].decimals
            token_b_decimals = prices[1].decimals
        else:
            token_a_decimals = prices[1].decimals
            token_b_decimals = prices[0].decimals

        sell_book = []
        for index in range(len(ob.sell_side_price)):
            sell_book.append(
                OrderBookOrder(
                    price=ob.sell_side_price[index]
                    * 10 ** (token_a_decimals - token_b_decimals),
                    quantity=int(ob.sell_side_amount[index] * 10**token_b_decimals),
                ),
            )
        buy_book = []
        for index in range(len(ob.buy_side_price)):
            buy_book.append(
                OrderBookOrder(
                    price=ob.buy_side_price[index] ** -1
                    * 10 ** (token_b_decimals - token_a_decimals),
                    quantity=int(
                        ob.buy_side_amount[index]
                        * 10**token_a_decimals
                        * ob.buy_side_price[index],
                    ),
                ),
            )

        return BuyOrderBook(buy_book), SellOrderBook(sell_book)

    @classmethod
    def get_book(cls, assets: Assets) -> "AxoOBMarketState":
        aob, ob, spot = cls._client.get_ob_info(assets)

        if spot is None:
            raise InvalidPoolError

        try:
            buy_book, sell_book = cls._process_ob(ob=aob)
            buy_book_full, sell_book_full = cls._process_ob(ob=ob)
        except IndexError:
            logger.error(f"Error getting Axo order book for assets: {assets}")
            raise InvalidPoolError

        if "lovelace" in assets:
            spot = 1.0

        instance = cls(
            assets=assets,
            spot=spot,
            block_time=int(datetime.now().timestamp()),
            block_index=0,
            buy_book_full=buy_book_full,
            sell_book_full=sell_book_full,
        )

        return instance

    @classmethod
    def order_datum_class(self) -> type[PlutusData]:
        return AxoOrderDatum

    @property
    def stake_address(self) -> Address:
        return self._stake_address

    def swap_utxo(
        self,
        address_source: Address,
        in_assets: Assets,
        out_assets: Assets,
        tx_builder: TransactionBuilder,
        extra_assets: Assets | None = None,
        address_target: Address | None = None,
        datum_target: PlutusData | None = None,
    ) -> tuple[TransactionOutput, PlutusData, UTxO]:
        # Basic checks
        if len(in_assets) != 1 or len(out_assets) != 1:
            raise ValueError(
                "Only one asset can be supplied as input, "
                + "and one asset supplied as output.",
            )

        # Get the mint input UTxO
        utxo_input = None
        for utxo in tx_builder.inputs:
            if utxo.output.amount.coin > 1200000:
                if utxo_input is None or len(utxo_input.output.to_cbor_hex()) < len(
                    utxo.output.to_cbor_hex(),
                ):
                    utxo_input = utxo

        # Get the order build info
        prices = get_token_prices(assets=[in_assets.unit(), out_assets.unit()])
        if prices[0].policy_id + prices[0].policy_name == in_assets.unit():
            in_decimals = prices[0].decimals
            out_decimals = prices[1].decimals
        else:
            in_decimals = prices[1].decimals
            out_decimals = prices[0].decimals
        params = AxoCreateParams(
            left=in_assets.unit() if in_assets.unit() != "lovelace" else "",
            right=out_assets.unit() if out_assets.unit() != "lovelace" else "",
            amount=in_assets.quantity() / 10**in_decimals,
            slippage=self.slippage(in_assets=in_assets, out_assets=out_assets) + 2.0,
        )
        create: AxoCreateResponse = self._client.create(
            wallet_addr=address_source.encode(),
            tx_hash=utxo_input.input.transaction_id.payload.hex(),
            tx_idx=utxo_input.input.index,
            params=params,
        )

        # Create the mint metadata
        metadata = AuxiliaryData(
            AlonzoMetadata(metadata=Metadata({721: create.nft_metadata})),
        )
        tx_builder.auxiliary_data = metadata

        # Create the axo receipt
        axo_receipt = MultiAsset.from_primitive(
            {
                bytes.fromhex(create.strat_id): {
                    bytes.fromhex(create.token_name): 2,
                },
            },
        )
        tx_builder.mint = axo_receipt
        redeemer = Redeemer(CancelRedeemer())
        redeemer.tag = RedeemerTag.MINT
        tx_builder.add_minting_script(
            PlutusV2Script(bytes.fromhex(create.policy_script)),
            redeemer,
        )

        # Add in the recommended lovelace to the input and the axo receipt
        in_assets.root["lovelace"] = (
            in_assets["lovelace"]
            + self.batcher_fee(
                in_assets=in_assets,
                out_assets=out_assets,
                extra_assets=extra_assets,
            ).quantity()
            + self.deposit(in_assets=in_assets, out_assets=out_assets).quantity()
        )
        in_assets.root[create.strat_id + create.token_name] = 1

        # Create the swap utxo
        order_datum = AxoOrderDatum.from_cbor(create.datum)
        output = TransactionOutput(
            address=create.algo_addr,
            amount=asset_to_value(in_assets),
            datum=order_datum,
        )
        tx_builder.add_output(output)

        # Create the receipt UTxO
        utxo = TransactionOutput(
            address=address_source,
            amount=Value(
                coin=1000000,
                multi_asset=MultiAsset.from_primitive(
                    {
                        bytes.fromhex(create.strat_id): {
                            bytes.fromhex(create.token_name): 1,
                        },
                    },
                ),
            ),
        )
        utxo.amount.coin = min_lovelace(context=tx_builder.context, output=utxo)
        tx_builder.add_output(utxo)

        return output, order_datum, utxo_input

    @property
    def volume_fee(self) -> int:
        return 10

    @classmethod
    def cancel_redeemer(cls) -> PlutusData:
        return Redeemer(AxoCancelRedeemer())

    def batcher_fee(
        self,
        in_assets: Assets | None = None,
        out_assets: Assets | None = None,
        extra_assets: Assets | None = None,
    ) -> Assets:
        """Batcher fee.

        Args:
            in_assets: The input assets for the swap
            out_assets: The output assets for the swap
            extra_assets: Extra assets included in the transaction
        """
        if in_assets.unit() == "lovelace":
            fees = max(self.volume_fee * in_assets.quantity() // 10000, 1200000)
        elif out_assets.unit() == "lovelace":
            fees = max(self.volume_fee * out_assets.quantity() // 10000, 1200000)
        else:
            fees = max(
                self.volume_fee * in_assets.quantity() * self.spot // 10000,
                1200000,
            )

        # The below code estimates the Cardano cost of executing the tx
        fees += 250000  # ~cost of output tx

        if in_assets.unit() == self.unit_a:
            book = self.sell_book_full
        else:
            book = self.buy_book_full

        # Each fill order incurs ~0.6 ada cost
        index = 0
        in_quantity = in_assets.quantity()
        while in_quantity > 0 and index < len(book):
            available = book[index].quantity * book[index].price
            fees += 600000
            if available > in_quantity:
                in_quantity = 0
            else:
                in_quantity -= book[index].price * book[index].quantity
            index += 1

        return Assets(lovelace=fees)

    def slippage(
        self,
        in_assets: Assets | None = None,
        out_assets: Assets | None = None,
    ) -> Assets:
        """Calculate slippage.

        Args:
            in_assets: The input assets for the swap
            out_assets: The output assets for the swap
            extra_assets: Extra assets included in the transaction
        """
        if in_assets.unit() == "lovelace":
            fees = max(self.volume_fee * in_assets.quantity() // 10000, 1200000)
        elif out_assets.unit() == "lovelace":
            fees = max(self.volume_fee * out_assets.quantity() // 10000, 1200000)
        else:
            fees = max(
                self.volume_fee * in_assets.quantity() * self.spot // 10000,
                1200000,
            )

        # The below code estimates the Cardano cost of executing the tx
        fees += 250000  # ~cost of output tx

        if in_assets.unit() == self.unit_a:
            book = self.sell_book_full
        else:
            book = self.buy_book_full

        # Each fill order incurs ~0.5 ada cost
        index = 0
        best_price = book[index].price
        in_quantity = in_assets.quantity()
        while in_quantity > 0 and index < len(book):
            available = book[index].quantity * book[index].price
            fees += 500000
            if available > in_quantity:
                in_quantity = 0
            else:
                in_quantity -= book[index].price * book[index].quantity
            last_price = book[index].price
            index += 1

        return 100 * abs(1 - (best_price / last_price))

    @property
    def pool_id(self):
        return ".".join([self.dex, self.unit_a, self.unit_b])

batcher_fee(in_assets: Assets | None = None, out_assets: Assets | None = None, extra_assets: Assets | None = None) -> Assets

Batcher fee.

Parameters:

Name	Type	Description	Default
in_assets	Assets | None	The input assets for the swap	None
out_assets	Assets | None	The output assets for the swap	None
extra_assets	Assets | None	Extra assets included in the transaction	None

Source code in src/charli3_dendrite/dexs/ob/axo.py

def batcher_fee(
    self,
    in_assets: Assets | None = None,
    out_assets: Assets | None = None,
    extra_assets: Assets | None = None,
) -> Assets:
    """Batcher fee.

    Args:
        in_assets: The input assets for the swap
        out_assets: The output assets for the swap
        extra_assets: Extra assets included in the transaction
    """
    if in_assets.unit() == "lovelace":
        fees = max(self.volume_fee * in_assets.quantity() // 10000, 1200000)
    elif out_assets.unit() == "lovelace":
        fees = max(self.volume_fee * out_assets.quantity() // 10000, 1200000)
    else:
        fees = max(
            self.volume_fee * in_assets.quantity() * self.spot // 10000,
            1200000,
        )

    # The below code estimates the Cardano cost of executing the tx
    fees += 250000  # ~cost of output tx

    if in_assets.unit() == self.unit_a:
        book = self.sell_book_full
    else:
        book = self.buy_book_full

    # Each fill order incurs ~0.6 ada cost
    index = 0
    in_quantity = in_assets.quantity()
    while in_quantity > 0 and index < len(book):
        available = book[index].quantity * book[index].price
        fees += 600000
        if available > in_quantity:
            in_quantity = 0
        else:
            in_quantity -= book[index].price * book[index].quantity
        index += 1

    return Assets(lovelace=fees)

order_selector() -> list[str] classmethod

Order selection information.

Source code in src/charli3_dendrite/dexs/ob/axo.py

@classmethod
def order_selector(self) -> list[str]:
    """Order selection information."""
    addresses = [
        "addr1z92l7rnra7sxjn5qv5fzc4fwsrrm29mgkleqj9a0y46j5lrryf9mtf9layje8u7u7wmap6alr28l90ry5t9nlyldjjsse4mxc9",
    ]
    return addresses

pool_selector() -> PoolSelector classmethod

Pool selection information.

Source code in src/charli3_dendrite/dexs/ob/axo.py

@classmethod
def pool_selector(self) -> PoolSelector:
    """Pool selection information."""
    return []

slippage(in_assets: Assets | None = None, out_assets: Assets | None = None) -> Assets

Calculate slippage.

Parameters:

Name	Type	Description	Default
in_assets	Assets | None	The input assets for the swap	None
out_assets	Assets | None	The output assets for the swap	None
extra_assets		Extra assets included in the transaction	required

Source code in src/charli3_dendrite/dexs/ob/axo.py

def slippage(
    self,
    in_assets: Assets | None = None,
    out_assets: Assets | None = None,
) -> Assets:
    """Calculate slippage.

    Args:
        in_assets: The input assets for the swap
        out_assets: The output assets for the swap
        extra_assets: Extra assets included in the transaction
    """
    if in_assets.unit() == "lovelace":
        fees = max(self.volume_fee * in_assets.quantity() // 10000, 1200000)
    elif out_assets.unit() == "lovelace":
        fees = max(self.volume_fee * out_assets.quantity() // 10000, 1200000)
    else:
        fees = max(
            self.volume_fee * in_assets.quantity() * self.spot // 10000,
            1200000,
        )

    # The below code estimates the Cardano cost of executing the tx
    fees += 250000  # ~cost of output tx

    if in_assets.unit() == self.unit_a:
        book = self.sell_book_full
    else:
        book = self.buy_book_full

    # Each fill order incurs ~0.5 ada cost
    index = 0
    best_price = book[index].price
    in_quantity = in_assets.quantity()
    while in_quantity > 0 and index < len(book):
        available = book[index].quantity * book[index].price
        fees += 500000
        if available > in_quantity:
            in_quantity = 0
        else:
            in_quantity -= book[index].price * book[index].quantity
        last_price = book[index].price
        index += 1

    return 100 * abs(1 - (best_price / last_price))