Module orca_whirlpool.internal.anchor.accounts.whirlpool

Classes

class Whirlpool (whirlpools_config: solders.pubkey.Pubkey, whirlpool_bump: list[int], tick_spacing: int, tick_spacing_seed: list[int], fee_rate: int, protocol_fee_rate: int, liquidity: int, sqrt_price: int, tick_current_index: int, protocol_fee_owed_a: int, protocol_fee_owed_b: int, token_mint_a: solders.pubkey.Pubkey, token_vault_a: solders.pubkey.Pubkey, fee_growth_global_a: int, token_mint_b: solders.pubkey.Pubkey, token_vault_b: solders.pubkey.Pubkey, fee_growth_global_b: int, reward_last_updated_timestamp: int, reward_infos: list[WhirlpoolRewardInfo])

Whirlpool(whirlpools_config: solders.pubkey.Pubkey, whirlpool_bump: list[int], tick_spacing: int, tick_spacing_seed: list[int], fee_rate: int, protocol_fee_rate: int, liquidity: int, sqrt_price: int, tick_current_index: int, protocol_fee_owed_a: int, protocol_fee_owed_b: int, token_mint_a: solders.pubkey.Pubkey, token_vault_a: solders.pubkey.Pubkey, fee_growth_global_a: int, token_mint_b: solders.pubkey.Pubkey, token_vault_b: solders.pubkey.Pubkey, fee_growth_global_b: int, reward_last_updated_timestamp: int, reward_infos: list[orca_whirlpool.internal.anchor.types.whirlpool_reward_info.WhirlpoolRewardInfo])

Expand source code

