Merge pull request #2731 from opentensor/feat/zyzniewski/set_delegate_take

Feat: set_delegate_take

Author: zyzniewski-reef

bittensor/core/async_subtensor.py

@@ -31,7 +31,7 @@
 from bittensor.core.chain_data.delegate_info import DelegatedInfo
 from bittensor.core.chain_data.utils import decode_metadata
 from bittensor.core.config import Config
-from bittensor.core.errors import SubstrateRequestException
+from bittensor.core.errors import ChainError, SubstrateRequestException
 from bittensor.core.extrinsics.asyncex.commit_reveal import commit_reveal_v3_extrinsic
 from bittensor.core.extrinsics.asyncex.registration import (
     burned_register_extrinsic,
@@ -57,6 +57,10 @@
     add_stake_extrinsic,
     add_stake_multiple_extrinsic,
 )
+from bittensor.core.extrinsics.asyncex.take import (
+    decrease_take_extrinsic,
+    increase_take_extrinsic,
+)
 from bittensor.core.extrinsics.asyncex.transfer import transfer_extrinsic
 from bittensor.core.extrinsics.asyncex.unstaking import (
     unstake_extrinsic,
@@ -1111,7 +1115,7 @@ async def get_delegate_take(
         block: Optional[int] = None,
         block_hash: Optional[str] = None,
         reuse_block: bool = False,
-    ) -> Optional[float]:
+    ) -> float:
         """
         Retrieves the delegate 'take' percentage for a neuron identified by its hotkey. The 'take' represents the
             percentage of rewards that the delegate claims from its nominators' stakes.
@@ -1123,7 +1127,7 @@ async def get_delegate_take(
             reuse_block (bool): Whether to reuse the last-used block hash.
 
         Returns:
-            Optional[float]: The delegate take percentage, None if not available.
+            float: The delegate take percentage.
 
         The delegate take is a critical parameter in the network's incentive structure, influencing the distribution of
             rewards among neurons and their nominators.
@@ -1135,11 +1139,8 @@ async def get_delegate_take(
             reuse_block=reuse_block,
             params=[hotkey_ss58],
         )
-        return (
-            None
-            if result is None
-            else u16_normalized_float(getattr(result, "value", 0))
-        )
+
+        return u16_normalized_float(result.value)  # type: ignore
 
     async def get_delegated(
         self,
@@ -2748,6 +2749,7 @@ async def sign_and_send_extrinsic(
         use_nonce: bool = False,
         period: Optional[int] = None,
         nonce_key: str = "hotkey",
+        raise_error: bool = False,
     ) -> tuple[bool, str]:
         """
         Helper method to sign and submit an extrinsic call to chain.
