feat(wallets.py): add wallet history command

bittensor/cli.py

@@ -103,6 +103,7 @@
             "swap_hotkey": SwapHotkeyCommand,
             "set_identity": SetIdentityCommand,
             "get_identity": GetIdentityCommand,
+            "history": GetWalletHistoryCommand,
         },
     },
     "stake": {

bittensor/commands/__init__.py

@@ -87,6 +87,7 @@
     UpdateWalletCommand,
     WalletCreateCommand,
     WalletBalanceCommand,
+    GetWalletHistoryCommand,
 )
 from .transfer import TransferCommand
 from .inspect import InspectCommand

bittensor/commands/wallets.py

@@ -23,6 +23,7 @@
 from rich.table import Table
 from typing import Optional, List
 from . import defaults
+import requests
 
 
 class RegenColdkeyCommand:
@@ -827,3 +828,163 @@ def check_config(config: "bittensor.config"):
                 _,
                 config.subtensor.chain_endpoint,
             ) = bittensor.subtensor.determine_chain_endpoint_and_network(str(network))
+
+
+API_URL = "https://api.subquery.network/sq/TaoStats/bittensor-indexer"
+
+GRAPHQL_QUERY = """
+query ($first: Int!, $after: Cursor, $filter: TransferFilter, $order: [TransfersOrderBy!]!) {
+    transfers(first: $first, after: $after, filter: $filter, orderBy: $order) {
+        nodes {
+            id
+            from
+            to
+            amount
+            extrinsicId
+            blockNumber
+        }
+        pageInfo {
+            endCursor
+            hasNextPage
+            hasPreviousPage
+        }
+        totalCount
+    }
+}
+"""
+
+
+class GetWalletHistoryCommand:
+    """
+    Executes the 'history' command to fetch the latest transfers of the provided wallet on the Bittensor network.
+
+    This command provides a detailed view of the transfers carried out on the wallet.
+
+    Usage:
+    The command lists the latest transfers of the provided wallet, showing the From, To, Amount, Extrinsic Id and Block Number.
+
+    Optional arguments:
+    None. The command uses the wallet and subtensor configurations to fetch latest transfer data associated with a wallet.
+
+    Example usage:
+    >>> btcli wallet history
+
+    Note:
+    This command is essential for users to monitor their financial status on the Bittensor network.
+    It helps in fetching info on all the transfers so that user can easily tally and cross check the transactions.
+    """
+
+    @staticmethod
+    def run(cli):
+        r"""Check the transfer history of the provided wallet."""
+        wallet = bittensor.wallet(config=cli.config)
+        wallet_address = wallet.coldkey.ss58_address
+
+        # Fetch all transfers
+        transfers = get_wallet_transfers(wallet_address)
+
+        # Create output table
+        table = create_transfer_history_table(transfers)
+
+        bittensor.__console__.print(table)
+
+    @staticmethod
+    def add_args(parser: argparse.ArgumentParser):
+        history_parser = parser.add_parser(
+            "history",
+            help="""Fetch transfer history associated with the provided wallet""",
+        )
+        bittensor.wallet.add_args(history_parser)
+        bittensor.subtensor.add_args(history_parser)
+
+    @staticmethod
+    def check_config(config: "bittensor.config"):
+        if not config.is_set("wallet.name") and not config.no_prompt:
+            wallet_name = Prompt.ask("Enter wallet name", default=defaults.wallet.name)
+            config.wallet.name = str(wallet_name)
+
+
+def get_wallet_transfers(wallet_address) -> List[dict]:
+    """Get all transfers associated with the provided wallet address."""
+
+    variables = {
+        "first": 10,
+        "filter": {
+            "or": [
+                {"from": {"equalTo": wallet_address}},
+                {"to": {"equalTo": wallet_address}},
+            ]
+        },
+        "order": "BLOCK_NUMBER_DESC",
+    }
+
+    transfers = []
+
+    # Make the request with the provided query and variables until all transfers are fetched
+    while True:
+        response = requests.post(
+            API_URL, json={"query": GRAPHQL_QUERY, "variables": variables}
+        )
+        data = response.json()
+
+        # Extract nodes and pageInfo from the response
+        transfer_data = data.get("data", {}).get("transfers", {})
+        nodes = transfer_data.get("nodes", [])
+        page_info = transfer_data.get("pageInfo", {})
+
+        # Add the nodes to the result list
+        transfers.extend(nodes)
+
+        # Check if there are more pages
+        if not page_info.get("hasNextPage"):
+            break
+
+        # Update variables for the next request with the endCursor from the current response
+        variables["after"] = page_info["endCursor"]
+
+    return transfers
+
+
+def create_transfer_history_table(transfers):
+    """Get output transfer table"""
+
+    table = Table(show_footer=False)
+    # Define the column names
+    column_names = ["Id", "From", "To", "Amount", "Extrinsic Id", "Block Number"]
+
+    # Create a table
+    table = Table(show_footer=False)
+    table.title = "[white]Wallet Transfers"
+
+    # Define the column styles
+    header_style = "overline white"
+    footer_style = "overline white"
+    column_style = "rgb(50,163,219)"
+    no_wrap = True
+
+    # Add columns to the table
+    for column_name in column_names:
+        table.add_column(
+            f"[white]{column_name}",
+            header_style=header_style,
+            footer_style=footer_style,
+            style=column_style,
+            no_wrap=no_wrap,
+        )
+
+    # Add rows to the table
+    for item in transfers:
+        table.add_row(
+            item["id"],
+            item["from"],
+            item["to"],
+            item["amount"],
+            str(item["extrinsicId"]),
+            item["blockNumber"],
+        )
+    table.add_row()
+    table.show_footer = True
+    table.box = None
+    table.pad_edge = False
+    table.width = None
+    return table