ManualTrader/utils/hyperliquid_account_analyzer.py

#!/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())