# -*- coding: utf-8 -*- """ Script principal pour lancer le trading avec le système de signaux pragmatique """ import sys import os sys.path.append('..') sys.path.append('.') import logging import time from datetime import datetime # Imports depuis le dossier parent from hyperliquid_client import HyperliquidClient import trading_config # Imports locaux from simple_trading_system.simple_trading_engine import SimpleTradingEngine from simple_trading_system.pragmatic_signals import PragmaticSignalGenerator, create_signal_provider def setup_logger(): """Configure le logger principal""" logger = logging.getLogger("PragmaticTrading") logger.setLevel(logging.INFO) # Handler console console_handler = logging.StreamHandler() console_handler.setLevel(logging.INFO) # Format détaillé formatter = logging.Formatter( '%(asctime)s - [%(name)s] - %(levelname)s - %(message)s', datefmt='%Y-%m-%d %H:%M:%S' ) console_handler.setFormatter(formatter) if not logger.handlers: logger.addHandler(console_handler) return logger def main(): """Fonction principale""" logger = setup_logger() logger.info("=" * 80) logger.info("DEMARRAGE DU SYSTEME DE TRADING PRAGMATIQUE") logger.info("=" * 80) # Choisir le mode de trading print("\nChoisissez le mode de trading:") print("1. SIMULATION - Test sans risque") print("2. LIVE_MANUAL - Trading réel avec validation manuelle") print("3. LIVE_AUTO - Trading automatique (DANGER)") choice = input("\nVotre choix (1/2/3): ").strip() mode_map = { '1': 'SIMULATION', '2': 'LIVE_MANUAL', '3': 'LIVE_AUTO' } trading_mode = mode_map.get(choice, 'SIMULATION') if trading_mode in ['LIVE_MANUAL', 'LIVE_AUTO']: confirm = input(f"\nATTENTION: Mode {trading_mode} sélectionné. Confirmer? (oui/non): ") if confirm.lower() != 'oui': logger.info("Trading annulé") return logger.info(f"Mode sélectionné: {trading_mode}") # Initialiser Hyperliquid client si nécessaire hyperliquid_client = None if trading_mode in ['LIVE_MANUAL', 'LIVE_AUTO']: try: hyperliquid_client = HyperliquidClient( api_key=trading_config.HYPERLIQUID_API_KEY, api_secret=trading_config.HYPERLIQUID_API_SECRET ) logger.info("Client Hyperliquid initialisé") except Exception as e: logger.error(f"Erreur initialisation Hyperliquid: {e}") logger.info("Basculement en mode SIMULATION") trading_mode = 'SIMULATION' # Créer le générateur de signaux signal_generator = PragmaticSignalGenerator( hyperliquid_client=hyperliquid_client, logger=logger ) # Créer le trading engine trading_engine = SimpleTradingEngine( trading_mode=trading_mode, initial_balance=1000, hyperliquid_client=hyperliquid_client ) # Connecter le générateur de signaux trading_engine.set_signals_provider(signal_generator.generate_signals) logger.info("Configuration terminée") logger.info("-" * 40) logger.info(f"Symboles surveillés: {', '.join(signal_generator.symbols)}") logger.info(f"Max positions: {trading_engine.max_positions}") logger.info(f"Taille position: {trading_engine.position_size_pct * 100:.1f}%") logger.info("-" * 40) # Mode manuel: demander confirmation pour chaque signal if trading_mode == 'LIVE_MANUAL': logger.info("\nMODE MANUEL ACTIF") logger.info("Les signaux seront affichés pour validation") logger.info("Appuyez sur Ctrl+C pour arrêter\n") try: while True: # Générer les signaux signals = signal_generator.generate_signals() if signals: logger.info(f"\n{'='*60}") logger.info(f"NOUVEAUX SIGNAUX DETECTES: {len(signals)}") logger.info(f"{'='*60}") for i, signal in enumerate(signals, 1): logger.info(f"\nSignal #{i}:") logger.info(f" Symbole: {signal['symbol']}") logger.info(f" Direction: {signal['side']}") logger.info(f" Prix: ${signal['current_price']:.4f}") logger.info(f" Leverage: {signal['leverage']}x") logger.info(f" Qualité: {signal['quality_score']:.2%}") logger.info(f" Raison: {signal['reason']}") if trading_engine.should_open_position(signal): action = input(f"\nOuvrir cette position? (o/n): ").strip().lower() if action == 'o': position = trading_engine.open_position(signal) if position: logger.info(f"Position ouverte: {position.id}") else: logger.error("Échec ouverture position") else: logger.info("Signal ignoré") else: logger.info("Signal rejeté par le système de filtres") # Afficher le résumé summary = signal_generator.get_signal_summary() status = trading_engine.get_status_dict() logger.info(f"\n--- Status ---") logger.info(f"Positions actives: {status['metrics']['active_positions']}/{status['metrics']['max_positions']}") logger.info(f"Signaux en cours: {summary['total_signals']}") # Pause avant la prochaine vérification time.sleep(30) except KeyboardInterrupt: logger.info("\n\nArrêt demandé par l'utilisateur") # Mode automatique elif trading_mode == 'LIVE_AUTO': logger.info("\nMODE AUTOMATIQUE ACTIF") logger.info("Le trading va commencer automatiquement") logger.info("Appuyez sur Ctrl+C pour arrêter\n") # Démarrer le trading engine trading_engine.start() try: while True: # Afficher le résumé périodiquement summary = signal_generator.get_signal_summary() status = trading_engine.get_status_dict() logger.info(f"\n--- Status Auto ---") logger.info(f"Timestamp: {datetime.now().strftime('%H:%M:%S')}") logger.info(f"Positions: {status['metrics']['active_positions']}/{status['metrics']['max_positions']}") logger.info(f"Signaux détectés: {summary['total_signals']}") if summary['signals']: for sig in summary['signals']: logger.info(f" • {sig['symbol']} {sig['side']}: {sig['reason']}") time.sleep(60) # Status toutes les minutes except KeyboardInterrupt: logger.info("\n\nArrêt demandé par l'utilisateur") trading_engine.stop() # Mode simulation else: logger.info("\nMODE SIMULATION") logger.info("Test du système sans risque réel") logger.info("Appuyez sur Ctrl+C pour arrêter\n") # En simulation, on peut tester rapidement try: iteration = 0 while True: iteration += 1 logger.info(f"\n--- Itération #{iteration} ---") # Générer et analyser les signaux signals = signal_generator.generate_signals() if signals: logger.info(f"Signaux générés: {len(signals)}") for signal in signals: logger.info(f" • {signal['symbol']} {signal['side']}: {signal['reason']} (Q:{signal['quality_score']:.2f})") if trading_engine.should_open_position(signal): position = trading_engine.open_position(signal) if position: logger.info(f" → Position simulée ouverte") else: logger.info("Aucun signal détecté") # Status status = trading_engine.get_status_dict() logger.info(f"Positions simulées: {status['metrics']['active_positions']}/{status['metrics']['max_positions']}") logger.info(f"Balance simulée: ${status['metrics']['balance']:.2f}") time.sleep(10) # Test rapide en simulation except KeyboardInterrupt: logger.info("\n\nSimulation terminée") logger.info("\n" + "=" * 80) logger.info("ARRET DU SYSTEME DE TRADING") logger.info("=" * 80) if __name__ == "__main__": main()