Skip to content
Draft
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
74 changes: 74 additions & 0 deletions sdk/python/bittensor/cli/commands/wallet.py
Original file line number Diff line number Diff line change
Expand Up @@ -44,6 +44,7 @@
wallet_balance_rows,
wallet_inspect_data,
wallet_overview_rows,
wallet_registration_rows,
)
from ..prompt import PromptSpec, confirm_wallet, fill_missing, interactive
from ..secrets import copy_secret_to_clipboard, warn_argv_secrets
Expand Down Expand Up @@ -1051,6 +1052,79 @@ def _amount(display: object, tao: float) -> str:
app_ctx.output.detail(None, human_balance_fields(row), json_fields=row)


@app.command("registrations", rich_help_panel=PANEL_INFO)
@with_globals
def wallet_registrations(
ctx: typer.Context,
all_wallets: bool = typer.Option(
False, "--all", "-a", help="Show registrations for every local wallet."
),
netuid: Optional[int] = typer.Option(None, "--netuid", help="Filter to one subnet."),
):
"""Show owned hotkey registrations, UIDs, and neuron metrics.

Unlike ``wallet overview``, this view is based on hotkey ownership and
subnet registration, not stake positions. It therefore includes miners
even when the wallet coldkey has no alpha staked to them.
"""
app_ctx: AppContext = ctx_of(ctx)
targets: list[tuple[str, str]]
if all_wallets:
targets = list_coldkeys(app_ctx.wallet_path)
if not targets:
app_ctx.output.error(f"no wallets found in {app_ctx.wallet_path}")
raise typer.Exit(1)
else:
coldkey = app_ctx.resolve_address("coldkey_ss58", None)
assert coldkey is not None
targets = [(app_ctx.wallet_name, coldkey)]

records = app_ctx.run(lambda client: wallet_registration_rows(client, targets, netuid=netuid))
names = local_address_names(app_ctx.wallet_path)
for record in records:
hotkey = str(record["hotkey"])
record["hotkey_name"] = names.get(hotkey)
app_ctx.output.classify_address(hotkey, "hotkey")
app_ctx.output.classify_address(str(record["coldkey"]), "coldkey")

columns = [
"wallet",
"hotkey",
"netuid",
"uid",
"active",
"incentive",
"dividends",
"emission",
"updated",
"axon",
"hotkey ss58",
]
rows = [
[
record["wallet"],
record["hotkey_name"] or record["hotkey"],
record["netuid"],
record["uid"],
"yes" if record["active"] else "no",
f"{record['incentive']:.4f}",
f"{record['dividends']:.4f}",
record["emission"],
record["updated"],
record["axon"] or "none",
record["hotkey"],
]
for record in records
]
app_ctx.output.columns(
"wallet registrations",
columns,
rows,
records,
right_align={2, 3, 5, 6, 8},
)


