652 lines
26 KiB
Python
652 lines
26 KiB
Python
#!/usr/bin/env python3
|
|
"""
|
|
Hyperliquid Wallet Data Fetcher - FINAL Perfect 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 perfectly aligned columns.
|
|
NO emojis in data cells - keeps them simple text only for perfect alignment.
|
|
|
|
Args:
|
|
positions: List of position dictionaries
|
|
"""
|
|
print(f"\n{'='*130}")
|
|
print("POSITIONS SUMMARY TABLE")
|
|
print('='*130)
|
|
|
|
# Print header
|
|
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'
|
|
|
|
# Determine side - NO EMOJIS in data
|
|
side_text = "LONG" if size > 0 else "SHORT"
|
|
|
|
# Format PnL and ROE with signs
|
|
pnl_sign = "+" if unrealized_pnl >= 0 else ""
|
|
roe_sign = "+" if return_on_equity >= 0 else ""
|
|
|
|
# Accumulate totals
|
|
total_position_value += abs(position_value)
|
|
total_pnl += unrealized_pnl
|
|
|
|
# Format all values as strings with proper width
|
|
asset_str = f"{coin[:8]:<8}"
|
|
side_str = f"{side_text:<5}"
|
|
size_str = f"{abs(size):>17,.4f}"
|
|
entry_str = f"${entry_px:>16,.2f}"
|
|
value_str = f"${abs(position_value):>16,.2f}"
|
|
pnl_str = f"{pnl_sign}${unrealized_pnl:>15,.2f}"
|
|
roe_str = f"{roe_sign}{return_on_equity:>9.2%}"
|
|
lev_str = f"{leverage_value}x {leverage_type[:4]}"
|
|
|
|
# Print row with exact spacing
|
|
print(f"| {asset_str} | {side_str} | {size_str} | {entry_str} | {value_str} | {pnl_str} | {roe_str} | {lev_str:<10} |")
|
|
|
|
# Separator before totals
|
|
print("|==========|=======|===================|===================|===================|===================|============|============|")
|
|
|
|
# Total row
|
|
total_value_str = f"${total_position_value:>16,.2f}"
|
|
total_pnl_sign = "+" if total_pnl >= 0 else ""
|
|
total_pnl_str = f"{total_pnl_sign}${total_pnl:>15,.2f}"
|
|
|
|
print(f"| TOTAL | | | | {total_value_str} | {total_pnl_str} | | |")
|
|
print('='*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() |