hyperliquid_client.py

"""
Hyperliquid API Client
A comprehensive client for interacting with the Hyperliquid DEX API
"""

import asyncio
import json
import time
import hmac
import hashlib
from typing import Dict, List, Optional, Any
from datetime import datetime
import requests
import websockets
from dotenv import load_dotenv
import os

# Load environment variables
load_dotenv()

class HyperliquidClient:
    """Main client for Hyperliquid API interactions"""
    
    def __init__(self, api_key: str = None, secret_key: str = None, base_url: str = None):
        self.api_key = api_key or os.getenv('HYPERLIQUID_API_KEY')
        self.secret_key = secret_key or os.getenv('HYPERLIQUID_SECRET_KEY')
        self.base_url = base_url or os.getenv('HYPERLIQUID_BASE_URL', 'https://api.hyperliquid.xyz')
        
        if not self.api_key or not self.secret_key:
            raise ValueError("API key and secret key must be provided")
    
    def _generate_signature(self, data: str) -> str:
        """Generate HMAC signature for authentication"""
        return hmac.new(
            self.secret_key.encode('utf-8'),
            data.encode('utf-8'),
            hashlib.sha256
        ).hexdigest()
    
    def _make_request(self, endpoint: str, data: Dict[str, Any], method: str = 'POST') -> Dict[str, Any]:
        """Make authenticated request to Hyperliquid API"""
        url = f"{self.base_url}{endpoint}"
        
        headers = {
            'Content-Type': 'application/json'
        }
        
        try:
            if method == 'POST':
                response = requests.post(url, json=data, headers=headers)
            else:
                response = requests.get(url, params=data, headers=headers)
            
            response.raise_for_status()
            return response.json()
        except requests.exceptions.RequestException as e:
            print(f"Request failed: {e}")
            if hasattr(e, 'response') and e.response is not None:
                try:
                    error_detail = e.response.json()
                    return {"error": str(e), "details": error_detail}
                except:
                    return {"error": str(e), "response_text": e.response.text}
            return {"error": str(e)}
    
    # Info Endpoint Methods
    def get_user_state(self) -> Dict[str, Any]:
        """Get user account state"""
        data = {"type": "clearinghouseState", "user": self.api_key}
        return self._make_request("/info", data)
    
    def get_open_orders(self) -> Dict[str, Any]:
        """Get open orders for the user"""
        data = {"type": "openOrders", "user": self.api_key}
        return self._make_request("/info", data)
    
    def get_user_fills(self) -> Dict[str, Any]:
        """Get user's fill history"""
        data = {"type": "userFills", "user": self.api_key}
        return self._make_request("/info", data)
    
    def get_meta(self) -> Dict[str, Any]:
        """Get exchange metadata"""
        data = {"type": "meta"}
        return self._make_request("/info", data)
    
    def get_clearinghouse_state(self, user: str) -> Dict[str, Any]:
        """Get clearinghouse state for a user"""
        data = {"type": "clearinghouseState", "user": user}
        return self._make_request("/info", data)
    
    def get_l2_book(self, coin: str) -> Dict[str, Any]:
        """Get level 2 order book for a coin"""
        data = {"type": "l2Book", "coin": coin}
        return self._make_request("/info", data)
    
    def get_candles(self, coin: str, interval: str, start_time: int, end_time: int) -> Dict[str, Any]:
        """Get historical candle data"""
        data = {
            "type": "candleSnapshot",
            "req": {
                "coin": coin,
                "interval": interval,
                "startTime": start_time,
                "endTime": end_time
            }
        }
        return self._make_request("/info", data)
    
    def get_trades(self, coin: str) -> Dict[str, Any]:
        """Get recent trades for a coin"""
        data = {"type": "recentTrades", "coin": coin}
        return self._make_request("/info", data)
    
    # Exchange Endpoint Methods
    def place_order(self, coin: str, is_buy: bool, sz: float, limit_px: float, 
                   reduce_only: bool = False, post_only: bool = False, leverage: float = 1.0) -> Dict[str, Any]:
        """Place a limit order"""
        try:
            # Import the actual Hyperliquid SDK
            from hyperliquid.exchange import Exchange
            from hyperliquid.utils.constants import MAINNET_API_URL
            from eth_account import Account
            
            # Create account from private key
            account = Account.from_key(self.secret_key)
            
            # Create exchange instance
            exchange_client = Exchange(
                wallet=account,
                base_url=MAINNET_API_URL,
                account_address=self.api_key
            )
            
            # Import order types
            from hyperliquid.utils.signing import OrderType, LimitOrderType
            
            # Get asset metadata to determine correct size precision
            meta = self.get_meta()
            asset_meta = None
            for asset in meta.get('universe', []):
                if asset['name'] == coin:
                    asset_meta = asset
                    break
            
            if asset_meta:
                sz_decimals = asset_meta.get('szDecimals', 4)
                sz_rounded = round(sz, sz_decimals)
            else:
                # Default to 4 decimal places if metadata not found
                sz_rounded = round(sz, 4)
            
            # Set leverage before placing order
            if leverage != 1.0:
                try:
                    leverage_result = exchange_client.update_leverage(coin, leverage)
                    if isinstance(leverage_result, (int, float)):
                        print(f"Leverage set to {leverage}x")
                    elif isinstance(leverage_result, dict) and "error" in leverage_result:
                        print(f"Warning: Could not set leverage: {leverage_result['error']}")
                    else:
                        print(f"Leverage set to {leverage}x")
                except Exception as e:
                    # If the exception is just the leverage value, it's actually success
                    if str(e) == str(leverage):
                        print(f"Leverage set to {leverage}x")
                    else:
                        print(f"Warning: Could not set leverage: {e}")
            
            # Create order type with limit order
            order_type = OrderType["limit"](LimitOrderType(tif="Gtc"))
            
            # Place the order
            result = exchange_client.order(
                name=coin,
                is_buy=is_buy,
                sz=sz_rounded,
                limit_px=limit_px,
                order_type=order_type,
                reduce_only=reduce_only
            )
            
            return {"success": True, "result": result}
            
        except ImportError as e:
            return {"error": f"Import error: {str(e)}"}
        except Exception as e:
            return {"error": f"Trading error: {str(e)}"}
    
    def place_market_order(self, coin: str, is_buy: bool, sz: float, leverage: float = 1.0) -> Dict[str, Any]:
        """Place a market order"""
        try:
            # Import the actual Hyperliquid SDK
            from hyperliquid.exchange import Exchange
            from hyperliquid.utils.constants import MAINNET_API_URL
            from eth_account import Account
            
            # Create account from private key
            account = Account.from_key(self.secret_key)
            
            # Create exchange instance
            exchange_client = Exchange(
                wallet=account,
                base_url=MAINNET_API_URL,
                account_address=self.api_key
            )
            
            # Get current market price for market-like execution
            l2_book = self.get_l2_book(coin)
            if "error" in l2_book:
                return {"error": f"Could not get market price: {l2_book['error']}"}
            
            # Get current price and set aggressive limit price for market-like execution
            if "levels" in l2_book and len(l2_book["levels"]) >= 2:
                levels = l2_book["levels"]
                if levels[0] and levels[1]:
                    bid_price = float(levels[0][0]["px"])
                    ask_price = float(levels[1][0]["px"])
                    
                    if is_buy:
                        # For buy orders, use ask price + small buffer to ensure execution
                        limit_px = ask_price * 1.001  # 0.1% above ask
                    else:
                        # For sell orders, use bid price - small buffer to ensure execution
                        limit_px = bid_price * 0.999  # 0.1% below bid
                else:
                    return {"error": "No price data available"}
            else:
                return {"error": "Invalid price data"}
            
            # Import order types
            from hyperliquid.utils.signing import OrderType, LimitOrderType
            
            # Get asset metadata to determine correct size precision
            meta = self.get_meta()
            asset_meta = None
            for asset in meta.get('universe', []):
                if asset['name'] == coin:
                    asset_meta = asset
                    break
            
            if asset_meta:
                sz_decimals = asset_meta.get('szDecimals', 4)
                sz_rounded = round(sz, sz_decimals)
            else:
                # Default to 4 decimal places if metadata not found
                sz_rounded = round(sz, 4)
            
            # Set leverage before placing order
            if leverage != 1.0:
                try:
                    leverage_result = exchange_client.update_leverage(coin, leverage)
                    if isinstance(leverage_result, (int, float)):
                        print(f"Leverage set to {leverage}x")
                    elif isinstance(leverage_result, dict) and "error" in leverage_result:
                        print(f"Warning: Could not set leverage: {leverage_result['error']}")
                    else:
                        print(f"Leverage set to {leverage}x")
                except Exception as e:
                    # If the exception is just the leverage value, it's actually success
                    if str(e) == str(leverage):
                        print(f"Leverage set to {leverage}x")
                    else:
                        print(f"Warning: Could not set leverage: {e}")
            
            # For market orders, use a limit order with aggressive pricing
            order_type = OrderType["limit"](LimitOrderType(tif="Gtc"))
            
            # Use market_open for market orders
            result = exchange_client.market_open(
                name=coin,
                is_buy=is_buy,
                sz=sz_rounded,
                px=limit_px,
                slippage=0.05  # 5% slippage tolerance
            )
            
            return {"success": True, "result": result}
            
        except ImportError as e:
            return {"error": f"Import error: {str(e)}"}
        except Exception as e:
            return {"error": f"Trading error: {str(e)}"}
    
    def cancel_order(self, oid: int) -> Dict[str, Any]:
        """Cancel an order by order ID"""
        return {"error": "Exchange endpoints require proper signature implementation"}
    
    def cancel_all_orders(self, coin: str = None) -> Dict[str, Any]:
        """Cancel all orders, optionally for a specific coin"""
        return {"error": "Exchange endpoints require proper signature implementation"}
    
    def update_leverage(self, coin: str, leverage: float) -> Dict[str, Any]:
        """Update leverage for a position"""
        try:
            # Import the actual Hyperliquid SDK
            from hyperliquid.exchange import Exchange
            from hyperliquid.utils.constants import MAINNET_API_URL
            from eth_account import Account
            
            # Create account from private key
            account = Account.from_key(self.secret_key)
            
            # Create exchange instance with correct parameters
            exchange_client = Exchange(
                wallet=account,
                base_url=MAINNET_API_URL,
                account_address=self.api_key
            )
            
            # Use the SDK's update_leverage method with correct signature
            result = exchange_client.update_leverage(coin, leverage)
            
            # The SDK method should return a response
            return {"success": True, "result": result}
            
        except ImportError as e:
            return {"error": f"Import error: {str(e)}"}
        except Exception as e:
            # If the exception is just the leverage value, it's actually success
            if str(e) == str(leverage):
                return {"success": True, "result": {"leverage": leverage}}
            else:
                return {"error": f"Leverage update error: {str(e)}"}
    
    def update_isolated_margin(self, coin: str, is_isolated: bool, leverage: float) -> Dict[str, Any]:
        """Update isolated margin settings"""
        return {"error": "Exchange endpoints require proper signature implementation"}
    
    # WebSocket Methods
    async def connect_websocket(self, callback):
        """Connect to Hyperliquid WebSocket for real-time updates"""
        ws_url = "wss://api.hyperliquid.xyz/ws"
        
        try:
            async with websockets.connect(ws_url) as websocket:
                pass  # WebSocket connected
                
                # Subscribe to user updates with correct format
                subscribe_msg = {
                    "method": "subscribe",
                    "subscription": {
                        "type": "userUpdates", 
                        "user": self.api_key
                    }
                }
                await websocket.send(json.dumps(subscribe_msg))
                
                async for message in websocket:
                    try:
                        data = json.loads(message)
                        await callback(data)
                    except json.JSONDecodeError as e:
                        continue
                    except Exception as e:
                        continue
                    
        except Exception as e:
            pass  # WebSocket connection failed
    
    # Utility Methods
    def get_account_value(self) -> float:
        """Get total account value in USD"""
        user_state = self.get_user_state()
        if "error" in user_state:
            return 0.0
        
        # Parse the clearinghouse state response
        if "marginSummary" in user_state:
            return float(user_state.get("marginSummary", {}).get("accountValue", 0))
        elif "accountValue" in user_state:
            return float(user_state.get("accountValue", 0))
        else:
            return 0.0
    
    def get_position(self, coin: str) -> Dict[str, Any]:
        """Get position for a specific coin"""
        user_state = self.get_user_state()
        if "error" in user_state:
            return {}
        
        # Parse positions from clearinghouse state
        positions = user_state.get("assetPositions", [])
        for position in positions:
            if position.get("position", {}).get("coin") == coin:
                return position
        return {}
    
    def get_available_balance(self) -> float:
        """Get available balance for trading"""
        user_state = self.get_user_state()
        if "error" in user_state:
            return 0.0
        
        # Parse available balance from clearinghouse state
        if "marginSummary" in user_state:
            return float(user_state.get("marginSummary", {}).get("totalMarginUsed", 0))
        else:
            return 0.0