Skip to content
128 changes: 124 additions & 4 deletions bittensor/core/async_subtensor.py
Original file line number Diff line number Diff line change
Expand Up @@ -96,6 +96,7 @@
claim_root_extrinsic,
root_register_extrinsic,
set_root_claim_type_extrinsic,
set_validator_claim_type_extrinsic,
)
from bittensor.core.extrinsics.asyncex.serving import (
publish_metadata_extrinsic,
Expand Down Expand Up @@ -3824,7 +3825,9 @@ async def get_root_claim_type(
The root claim type controls how dividends from staking to the Root Subnet (subnet 0) are processed when they
are claimed:

- `Swap` (default): Alpha dividends are swapped to TAO at claim time and restaked on the root subnet.
- `Delegated` (default): Delegate the choice to the validator. Stakers with this setting will inherit
the validator's claim type (Swap or Keep) for each subnet.
- `Swap`: Alpha dividends are swapped to TAO at claim time and restaked on the root subnet.
- `Keep`: Alpha dividends remain as Alpha on the originating subnets.

Parameters:
Expand All @@ -3836,12 +3839,14 @@ async def get_root_claim_type(

Returns:

The root claim type as a string, either `Swap` or `Keep`,
The root claim type as a string, either `Swap`, `Keep`, or `Delegated`,
or dict for "KeepSubnets" in format {"KeepSubnets": {"subnets": [1, 2, 3]}}.
If not set, returns `"Delegated"` (the default).

Notes:
- The claim type applies to both automatic and manual root claims; it does not affect the original TAO stake
on subnet 0, only how Alpha dividends are treated.
- Stakers with `Delegated` will inherit the validator's claim type for each subnet when claiming.
- See: <https://docs.learnbittensor.org/staking-and-delegation/root-claims>
- See also: <https://docs.learnbittensor.org/staking-and-delegation/root-claims/managing-root-claims>
"""
Expand All @@ -3853,11 +3858,14 @@ async def get_root_claim_type(
block_hash=block_hash,
reuse_block_hash=reuse_block,
)
# Query returns enum as dict: {"Swap": ()} or {"Keep": ()} or {"KeepSubnets": {"subnets": [1, 2, 3]}}

# Query returns enum as dict:
# - {"Swap": ()}, {"Keep": ()}, {"Delegated": ()}, or
# - {"KeepSubnets": {"subnets": [1, 2, 3]}}
variant_name = next(iter(query.keys()))
variant_value = query[variant_name]

# For simple variants (Swap, Keep), value is empty tuple, return string
# For simple variants (Swap, Keep, Delegated), value is empty tuple, return string.
if not variant_value or variant_value == ():
return variant_name

Expand Down Expand Up @@ -4839,6 +4847,61 @@ async def get_unstake_fee(
)
return sim_swap_result.alpha_fee.set_unit(netuid=netuid)

async def get_validator_claim_type(
self,
hotkey_ss58: str,
netuid: int,
block: Optional[int] = None,
block_hash: Optional[str] = None,
reuse_block: bool = False,
) -> Union[str, dict]:
"""Retrieves the validator claim type for a given hotkey and netuid.

This returns the claim type that validators set for their hotkey on a specific subnet.
Stakers with "Delegated" root claim type will inherit this value when claiming root emissions.

Parameters:
hotkey_ss58: The ss58 address of the hotkey.
netuid: The netuid of the subnet.
block: The block number to query. Do not specify if using block_hash or reuse_block.
block_hash: The block hash at which to check the parameter. Do not set if using block or reuse_block.
reuse_block: Whether to reuse the last-used block hash. Do not set if using block_hash or block.

Returns:
Union[str, dict]: ValidatorClaimType value. Returns string for "Swap" or "Keep",
or dict for "KeepSubnets" in format {"KeepSubnets": {"subnets": [1, 2, 3]}}.
If not set, returns "Keep" (the default).
"""
block_hash = await self.determine_block_hash(block, block_hash, reuse_block)
query = await self.substrate.query(
module="SubtensorModule",
storage_function="ValidatorClaimType",
params=[hotkey_ss58, netuid],
block_hash=block_hash,
reuse_block_hash=reuse_block,
)

# If query returns None or empty, return default "Keep"
if not query:
return "Keep"

# Query returns enum as dict: {"Swap": ()}, {"Keep": ()}, or {"KeepSubnets": {"subnets": [1, 2, 3]}}
variant_name = next(iter(query.keys()))
variant_value = query[variant_name]

# For simple variants (Swap, Keep), value is empty tuple, return string
if not variant_value or variant_value == ():
return variant_name

# For KeepSubnets, value contains the data, return full dict structure
if isinstance(variant_value, dict) and "subnets" in variant_value:
subnets_raw = variant_value["subnets"]
subnets = list(subnets_raw[0])

return {variant_name: {"subnets": subnets}}

return {variant_name: variant_value}

async def get_vote_data(
self,
proposal_hash: str,
Expand Down Expand Up @@ -8343,6 +8406,63 @@ async def set_root_claim_type(
wait_for_revealed_execution=wait_for_revealed_execution,
)

async def set_validator_claim_type(
self,
wallet: "Wallet",
hotkey_ss58: str,
netuid: int,
new_claim_type: "Literal['Swap', 'Keep'] | RootClaimType | dict",
*,
mev_protection: bool = DEFAULT_MEV_PROTECTION,
period: Optional[int] = DEFAULT_PERIOD,
raise_error: bool = False,
wait_for_inclusion: bool = True,
wait_for_finalization: bool = True,
wait_for_revealed_execution: bool = True,
) -> ExtrinsicResponse:
"""Sets the validator claim type for a hotkey on a specific subnet.

This allows validators to set a default claim type that will be inherited by stakers
who have set their root claim type to "Delegated" (the default).

Parameters:
wallet: Bittensor Wallet instance (must own the hotkey).
hotkey_ss58: The SS58 address of the hotkey to set claim type for.
netuid: The netuid of the subnet.
new_claim_type: The new validator claim type. Can be:
- String: "Swap" or "Keep"
- RootClaimType: RootClaimType.Swap, RootClaimType.Keep
- Dict: {"KeepSubnets": {"subnets": [1, 2, 3]}}
- Callable: RootClaimType.KeepSubnets([1, 2, 3])
Note: "Delegated" is not allowed for validators.
mev_protection: If True, encrypts and submits the transaction through the MEV Shield pallet to protect
against front-running and MEV attacks. The transaction remains encrypted in the mempool until validators
decrypt and execute it. If False, submits the transaction directly without encryption.
period: The number of blocks during which the transaction will remain valid after it's submitted. If the
transaction is not included in a block within that number of blocks, it will expire and be rejected. You
can think of it as an expiration date for the transaction.
raise_error: Raises a relevant exception rather than returning `False` if unsuccessful.
wait_for_inclusion: Whether to wait for the inclusion of the transaction.
wait_for_finalization: Whether to wait for the finalization of the transaction.
wait_for_revealed_execution: Whether to wait for the revealed execution of transaction if mev_protection used.

Returns:
ExtrinsicResponse: The result object of the extrinsic execution.
"""
return await set_validator_claim_type_extrinsic(
subtensor=self,
wallet=wallet,
hotkey_ss58=hotkey_ss58,
netuid=netuid,
new_claim_type=new_claim_type,
mev_protection=mev_protection,
period=period,
raise_error=raise_error,
wait_for_inclusion=wait_for_inclusion,
wait_for_finalization=wait_for_finalization,
wait_for_revealed_execution=wait_for_revealed_execution,
)

async def set_subnet_identity(
self,
wallet: "Wallet",
Expand Down
20 changes: 11 additions & 9 deletions bittensor/core/chain_data/root_claim.py
Original file line number Diff line number Diff line change
Expand Up @@ -40,26 +40,28 @@ class RootClaimType(str, Enum):
Enumeration of root claim types in the Bittensor network.

This enum defines how coldkeys manage their root alpha emissions:
- Swap: Swap any alpha emission for TAO
- Keep: Keep all alpha emission
- KeepSubnets: Keep alpha emission for specified subnets, swap everything else
- Delegated: Delegate the choice to the validator (inherit validator's claim type).
- Keep: Keep all alpha emission.
- KeepSubnets: Keep alpha emission for specified subnets, swap everything else.
- Swap: Swap any alpha emission for TAO.

The values match exactly with the RootClaimTypeEnum defined in the Subtensor runtime.
"""

Swap = "Swap"
Delegated = "Delegated"
Keep = "Keep"
KeepSubnets = KeepSubnetsDescriptor
Swap = "Swap"

@classmethod
def normalize(
cls, value: "Literal['Swap', 'Keep'] | RootClaimType | dict"
cls, value: "Literal['Swap', 'Keep', 'Delegated'] | RootClaimType | dict"
) -> str | dict:
"""
Normalizes a root claim type to a format suitable for Substrate calls.

This method handles various input formats:
- String values ("Swap", "Keep") → returns string
- String values ("Swap", "Keep", "Delegated") → returns string
- Enum values (RootClaimType.Swap) → returns string
- Dict values ({"KeepSubnets": {"subnets": [1, 2, 3]}}) → returns dict as-is
- Callable KeepSubnets([1, 2, 3]) → returns dict
Expand All @@ -68,7 +70,7 @@ def normalize(
value: The root claim type in any supported format.

Returns:
Normalized value - string for Swap/Keep or dict for KeepSubnets.
Normalized value - string for Swap/Keep/Delegated or dict for KeepSubnets.

Raises:
ValueError: If the value is not a valid root claim type or KeepSubnets has no subnets.
Expand All @@ -90,7 +92,7 @@ def normalize(

# Handle string values
if isinstance(value, str):
if value in ("Swap", "Keep"):
if value in ("Swap", "Keep", "Delegated"):
return value
elif value == "KeepSubnets":
raise ValueError(
Expand All @@ -99,7 +101,7 @@ def normalize(
else:
raise ValueError(
f"Invalid root claim type: {value}. "
f"Valid types are: 'Swap', 'Keep', or KeepSubnets dict/callable"
f"Valid types are: 'Swap', 'Keep', 'Delegated', or KeepSubnets dict/callable"
)

# Handle dict values (for KeepSubnets)
Expand Down
98 changes: 95 additions & 3 deletions bittensor/core/extrinsics/asyncex/root.py
Original file line number Diff line number Diff line change
Expand Up @@ -169,7 +169,7 @@ async def root_register_extrinsic(
async def set_root_claim_type_extrinsic(
subtensor: "AsyncSubtensor",
wallet: "Wallet",
new_root_claim_type: "Literal['Swap', 'Keep'] | RootClaimType | dict",
new_root_claim_type: "Literal['Swap', 'Keep', 'Delegated'] | RootClaimType | dict",
*,
mev_protection: bool = DEFAULT_MEV_PROTECTION,
period: Optional[int] = None,
Expand All @@ -184,8 +184,8 @@ async def set_root_claim_type_extrinsic(
subtensor: Subtensor instance to interact with the blockchain.
wallet: Bittensor Wallet instance.
new_root_claim_type: The new root claim type to set. Can be:
- String: "Swap" or "Keep"
- RootClaimType: RootClaimType.Swap, RootClaimType.Keep
- String: "Swap", "Keep", "Delegated"
- RootClaimType: RootClaimType.Swap, RootClaimType.Keep, RootClaimType.Delegated
- Dict: {"KeepSubnets": {"subnets": [1, 2, 3]}}
- Callable: RootClaimType.KeepSubnets([1, 2, 3])
mev_protection: If True, encrypts and submits the transaction through the MEV Shield pallet to protect
Expand Down Expand Up @@ -302,3 +302,95 @@ async def claim_root_extrinsic(

except Exception as error:
return ExtrinsicResponse.from_exception(raise_error=raise_error, error=error)


async def set_validator_claim_type_extrinsic(
subtensor: "AsyncSubtensor",
wallet: "Wallet",
hotkey_ss58: str,
netuid: int,
new_claim_type: "Literal['Swap', 'Keep'] | RootClaimType | dict",
*,
mev_protection: bool = DEFAULT_MEV_PROTECTION,
period: Optional[int] = None,
raise_error: bool = False,
wait_for_inclusion: bool = True,
wait_for_finalization: bool = True,
wait_for_revealed_execution: bool = True,
) -> ExtrinsicResponse:
"""Sets the validator claim type for a hotkey on a specific subnet.

This allows validators to set a default claim type that will be inherited by stakers who have set their root claim
type to "Delegated" (the default).

Parameters:
subtensor: AsyncSubtensor instance to interact with the blockchain.
wallet: Bittensor Wallet instance (must own the hotkey).
hotkey_ss58: The SS58 address of the hotkey to set claim type for.
netuid: The netuid of the subnet.
new_claim_type: The new validator claim type. Can be:
- String: "Swap" or "Keep"
- RootClaimType: RootClaimType.Swap, RootClaimType.Keep
- Dict: {"KeepSubnets": {"subnets": [1, 2, 3]}}
- Callable: RootClaimType.KeepSubnets([1, 2, 3])
Note: "Delegated" is not allowed for validators.
mev_protection: If True, encrypts and submits the transaction through the MEV Shield pallet to protect
against front-running and MEV attacks. The transaction remains encrypted in the mempool until validators
decrypt and execute it. If False, submits the transaction directly without encryption.
period: The number of blocks during which the transaction will remain valid after it's submitted. If the
transaction is not included in a block within that number of blocks, it will expire and be rejected. You can
think of it as an expiration date for the transaction.
raise_error: Raises a relevant exception rather than returning `False` if unsuccessful.
wait_for_inclusion: Whether to wait for the inclusion of the transaction.
wait_for_finalization: Whether to wait for the finalization of the transaction.
wait_for_revealed_execution: Whether to wait for the revealed execution of transaction if mev_protection used.

Returns:
ExtrinsicResponse: The result object of the extrinsic execution.

Raises:
ValueError: If new_claim_type is "Delegated" (not allowed for validators).
"""
try:
if not (
unlocked := ExtrinsicResponse.unlock_wallet(wallet, raise_error)
).success:
return unlocked

normalized_type = RootClaimType.normalize(new_claim_type)

# Validators cannot set Delegated claim type
if normalized_type == "Delegated":
raise ValueError(
"Delegated claim type cannot be set for validators. Validators must use Swap, Keep, or KeepSubnets."
)

call = await SubtensorModule(subtensor).set_validator_claim_type(
hotkey=hotkey_ss58,
netuid=netuid,
new_claim_type=normalized_type,
)

if mev_protection:
return await submit_encrypted_extrinsic(
subtensor=subtensor,
wallet=wallet,
call=call,
period=period,
raise_error=raise_error,
wait_for_inclusion=wait_for_inclusion,
wait_for_finalization=wait_for_finalization,
wait_for_revealed_execution=wait_for_revealed_execution,
)
else:
return await subtensor.sign_and_send_extrinsic(
call=call,
wallet=wallet,
period=period,
raise_error=raise_error,
wait_for_inclusion=wait_for_inclusion,
wait_for_finalization=wait_for_finalization,
)

except Exception as error:
return ExtrinsicResponse.from_exception(raise_error=raise_error, error=error)
25 changes: 25 additions & 0 deletions bittensor/core/extrinsics/pallets/subtensor_module.py
Original file line number Diff line number Diff line change
Expand Up @@ -565,6 +565,31 @@ def set_subnet_identity(
additional=additional,
)

def set_validator_claim_type(
self,
hotkey: str,
netuid: int,
new_claim_type: Literal["Swap", "Keep"] | dict,
) -> Call:
"""Returns GenericCall instance for Subtensor function SubtensorModule.set_validator_claim_type.

Parameters:
hotkey: The hotkey SS58 address associated with the validator.
netuid: The netuid of the subnet.
new_claim_type: The new validator claim type. Can be:
- String: "Swap" or "Keep"
- Dict: {"KeepSubnets": {"subnets": [1, 2, 3]}}
Note: "Delegated" is not allowed for validators.

Returns:
GenericCall instance.
"""
return self.create_composed_call(
hotkey=hotkey,
netuid=netuid,
new_claim_type=new_claim_type,
)

def start_call(self, netuid: int) -> Call:
"""Returns GenericCall instance for Subtensor function SubtensorModule.start_call.

Expand Down
Loading
Loading