Skip to content

Core base

Abstract base class and common functions for handling token pairs.

AbstractPairState

Bases: DendriteBaseModel, ABC

Abstract base class representing the state of a pair.

Source code in src/charli3_dendrite/dexs/core/base.py 
 21
 22
 23
 24
 25
 26
 27
 28
 29
 30
 31
 32
 33
 34
 35
 36
 37
 38
 39
 40
 41
 42
 43
 44
 45
 46
 47
 48
 49
 50
 51
 52
 53
 54
 55
 56
 57
 58
 59
 60
 61
 62
 63
 64
 65
 66
 67
 68
 69
 70
 71
 72
 73
 74
 75
 76
 77
 78
 79
 80
 81
 82
 83
 84
 85
 86
 87
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
138
139
140
141
142
143
144
145
146
147
148
149
150
151
152
153
154
155
156
157
158
159
160
161
162
163
164
165
166
167
168
169
170
171
172
173
174
175
176
177
178
179
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
202
203
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
class AbstractPairState(DendriteBaseModel, ABC):
    """Abstract base class representing the state of a pair."""

    assets: Assets
    block_time: int
    block_index: int
    fee: int | list[int] | None = None
    plutus_v2: bool
    tx_index: int | None = None
    tx_hash: str | None = None
    datum_cbor: str | None = None
    datum_hash: str | None = None
    dex_nft: Assets | None = None

    _batcher_fee: Assets
    _datum_parsed: PlutusData

    @classmethod
    @abstractmethod
    def dex(cls) -> str:
        """Official dex name."""
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def order_selector(cls) -> list[str]:
        """Order selection information."""
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def pool_selector(cls) -> PoolSelector:
        """Pool selection information."""
        raise NotImplementedError

    @abstractmethod
    def get_amount_out(self, asset: Assets) -> tuple[Assets, float]:
        """Calculate the output amount of assets for given input.

        Args:
            asset: An asset with a defined quantity.

        Returns:
            A tuple where the first value is the estimated asset returned and
            the second value is the price impact ratio.
        """
        raise NotImplementedError

    @abstractmethod
    def get_amount_in(self, asset: Assets) -> tuple[Assets, float]:
        """Get the input asset amount given a desired output asset amount.

        Args:
            asset: An asset with a defined quantity.

        Returns:
           A tuple where the first value is the the estimated asset needed and
           the second value is the slippage fee.
        """
        raise NotImplementedError

    @property
    @abstractmethod
    def swap_forward(self) -> bool:
        """Returns if swap forwarding is enabled."""
        raise NotImplementedError

    @property
    def inline_datum(self) -> bool:
        """Determine whether the datum should be inline."""
        return self.plutus_v2

    @classmethod
    def reference_utxo(cls) -> UTxO | None:
        """Get Reference UTXO.

        Returns:
            UTxO | None: UTxO object if it exists, otherwise None.
        """
        return None

    @property
    @abstractmethod
    def stake_address(self) -> Address:
        """Return the staking address."""
        raise NotImplementedError

    @classmethod
    @abstractmethod
    def order_datum_class(cls) -> type[PlutusData]:
        """Returns data class used for handling order datums."""
        raise NotImplementedError

    @classmethod
    def default_script_class(cls) -> type[PlutusV1Script] | type[PlutusV2Script]:
        """Get default script class as Plutus V1 unless overridden.

        Returns:
            type[PlutusV1Script] | type[PlutusV2Script]: The default script class.
        """
        return PlutusV1Script

    @property
    def script_class(self) -> type[PlutusV1Script] | type[PlutusV2Script]:
        """Returns the script class based on the Plutus version being used."""
        if self.plutus_v2:
            return PlutusV2Script
        return PlutusV1Script

    def swap_datum(
        self,
        address_source: Address,
        in_assets: Assets,
        out_assets: Assets,
        extra_assets: Assets | None = None,
        address_target: Address | None = None,
        datum_target: PlutusData | None = None,
    ) -> PlutusData:
        """Constructs the datum for a swap transaction."""
        raise NotImplementedError

    @abstractmethod
    def swap_utxo(
        self,
        address_source: Address,
        in_assets: Assets,
        out_assets: Assets,
        extra_assets: Assets | None = None,
        address_target: Address | None = None,
        datum_target: PlutusData | None = None,
    ) -> TransactionOutput:
        """Constructs the transaction output for a swap."""
        raise NotImplementedError

    @property
    def volume_fee(self) -> int | float | list[int] | list[float] | None:
        """Swap fee of swap in basis points."""
        return self.fee

    @classmethod
    def cancel_redeemer(cls) -> PlutusData:
        """Returns the redeemer data for canceling transaction."""
        return Redeemer(CancelRedeemer())

    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
        """
        return self._batcher

    def deposit(
        self,
        in_assets: Assets | None = None,
        out_assets: Assets | None = None,
    ) -> Assets:
        """Batcher fee."""
        return self._deposit

    @classmethod
    def dex_policy(cls) -> list[str] | None:
        """The dex nft policy.

        This should be the policy or policy+name of the dex nft.

        If None, then the default dex nft check is skipped.

        Returns:
            Optional[str]: policy or policy+name of dex nft
        """
        return None

    @property
    def unit_a(self) -> str:
        """Token name of asset A."""
        return self.assets.unit(0)

    @property
    def unit_b(self) -> str:
        """Token name of asset b."""
        return self.assets.unit(1)

    @property
    def reserve_a(self) -> int:
        """Reserve amount of asset A."""
        return self.assets.quantity(0)

    @property
    def reserve_b(self) -> int:
        """Reserve amount of asset B."""
        return self.assets.quantity(1)

    @property
    @abstractmethod
    def price(self) -> tuple[Decimal, Decimal]:
        """Price of assets.

        Returns:
            A `Tuple[Decimal, Decimal] where the first `Decimal` is the price to buy
                1 of token B in units of token A, and the second `Decimal` is the price
                to buy 1 of token A in units of token B.
        """
        raise NotImplementedError

    @property
    @abstractmethod
    def tvl(self) -> Decimal:
        """Return the total value locked for the pool.

        Raises:
            NotImplementedError: Only ADA pool TVL is implemented.
        """
        raise NotImplementedError

    @property
    @abstractmethod
    def pool_id(self) -> str:
        """A unique identifier for the pool or ob.

        Raises:
            NotImplementedError: Only ADA pool TVL is implemented.
        """
        raise NotImplementedError

inline_datum: bool property

Determine whether the datum should be inline.

pool_id: str abstractmethod property

A unique identifier for the pool or ob.

Raises:

Type Description
NotImplementedError

Only ADA pool TVL is implemented.

price: tuple[Decimal, Decimal] abstractmethod property

Price of assets.

Returns:

Type Description
tuple[Decimal, Decimal]

A Tuple[Decimal, Decimal] where the firstDecimalis the price to buy 1 of token B in units of token A, and the secondDecimal` is the price to buy 1 of token A in units of token B.