@dataclass
class Whirlpool:
    discriminator: typing.ClassVar = b"?\x95\xd1\x0c\xe1\x80c\t"
    layout: typing.ClassVar = borsh.CStruct(
        "whirlpools_config" / BorshPubkey,
        "whirlpool_bump" / borsh.U8[1],
        "tick_spacing" / borsh.U16,
        "tick_spacing_seed" / borsh.U8[2],
        "fee_rate" / borsh.U16,
        "protocol_fee_rate" / borsh.U16,
        "liquidity" / borsh.U128,
        "sqrt_price" / borsh.U128,
        "tick_current_index" / borsh.I32,
        "protocol_fee_owed_a" / borsh.U64,
        "protocol_fee_owed_b" / borsh.U64,
        "token_mint_a" / BorshPubkey,
        "token_vault_a" / BorshPubkey,
        "fee_growth_global_a" / borsh.U128,
        "token_mint_b" / BorshPubkey,
        "token_vault_b" / BorshPubkey,
        "fee_growth_global_b" / borsh.U128,
        "reward_last_updated_timestamp" / borsh.U64,
        "reward_infos" / types.whirlpool_reward_info.WhirlpoolRewardInfo.layout[3],
    )
    whirlpools_config: Pubkey
    whirlpool_bump: list[int]
    tick_spacing: int
    tick_spacing_seed: list[int]
    fee_rate: int
    protocol_fee_rate: int
    liquidity: int
    sqrt_price: int
    tick_current_index: int
    protocol_fee_owed_a: int
    protocol_fee_owed_b: int
    token_mint_a: Pubkey
    token_vault_a: Pubkey
    fee_growth_global_a: int
    token_mint_b: Pubkey
    token_vault_b: Pubkey
    fee_growth_global_b: int
    reward_last_updated_timestamp: int
    reward_infos: list[types.whirlpool_reward_info.WhirlpoolRewardInfo]

    @classmethod
    async def fetch(
        cls,
        conn: AsyncClient,
        address: Pubkey,
        commitment: typing.Optional[Commitment] = None,
        program_id: Pubkey = PROGRAM_ID,
    ) -> typing.Optional["Whirlpool"]:
        resp = await conn.get_account_info(address, commitment=commitment)
        info = resp.value
        if info is None:
            return None
        if info.owner != program_id:
            raise ValueError("Account does not belong to this program")
        bytes_data = info.data
        return cls.decode(bytes_data)

    @classmethod
    async def fetch_multiple(
        cls,
        conn: AsyncClient,
        addresses: list[Pubkey],
        commitment: typing.Optional[Commitment] = None,
        program_id: Pubkey = PROGRAM_ID,
    ) -> typing.List[typing.Optional["Whirlpool"]]:
        infos = await get_multiple_accounts(conn, addresses, commitment=commitment)
        res: typing.List[typing.Optional["Whirlpool"]] = []
        for info in infos:
            if info is None:
                res.append(None)
                continue
            if info.account.owner != program_id:
                raise ValueError("Account does not belong to this program")
            res.append(cls.decode(info.account.data))
        return res

    @classmethod
    def decode(cls, data: bytes) -> "Whirlpool":
        if data[:ACCOUNT_DISCRIMINATOR_SIZE] != cls.discriminator:
            raise AccountInvalidDiscriminator(
                "The discriminator for this account is invalid"
            )
        dec = Whirlpool.layout.parse(data[ACCOUNT_DISCRIMINATOR_SIZE:])
        return cls(
            whirlpools_config=dec.whirlpools_config,
            whirlpool_bump=dec.whirlpool_bump,
            tick_spacing=dec.tick_spacing,
            tick_spacing_seed=dec.tick_spacing_seed,
            fee_rate=dec.fee_rate,
            protocol_fee_rate=dec.protocol_fee_rate,
            liquidity=dec.liquidity,
            sqrt_price=dec.sqrt_price,
            tick_current_index=dec.tick_current_index,
            protocol_fee_owed_a=dec.protocol_fee_owed_a,
            protocol_fee_owed_b=dec.protocol_fee_owed_b,
            token_mint_a=dec.token_mint_a,
            token_vault_a=dec.token_vault_a,
            fee_growth_global_a=dec.fee_growth_global_a,
            token_mint_b=dec.token_mint_b,
            token_vault_b=dec.token_vault_b,
            fee_growth_global_b=dec.fee_growth_global_b,
            reward_last_updated_timestamp=dec.reward_last_updated_timestamp,
            reward_infos=list(
                map(
                    lambda item: types.whirlpool_reward_info.WhirlpoolRewardInfo.from_decoded(
                        item
                    ),
                    dec.reward_infos,
                )
            ),
        )

    def to_json(self) -> WhirlpoolJSON:
        return {
            "whirlpools_config": str(self.whirlpools_config),
            "whirlpool_bump": self.whirlpool_bump,
            "tick_spacing": self.tick_spacing,
            "tick_spacing_seed": self.tick_spacing_seed,
            "fee_rate": self.fee_rate,
            "protocol_fee_rate": self.protocol_fee_rate,
            "liquidity": self.liquidity,
            "sqrt_price": self.sqrt_price,
            "tick_current_index": self.tick_current_index,
            "protocol_fee_owed_a": self.protocol_fee_owed_a,
            "protocol_fee_owed_b": self.protocol_fee_owed_b,
            "token_mint_a": str(self.token_mint_a),
            "token_vault_a": str(self.token_vault_a),
            "fee_growth_global_a": self.fee_growth_global_a,
            "token_mint_b": str(self.token_mint_b),
            "token_vault_b": str(self.token_vault_b),
            "fee_growth_global_b": self.fee_growth_global_b,
            "reward_last_updated_timestamp": self.reward_last_updated_timestamp,
            "reward_infos": list(map(lambda item: item.to_json(), self.reward_infos)),
        }

    @classmethod
    def from_json(cls, obj: WhirlpoolJSON) -> "Whirlpool":
        return cls(
            whirlpools_config=Pubkey.from_string(obj["whirlpools_config"]),
            whirlpool_bump=obj["whirlpool_bump"],
            tick_spacing=obj["tick_spacing"],
            tick_spacing_seed=obj["tick_spacing_seed"],
            fee_rate=obj["fee_rate"],
            protocol_fee_rate=obj["protocol_fee_rate"],
            liquidity=obj["liquidity"],
            sqrt_price=obj["sqrt_price"],
            tick_current_index=obj["tick_current_index"],
            protocol_fee_owed_a=obj["protocol_fee_owed_a"],
            protocol_fee_owed_b=obj["protocol_fee_owed_b"],
            token_mint_a=Pubkey.from_string(obj["token_mint_a"]),
            token_vault_a=Pubkey.from_string(obj["token_vault_a"]),
            fee_growth_global_a=obj["fee_growth_global_a"],
            token_mint_b=Pubkey.from_string(obj["token_mint_b"]),
            token_vault_b=Pubkey.from_string(obj["token_vault_b"]),
            fee_growth_global_b=obj["fee_growth_global_b"],
            reward_last_updated_timestamp=obj["reward_last_updated_timestamp"],
            reward_infos=list(
                map(
                    lambda item: types.whirlpool_reward_info.WhirlpoolRewardInfo.from_json(
                        item
                    ),
                    obj["reward_infos"],
                )
            ),
        )