@@ -2758,6 +2760,7 @@ async def sign_and_send_extrinsic(
             wait_for_inclusion (bool): whether to wait until the extrinsic call is included on the chain
             wait_for_finalization (bool): whether to wait until the extrinsic call is finalized on the chain
             sign_with: the wallet's keypair to use for the signing. Options are "coldkey", "hotkey", "coldkeypub"
+            raise_error: raises relevant exception rather than returning `False` if unsuccessful.
 
         Returns:
             (success, error message)
@@ -2795,9 +2798,15 @@ async def sign_and_send_extrinsic(
             if await response.is_success:
                 return True, ""
 
+            if raise_error:
+                raise ChainError.from_error(response.error_message)
+
             return False, format_error_message(await response.error_message)
 
         except SubstrateRequestException as e:
+            if raise_error:
+                raise
+
             return False, format_error_message(e)
 
     # Extrinsics =======================================================================================================
@@ -3260,6 +3269,82 @@ async def root_set_weights(
             wait_for_inclusion=wait_for_inclusion,
         )
 
+    async def set_delegate_take(
+        self,
+        wallet: "Wallet",
+        hotkey_ss58: str,
+        take: float,
+        wait_for_inclusion: bool = True,
+        wait_for_finalization: bool = True,
+        raise_error: bool = False,
+    ) -> tuple[bool, str]:
+        """
+        Sets the delegate 'take' percentage for a neuron identified by its hotkey.
+        The 'take' represents the percentage of rewards that the delegate claims from its nominators' stakes.
+
+        Arguments:
+            wallet (bittensor_wallet.Wallet): bittensor wallet instance.
+            hotkey_ss58 (str): The ``SS58`` address of the neuron's hotkey.
+            take (float): Percentage reward for the delegate.
+            wait_for_inclusion (bool): Waits for the transaction to be included in a block.
+            wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain.
+            raise_error: Raises relevant exception rather than returning `False` if unsuccessful.
+
+        Returns:
+            tuple[bool, str]: A tuple where the first element is a boolean indicating success or failure of the
+             operation, and the second element is a message providing additional information.
+
+        Raises:
+            DelegateTakeTooHigh: Delegate take is too high.
+            DelegateTakeTooLow: Delegate take is too low.
+            DelegateTxRateLimitExceeded: A transactor exceeded the rate limit for delegate transaction.
+            HotKeyAccountNotExists: The hotkey does not exists.
+            NonAssociatedColdKey: Request to stake, unstake or subscribe is made by a coldkey that is not associated with the hotkey account.
+            bittensor_wallet.errors.PasswordError: Decryption failed or wrong password for decryption provided.
+            bittensor_wallet.errors.KeyFileError: Failed to decode keyfile data.
+
+        The delegate take is a critical parameter in the network's incentive structure, influencing the distribution of
+            rewards among neurons and their nominators.
+        """
+
+        # u16 representation of the take
+        take_u16 = int(take * 0xFFFF)
+
+        current_take = await self.get_delegate_take(hotkey_ss58)
+        current_take_u16 = int(current_take * 0xFFFF)
+
+        if current_take_u16 == take_u16:
+            logging.info(":white_heavy_check_mark: [green]Already Set[/green]")
+            return True, ""
+
+        logging.info(f"Updating {hotkey_ss58} take: current={current_take} new={take}")
+
+        if current_take_u16 < take_u16:
+            success, error = await increase_take_extrinsic(
+                self,
+                wallet,
+                hotkey_ss58,
+                take_u16,
+                wait_for_finalization=wait_for_finalization,
+                wait_for_inclusion=wait_for_inclusion,
+                raise_error=raise_error,
+            )
+        else:
+            success, error = await decrease_take_extrinsic(
+                self,
+                wallet,
+                hotkey_ss58,
+                take_u16,
+                wait_for_finalization=wait_for_finalization,
+                wait_for_inclusion=wait_for_inclusion,
+                raise_error=raise_error,
+            )
+
+        if success:
+            logging.info(":white_heavy_check_mark: [green]Take Updated[/green]")
+
+        return success, error
+
     async def set_subnet_identity(
         self,
         wallet: "Wallet",

bittensor/core/errors.py

@@ -17,6 +17,21 @@
 ExtrinsicNotFound = ExtrinsicNotFound
 
 
+class _ChainErrorMeta(type):
+    _exceptions: dict[str, Exception] = {}
+
+    def __new__(mcs, name, bases, attrs):
+        cls = super().__new__(mcs, name, bases, attrs)
+
+        mcs._exceptions.setdefault(cls.__name__, cls)
+
+        return cls
+
+    @classmethod
+    def get_exception_class(mcs, exception_name):
+        return mcs._exceptions[exception_name]
+
+
 class MaxSuccessException(Exception):
     """Raised when the POW Solver has reached the max number of successful solutions."""
 
@@ -25,9 +40,20 @@ class MaxAttemptsException(Exception):
     """Raised when the POW Solver has reached the max number of attempts."""
 
 
-class ChainError(SubstrateRequestException):
+class ChainError(SubstrateRequestException, metaclass=_ChainErrorMeta):
     """Base error for any chain related errors."""
 
+    @classmethod
+    def from_error(cls, error):
+        try:
+            error_cls = _ChainErrorMeta.get_exception_class(
+                error["name"],
+            )
+        except KeyError:
+            return cls(error)
+        else:
+            return error_cls(" ".join(error["docs"]))
+
 
 class ChainConnectionError(ChainError):
     """Error for any chain connection related errors."""
@@ -41,6 +67,36 @@ class ChainQueryError(ChainError):
     """Error for any chain query related errors."""
 
 
+class DelegateTakeTooHigh(ChainTransactionError):
+    """
+    Delegate take is too high.
+    """
+
+
+class DelegateTakeTooLow(ChainTransactionError):
+    """
+    Delegate take is too low.
+    """
+
+
+class DelegateTxRateLimitExceeded(ChainTransactionError):
+    """
+    A transactor exceeded the rate limit for delegate transaction.
+    """
+
+
+class HotKeyAccountNotExists(ChainTransactionError):
+    """
+    The hotkey does not exist.
+    """
+
+
+class NonAssociatedColdKey(ChainTransactionError):
+    """
+    Request to stake, unstake or subscribe is made by a coldkey that is not associated with the hotkey account.
+    """
+
+
 class StakeError(ChainTransactionError):
     """Error raised when a stake transaction fails."""
 

bittensor/core/extrinsics/asyncex/take.py

@@ -0,0 +1,72 @@
+from typing import TYPE_CHECKING
+
+from bittensor_wallet.bittensor_wallet import Wallet
+
+from bittensor.utils import unlock_key
+
+if TYPE_CHECKING:
+    from bittensor.core.async_subtensor import AsyncSubtensor
+
+
+async def increase_take_extrinsic(
+    subtensor: "AsyncSubtensor",
+    wallet: Wallet,
+    hotkey_ss58: str,
+    take: int,
+    wait_for_inclusion: bool = True,
+    wait_for_finalization: bool = True,
+    raise_error: bool = False,
+) -> tuple[bool, str]:
+    unlock = unlock_key(wallet, raise_error=raise_error)
+
+    if not unlock.success:
+        return False, unlock.message
+
+    call = await subtensor.substrate.compose_call(
+        call_module="SubtensorModule",
+        call_function="increase_take",
+        call_params={
+            "hotkey": hotkey_ss58,
+            "take": take,
+        },
+    )
+
+    return await subtensor.sign_and_send_extrinsic(
+        call,
+        wallet,
+        wait_for_inclusion=wait_for_inclusion,
+        wait_for_finalization=wait_for_finalization,
+        raise_error=raise_error,
+    )
+
+
+async def decrease_take_extrinsic(
+    subtensor: "AsyncSubtensor",
+    wallet: Wallet,
+    hotkey_ss58: str,
+    take: int,
+    wait_for_inclusion: bool = True,
+    wait_for_finalization: bool = True,
+    raise_error: bool = False,
+) -> tuple[bool, str]:
+    unlock = unlock_key(wallet, raise_error=raise_error)
+
+    if not unlock.success:
+        return False, unlock.message
+
+    call = await subtensor.substrate.compose_call(
+        call_module="SubtensorModule",
+        call_function="decrease_take",
+        call_params={
+            "hotkey": hotkey_ss58,
+            "take": take,
+        },
+    )
+
+    return await subtensor.sign_and_send_extrinsic(
+        call,
+        wallet,
+        wait_for_inclusion=wait_for_inclusion,
+        wait_for_finalization=wait_for_finalization,
+        raise_error=raise_error,
+    )

bittensor/core/extrinsics/take.py

@@ -0,0 +1,72 @@
+from typing import TYPE_CHECKING
+
+from bittensor_wallet.bittensor_wallet import Wallet
+
+from bittensor.utils import unlock_key
+
+if TYPE_CHECKING:
+    from bittensor.core.subtensor import Subtensor
+
+
+def increase_take_extrinsic(
+    subtensor: "Subtensor",
+    wallet: Wallet,
+    hotkey_ss58: str,
+    take: int,
+    wait_for_inclusion: bool = True,
+    wait_for_finalization: bool = True,
+    raise_error: bool = False,
+) -> tuple[bool, str]:
+    unlock = unlock_key(wallet, raise_error=raise_error)
+
+    if not unlock.success:
+        return False, unlock.message
+
+    call = subtensor.substrate.compose_call(
+        call_module="SubtensorModule",
+        call_function="increase_take",
+        call_params={
+            "hotkey": hotkey_ss58,
+            "take": take,
+        },
+    )
+
+    return subtensor.sign_and_send_extrinsic(
+        call,
+        wallet,
+        wait_for_inclusion=wait_for_inclusion,
+        wait_for_finalization=wait_for_finalization,
+        raise_error=raise_error,
+    )
+
+
+def decrease_take_extrinsic(
+    subtensor: "Subtensor",
+    wallet: Wallet,
+    hotkey_ss58: str,
+    take: int,
+    wait_for_inclusion: bool = True,
+    wait_for_finalization: bool = True,
+    raise_error: bool = False,
+) -> tuple[bool, str]:
+    unlock = unlock_key(wallet, raise_error=raise_error)
+
+    if not unlock.success:
+        return False, unlock.message
+
+    call = subtensor.substrate.compose_call(
+        call_module="SubtensorModule",
+        call_function="decrease_take",
+        call_params={
+            "hotkey": hotkey_ss58,
+            "take": take,
+        },
+    )
+
+    return subtensor.sign_and_send_extrinsic(
+        call,
+        wallet,
+        wait_for_inclusion=wait_for_inclusion,
+        wait_for_finalization=wait_for_finalization,
+        raise_error=raise_error,
+    )