Create a mechanism for downloading hive power data

clive/__private/core/commands/commands.py

@@ -8,6 +8,7 @@ from clive.__private.core.commands.broadcast import Broadcast
 from clive.__private.core.commands.build_transaction import BuildTransaction
 from clive.__private.core.commands.command_wrappers import CommandWithResultWrapper, CommandWrapper
 from clive.__private.core.commands.create_wallet import CreateWallet
+from clive.__private.core.commands.data_retrieval.hive_power_data import HivePowerData, HivePowerDataRetrieval
 from clive.__private.core.commands.data_retrieval.proposals_data import ProposalsData, ProposalsDataRetrieval
 from clive.__private.core.commands.data_retrieval.savings_data import SavingsData, SavingsDataRetrieval
 from clive.__private.core.commands.data_retrieval.update_node_data import UpdateNodeData
@@ -298,6 +299,15 @@ class Commands(Generic[WorldT]):
             )
         )
 
+    async def retrieve_hp_data(
+        self,
+        *,
+        account_name: str,
+    ) -> CommandWithResultWrapper[HivePowerData]:
+        return await self.__surround_with_exception_handlers(
+            HivePowerDataRetrieval(node=self._world.node, account_name=account_name)
+        )
+
     async def find_transaction(self, *, transaction_id: str) -> CommandWithResultWrapper[TransactionStatus]:
         return await self.__surround_with_exception_handlers(
             FindTransaction(node=self._world.node, transaction_id=transaction_id)

clive/__private/core/commands/data_retrieval/hive_power_data.py

+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import TYPE_CHECKING, Final
+
+from clive.__private.core.calcluate_hive_power import calculate_hive_power
+from clive.__private.core.commands.abc.command_data_retrieval import CommandDataRetrieval
+from clive.models import Asset
+
+if TYPE_CHECKING:
+    from datetime import datetime
+
+    from clive.__private.core.node import Node
+    from clive.models.aliased import DynamicGlobalProperties, SchemasAccount
+    from schemas.apis.database_api import FindAccounts
+
+
+@dataclass
+class HarvestedDataRaw:
+    dgpo: DynamicGlobalProperties | None = None
+    core_account: FindAccounts | None = None
+
+
+@dataclass
+class SanitizedData:
+    dgpo: DynamicGlobalProperties
+    core_account: SchemasAccount
+
+
+@dataclass
+class SharesBalance:
+    """Class to store the balance of shares in HP and VESTS."""
+
+    hp_balance: Asset.Hive
+    vests_balance: Asset.Vests
+
+
+@dataclass
+class HivePowerData:
+    owned_balance: SharesBalance
+    total_balance: SharesBalance
+    received_balance: SharesBalance
+    delegated_balance: SharesBalance
+    next_vesting_withdrawal: datetime
+    to_withdraw: SharesBalance
+    next_power_down: Asset.Hive
+    current_hp_apr: str
+
+
+@dataclass(kw_only=True)
+class HivePowerDataRetrieval(CommandDataRetrieval[HarvestedDataRaw, SanitizedData, HivePowerData]):
+    node: Node
+    account_name: str
+
+    async def _harvest_data_from_api(self) -> HarvestedDataRaw:
+        async with self.node.batch() as node:
+            return HarvestedDataRaw(
+                await node.api.database_api.get_dynamic_global_properties(),
+                await node.api.database_api.find_accounts(accounts=[self.account_name]),
+            )
+
+    async def _sanitize_data(self, data: HarvestedDataRaw) -> SanitizedData:
+        return SanitizedData(
+            dgpo=self._assert_gdpo(data.dgpo),
+            core_account=self._assert_core_account(data.core_account),
+        )
+
+    async def _process_data(self, data: SanitizedData) -> HivePowerData:
+        owned_shares = data.core_account.vesting_shares
+        received_shares = data.core_account.received_vesting_shares
+        delegated_shares = data.core_account.delegated_vesting_shares
+        total_shares = owned_shares + received_shares - delegated_shares
+
+        return HivePowerData(
+            owned_balance=self._create_balance_representation(data.dgpo, owned_shares),
+            total_balance=self._create_balance_representation(data.dgpo, total_shares),
+            received_balance=self._create_balance_representation(data.dgpo, received_shares),
+            delegated_balance=self._create_balance_representation(data.dgpo, delegated_shares),
+            next_vesting_withdrawal=data.core_account.next_vesting_withdrawal,
+            to_withdraw=self._create_balance_representation(
+                data.dgpo, Asset.vests(data.core_account.to_withdraw / 10**data.dgpo.total_vesting_shares.precision)
+            ),
+            next_power_down=Asset.hive(
+                calculate_hive_power(data.dgpo, data.core_account.vesting_withdraw_rate)
+                / 10**data.dgpo.total_vesting_fund_hive.precision
+            ),
+            current_hp_apr=self._calculate_current_hp_apr(data.dgpo),
+        )
+
+    def _assert_gdpo(self, data: DynamicGlobalProperties | None) -> DynamicGlobalProperties:
+        assert data is not None, "DynamicGlobalProperties data is missing"
+        return data
+
+    def _assert_core_account(self, data: FindAccounts | None) -> SchemasAccount:
+        assert data is not None, "FindAccounts data is missing"
+        assert len(data.accounts) == 1, "Invalid amount of accounts"
+
+        account = data.accounts[0]
+        assert account.name == self.account_name, "Invalid account name"
+        return account
+
+    def _create_balance_representation(self, dgdpo: DynamicGlobalProperties, vests_value: Asset.Vests) -> SharesBalance:
+        return SharesBalance(
+            hp_balance=Asset.hive(
+                calculate_hive_power(dgdpo, vests_value) / 10**dgdpo.total_vesting_fund_hive.precision
+            ),
+            vests_balance=vests_value,
+        )
+
+    def _calculate_current_hp_apr(self, gdpo: DynamicGlobalProperties) -> str:
+        # The inflation was set to 9.5% at block 7m
+        initial_inflation_rate: Final[float] = 9.5
+        initial_block: Final[int] = 7000000
+
+        # It decreases by 0.01% every 250k blocks
+        decrease_rate: Final[int] = 250000
+        decrease_percent_per_increment: Final[float] = 0.01
+
+        # How many increments have happened since block 7m?
+        head_block = gdpo.head_block_number
+        delta_blocks = head_block - initial_block
+        decrease_increments = delta_blocks / decrease_rate
+
+        current_inflation_rate = initial_inflation_rate - decrease_increments * decrease_percent_per_increment
+
+        # Cannot go lower than 0.95 %
+        minimum_inflation_rate: Final[float] = 0.95
+
+        if current_inflation_rate < minimum_inflation_rate:
+            current_inflation_rate = 0.95
+
+        # Calculate the APR
+        vesting_reward_percent = gdpo.vesting_reward_percent / 10000
+        virtual_supply = int(gdpo.virtual_supply.amount) / 10**gdpo.virtual_supply.precision
+        total_vesting_funds = int(gdpo.total_vesting_fund_hive.amount) / 10**gdpo.total_vesting_fund_hive.precision
+
+        return f"{virtual_supply * current_inflation_rate * vesting_reward_percent / total_vesting_funds :.2f}"