#!/usr/bin/env python3 """ Client Hyperliquid pour récupérer les balances et positions réelles Version améliorée avec patterns éprouvés de l'archive """ import requests import json import time import logging from typing import Dict, List, Optional from hyperliquid_config import MASTER_ACCOUNT_ADDRESS, BASE_URL, DEFAULT_HEADERS # Configuration du logging logger = logging.getLogger(__name__) class HyperliquidClient: """Client Hyperliquid avec patterns éprouvés de l'archive""" def __init__(self, api_wallet_key: str = None, master_account: str = None): """ Initialise le client Hyperliquid Args: api_wallet_key: Clé privée du wallet API (pour signer) master_account: Adresse du compte maître (pour les données) """ self.base_url = BASE_URL self.account_address = master_account or MASTER_ACCOUNT_ADDRESS self.headers = DEFAULT_HEADERS self.api_wallet_key = api_wallet_key # Initialize Info API for read-only operations try: from hyperliquid.info import Info from hyperliquid.utils import constants self.info = Info(constants.MAINNET_API_URL, skip_ws=True) logger.info(f"Client Hyperliquid initialisé pour compte: {self.account_address[:8]}...") except Exception as e: logger.error(f"Erreur initialisation Info API: {e}") self.info = None # Initialize Exchange API only if we have a wallet key self.exchange = None if api_wallet_key: try: from hyperliquid.exchange import Exchange from hyperliquid.utils import constants from hyperliquid_config import API_WALLET_PRIVATE_KEY from eth_account import Account # Use the provided key or fall back to config wallet_key = api_wallet_key or API_WALLET_PRIVATE_KEY # Create LocalAccount from private key wallet_account = Account.from_key(wallet_key) self.exchange = Exchange( wallet_account, constants.MAINNET_API_URL ) logger.info("Exchange API initialisé pour trading live") except Exception as e: logger.error(f"Erreur initialisation Exchange API: {e}") self.exchange = None def _make_request(self, payload: Dict) -> Optional[Dict]: """Effectue une requête vers l'API Hyperliquid""" try: response = requests.post( f"{self.base_url}/info", headers=self.headers, json=payload, timeout=10 ) response.raise_for_status() return response.json() except Exception as e: print(f"Erreur requête Hyperliquid: {e}") return None def get_user_state(self) -> Optional[Dict]: """Récupère l'état du compte utilisateur via l'API Info moderne""" try: if not self.info: logger.error("Info API non disponible") return None user_state = self.info.user_state(self.account_address) if user_state: logger.debug(f"État utilisateur récupéré: {len(user_state)} champs") return user_state except Exception as e: logger.error(f"Erreur récupération état utilisateur: {e}") # Fallback vers l'ancienne méthode payload = { "type": "clearinghouseState", "user": self.account_address } return self._make_request(payload) def get_balances(self) -> Dict[str, float]: """Récupère les balances du compte avec format amélioré""" try: user_state = self.get_user_state() if not user_state: logger.warning("Impossible de récupérer l'état utilisateur") return {} balances = {} # Balance principale (USDC) - Format moderne margin_summary = user_state.get('marginSummary', {}) cross_margin_summary = user_state.get('crossMarginSummary', {}) if margin_summary: account_value = float(margin_summary.get('accountValue', '0')) total_raw_usd = float(margin_summary.get('totalRawUsd', '0')) balances['USDC'] = account_value balances['USDC_available'] = total_raw_usd logger.debug(f"Balance principale: ${account_value:.2f} (disponible: ${total_raw_usd:.2f})") elif cross_margin_summary: account_value = float(cross_margin_summary.get('accountValue', '0')) total_raw_usd = float(cross_margin_summary.get('totalRawUsd', '0')) balances['USDC'] = account_value balances['USDC_available'] = total_raw_usd # Marge utilisée cross_margin_used = user_state.get("crossMaintenanceMarginUsed", "0") balances["USDC_margin_used"] = float(cross_margin_used) # Balances Spot si présentes if "spotUser" in user_state: spot_balances = user_state["spotUser"].get("balances", []) for balance in spot_balances: token = balance.get("coin", "") total = float(balance.get("total", "0")) if total > 0: balances[token] = total logger.debug(f"Balance spot {token}: {total}") return balances except Exception as e: logger.error(f"Erreur récupération balances: {e}") return {} def get_positions(self) -> List[Dict]: """Récupère les positions ouvertes""" user_state = self.get_user_state() if not user_state: return [] positions = [] # Positions Perp asset_positions = user_state.get("assetPositions", []) for pos in asset_positions: position_info = pos.get("position", {}) coin = position_info.get("coin", "") size = float(position_info.get("szi", "0")) if size != 0: # Position ouverte entry_px = float(position_info.get("entryPx", "0")) unrealized_pnl = float(pos.get("unrealizedPnl", "0")) # Examen des données de position complètes logger.debug(f"Position {coin} - Données complètes:") logger.debug(f" position_info: {position_info}") logger.debug(f" pos: {pos}") # Essayer de trouver le leverage dans les données leverage = position_info.get("leverage", None) if not leverage: # Chercher dans d'autres champs possibles leverage = pos.get("leverage", None) positions.append({ "symbol": f"{coin}-PERP", "side": "LONG" if size > 0 else "SHORT", "size": abs(size), "entry_price": entry_px, "unrealized_pnl": unrealized_pnl, "type": "perpetual", "leverage": leverage, # Ajouter le leverage s'il existe "raw_position_data": position_info, # Données brutes pour debug "raw_asset_data": pos # Données complètes pour debug }) return positions def get_meta_info(self) -> Optional[Dict]: """Récupère les informations meta (leverages disponibles, etc.)""" payload = {"type": "meta"} return self._make_request(payload) def get_current_market_price(self, symbol: str) -> Optional[float]: """Récupère le mark price actuel depuis Hyperliquid""" try: clean_symbol = symbol.replace('USDT', '') # Récupérer les données avec mark prices meta_and_ctxs = self.info.meta_and_asset_ctxs() if len(meta_and_ctxs) < 2: return None universe = meta_and_ctxs[0]['universe'] asset_ctxs = meta_and_ctxs[1] # Trouver l'index du symbole symbol_index = None for i, asset in enumerate(universe): if asset['name'] == clean_symbol: symbol_index = i break if symbol_index is not None and symbol_index < len(asset_ctxs): mark_price = float(asset_ctxs[symbol_index]['markPx']) return mark_price return None except Exception as e: logger.error(f"Erreur récupération mark price {symbol}: {e}") return None def get_symbol_decimals(self, symbol: str) -> int: """Récupère le nombre de décimales autorisées pour un symbole (szDecimals)""" meta_info = self.get_meta_info() if not meta_info or "universe" not in meta_info: return 6 # Fallback par défaut for token_info in meta_info["universe"]: if token_info.get("name") == symbol: return token_info.get("szDecimals", 6) return 6 # Fallback si symbole non trouvé def get_available_leverages(self) -> Dict[str, int]: """Récupère les leverages maximum disponibles pour chaque token""" meta_info = self.get_meta_info() if not meta_info or "universe" not in meta_info: return {} leverages = {} for token_info in meta_info["universe"]: if "name" in token_info and "maxLeverage" in token_info: symbol = token_info["name"] max_leverage = int(token_info["maxLeverage"]) leverages[symbol] = max_leverage return leverages def get_active_positions_count(self) -> int: """Récupère le nombre de positions actives sur Hyperliquid""" try: positions = self.get_positions() if not positions: return 0 # Compter seulement les positions avec une taille non nulle active_count = 0 for pos in positions: size = abs(float(pos.get('size', 0))) if size > 0: active_count += 1 return active_count except Exception as e: print(f"Erreur lors de la récupération du nombre de positions: {e}") return 0 def get_current_market_price(self, symbol: str) -> float: """Récupère le mark price actuel pour un symbole""" try: # Utiliser l'API metaAndAssetCtxs pour récupérer le mark price url = f"{self.base_url}/info" payload = {"type": "metaAndAssetCtxs"} response = requests.post(url, json=payload) if response.status_code != 200: return 0.0 meta_and_ctxs = response.json() if len(meta_and_ctxs) < 2: return 0.0 universe = meta_and_ctxs[0]['universe'] asset_ctxs = meta_and_ctxs[1] # Trouver l'index du symbole symbol_index = -1 for i, token_info in enumerate(universe): if token_info.get("name") == symbol: symbol_index = i break if symbol_index == -1 or symbol_index >= len(asset_ctxs): return 0.0 # Récupérer le mark price mark_price = float(asset_ctxs[symbol_index]['markPx']) return mark_price except Exception as e: print(f"Erreur récupération mark price pour {symbol}: {e}") return 0.0 def get_active_positions_details(self) -> Dict[str, Dict]: """Récupère les détails des positions actives groupées par symbol_side""" try: positions = self.get_positions() if not positions: return {} active_positions = {} for pos in positions: size = abs(float(pos.get('size', 0))) if size > 0: symbol = pos.get('symbol', '').replace('-PERP', '') side = pos.get('side', '').upper() key = f"{symbol}_{side}" active_positions[key] = { 'symbol': symbol, 'side': side, 'size': size, 'entry_price': float(pos.get('entry_price', 0)), 'unrealized_pnl': float(pos.get('unrealized_pnl', 0)), 'leverage': int(pos.get('leverage', {}).get('value', 1)), 'raw_data': pos } return active_positions except Exception as e: print(f"Erreur lors de la récupération des détails de positions: {e}") return {} def get_portfolio_summary(self) -> Dict: """Récupère un résumé complet du portfolio""" try: balances = self.get_balances() positions = self.get_positions() user_state = self.get_user_state() # Calcul des métriques total_account_value = balances.get("USDC", 0) margin_used = balances.get("USDC_margin_used", 0) available_balance = total_account_value - margin_used # PnL total des positions total_unrealized_pnl = sum(pos["unrealized_pnl"] for pos in positions) # Nombre de positions active_positions_count = len([p for p in positions if p["size"] > 0]) return { "account_value": total_account_value, "available_balance": available_balance, "margin_used": margin_used, "unrealized_pnl": total_unrealized_pnl, "active_positions": active_positions_count, "positions": positions, "balances": balances, "account_address": self.account_address, "timestamp": user_state.get("time", 0) if user_state else 0 } except Exception as e: print(f"Erreur portfolio summary: {e}") return { "account_value": 0, "available_balance": 0, "margin_used": 0, "unrealized_pnl": 0, "active_positions": 0, "positions": [], "balances": {}, "error": str(e) } def place_market_order(self, symbol: str, is_buy: bool, size: float, reduce_only: bool = False, leverage: int = None, order_tag: str = None) -> Dict: """ Place un ordre au marché sur Hyperliquid avec gestion d'erreurs améliorée Args: symbol: Symbole du token (ex: "BTC", "ETH") is_buy: True pour acheter, False pour vendre size: Taille de l'ordre reduce_only: True pour fermer une position seulement leverage: Effet de levier à appliquer (ex: 3, 5, 10) order_tag: Tag/label pour identifier l'ordre (ex: "AUTO_BUY_SIGNAL", "CLOSE_TP") Returns: Dict avec le résultat de l'ordre """ try: if not self.exchange: error_msg = "Exchange API non initialisé - impossible de passer des ordres" logger.error(error_msg) return { "success": False, "error": error_msg } # Validation des paramètres if size <= 0: error_msg = f"Taille d'ordre invalide: {size}" logger.error(error_msg) return {"success": False, "error": error_msg} # Clean symbol (remove -PERP or -USD suffix if present) clean_symbol = symbol.replace('-PERP', '').replace('-USD', '') logger.info(f"LIVE: Passage ordre marché {clean_symbol} {'BUY' if is_buy else 'SELL'} {size}" + (f" leverage {leverage}x" if leverage else "") + (" (reduce_only)" if reduce_only else "")) # Définir le leverage si spécifié (pour positions futures) if leverage and leverage > 1 and not reduce_only: try: logger.info(f"Définition leverage {leverage}x pour {clean_symbol}") leverage_result = self.exchange.update_leverage( leverage=int(leverage), name=clean_symbol, is_cross=False # Isolated margin pour plus de sécurité ) logger.info(f"Leverage défini: {leverage_result}") if leverage_result and leverage_result.get("status") == "err": logger.warning(f"Erreur leverage: {leverage_result.get('response', 'Erreur inconnue')}") # On continue quand même, le leverage existant sera utilisé except Exception as e: logger.warning(f"Erreur définition leverage: {e}") # On continue quand même # Arrondir selon les contraintes Hyperliquid exactes (szDecimals) decimals = self.get_symbol_decimals(clean_symbol) rounded_size = round(size, decimals) if rounded_size != size: logger.info(f"Taille arrondie {clean_symbol}: {size} -> {rounded_size} ({decimals} décimales)") # Créer un tag descriptif pour l'ordre if not order_tag: action = "BUY" if is_buy else "SELL" order_tag = f"AUTO_{action}_{clean_symbol}" if reduce_only: order_tag = f"CLOSE_{action}_{clean_symbol}" # Ordre au marché avec gestion d'erreurs robuste logger.info(f"Envoi ordre Hyperliquid: {clean_symbol} {is_buy} taille={rounded_size} tag={order_tag}") # Tenter d'ajouter le tag si l'API le supporte try: order_result = self.exchange.market_open( name=clean_symbol, is_buy=is_buy, sz=rounded_size, cloid=order_tag # Client Order ID pour tracking ) except Exception as tag_error: # Fallback sans tag si l'API ne le supporte pas logger.warning(f"Tag non supporté, ordre sans tag: {tag_error}") order_result = self.exchange.market_open( name=clean_symbol, is_buy=is_buy, sz=rounded_size ) logger.info(f"Résultat ordre: {order_result}") # Analyse du résultat avec patterns éprouvés if order_result and isinstance(order_result, dict): # Vérifier les erreurs d'authentification spécifiques if order_result.get("status") == "err": error_msg = order_result.get("response", "Erreur inconnue") if "does not exist" in str(error_msg): error_msg = f"API Wallet non autorisé: {error_msg} - Veuillez enregistrer l'API wallet sur Hyperliquid" logger.error(f"Erreur API Hyperliquid: {error_msg}") return { "success": False, "error": error_msg, "result": order_result } elif order_result.get("status") == "ok": # Vérifier les statuses pour des erreurs spécifiques response_data = order_result.get("result", {}).get("response", {}).get("data", {}) statuses = response_data.get("statuses", []) # Chercher des erreurs dans les statuses for status in statuses: if status.get("error"): error_msg = status.get("error") logger.error(f"Erreur ordre Hyperliquid: {error_msg}") return { "success": False, "error": f"Erreur ordre: {error_msg}", "result": order_result } response_data = order_result.get("response", {}).get("data", {}) statuses = response_data.get("statuses", []) # Récupère l'ID de l'ordre si disponible order_id = None if statuses: order_id = statuses[0].get("resting") or statuses[0].get("filled") logger.info(f"SUCCES: Ordre exécuté avec succès (ID: {order_id})") return { "success": True, "order_id": order_id, "statuses": statuses, "result": order_result, "symbol": clean_symbol, "side": "BUY" if is_buy else "SELL", "size": size, "timestamp": time.time() } else: error_details = order_result.get("response", {}).get("error", str(order_result)) error_msg = f"Ordre rejeté: {error_details}" logger.error(error_msg) return { "success": False, "error": error_msg, "result": order_result } except Exception as e: error_msg = f"Erreur critique lors du passage d'ordre: {e}" logger.error(error_msg) return { "success": False, "error": error_msg, "exception": str(e) } def get_fill_info(self, user_address: str = None, limit: int = 100) -> List[Dict]: """ Récupère les informations de fills (ordres exécutés) récents Args: user_address: Adresse utilisateur (utilise self.account_address par défaut) limit: Nombre maximum de fills à récupérer Returns: List des fills récents """ try: if not self.info: logger.error("Info API non disponible") return [] user_addr = user_address or self.account_address # Récupérer les fills récents avec l'API moderne fills = self.info.user_fills(user_addr) if fills: # Limite le nombre de résultats et trie par timestamp décroissant fills_sorted = sorted(fills, key=lambda x: x.get('time', 0), reverse=True) fills_limited = fills_sorted[:limit] logger.debug(f"Récupéré {len(fills_limited)} fills récents") return fills_limited else: logger.debug("Aucun fill trouvé") return [] except Exception as e: logger.error(f"Erreur récupération fills: {e}") return [] def get_execution_price_from_fills(self, symbol: str, order_time: float, tolerance_seconds: int = 30) -> Optional[float]: """ Récupère le prix d'exécution réel depuis les fills récents Args: symbol: Symbole recherché order_time: Timestamp de l'ordre (epoch) tolerance_seconds: Tolérance en secondes pour trouver le fill Returns: Prix d'exécution réel ou None si non trouvé """ try: fills = self.get_fill_info() for fill in fills: # Vérifier le symbole et le timing if (fill.get("coin") == symbol and abs(fill.get("time", 0) / 1000 - order_time) <= tolerance_seconds): execution_price = float(fill.get("px", 0)) print(f"Prix d'exécution trouvé: {symbol} à ${execution_price}") return execution_price print(f"WARNING: Aucun prix d'exécution trouvé pour {symbol}") return None except Exception as e: print(f"ERROR: Erreur récupération prix d'exécution: {e}") return None def group_trades_to_positions(self, trades: List[Dict]) -> List[Dict]: """ Regroupe les trades individuels en positions complètes Une position = un trade OPEN + un trade CLOSE sur le même symbole """ positions = [] open_trades = {} # Stockage temporaire des trades d'ouverture # Trier les trades par timestamp (plus ancien en premier) sorted_trades = sorted(trades, key=lambda x: x['timestamp']) for trade in sorted_trades: symbol = trade['symbol'] operation = trade.get('operation', '') if operation == 'OPEN' or 'Open' in str(trade.get('dir', '')): # Trade d'ouverture - stocker pour matcher avec fermeture if symbol not in open_trades: open_trades[symbol] = [] open_trades[symbol].append(trade) elif operation == 'CLOSE' or 'Close' in str(trade.get('dir', '')): # Trade de fermeture - chercher le trade d'ouverture correspondant if symbol in open_trades and open_trades[symbol]: # Prendre le trade d'ouverture le plus ancien (FIFO) open_trade = open_trades[symbol].pop(0) # Créer une position complète position = { 'id': f"pos_{open_trade['id']}_{trade['id']}", 'symbol': symbol, 'side': 'LONG' if open_trade['side'] == 'BUY' else 'SHORT', 'entry_price': open_trade['price'], 'exit_price': trade['price'], 'size': open_trade['size'], 'open_time': open_trade['timestamp'], 'close_time': trade['timestamp'], 'duration': self._calculate_duration(open_trade['timestamp'], trade['timestamp']), 'pnl': trade.get('pnl', 0) or self._calculate_pnl( open_trade['price'], trade['price'], open_trade['size'], open_trade['side'] ), 'fees': (open_trade.get('fee', 0) + trade.get('fee', 0)), 'source': 'HYPERLIQUID' } positions.append(position) # Ajouter les positions encore ouvertes (trades OPEN sans CLOSE) for symbol, trades in open_trades.items(): for trade in trades: # Position toujours ouverte position = { 'id': f"pos_open_{trade['id']}", 'symbol': symbol, 'side': 'LONG' if trade['side'] == 'BUY' else 'SHORT', 'entry_price': trade['price'], 'exit_price': None, # Pas encore fermée 'size': trade['size'], 'open_time': trade['timestamp'], 'close_time': None, 'duration': self._calculate_duration(trade['timestamp'], time.time()), 'pnl': 0, # PnL non réalisé 'fees': trade.get('fee', 0), 'status': 'OPEN', 'source': 'HYPERLIQUID' } positions.append(position) return positions def _calculate_duration(self, start_time: float, end_time: float) -> str: """Calcule la durée entre deux timestamps""" duration_seconds = int(end_time - start_time) hours = duration_seconds // 3600 minutes = (duration_seconds % 3600) // 60 seconds = duration_seconds % 60 if hours > 0: return f"{hours}h {minutes}m {seconds}s" else: return f"{minutes}m {seconds}s" def _calculate_pnl(self, entry_price: float, exit_price: float, size: float, side: str) -> float: """Calcule le PnL d'une position""" if side == 'BUY': # LONG return (exit_price - entry_price) * size else: # SHORT return (entry_price - exit_price) * size def get_position_history(self, limit: int = 100) -> List[Dict]: """ Récupère l'historique des positions (trades regroupés) Returns: Liste des positions fermées et ouvertes """ try: # Récupérer les trades individuels trades = self.sync_trade_history(limit=limit * 2) # Plus de trades pour avoir assez de positions # Regrouper en positions positions = self.group_trades_to_positions(trades) # Trier par timestamp décroissant (plus récent en premier) positions.sort(key=lambda x: x.get('close_time') or x.get('open_time', 0), reverse=True) # Limiter le nombre de positions retournées return positions[:limit] except Exception as e: logger.error(f"Erreur récupération historique positions: {e}") return [] def get_candles_data(self, symbol: str, interval: str = "1h", lookback_hours: int = 24) -> Optional[List[Dict]]: """ Récupère les données de chandeliers pour un symbole Args: symbol: Symbole (ex: 'BTC', 'ETH') interval: Intervalle de temps ('1m', '5m', '15m', '1h', '4h', '1d') lookback_hours: Nombre d'heures à récupérer Returns: Liste des chandeliers avec 'time', 'open', 'high', 'low', 'close', 'volume' """ try: if not self.info: logger.error("Info API non disponible pour récupérer les chandeliers") return None # Calculer les timestamps import time as time_module end_time = int(time_module.time() * 1000) # maintenant en ms start_time = end_time - (lookback_hours * 60 * 60 * 1000) # lookback_hours heures en arrière # Récupérer les données via l'API Info candles = self.info.candles_snapshot( coin=symbol, interval=interval, startTime=start_time, endTime=end_time ) if not candles: logger.warning(f"Aucune donnée chandelier pour {symbol}") return None # Convertir au format standard formatted_candles = [] for candle in candles: if len(candle) >= 6: # [time, open, high, low, close, volume] formatted_candles.append({ 'time': int(candle[0]), 'open': float(candle[1]), 'high': float(candle[2]), 'low': float(candle[3]), 'close': float(candle[4]), 'volume': float(candle[5]) if len(candle) > 5 else 0.0 }) logger.debug(f"Récupéré {len(formatted_candles)} chandeliers pour {symbol}") return formatted_candles except Exception as e: logger.error(f"Erreur récupération chandeliers {symbol}: {e}") return None def calculate_correlation(self, symbol1: str, symbol2: str, lookback_hours: int = 24) -> Optional[float]: """ Calcule la corrélation entre deux symboles Args: symbol1: Premier symbole (ex: 'BTC') symbol2: Second symbole (ex: 'ETH') lookback_hours: Période de calcul en heures Returns: Corrélation entre -1 et 1, ou None si erreur """ try: # Récupérer les données candles1 = self.get_candles_data(symbol1, interval="1h", lookback_hours=lookback_hours) candles2 = self.get_candles_data(symbol2, interval="1h", lookback_hours=lookback_hours) if not candles1 or not candles2: logger.warning(f"Données manquantes pour corrélation {symbol1}-{symbol2}") return None # Extraire les prix de clôture et les aligner sur les timestamps prices1 = {} prices2 = {} for candle in candles1: prices1[candle['time']] = candle['close'] for candle in candles2: prices2[candle['time']] = candle['close'] # Prendre les timestamps communs common_times = sorted(set(prices1.keys()) & set(prices2.keys())) if len(common_times) < 2: logger.warning(f"Pas assez de données communes pour {symbol1}-{symbol2}") return None # Extraire les prix alignés aligned_prices1 = [prices1[t] for t in common_times] aligned_prices2 = [prices2[t] for t in common_times] # Calculer les variations returns1 = [(aligned_prices1[i] - aligned_prices1[i-1]) / aligned_prices1[i-1] for i in range(1, len(aligned_prices1))] returns2 = [(aligned_prices2[i] - aligned_prices2[i-1]) / aligned_prices2[i-1] for i in range(1, len(aligned_prices2))] if len(returns1) < 2: return None # Calcul de la corrélation de Pearson import statistics mean1 = statistics.mean(returns1) mean2 = statistics.mean(returns2) numerator = sum((returns1[i] - mean1) * (returns2[i] - mean2) for i in range(len(returns1))) sum_sq1 = sum((returns1[i] - mean1) ** 2 for i in range(len(returns1))) sum_sq2 = sum((returns2[i] - mean2) ** 2 for i in range(len(returns2))) denominator = (sum_sq1 * sum_sq2) ** 0.5 if denominator == 0: return 0.0 correlation = numerator / denominator logger.debug(f"Corrélation {symbol1}-{symbol2}: {correlation:.3f} sur {len(returns1)} périodes") return correlation except Exception as e: logger.error(f"Erreur calcul corrélation {symbol1}-{symbol2}: {e}") return None def get_btc_correlation(self, symbol: str, lookback_hours: int = 24) -> Optional[float]: """ Calcule la corrélation entre BTC et un autre symbole Args: symbol: Symbole à comparer avec BTC lookback_hours: Période de calcul Returns: Corrélation avec BTC (-1 à 1) ou None """ if symbol == 'BTC': return 1.0 # BTC est parfaitement corrélé avec lui-même return self.calculate_correlation('BTC', symbol, lookback_hours) def sync_trade_history(self, limit: int = 100) -> List[Dict]: """ Récupère l'historique des trades Hyperliquid et les convertit au format du système Args: limit: Nombre maximum de fills à récupérer Returns: Liste des trades au format du système """ try: fills = self.get_fill_info(limit=limit) if not fills: logger.info("Aucun fill Hyperliquid à synchroniser") return [] synced_trades = [] for fill in fills: # Conversion timestamp (ms to seconds) timestamp = fill.get('time', 0) / 1000 # Détection du côté (B = Buy/Long, A = Sell/Short dans Hyperliquid) side = 'BUY' if fill.get('side') == 'B' else 'SELL' # Type d'opération depuis 'dir' direction = fill.get('dir', '') operation_type = 'OPEN' if 'Open' in direction else 'CLOSE' # Création du trade au format système trade = { 'id': f"hl_{fill.get('oid', 0)}", # Préfixe Hyperliquid 'symbol': fill.get('coin', ''), 'side': side, 'operation': operation_type, 'size': float(fill.get('sz', 0)), 'price': float(fill.get('px', 0)), 'timestamp': timestamp, 'fee': float(fill.get('fee', 0)), 'pnl': float(fill.get('closedPnl', 0)) if fill.get('closedPnl') else None, 'order_id': fill.get('oid'), 'hash': fill.get('hash'), 'source': 'HYPERLIQUID', 'raw_fill': fill # Données brutes pour debug } synced_trades.append(trade) logger.info(f"Synchronisé {len(synced_trades)} trades depuis Hyperliquid") return synced_trades except Exception as e: logger.error(f"Erreur synchronisation historique: {e}") return [] def get_account_state(self) -> Optional[Dict]: """ Récupère l'état complet du compte Hyperliquid Compatible avec l'API dashboard Returns: Dict avec marginSummary, positions, etc. """ try: user_state = self.get_user_state() if not user_state: logger.warning("Impossible de récupérer l'état utilisateur") return None # Retourner les données complètes pour le dashboard return user_state except Exception as e: logger.error(f"Erreur récupération état compte: {e}") return None def update_position_stop_loss(self, symbol: str, stop_loss_price: float) -> Dict: """ Met à jour le stop loss d'une position en créant un ordre stop limit Utilise exchange.order() pour créer un ordre de protection Args: symbol: Symbole du token (ex: "BTC", "ETH") stop_loss_price: Nouveau prix de stop loss Returns: Dict avec le résultat de la mise à jour """ try: if not self.exchange: return {"success": False, "error": "Exchange API non disponible"} clean_symbol = symbol.replace('-PERP', '') rounded_price = round(float(stop_loss_price), 2) logger.info(f"Création ordre stop loss: {clean_symbol} stop=${rounded_price:.2f}") # Récupérer la position pour déterminer le côté et la taille positions = self.get_positions() target_position = None for pos in positions: if pos.get('symbol', '').replace('-PERP', '') == clean_symbol: target_position = pos break if not target_position: return {"success": False, "error": f"Position {clean_symbol} non trouvée"} position_size = float(target_position.get('size', 0)) if position_size == 0: return {"success": False, "error": f"Position {clean_symbol} fermée"} # Déterminer le sens de l'ordre stop if position_size > 0: # Position longue is_buy = False # Vendre pour fermer order_side = "Sell" else: # Position courte is_buy = True # Acheter pour fermer order_side = "Buy" order_size = abs(position_size) logger.info(f"Ordre stop: {order_side} {order_size} {clean_symbol} @ ${rounded_price}") # D'abord annuler les ordres stop existants pour ce symbole try: existing_orders = self.info.open_orders(self.account_address) if self.info else [] if existing_orders: for order in existing_orders: if (order.get('coin') == clean_symbol and order.get('reduceOnly', False)): order_id = order.get('oid') if order_id: try: self.exchange.cancel(clean_symbol, int(order_id)) logger.info(f"Annulé ordre stop existant: {order_id}") except Exception as cancel_e: logger.warning(f"Échec annulation ordre {order_id}: {cancel_e}") except Exception as e: logger.warning(f"Échec vérification ordres existants: {e}") # Créer un ordre stop loss comme ceux créés manuellement try: # D'après l'analyse de l'ordre manuel, utiliser le bon format result = self.exchange.order( name=clean_symbol, is_buy=is_buy, sz=order_size, limit_px=rounded_price, order_type={"trigger": {"triggerPx": str(rounded_price), "isMarket": False, "tpsl": "sl"}}, reduce_only=True ) logger.info(f"Résultat ordre stop: {result}") if result and result.get("status") == "ok": return { "success": True, "stop_loss_price": rounded_price, "symbol": clean_symbol, "order_side": order_side, "order_size": order_size, "result": result, "method": "stop_trigger_order" } else: # Log de l'erreur détaillée error_details = "" if result and "response" in result and "data" in result["response"]: data = result["response"]["data"] if "statuses" in data and data["statuses"]: error_details = data["statuses"][0].get("error", "Erreur inconnue") logger.warning(f"Échec ordre stop: {error_details}") # Fallback vers système virtuel logger.info(f"Activation stop loss virtuel pour {clean_symbol} à ${rounded_price:.2f}") return { "success": True, "stop_loss_price": rounded_price, "symbol": clean_symbol, "order_side": order_side, "order_size": order_size, "virtual": True, "method": "trailing_stop_fallback", "note": f"Fallback - ordre échoué: {error_details}" } except Exception as order_e: logger.warning(f"Exception ordre stop: {order_e}") # Fallback vers système virtuel logger.info(f"Activation stop loss virtuel pour {clean_symbol} à ${rounded_price:.2f}") return { "success": True, "stop_loss_price": rounded_price, "symbol": clean_symbol, "order_side": order_side, "order_size": order_size, "virtual": True, "method": "trailing_stop_fallback", "note": "Fallback - exception ordre" } except Exception as e: safe_e = str(e).replace('{', '{{').replace('}', '}}') error_msg = f"Erreur update stop loss: {safe_e}" logger.error(error_msg) return { "success": False, "error": error_msg, "exception": str(e) } def modify_order(self, order_id: str, new_price: float = None, new_size: float = None) -> Dict: """ Modifie un ordre existant sur Hyperliquid Args: order_id: ID de l'ordre à modifier new_price: Nouveau prix (optionnel) new_size: Nouvelle taille (optionnel) Returns: Dict avec le résultat de la modification """ try: if not self.exchange: return {"success": False, "error": "Exchange API non disponible"} if not order_id: return {"success": False, "error": "Order ID requis"} logger.info(f"Modification ordre {order_id}: prix={new_price}, taille={new_size}") # Préparer les modifications modifications = { "oid": int(order_id) } if new_price is not None: modifications["px"] = str(new_price) if new_size is not None: modifications["sz"] = str(new_size) # Utiliser la méthode modify de l'exchange result = self.exchange.modify_order(modifications) logger.info(f"Résultat modification: {result}") if result and result.get("status") == "ok": return { "success": True, "order_id": order_id, "new_price": new_price, "new_size": new_size, "result": result } else: error_msg = result.get("response", {}).get("error", str(result)) return { "success": False, "error": f"Échec modification ordre: {error_msg}", "result": result } except Exception as e: error_msg = f"Erreur modification ordre: {e}" logger.error(error_msg) return { "success": False, "error": error_msg, "exception": str(e) } def cancel_order(self, order_id: str = None, cloid: str = None) -> Dict: """ Annule un ordre sur Hyperliquid Args: order_id: ID de l'ordre à annuler cloid: Client Order ID (alternative à order_id) Returns: Dict avec le résultat de l'annulation """ try: if not self.exchange: return {"success": False, "error": "Exchange API non disponible"} if not order_id and not cloid: return {"success": False, "error": "Order ID ou CLOID requis"} logger.info(f"Annulation ordre: ID={order_id}, CLOID={cloid}") # Préparer la requête d'annulation if order_id: cancel_request = {"oid": int(order_id)} else: cancel_request = {"cloid": cloid} # Annuler l'ordre result = self.exchange.cancel_order(cancel_request) logger.info(f"Résultat annulation: {result}") if result and result.get("status") == "ok": return { "success": True, "order_id": order_id, "cloid": cloid, "result": result } else: error_msg = result.get("response", {}).get("error", str(result)) return { "success": False, "error": f"Échec annulation ordre: {error_msg}", "result": result } except Exception as e: error_msg = f"Erreur annulation ordre: {e}" logger.error(error_msg) return { "success": False, "error": error_msg, "exception": str(e) } def get_open_orders(self, symbol: str = None) -> List[Dict]: """ Récupère les ordres ouverts Args: symbol: Symbole spécifique (optionnel) Returns: Liste des ordres ouverts """ try: if not self.info: return [] open_orders = self.info.open_orders(self.account_address) if symbol: clean_symbol = symbol.replace('-PERP', '') open_orders = [order for order in open_orders if order.get('coin') == clean_symbol] logger.debug(f"Ordres ouverts trouvés: {len(open_orders)}") return open_orders except Exception as e: logger.error(f"Erreur récupération ordres ouverts: {e}") return [] # Instance globale avec initialisation améliorée try: from hyperliquid_config import API_WALLET_PRIVATE_KEY hyperliquid_client = HyperliquidClient(api_wallet_key=API_WALLET_PRIVATE_KEY) logger.info("Client Hyperliquid global initialisé avec support Exchange") except Exception as e: logger.warning(f"Initialisation Exchange échouée: {e}") hyperliquid_client = HyperliquidClient() # Mode lecture seule logger.info("Client Hyperliquid global initialisé en mode lecture seule")