from . import xdr as stellar_xdr
from .asset import Asset
from .utils import sha256
__all__ = ["LiquidityPoolAsset", "LIQUIDITY_POOL_FEE_V18"]
#: LIQUIDITY_POOL_FEE_V18 is the default liquidity pool fee in protocol v18.
#: It defaults to 30 base points (0.3%).
LIQUIDITY_POOL_FEE_V18 = stellar_xdr.LIQUIDITY_POOL_FEE_V18
[docs]
class LiquidityPoolAsset:
"""The :class:`LiquidityPoolAsset` object, which represents a liquidity pool trustline change.
:param asset_a: The first asset in the Pool, it must respect the rule asset_a < asset_b.
See :func:`stellar_sdk.liquidity_pool_asset.LiquidityPoolAsset.is_valid_lexicographic_order`
for more details on how assets are sorted.
:param asset_b: The second asset in the Pool, it must respect the rule asset_a < asset_b.
See :func:`stellar_sdk.liquidity_pool_asset.LiquidityPoolAsset.is_valid_lexicographic_order`
for more details on how assets are sorted.
:param fee: The liquidity pool fee. For now the only fee supported is `30`.
:raise: :exc:`ValueError <stellar_sdk.exceptions.ValueError>`
"""
def __init__(
self, asset_a: Asset, asset_b: Asset, fee: int = LIQUIDITY_POOL_FEE_V18
) -> None:
if not self.is_valid_lexicographic_order(asset_a, asset_b):
raise ValueError("`Assets are not in lexicographic order.")
if fee != LIQUIDITY_POOL_FEE_V18:
raise ValueError("`fee` is invalid.")
self.type: str = "liquidity_pool_shares"
self.asset_a: Asset = asset_a
self.asset_b: Asset = asset_b
self.fee: int = fee
@property
def liquidity_pool_id(self) -> str:
"""Computes the liquidity pool id for current instance.
:return: Liquidity pool id.
"""
return self._liquidity_pool_id_bytes.hex()
[docs]
@staticmethod
def is_valid_lexicographic_order(asset_a: Asset, asset_b: Asset) -> bool:
"""Compares if asset_a < asset_b according with the criteria:
1. First compare the type (eg. native before alphanum4 before alphanum12).
2. If the types are equal, compare the assets codes.
3. If the asset codes are equal, compare the issuers.
:param asset_a: The first asset in the lexicographic order.
:param asset_b: The second asset in the lexicographic order.
:return: return `True` if asset_a < asset_b
"""
if asset_a == asset_b:
return False
# Compare asset types.
if asset_a.type == "native":
return True
elif asset_a.type == "credit_alphanum4":
if asset_b.type == "native":
return False
if asset_b.type == "credit_alphanum12":
return True
else:
# asset_a.type == "credit_alphanum12"
if asset_b.type != "credit_alphanum12":
return False
# Compare asset codes.
if asset_a.code != asset_b.code:
assert asset_a.code is not None
assert asset_b.code is not None
return asset_a.code < asset_b.code
# Compare asset issuer.
assert asset_a.issuer is not None
assert asset_b.issuer is not None
return asset_a.issuer < asset_b.issuer
[docs]
def to_change_trust_asset_xdr_object(self) -> stellar_xdr.ChangeTrustAsset:
"""Returns the xdr object for this ChangeTrustAsset object.
:return: XDR ChangeTrustAsset object
"""
liquidity_pool_parameters = self._liquidity_pool_parameters()
return stellar_xdr.ChangeTrustAsset(
stellar_xdr.AssetType.ASSET_TYPE_POOL_SHARE,
liquidity_pool=liquidity_pool_parameters,
)
[docs]
@classmethod
def from_xdr_object(
cls, xdr_object: stellar_xdr.ChangeTrustAsset
) -> "LiquidityPoolAsset":
"""Create a :class:`LiquidityPoolAsset` from an XDR ChangeTrustAsset object.
:param xdr_object: The XDR ChangeTrustAsset object.
:return: A new :class:`LiquidityPoolAsset` object from the given XDR ChangeTrustAsset object.
"""
asset_type = xdr_object.type
if asset_type == stellar_xdr.AssetType.ASSET_TYPE_POOL_SHARE:
assert xdr_object.liquidity_pool is not None
assert xdr_object.liquidity_pool.constant_product is not None
asset_a = Asset.from_xdr_object(
xdr_object.liquidity_pool.constant_product.asset_a
)
asset_b = Asset.from_xdr_object(
xdr_object.liquidity_pool.constant_product.asset_b
)
fee = xdr_object.liquidity_pool.constant_product.fee.int32
return cls(asset_a, asset_b, fee)
else:
raise ValueError(f"Unexpected asset type: {asset_type}")
@property
def _liquidity_pool_id_bytes(self) -> bytes:
liquidity_pool_parameters = self._liquidity_pool_parameters()
return sha256(liquidity_pool_parameters.to_xdr_bytes())
def _liquidity_pool_parameters(self) -> stellar_xdr.LiquidityPoolParameters:
liquidity_pool_constant_product_parameters = (
stellar_xdr.LiquidityPoolConstantProductParameters(
self.asset_a.to_xdr_object(),
self.asset_b.to_xdr_object(),
stellar_xdr.Int32(self.fee),
)
)
return stellar_xdr.LiquidityPoolParameters(
stellar_xdr.LiquidityPoolType.LIQUIDITY_POOL_CONSTANT_PRODUCT,
liquidity_pool_constant_product_parameters,
)
def __hash__(self):
return hash((self.asset_a, self.asset_b, self.fee, self.type))
def __eq__(self, other: object) -> bool:
if not isinstance(other, self.__class__):
return NotImplemented
return (
self.asset_a == other.asset_a
and self.asset_b == other.asset_b
and self.fee == other.fee
)
def __str__(self):
return f"<LiquidityPoolAsset [asset_a={self.asset_a}, asset_b={self.asset_b}, fee={self.fee}, type={self.type}]>"