diff --git a/wallet_data.py b/wallet_data.py new file mode 100644 index 0000000..0d0a5ab --- /dev/null +++ b/wallet_data.py @@ -0,0 +1,664 @@ +#!/usr/bin/env python3 +""" +Hyperliquid Wallet Data Fetcher - Perfect Table Alignment +========================================================== +Complete Python script to pull all available data for a Hyperliquid wallet via API. + +Requirements: + pip install hyperliquid-python-sdk + +Usage: + python hyperliquid_wallet_data.py + +Example: + python hyperliquid_wallet_data.py 0xcd5051944f780a621ee62e39e493c489668acf4d +""" + +import sys +import json +from datetime import datetime, timedelta +from typing import Optional, Dict, Any +from hyperliquid.info import Info +from hyperliquid.utils import constants + + +class HyperliquidWalletAnalyzer: + """ + Comprehensive wallet data analyzer for Hyperliquid exchange. + Fetches all available information about a specific wallet address. + """ + + def __init__(self, wallet_address: str, use_testnet: bool = False): + """ + Initialize the analyzer with a wallet address. + + Args: + wallet_address: Ethereum-style address (0x...) + use_testnet: If True, use testnet instead of mainnet + """ + self.wallet_address = wallet_address + api_url = constants.TESTNET_API_URL if use_testnet else constants.MAINNET_API_URL + + # Initialize Info API (read-only, no private keys needed) + self.info = Info(api_url, skip_ws=True) + print(f"Initialized Hyperliquid API: {'Testnet' if use_testnet else 'Mainnet'}") + print(f"Target wallet: {wallet_address}\n") + + def print_position_details(self, position: Dict[str, Any], index: int): + """ + Print detailed information about a single position. + + Args: + position: Position data dictionary + index: Position number for display + """ + pos = position.get('position', {}) + + # Extract all position details + coin = pos.get('coin', 'Unknown') + size = float(pos.get('szi', 0)) + entry_px = float(pos.get('entryPx', 0)) + position_value = float(pos.get('positionValue', 0)) + unrealized_pnl = float(pos.get('unrealizedPnl', 0)) + return_on_equity = float(pos.get('returnOnEquity', 0)) + + # Leverage details + leverage = pos.get('leverage', {}) + leverage_type = leverage.get('type', 'unknown') if isinstance(leverage, dict) else 'cross' + leverage_value = leverage.get('value', 0) if isinstance(leverage, dict) else 0 + + # Margin and liquidation + margin_used = float(pos.get('marginUsed', 0)) + liquidation_px = pos.get('liquidationPx') + max_trade_szs = pos.get('maxTradeSzs', [0, 0]) + + # Cumulative funding + cumulative_funding = float(pos.get('cumFunding', {}).get('allTime', 0)) + + # Determine if long or short + side = "LONG šŸ“ˆ" if size > 0 else "SHORT šŸ“‰" + side_color = "🟢" if size > 0 else "šŸ”“" + + # PnL color + pnl_symbol = "🟢" if unrealized_pnl >= 0 else "šŸ”“" + pnl_sign = "+" if unrealized_pnl >= 0 else "" + + # ROE color + roe_symbol = "🟢" if return_on_equity >= 0 else "šŸ”“" + roe_sign = "+" if return_on_equity >= 0 else "" + + print(f"\n{'='*80}") + print(f"POSITION #{index}: {coin} {side} {side_color}") + print(f"{'='*80}") + + print(f"\nšŸ“Š POSITION DETAILS:") + print(f" Size: {abs(size):.6f} {coin}") + print(f" Side: {side}") + print(f" Entry Price: ${entry_px:,.4f}") + print(f" Position Value: ${abs(position_value):,.2f}") + + print(f"\nšŸ’° PROFITABILITY:") + print(f" Unrealized PnL: {pnl_symbol} {pnl_sign}${unrealized_pnl:,.2f}") + print(f" Return on Equity: {roe_symbol} {roe_sign}{return_on_equity:.2%}") + print(f" Cumulative Funding: ${cumulative_funding:,.4f}") + + print(f"\nāš™ļø LEVERAGE & MARGIN:") + print(f" Leverage Type: {leverage_type.upper()}") + print(f" Leverage: {leverage_value}x") + print(f" Margin Used: ${margin_used:,.2f}") + + print(f"\nāš ļø RISK MANAGEMENT:") + if liquidation_px: + liquidation_px_float = float(liquidation_px) if liquidation_px else 0 + print(f" Liquidation Price: ${liquidation_px_float:,.4f}") + + # Calculate distance to liquidation + if entry_px > 0 and liquidation_px_float > 0: + if size > 0: # Long position + distance = ((entry_px - liquidation_px_float) / entry_px) * 100 + else: # Short position + distance = ((liquidation_px_float - entry_px) / entry_px) * 100 + + distance_symbol = "🟢" if abs(distance) > 20 else "🟔" if abs(distance) > 10 else "šŸ”“" + print(f" Distance to Liq: {distance_symbol} {abs(distance):.2f}%") + else: + print(f" Liquidation Price: N/A (Cross margin)") + + if max_trade_szs and len(max_trade_szs) == 2: + print(f" Max Long Trade: {max_trade_szs[0]}") + print(f" Max Short Trade: {max_trade_szs[1]}") + + print(f"\n{'='*80}") + + def get_user_state(self) -> Dict[str, Any]: + """ + Get complete user state including positions and margin summary. + + Returns: + Dict containing: + - assetPositions: List of open perpetual positions + - marginSummary: Account value, margin used, withdrawable + - crossMarginSummary: Cross margin details + - withdrawable: Available balance to withdraw + """ + print("šŸ“Š Fetching User State (Perpetuals)...") + try: + data = self.info.user_state(self.wallet_address) + + if data: + margin_summary = data.get('marginSummary', {}) + positions = data.get('assetPositions', []) + + account_value = float(margin_summary.get('accountValue', 0)) + total_margin_used = float(margin_summary.get('totalMarginUsed', 0)) + total_ntl_pos = float(margin_summary.get('totalNtlPos', 0)) + total_raw_usd = float(margin_summary.get('totalRawUsd', 0)) + withdrawable = float(data.get('withdrawable', 0)) + + print(f" āœ“ Account Value: ${account_value:,.2f}") + print(f" āœ“ Total Margin Used: ${total_margin_used:,.2f}") + print(f" āœ“ Total Position Value: ${total_ntl_pos:,.2f}") + print(f" āœ“ Withdrawable: ${withdrawable:,.2f}") + print(f" āœ“ Open Positions: {len(positions)}") + + # Calculate margin utilization + if account_value > 0: + margin_util = (total_margin_used / account_value) * 100 + util_symbol = "🟢" if margin_util < 50 else "🟔" if margin_util < 75 else "šŸ”“" + print(f" āœ“ Margin Utilization: {util_symbol} {margin_util:.2f}%") + + # Print detailed information for each position + if positions: + print(f"\n{'='*80}") + print(f"DETAILED POSITION BREAKDOWN ({len(positions)} positions)") + print(f"{'='*80}") + + for idx, position in enumerate(positions, 1): + self.print_position_details(position, idx) + + # Summary table with perfect alignment + self.print_positions_summary_table(positions) + + else: + print(" ⚠ No perpetual positions found") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return {} + + def print_positions_summary_table(self, positions: list): + """ + Print a summary table of all positions with properly aligned vertical separators. + + Args: + positions: List of position dictionaries + """ + print(f"\n{'='*130}") + print(f"POSITIONS SUMMARY TABLE") + print(f"{'='*130}") + + # Header with vertical separators + print("| Asset | Side | Size | Entry Price | Position Value | Unrealized PnL | ROE | Leverage |") + print("|----------|-----------|------------------|------------------|--------------------|--------------------|------------|------------|") + + total_position_value = 0 + total_pnl = 0 + + for position in positions: + pos = position.get('position', {}) + + coin = pos.get('coin', 'Unknown') + size = float(pos.get('szi', 0)) + entry_px = float(pos.get('entryPx', 0)) + position_value = float(pos.get('positionValue', 0)) + unrealized_pnl = float(pos.get('unrealizedPnl', 0)) + return_on_equity = float(pos.get('returnOnEquity', 0)) + + # Get leverage + leverage = pos.get('leverage', {}) + leverage_value = leverage.get('value', 0) if isinstance(leverage, dict) else 0 + leverage_type = leverage.get('type', 'cross') if isinstance(leverage, dict) else 'cross' + + side_text = "LONG" if size > 0 else "SHORT" + side_emoji = "šŸ“ˆ" if size > 0 else "šŸ“‰" + + # Add color indicators (using text instead of emojis for alignment) + pnl_sign = "+" if unrealized_pnl >= 0 else "" + + # Accumulate totals + total_position_value += abs(position_value) + total_pnl += unrealized_pnl + + # Format numbers with proper width - no emojis in the data + size_str = f"{abs(size):,.4f}" + entry_str = f"${entry_px:,.2f}" + value_str = f"${abs(position_value):,.2f}" + pnl_str = f"{pnl_sign}${unrealized_pnl:,.2f}" + roe_str = f"{return_on_equity:+.2%}" + lev_str = f"{leverage_value}x {leverage_type[:4]}" + + # Use fixed width with ljust/rjust for proper alignment + row = (f"| {coin[:8]:<8} " + f"| {side_text:<5} {side_emoji} " + f"| {size_str:>16} " + f"| {entry_str:>16} " + f"| {value_str:>18} " + f"| {pnl_str:>18} " + f"| {roe_str:>10} " + f"| {lev_str:<10} |") + print(row) + + # Separator before totals + print("|==========|===========|==================|==================|====================|====================|============|============|") + + # Total row + total_value_str = f"${total_position_value:,.2f}" + total_pnl_sign = "+" if total_pnl >= 0 else "" + total_pnl_str = f"{total_pnl_sign}${total_pnl:,.2f}" + + total_row = (f"| {'TOTAL':<8} " + f"| {'':<9} " + f"| {'':<16} " + f"| {'':<16} " + f"| {total_value_str:>18} " + f"| {total_pnl_str:>18} " + f"| {'':<10} " + f"| {'':<10} |") + print(total_row) + print(f"{'='*130}\n") + + def get_spot_state(self) -> Dict[str, Any]: + """ + Get spot trading state including token balances. + + Returns: + Dict containing: + - balances: List of spot token holdings + """ + print("\nšŸ’° Fetching Spot State...") + try: + data = self.info.spot_user_state(self.wallet_address) + + if data and data.get('balances'): + print(f" āœ“ Spot Holdings: {len(data['balances'])} tokens") + for balance in data['balances'][:5]: # Show first 5 + print(f" - {balance.get('coin', 'Unknown')}: {balance.get('total', 0)}") + else: + print(" ⚠ No spot holdings found") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return {} + + def get_open_orders(self) -> list: + """ + Get all open orders for the user. + + Returns: + List of open orders with details (price, size, side, etc.) + """ + print("\nšŸ“‹ Fetching Open Orders...") + try: + data = self.info.open_orders(self.wallet_address) + + if data: + print(f" āœ“ Open Orders: {len(data)}") + for order in data[:3]: # Show first 3 + coin = order.get('coin', 'Unknown') + side = order.get('side', 'Unknown') + size = order.get('sz', 0) + price = order.get('limitPx', 0) + print(f" - {coin} {side}: {size} @ ${price}") + else: + print(" ⚠ No open orders") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return [] + + def get_user_fills(self, limit: int = 100) -> list: + """ + Get recent trade fills (executions). + + Args: + limit: Maximum number of fills to retrieve (max 2000) + + Returns: + List of fills with execution details, PnL, timestamps + """ + print(f"\nšŸ“ˆ Fetching Recent Fills (last {limit})...") + try: + data = self.info.user_fills(self.wallet_address) + + if data: + fills = data[:limit] + print(f" āœ“ Total Fills Retrieved: {len(fills)}") + + # Show summary stats + total_pnl = sum(float(f.get('closedPnl', 0)) for f in fills if f.get('closedPnl')) + print(f" āœ“ Total Closed PnL: ${total_pnl:.2f}") + + # Show most recent + if fills: + recent = fills[0] + print(f" āœ“ Most Recent: {recent.get('coin')} {recent.get('side')} {recent.get('sz')} @ ${recent.get('px')}") + else: + print(" ⚠ No fills found") + + return data[:limit] if data else [] + except Exception as e: + print(f" āœ— Error: {e}") + return [] + + def get_user_fills_by_time(self, start_time: Optional[int] = None, + end_time: Optional[int] = None) -> list: + """ + Get fills within a specific time range. + + Args: + start_time: Start timestamp in milliseconds (default: 7 days ago) + end_time: End timestamp in milliseconds (default: now) + + Returns: + List of fills within the time range + """ + if not start_time: + start_time = int((datetime.now() - timedelta(days=7)).timestamp() * 1000) + if not end_time: + end_time = int(datetime.now().timestamp() * 1000) + + print(f"\nšŸ“… Fetching Fills by Time Range...") + print(f" From: {datetime.fromtimestamp(start_time/1000)}") + print(f" To: {datetime.fromtimestamp(end_time/1000)}") + + try: + data = self.info.user_fills_by_time(self.wallet_address, start_time, end_time) + + if data: + print(f" āœ“ Fills in Range: {len(data)}") + else: + print(" ⚠ No fills in this time range") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return [] + + def get_user_fees(self) -> Dict[str, Any]: + """ + Get user's fee schedule and trading volume. + + Returns: + Dict containing: + - feeSchedule: Fee rates by tier + - userCrossRate: User's current cross trading fee rate + - userAddRate: User's maker fee rate + - userWithdrawRate: Withdrawal fee rate + - dailyUserVlm: Daily trading volume + """ + print("\nšŸ’³ Fetching Fee Information...") + try: + data = self.info.user_fees(self.wallet_address) + + if data: + print(f" āœ“ Maker Fee: {data.get('userAddRate', 0)}%") + print(f" āœ“ Taker Fee: {data.get('userCrossRate', 0)}%") + print(f" āœ“ Daily Volume: ${data.get('dailyUserVlm', [0])[0] if data.get('dailyUserVlm') else 0}") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return {} + + def get_user_rate_limit(self) -> Dict[str, Any]: + """ + Get API rate limit information. + + Returns: + Dict containing: + - cumVlm: Cumulative trading volume + - nRequestsUsed: Number of requests used + - nRequestsCap: Request capacity + """ + print("\nā±ļø Fetching Rate Limit Info...") + try: + data = self.info.user_rate_limit(self.wallet_address) + + if data: + used = data.get('nRequestsUsed', 0) + cap = data.get('nRequestsCap', 0) + print(f" āœ“ API Requests: {used}/{cap}") + print(f" āœ“ Cumulative Volume: ${data.get('cumVlm', 0)}") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return {} + + def get_funding_history(self, coin: str, days: int = 7) -> list: + """ + Get funding rate history for a specific coin. + + Args: + coin: Asset symbol (e.g., 'BTC', 'ETH') + days: Number of days of history (default: 7) + + Returns: + List of funding rate entries + """ + end_time = int(datetime.now().timestamp() * 1000) + start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000) + + print(f"\nšŸ“Š Fetching Funding History for {coin}...") + try: + data = self.info.funding_history(coin, start_time, end_time) + + if data: + print(f" āœ“ Funding Entries: {len(data)}") + if data: + latest = data[-1] + print(f" āœ“ Latest Rate: {latest.get('fundingRate', 0)}") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return [] + + def get_user_funding_history(self, days: int = 7) -> list: + """ + Get user's funding payments history. + + Args: + days: Number of days of history (default: 7) + + Returns: + List of funding payments + """ + end_time = int(datetime.now().timestamp() * 1000) + start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000) + + print(f"\nšŸ’ø Fetching User Funding Payments (last {days} days)...") + try: + data = self.info.user_funding_history(self.wallet_address, start_time, end_time) + + if data: + print(f" āœ“ Funding Payments: {len(data)}") + total_funding = sum(float(f.get('usdc', 0)) for f in data) + print(f" āœ“ Total Funding P&L: ${total_funding:.2f}") + else: + print(" ⚠ No funding payments found") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return [] + + def get_user_non_funding_ledger_updates(self, days: int = 7) -> list: + """ + Get non-funding ledger updates (deposits, withdrawals, liquidations). + + Args: + days: Number of days of history (default: 7) + + Returns: + List of ledger updates + """ + end_time = int(datetime.now().timestamp() * 1000) + start_time = int((datetime.now() - timedelta(days=days)).timestamp() * 1000) + + print(f"\nšŸ“’ Fetching Ledger Updates (last {days} days)...") + try: + data = self.info.user_non_funding_ledger_updates(self.wallet_address, start_time, end_time) + + if data: + print(f" āœ“ Ledger Updates: {len(data)}") + # Categorize updates + deposits = [u for u in data if 'deposit' in str(u.get('delta', {})).lower()] + withdrawals = [u for u in data if 'withdraw' in str(u.get('delta', {})).lower()] + print(f" āœ“ Deposits: {len(deposits)}, Withdrawals: {len(withdrawals)}") + else: + print(" ⚠ No ledger updates found") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return [] + + def get_referral_state(self) -> Dict[str, Any]: + """ + Get referral program state for the user. + + Returns: + Dict with referral status and earnings + """ + print("\nšŸŽ Fetching Referral State...") + try: + data = self.info.query_referral_state(self.wallet_address) + + if data: + print(f" āœ“ Referral Code: {data.get('referralCode', 'N/A')}") + print(f" āœ“ Referees: {len(data.get('referees', []))}") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return {} + + def get_sub_accounts(self) -> list: + """ + Get list of sub-accounts for the user. + + Returns: + List of sub-account addresses + """ + print("\nšŸ‘„ Fetching Sub-Accounts...") + try: + data = self.info.query_sub_accounts(self.wallet_address) + + if data: + print(f" āœ“ Sub-Accounts: {len(data)}") + else: + print(" ⚠ No sub-accounts found") + + return data + except Exception as e: + print(f" āœ— Error: {e}") + return [] + + def fetch_all_data(self, save_to_file: bool = True) -> Dict[str, Any]: + """ + Fetch all available data for the wallet. + + Args: + save_to_file: If True, save results to JSON file + + Returns: + Dict containing all fetched data + """ + print("=" * 80) + print("HYPERLIQUID WALLET DATA FETCHER") + print("=" * 80) + + all_data = { + 'wallet_address': self.wallet_address, + 'timestamp': datetime.now().isoformat(), + 'data': {} + } + + # Fetch all data sections + all_data['data']['user_state'] = self.get_user_state() + all_data['data']['spot_state'] = self.get_spot_state() + all_data['data']['open_orders'] = self.get_open_orders() + all_data['data']['recent_fills'] = self.get_user_fills(limit=50) + all_data['data']['fills_last_7_days'] = self.get_user_fills_by_time() + all_data['data']['user_fees'] = self.get_user_fees() + all_data['data']['rate_limit'] = self.get_user_rate_limit() + all_data['data']['funding_payments'] = self.get_user_funding_history(days=7) + all_data['data']['ledger_updates'] = self.get_user_non_funding_ledger_updates(days=7) + all_data['data']['referral_state'] = self.get_referral_state() + all_data['data']['sub_accounts'] = self.get_sub_accounts() + + # Optional: Fetch funding history for positions + user_state = all_data['data']['user_state'] + if user_state and user_state.get('assetPositions'): + all_data['data']['funding_history'] = {} + for position in user_state['assetPositions'][:3]: # First 3 positions + coin = position.get('position', {}).get('coin') + if coin: + all_data['data']['funding_history'][coin] = self.get_funding_history(coin, days=7) + + print("\n" + "=" * 80) + print("DATA COLLECTION COMPLETE") + print("=" * 80) + + # Save to file + if save_to_file: + filename = f"hyperliquid_wallet_data_{self.wallet_address[:10]}_{datetime.now().strftime('%Y%m%d_%H%M%S')}.json" + with open(filename, 'w') as f: + json.dump(all_data, f, indent=2, default=str) + print(f"\nšŸ’¾ Data saved to: {filename}") + + return all_data + + +def main(): + """Main execution function.""" + if len(sys.argv) < 2: + print("Usage: python hyperliquid_wallet_data.py [--testnet]") + print("\nExample:") + print(" python hyperliquid_wallet_data.py 0xcd5051944f780a621ee62e39e493c489668acf4d") + sys.exit(1) + + wallet_address = sys.argv[1] + use_testnet = '--testnet' in sys.argv + + # Validate wallet address format + if not wallet_address.startswith('0x') or len(wallet_address) != 42: + print("āŒ Error: Invalid wallet address format") + print(" Address must be in format: 0x followed by 40 hexadecimal characters") + sys.exit(1) + + try: + analyzer = HyperliquidWalletAnalyzer(wallet_address, use_testnet=use_testnet) + data = analyzer.fetch_all_data(save_to_file=True) + + print("\nāœ… All data fetched successfully!") + print(f"\nšŸ“Š Summary:") + print(f" - Account Value: ${data['data']['user_state'].get('marginSummary', {}).get('accountValue', 0)}") + print(f" - Open Positions: {len(data['data']['user_state'].get('assetPositions', []))}") + print(f" - Spot Holdings: {len(data['data']['spot_state'].get('balances', []))}") + print(f" - Open Orders: {len(data['data']['open_orders'])}") + print(f" - Recent Fills: {len(data['data']['recent_fills'])}") + + except Exception as e: + print(f"\nāŒ Fatal Error: {e}") + import traceback + traceback.print_exc() + sys.exit(1) + + +if __name__ == "__main__": + main() \ No newline at end of file