wallet info
This commit is contained in:
664
wallet_data.py
Normal file
664
wallet_data.py
Normal file
@ -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 <wallet_address>
|
||||
|
||||
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 <wallet_address> [--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()
|
||||
Reference in New Issue
Block a user