Skip to content
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension


Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
14 changes: 13 additions & 1 deletion CHANGELOG.md
Original file line number Diff line number Diff line change
@@ -1,6 +1,18 @@
# Changelog

## 9.19.0 /2026-03-02
## 9.20.0 /2026-03-24

## What's Changed
* feat: Add batch extrinsic support to reduce transaction fees by @bittoby in https://github.com/opentensor/btcli/pull/863
* Fixes an issue (my fault) where `process_nested` can really be `Any`,… by @thewhaleking in https://github.com/opentensor/btcli/pull/869
* Feat/coldkey swap clear by @ibraheem-abe in https://github.com/opentensor/btcli/pull/871

## New Contributors
* @bittoby made their first contribution in https://github.com/opentensor/btcli/pull/863

**Full Changelog**: https://github.com/opentensor/btcli/compare/v9.19.0...v9.20.0

## 9.19.0 /2026-03-19

## What's Changed
* fix: JSON output empty for `btcli subnets list --json-out` command by @GlobalStar117 in https://github.com/opentensor/btcli/pull/800
Expand Down
48 changes: 39 additions & 9 deletions bittensor_cli/cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -4144,7 +4144,11 @@ def wallet_swap_coldkey(
self,
action: str = typer.Argument(
None,
help="Action to perform: 'announce' to announce intent, 'execute' to complete swap after delay, 'dispute' to freeze the swap.",
help=(
"Action to perform: 'announce' to announce intent, "
"'execute' to complete swap after delay, 'dispute' to freeze the swap, "
"'clear' to withdraw announcement, 'check' to view status."
),
),
wallet_name: Optional[str] = Options.wallet_name,
wallet_path: Optional[str] = Options.wallet_path,
Expand Down Expand Up @@ -4176,6 +4180,9 @@ def wallet_swap_coldkey(
If you suspect compromise, you can [bold]Dispute[/bold] an active announcement to freeze
all activity for the coldkey until the triumvirate can intervene.

If you want to withdraw your announcement, you can [bold]Clear[/bold] (withdraw) an announcement once the
reannouncement delay has elapsed.

EXAMPLES

Step 1 - Announce your intent to swap:
Expand All @@ -4190,9 +4197,13 @@ def wallet_swap_coldkey(

[green]$[/green] btcli wallet swap-coldkey dispute

Check status of pending swaps:
Clear (withdraw) an announcement:

[green]$[/green] btcli wallet swap-coldkey clear

[green]$[/green] btcli wallet swap-check
Check status of your swap:

[green]$[/green] btcli wallet swap-coldkey check
"""
self.verbosity_handler(quiet, verbose, prompt=False, json_output=False)

Expand All @@ -4201,18 +4212,19 @@ def wallet_swap_coldkey(
"\n[bold][blue]Coldkey Swap Actions:[/blue][/bold]\n"
" [dark_sea_green3]announce[/dark_sea_green3] - Start the swap process (pays fee, starts delay timer)\n"
" [dark_sea_green3]execute[/dark_sea_green3] - Complete the swap (after delay period)\n"
" [dark_sea_green3]dispute[/dark_sea_green3] - Freeze the swap process if you suspect compromise\n\n"
" [dim]You can check the current status of your swap with 'btcli wallet swap-check'.[/dim]\n"
" [dark_sea_green3]dispute[/dark_sea_green3] - Freeze the swap process if you suspect compromise\n"
" [dark_sea_green3]clear[/dark_sea_green3] - Withdraw your swap announcement\n"
" [dark_sea_green3]check[/dark_sea_green3] - Check the status of your swap\n\n"
)
action = Prompt.ask(
"Select action",
choices=["announce", "execute", "dispute"],
choices=["announce", "execute", "dispute", "clear", "check"],
default="announce",
)

if action.lower() not in ("announce", "execute", "dispute"):
if action.lower() not in ("announce", "execute", "dispute", "clear", "check"):
print_error(
f"Invalid action: {action}. Must be 'announce', 'execute', or 'dispute'."
f"Invalid action: {action}. Must be 'announce', 'execute', 'dispute', 'clear', or 'check'."
)
raise typer.Exit(1)

Expand All @@ -4233,7 +4245,7 @@ def wallet_swap_coldkey(
)

new_wallet_coldkey_ss58 = None
if action != "dispute":
if action not in ("dispute", "clear", "check"):
if not new_wallet_or_ss58:
new_wallet_or_ss58 = Prompt.ask(
"Enter the [blue]new wallet name[/blue] or [blue]SS58 address[/blue] of the new coldkey",
Expand Down Expand Up @@ -4285,6 +4297,24 @@ def wallet_swap_coldkey(
mev_protection=mev_protection,
)
)
elif action == "clear":
return self._run_command(
wallets.clear_coldkey_swap_announcement(
wallet=wallet,
subtensor=self.initialize_chain(network),
decline=decline,
quiet=quiet,
prompt=prompt,
mev_protection=mev_protection,
)
)
elif action == "check":
return self._run_command(
wallets.check_swap_status(
subtensor=self.initialize_chain(network),
origin_ss58=wallet.coldkeypub.ss58_address,
)
)
else:
return self._run_command(
wallets.execute_coldkey_swap(
Expand Down
6 changes: 3 additions & 3 deletions bittensor_cli/src/bittensor/chain_data.py
Original file line number Diff line number Diff line change
Expand Up @@ -76,9 +76,9 @@ def _chr_str(codes: tuple[int]) -> str:


def process_nested(
data: Sequence[dict[Hashable, tuple[int]]] | dict,
data: Sequence[dict[Hashable, tuple[int]]] | dict | Any,
chr_transform: Callable[[tuple[int]], str],
) -> list[dict[Hashable, str]] | dict[Hashable, str]:
) -> list[dict[Hashable, str]] | dict[Hashable, str] | Any:
"""Processes nested data structures by applying a transformation function to their elements."""
if isinstance(data, Sequence):
if len(data) > 0 and isinstance(data[0], dict):
Expand All @@ -93,7 +93,7 @@ def process_nested(
elif isinstance(data, dict):
return {k: chr_transform(v) for k, v in data.items()}
else:
raise TypeError(f"Unsupported data type {type(data)}")
return data


@dataclass
Expand Down
82 changes: 80 additions & 2 deletions bittensor_cli/src/bittensor/subtensor_interface.py
Original file line number Diff line number Diff line change
Expand Up @@ -1271,8 +1271,8 @@ async def create_signed(call_to_sign, n):
err_msg = format_error_message(e)
if mev_protection and "'result': 'invalid'" in str(e).lower():
err_msg = (
f"MEV Shield extrinsic rejected as invalid. "
f"This usually means the MEV Shield NextKey changed between fetching and submission."
"MEV Shield extrinsic rejected as invalid. "
"This usually means the MEV Shield NextKey changed between fetching and submission."
)
if proxy and "Invalid Transaction" in err_msg:
extrinsic_fee, signer_balance = await asyncio.gather(
Expand All @@ -1289,6 +1289,84 @@ async def create_signed(call_to_sign, n):
)
return False, err_msg, None

async def sign_and_send_batch_extrinsic(
self,
calls: list[GenericCall],
wallet: Wallet,
wait_for_inclusion: bool = True,
wait_for_finalization: bool = False,
era: Optional[dict[str, int]] = None,
proxy: Optional[str] = None,
nonce: Optional[int] = None,
sign_with: Literal["coldkey", "hotkey", "coldkeypub"] = "coldkey",
announce_only: bool = False,
mev_protection: bool = False,
block_hash: Optional[str] = None,
) -> tuple[bool, str, Optional[AsyncExtrinsicReceipt]]:
"""
Wraps multiple extrinsic calls into a single Utility.batch_all transaction
and submits it. This reduces fees by combining N separate transactions into one.

batch_all is atomic: if any call in the batch fails, the entire batch reverts.

For a single call, this delegates directly to sign_and_send_extrinsic without
wrapping, so there's no overhead for non-batch use cases.

:param calls: list of prepared GenericCall objects to batch together.
:param wallet: the wallet whose key will sign the extrinsic.
:param wait_for_inclusion: wait until the extrinsic is included on chain.
:param wait_for_finalization: wait until the extrinsic is finalized on chain.
:param era: validity period in blocks for the transaction.
:param proxy: the real account if using a proxy. None otherwise.
:param nonce: explicit nonce for submission. Fetched automatically if None.
:param sign_with: which wallet keypair signs the extrinsic.
:param announce_only: make the call as a proxy announcement.
:param mev_protection: encrypt the extrinsic via MEV Shield.
:param block_hash: cached block hash for compose_call. Fetched if None.

:return: (success, error message or inner hash, extrinsic receipt | None)
"""
if not calls:
return False, "No calls to batch", None

# No need to wrap a single call in a batch
if len(calls) == 1:
return await self.sign_and_send_extrinsic(
call=calls[0],
wallet=wallet,
wait_for_inclusion=wait_for_inclusion,
wait_for_finalization=wait_for_finalization,
era=era,
proxy=proxy,
nonce=nonce,
sign_with=sign_with,
announce_only=announce_only,
mev_protection=mev_protection,
)

if block_hash is None:
block_hash = await self.substrate.get_chain_head()

batch_call = await self.substrate.compose_call(
call_module="Utility",
call_function="batch_all",
call_params={"calls": calls},
block_hash=block_hash,
)

return await self.sign_and_send_extrinsic(
call=batch_call,
wallet=wallet,
wait_for_inclusion=wait_for_inclusion,
wait_for_finalization=wait_for_finalization,
era=era,
proxy=proxy,
nonce=nonce,
sign_with=sign_with,
announce_only=announce_only,
mev_protection=mev_protection,
)

async def get_children(self, hotkey, netuid) -> tuple[bool, list, str]:
"""
This method retrieves the children of a given hotkey and netuid. It queries the SubtensorModule's ChildKeys
Expand Down
Loading
Loading