diff --git a/bittensor/core/async_subtensor.py b/bittensor/core/async_subtensor.py index f39857db87..4c456ac83a 100644 --- a/bittensor/core/async_subtensor.py +++ b/bittensor/core/async_subtensor.py @@ -192,6 +192,41 @@ async def encode_params( return param_data.to_hex() + async def get_hyperparameter( + self, + param_name: str, + netuid: int, + block_hash: Optional[str] = None, + reuse_block: bool = False, + ) -> Optional[Any]: + """ + Retrieves a specified hyperparameter for a specific subnet. + + Args: + param_name (str): The name of the hyperparameter to retrieve. + netuid (int): The unique identifier of the subnet. + block_hash (Optional[str]): The hash of blockchain block number for the query. + reuse_block (bool): Whether to reuse the last-used block hash. + + Returns: + The value of the specified hyperparameter if the subnet exists, or None + """ + if not await self.subnet_exists(netuid, block_hash): + logging.error(f"subnet {netuid} does not exist") + return None + + result = await self.substrate.query( + module="SubtensorModule", + storage_function=param_name, + params=[netuid], + block_hash=block_hash, + reuse_block_hash=reuse_block, + ) + + return result + + # Common subtensor methods ========================================================================================= + async def get_current_block(self) -> int: """ Returns the current block number on the Bittensor blockchain. This function provides the latest block number, indicating the most recent state of the blockchain. @@ -663,39 +698,6 @@ async def subnet_exists( ) return result - async def get_hyperparameter( - self, - param_name: str, - netuid: int, - block_hash: Optional[str] = None, - reuse_block: bool = False, - ) -> Optional[Any]: - """ - Retrieves a specified hyperparameter for a specific subnet. - - Args: - param_name (str): The name of the hyperparameter to retrieve. - netuid (int): The unique identifier of the subnet. - block_hash (Optional[str]): The hash of blockchain block number for the query. - reuse_block (bool): Whether to reuse the last-used block hash. - - Returns: - The value of the specified hyperparameter if the subnet exists, or None - """ - if not await self.subnet_exists(netuid, block_hash): - print("subnet does not exist") - return None - - result = await self.substrate.query( - module="SubtensorModule", - storage_function=param_name, - params=[netuid], - block_hash=block_hash, - reuse_block_hash=reuse_block, - ) - - return result - async def filter_netuids_by_registered_hotkeys( self, all_netuids: Iterable[int], @@ -1357,7 +1359,7 @@ async def blocks_since_last_update(self, netuid: int, uid: int) -> Optional[int] call = await self.get_hyperparameter(param_name="LastUpdate", netuid=netuid) return None if call is None else await self.get_current_block() - int(call[uid]) - # extrinsics + # Extrinsics ======================================================================================================= async def transfer( self, diff --git a/bittensor/core/extrinsics/staking.py b/bittensor/core/extrinsics/staking.py new file mode 100644 index 0000000000..769043c857 --- /dev/null +++ b/bittensor/core/extrinsics/staking.py @@ -0,0 +1,479 @@ +from time import sleep +from typing import Union, Optional, TYPE_CHECKING + +from bittensor.core.errors import NotDelegateError, StakeError, NotRegisteredError +from bittensor.utils import format_error_message, unlock_key +from bittensor.utils.balance import Balance +from bittensor.utils.btlogging import logging +from bittensor.utils.networking import ensure_connected + +if TYPE_CHECKING: + from bittensor_wallet import Wallet + from bittensor.core.subtensor import Subtensor + + +@ensure_connected +def _do_stake( + self: "Subtensor", + wallet: "Wallet", + hotkey_ss58: str, + amount: "Balance", + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, +) -> bool: + """Sends a stake extrinsic to the chain. + + Args: + self (subtensor): Subtensor instance. + wallet (bittensor_wallet.Wallet): Wallet object that can sign the extrinsic. + hotkey_ss58 (str): Hotkey ``ss58`` address to stake to. + amount (bittensor.utils.balance.Balance): Amount to stake. + wait_for_inclusion (bool): If ``true``, waits for inclusion before returning. + wait_for_finalization (bool): If ``true``, waits for finalization before returning. + + Returns: + success (bool): ``True`` if the extrinsic was successful. + + Raises: + bittensor.core.errors.StakeError: If the extrinsic failed. + """ + + call = self.substrate.compose_call( + call_module="SubtensorModule", + call_function="add_stake", + call_params={"hotkey": hotkey_ss58, "amount_staked": amount.rao}, + ) + extrinsic = self.substrate.create_signed_extrinsic( + call=call, keypair=wallet.coldkey + ) + response = self.substrate.submit_extrinsic( + extrinsic, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + # We only wait here if we expect finalization. + if not wait_for_finalization and not wait_for_inclusion: + return True + + response.process_events() + if response.is_success: + return True + else: + raise StakeError( + format_error_message(response.error_message, substrate=self.substrate) + ) + + +def _check_threshold_amount( + subtensor: "Subtensor", stake_balance: Balance +) -> tuple[bool, Balance]: + """ + Checks if the new stake balance will be above the minimum required stake threshold. + + Args: + subtensor (bittensor.core.subtensor.Subtensor): Subtensor instance. + stake_balance (Balance): the balance to check for threshold limits. + + Returns: + success, threshold (bool, Balance): ``true`` if the staking balance is above the threshold, or ``false`` if the staking balance is below the threshold. The threshold balance required to stake. + """ + min_req_stake: Balance = subtensor.get_minimum_required_stake() + if min_req_stake > stake_balance: + return False, min_req_stake + else: + return True, min_req_stake + + +def __do_add_stake_single( + subtensor: "Subtensor", + wallet: "Wallet", + hotkey_ss58: str, + amount: "Balance", + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, +) -> bool: + """ + Executes a stake call to the chain using the wallet and the amount specified. + + Args: + wallet (bittensor_wallet.Wallet): Bittensor wallet object. + hotkey_ss58 (str): Hotkey to stake to. + amount (bittensor.utils.balance.Balance): Amount to stake as Bittensor balance object. + wait_for_inclusion (bool): If set, waits for the extrinsic to enter a block before returning ``true``, or returns ``false`` if the extrinsic fails to enter the block within the timeout. + wait_for_finalization (bool): If set, waits for the extrinsic to be finalized on the chain before returning ``true``, or returns ``false`` if the extrinsic fails to be finalized within the timeout. + + Returns: + success (bool): Flag is ``true`` if extrinsic was finalized or uncluded in the block. If we did not wait for finalization / inclusion, the response is ``true``. + + Raises: + bittensor.core.errors.StakeError: If the extrinsic fails to be finalized or included in the block. + bittensor.core.errors.NotDelegateError: If the hotkey is not a delegate. + bittensor.core.errors.NotRegisteredError: If the hotkey is not registered in any subnets. + + """ + # Decrypt keys, + if not (unlock := unlock_key(wallet)).success: + logging.error(unlock.message) + return False + + hotkey_owner = subtensor.get_hotkey_owner(hotkey_ss58) + own_hotkey = wallet.coldkeypub.ss58_address == hotkey_owner + if not own_hotkey: + # We are delegating. Verify that the hotkey is a delegate. + if not subtensor.is_hotkey_delegate(hotkey_ss58=hotkey_ss58): + raise NotDelegateError("Hotkey: {} is not a delegate.".format(hotkey_ss58)) + + success = _do_stake( + wallet=wallet, + hotkey_ss58=hotkey_ss58, + amount=amount, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + return success + + +def add_stake_extrinsic( + subtensor: "Subtensor", + wallet: "Wallet", + hotkey_ss58: Optional[str] = None, + amount: Optional[Union[Balance, float]] = None, + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, +) -> bool: + """Adds the specified amount of stake to passed hotkey ``uid``. + + Args: + subtensor (bittensor.core.subtensor.Subtensor): Subtensor instance. + wallet (Wallet): Bittensor wallet object. + hotkey_ss58 (Optional[str]): The ``ss58`` address of the hotkey account to stake to defaults to the wallet's hotkey. + amount (Union[Balance, float]): Amount to stake as Bittensor balance, or ``float`` interpreted as Tao. + wait_for_inclusion (bool): If set, waits for the extrinsic to enter a block before returning ``true``, or returns ``false`` if the extrinsic fails to enter the block within the timeout. + wait_for_finalization (bool): If set, waits for the extrinsic to be finalized on the chain before returning ``true``, or returns ``false`` if the extrinsic fails to be finalized within the timeout. + + Returns: + success (bool): Flag is ``true`` if extrinsic was finalized or uncluded in the block. If we did not wait for finalization / inclusion, the response is ``true``. + + Raises: + bittensor.core.errors.NotRegisteredError: If the wallet is not registered on the chain. + bittensor.core.errors.NotDelegateError: If the hotkey is not a delegate on the chain. + """ + # Decrypt keys, + if not (unlock := unlock_key(wallet)).success: + logging.error(unlock.message) + return False + + # Default to wallet's own hotkey if the value is not passed. + if hotkey_ss58 is None: + hotkey_ss58 = wallet.hotkey.ss58_address + + # Flag to indicate if we are using the wallet's own hotkey. + own_hotkey: bool + + logging.info( + f":satellite: [magenta]Syncing with chain:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + old_balance = subtensor.get_balance(wallet.coldkeypub.ss58_address) + # Get hotkey owner + hotkey_owner = subtensor.get_hotkey_owner(hotkey_ss58) + own_hotkey = wallet.coldkeypub.ss58_address == hotkey_owner + if not own_hotkey: + # This is not the wallet's own hotkey so we are delegating. + if not subtensor.is_hotkey_delegate(hotkey_ss58): + raise NotDelegateError("Hotkey: {} is not a delegate.".format(hotkey_ss58)) + + # Get current stake + old_stake = subtensor.get_stake_for_coldkey_and_hotkey( + coldkey_ss58=wallet.coldkeypub.ss58_address, hotkey_ss58=hotkey_ss58 + ) + + # Grab the existential deposit. + existential_deposit = subtensor.get_existential_deposit() + + # Convert to bittensor.Balance + if amount is None: + # Stake it all. + staking_balance = Balance.from_tao(old_balance.tao) + elif not isinstance(amount, Balance): + staking_balance = Balance.from_tao(amount) + else: + staking_balance = amount + + # Leave existential balance to keep key alive. + if staking_balance > old_balance - existential_deposit: + # If we are staking all, we need to leave at least the existential deposit. + staking_balance = old_balance - existential_deposit + else: + staking_balance = staking_balance + + # Check enough to stake. + if staking_balance > old_balance: + logging.error(":cross_mark: [red]Not enough stake:[/red]") + logging.error(f"\t\tbalance:{old_balance}") + logging.error(f"\t\tamount: {staking_balance}") + logging.error(f"\t\twallet: {wallet.name}") + return False + + # If nominating, we need to check if the new stake balance will be above the minimum required stake threshold. + if not own_hotkey: + new_stake_balance = old_stake + staking_balance + is_above_threshold, threshold = _check_threshold_amount( + subtensor, new_stake_balance + ) + if not is_above_threshold: + logging.error( + f":cross_mark: [red]New stake balance of {new_stake_balance} is below the minimum required nomination stake threshold {threshold}.[/red]" + ) + return False + + try: + logging.info( + f":satellite: [magenta]Staking to:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + staking_response: bool = __do_add_stake_single( + subtensor=subtensor, + wallet=wallet, + hotkey_ss58=hotkey_ss58, + amount=staking_balance, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + if staking_response is True: # If we successfully staked. + # We only wait here if we expect finalization. + if not wait_for_finalization and not wait_for_inclusion: + return True + + logging.success(":white_heavy_check_mark: [green]Finalized[/green]") + + logging.info( + f":satellite: [magenta]Checking Balance on:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + new_balance = subtensor.get_balance(address=wallet.coldkeypub.ss58_address) + block = subtensor.get_current_block() + new_stake = subtensor.get_stake_for_coldkey_and_hotkey( + coldkey_ss58=wallet.coldkeypub.ss58_address, + hotkey_ss58=hotkey_ss58, + block=block, + ) # Get current stake + + logging.info("Balance:") + logging.info( + f"[blue]{old_balance}[/blue] :arrow_right: {new_balance}[/green]" + ) + logging.info("Stake:") + logging.info( + f"[blue]{old_stake}[/blue] :arrow_right: [green]{new_stake}[/green]" + ) + return True + else: + logging.error(":cross_mark: [red]Failed[/red]: Error unknown.") + return False + + except NotRegisteredError: + logging.error( + ":cross_mark: [red]Hotkey: {} is not registered.[/red]".format( + wallet.hotkey_str + ) + ) + return False + except StakeError as e: + logging.error(f":cross_mark: [red]Stake Error: {e}[/red]") + return False + + +def add_stake_multiple_extrinsic( + subtensor: "Subtensor", + wallet: "Wallet", + hotkey_ss58s: list[str], + amounts: Optional[list[Union[Balance, float]]] = None, + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, +) -> bool: + """Adds stake to each ``hotkey_ss58`` in the list, using each amount, from a common coldkey. + + Args: + subtensor (bittensor.core.subtensor.Subtensor): Subtensor instance. + wallet (bittensor_wallet.Wallet): Bittensor wallet object for the coldkey. + hotkey_ss58s (List[str]): List of hotkeys to stake to. + amounts (List[Union[Balance, float]]): List of amounts to stake. If ``None``, stake all to the first hotkey. + wait_for_inclusion (bool): If set, waits for the extrinsic to enter a block before returning ``true``, or returns ``false`` if the extrinsic fails to enter the block within the timeout. + wait_for_finalization (bool): If set, waits for the extrinsic to be finalized on the chain before returning ``true``, or returns ``false`` if the extrinsic fails to be finalized within the timeout. + + Returns: + success (bool): Flag is ``true`` if extrinsic was finalized or included in the block. Flag is ``true`` if any wallet was staked. If we did not wait for finalization / inclusion, the response is ``true``. + """ + if not isinstance(hotkey_ss58s, list) or not all( + isinstance(hotkey_ss58, str) for hotkey_ss58 in hotkey_ss58s + ): + raise TypeError("hotkey_ss58s must be a list of str") + + if len(hotkey_ss58s) == 0: + return True + + if amounts is not None and len(amounts) != len(hotkey_ss58s): + raise ValueError("amounts must be a list of the same length as hotkey_ss58s") + + if amounts is not None and not all( + isinstance(amount, (Balance, float)) for amount in amounts + ): + raise TypeError( + "amounts must be a [list of bittensor.Balance or float] or None" + ) + + if amounts is None: + amounts = [None] * len(hotkey_ss58s) + else: + # Convert to Balance + amounts = [ + Balance.from_tao(amount) if isinstance(amount, float) else amount + for amount in amounts + ] + + if sum(amount.tao for amount in amounts) == 0: + # Staking 0 tao + return True + + # Decrypt keys, + if not (unlock := unlock_key(wallet)).success: + logging.error(unlock.message) + return False + + old_stakes = [] + + logging.info( + f":satellite: [magenta]Syncing with chain:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + old_balance = subtensor.get_balance(wallet.coldkeypub.ss58_address) + + # Get the old stakes. + for hotkey_ss58 in hotkey_ss58s: + old_stakes.append( + subtensor.get_stake_for_coldkey_and_hotkey( + coldkey_ss58=wallet.coldkeypub.ss58_address, hotkey_ss58=hotkey_ss58 + ) + ) + + # Remove existential balance to keep key alive. + # Keys must maintain a balance of at least 1000 rao to stay alive. + total_staking_rao = sum( + [amount.rao if amount is not None else 0 for amount in amounts] + ) + if total_staking_rao == 0: + # Staking all to the first wallet. + if old_balance.rao > 1000: + old_balance -= Balance.from_rao(1000) + + elif total_staking_rao < 1000: + # Staking less than 1000 rao to the wallets. + pass + else: + # Staking more than 1000 rao to the wallets. + # Reduce the amount to stake to each wallet to keep the balance above 1000 rao. + percent_reduction = 1 - (1000 / total_staking_rao) + amounts = [ + Balance.from_tao(amount.tao * percent_reduction) for amount in amounts + ] + + successful_stakes = 0 + for idx, (hotkey_ss58, amount, old_stake) in enumerate( + zip(hotkey_ss58s, amounts, old_stakes) + ): + staking_all = False + # Convert to bittensor.Balance + if amount is None: + # Stake it all. + staking_balance = Balance.from_tao(old_balance.tao) + staking_all = True + else: + # Amounts are cast to balance earlier in the function + assert isinstance(amount, Balance) + staking_balance = amount + + # Check enough to stake + if staking_balance > old_balance: + logging.error( + f":cross_mark: [red]Not enough balance[/red]: [green]{old_balance}[/green] to stake: [blue]{staking_balance}[/blue] from wallet: [white]{wallet.name}[/white]" + ) + continue + + try: + staking_response: bool = __do_add_stake_single( + subtensor=subtensor, + wallet=wallet, + hotkey_ss58=hotkey_ss58, + amount=staking_balance, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + # If we successfully staked. + if staking_response: + # We only wait here if we expect finalization. + + if idx < len(hotkey_ss58s) - 1: + # Wait for tx rate limit. + tx_rate_limit_blocks = subtensor.tx_rate_limit() + if tx_rate_limit_blocks > 0: + logging.error( + f":hourglass: [yellow]Waiting for tx rate limit: [white]{tx_rate_limit_blocks}[/white] blocks[/yellow]" + ) + sleep(tx_rate_limit_blocks * 12) # 12 seconds per block + + if not wait_for_finalization and not wait_for_inclusion: + old_balance -= staking_balance + successful_stakes += 1 + if staking_all: + # If staked all, no need to continue + break + + continue + + logging.success(":white_heavy_check_mark: [green]Finalized[/green]") + + block = subtensor.get_current_block() + new_stake = subtensor.get_stake_for_coldkey_and_hotkey( + coldkey_ss58=wallet.coldkeypub.ss58_address, + hotkey_ss58=hotkey_ss58, + block=block, + ) + new_balance = subtensor.get_balance( + wallet.coldkeypub.ss58_address, block=block + ) + logging.info( + "Stake ({}): [blue]{}[/blue] :arrow_right: [green]{}[/green]".format( + hotkey_ss58, old_stake, new_stake + ) + ) + old_balance = new_balance + successful_stakes += 1 + if staking_all: + # If staked all, no need to continue + break + + else: + logging.error(":cross_mark: [red]Failed[/red]: Error unknown.") + continue + + except NotRegisteredError: + logging.error( + ":cross_mark: [red]Hotkey: {} is not registered.[/red]".format( + hotkey_ss58 + ) + ) + continue + except StakeError as e: + logging.error(":cross_mark: [red]Stake Error: {}[/red]".format(e)) + continue + + if successful_stakes != 0: + logging.info( + f":satellite: [magenta]Checking Balance on:[/magenta] ([blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + new_balance = subtensor.get_balance(wallet.coldkeypub.ss58_address) + logging.info( + f"Balance: [blue]{old_balance}[/blue] :arrow_right: [green]{new_balance}[/green]" + ) + return True + + return False diff --git a/bittensor/core/extrinsics/unstaking.py b/bittensor/core/extrinsics/unstaking.py new file mode 100644 index 0000000000..5089410187 --- /dev/null +++ b/bittensor/core/extrinsics/unstaking.py @@ -0,0 +1,416 @@ +from time import sleep +from typing import Union, Optional, TYPE_CHECKING + +from bittensor.core.errors import StakeError, NotRegisteredError +from bittensor.utils import format_error_message, unlock_key +from bittensor.utils.balance import Balance +from bittensor.utils.btlogging import logging +from bittensor.utils.networking import ensure_connected + +if TYPE_CHECKING: + from bittensor_wallet import Wallet + from bittensor.core.subtensor import Subtensor + + +@ensure_connected +def _do_unstake( + self: "Subtensor", + wallet: "Wallet", + hotkey_ss58: str, + amount: "Balance", + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, +) -> bool: + """Sends an unstake extrinsic to the chain. + + Args: + wallet (bittensor_wallet.Wallet): Wallet object that can sign the extrinsic. + hotkey_ss58 (str): Hotkey ``ss58`` address to unstake from. + amount (bittensor.utils.balance.Balance): Amount to unstake. + wait_for_inclusion (bool): If ``true``, waits for inclusion before returning. + wait_for_finalization (bool): If ``true``, waits for finalization before returning. + + Returns: + success (bool): ``True`` if the extrinsic was successful. + + Raises: + StakeError: If the extrinsic failed. + """ + + call = self.substrate.compose_call( + call_module="SubtensorModule", + call_function="remove_stake", + call_params={"hotkey": hotkey_ss58, "amount_unstaked": amount.rao}, + ) + extrinsic = self.substrate.create_signed_extrinsic( + call=call, keypair=wallet.coldkey + ) + response = self.substrate.submit_extrinsic( + extrinsic, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + # We only wait here if we expect finalization. + if not wait_for_finalization and not wait_for_inclusion: + return True + + response.process_events() + if response.is_success: + return True + else: + raise StakeError( + format_error_message(response.error_message, substrate=self.substrate) + ) + + +def _check_threshold_amount(subtensor: "Subtensor", stake_balance: "Balance") -> bool: + """ + Checks if the remaining stake balance is above the minimum required stake threshold. + + Args: + subtensor (bittensor.core.subtensor.Subtensor): Subtensor instance. + stake_balance (bittensor.utils.balance.Balance): the balance to check for threshold limits. + + Returns: + success (bool): ``true`` if the unstaking is above the threshold or 0, or ``false`` if the unstaking is below the threshold, but not 0. + """ + min_req_stake: Balance = subtensor.get_minimum_required_stake() + + if min_req_stake > stake_balance > 0: + logging.warning( + f":cross_mark: [yellow]Remaining stake balance of {stake_balance} less than minimum of {min_req_stake} TAO[/yellow]" + ) + return False + else: + return True + + +def __do_remove_stake_single( + subtensor: "Subtensor", + wallet: "Wallet", + hotkey_ss58: str, + amount: "Balance", + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, +) -> bool: + """ + Executes an unstake call to the chain using the wallet and the amount specified. + + Args: + wallet (bittensor_wallet.Wallet): Bittensor wallet object. + hotkey_ss58 (str): Hotkey address to unstake from. + amount (bittensor.utils.balance.Balance): Amount to unstake as Bittensor balance object. + wait_for_inclusion (bool): If set, waits for the extrinsic to enter a block before returning ``true``, or returns ``false`` if the extrinsic fails to enter the block within the timeout. + wait_for_finalization (bool): If set, waits for the extrinsic to be finalized on the chain before returning ``true``, or returns ``false`` if the extrinsic fails to be finalized within the timeout. + + Returns: + success (bool): Flag is ``true`` if extrinsic was finalized or uncluded in the block. If we did not wait for finalization / inclusion, the response is ``true``. + + Raises: + bittensor.core.errors.StakeError: If the extrinsic fails to be finalized or included in the block. + bittensor.core.errors.NotRegisteredError: If the hotkey is not registered in any subnets. + + """ + if not (unlock := unlock_key(wallet)).success: + logging.error(unlock.message) + return False + + success = _do_unstake( + self=subtensor, + wallet=wallet, + hotkey_ss58=hotkey_ss58, + amount=amount, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + return success + + +def unstake_extrinsic( + subtensor: "Subtensor", + wallet: "Wallet", + hotkey_ss58: Optional[str] = None, + amount: Optional[Union[Balance, float]] = None, + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, +) -> bool: + """Removes stake into the wallet coldkey from the specified hotkey ``uid``. + + Args: + subtensor (bittensor.core.subtensor.Subtensor): Subtensor instance. + wallet (bittensor_wallet.Wallet): Bittensor wallet object. + hotkey_ss58 (Optional[str]): The ``ss58`` address of the hotkey to unstake from. By default, the wallet hotkey is used. + amount (Union[Balance, float]): Amount to stake as Bittensor balance, or ``float`` interpreted as Tao. + wait_for_inclusion (bool): If set, waits for the extrinsic to enter a block before returning ``true``, or returns ``false`` if the extrinsic fails to enter the block within the timeout. + wait_for_finalization (bool): If set, waits for the extrinsic to be finalized on the chain before returning ``true``, or returns ``false`` if the extrinsic fails to be finalized within the timeout. + + Returns: + success (bool): Flag is ``true`` if extrinsic was finalized or uncluded in the block. If we did not wait for finalization / inclusion, the response is ``true``. + """ + # Decrypt keys, + if not (unlock := unlock_key(wallet)).success: + logging.error(unlock.message) + return False + + if hotkey_ss58 is None: + hotkey_ss58 = wallet.hotkey.ss58_address # Default to wallet's own hotkey. + + logging.info( + f":satellite: [magenta]Syncing with chain:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + old_balance = subtensor.get_balance(wallet.coldkeypub.ss58_address) + old_stake = subtensor.get_stake_for_coldkey_and_hotkey( + coldkey_ss58=wallet.coldkeypub.ss58_address, hotkey_ss58=hotkey_ss58 + ) + + hotkey_owner = subtensor.get_hotkey_owner(hotkey_ss58) + own_hotkey: bool = wallet.coldkeypub.ss58_address == hotkey_owner + + # Convert to bittensor.Balance + if amount is None: + # Unstake it all. + unstaking_balance = old_stake + elif not isinstance(amount, Balance): + unstaking_balance = Balance.from_tao(amount) + else: + unstaking_balance = amount + + # Check enough to unstake. + stake_on_uid = old_stake + if unstaking_balance > stake_on_uid: + logging.error( + f":cross_mark: [red]Not enough stake[/red]: [green]{stake_on_uid}[/green] to unstake: [blue]{unstaking_balance}[/blue] from hotkey: [yellow]{wallet.hotkey_str}[/yellow]" + ) + return False + + # If nomination stake, check threshold. + if not own_hotkey and not _check_threshold_amount( + subtensor=subtensor, stake_balance=(stake_on_uid - unstaking_balance) + ): + logging.warning( + ":warning: [yellow]This action will unstake the entire staked balance![/yellow]" + ) + unstaking_balance = stake_on_uid + + try: + logging.info( + f":satellite: [magenta]Unstaking from chain:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + staking_response: bool = __do_remove_stake_single( + subtensor=subtensor, + wallet=wallet, + hotkey_ss58=hotkey_ss58, + amount=unstaking_balance, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + if staking_response is True: # If we successfully unstaked. + # We only wait here if we expect finalization. + if not wait_for_finalization and not wait_for_inclusion: + return True + + logging.success(":white_heavy_check_mark: [green]Finalized[/green]") + + logging.info( + f":satellite: [magenta]Checking Balance on:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + new_balance = subtensor.get_balance(address=wallet.coldkeypub.ss58_address) + new_stake = subtensor.get_stake_for_coldkey_and_hotkey( + coldkey_ss58=wallet.coldkeypub.ss58_address, hotkey_ss58=hotkey_ss58 + ) # Get stake on hotkey. + logging.info(f"Balance:") + logging.info( + f"\t\t[blue]{old_balance}[/blue] :arrow_right: [green]{new_balance}[/green]" + ) + logging.info("Stake:") + logging.info( + f"\t\t[blue]{old_stake}[/blue] :arrow_right: [green]{new_stake}[/green]" + ) + return True + else: + logging.error(":cross_mark: [red]Failed[/red]: Unknown Error.") + return False + + except NotRegisteredError: + logging.error( + f":cross_mark: [red]Hotkey: {wallet.hotkey_str} is not registered.[/red]" + ) + return False + except StakeError as e: + logging.error(":cross_mark: [red]Stake Error: {}[/red]".format(e)) + return False + + +def unstake_multiple_extrinsic( + subtensor: "Subtensor", + wallet: "Wallet", + hotkey_ss58s: list[str], + amounts: Optional[list[Union[Balance, float]]] = None, + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, +) -> bool: + """Removes stake from each ``hotkey_ss58`` in the list, using each amount, to a common coldkey. + + Args: + subtensor (bittensor.core.subtensor.Subtensor): Subtensor instance. + wallet (bittensor_wallet.Wallet): The wallet with the coldkey to unstake to. + hotkey_ss58s (List[str]): List of hotkeys to unstake from. + amounts (List[Union[Balance, float]]): List of amounts to unstake. If ``None``, unstake all. + wait_for_inclusion (bool): If set, waits for the extrinsic to enter a block before returning ``true``, or returns ``false`` if the extrinsic fails to enter the block within the timeout. + wait_for_finalization (bool): If set, waits for the extrinsic to be finalized on the chain before returning ``true``, or returns ``false`` if the extrinsic fails to be finalized within the timeout. + + Returns: + success (bool): Flag is ``true`` if extrinsic was finalized or included in the block. Flag is ``true`` if any wallet was unstaked. If we did not wait for finalization / inclusion, the response is ``true``. + """ + if not isinstance(hotkey_ss58s, list) or not all( + isinstance(hotkey_ss58, str) for hotkey_ss58 in hotkey_ss58s + ): + raise TypeError("hotkey_ss58s must be a list of str") + + if len(hotkey_ss58s) == 0: + return True + + if amounts is not None and len(amounts) != len(hotkey_ss58s): + raise ValueError("amounts must be a list of the same length as hotkey_ss58s") + + if amounts is not None and not all( + isinstance(amount, (Balance, float)) for amount in amounts + ): + raise TypeError( + "amounts must be a [list of bittensor.Balance or float] or None" + ) + + if amounts is None: + amounts = [None] * len(hotkey_ss58s) + else: + # Convert to Balance + amounts = [ + Balance.from_tao(amount) if isinstance(amount, float) else amount + for amount in amounts + ] + + if sum(amount.tao for amount in amounts) == 0: + # Staking 0 tao + return True + + # Unlock coldkey. + if not (unlock := unlock_key(wallet)).success: + logging.error(unlock.message) + return False + + old_stakes = [] + own_hotkeys = [] + logging.info( + f":satellite: [magenta]Syncing with chain:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + old_balance = subtensor.get_balance(wallet.coldkeypub.ss58_address) + + for hotkey_ss58 in hotkey_ss58s: + old_stake = subtensor.get_stake_for_coldkey_and_hotkey( + coldkey_ss58=wallet.coldkeypub.ss58_address, hotkey_ss58=hotkey_ss58 + ) # Get stake on hotkey. + old_stakes.append(old_stake) # None if not registered. + + hotkey_owner = subtensor.get_hotkey_owner(hotkey_ss58) + own_hotkeys.append(wallet.coldkeypub.ss58_address == hotkey_owner) + + successful_unstakes = 0 + for idx, (hotkey_ss58, amount, old_stake, own_hotkey) in enumerate( + zip(hotkey_ss58s, amounts, old_stakes, own_hotkeys) + ): + # Covert to bittensor.Balance + if amount is None: + # Unstake it all. + unstaking_balance = old_stake + else: + unstaking_balance = ( + amount if isinstance(amount, Balance) else Balance.from_tao(amount) + ) + + # Check enough to unstake. + stake_on_uid = old_stake + if unstaking_balance > stake_on_uid: + logging.error( + f":cross_mark: [red]Not enough stake[/red]: [green]{stake_on_uid}[/green] to unstake: [blue]{unstaking_balance}[/blue] from hotkey: [blue]{wallet.hotkey_str}[/blue]." + ) + continue + + # If nomination stake, check threshold. + if not own_hotkey and not _check_threshold_amount( + subtensor=subtensor, stake_balance=(stake_on_uid - unstaking_balance) + ): + logging.warning( + f":warning: [yellow]This action will unstake the entire staked balance![/yellow]" + ) + unstaking_balance = stake_on_uid + + try: + logging.info( + f":satellite: [magenta]Unstaking from chain:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + staking_response: bool = __do_remove_stake_single( + subtensor=subtensor, + wallet=wallet, + hotkey_ss58=hotkey_ss58, + amount=unstaking_balance, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + if staking_response is True: # If we successfully unstaked. + # We only wait here if we expect finalization. + + if idx < len(hotkey_ss58s) - 1: + # Wait for tx rate limit. + tx_rate_limit_blocks = subtensor.tx_rate_limit() + if tx_rate_limit_blocks > 0: + logging.info( + f":hourglass: [yellow]Waiting for tx rate limit: [white]{tx_rate_limit_blocks}[/white] blocks[/yellow]" + ) + sleep(tx_rate_limit_blocks * 12) # 12 seconds per block + + if not wait_for_finalization and not wait_for_inclusion: + successful_unstakes += 1 + continue + + logging.info(":white_heavy_check_mark: [green]Finalized[/green]") + + logging.info( + f":satellite: [magenta]Checking Balance on:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]..." + ) + block = subtensor.get_current_block() + new_stake = subtensor.get_stake_for_coldkey_and_hotkey( + coldkey_ss58=wallet.coldkeypub.ss58_address, + hotkey_ss58=hotkey_ss58, + block=block, + ) + logging.info( + f"Stake ({hotkey_ss58}): [blue]{stake_on_uid}[/blue] :arrow_right: [green]{new_stake}[/green]" + ) + successful_unstakes += 1 + else: + logging.error(":cross_mark: [red]Failed: Unknown Error.[/red]") + continue + + except NotRegisteredError: + logging.error( + f":cross_mark: [red]Hotkey[/red] [blue]{hotkey_ss58}[/blue] [red]is not registered.[/red]" + ) + continue + except StakeError as e: + logging.error(":cross_mark: [red]Stake Error: {}[/red]".format(e)) + continue + + if successful_unstakes != 0: + logging.info( + f":satellite: [magenta]Checking Balance on:[/magenta] ([blue]{subtensor.network}[/blue] [magenta]...[/magenta]" + ) + new_balance = subtensor.get_balance(wallet.coldkeypub.ss58_address) + logging.info( + f"Balance: [blue]{old_balance}[/blue] :arrow_right: [green]{new_balance}[/green]" + ) + return True + + return False diff --git a/bittensor/core/subtensor.py b/bittensor/core/subtensor.py index 8e4be9dd29..a26a02638c 100644 --- a/bittensor/core/subtensor.py +++ b/bittensor/core/subtensor.py @@ -53,6 +53,14 @@ from bittensor.core.extrinsics.transfer import ( transfer_extrinsic, ) +from bittensor.core.extrinsics.staking import ( + add_stake_extrinsic, + add_stake_multiple_extrinsic, +) +from bittensor.core.extrinsics.unstaking import ( + unstake_extrinsic, + unstake_multiple_extrinsic, +) from bittensor.core.metagraph import Metagraph from bittensor.utils import ( networking, @@ -420,7 +428,7 @@ def _get_hyperparameter( return result.value - # Calls methods + # Chain calls methods ============================================================================================== @networking.ensure_connected def query_subtensor( self, name: str, block: Optional[int] = None, params: Optional[list] = None @@ -634,7 +642,7 @@ def query_module( ), ) - # Common subtensor methods + # Common subtensor methods ========================================================================================= def metagraph( self, netuid: int, lite: bool = True, block: Optional[int] = None ) -> "Metagraph": # type: ignore @@ -787,236 +795,6 @@ def is_hotkey_registered( else: return self.is_hotkey_registered_on_subnet(hotkey_ss58, netuid, block) - # Not used in Bittensor, but is actively used by the community in almost all subnets - def set_weights( - self, - wallet: "Wallet", - netuid: int, - uids: Union[NDArray[np.int64], "torch.LongTensor", list], - weights: Union[NDArray[np.float32], "torch.FloatTensor", list], - version_key: int = settings.version_as_int, - wait_for_inclusion: bool = False, - wait_for_finalization: bool = False, - max_retries: int = 5, - ) -> tuple[bool, str]: - """ - Sets the inter-neuronal weights for the specified neuron. This process involves specifying the influence or trust a neuron places on other neurons in the network, which is a fundamental aspect of Bittensor's decentralized learning architecture. - - Args: - wallet (bittensor_wallet.Wallet): The wallet associated with the neuron setting the weights. - netuid (int): The unique identifier of the subnet. - uids (Union[NDArray[np.int64], torch.LongTensor, list]): The list of neuron UIDs that the weights are being set for. - weights (Union[NDArray[np.float32], torch.FloatTensor, list]): The corresponding weights to be set for each UID. - version_key (int): Version key for compatibility with the network. Default is ``int representation of Bittensor version.``. - wait_for_inclusion (bool): Waits for the transaction to be included in a block. Default is ``False``. - wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Default is ``False``. - max_retries (int): The number of maximum attempts to set weights. Default is ``5``. - - Returns: - tuple[bool, str]: ``True`` if the setting of weights is successful, False otherwise. And `msg`, a string value describing the success or potential error. - - This function is crucial in shaping the network's collective intelligence, where each neuron's learning and contribution are influenced by the weights it sets towards others【81†source】. - """ - uid = self.get_uid_for_hotkey_on_subnet(wallet.hotkey.ss58_address, netuid) - retries = 0 - success = False - message = "No attempt made. Perhaps it is too soon to set weights!" - while ( - self.blocks_since_last_update(netuid, uid) > self.weights_rate_limit(netuid) # type: ignore - and retries < max_retries - ): - try: - logging.info( - f"Setting weights for subnet #{netuid}. Attempt {retries + 1} of {max_retries}." - ) - success, message = set_weights_extrinsic( - subtensor=self, - wallet=wallet, - netuid=netuid, - uids=uids, - weights=weights, - version_key=version_key, - wait_for_inclusion=wait_for_inclusion, - wait_for_finalization=wait_for_finalization, - ) - except Exception as e: - logging.error(f"Error setting weights: {e}") - finally: - retries += 1 - - return success, message - - @legacy_torch_api_compat - def root_set_weights( - self, - wallet: "Wallet", - netuids: Union[NDArray[np.int64], "torch.LongTensor", list], - weights: Union[NDArray[np.float32], "torch.FloatTensor", list], - version_key: int = 0, - wait_for_inclusion: bool = False, - wait_for_finalization: bool = False, - ) -> bool: - """ - Sets the weights for neurons on the root network. This action is crucial for defining the influence and interactions of neurons at the root level of the Bittensor network. - - Args: - wallet (bittensor_wallet.Wallet): The wallet associated with the neuron setting the weights. - netuids (Union[NDArray[np.int64], torch.LongTensor, list]): The list of neuron UIDs for which weights are being set. - weights (Union[NDArray[np.float32], torch.FloatTensor, list]): The corresponding weights to be set for each UID. - version_key (int, optional): Version key for compatibility with the network. Default is ``0``. - wait_for_inclusion (bool, optional): Waits for the transaction to be included in a block. Defaults to ``False``. - wait_for_finalization (bool, optional): Waits for the transaction to be finalized on the blockchain. Defaults to ``False``. - - Returns: - bool: ``True`` if the setting of root-level weights is successful, False otherwise. - - This function plays a pivotal role in shaping the root network's collective intelligence and decision-making processes, reflecting the principles of decentralized governance and collaborative learning in Bittensor. - """ - return set_root_weights_extrinsic( - subtensor=self, - wallet=wallet, - netuids=netuids, - weights=weights, - version_key=version_key, - wait_for_inclusion=wait_for_inclusion, - wait_for_finalization=wait_for_finalization, - ) - - def register( - self, - wallet: "Wallet", - netuid: int, - wait_for_inclusion: bool = False, - wait_for_finalization: bool = True, - max_allowed_attempts: int = 3, - output_in_place: bool = True, - cuda: bool = False, - dev_id: Union[list[int], int] = 0, - tpb: int = 256, - num_processes: Optional[int] = None, - update_interval: Optional[int] = None, - log_verbose: bool = False, - ) -> bool: - """ - Registers a neuron on the Bittensor network using the provided wallet. - - Registration is a critical step for a neuron to become an active participant in the network, enabling it to stake, set weights, and receive incentives. - - Args: - wallet (bittensor_wallet.Wallet): The wallet associated with the neuron to be registered. - netuid (int): The unique identifier of the subnet. - wait_for_inclusion (bool): Waits for the transaction to be included in a block. Defaults to `False`. - wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Defaults to `True`. - max_allowed_attempts (int): Maximum number of attempts to register the wallet. - output_in_place (bool): If true, prints the progress of the proof of work to the console in-place. Meaning the progress is printed on the same lines. Defaults to `True`. - cuda (bool): If ``true``, the wallet should be registered using CUDA device(s). Defaults to `False`. - dev_id (Union[List[int], int]): The CUDA device id to use, or a list of device ids. Defaults to `0` (zero). - tpb (int): The number of threads per block (CUDA). Default to `256`. - num_processes (Optional[int]): The number of processes to use to register. Default to `None`. - update_interval (Optional[int]): The number of nonces to solve between updates. Default to `None`. - log_verbose (bool): If ``true``, the registration process will log more information. Default to `False`. - - Returns: - bool: ``True`` if the registration is successful, False otherwise. - - This function facilitates the entry of new neurons into the network, supporting the decentralized - growth and scalability of the Bittensor ecosystem. - """ - return register_extrinsic( - subtensor=self, - wallet=wallet, - netuid=netuid, - wait_for_inclusion=wait_for_inclusion, - wait_for_finalization=wait_for_finalization, - max_allowed_attempts=max_allowed_attempts, - output_in_place=output_in_place, - cuda=cuda, - dev_id=dev_id, - tpb=tpb, - num_processes=num_processes, - update_interval=update_interval, - log_verbose=log_verbose, - ) - - def root_register( - self, - wallet: "Wallet", - wait_for_inclusion: bool = False, - wait_for_finalization: bool = True, - ) -> bool: - """ - Registers the neuron associated with the wallet on the root network. This process is integral for participating in the highest layer of decision-making and governance within the Bittensor network. - - Args: - wallet (bittensor.wallet): The wallet associated with the neuron to be registered on the root network. - wait_for_inclusion (bool): Waits for the transaction to be included in a block. Defaults to `False`. - wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Defaults to `True`. - - Returns: - bool: ``True`` if the registration on the root network is successful, False otherwise. - - This function enables neurons to engage in the most critical and influential aspects of the network's governance, signifying a high level of commitment and responsibility in the Bittensor ecosystem. - """ - return root_register_extrinsic( - subtensor=self, - wallet=wallet, - wait_for_inclusion=wait_for_inclusion, - wait_for_finalization=wait_for_finalization, - ) - - def burned_register( - self, - wallet: "Wallet", - netuid: int, - wait_for_inclusion: bool = False, - wait_for_finalization: bool = True, - ) -> bool: - """ - Registers a neuron on the Bittensor network by recycling TAO. This method of registration involves recycling TAO tokens, allowing them to be re-mined by performing work on the network. - - Args: - wallet (bittensor_wallet.Wallet): The wallet associated with the neuron to be registered. - netuid (int): The unique identifier of the subnet. - wait_for_inclusion (bool, optional): Waits for the transaction to be included in a block. Defaults to `False`. - wait_for_finalization (bool, optional): Waits for the transaction to be finalized on the blockchain. Defaults to `True`. - - Returns: - bool: ``True`` if the registration is successful, False otherwise. - """ - return burned_register_extrinsic( - subtensor=self, - wallet=wallet, - netuid=netuid, - wait_for_inclusion=wait_for_inclusion, - wait_for_finalization=wait_for_finalization, - ) - - def serve_axon( - self, - netuid: int, - axon: "Axon", - wait_for_inclusion: bool = False, - wait_for_finalization: bool = True, - certificate: Optional[Certificate] = None, - ) -> bool: - """ - Registers an ``Axon`` serving endpoint on the Bittensor network for a specific neuron. This function is used to set up the Axon, a key component of a neuron that handles incoming queries and data processing tasks. - - Args: - netuid (int): The unique identifier of the subnetwork. - axon (bittensor.core.axon.Axon): The Axon instance to be registered for serving. - wait_for_inclusion (bool): Waits for the transaction to be included in a block. Default is ``False``. - wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Default is ``True``. - - Returns: - bool: ``True`` if the Axon serve registration is successful, False otherwise. - - By registering an Axon, the neuron becomes an active part of the network's distributed computing infrastructure, contributing to the collective intelligence of Bittensor. - """ - return serve_axon_extrinsic( - self, netuid, axon, wait_for_inclusion, wait_for_finalization, certificate - ) - # metagraph @property def block(self) -> int: @@ -1069,8 +847,6 @@ def weights_rate_limit(self, netuid: int) -> Optional[int]: call = self._get_hyperparameter(param_name="WeightsSetRateLimit", netuid=netuid) return None if call is None else int(call) - # Keep backwards compatibility for community usage. - # Make some commitment on-chain about arbitrary data. def commit(self, wallet, netuid: int, data: str): """ Commits arbitrary data to the Bittensor network by publishing metadata. @@ -1082,7 +858,6 @@ def commit(self, wallet, netuid: int, data: str): """ publish_metadata(self, wallet, netuid, f"Raw{len(data)}", data.encode()) - # Keep backwards compatibility for community usage. def subnetwork_n(self, netuid: int, block: Optional[int] = None) -> Optional[int]: """ Returns network SubnetworkN hyperparameter. @@ -1099,55 +874,21 @@ def subnetwork_n(self, netuid: int, block: Optional[int] = None) -> Optional[int ) return None if call is None else int(call) - # Community uses this method - def transfer( - self, - wallet: "Wallet", - dest: str, - amount: Union["Balance", float], - wait_for_inclusion: bool = True, - wait_for_finalization: bool = False, - ) -> bool: + def get_neuron_for_pubkey_and_subnet( + self, hotkey_ss58: str, netuid: int, block: Optional[int] = None + ) -> Optional["NeuronInfo"]: """ - Executes a transfer of funds from the provided wallet to the specified destination address. This function is used to move TAO tokens within the Bittensor network, facilitating transactions between neurons. + Retrieves information about a neuron based on its public key (hotkey SS58 address) and the specific subnet UID (netuid). This function provides detailed neuron information for a particular subnet within the Bittensor network. Args: - wallet (bittensor_wallet.Wallet): The wallet from which funds are being transferred. - dest (str): The destination public key address. - amount (Union[bittensor.utils.balance.Balance, float]): The amount of TAO to be transferred. - wait_for_inclusion (bool): Waits for the transaction to be included in a block. Default is ``True``. - wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Default is ``False``. + hotkey_ss58 (str): The ``SS58`` address of the neuron's hotkey. + netuid (int): The unique identifier of the subnet. + block (Optional[int]): The blockchain block number at which to perform the query. Returns: - transfer_extrinsic (bool): ``True`` if the transfer is successful, False otherwise. + Optional[bittensor.core.chain_data.neuron_info.NeuronInfo]: Detailed information about the neuron if found, ``None`` otherwise. - This function is essential for the fluid movement of tokens in the network, supporting various economic activities such as staking, delegation, and reward distribution. - """ - return transfer_extrinsic( - subtensor=self, - wallet=wallet, - dest=dest, - amount=amount, - wait_for_inclusion=wait_for_inclusion, - wait_for_finalization=wait_for_finalization, - ) - - # Community uses this method via `bittensor.api.extrinsics.prometheus.prometheus_extrinsic` - def get_neuron_for_pubkey_and_subnet( - self, hotkey_ss58: str, netuid: int, block: Optional[int] = None - ) -> Optional["NeuronInfo"]: - """ - Retrieves information about a neuron based on its public key (hotkey SS58 address) and the specific subnet UID (netuid). This function provides detailed neuron information for a particular subnet within the Bittensor network. - - Args: - hotkey_ss58 (str): The ``SS58`` address of the neuron's hotkey. - netuid (int): The unique identifier of the subnet. - block (Optional[int]): The blockchain block number at which to perform the query. - - Returns: - Optional[bittensor.core.chain_data.neuron_info.NeuronInfo]: Detailed information about the neuron if found, ``None`` otherwise. - - This function is crucial for accessing specific neuron data and understanding its status, stake, and other attributes within a particular subnet of the Bittensor ecosystem. + This function is crucial for accessing specific neuron data and understanding its status, stake, and other attributes within a particular subnet of the Bittensor ecosystem. """ return self.neuron_for_uid( self.get_uid_for_hotkey_on_subnet(hotkey_ss58, netuid, block=block), @@ -1225,7 +966,6 @@ def neuron_for_uid( return NeuronInfo.from_vec_u8(result) - # Community uses this method def get_subnet_hyperparameters( self, netuid: int, block: Optional[int] = None ) -> Optional[Union[list, "SubnetHyperparameters"]]: @@ -1253,8 +993,6 @@ def get_subnet_hyperparameters( return SubnetHyperparameters.from_vec_u8(hex_to_bytes(hex_bytes_result)) - # Community uses this method - # Returns network ImmunityPeriod hyper parameter. def immunity_period( self, netuid: int, block: Optional[int] = None ) -> Optional[int]: @@ -1275,7 +1013,6 @@ def immunity_period( ) return None if call is None else int(call) - # Community uses this method def get_uid_for_hotkey_on_subnet( self, hotkey_ss58: str, netuid: int, block: Optional[int] = None ) -> Optional[int]: @@ -1295,7 +1032,6 @@ def get_uid_for_hotkey_on_subnet( _result = self.query_subtensor("Uids", block, [netuid, hotkey_ss58]) return getattr(_result, "value", None) - # Community uses this method def tempo(self, netuid: int, block: Optional[int] = None) -> Optional[int]: """ Returns network Tempo hyperparameter. @@ -1310,7 +1046,6 @@ def tempo(self, netuid: int, block: Optional[int] = None) -> Optional[int]: call = self._get_hyperparameter(param_name="Tempo", netuid=netuid, block=block) return None if call is None else int(call) - # Community uses this method def get_commitment(self, netuid: int, uid: int, block: Optional[int] = None) -> str: """ Retrieves the on-chain commitment for a specific neuron in the Bittensor network. @@ -1335,7 +1070,6 @@ def get_commitment(self, netuid: int, uid: int, block: Optional[int] = None) -> except TypeError: return "" - # Community uses this via `bittensor.utils.weight_utils.process_weights_for_netuid` function. def min_allowed_weights( self, netuid: int, block: Optional[int] = None ) -> Optional[int]: @@ -1354,7 +1088,6 @@ def min_allowed_weights( ) return None if call is None else int(call) - # Community uses this via `bittensor.utils.weight_utils.process_weights_for_netuid` function. def max_weight_limit( self, netuid: int, block: Optional[int] = None ) -> Optional[float]: @@ -1373,7 +1106,6 @@ def max_weight_limit( ) return None if call is None else u16_normalized_float(int(call)) - # # Community uses this method. It is used in subtensor in neuron_info, and serving. def get_prometheus_info( self, netuid: int, hotkey_ss58: str, block: Optional[int] = None ) -> Optional["PrometheusInfo"]: @@ -1399,7 +1131,6 @@ def get_prometheus_info( ) return None - # Community uses this method def subnet_exists(self, netuid: int, block: Optional[int] = None) -> bool: """ Checks if a subnet with the specified unique identifier (netuid) exists within the Bittensor network. @@ -1437,7 +1168,6 @@ def get_all_subnets_info(self, block: Optional[int] = None) -> list[SubnetInfo]: else: return SubnetInfo.list_from_vec_u8(hex_to_bytes(hex_bytes_result)) - # Metagraph uses this method def bonds( self, netuid: int, block: Optional[int] = None ) -> list[tuple[int, list[tuple[int, int]]]]: @@ -1487,7 +1217,6 @@ def get_subnet_burn_cost(self, block: Optional[int] = None) -> Optional[str]: return lock_cost - # Metagraph uses this method def neurons(self, netuid: int, block: Optional[int] = None) -> list["NeuronInfo"]: """ Retrieves a list of all neurons within a specified subnet of the Bittensor network. This function provides a snapshot of the subnet's neuron population, including each neuron's attributes and network interactions. @@ -1517,7 +1246,6 @@ def neurons(self, netuid: int, block: Optional[int] = None) -> list["NeuronInfo" return neurons - # Metagraph uses this method def get_total_subnets(self, block: Optional[int] = None) -> Optional[int]: """ Retrieves the total number of subnets within the Bittensor network as of a specific blockchain block. @@ -1533,7 +1261,6 @@ def get_total_subnets(self, block: Optional[int] = None) -> Optional[int]: _result = self.query_subtensor("TotalNetworks", block) return getattr(_result, "value", None) - # Metagraph uses this method def get_subnets(self, block: Optional[int] = None) -> list[int]: """ Retrieves a list of all subnets currently active within the Bittensor network. This function provides an overview of the various subnets and their identifiers. @@ -1553,7 +1280,6 @@ def get_subnets(self, block: Optional[int] = None) -> list[int]: else [] ) - # Metagraph uses this method def neurons_lite( self, netuid: int, block: Optional[int] = None ) -> list["NeuronInfoLite"]: @@ -1581,7 +1307,6 @@ def neurons_lite( return NeuronInfoLite.list_from_vec_u8(hex_to_bytes(hex_bytes_result)) # type: ignore - # Used in the `neurons` method which is used in metagraph.py def weights( self, netuid: int, block: Optional[int] = None ) -> list[tuple[int, list[tuple[int, int]]]]: @@ -1607,7 +1332,6 @@ def weights( return w_map - # Used by community via `transfer_extrinsic` @networking.ensure_connected def get_balance(self, address: str, block: Optional[int] = None) -> "Balance": """ @@ -1640,7 +1364,6 @@ def get_balance(self, address: str, block: Optional[int] = None) -> "Balance": return Balance(result.value["data"]["free"]) - # Used in community via `bittensor.core.subtensor.Subtensor.transfer` @networking.ensure_connected def get_transfer_fee( self, wallet: "Wallet", dest: str, value: Union["Balance", float, int] @@ -1690,7 +1413,6 @@ def get_transfer_fee( ) return fee - # Used in community via `bittensor.core.subtensor.Subtensor.transfer` def get_existential_deposit( self, block: Optional[int] = None ) -> Optional["Balance"]: @@ -1712,142 +1434,6 @@ def get_existential_deposit( return None return Balance.from_rao(result.value) - # Community uses this method - def commit_weights( - self, - wallet: "Wallet", - netuid: int, - salt: list[int], - uids: Union[NDArray[np.int64], list], - weights: Union[NDArray[np.int64], list], - version_key: int = settings.version_as_int, - wait_for_inclusion: bool = False, - wait_for_finalization: bool = False, - max_retries: int = 5, - ) -> tuple[bool, str]: - """ - Commits a hash of the neuron's weights to the Bittensor blockchain using the provided wallet. - This action serves as a commitment or snapshot of the neuron's current weight distribution. - - Args: - wallet (bittensor_wallet.Wallet): The wallet associated with the neuron committing the weights. - netuid (int): The unique identifier of the subnet. - salt (list[int]): list of randomly generated integers as salt to generated weighted hash. - uids (np.ndarray): NumPy array of neuron UIDs for which weights are being committed. - weights (np.ndarray): NumPy array of weight values corresponding to each UID. - version_key (int): Version key for compatibility with the network. Default is ``int representation of Bittensor version.``. - wait_for_inclusion (bool): Waits for the transaction to be included in a block. Default is ``False``. - wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Default is ``False``. - max_retries (int): The number of maximum attempts to commit weights. Default is ``5``. - - Returns: - tuple[bool, str]: ``True`` if the weight commitment is successful, False otherwise. And `msg`, a string - value describing the success or potential error. - - This function allows neurons to create a tamper-proof record of their weight distribution at a specific point in time, - enhancing transparency and accountability within the Bittensor network. - """ - retries = 0 - success = False - message = "No attempt made. Perhaps it is too soon to commit weights!" - - logging.info( - f"Committing weights with params: netuid={netuid}, uids={uids}, weights={weights}, version_key={version_key}" - ) - - # Generate the hash of the weights - commit_hash = generate_weight_hash( - address=wallet.hotkey.ss58_address, - netuid=netuid, - uids=list(uids), - values=list(weights), - salt=salt, - version_key=version_key, - ) - - logging.info(f"Commit Hash: {commit_hash}") - - while retries < max_retries: - try: - success, message = commit_weights_extrinsic( - subtensor=self, - wallet=wallet, - netuid=netuid, - commit_hash=commit_hash, - wait_for_inclusion=wait_for_inclusion, - wait_for_finalization=wait_for_finalization, - ) - if success: - break - except Exception as e: - logging.error(f"Error committing weights: {e}") - finally: - retries += 1 - - return success, message - - # Community uses this method - def reveal_weights( - self, - wallet: "Wallet", - netuid: int, - uids: Union[NDArray[np.int64], list], - weights: Union[NDArray[np.int64], list], - salt: Union[NDArray[np.int64], list], - version_key: int = settings.version_as_int, - wait_for_inclusion: bool = False, - wait_for_finalization: bool = False, - max_retries: int = 5, - ) -> tuple[bool, str]: - """ - Reveals the weights for a specific subnet on the Bittensor blockchain using the provided wallet. - This action serves as a revelation of the neuron's previously committed weight distribution. - - Args: - wallet (bittensor_wallet.Wallet): The wallet associated with the neuron revealing the weights. - netuid (int): The unique identifier of the subnet. - uids (np.ndarray): NumPy array of neuron UIDs for which weights are being revealed. - weights (np.ndarray): NumPy array of weight values corresponding to each UID. - salt (np.ndarray): NumPy array of salt values corresponding to the hash function. - version_key (int): Version key for compatibility with the network. Default is ``int representation of Bittensor version``. - wait_for_inclusion (bool): Waits for the transaction to be included in a block. Default is ``False``. - wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Default is ``False``. - max_retries (int): The number of maximum attempts to reveal weights. Default is ``5``. - - Returns: - tuple[bool, str]: ``True`` if the weight revelation is successful, False otherwise. And `msg`, a string - value describing the success or potential error. - - This function allows neurons to reveal their previously committed weight distribution, ensuring transparency - and accountability within the Bittensor network. - """ - - retries = 0 - success = False - message = "No attempt made. Perhaps it is too soon to reveal weights!" - - while retries < max_retries: - try: - success, message = reveal_weights_extrinsic( - subtensor=self, - wallet=wallet, - netuid=netuid, - uids=list(uids), - weights=list(weights), - salt=list(salt), - version_key=version_key, - wait_for_inclusion=wait_for_inclusion, - wait_for_finalization=wait_for_finalization, - ) - if success: - break - except Exception as e: - logging.error(f"Error revealing weights: {e}") - finally: - retries += 1 - - return success, message - def difficulty(self, netuid: int, block: Optional[int] = None) -> Optional[int]: """ Retrieves the 'Difficulty' hyperparameter for a specified subnet in the Bittensor network. @@ -1938,5 +1524,667 @@ def get_delegate_by_hotkey( return DelegateInfo.from_vec_u8(bytes(result)) - # Subnet 27 uses this method name - _do_serve_axon = do_serve_axon + def get_stake_for_coldkey_and_hotkey( + self, hotkey_ss58: str, coldkey_ss58: str, block: Optional[int] = None + ) -> Optional["Balance"]: + """ + Returns the stake under a coldkey - hotkey pairing. + + Args: + hotkey_ss58 (str): The SS58 address of the hotkey. + coldkey_ss58 (str): The SS58 address of the coldkey. + block (Optional[int]): The block number to retrieve the stake from. If ``None``, the latest block is used. Default is ``None``. + + Returns: + Optional[Balance]: The stake under the coldkey - hotkey pairing, or ``None`` if the pairing does not exist or the stake is not found. + """ + result = self.query_subtensor("Stake", block, [hotkey_ss58, coldkey_ss58]) + return ( + None + if getattr(result, "value", None) is None + else Balance.from_rao(result.value) + ) + + def does_hotkey_exist(self, hotkey_ss58: str, block: Optional[int] = None) -> bool: + """ + Returns true if the hotkey is known by the chain and there are accounts. + + Args: + hotkey_ss58 (str): The SS58 address of the hotkey. + block (Optional[int]): The block number to check the hotkey against. If ``None``, the latest block is used. Default is ``None``. + + Returns: + bool: ``True`` if the hotkey is known by the chain and there are accounts, ``False`` otherwise. + """ + result = self.query_subtensor("Owner", block, [hotkey_ss58]) + return ( + False + if getattr(result, "value", None) is None + else result.value != "5C4hrfjw9DjXZTzV3MwzrrAr9P1MJhSrvWGWqi1eSuyUpnhM" + ) + + def get_hotkey_owner( + self, hotkey_ss58: str, block: Optional[int] = None + ) -> Optional[str]: + """ + Returns the coldkey owner of the passed hotkey. + + Args: + hotkey_ss58 (str): The SS58 address of the hotkey. + block (Optional[int]): The block number to check the hotkey owner against. If ``None``, the latest block is used. Default is ``None``. + + Returns: + Optional[str]: The SS58 address of the coldkey owner, or ``None`` if the hotkey does not exist or the owner is not found. + """ + result = self.query_subtensor("Owner", block, [hotkey_ss58]) + return ( + None + if getattr(result, "value", None) is None + or not self.does_hotkey_exist(hotkey_ss58, block) + else result.value + ) + + @networking.ensure_connected + def get_minimum_required_stake( + self, + ) -> Balance: + """ + Returns the minimum required stake for nominators in the Subtensor network. + + This method retries the substrate call up to three times with exponential backoff in case of failures. + + Returns: + Balance: The minimum required stake as a Balance object. + + Raises: + Exception: If the substrate call fails after the maximum number of retries. + """ + + result = self.substrate.query( + module="SubtensorModule", storage_function="NominatorMinRequiredStake" + ) + return Balance.from_rao(result.decode()) + + def tx_rate_limit(self, block: Optional[int] = None) -> Optional[int]: + """ + Retrieves the transaction rate limit for the Bittensor network as of a specific blockchain block. + This rate limit sets the maximum number of transactions that can be processed within a given time frame. + + Args: + block (Optional[int]): The blockchain block number at which to perform the query. + + Returns: + Optional[int]: The transaction rate limit of the network, None if not available. + + The transaction rate limit is an essential parameter for ensuring the stability and scalability of the Bittensor network. It helps in managing network load and preventing congestion, thereby maintaining efficient and timely transaction processing. + """ + result = self.query_subtensor("TxRateLimit", block) + return getattr(result, "value", None) + + @networking.ensure_connected + def get_delegates(self, block: Optional[int] = None) -> list[DelegateInfo]: + """ + Retrieves a list of all delegate neurons within the Bittensor network. This function provides an overview of the neurons that are actively involved in the network's delegation system. + + Analyzing the delegate population offers insights into the network's governance dynamics and the distribution of trust and responsibility among participating neurons. + + Args: + block (Optional[int]): The blockchain block number for the query. + + Returns: + list[DelegateInfo]: A list of DelegateInfo objects detailing each delegate's characteristics. + + """ + block_hash = None if block is None else self.substrate.get_block_hash(block) + + json_body = self.substrate.rpc_request( + method="delegateInfo_getDelegates", + params=[block_hash] if block_hash else [], + ) + + if not (result := json_body.get("result", None)): + return [] + + return DelegateInfo.list_from_vec_u8(result) + + def is_hotkey_delegate(self, hotkey_ss58: str, block: Optional[int] = None) -> bool: + """ + Determines whether a given hotkey (public key) is a delegate on the Bittensor network. This function checks if the neuron associated with the hotkey is part of the network's delegation system. + + Args: + hotkey_ss58 (str): The SS58 address of the neuron's hotkey. + block (Optional[int]): The blockchain block number for the query. + + Returns: + bool: ``True`` if the hotkey is a delegate, ``False`` otherwise. + + Being a delegate is a significant status within the Bittensor network, indicating a neuron's involvement in consensus and governance processes. + """ + return hotkey_ss58 in [ + info.hotkey_ss58 for info in self.get_delegates(block=block) + ] + + # Extrinsics ======================================================================================================= + + def set_weights( + self, + wallet: "Wallet", + netuid: int, + uids: Union[NDArray[np.int64], "torch.LongTensor", list], + weights: Union[NDArray[np.float32], "torch.FloatTensor", list], + version_key: int = settings.version_as_int, + wait_for_inclusion: bool = False, + wait_for_finalization: bool = False, + max_retries: int = 5, + ) -> tuple[bool, str]: + """ + Sets the inter-neuronal weights for the specified neuron. This process involves specifying the influence or trust a neuron places on other neurons in the network, which is a fundamental aspect of Bittensor's decentralized learning architecture. + + Args: + wallet (bittensor_wallet.Wallet): The wallet associated with the neuron setting the weights. + netuid (int): The unique identifier of the subnet. + uids (Union[NDArray[np.int64], torch.LongTensor, list]): The list of neuron UIDs that the weights are being set for. + weights (Union[NDArray[np.float32], torch.FloatTensor, list]): The corresponding weights to be set for each UID. + version_key (int): Version key for compatibility with the network. Default is ``int representation of Bittensor version.``. + wait_for_inclusion (bool): Waits for the transaction to be included in a block. Default is ``False``. + wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Default is ``False``. + max_retries (int): The number of maximum attempts to set weights. Default is ``5``. + + Returns: + tuple[bool, str]: ``True`` if the setting of weights is successful, False otherwise. And `msg`, a string value describing the success or potential error. + + This function is crucial in shaping the network's collective intelligence, where each neuron's learning and contribution are influenced by the weights it sets towards others【81†source】. + """ + uid = self.get_uid_for_hotkey_on_subnet(wallet.hotkey.ss58_address, netuid) + retries = 0 + success = False + message = "No attempt made. Perhaps it is too soon to set weights!" + while ( + self.blocks_since_last_update(netuid, uid) > self.weights_rate_limit(netuid) # type: ignore + and retries < max_retries + ): + try: + logging.info( + f"Setting weights for subnet #{netuid}. Attempt {retries + 1} of {max_retries}." + ) + success, message = set_weights_extrinsic( + subtensor=self, + wallet=wallet, + netuid=netuid, + uids=uids, + weights=weights, + version_key=version_key, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + except Exception as e: + logging.error(f"Error setting weights: {e}") + finally: + retries += 1 + + return success, message + + @legacy_torch_api_compat + def root_set_weights( + self, + wallet: "Wallet", + netuids: Union[NDArray[np.int64], "torch.LongTensor", list], + weights: Union[NDArray[np.float32], "torch.FloatTensor", list], + version_key: int = 0, + wait_for_inclusion: bool = False, + wait_for_finalization: bool = False, + ) -> bool: + """ + Sets the weights for neurons on the root network. This action is crucial for defining the influence and interactions of neurons at the root level of the Bittensor network. + + Args: + wallet (bittensor_wallet.Wallet): The wallet associated with the neuron setting the weights. + netuids (Union[NDArray[np.int64], torch.LongTensor, list]): The list of neuron UIDs for which weights are being set. + weights (Union[NDArray[np.float32], torch.FloatTensor, list]): The corresponding weights to be set for each UID. + version_key (int, optional): Version key for compatibility with the network. Default is ``0``. + wait_for_inclusion (bool, optional): Waits for the transaction to be included in a block. Defaults to ``False``. + wait_for_finalization (bool, optional): Waits for the transaction to be finalized on the blockchain. Defaults to ``False``. + + Returns: + bool: ``True`` if the setting of root-level weights is successful, False otherwise. + + This function plays a pivotal role in shaping the root network's collective intelligence and decision-making processes, reflecting the principles of decentralized governance and collaborative learning in Bittensor. + """ + return set_root_weights_extrinsic( + subtensor=self, + wallet=wallet, + netuids=netuids, + weights=weights, + version_key=version_key, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + def register( + self, + wallet: "Wallet", + netuid: int, + wait_for_inclusion: bool = False, + wait_for_finalization: bool = True, + max_allowed_attempts: int = 3, + output_in_place: bool = True, + cuda: bool = False, + dev_id: Union[list[int], int] = 0, + tpb: int = 256, + num_processes: Optional[int] = None, + update_interval: Optional[int] = None, + log_verbose: bool = False, + ) -> bool: + """ + Registers a neuron on the Bittensor network using the provided wallet. + + Registration is a critical step for a neuron to become an active participant in the network, enabling it to stake, set weights, and receive incentives. + + Args: + wallet (bittensor_wallet.Wallet): The wallet associated with the neuron to be registered. + netuid (int): The unique identifier of the subnet. + wait_for_inclusion (bool): Waits for the transaction to be included in a block. Defaults to `False`. + wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Defaults to `True`. + max_allowed_attempts (int): Maximum number of attempts to register the wallet. + output_in_place (bool): If true, prints the progress of the proof of work to the console in-place. Meaning the progress is printed on the same lines. Defaults to `True`. + cuda (bool): If ``true``, the wallet should be registered using CUDA device(s). Defaults to `False`. + dev_id (Union[List[int], int]): The CUDA device id to use, or a list of device ids. Defaults to `0` (zero). + tpb (int): The number of threads per block (CUDA). Default to `256`. + num_processes (Optional[int]): The number of processes to use to register. Default to `None`. + update_interval (Optional[int]): The number of nonces to solve between updates. Default to `None`. + log_verbose (bool): If ``true``, the registration process will log more information. Default to `False`. + + Returns: + bool: ``True`` if the registration is successful, False otherwise. + + This function facilitates the entry of new neurons into the network, supporting the decentralized + growth and scalability of the Bittensor ecosystem. + """ + return register_extrinsic( + subtensor=self, + wallet=wallet, + netuid=netuid, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + max_allowed_attempts=max_allowed_attempts, + output_in_place=output_in_place, + cuda=cuda, + dev_id=dev_id, + tpb=tpb, + num_processes=num_processes, + update_interval=update_interval, + log_verbose=log_verbose, + ) + + def root_register( + self, + wallet: "Wallet", + wait_for_inclusion: bool = False, + wait_for_finalization: bool = True, + ) -> bool: + """ + Registers the neuron associated with the wallet on the root network. This process is integral for participating in the highest layer of decision-making and governance within the Bittensor network. + + Args: + wallet (bittensor_wallet.wallet): The wallet associated with the neuron to be registered on the root network. + wait_for_inclusion (bool): Waits for the transaction to be included in a block. Defaults to `False`. + wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Defaults to `True`. + + Returns: + bool: ``True`` if the registration on the root network is successful, False otherwise. + + This function enables neurons to engage in the most critical and influential aspects of the network's governance, signifying a high level of commitment and responsibility in the Bittensor ecosystem. + """ + return root_register_extrinsic( + subtensor=self, + wallet=wallet, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + def burned_register( + self, + wallet: "Wallet", + netuid: int, + wait_for_inclusion: bool = False, + wait_for_finalization: bool = True, + ) -> bool: + """ + Registers a neuron on the Bittensor network by recycling TAO. This method of registration involves recycling TAO tokens, allowing them to be re-mined by performing work on the network. + + Args: + wallet (bittensor_wallet.Wallet): The wallet associated with the neuron to be registered. + netuid (int): The unique identifier of the subnet. + wait_for_inclusion (bool, optional): Waits for the transaction to be included in a block. Defaults to `False`. + wait_for_finalization (bool, optional): Waits for the transaction to be finalized on the blockchain. Defaults to `True`. + + Returns: + bool: ``True`` if the registration is successful, False otherwise. + """ + return burned_register_extrinsic( + subtensor=self, + wallet=wallet, + netuid=netuid, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + def serve_axon( + self, + netuid: int, + axon: "Axon", + wait_for_inclusion: bool = False, + wait_for_finalization: bool = True, + certificate: Optional[Certificate] = None, + ) -> bool: + """ + Registers an ``Axon`` serving endpoint on the Bittensor network for a specific neuron. This function is used to set up the Axon, a key component of a neuron that handles incoming queries and data processing tasks. + + Args: + netuid (int): The unique identifier of the subnetwork. + axon (bittensor.core.axon.Axon): The Axon instance to be registered for serving. + wait_for_inclusion (bool): Waits for the transaction to be included in a block. Default is ``False``. + wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Default is ``True``. + + Returns: + bool: ``True`` if the Axon serve registration is successful, False otherwise. + + By registering an Axon, the neuron becomes an active part of the network's distributed computing infrastructure, contributing to the collective intelligence of Bittensor. + """ + return serve_axon_extrinsic( + self, netuid, axon, wait_for_inclusion, wait_for_finalization, certificate + ) + + _do_serve_axon = do_serve_axon + + def transfer( + self, + wallet: "Wallet", + dest: str, + amount: Union["Balance", float], + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, + ) -> bool: + """ + Executes a transfer of funds from the provided wallet to the specified destination address. This function is used to move TAO tokens within the Bittensor network, facilitating transactions between neurons. + + Args: + wallet (bittensor_wallet.Wallet): The wallet from which funds are being transferred. + dest (str): The destination public key address. + amount (Union[bittensor.utils.balance.Balance, float]): The amount of TAO to be transferred. + wait_for_inclusion (bool): Waits for the transaction to be included in a block. Default is ``True``. + wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Default is ``False``. + + Returns: + transfer_extrinsic (bool): ``True`` if the transfer is successful, False otherwise. + + This function is essential for the fluid movement of tokens in the network, supporting various economic activities such as staking, delegation, and reward distribution. + """ + return transfer_extrinsic( + subtensor=self, + wallet=wallet, + dest=dest, + amount=amount, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + def commit_weights( + self, + wallet: "Wallet", + netuid: int, + salt: list[int], + uids: Union[NDArray[np.int64], list], + weights: Union[NDArray[np.int64], list], + version_key: int = settings.version_as_int, + wait_for_inclusion: bool = False, + wait_for_finalization: bool = False, + max_retries: int = 5, + ) -> tuple[bool, str]: + """ + Commits a hash of the neuron's weights to the Bittensor blockchain using the provided wallet. + This action serves as a commitment or snapshot of the neuron's current weight distribution. + + Args: + wallet (bittensor_wallet.Wallet): The wallet associated with the neuron committing the weights. + netuid (int): The unique identifier of the subnet. + salt (list[int]): list of randomly generated integers as salt to generated weighted hash. + uids (np.ndarray): NumPy array of neuron UIDs for which weights are being committed. + weights (np.ndarray): NumPy array of weight values corresponding to each UID. + version_key (int): Version key for compatibility with the network. Default is ``int representation of Bittensor version.``. + wait_for_inclusion (bool): Waits for the transaction to be included in a block. Default is ``False``. + wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Default is ``False``. + max_retries (int): The number of maximum attempts to commit weights. Default is ``5``. + + Returns: + tuple[bool, str]: ``True`` if the weight commitment is successful, False otherwise. And `msg`, a string value describing the success or potential error. + + This function allows neurons to create a tamper-proof record of their weight distribution at a specific point in time, enhancing transparency and accountability within the Bittensor network. + """ + retries = 0 + success = False + message = "No attempt made. Perhaps it is too soon to commit weights!" + + logging.info( + f"Committing weights with params: netuid={netuid}, uids={uids}, weights={weights}, version_key={version_key}" + ) + + # Generate the hash of the weights + commit_hash = generate_weight_hash( + address=wallet.hotkey.ss58_address, + netuid=netuid, + uids=list(uids), + values=list(weights), + salt=salt, + version_key=version_key, + ) + + logging.info(f"Commit Hash: {commit_hash}") + + while retries < max_retries: + try: + success, message = commit_weights_extrinsic( + subtensor=self, + wallet=wallet, + netuid=netuid, + commit_hash=commit_hash, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + if success: + break + except Exception as e: + logging.error(f"Error committing weights: {e}") + finally: + retries += 1 + + return success, message + + def reveal_weights( + self, + wallet: "Wallet", + netuid: int, + uids: Union[NDArray[np.int64], list], + weights: Union[NDArray[np.int64], list], + salt: Union[NDArray[np.int64], list], + version_key: int = settings.version_as_int, + wait_for_inclusion: bool = False, + wait_for_finalization: bool = False, + max_retries: int = 5, + ) -> tuple[bool, str]: + """ + Reveals the weights for a specific subnet on the Bittensor blockchain using the provided wallet. + This action serves as a revelation of the neuron's previously committed weight distribution. + + Args: + wallet (bittensor_wallet.Wallet): The wallet associated with the neuron revealing the weights. + netuid (int): The unique identifier of the subnet. + uids (np.ndarray): NumPy array of neuron UIDs for which weights are being revealed. + weights (np.ndarray): NumPy array of weight values corresponding to each UID. + salt (np.ndarray): NumPy array of salt values corresponding to the hash function. + version_key (int): Version key for compatibility with the network. Default is ``int representation of Bittensor version``. + wait_for_inclusion (bool): Waits for the transaction to be included in a block. Default is ``False``. + wait_for_finalization (bool): Waits for the transaction to be finalized on the blockchain. Default is ``False``. + max_retries (int): The number of maximum attempts to reveal weights. Default is ``5``. + + Returns: + tuple[bool, str]: ``True`` if the weight revelation is successful, False otherwise. And `msg`, a string value describing the success or potential error. + + This function allows neurons to reveal their previously committed weight distribution, ensuring transparency and accountability within the Bittensor network. + """ + + retries = 0 + success = False + message = "No attempt made. Perhaps it is too soon to reveal weights!" + + while retries < max_retries: + try: + success, message = reveal_weights_extrinsic( + subtensor=self, + wallet=wallet, + netuid=netuid, + uids=list(uids), + weights=list(weights), + salt=list(salt), + version_key=version_key, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + if success: + break + except Exception as e: + logging.error(f"Error revealing weights: {e}") + finally: + retries += 1 + + return success, message + + def add_stake( + self, + wallet: "Wallet", + hotkey_ss58: Optional[str] = None, + amount: Optional[Union["Balance", float]] = None, + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, + ) -> bool: + """ + Adds the specified amount of stake to a neuron identified by the hotkey ``SS58`` address. + Staking is a fundamental process in the Bittensor network that enables neurons to participate actively and earn incentives. + + Args: + wallet (bittensor_wallet.Wallet): The wallet to be used for staking. + hotkey_ss58 (Optional[str]): The ``SS58`` address of the hotkey associated with the neuron. + amount (Union[Balance, float]): The amount of TAO to stake. + 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. + + Returns: + bool: ``True`` if the staking is successful, False otherwise. + + This function enables neurons to increase their stake in the network, enhancing their influence and potential rewards in line with Bittensor's consensus and reward mechanisms. + """ + return add_stake_extrinsic( + subtensor=self, + wallet=wallet, + hotkey_ss58=hotkey_ss58, + amount=amount, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + def add_stake_multiple( + self, + wallet: "Wallet", + hotkey_ss58s: list[str], + amounts: Optional[list[Union["Balance", float]]] = None, + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, + ) -> bool: + """ + Adds stakes to multiple neurons identified by their hotkey SS58 addresses. + This bulk operation allows for efficient staking across different neurons from a single wallet. + + Args: + wallet (bittensor_wallet.Wallet): The wallet used for staking. + hotkey_ss58s (list[str]): List of ``SS58`` addresses of hotkeys to stake to. + amounts (list[Union[Balance, float]]): Corresponding amounts of TAO to stake for each hotkey. + 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. + + Returns: + bool: ``True`` if the staking is successful for all specified neurons, False otherwise. + + This function is essential for managing stakes across multiple neurons, reflecting the dynamic and collaborative nature of the Bittensor network. + """ + return add_stake_multiple_extrinsic( + subtensor=self, + wallet=wallet, + hotkey_ss58s=hotkey_ss58s, + amounts=amounts, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + def unstake( + self, + wallet: "Wallet", + hotkey_ss58: Optional[str] = None, + amount: Optional[Union["Balance", float]] = None, + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, + ) -> bool: + """ + Removes a specified amount of stake from a single hotkey account. This function is critical for adjusting individual neuron stakes within the Bittensor network. + + Args: + wallet (bittensor_wallet.wallet): The wallet associated with the neuron from which the stake is being removed. + hotkey_ss58 (Optional[str]): The ``SS58`` address of the hotkey account to unstake from. + amount (Union[Balance, float]): The amount of TAO to unstake. If not specified, unstakes all. + 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. + + Returns: + bool: ``True`` if the unstaking process is successful, False otherwise. + + This function supports flexible stake management, allowing neurons to adjust their network participation and potential reward accruals. + """ + return unstake_extrinsic( + subtensor=self, + wallet=wallet, + hotkey_ss58=hotkey_ss58, + amount=amount, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) + + def unstake_multiple( + self, + wallet: "Wallet", + hotkey_ss58s: list[str], + amounts: Optional[list[Union["Balance", float]]] = None, + wait_for_inclusion: bool = True, + wait_for_finalization: bool = False, + ) -> bool: + """ + Performs batch unstaking from multiple hotkey accounts, allowing a neuron to reduce its staked amounts efficiently. This function is useful for managing the distribution of stakes across multiple neurons. + + Args: + wallet (bittensor_wallet.Wallet): The wallet linked to the coldkey from which the stakes are being withdrawn. + hotkey_ss58s (List[str]): A list of hotkey ``SS58`` addresses to unstake from. + amounts (List[Union[Balance, float]]): The amounts of TAO to unstake from each hotkey. If not provided, unstakes all available stakes. + 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. + + Returns: + bool: ``True`` if the batch unstaking is successful, False otherwise. + + This function allows for strategic reallocation or withdrawal of stakes, aligning with the dynamic stake management aspect of the Bittensor network. + """ + return unstake_multiple_extrinsic( + subtensor=self, + wallet=wallet, + hotkey_ss58s=hotkey_ss58s, + amounts=amounts, + wait_for_inclusion=wait_for_inclusion, + wait_for_finalization=wait_for_finalization, + ) diff --git a/bittensor/utils/btlogging/format.py b/bittensor/utils/btlogging/format.py index 819953425e..1fdcd7764e 100644 --- a/bittensor/utils/btlogging/format.py +++ b/bittensor/utils/btlogging/format.py @@ -56,6 +56,7 @@ def _success(self, message: str, *args, **kws): ":satellite:": "🛰️", ":warning:": "⚠️", ":arrow_right:": "➡️", + ":hourglass:": "⏳", } diff --git a/tests/unit_tests/test_subtensor.py b/tests/unit_tests/test_subtensor.py index b06ec061c0..73c63b30b9 100644 --- a/tests/unit_tests/test_subtensor.py +++ b/tests/unit_tests/test_subtensor.py @@ -2153,3 +2153,641 @@ def test_networks_during_connection(mocker): # Assertions sub.network = network sub.chain_endpoint = settings.NETWORK_MAP.get(network) + + +@pytest.mark.parametrize( + "fake_value_result", + [1, None], + ids=["result has value attr", "result has not value attr"], +) +def test_get_stake_for_coldkey_and_hotkey(subtensor, mocker, fake_value_result): + """Test get_stake_for_coldkey_and_hotkey calls right method with correct arguments.""" + # Preps + fake_hotkey_ss58 = "FAKE_H_SS58" + fake_coldkey_ss58 = "FAKE_C_SS58" + fake_block = 123 + + return_value = ( + mocker.Mock(value=fake_value_result) + if fake_value_result is not None + else fake_value_result + ) + + subtensor.query_subtensor = mocker.patch.object( + subtensor, "query_subtensor", return_value=return_value + ) + spy_balance_from_rao = mocker.spy(subtensor_module.Balance, "from_rao") + + # Call + result = subtensor.get_stake_for_coldkey_and_hotkey( + hotkey_ss58=fake_hotkey_ss58, + coldkey_ss58=fake_coldkey_ss58, + block=fake_block, + ) + + # Asserts + subtensor.query_subtensor.assert_called_once_with( + "Stake", fake_block, [fake_hotkey_ss58, fake_coldkey_ss58] + ) + if fake_value_result is not None: + spy_balance_from_rao.assert_called_once_with(fake_value_result) + else: + spy_balance_from_rao.assert_not_called() + assert result == fake_value_result + + +def test_does_hotkey_exist_true(mocker, subtensor): + """Test when the hotkey exists.""" + # Mock data + fake_hotkey_ss58 = "fake_hotkey" + fake_owner = "valid_owner" + fake_block = 123 + + # Mocks + mock_query_subtensor = mocker.patch.object( + subtensor, + "query_subtensor", + return_value=mocker.Mock(value=fake_owner), + ) + + # Call + result = subtensor.does_hotkey_exist(fake_hotkey_ss58, block=fake_block) + + # Assertions + mock_query_subtensor.assert_called_once_with( + "Owner", fake_block, [fake_hotkey_ss58] + ) + assert result is True + + +def test_does_hotkey_exist_no_value(mocker, subtensor): + """Test when query_subtensor returns no value.""" + # Mock data + fake_hotkey_ss58 = "fake_hotkey" + fake_block = 123 + + # Mocks + mock_query_subtensor = mocker.patch.object( + subtensor, + "query_subtensor", + return_value=None, + ) + + # Call + result = subtensor.does_hotkey_exist(fake_hotkey_ss58, block=fake_block) + + # Assertions + mock_query_subtensor.assert_called_once_with( + "Owner", fake_block, [fake_hotkey_ss58] + ) + assert result is False + + +def test_does_hotkey_exist_special_id(mocker, subtensor): + """Test when query_subtensor returns the special invalid owner identifier.""" + # Mock data + fake_hotkey_ss58 = "fake_hotkey" + fake_owner = "5C4hrfjw9DjXZTzV3MwzrrAr9P1MJhSrvWGWqi1eSuyUpnhM" + fake_block = 123 + + # Mocks + mock_query_subtensor = mocker.patch.object( + subtensor, + "query_subtensor", + return_value=mocker.Mock(value=fake_owner), + ) + + # Call + result = subtensor.does_hotkey_exist(fake_hotkey_ss58, block=fake_block) + + # Assertions + mock_query_subtensor.assert_called_once_with( + "Owner", fake_block, [fake_hotkey_ss58] + ) + assert result is False + + +def test_does_hotkey_exist_latest_block(mocker, subtensor): + """Test when no block is provided (latest block).""" + # Mock data + fake_hotkey_ss58 = "fake_hotkey" + fake_owner = "valid_owner" + + # Mocks + mock_query_subtensor = mocker.patch.object( + subtensor, + "query_subtensor", + return_value=mocker.Mock(value=fake_owner), + ) + + # Call + result = subtensor.does_hotkey_exist(fake_hotkey_ss58) + + # Assertions + mock_query_subtensor.assert_called_once_with("Owner", None, [fake_hotkey_ss58]) + assert result is True + + +def test_get_hotkey_owner_success(mocker, subtensor): + """Test when hotkey exists and owner is found.""" + # Mock data + fake_hotkey_ss58 = "fake_hotkey" + fake_coldkey_ss58 = "fake_coldkey" + fake_block = 123 + + # Mocks + mock_query_subtensor = mocker.patch.object( + subtensor, + "query_subtensor", + return_value=mocker.Mock(value=fake_coldkey_ss58), + ) + mock_does_hotkey_exist = mocker.patch.object( + subtensor, "does_hotkey_exist", return_value=True + ) + + # Call + result = subtensor.get_hotkey_owner(fake_hotkey_ss58, block=fake_block) + + # Assertions + mock_query_subtensor.assert_called_once_with( + "Owner", fake_block, [fake_hotkey_ss58] + ) + mock_does_hotkey_exist.assert_called_once_with(fake_hotkey_ss58, fake_block) + assert result == fake_coldkey_ss58 + + +def test_get_hotkey_owner_no_value(mocker, subtensor): + """Test when query_subtensor returns no value.""" + # Mock data + fake_hotkey_ss58 = "fake_hotkey" + fake_block = 123 + + # Mocks + mock_query_subtensor = mocker.patch.object( + subtensor, + "query_subtensor", + return_value=None, + ) + mock_does_hotkey_exist = mocker.patch.object( + subtensor, "does_hotkey_exist", return_value=True + ) + + # Call + result = subtensor.get_hotkey_owner(fake_hotkey_ss58, block=fake_block) + + # Assertions + mock_query_subtensor.assert_called_once_with( + "Owner", fake_block, [fake_hotkey_ss58] + ) + mock_does_hotkey_exist.assert_not_called() + assert result is None + + +def test_get_hotkey_owner_does_not_exist(mocker, subtensor): + """Test when hotkey does not exist.""" + # Mock data + fake_hotkey_ss58 = "fake_hotkey" + fake_block = 123 + + # Mocks + mock_query_subtensor = mocker.patch.object( + subtensor, + "query_subtensor", + return_value=mocker.Mock(value="fake_coldkey"), + ) + mock_does_hotkey_exist = mocker.patch.object( + subtensor, "does_hotkey_exist", return_value=False + ) + + # Call + result = subtensor.get_hotkey_owner(fake_hotkey_ss58, block=fake_block) + + # Assertions + mock_query_subtensor.assert_called_once_with( + "Owner", fake_block, [fake_hotkey_ss58] + ) + mock_does_hotkey_exist.assert_called_once_with(fake_hotkey_ss58, fake_block) + assert result is None + + +def test_get_hotkey_owner_latest_block(mocker, subtensor): + """Test when no block is provided (latest block).""" + # Mock data + fake_hotkey_ss58 = "fake_hotkey" + fake_coldkey_ss58 = "fake_coldkey" + + # Mocks + mock_query_subtensor = mocker.patch.object( + subtensor, + "query_subtensor", + return_value=mocker.Mock(value=fake_coldkey_ss58), + ) + mock_does_hotkey_exist = mocker.patch.object( + subtensor, "does_hotkey_exist", return_value=True + ) + + # Call + result = subtensor.get_hotkey_owner(fake_hotkey_ss58) + + # Assertions + mock_query_subtensor.assert_called_once_with("Owner", None, [fake_hotkey_ss58]) + mock_does_hotkey_exist.assert_called_once_with(fake_hotkey_ss58, None) + assert result == fake_coldkey_ss58 + + +def test_get_minimum_required_stake_success(mocker, subtensor): + """Test successful call to get_minimum_required_stake.""" + # Mock data + fake_min_stake = "1000000000" # Example value in rao + + # Mocking + mock_query = mocker.patch.object( + subtensor.substrate, + "query", + return_value=mocker.Mock(decode=mocker.Mock(return_value=fake_min_stake)), + ) + mock_balance_from_rao = mocker.patch("bittensor.utils.balance.Balance.from_rao") + + # Call + result = subtensor.get_minimum_required_stake() + + # Assertions + mock_query.assert_called_once_with( + module="SubtensorModule", storage_function="NominatorMinRequiredStake" + ) + mock_balance_from_rao.assert_called_once_with(fake_min_stake) + assert result == mock_balance_from_rao.return_value + + +def test_get_minimum_required_stake_query_failure(mocker, subtensor): + """Test query failure in get_minimum_required_stake.""" + # Mocking + mock_query = mocker.patch.object( + subtensor.substrate, + "query", + side_effect=Exception("Query failed"), + ) + + # Call and Assertions + with pytest.raises(Exception, match="Query failed"): + subtensor.get_minimum_required_stake() + mock_query.assert_called_once_with( + module="SubtensorModule", storage_function="NominatorMinRequiredStake" + ) + + +def test_get_minimum_required_stake_invalid_result(mocker, subtensor): + """Test when the result cannot be decoded.""" + # Mock data + fake_invalid_stake = None # Simulate a failure in decoding + + # Mocking + mock_query = mocker.patch.object( + subtensor.substrate, + "query", + return_value=mocker.Mock(decode=mocker.Mock(return_value=fake_invalid_stake)), + ) + mock_balance_from_rao = mocker.patch("bittensor.utils.balance.Balance.from_rao") + + # Call + result = subtensor.get_minimum_required_stake() + + # Assertions + mock_query.assert_called_once_with( + module="SubtensorModule", storage_function="NominatorMinRequiredStake" + ) + mock_balance_from_rao.assert_called_once_with(fake_invalid_stake) + assert result == mock_balance_from_rao.return_value + + +def test_tx_rate_limit_success(mocker, subtensor): + """Test when tx_rate_limit is successfully retrieved.""" + # Mock data + fake_rate_limit = 100 + fake_block = 123 + + # Mocks + mock_query_subtensor = mocker.patch.object( + subtensor, + "query_subtensor", + return_value=mocker.Mock(value=fake_rate_limit), + ) + + # Call + result = subtensor.tx_rate_limit(block=fake_block) + + # Assertions + mock_query_subtensor.assert_called_once_with("TxRateLimit", fake_block) + assert result == fake_rate_limit + + +def test_tx_rate_limit_no_value(mocker, subtensor): + """Test when query_subtensor returns None.""" + # Mock data + fake_block = 123 + + # Mocks + mock_query_subtensor = mocker.patch.object( + subtensor, + "query_subtensor", + return_value=None, + ) + + # Call + result = subtensor.tx_rate_limit(block=fake_block) + + # Assertions + mock_query_subtensor.assert_called_once_with("TxRateLimit", fake_block) + assert result is None + + +def test_get_delegates_success(mocker, subtensor): + """Test when delegates are successfully retrieved.""" + # Mock data + fake_block = 123 + fake_block_hash = "0xabc123" + fake_json_body = { + "result": "mock_encoded_delegates", + } + + # Mocks + mock_get_block_hash = mocker.patch.object( + subtensor.substrate, + "get_block_hash", + return_value=fake_block_hash, + ) + mock_rpc_request = mocker.patch.object( + subtensor.substrate, + "rpc_request", + return_value=fake_json_body, + ) + mock_list_from_vec_u8 = mocker.patch.object( + subtensor_module.DelegateInfo, + "list_from_vec_u8", + return_value=["delegate1", "delegate2"], + ) + + # Call + result = subtensor.get_delegates(block=fake_block) + + # Assertions + mock_get_block_hash.assert_called_once_with(fake_block) + mock_rpc_request.assert_called_once_with( + method="delegateInfo_getDelegates", + params=[fake_block_hash], + ) + mock_list_from_vec_u8.assert_called_once_with(fake_json_body["result"]) + assert result == ["delegate1", "delegate2"] + + +def test_get_delegates_no_result(mocker, subtensor): + """Test when rpc_request returns no result.""" + # Mock data + fake_block = 123 + fake_block_hash = "0xabc123" + fake_json_body = {} + + # Mocks + mock_get_block_hash = mocker.patch.object( + subtensor.substrate, + "get_block_hash", + return_value=fake_block_hash, + ) + mock_rpc_request = mocker.patch.object( + subtensor.substrate, + "rpc_request", + return_value=fake_json_body, + ) + + # Call + result = subtensor.get_delegates(block=fake_block) + + # Assertions + mock_get_block_hash.assert_called_once_with(fake_block) + mock_rpc_request.assert_called_once_with( + method="delegateInfo_getDelegates", + params=[fake_block_hash], + ) + assert result == [] + + +def test_get_delegates_latest_block(mocker, subtensor): + """Test when no block is provided (latest block).""" + # Mock data + fake_json_body = { + "result": "mock_encoded_delegates", + } + + # Mocks + mock_rpc_request = mocker.patch.object( + subtensor.substrate, + "rpc_request", + return_value=fake_json_body, + ) + mock_list_from_vec_u8 = mocker.patch.object( + subtensor_module.DelegateInfo, + "list_from_vec_u8", + return_value=["delegate1", "delegate2"], + ) + + # Call + result = subtensor.get_delegates() + + # Assertions + mock_rpc_request.assert_called_once_with( + method="delegateInfo_getDelegates", + params=[], + ) + mock_list_from_vec_u8.assert_called_once_with(fake_json_body["result"]) + assert result == ["delegate1", "delegate2"] + + +def test_is_hotkey_delegate_true(mocker, subtensor): + """Test when hotkey is a delegate.""" + # Mock data + fake_hotkey_ss58 = "hotkey_1" + fake_block = 123 + fake_delegates = [ + mocker.Mock(hotkey_ss58="hotkey_1"), + mocker.Mock(hotkey_ss58="hotkey_2"), + ] + + # Mocks + mock_get_delegates = mocker.patch.object( + subtensor, "get_delegates", return_value=fake_delegates + ) + + # Call + result = subtensor.is_hotkey_delegate(fake_hotkey_ss58, block=fake_block) + + # Assertions + mock_get_delegates.assert_called_once_with(block=fake_block) + assert result is True + + +def test_is_hotkey_delegate_false(mocker, subtensor): + """Test when hotkey is not a delegate.""" + # Mock data + fake_hotkey_ss58 = "hotkey_3" + fake_block = 123 + fake_delegates = [ + mocker.Mock(hotkey_ss58="hotkey_1"), + mocker.Mock(hotkey_ss58="hotkey_2"), + ] + + # Mocks + mock_get_delegates = mocker.patch.object( + subtensor, "get_delegates", return_value=fake_delegates + ) + + # Call + result = subtensor.is_hotkey_delegate(fake_hotkey_ss58, block=fake_block) + + # Assertions + mock_get_delegates.assert_called_once_with(block=fake_block) + assert result is False + + +def test_is_hotkey_delegate_empty_list(mocker, subtensor): + """Test when delegate list is empty.""" + # Mock data + fake_hotkey_ss58 = "hotkey_1" + fake_block = 123 + + # Mocks + mock_get_delegates = mocker.patch.object( + subtensor, "get_delegates", return_value=[] + ) + + # Call + result = subtensor.is_hotkey_delegate(fake_hotkey_ss58, block=fake_block) + + # Assertions + mock_get_delegates.assert_called_once_with(block=fake_block) + assert result is False + + +def test_add_stake_success(mocker, subtensor): + """Test add_stake returns True on successful staking.""" + # Prep + fake_wallet = mocker.Mock() + fake_hotkey_ss58 = "fake_hotkey" + fake_amount = 10.0 + + mock_add_stake_extrinsic = mocker.patch.object( + subtensor_module, "add_stake_extrinsic" + ) + + # Call + result = subtensor.add_stake( + wallet=fake_wallet, + hotkey_ss58=fake_hotkey_ss58, + amount=fake_amount, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + + # Assertions + mock_add_stake_extrinsic.assert_called_once_with( + subtensor=subtensor, + wallet=fake_wallet, + hotkey_ss58=fake_hotkey_ss58, + amount=fake_amount, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + assert result == mock_add_stake_extrinsic.return_value + + +def test_add_stake_multiple_success(mocker, subtensor): + """Test add_stake_multiple successfully stakes for all hotkeys.""" + # Prep + fake_wallet = mocker.Mock() + fake_hotkey_ss58 = ["fake_hotkey"] + fake_amount = [10.0] + + mock_add_stake_multiple_extrinsic = mocker.patch.object( + subtensor_module, "add_stake_multiple_extrinsic" + ) + + # Call + result = subtensor.add_stake_multiple( + wallet=fake_wallet, + hotkey_ss58s=fake_hotkey_ss58, + amounts=fake_amount, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + + # Assertions + mock_add_stake_multiple_extrinsic.assert_called_once_with( + subtensor=subtensor, + wallet=fake_wallet, + hotkey_ss58s=fake_hotkey_ss58, + amounts=fake_amount, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + assert result == mock_add_stake_multiple_extrinsic.return_value + + +def test_unstake_success(mocker, subtensor): + """Test unstake operation is successful.""" + # Preps + fake_wallet = mocker.Mock() + fake_hotkey_ss58 = "hotkey_1" + fake_amount = 10.0 + + mock_unstake_extrinsic = mocker.patch.object(subtensor_module, "unstake_extrinsic") + + # Call + result = subtensor.unstake( + wallet=fake_wallet, + hotkey_ss58=fake_hotkey_ss58, + amount=fake_amount, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + + # Assertions + mock_unstake_extrinsic.assert_called_once_with( + subtensor=subtensor, + wallet=fake_wallet, + hotkey_ss58=fake_hotkey_ss58, + amount=fake_amount, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + assert result == mock_unstake_extrinsic.return_value + + +def test_unstake_multiple_success(mocker, subtensor): + """Test unstake_multiple succeeds for all hotkeys.""" + # Preps + fake_wallet = mocker.Mock() + fake_hotkeys = ["hotkey_1", "hotkey_2"] + fake_amounts = [10.0, 20.0] + + mock_unstake_multiple_extrinsic = mocker.patch( + "bittensor.core.subtensor.unstake_multiple_extrinsic", return_value=True + ) + + # Call + result = subtensor.unstake_multiple( + wallet=fake_wallet, + hotkey_ss58s=fake_hotkeys, + amounts=fake_amounts, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + + # Assertions + mock_unstake_multiple_extrinsic.assert_called_once_with( + subtensor=subtensor, + wallet=fake_wallet, + hotkey_ss58s=fake_hotkeys, + amounts=fake_amounts, + wait_for_inclusion=True, + wait_for_finalization=False, + ) + assert result == mock_unstake_multiple_extrinsic.return_value