Class variables

var discriminator : ClassVar

var fee_growth_global_a : int

var fee_growth_global_b : int

var fee_rate : int

var layout : ClassVar

var liquidity : int

var protocol_fee_owed_a : int

var protocol_fee_owed_b : int

var protocol_fee_rate : int

var reward_infos : list[WhirlpoolRewardInfo]

var reward_last_updated_timestamp : int

var sqrt_price : int

var tick_current_index : int

var tick_spacing : int

var tick_spacing_seed : list[int]

var token_mint_a : solders.pubkey.Pubkey

var token_mint_b : solders.pubkey.Pubkey

var token_vault_a : solders.pubkey.Pubkey

var token_vault_b : solders.pubkey.Pubkey

var whirlpool_bump : list[int]

var whirlpools_config : solders.pubkey.Pubkey

Static methods

def decode(data: bytes) ‑> Whirlpool

Expand source code

@classmethod
def decode(cls, data: bytes) -> "Whirlpool":
    if data[:ACCOUNT_DISCRIMINATOR_SIZE] != cls.discriminator:
        raise AccountInvalidDiscriminator(
            "The discriminator for this account is invalid"
        )
    dec = Whirlpool.layout.parse(data[ACCOUNT_DISCRIMINATOR_SIZE:])
    return cls(
        whirlpools_config=dec.whirlpools_config,
        whirlpool_bump=dec.whirlpool_bump,
        tick_spacing=dec.tick_spacing,
        tick_spacing_seed=dec.tick_spacing_seed,
        fee_rate=dec.fee_rate,
        protocol_fee_rate=dec.protocol_fee_rate,
        liquidity=dec.liquidity,
        sqrt_price=dec.sqrt_price,
        tick_current_index=dec.tick_current_index,
        protocol_fee_owed_a=dec.protocol_fee_owed_a,
        protocol_fee_owed_b=dec.protocol_fee_owed_b,
        token_mint_a=dec.token_mint_a,
        token_vault_a=dec.token_vault_a,
        fee_growth_global_a=dec.fee_growth_global_a,
        token_mint_b=dec.token_mint_b,
        token_vault_b=dec.token_vault_b,
        fee_growth_global_b=dec.fee_growth_global_b,
        reward_last_updated_timestamp=dec.reward_last_updated_timestamp,
        reward_infos=list(
            map(
                lambda item: types.whirlpool_reward_info.WhirlpoolRewardInfo.from_decoded(
                    item
                ),
                dec.reward_infos,
            )
        ),
    )

async def fetch(conn: solana.rpc.async_api.AsyncClient, address: solders.pubkey.Pubkey, commitment: Optional[solana.rpc.commitment.Commitment] = None, program_id: solders.pubkey.Pubkey = Pubkey( whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc, )) ‑> Optional[Whirlpool]

Expand source code

@classmethod
async def fetch(
    cls,
    conn: AsyncClient,
    address: Pubkey,
    commitment: typing.Optional[Commitment] = None,
    program_id: Pubkey = PROGRAM_ID,
) -> typing.Optional["Whirlpool"]:
    resp = await conn.get_account_info(address, commitment=commitment)
    info = resp.value
    if info is None:
        return None
    if info.owner != program_id:
        raise ValueError("Account does not belong to this program")
    bytes_data = info.data
    return cls.decode(bytes_data)

async def fetch_multiple(conn: solana.rpc.async_api.AsyncClient, addresses: list[solders.pubkey.Pubkey], commitment: Optional[solana.rpc.commitment.Commitment] = None, program_id: solders.pubkey.Pubkey = Pubkey( whirLbMiicVdio4qvUfM5KAg6Ct8VwpYzGff3uctyCc, )) ‑> List[Optional[Whirlpool]]

Expand source code

@classmethod
async def fetch_multiple(
    cls,
    conn: AsyncClient,
    addresses: list[Pubkey],
    commitment: typing.Optional[Commitment] = None,
    program_id: Pubkey = PROGRAM_ID,
) -> typing.List[typing.Optional["Whirlpool"]]:
    infos = await get_multiple_accounts(conn, addresses, commitment=commitment)
    res: typing.List[typing.Optional["Whirlpool"]] = []
    for info in infos:
        if info is None:
            res.append(None)
            continue
        if info.account.owner != program_id:
            raise ValueError("Account does not belong to this program")
        res.append(cls.decode(info.account.data))
    return res

def from_json(obj: WhirlpoolJSON) ‑> Whirlpool

Expand source code