reserve_a: int property

Reserve amount of asset A.

reserve_b: int property

Reserve amount of asset B.

script_class: type[PlutusV1Script] | type[PlutusV2Script] property

Returns the script class based on the Plutus version being used.

stake_address: Address abstractmethod property

Return the staking address.

swap_forward: bool abstractmethod property

Returns if swap forwarding is enabled.

tvl: Decimal abstractmethod property

Return the total value locked for the pool.

Raises:

Type Description
NotImplementedError

Only ADA pool TVL is implemented.

unit_a: str property

Token name of asset A.

unit_b: str property

Token name of asset b.

volume_fee: int | float | list[int] | list[float] | None property

Swap fee of swap in basis points.

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/core/base.py 
165
166
167
168
169
170
171
172
173
174
175
176
177
178
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
    """
    return self._batcher

cancel_redeemer() -> PlutusData classmethod

Returns the redeemer data for canceling transaction.

Source code in src/charli3_dendrite/dexs/core/base.py 
160
161
162
163
@classmethod
def cancel_redeemer(cls) -> PlutusData:
    """Returns the redeemer data for canceling transaction."""
    return Redeemer(CancelRedeemer())

default_script_class() -> type[PlutusV1Script] | type[PlutusV2Script] classmethod

Get default script class as Plutus V1 unless overridden.

Returns:

Type Description
type[PlutusV1Script] | type[PlutusV2Script]

type[PlutusV1Script] | type[PlutusV2Script]: The default script class.
Source code in src/charli3_dendrite/dexs/core/base.py 
114
115
116
117
118
119
120
121
@classmethod
def default_script_class(cls) -> type[PlutusV1Script] | type[PlutusV2Script]:
    """Get default script class as Plutus V1 unless overridden.

    Returns:
        type[PlutusV1Script] | type[PlutusV2Script]: The default script class.
    """
    return PlutusV1Script

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

Batcher fee.

Source code in src/charli3_dendrite/dexs/core/base.py 
180
181
182
183
184
185
186
def deposit(
    self,
    in_assets: Assets | None = None,
    out_assets: Assets | None = None,
) -> Assets:
    """Batcher fee."""
    return self._deposit

dex() -> str abstractmethod classmethod

Official dex name.

Source code in src/charli3_dendrite/dexs/core/base.py 
38
39
40
41
42
@classmethod
@abstractmethod
def dex(cls) -> str:
    """Official dex name."""
    raise NotImplementedError

dex_policy() -> list[str] | None classmethod

The dex nft policy.

This should be the policy or policy+name of the dex nft.

If None, then the default dex nft check is skipped.

Returns:

Type Description
list[str] | None

Optional[str]: policy or policy+name of dex nft
Source code in src/charli3_dendrite/dexs/core/base.py 
188
189
190
191
192
193
194
195
196
197
198
199
@classmethod
def dex_policy(cls) -> list[str] | None:
    """The dex nft policy.

    This should be the policy or policy+name of the dex nft.

    If None, then the default dex nft check is skipped.

    Returns:
        Optional[str]: policy or policy+name of dex nft
    """
    return None

get_amount_in(asset: Assets) -> tuple[Assets, float] abstractmethod

Get the input asset amount given a desired output asset amount.

Parameters:

Name Type Description Default
asset Assets

An asset with a defined quantity.

 required

Returns:

Type Description
Assets

A tuple where the first value is the the estimated asset needed and
float

the second value is the slippage fee.
Source code in src/charli3_dendrite/dexs/core/base.py 
69
70
71
72
73
74
75
76
77
78
79
80
@abstractmethod
def get_amount_in(self, asset: Assets) -> tuple[Assets, float]:
    """Get the input asset amount given a desired output asset amount.

    Args:
        asset: An asset with a defined quantity.

    Returns:
       A tuple where the first value is the the estimated asset needed and
       the second value is the slippage fee.
    """
    raise NotImplementedError

get_amount_out(asset: Assets) -> tuple[Assets, float] abstractmethod

Calculate the output amount of assets for given input.

Parameters:

Name Type Description Default
asset Assets

An asset with a defined quantity.

 required

Returns:

Type Description
Assets

A tuple where the first value is the estimated asset returned and
float

the second value is the price impact ratio.
Source code in src/charli3_dendrite/dexs/core/base.py 
56
57
58
59
60
61
62
63
64
65
66
67
@abstractmethod
def get_amount_out(self, asset: Assets) -> tuple[Assets, float]:
    """Calculate the output amount of assets for given input.

    Args:
        asset: An asset with a defined quantity.

    Returns:
        A tuple where the first value is the estimated asset returned and
        the second value is the price impact ratio.
    """
    raise NotImplementedError

order_datum_class() -> type[PlutusData] abstractmethod classmethod

Returns data class used for handling order datums.

Source code in src/charli3_dendrite/dexs/core/base.py 
108
109
110
111
112
@classmethod
@abstractmethod
def order_datum_class(cls) -> type[PlutusData]:
    """Returns data class used for handling order datums."""
    raise NotImplementedError

order_selector() -> list[str] abstractmethod classmethod

Order selection information.

Source code in src/charli3_dendrite/dexs/core/base.py 
44
45
46
47
48
@classmethod
@abstractmethod
def order_selector(cls) -> list[str]:
    """Order selection information."""
    raise NotImplementedError

pool_selector() -> PoolSelector abstractmethod classmethod

Pool selection information.

Source code in src/charli3_dendrite/dexs/core/base.py 
50
51
52
53
54
@classmethod
@abstractmethod
def pool_selector(cls) -> PoolSelector:
    """Pool selection information."""
    raise NotImplementedError

reference_utxo() -> UTxO | None classmethod

Get Reference UTXO.

Returns:

Type Description
UTxO | None

UTxO | None: UTxO object if it exists, otherwise None.
Source code in src/charli3_dendrite/dexs/core/base.py 
 93
 94
 95
 96
 97
 98
 99
100
@classmethod
def reference_utxo(cls) -> UTxO | None:
    """Get Reference UTXO.

    Returns:
        UTxO | None: UTxO object if it exists, otherwise None.
    """
    return None

swap_datum(address_source: Address, in_assets: Assets, out_assets: Assets, extra_assets: Assets | None = None, address_target: Address | None = None, datum_target: PlutusData | None = None) -> PlutusData

Constructs the datum for a swap transaction.

Source code in src/charli3_dendrite/dexs/core/base.py 
130
131
132
133
134
135
136
137
138
139
140
def swap_datum(
    self,
    address_source: Address,
    in_assets: Assets,
    out_assets: Assets,
    extra_assets: Assets | None = None,
    address_target: Address | None = None,
    datum_target: PlutusData | None = None,
) -> PlutusData:
    """Constructs the datum for a swap transaction."""
    raise NotImplementedError

swap_utxo(address_source: Address, in_assets: Assets, out_assets: Assets, extra_assets: Assets | None = None, address_target: Address | None = None, datum_target: PlutusData | None = None) -> TransactionOutput abstractmethod

Constructs the transaction output for a swap.

Source code in src/charli3_dendrite/dexs/core/base.py 
142
143
144
145
146
147
148
149
150
151
152
153
@abstractmethod
def swap_utxo(
    self,
    address_source: Address,
    in_assets: Assets,
    out_assets: Assets,
    extra_assets: Assets | None = None,
    address_target: Address | None = None,
    datum_target: PlutusData | None = None,
) -> TransactionOutput:
    """Constructs the transaction output for a swap."""
    raise NotImplementedError