Module orca_whirlpool.internal.anchor.accounts.position

Classes

class Position (whirlpool: solders.pubkey.Pubkey, position_mint: solders.pubkey.Pubkey, liquidity: int, tick_lower_index: int, tick_upper_index: int, fee_growth_checkpoint_a: int, fee_owed_a: int, fee_growth_checkpoint_b: int, fee_owed_b: int, reward_infos: list[PositionRewardInfo])

Position(whirlpool: solders.pubkey.Pubkey, position_mint: solders.pubkey.Pubkey, liquidity: int, tick_lower_index: int, tick_upper_index: int, fee_growth_checkpoint_a: int, fee_owed_a: int, fee_growth_checkpoint_b: int, fee_owed_b: int, reward_infos: list[orca_whirlpool.internal.anchor.types.position_reward_info.PositionRewardInfo])

Expand source code

@dataclass
class Position:
    discriminator: typing.ClassVar = b"\xaa\xbc\x8f\xe4z@\xf7\xd0"
    layout: typing.ClassVar = borsh.CStruct(
        "whirlpool" / BorshPubkey,
        "position_mint" / BorshPubkey,
        "liquidity" / borsh.U128,
        "tick_lower_index" / borsh.I32,
        "tick_upper_index" / borsh.I32,
        "fee_growth_checkpoint_a" / borsh.U128,
        "fee_owed_a" / borsh.U64,
        "fee_growth_checkpoint_b" / borsh.U128,
        "fee_owed_b" / borsh.U64,
        "reward_infos" / types.position_reward_info.PositionRewardInfo.layout[3],
    )
    whirlpool: Pubkey
    position_mint: Pubkey
    liquidity: int
    tick_lower_index: int
    tick_upper_index: int
    fee_growth_checkpoint_a: int
    fee_owed_a: int
    fee_growth_checkpoint_b: int
    fee_owed_b: int
    reward_infos: list[types.position_reward_info.PositionRewardInfo]

    @classmethod
    async def fetch(
        cls,
        conn: AsyncClient,
        address: Pubkey,
        commitment: typing.Optional[Commitment] = None,
        program_id: Pubkey = PROGRAM_ID,
    ) -> typing.Optional["Position"]:
        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["Position"]]:
        infos = await get_multiple_accounts(conn, addresses, commitment=commitment)
        res: typing.List[typing.Optional["Position"]] = []
        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) -> "Position":
        if data[:ACCOUNT_DISCRIMINATOR_SIZE] != cls.discriminator:
            raise AccountInvalidDiscriminator(
                "The discriminator for this account is invalid"
            )
        dec = Position.layout.parse(data[ACCOUNT_DISCRIMINATOR_SIZE:])
        return cls(
            whirlpool=dec.whirlpool,
            position_mint=dec.position_mint,
            liquidity=dec.liquidity,
            tick_lower_index=dec.tick_lower_index,
            tick_upper_index=dec.tick_upper_index,
            fee_growth_checkpoint_a=dec.fee_growth_checkpoint_a,
            fee_owed_a=dec.fee_owed_a,
            fee_growth_checkpoint_b=dec.fee_growth_checkpoint_b,
            fee_owed_b=dec.fee_owed_b,
            reward_infos=list(
                map(
                    lambda item: types.position_reward_info.PositionRewardInfo.from_decoded(
                        item
                    ),
                    dec.reward_infos,
                )
            ),
        )

    def to_json(self) -> PositionJSON:
        return {
            "whirlpool": str(self.whirlpool),
            "position_mint": str(self.position_mint),
            "liquidity": self.liquidity,
            "tick_lower_index": self.tick_lower_index,
            "tick_upper_index": self.tick_upper_index,
            "fee_growth_checkpoint_a": self.fee_growth_checkpoint_a,
            "fee_owed_a": self.fee_owed_a,
            "fee_growth_checkpoint_b": self.fee_growth_checkpoint_b,
            "fee_owed_b": self.fee_owed_b,
            "reward_infos": list(map(lambda item: item.to_json(), self.reward_infos)),
        }

    @classmethod
    def from_json(cls, obj: PositionJSON) -> "Position":
        return cls(
            whirlpool=Pubkey.from_string(obj["whirlpool"]),
            position_mint=Pubkey.from_string(obj["position_mint"]),
            liquidity=obj["liquidity"],
            tick_lower_index=obj["tick_lower_index"],
            tick_upper_index=obj["tick_upper_index"],
            fee_growth_checkpoint_a=obj["fee_growth_checkpoint_a"],
            fee_owed_a=obj["fee_owed_a"],
            fee_growth_checkpoint_b=obj["fee_growth_checkpoint_b"],
            fee_owed_b=obj["fee_owed_b"],
            reward_infos=list(
                map(
                    lambda item: types.position_reward_info.PositionRewardInfo.from_json(
                        item
                    ),
                    obj["reward_infos"],
                )
            ),
        )