@classmethod
def from_json(cls, obj: WhirlpoolJSON) -> "Whirlpool":
    return cls(
        whirlpools_config=Pubkey.from_string(obj["whirlpools_config"]),
        whirlpool_bump=obj["whirlpool_bump"],
        tick_spacing=obj["tick_spacing"],
        tick_spacing_seed=obj["tick_spacing_seed"],
        fee_rate=obj["fee_rate"],
        protocol_fee_rate=obj["protocol_fee_rate"],
        liquidity=obj["liquidity"],
        sqrt_price=obj["sqrt_price"],
        tick_current_index=obj["tick_current_index"],
        protocol_fee_owed_a=obj["protocol_fee_owed_a"],
        protocol_fee_owed_b=obj["protocol_fee_owed_b"],
        token_mint_a=Pubkey.from_string(obj["token_mint_a"]),
        token_vault_a=Pubkey.from_string(obj["token_vault_a"]),
        fee_growth_global_a=obj["fee_growth_global_a"],
        token_mint_b=Pubkey.from_string(obj["token_mint_b"]),
        token_vault_b=Pubkey.from_string(obj["token_vault_b"]),
        fee_growth_global_b=obj["fee_growth_global_b"],
        reward_last_updated_timestamp=obj["reward_last_updated_timestamp"],
        reward_infos=list(
            map(
                lambda item: types.whirlpool_reward_info.WhirlpoolRewardInfo.from_json(
                    item
                ),
                obj["reward_infos"],
            )
        ),
    )

Methods

def to_json(self) ‑> WhirlpoolJSON

Expand source code

def to_json(self) -> WhirlpoolJSON:
    return {
        "whirlpools_config": str(self.whirlpools_config),
        "whirlpool_bump": self.whirlpool_bump,
        "tick_spacing": self.tick_spacing,
        "tick_spacing_seed": self.tick_spacing_seed,
        "fee_rate": self.fee_rate,
        "protocol_fee_rate": self.protocol_fee_rate,
        "liquidity": self.liquidity,
        "sqrt_price": self.sqrt_price,
        "tick_current_index": self.tick_current_index,
        "protocol_fee_owed_a": self.protocol_fee_owed_a,
        "protocol_fee_owed_b": self.protocol_fee_owed_b,
        "token_mint_a": str(self.token_mint_a),
        "token_vault_a": str(self.token_vault_a),
        "fee_growth_global_a": self.fee_growth_global_a,
        "token_mint_b": str(self.token_mint_b),
        "token_vault_b": str(self.token_vault_b),
        "fee_growth_global_b": self.fee_growth_global_b,
        "reward_last_updated_timestamp": self.reward_last_updated_timestamp,
        "reward_infos": list(map(lambda item: item.to_json(), self.reward_infos)),
    }

class WhirlpoolJSON (*args, **kwargs)

dict() -> new empty dictionary dict(mapping) -> new dictionary initialized from a mapping object's (key, value) pairs dict(iterable) -> new dictionary initialized as if via: d = {} for k, v in iterable: d[k] = v dict(**kwargs) -> new dictionary initialized with the name=value pairs in the keyword argument list. For example: dict(one=1, two=2)

Expand source code

class WhirlpoolJSON(typing.TypedDict):
    whirlpools_config: str
    whirlpool_bump: list[int]
    tick_spacing: int
    tick_spacing_seed: list[int]
    fee_rate: int
    protocol_fee_rate: int
    liquidity: int
    sqrt_price: int
    tick_current_index: int
    protocol_fee_owed_a: int
    protocol_fee_owed_b: int
    token_mint_a: str
    token_vault_a: str
    fee_growth_global_a: int
    token_mint_b: str
    token_vault_b: str
    fee_growth_global_b: int
    reward_last_updated_timestamp: int
    reward_infos: list[types.whirlpool_reward_info.WhirlpoolRewardInfoJSON]

Ancestors

• builtins.dict

Class variables

var fee_growth_global_a : int

var fee_growth_global_b : int

var fee_rate : int

var liquidity : int

var protocol_fee_owed_a : int

var protocol_fee_owed_b : int

var protocol_fee_rate : int

var reward_infos : list[WhirlpoolRewardInfoJSON]

var reward_last_updated_timestamp : int

var sqrt_price : int

var tick_current_index : int

var tick_spacing : int

var tick_spacing_seed : list[int]

var token_mint_a : str

var token_mint_b : str

var token_vault_a : str

var token_vault_b : str

var whirlpool_bump : list[int]

var whirlpools_config : str