@app.command("overview", rich_help_panel=PANEL_INFO)
@with_globals
def wallet_overview(
Expand Down
114 changes: 113 additions & 1 deletion sdk/python/bittensor/cli/helpers.py
Original file line number Diff line number Diff line change
Expand Up @@ -10,7 +10,8 @@
from __future__ import annotations

import asyncio
from typing import Optional
import ipaddress
from typing import Any, Optional

from .. import config as cfg
from .. import wallets
Expand All @@ -19,6 +20,7 @@
from ..balance import Balance
from ..client import Client
from ..reads import StakePosition, StakeValuation
from ..settings import U16_MAX

STAKE_VALUE_BASIS = "spot price; excludes slippage/fees of an actual unstake"

Expand Down Expand Up @@ -411,6 +413,116 @@ async def position_uids(
return {pair: int(value) for pair, value in zip(pairs, values) if value is not None}


def _registration_axon_endpoint(axon: Any) -> Optional[str]:
"""Return the served axon endpoint from a decoded ``AxonInfo`` record."""
if not isinstance(axon, dict):
return None
ip = int(axon.get("ip") or 0)
port = int(axon.get("port") or 0)
if not ip or not port:
return None
if int(axon.get("ip_type") or 4) == 6:
return f"[{ipaddress.IPv6Address(ip)}]:{port}"
return f"{ipaddress.IPv4Address(ip)}:{port}"


async def wallet_registration_rows(
client: Client,
coldkeys: list[tuple[str, str]],
netuid: Optional[int] = None,
) -> list[dict[str, Any]]:
"""Registered owned hotkeys and their neuron metrics at one block.

Registration discovery is deliberately ownership-based rather than
stake-based. A coldkey's miner remains visible even when that coldkey has
no alpha staked to the hotkey, which is the behavior the pre-v11 wallet
overview provided.
"""
if not coldkeys:
return []

view = await client.at()
owned_by_coldkey = await view.query_batch(
st.SubtensorModule.OwnedHotkeys, [[coldkey] for _, coldkey in coldkeys]
)
owned = [
(wallet_name, coldkey, str(hotkey))
for (wallet_name, coldkey), hotkeys in zip(coldkeys, owned_by_coldkey)
for hotkey in hotkeys or []
]
if not owned:
return []

memberships = await asyncio.gather(
*[view.query_map(st.SubtensorModule.IsNetworkMember, [hotkey]) for _, _, hotkey in owned]
)
registered = [
(wallet_name, coldkey, hotkey, int(member_netuid))
for (wallet_name, coldkey, hotkey), rows in zip(owned, memberships)
for member_netuid, is_member in rows
if is_member and (netuid is None or int(member_netuid) == netuid)
]
if not registered:
return []

uids = await view.query_batch(
st.SubtensorModule.Uids,
[[member_netuid, hotkey] for _, _, hotkey, member_netuid in registered],
)
registrations_with_uids = [
(*registration, int(uid)) for registration, uid in zip(registered, uids) if uid is not None
]
neurons = await asyncio.gather(
*[
view.runtime(api.NeuronInfoRuntimeApi.get_neuron_lite, [member_netuid, uid])
for _, _, _, member_netuid, uid in registrations_with_uids
]
)

records: list[dict[str, Any]] = []
for registration, neuron in zip(registrations_with_uids, neurons):
if not neuron:
continue
wallet_name, coldkey, hotkey, member_netuid, uid = registration
stake = view.balance(
sum(int(amount) for _, amount in neuron.get("stake") or []), member_netuid
)
emission = view.balance(int(neuron.get("emission") or 0), member_netuid)
last_update = int(neuron.get("last_update") or 0)
records.append(
{
"wallet": wallet_name,
"coldkey": coldkey,
"hotkey": hotkey,
"netuid": member_netuid,
"uid": uid,
"active": bool(neuron.get("active")),
"stake": str(stake),
"stake_amount": stake.amount,
"rank": int(neuron.get("rank") or 0) / U16_MAX,
"trust": int(neuron.get("trust") or 0) / U16_MAX,
"consensus": int(neuron.get("consensus") or 0) / U16_MAX,
"incentive": int(neuron.get("incentive") or 0) / U16_MAX,
"dividends": int(neuron.get("dividends") or 0) / U16_MAX,
"emission": str(emission),
"emission_amount": emission.amount,
"validator_trust": int(neuron.get("validator_trust") or 0) / U16_MAX,
"validator_permit": bool(neuron.get("validator_permit")),
"last_update": last_update,
"updated": max(0, view.block - last_update),
"axon": _registration_axon_endpoint(neuron.get("axon_info")),
}
)
return sorted(
records,
key=lambda record: (
str(record["wallet"]),
int(record["netuid"]),
int(record["uid"]),
),
)


async def wallet_overview_rows(
client: Client,
coldkeys: list[tuple[str, str]],
Expand Down
67 changes: 67 additions & 0 deletions sdk/python/tests/unit/test_cli.py
Original file line number Diff line number Diff line change
Expand Up @@ -246,6 +246,73 @@ def test_wallet_balance_by_address(self, fake: FakeSubstrate):
assert payload["coldkey"] == BOB
assert payload["free_tao"] == pytest.approx(2.5)

def test_wallet_registrations_includes_unstaked_owned_hotkey(
self, fake: FakeSubstrate, wallet_dir: str
):
coldkey = wallets.list_wallets_detailed(wallet_dir)[0].ss58
fake.seed("SubtensorModule", "OwnedHotkeys", [coldkey], [BOB])
fake.seed_map("SubtensorModule", "IsNetworkMember", [(4, True), (7, False)])
fake.seed("SubtensorModule", "Uids", [4, BOB], 12)
fake.seed_runtime(
"NeuronInfoRuntimeApi",
"get_neuron_lite",
lambda params: (
{
"hotkey": BOB,
"coldkey": coldkey,
"uid": 12,
"netuid": 4,
"active": True,
"axon_info": {"ip": 2_130_706_433, "port": 8091, "ip_type": 4},
"stake": [],
"rank": 16_384,
"emission": 250_000_000,
"incentive": 32_768,
"consensus": 8_192,
"trust": 4_096,
"validator_trust": 2_048,
"dividends": 1_024,
"last_update": 90,
"validator_permit": False,
}
if params == [4, 12]
else None
),
)

result = invoke("--json", "wallet", "registrations", "--all")
assert result.exit_code == 0, result.output
assert json.loads(result.output) == [
{
"wallet": _WALLET_NAME,
"coldkey": coldkey,
"hotkey": BOB,
"netuid": 4,
"uid": 12,
"active": True,
"stake": "0.000000000α₄",
"stake_amount": 0.0,
"rank": pytest.approx(16_384 / 65_535),
"trust": pytest.approx(4_096 / 65_535),
"consensus": pytest.approx(8_192 / 65_535),
"incentive": pytest.approx(32_768 / 65_535),
"dividends": pytest.approx(1_024 / 65_535),
"emission": "0.250000000α₄",
"emission_amount": 0.25,
"validator_trust": pytest.approx(2_048 / 65_535),
"validator_permit": False,
"last_update": 90,
"updated": 10,
"axon": "127.0.0.1:8091",
"hotkey_name": None,
}
]

human = invoke("wallet", "registrations", "--all")
assert human.exit_code == 0, human.output
assert "0.5000" in human.output
assert "127.0.0.1:8091" in human.output


class TestAddressResolution:
@staticmethod
Expand Down
Loading