Class variables

var discriminator : ClassVar

var fee_growth_checkpoint_a : int

var fee_growth_checkpoint_b : int

var fee_owed_a : int

var fee_owed_b : int

var layout : ClassVar

var liquidity : int

var position_mint : solders.pubkey.Pubkey

var reward_infos : list[PositionRewardInfo]

var tick_lower_index : int

var tick_upper_index : int

var whirlpool : solders.pubkey.Pubkey

Static methods

def decode(data: bytes) ‑> Position

Expand source code

@classmethod
def decode(cls, data: bytes) -> "Position":
    if data[:ACCOUNT_DISCRIMINATOR_SIZE] != cls.discriminator:
        raise AccountInvalidDiscriminator(
            "The discriminator for this account is invalid"
        )
    dec = Position.layout.parse(data[ACCOUNT_DISCRIMINATOR_SIZE:])
    return cls(
        whirlpool=dec.whirlpool,
        position_mint=dec.position_mint,
        liquidity=dec.liquidity,
        tick_lower_index=dec.tick_lower_index,
        tick_upper_index=dec.tick_upper_index,
        fee_growth_checkpoint_a=dec.fee_growth_checkpoint_a,
        fee_owed_a=dec.fee_owed_a,
        fee_growth_checkpoint_b=dec.fee_growth_checkpoint_b,
        fee_owed_b=dec.fee_owed_b,
        reward_infos=list(
            map(
                lambda item: types.position_reward_info.PositionRewardInfo.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[Position]

Expand source code

@classmethod
async def fetch(
    cls,
    conn: AsyncClient,
    address: Pubkey,
    commitment: typing.Optional[Commitment] = None,
    program_id: Pubkey = PROGRAM_ID,
) -> typing.Optional["Position"]:
    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[Position]]

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["Position"]]:
    infos = await get_multiple_accounts(conn, addresses, commitment=commitment)
    res: typing.List[typing.Optional["Position"]] = []
    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: PositionJSON) ‑> Position

Expand source code

@classmethod
def from_json(cls, obj: PositionJSON) -> "Position":
    return cls(
        whirlpool=Pubkey.from_string(obj["whirlpool"]),
        position_mint=Pubkey.from_string(obj["position_mint"]),
        liquidity=obj["liquidity"],
        tick_lower_index=obj["tick_lower_index"],
        tick_upper_index=obj["tick_upper_index"],
        fee_growth_checkpoint_a=obj["fee_growth_checkpoint_a"],
        fee_owed_a=obj["fee_owed_a"],
        fee_growth_checkpoint_b=obj["fee_growth_checkpoint_b"],
        fee_owed_b=obj["fee_owed_b"],
        reward_infos=list(
            map(
                lambda item: types.position_reward_info.PositionRewardInfo.from_json(
                    item
                ),
                obj["reward_infos"],
            )
        ),
    )

Methods

def to_json(self) ‑> PositionJSON

Expand source code

def to_json(self) -> PositionJSON:
    return {
        "whirlpool": str(self.whirlpool),
        "position_mint": str(self.position_mint),
        "liquidity": self.liquidity,
        "tick_lower_index": self.tick_lower_index,
        "tick_upper_index": self.tick_upper_index,
        "fee_growth_checkpoint_a": self.fee_growth_checkpoint_a,
        "fee_owed_a": self.fee_owed_a,
        "fee_growth_checkpoint_b": self.fee_growth_checkpoint_b,
        "fee_owed_b": self.fee_owed_b,
        "reward_infos": list(map(lambda item: item.to_json(), self.reward_infos)),
    }

class PositionJSON (*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 PositionJSON(typing.TypedDict):
    whirlpool: str
    position_mint: str
    liquidity: int
    tick_lower_index: int
    tick_upper_index: int
    fee_growth_checkpoint_a: int
    fee_owed_a: int
    fee_growth_checkpoint_b: int
    fee_owed_b: int
    reward_infos: list[types.position_reward_info.PositionRewardInfoJSON]

Ancestors

• builtins.dict

Class variables

var fee_growth_checkpoint_a : int

var fee_growth_checkpoint_b : int

var fee_owed_a : int

var fee_owed_b : int

var liquidity : int

var position_mint : str

var reward_infos : list[PositionRewardInfoJSON]

var tick_lower_index : int

var tick_upper_index : int

var whirlpool : str