Adds move_stake, transfer_stake, swap_stake to sync and async subtensor

Author: ibraheem-abe

bittensor/core/async_subtensor.py

@@ -526,6 +526,268 @@ async def add_stake(
 
     stake = add_stake
 
+    async def transfer_stake(
+        self,
+        wallet: "Wallet",
+        destination_coldkey_ss58: str,
+        hotkey_ss58: str,
+        origin_netuid: int,
+        destination_netuid: int,
+        amount: Union["Balance", float, int],
+        wait_for_inclusion: bool = True,
+        wait_for_finalization: bool = False,
+    ) -> bool:
+        """
+        Transfers stake from one subnet to another. Keeps the same hotkey but destination coldkey is different.
+        Allows moving stake to a different coldkey's control while also having the option to change the subnet.
+
+        Hotkey is the same. Coldkeys are different.
+
+        Args:
+            wallet (bittensor.wallet): The wallet to transfer stake from.
+            destination_coldkey_ss58 (str): The destination coldkey SS58 address. Different from the origin coldkey.
+            hotkey_ss58 (str): The hotkey SS58 address associated with the stake. This is owned by the origin coldkey.
+            origin_netuid (int): The source subnet UID.
+            destination_netuid (int): The destination subnet UID.
+            amount (Union[Balance, float]): Amount to transfer.
+            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:
+            success (bool): True if the extrinsic was included in a block.
+
+        Raises:
+            StakeError: If the transfer fails due to insufficient stake or other reasons.
+        """
+        if isinstance(amount, (float, int)):
+            amount = Balance.from_tao(amount)
+
+        hotkey_owner = await self.get_hotkey_owner(hotkey_ss58)
+        if hotkey_owner != wallet.coldkeypub.ss58_address:
+            logging.error(
+                f":cross_mark: [red]Failed[/red]: Hotkey: {hotkey_ss58} does not belong to the origin coldkey owner: {wallet.coldkeypub.ss58_address}"
+            )
+            return False
+
+        stake_in_origin = await self.get_stake(
+            hotkey_ss58=hotkey_ss58,
+            coldkey_ss58=wallet.coldkeypub.ss58_address,
+            netuid=origin_netuid,
+        )
+        if stake_in_origin < amount:
+            logging.error(
+                f":cross_mark: [red]Failed[/red]: Insufficient stake in origin hotkey: {hotkey_ss58}. Stake: {stake_in_origin}, amount: {amount}"
+            )
+            return False
+
+        call = await self.substrate.compose_call(
+            call_module="SubtensorModule",
+            call_function="transfer_stake",
+            call_params={
+                "destination_coldkey": destination_coldkey_ss58,
+                "hotkey": hotkey_ss58,
+                "origin_netuid": origin_netuid,
+                "destination_netuid": destination_netuid,
+                "alpha_amount": amount.rao,
+            },
+        )
+        next_nonce = await self.substrate.get_account_next_index(
+            wallet.coldkeypub.ss58_address
+        )
+        extrinsic = await self.substrate.create_signed_extrinsic(
+            call=call, keypair=wallet.coldkey, nonce=next_nonce
+        )
+        response = await self.substrate.submit_extrinsic(
+            extrinsic,
+            wait_for_inclusion=wait_for_inclusion,
+            wait_for_finalization=wait_for_finalization,
+        )
+        if not wait_for_finalization and not wait_for_inclusion:
+            return True
+
+        if await response.is_success:
+            return True
+        else:
+            logging.error(
+                f":cross_mark: [red]Failed[/red]: {format_error_message(await response.error_message)}"
+            )
+            return False
+
+    async def swap_stake(
+        self,
+        wallet: "Wallet",
+        hotkey_ss58: str,
+        origin_netuid: int,
+        destination_netuid: int,
+        amount: Union["Balance", float, int],
+        wait_for_inclusion: bool = True,
+        wait_for_finalization: bool = False,
+    ) -> bool:
+        """
+        Moves stake between subnets while keeping the same coldkey-hotkey pair ownership.
+        Like subnet hopping - same owner, same hotkey, just changing which subnet the stake is in.
+
+        Both hotkey and coldkey are the same.
+
+        Args:
+            wallet (bittensor.wallet): The wallet to transfer stake from.
+            hotkey_ss58 (str): The SS58 address of the hotkey whose stake is being swapped.
+            origin_netuid (int): The netuid from which stake is removed.
+            destination_netuid (int): The netuid to which stake is added.
+            amount (Union[Balance, float, int]): The amount to swap.
+            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:
+            success (bool): True if the extrinsic was successful.
+        """
+        if isinstance(amount, (float, int)):
+            amount = Balance.from_tao(amount)
+
+        hotkey_owner = await self.get_hotkey_owner(hotkey_ss58)
+        if hotkey_owner != wallet.coldkeypub.ss58_address:
+            logging.error(
+                f":cross_mark: [red]Failed[/red]: Hotkey: {hotkey_ss58} does not belong to the origin coldkey owner: {wallet.coldkeypub.ss58_address}"
+            )
+            return False
+
+        stake_in_origin = await self.get_stake(
+            hotkey_ss58=hotkey_ss58,
+            coldkey_ss58=wallet.coldkeypub.ss58_address,
+            netuid=origin_netuid,
+        )
+        if stake_in_origin < amount:
+            logging.error(
+                f":cross_mark: [red]Failed[/red]: Insufficient stake in origin hotkey: {hotkey_ss58}. Stake: {stake_in_origin}, amount: {amount}"
+            )
+            return False
+
+        call = await self.substrate.compose_call(
+            call_module="SubtensorModule",
+            call_function="swap_stake",
+            call_params={
+                "hotkey": hotkey_ss58,
+                "origin_netuid": origin_netuid,
+                "destination_netuid": destination_netuid,
+                "alpha_amount": amount.rao,
+            },
+        )
+        next_nonce = await self.substrate.get_account_next_index(
+            wallet.coldkeypub.ss58_address
+        )
+        extrinsic = await self.substrate.create_signed_extrinsic(
+            call=call,
+            keypair=wallet.coldkey,
+            nonce=next_nonce,
+        )
+        response = await self.substrate.submit_extrinsic(
+            extrinsic,
+            wait_for_inclusion=wait_for_inclusion,
+            wait_for_finalization=wait_for_finalization,
+        )
+        if not wait_for_finalization and not wait_for_inclusion:
+            return True
+
+        if await response.is_success:
+            return True
+        else:
+            logging.error(
+                f":cross_mark: [red]Failed[/red]: {format_error_message(await response.error_message)}"
+            )
+            return False
+
+    async def move_stake(
+        self,
+        wallet: "Wallet",
+        origin_hotkey: str,
+        origin_netuid: int,
+        destination_hotkey: str,
+        destination_netuid: int,
+        amount: Union["Balance", float, int],
+        wait_for_inclusion: bool = True,
+        wait_for_finalization: bool = False,
+    ) -> bool:
+        """
+        Moves stake to a different hotkey and/or subnet while keeping the same coldkey owner.
+        Flexible movement allowing changes to both hotkey and subnet under the same coldkey's control.
+
+        Coldkey is the same. Hotkeys can be different.
+
+        Args:
+            wallet (bittensor.wallet): The wallet to transfer stake from.
+            origin_hotkey (str): The SS58 address of the source hotkey.
+            origin_netuid (int): The netuid of the source subnet.
+            destination_hotkey (str): The SS58 address of the destination hotkey.
+            destination_netuid (int): The netuid of the destination subnet.
+            amount (Union[Balance, float, int]): Amount of stake to move.
+            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:
+            bool: True if the stake movement was successful, False otherwise.
+
+        Raises:
+            StakeError: If the movement fails due to insufficient stake or other reasons.
+        """
+        if isinstance(amount, (float, int)):
+            amount = Balance.from_tao(amount)
+
+        origin_owner = await self.get_hotkey_owner(origin_hotkey)
+        if origin_owner != wallet.coldkeypub.ss58_address:
+            logging.error(
+                f":cross_mark: [red]Failed[/red]: Origin hotkey: {origin_hotkey} does not belong to the coldkey owner: {wallet.coldkeypub.ss58_address}"
+            )
+            return False
+
+        stake_in_origin = await self.get_stake(
+            hotkey_ss58=origin_hotkey,
+            coldkey_ss58=wallet.coldkeypub.ss58_address,
+            netuid=origin_netuid,
+        )
+        if stake_in_origin < amount:
+            logging.error(
+                f":cross_mark: [red]Failed[/red]: Insufficient stake in origin hotkey: {origin_hotkey}. Stake: {stake_in_origin}, amount: {amount}"
+            )
+            return False
+
+        call = await self.substrate.compose_call(
+            call_module="SubtensorModule",
+            call_function="move_stake",
+            call_params={
+                "origin_hotkey": origin_hotkey,
+                "origin_netuid": origin_netuid,
+                "destination_hotkey": destination_hotkey,
+                "destination_netuid": destination_netuid,
+                "alpha_amount": amount.rao,
+            },
+        )
+
+        next_nonce = await self.substrate.get_account_next_index(
+            wallet.coldkeypub.ss58_address
+        )
+        extrinsic = await self.substrate.create_signed_extrinsic(
+            call=call,
+            keypair=wallet.coldkey,
+            nonce=next_nonce,
+        )
+
+        response = await self.substrate.submit_extrinsic(
+            extrinsic,
+            wait_for_inclusion=wait_for_inclusion,
+            wait_for_finalization=wait_for_finalization,
+        )
+
+        if not wait_for_finalization and not wait_for_inclusion:
+            return True
+
+        if await response.is_success:
+            return True
+        else:
+            logging.error(
+                f":cross_mark: [red]Failed[/red]: {format_error_message(await response.error_message)}"
+            )
+            return False
+
     async def is_hotkey_registered_any(
         self,
         hotkey_ss58: str,

bittensor/core/extrinsics/transfer.py

@@ -18,6 +18,7 @@
 from typing import Optional, Union, TYPE_CHECKING
 
 from bittensor.core.extrinsics.utils import submit_extrinsic
+from bittensor.core.errors import StakeError
 from bittensor.core.settings import NETWORK_EXPLORER_MAP
 from bittensor.utils import (
     get_explorer_url_for_network,
@@ -197,3 +198,106 @@ def transfer_extrinsic(
         return True
 
     return False
+
+
+def transfer_stake_extrinsic(
+    subtensor: "Subtensor",
+    wallet: "Wallet",
+    hotkey_ss58: str,
+    amount: Optional[Union[Balance, float, int]],
+    origin_netuid: int,
+    destination_netuid: int,
+    destination_coldkey_ss58: str,
+    wait_for_inclusion: bool = True,
+    wait_for_finalization: bool = False,
+) -> bool:
+    """Transfers stake from one network to another.
+
+    Args:
+        subtensor (bittensor.core.subtensor.Subtensor): Subtensor instance.
+        wallet (Wallet): Bittensor wallet object.
+        hotkey_ss58 (str): The ``ss58`` address of the hotkey account to transfer stake from.
+        amount (Union[Balance, float, int]): Amount to transfer as Bittensor balance, float or int.
+        origin_netuid (int): The netuid to transfer stake from.
+        destination_netuid (int): The netuid to transfer stake to.
+        destination_coldkey_ss58 (str): The destination coldkey to transfer stake to.
+        wait_for_inclusion (bool): If set, waits for the extrinsic to enter a block before returning.
+        wait_for_finalization (bool): If set, waits for the extrinsic to be finalized before returning.
+
+    Returns:
+        success (bool): True if the transfer was successful.
+    """
+    # Decrypt keys
+    if not (unlock := unlock_key(wallet)).success:
+        logging.error(unlock.message)
+        return False
+
+    if not isinstance(amount, Balance):
+        amount = Balance.from_tao(amount).set_unit(origin_netuid)
+
+    logging.info(
+        f":satellite: [magenta]Transferring stake on:[/magenta] [blue]{subtensor.network}[/blue] [magenta]...[/magenta]"
+    )
+
+    old_stake = subtensor.get_stake_for_coldkey_and_hotkey(
+        coldkey_ss58=wallet.coldkeypub.ss58_address,
+        hotkey_ss58=hotkey_ss58,
+        netuid=origin_netuid,
+    )
+    if old_stake < amount:
+        logging.error(
+            f":cross_mark: [red]Failed[/red]: Not enough stake on netuid {origin_netuid} to transfer. Stake: {old_stake} < Amount: {amount}"
+        )
+        return False
+    try:
+        logging.info(
+            f":satellite: [magenta]Transferring:[/magenta] [blue]{amount} from netuid: {origin_netuid} to netuid: {destination_netuid}[/blue]"
+        )
+
+        call = subtensor.substrate.compose_call(
+            call_module="SubtensorModule",
+            call_function="transfer_stake",
+            call_params={
+                "destination_coldkey": destination_coldkey_ss58,
+                "hotkey": hotkey_ss58,
+                "origin_netuid": origin_netuid,
+                "destination_netuid": destination_netuid,
+                "alpha_amount": amount.rao,
+            },
+        )
+        extrinsic = subtensor.substrate.create_signed_extrinsic(
+            call=call, keypair=wallet.coldkey
+        )
+        response = subtensor.substrate.submit_extrinsic(
+            extrinsic,
+            wait_for_inclusion=wait_for_inclusion,
+            wait_for_finalization=wait_for_finalization,
+        )
+
+        if not wait_for_finalization and not wait_for_inclusion:
+            return True
+
+        response.process_events()
+        if response.is_success:
+            logging.success(":white_heavy_check_mark: [green]Finalized[/green]")
+
+            # Get new stake
+            new_stake = subtensor.get_stake_for_coldkey_and_hotkey(
+                coldkey_ss58=wallet.coldkeypub.ss58_address,
+                hotkey_ss58=hotkey_ss58,
+                netuid=origin_netuid,
+            )
+
+            logging.info(
+                f"Origin Stake: [blue]{old_stake}[/blue] :arrow_right: [green]{new_stake}[/green]"
+            )
+            return True
+        else:
+            logging.error(
+                f":cross_mark: [red]Failed[/red]: {format_error_message(response.error_message)}"
+            )
+            return False
+
+    except StakeError as e:
+        logging.error(f":cross_mark: [red]Transfer Stake Error: {e}[/red]")
+        return False