""" DollarTaker - Arrache exactement 1$ net au marché par position Calcule précisément les frais pour garantir un profit net de 1.00$ USD """ import sys import logging import time import math from datetime import datetime, timedelta from threading import Thread, Lock from typing import Dict, List, Optional import os # Fix encoding for Windows console sys.stdout.reconfigure(encoding='utf-8') class DollarTaker: """Worker spécialisé dans l'arrachage de 1$ net par position""" def __init__(self, hyperliquid_client=None): self.hyperliquid_client = hyperliquid_client self.running = False # Configuration DollarTaker self.target_profit_usd = 1.00 # Exactement 1$ net self.hyperliquid_fee_rate = 0.00025 # 0.025% par trade (maker) self.check_interval = 5 # Vérifier toutes les 5 secondes # Storage des positions surveillées self.tracked_positions = {} # {symbol_side: DollarPosition} self.lock = Lock() # Statistiques self.dollars_taken = 0 self.total_positions_closed = 0 self.average_hold_time = 0 # Logger self.setup_logger() self.logger.info("=== DOLLAR TAKER STARTED ===") self.logger.info(f"Target: ${self.target_profit_usd:.2f} net per position") self.logger.info(f"Fee rate: {self.hyperliquid_fee_rate*100:.3f}% per trade") def setup_logger(self): """Configure le système de logging""" self.logger = logging.getLogger("DollarTaker") self.logger.setLevel(logging.INFO) # Créer le dossier logs si nécessaire logs_dir = "logs" if not os.path.exists(logs_dir): os.makedirs(logs_dir) # Handler fichier file_handler = logging.FileHandler(os.path.join(logs_dir, "dollar_taker.log")) file_handler.setLevel(logging.INFO) # Handler console console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # Format formatter = logging.Formatter('%(asctime)s - [%(name)s] - %(message)s') file_handler.setFormatter(formatter) console_handler.setFormatter(formatter) if not self.logger.handlers: self.logger.addHandler(file_handler) self.logger.addHandler(console_handler) def calculate_target_price(self, entry_price: float, side: str, size: float, leverage: int) -> float: """Calcule le prix de sortie exact pour 1$ net de profit""" # Valeur de la position position_value = entry_price * size margin_used = position_value / leverage # Frais d'ouverture (déjà payés) opening_fee = position_value * self.hyperliquid_fee_rate # Pour avoir 1$ net, il faut: # profit_brut - closing_fee = 1.00$ # donc: profit_brut = 1.00$ + closing_fee # Le closing_fee dépend du prix de sortie, donc on doit résoudre: # profit_brut = |exit_price - entry_price| * size # closing_fee = exit_price * size * fee_rate if side == 'LONG': # Pour LONG: profit = (exit_price - entry_price) * size # On veut: (exit_price - entry_price) * size - exit_price * size * fee_rate = 1.00 # Simplifie: exit_price * size * (1 - fee_rate) - entry_price * size = 1.00 # exit_price = (1.00 + entry_price * size) / (size * (1 - fee_rate)) target_exit_price = (self.target_profit_usd + entry_price * size) / (size * (1 - self.hyperliquid_fee_rate)) else: # SHORT # Pour SHORT: profit = (entry_price - exit_price) * size # On veut: (entry_price - exit_price) * size - exit_price * size * fee_rate = 1.00 # Simplifie: entry_price * size - exit_price * size * (1 + fee_rate) = 1.00 # exit_price = (entry_price * size - 1.00) / (size * (1 + fee_rate)) target_exit_price = (entry_price * size - self.target_profit_usd) / (size * (1 + self.hyperliquid_fee_rate)) return target_exit_price def verify_dollar_calculation(self, entry_price: float, exit_price: float, side: str, size: float) -> float: """Vérifie que le calcul donne bien 1$ net""" # Profit brut if side == 'LONG': gross_profit = (exit_price - entry_price) * size else: # SHORT gross_profit = (entry_price - exit_price) * size # Frais de fermeture closing_fee = exit_price * size * self.hyperliquid_fee_rate # Profit net net_profit = gross_profit - closing_fee return net_profit def update_tracked_positions(self): """Met à jour les positions surveillées depuis Hyperliquid""" if not self.hyperliquid_client: return try: hl_positions = self.hyperliquid_client.get_positions() if not hl_positions: return with self.lock: current_keys = set() for pos in hl_positions: symbol = pos.get('symbol', '').replace('-PERP', '') side = pos.get('side', '').upper() key = f"{symbol}_{side}" current_keys.add(key) entry_price = float(pos.get('entry_price', 0)) size = abs(float(pos.get('size', 0))) leverage = int(pos.get('leverage', {}).get('value', 1)) if key not in self.tracked_positions: # Nouvelle position à surveiller target_price = self.calculate_target_price(entry_price, side, size, leverage) self.tracked_positions[key] = { 'symbol': symbol, 'side': side, 'entry_price': entry_price, 'size': size, 'leverage': leverage, 'target_price': target_price, 'opened_at': datetime.now(), 'dollars_to_take': 1.00 } # Vérification du calcul net_profit = self.verify_dollar_calculation(entry_price, target_price, side, size) self.logger.info(f"NEW TARGET: {symbol} {side}") self.logger.info(f" Entry: ${entry_price:.4f} | Target: ${target_price:.4f}") self.logger.info(f" Size: {size:.6f} | Leverage: {leverage}x") self.logger.info(f" Expected net profit: ${net_profit:.6f}") # Supprimer les positions fermées for key in list(self.tracked_positions.keys()): if key not in current_keys: pos = self.tracked_positions[key] hold_time = datetime.now() - pos['opened_at'] self.logger.info(f"POSITION CLOSED: {pos['symbol']} {pos['side']} (held {hold_time})") del self.tracked_positions[key] except Exception as e: self.logger.error(f"Erreur update tracked positions: {e}") def check_dollar_opportunities(self): """Vérifie si une position peut être fermée pour 1$ net""" if not self.hyperliquid_client: return try: with self.lock: for key, tracked_pos in self.tracked_positions.items(): symbol = tracked_pos['symbol'] side = tracked_pos['side'] target_price = tracked_pos['target_price'] entry_price = tracked_pos['entry_price'] # Récupérer le prix actuel current_price = self.hyperliquid_client.get_current_market_price(symbol) if not current_price: continue # Vérifier si on peut prendre le dollar should_close = False if side == 'LONG' and current_price >= target_price: should_close = True elif side == 'SHORT' and current_price <= target_price: should_close = True if should_close: # Calculer le profit net exact avec le prix actuel net_profit = self.verify_dollar_calculation( entry_price, current_price, side, tracked_pos['size'] ) if net_profit >= 0.98: # Tolérance de 2 cents self.close_position_for_dollar(tracked_pos, current_price, net_profit) except Exception as e: self.logger.error(f"Erreur check dollar opportunities: {e}") def close_position_for_dollar(self, tracked_pos: Dict, current_price: float, expected_net: float): """Ferme une position pour arracher exactement 1$""" try: symbol = tracked_pos['symbol'] side = tracked_pos['side'] size = tracked_pos['size'] # Fermer la position result = self.hyperliquid_client.close_position(symbol) if result and result.get('success'): hold_time = datetime.now() - tracked_pos['opened_at'] # Statistiques self.dollars_taken += 1 self.total_positions_closed += 1 self.logger.info("=" * 60) self.logger.info(f"DOLLAR TAKEN! {symbol} {side}") self.logger.info(f"Entry: ${tracked_pos['entry_price']:.4f} -> Exit: ${current_price:.4f}") self.logger.info(f"Expected net: ${expected_net:.6f}") self.logger.info(f"Hold time: {hold_time}") self.logger.info(f"Total dollars taken: ${self.dollars_taken:.2f}") self.logger.info("=" * 60) else: self.logger.error(f"Échec fermeture {symbol}: {result.get('error') if result else 'No result'}") except Exception as e: self.logger.error(f"Erreur fermeture position: {e}") def log_status(self): """Affiche le statut des positions surveillées""" try: with self.lock: if not self.tracked_positions: self.logger.info("No positions being tracked for dollar taking") return self.logger.info("=" * 80) self.logger.info("DOLLAR TAKER STATUS:") self.logger.info(f"Dollars taken: ${self.dollars_taken:.2f} | Positions tracked: {len(self.tracked_positions)}") self.logger.info("=" * 80) for key, pos in self.tracked_positions.items(): current_price = self.hyperliquid_client.get_current_market_price(pos['symbol']) if current_price: distance_to_target = abs(current_price - pos['target_price']) distance_pct = (distance_to_target / pos['entry_price']) * 100 direction = "↗" if pos['side'] == 'LONG' else "↘" self.logger.info(f"{pos['symbol']:<6} {pos['side']:<5} {direction} | " f"Current: ${current_price:>8.4f} | Target: ${pos['target_price']:>8.4f} | " f"Distance: {distance_pct:>5.3f}%") self.logger.info("=" * 80) except Exception as e: self.logger.error(f"Erreur log status: {e}") def dollar_taking_loop(self): """Boucle principale du DollarTaker""" self.logger.info("Démarrage de la boucle DollarTaker...") status_counter = 0 while self.running: try: # Mettre à jour les positions surveillées self.update_tracked_positions() # Vérifier les opportunités self.check_dollar_opportunities() # Log status toutes les 12 fois (1 minute) status_counter += 1 if status_counter % 12 == 0: self.log_status() time.sleep(self.check_interval) except Exception as e: self.logger.error(f"Erreur dans dollar_taking_loop: {e}") time.sleep(self.check_interval) def start(self): """Démarre le DollarTaker""" if not self.running: self.running = True self.dollar_thread = Thread(target=self.dollar_taking_loop) self.dollar_thread.daemon = True self.dollar_thread.start() self.logger.info("DollarTaker démarré - Prêt à arracher des dollars!") def stop(self): """Arrête le DollarTaker""" self.running = False if hasattr(self, 'dollar_thread'): self.dollar_thread.join(timeout=5) self.logger.info(f"DollarTaker arrêté - Total: ${self.dollars_taken:.2f} arrachés!")