Liquid Staking

tinyman/liquid_staking/base_client.py

@@ -0,0 +1,122 @@
+import time
+from base64 import b64decode, b64encode
+
+from algosdk import transaction
+from algosdk.logic import get_application_address
+
+# TODO: move struct to parent.
+from tinyman.liquid_staking.struct import get_struct, get_box_costs
+from tinyman.utils import get_global_state, TransactionGroup
+
+
+class BaseClient():
+    def __init__(self, algod, app_id, user_address, user_sk) -> None:
+        self.algod = algod
+        self.app_id = app_id
+        self.application_address = get_application_address(self.app_id)
+        self.user_address = user_address
+        self.keys = {}
+        self.add_key(user_address, user_sk)
+        self.current_timestamp = None
+        self.simulate = False
+
+    def get_suggested_params(self):
+        return self.algod.suggested_params()
+
+    def get_current_timestamp(self):
+        return self.current_timestamp or time.time()
+
+    def _submit(self, transactions, additional_fees=0):
+        transactions = self.flatten_transactions(transactions)
+        fee = transactions[0].fee
+        n = 0
+        for txn in transactions:
+            if txn.fee == fee:
+                txn.fee = 0
+                n += 1
+        transactions[0].fee = (n + additional_fees) * fee
+        txn_group = TransactionGroup(transactions)
+        for address, key in self.keys.items():
+            if isinstance(key, transaction.LogicSigAccount):
+                txn_group.sign_with_logicsig(key, address=address)
+            else:
+                txn_group.sign_with_private_key(address, key)
+        if self.simulate:
+            txn_info = self.algod.simulate_raw_transactions(txn_group.signed_transactions)
+        else:
+            txn_info = txn_group.submit(self.algod, wait=True)
+        return txn_info
+
+    def flatten_transactions(self, txns):
+        result = []
+        if isinstance(txns, transaction.Transaction):
+            result = [txns]
+        elif isinstance(txns, list):
+            for txn in txns:
+                result += self.flatten_transactions(txn)
+        return result
+
+    def calculate_min_balance(self, accounts=0, assets=0, boxes=None):
+        cost = 0
+        cost += accounts * 100_000
+        cost += assets * 100_000
+        cost += get_box_costs(boxes or {})
+        return cost
+
+    def add_key(self, address, key):
+        self.keys[address] = key
+
+    def get_global(self, key, default=None, app_id=None):
+        app_id = app_id or self.app_id
+        global_state = {s["key"]: s["value"] for s in self.algod.application_info(app_id)["params"]["global-state"]}
+        key = b64encode(key).decode()
+        if key in global_state:
+            value = global_state[key]
+            if value["type"] == 2:
+                return value["uint"]
+            else:
+                return b64decode(value["bytes"])
+        else:
+            return default
+
+    def get_global_state(self, app_id=None):
+        app_id = app_id or self.app_id
+
+        return get_global_state(self.algod, app_id)
+
+    def get_box(self, box_name, struct_name, app_id=None):
+        app_id = app_id or self.app_id
+        box_value = b64decode(self.algod.application_box_by_name(app_id, box_name)["value"])
+        struct_class = get_struct(struct_name)
+        struct = struct_class(box_value)
+        return struct
+
+    def box_exists(self, box_name, app_id=None):
+        app_id = app_id or self.app_id
+        try:
+            self.algod.application_box_by_name(app_id, box_name)
+            return True
+        except Exception:
+            return False
+
+    def get_reward_slot(self, staking_asset_id, reward_asset_id):
+        asset_box = self.get_asset_box(staking_asset_id)
+        for i in range(8):
+            if asset_box.reward_slots[i].asset_id == reward_asset_id:
+                return i
+
+    def is_opted_in(self, address, asset_id):
+        try:
+            self.algod.account_asset_info(address, asset_id)
+            return True
+        except Exception:
+            return False
+
+    def get_optin_if_needed_txn(self, sender, asset_id):
+        if not self.is_opted_in(sender, asset_id):
+            txn = transaction.AssetOptInTxn(
+                sender=sender,
+                sp=self.get_suggested_params(),
+                index=asset_id,
+            )
+            return txn

tinyman/liquid_staking/constants.py

@@ -0,0 +1,41 @@
+TESTNET_TALGO_APP_ID = 724519988
+MAINNET_TALGO_APP_ID = 2537013674
+
+TESTNET_TALGO_STAKING_APP_ID = 724676904
+MAINNET_TALGO_STAKING_APP_ID = 2537022861
+
+TESTNET_TALGO_ASSET_ID = 724519992
+MAINNET_TALGO_ASSET_ID = 2537013734
+
+TESTNET_STALGO_ASSET_ID = 724676936
+MAINNET_STALGO_ASSET_ID = 2537023208
+
+# App Constants
+
+APP_LOCAL_INTS = 3
+APP_LOCAL_BYTES = 1
+APP_GLOBAL_INTS = 16
+APP_GLOBAL_BYTES = 16
+EXTRA_PAGES = 3
+
+VAULT_APP_ID_KEY = b"vault_app_id"
+TINY_ASSET_ID_KEY = b"tiny_asset_id"
+TALGO_ASSET_ID_KEY = b"talgo_asset_id"
+STALGO_ASSET_ID_KEY = b"stalgo_asset_id"
+
+TOTAL_REWARD_AMOUNT_SUM_KEY = b"total_reward_amount_sum"
+TOTAL_CLAIMED_REWARD_AMOUNT_KEY = b"total_claimed_reward_amount"
+CURRENT_REWARD_RATE_PER_TIME_KEY = b"current_reward_rate_per_time"
+CURRENT_REWARD_RATE_PER_TIME_END_TIMESTAMP_KEY = b"current_reward_rate_per_time_end_timestamp"
+
+TINY_POWER_THRESHOLD_KEY = b"tiny_power_threshold"
+LAST_UPDATE_TIMESTAMP_KEY = b"last_update_timestamp"
+ACCUMULATED_REWARDS_PER_UNIT = b"accumulated_rewards_per_unit"
+TOTAL_STAKED_AMOUNT_KEY = b"total_staked_amount"
+TOTAL_STAKER_COUNT_KEY = b"total_staker_count"
+
+PROPOSED_MANAGER_KEY = b"proposed_manager"
+MANAGER_KEY = b"manager"
+
+MAX_UINT64 = 18446744073709551615
+RATE_SCALER = 1_000_000_000_000