#!/usr/bin/env python3 """ Hyperliquid Account Analyzer Analyzes Hyperliquid trading accounts to evaluate: - Profitability and performance metrics - Average trade duration and trading patterns - Risk management quality - Win rates and consistency - Position sizing and leverage usage Usage: python utils/hyperliquid_account_analyzer.py [address1] [address2] ... Or run with the provided top 10 addresses: python utils/hyperliquid_account_analyzer.py --top10 """ import asyncio import aiohttp import json import sys import os from datetime import datetime, timedelta from typing import Dict, List, Optional, Any, Tuple from dataclasses import dataclass import statistics from collections import defaultdict import argparse # Add src to path to import our modules sys.path.append(os.path.join(os.path.dirname(__file__), '..', 'src')) @dataclass class Trade: """Represents a single trade""" timestamp: int coin: str side: str # 'buy' or 'sell' size: float price: float fee: float is_maker: bool @dataclass class Position: """Represents a position""" coin: str size: float side: str # 'long' or 'short' entry_price: float mark_price: float unrealized_pnl: float leverage: float margin_used: float @dataclass class AccountStats: """Comprehensive account statistics""" address: str total_pnl: float win_rate: float total_trades: int avg_trade_duration_hours: float max_drawdown: float sharpe_ratio: float avg_position_size: float max_leverage_used: float avg_leverage_used: float trading_frequency_per_day: float risk_reward_ratio: float consecutive_losses_max: int profit_factor: float largest_win: float largest_loss: float active_positions: int current_drawdown: float last_trade_timestamp: int analysis_period_days: int class HyperliquidAccountAnalyzer: """Analyzes Hyperliquid trading accounts""" def __init__(self): self.info_url = "https://api.hyperliquid.xyz/info" self.session = None async def __aenter__(self): self.session = aiohttp.ClientSession() return self async def __aexit__(self, exc_type, exc_val, exc_tb): if self.session: await self.session.close() async def get_account_state(self, address: str) -> Optional[Dict]: """Get current account state including positions and balance""" try: payload = { "type": "clearinghouseState", "user": address } async with self.session.post(self.info_url, json=payload) as response: if response.status == 200: return await response.json() else: print(f"❌ Error fetching account state for {address}: HTTP {response.status}") return None except Exception as e: print(f"❌ Exception fetching account state for {address}: {e}") return None async def get_user_fills(self, address: str, limit: int = 1000) -> Optional[List[Dict]]: """Get recent fills/trades for a user""" try: payload = { "type": "userFills", "user": address } async with self.session.post(self.info_url, json=payload) as response: if response.status == 200: data = await response.json() # Return only the most recent fills up to limit fills = data if isinstance(data, list) else [] return fills[:limit] else: print(f"❌ Error fetching fills for {address}: HTTP {response.status}") return None except Exception as e: print(f"❌ Exception fetching fills for {address}: {e}") return None async def get_funding_history(self, address: str) -> Optional[List[Dict]]: """Get funding payments history""" try: payload = { "type": "userFunding", "user": address } async with self.session.post(self.info_url, json=payload) as response: if response.status == 200: return await response.json() else: return [] except Exception as e: print(f"⚠️ Could not fetch funding history for {address}: {e}") return [] def parse_trades(self, fills: List[Dict]) -> List[Trade]: """Parse fills into Trade objects""" trades = [] for fill in fills: try: # Parse timestamp timestamp = int(fill.get('time', 0)) if timestamp == 0: continue # Parse trade data coin = fill.get('coin', 'UNKNOWN') side = fill.get('side', 'buy').lower() size = float(fill.get('sz', '0')) price = float(fill.get('px', '0')) fee = float(fill.get('fee', '0')) is_maker = fill.get('liquidation', False) == False # Simplified maker detection if size > 0 and price > 0: trades.append(Trade( timestamp=timestamp, coin=coin, side=side, size=size, price=price, fee=fee, is_maker=is_maker )) except (ValueError, KeyError) as e: print(f"⚠️ Warning: Could not parse fill: {fill} - {e}") continue return trades def parse_positions(self, account_state: Dict) -> List[Position]: """Parse account state into Position objects""" positions = [] if not account_state or 'assetPositions' not in account_state: return positions for asset_pos in account_state['assetPositions']: try: position_data = asset_pos.get('position', {}) coin = position_data.get('coin', 'UNKNOWN') size_str = position_data.get('szi', '0') size = float(size_str) if abs(size) < 1e-6: # Skip dust positions continue side = 'long' if size > 0 else 'short' entry_price = float(position_data.get('entryPx', '0')) mark_price = float(position_data.get('positionValue', '0')) / abs(size) if size != 0 else 0 unrealized_pnl = float(position_data.get('unrealizedPnl', '0')) leverage = float(position_data.get('leverage', {}).get('value', '1')) margin_used = float(position_data.get('marginUsed', '0')) positions.append(Position( coin=coin, size=abs(size), side=side, entry_price=entry_price, mark_price=mark_price, unrealized_pnl=unrealized_pnl, leverage=leverage, margin_used=margin_used )) except (ValueError, KeyError) as e: print(f"⚠️ Warning: Could not parse position: {asset_pos} - {e}") continue return positions def calculate_trade_performance(self, trades: List[Trade]) -> Tuple[float, float, int, int]: """Calculate more accurate trade performance metrics""" if len(trades) < 2: return 0.0, 0.0, 0, 0 # Group trades by coin and track P&L per completed round trip trades_by_coin = defaultdict(list) for trade in sorted(trades, key=lambda x: x.timestamp): trades_by_coin[trade.coin].append(trade) total_realized_pnl = 0.0 winning_trades = 0 losing_trades = 0 total_fees = 0.0 for coin, coin_trades in trades_by_coin.items(): position = 0.0 entry_price = 0.0 entry_cost = 0.0 for trade in coin_trades: total_fees += trade.fee if trade.side == 'buy': if position <= 0: # Opening long or closing short if position < 0: # Closing short position pnl = (entry_price - trade.price) * abs(position) - trade.fee total_realized_pnl += pnl if pnl > 0: winning_trades += 1 else: losing_trades += 1 # Start new long position new_size = trade.size - max(0, -position) if new_size > 0: entry_price = trade.price entry_cost = new_size * trade.price position = new_size else: # Adding to long position entry_cost += trade.size * trade.price position += trade.size entry_price = entry_cost / position elif trade.side == 'sell': if position >= 0: # Closing long or opening short if position > 0: # Closing long position pnl = (trade.price - entry_price) * min(position, trade.size) - trade.fee total_realized_pnl += pnl if pnl > 0: winning_trades += 1 else: losing_trades += 1 # Start new short position new_size = trade.size - max(0, position) if new_size > 0: entry_price = trade.price position = -new_size else: # Adding to short position position -= trade.size entry_price = trade.price # Simplified for shorts win_rate = winning_trades / (winning_trades + losing_trades) if (winning_trades + losing_trades) > 0 else 0 return total_realized_pnl, win_rate, winning_trades, losing_trades async def analyze_account(self, address: str) -> Optional[AccountStats]: """Analyze a single account and return comprehensive statistics""" print(f"\n🔍 Analyzing account: {address}") # Get account data account_state = await self.get_account_state(address) fills = await self.get_user_fills(address, limit=500) # Reduced limit for better analysis if not fills: print(f"❌ No trading data found for {address}") return None # Parse data trades = self.parse_trades(fills) positions = self.parse_positions(account_state) if account_state else [] if not trades: print(f"❌ No valid trades found for {address}") return None print(f"📊 Found {len(trades)} trades, {len(positions)} active positions") # Calculate time period trades_sorted = sorted(trades, key=lambda x: x.timestamp) oldest_trade = trades_sorted[0].timestamp newest_trade = trades_sorted[-1].timestamp analysis_period_ms = newest_trade - oldest_trade analysis_period_days = max(1, analysis_period_ms / (1000 * 60 * 60 * 24)) # Calculate improved metrics total_trades = len(trades) total_fees = sum(trade.fee for trade in trades) # Get better PnL calculation realized_pnl, win_rate, winning_trades, losing_trades = self.calculate_trade_performance(trades) unrealized_pnl = sum(pos.unrealized_pnl for pos in positions) total_pnl = realized_pnl + unrealized_pnl print(f"💰 Realized PnL: ${realized_pnl:.2f}, Unrealized: ${unrealized_pnl:.2f}, Fees: ${total_fees:.2f}") print(f"📈 Wins: {winning_trades}, Losses: {losing_trades}, Win Rate: {win_rate:.1%}") # Calculate trade durations (improved) durations = [] position_tracker = defaultdict(lambda: {'size': 0, 'start_time': 0}) for trade in trades_sorted: coin = trade.coin pos = position_tracker[coin] if trade.side == 'buy': if pos['size'] <= 0 and trade.size > abs(pos['size']): # Opening new long pos['start_time'] = trade.timestamp pos['size'] += trade.size else: # sell if pos['size'] > 0: # Closing long position if trade.size >= pos['size'] and pos['start_time'] > 0: # Fully closing duration_hours = (trade.timestamp - pos['start_time']) / (1000 * 3600) if duration_hours > 0: durations.append(duration_hours) pos['start_time'] = 0 pos['size'] -= trade.size elif pos['size'] <= 0: # Opening short pos['start_time'] = trade.timestamp pos['size'] -= trade.size avg_duration = statistics.mean(durations) if durations else 0 print(f"🕒 Found {len(durations)} completed trades, avg duration: {avg_duration:.1f} hours") # Calculate position size statistics position_sizes = [trade.size * trade.price for trade in trades] avg_position_size = statistics.mean(position_sizes) if position_sizes else 0 # Calculate leverage statistics from current positions leverages = [pos.leverage for pos in positions if pos.leverage > 0] max_leverage = max(leverages) if leverages else 0 avg_leverage = statistics.mean(leverages) if leverages else 1 # Calculate trading frequency trading_freq = total_trades / analysis_period_days if analysis_period_days > 0 else 0 # Simplified drawdown calculation max_drawdown = 0.0 current_drawdown = 0.0 if total_pnl < 0: max_drawdown = abs(total_pnl) / (avg_position_size * 10) if avg_position_size > 0 else 0 current_drawdown = max_drawdown # Risk metrics profit_factor = abs(realized_pnl) / total_fees if total_fees > 0 else 0 return AccountStats( address=address, total_pnl=total_pnl, win_rate=win_rate, total_trades=total_trades, avg_trade_duration_hours=avg_duration, max_drawdown=max_drawdown, sharpe_ratio=0, # Would need returns data avg_position_size=avg_position_size, max_leverage_used=max_leverage, avg_leverage_used=avg_leverage, trading_frequency_per_day=trading_freq, risk_reward_ratio=winning_trades / max(1, losing_trades), consecutive_losses_max=0, # Would need sequence analysis profit_factor=profit_factor, largest_win=0, # Would need individual trade P&L largest_loss=0, # Would need individual trade P&L active_positions=len(positions), current_drawdown=current_drawdown, last_trade_timestamp=newest_trade, analysis_period_days=int(analysis_period_days) ) async def analyze_multiple_accounts(self, addresses: List[str]) -> List[AccountStats]: """Analyze multiple accounts concurrently""" print(f"🚀 Starting analysis of {len(addresses)} accounts...\n") tasks = [self.analyze_account(addr) for addr in addresses] results = await asyncio.gather(*tasks, return_exceptions=True) # Filter out None results and exceptions valid_results = [] for i, result in enumerate(results): if isinstance(result, Exception): print(f"❌ Error analyzing {addresses[i]}: {result}") elif result is not None: valid_results.append(result) return valid_results def print_analysis_results(self, stats_list: List[AccountStats]): """Print comprehensive analysis results""" if not stats_list: print("❌ No valid analysis results to display") return print("\n" + "="*100) print("📊 HYPERLIQUID ACCOUNT ANALYSIS RESULTS") print("="*100) # Sort by a composite score (you can adjust this ranking) def calculate_score(stats: AccountStats) -> float: score = 0 # Profitability (40% weight) if stats.total_pnl > 0: score += 40 # Win rate (20% weight) score += stats.win_rate * 20 # Short duration trades (20% weight) - prefer < 24 hours if stats.avg_trade_duration_hours > 0: duration_score = max(0, 20 - (stats.avg_trade_duration_hours / 24) * 20) score += duration_score # Trading frequency (10% weight) - prefer active traders freq_score = min(10, stats.trading_frequency_per_day * 2) score += freq_score # Low drawdown (10% weight) drawdown_score = max(0, 10 - stats.max_drawdown * 100) score += drawdown_score return score sorted_stats = sorted(stats_list, key=calculate_score, reverse=True) for i, stats in enumerate(sorted_stats, 1): score = calculate_score(stats) print(f"\n{i}. 📋 ACCOUNT: {stats.address}") print(f" 🏆 SCORE: {score:.1f}/100") print(f" 💰 Total PnL: ${stats.total_pnl:.2f}") print(f" 📈 Win Rate: {stats.win_rate:.1%}") print(f" 🕒 Avg Trade Duration: {stats.avg_trade_duration_hours:.1f} hours") print(f" 📉 Max Drawdown: {stats.max_drawdown:.1%}") print(f" 🔄 Trading Frequency: {stats.trading_frequency_per_day:.1f} trades/day") print(f" 💵 Avg Position Size: ${stats.avg_position_size:.2f}") print(f" ⚡ Max Leverage: {stats.max_leverage_used:.1f}x") print(f" 📊 Total Trades: {stats.total_trades}") print(f" 📍 Active Positions: {stats.active_positions}") print(f" 📅 Analysis Period: {stats.analysis_period_days} days") # Evaluation evaluation = [] if stats.total_pnl > 0: evaluation.append("✅ Profitable") else: evaluation.append("❌ Not profitable") if stats.avg_trade_duration_hours < 24: evaluation.append("✅ Short-term trades") else: evaluation.append("⚠️ Longer-term trades") if stats.win_rate > 0.5: evaluation.append("✅ Good win rate") else: evaluation.append("⚠️ Low win rate") if stats.max_drawdown < 0.2: evaluation.append("✅ Good risk management") else: evaluation.append("⚠️ High drawdown risk") print(f" 🎯 Evaluation: {' | '.join(evaluation)}") # Recommendation print("\n" + "="*100) print("🎯 RECOMMENDATION FOR COPY TRADING") print("="*100) if sorted_stats: best_account = sorted_stats[0] best_score = calculate_score(best_account) print(f"\n🏆 TOP RECOMMENDATION: {best_account.address}") print(f" 📊 Overall Score: {best_score:.1f}/100") if best_score >= 70: recommendation = "🟢 HIGHLY RECOMMENDED" elif best_score >= 50: recommendation = "🟡 MODERATELY RECOMMENDED" else: recommendation = "🔴 NOT RECOMMENDED" print(f" {recommendation}") print(f"\n📋 Why this account:") if best_account.total_pnl > 0: print(f" ✅ Profitable: ${best_account.total_pnl:.2f} total PnL") if best_account.avg_trade_duration_hours < 24: print(f" ✅ Short trades: {best_account.avg_trade_duration_hours:.1f} hour average") if best_account.win_rate > 0.5: print(f" ✅ Good performance: {best_account.win_rate:.1%} win rate") if best_account.max_drawdown < 0.2: print(f" ✅ Risk management: {best_account.max_drawdown:.1%} max drawdown") print(f"\n⚙️ Suggested copy trading settings:") print(f" 📊 Portfolio allocation: 5-10% (conservative start)") print(f" ⚡ Max leverage limit: {min(5, best_account.avg_leverage_used):.0f}x") print(f" 💰 Min position size: $25") async def main(): """Main function""" parser = argparse.ArgumentParser(description='Analyze Hyperliquid trading accounts') parser.add_argument('addresses', nargs='*', help='Account addresses to analyze') parser.add_argument('--top10', action='store_true', help='Analyze the provided top 10 accounts') args = parser.parse_args() # Top 10 accounts from the user top10_addresses = [ "0xa10ec245b3483f83e350a9165a52ae23dbab01bc", "0xd11f5de0189d52b3abe6b0960b8377c20988e17e", "0xc62df97dcf96324adf4edd30a4a7bffd5402f4da", "0xa70434af5778038245d53da1b4d360a30307a827", "0x72fad4e75748b65566a3ebb555b6f6ee18ce08d1", "0x0487b5e806ac781508cb3272ebd83ad603ddcc0f", "0x59a15c79a007cd6e9965b949fcf04125c2212524", "0xeaa400abec7c62d315fd760cbba817fa35e4e0e8", "0x3104b7668f9e46fb13ec0b141d2902e144d67efe", "0x74dcdc6df25bd7ba70336632ecd76a053d0f8dd4" ] if args.top10: addresses = top10_addresses elif args.addresses: addresses = args.addresses else: addresses = top10_addresses print("ℹ️ No addresses specified, analyzing top 10 accounts") async with HyperliquidAccountAnalyzer() as analyzer: results = await analyzer.analyze_multiple_accounts(addresses) analyzer.print_analysis_results(results) if __name__ == "__main__": asyncio.run(main())