#!/usr/bin/env python3 """ Service de surveillance et reporting par email Envoie un récapitulatif des trades et analyse du marché toutes les 6h """ import smtplib import json import time from email.mime.text import MIMEText from email.mime.multipart import MIMEMultipart from datetime import datetime, timedelta from typing import List, Dict import logging import os import sys # Importer le client Hyperliquid from hyperliquid_client import HyperliquidClient class TradingReportMailer: """Service d'envoi de rapports de trading par email""" def __init__(self, recipient_email: str = "binet.benjamin@gmail.com"): self.recipient_email = recipient_email self.hyperliquid_client = HyperliquidClient() self.report_interval = 6 * 3600 # 6 heures en secondes self.last_report_time = 0 # Configuration logging self.setup_logger() def setup_logger(self): """Configure le logging""" self.logger = logging.getLogger("TradingReportMailer") self.logger.setLevel(logging.INFO) handler = logging.StreamHandler() formatter = logging.Formatter('%(asctime)s - [%(name)s] - %(message)s') handler.setFormatter(formatter) if not self.logger.handlers: self.logger.addHandler(handler) def get_recent_trades(self, hours: int = 6) -> List[Dict]: """Récupère les trades des X dernières heures""" try: # Utiliser l'API info pour récupérer les fills user_fills = self.hyperliquid_client.info.user_fills( self.hyperliquid_client.account_address ) if not user_fills: return [] # Log du nombre total de fills (limite API = 2000) total_fills = len(user_fills) if total_fills >= 2000: self.logger.warning(f"⚠️ Limite API atteinte: {total_fills} fills récupérés (max 2000)") # Filtrer les fills des X dernières heures cutoff_time = int((datetime.now() - timedelta(hours=hours)).timestamp() * 1000) recent_fills = [ fill for fill in user_fills if fill.get('time', 0) > cutoff_time ] # Log du nombre de fills dans la période self.logger.info(f"📊 Fills récupérés: {len(recent_fills)} dans les {hours}h (sur {total_fills} total)") return recent_fills except Exception as e: self.logger.error(f"Erreur récupération trades: {e}") return [] def get_current_positions(self) -> List[Dict]: """Récupère les positions actuelles""" try: positions = self.hyperliquid_client.get_positions() return positions if positions else [] except Exception as e: self.logger.error(f"Erreur récupération positions: {e}") return [] def get_account_balance(self) -> float: """Récupère la balance du compte""" try: user_state = self.hyperliquid_client.info.user_state( self.hyperliquid_client.account_address ) if user_state and 'marginSummary' in user_state: # accountValue = balance totale incluant PnL non réalisé balance = float(user_state['marginSummary'].get('accountValue', 0)) return balance return 0.0 except Exception as e: self.logger.error(f"Erreur récupération balance: {e}") return 0.0 def analyze_trades(self, trades: List[Dict]) -> Dict: """Analyse les trades et génère des statistiques""" if not trades: return { 'total_trades': 0, 'winning_trades': 0, 'losing_trades': 0, 'total_pnl': 0, 'win_rate': 0, 'avg_win': 0, 'avg_loss': 0 } total_pnl = 0 winning_trades = 0 losing_trades = 0 total_win_pnl = 0 total_loss_pnl = 0 for fill in trades: # Extraire le PnL si disponible (closedPnl existe dans l'API Hyperliquid) closed_pnl = float(fill.get('closedPnl', 0)) # Compter seulement les trades qui ferment une position (closedPnl != 0) if closed_pnl != 0: total_pnl += closed_pnl if closed_pnl > 0: winning_trades += 1 total_win_pnl += closed_pnl else: losing_trades += 1 total_loss_pnl += closed_pnl total_trades = winning_trades + losing_trades win_rate = (winning_trades / total_trades * 100) if total_trades > 0 else 0 avg_win = (total_win_pnl / winning_trades) if winning_trades > 0 else 0 avg_loss = (total_loss_pnl / losing_trades) if losing_trades > 0 else 0 return { 'total_trades': total_trades, 'winning_trades': winning_trades, 'losing_trades': losing_trades, 'total_pnl': total_pnl, 'win_rate': win_rate, 'avg_win': avg_win, 'avg_loss': avg_loss } def get_market_analysis(self) -> Dict: """Analyse le marché actuel""" try: # Récupérer les données BTC from binance_data_provider import BinanceDataProvider binance = BinanceDataProvider() # Utiliser BTC (qui sera mappé à BTCUSDT par le provider) btc_data = binance.get_candles('BTC', '1h', 24) if not btc_data or len(btc_data) < 2: return {'btc_change_24h': 0, 'btc_trend': 'UNKNOWN', 'btc_price': 0} btc_price_24h_ago = float(btc_data[0]['close']) # Close price il y a 24h btc_price_now = float(btc_data[-1]['close']) # Close price actuel btc_change_24h = ((btc_price_now - btc_price_24h_ago) / btc_price_24h_ago) * 100 # Déterminer la tendance if btc_change_24h > 2: trend = "BULLISH 📈" elif btc_change_24h < -2: trend = "BEARISH 📉" else: trend = "RANGE 📊" return { 'btc_price': btc_price_now, 'btc_change_24h': btc_change_24h, 'btc_trend': trend } except Exception as e: self.logger.error(f"Erreur analyse marché: {e}") return {'btc_change_24h': 0, 'btc_trend': 'UNKNOWN', 'btc_price': 0} def format_html_report(self, trades_stats: Dict, market_analysis: Dict, positions: List[Dict], account_balance: float) -> str: """Génère le rapport HTML""" now = datetime.now().strftime("%Y-%m-%d %H:%M") # Charger les peaks enregistrés peaks_data = {} try: with open('position_peaks.json', 'r') as f: peaks_data = json.load(f) except: pass # Récupérer les trades fermés des 6 dernières heures closed_trades_html = "" recent_trades = self.get_recent_trades(hours=6) # Filtrer pour avoir uniquement les trades de fermeture (tous les trades des 6h) closed_only = [t for t in recent_trades if float(t.get('closedPnl', 0)) != 0] # Log du nombre de trades fermés trouvés self.logger.info(f"📋 Trades fermés trouvés pour le rapport: {len(closed_only)}") for trade in closed_only: # Tous les trades fermés des 6 dernières heures coin = trade.get('coin', '') direction = trade.get('dir', '') # 'Buy' ou 'Sell' de l'API Hyperliquid pnl = float(trade.get('closedPnl', 0)) price = float(trade.get('px', 0)) size = float(trade.get('sz', 0)) # Convertir direction API en LONG/SHORT # Buy = ouverture de LONG (ou fermeture de SHORT) # Sell = fermeture de LONG (ou ouverture de SHORT) # Pour les trades fermés, si closedPnl existe, c'est une fermeture # On infère le side à partir du side initial sauvegardé side_display = "" if direction.lower() == 'buy': # Buy avec closedPnl = fermeture de SHORT side_display = "SHORT" else: # Sell avec closedPnl = fermeture de LONG side_display = "LONG" pnl_color = "green" if pnl >= 0 else "red" pnl_sign = "+" if pnl >= 0 else "" # Calculer le PnL% final # Note: closedPnl est en USD, on doit estimer la valeur de la position position_value = size * price # Valeur approximative final_pnl_pct = (pnl / position_value * 100) if position_value > 0 else 0 # Chercher le peak enregistré - essayer plusieurs clés peak_key_options = [ f"{coin}_{side_display}", # Ex: BTC_LONG f"{coin}_{direction}", # Ex: BTC_Buy f"{coin}_LONG", f"{coin}_SHORT" ] peak_pnl_pct = 0 for key in peak_key_options: if key in peaks_data: peak_pnl_pct = peaks_data[key].get('peak_pnl_percent', 0) if peak_pnl_pct != 0: break # Afficher le peak seulement s'il existe peak_display = f"{peak_pnl_pct:+.1f}%" if peak_pnl_pct != 0 else "-" closed_trades_html += f""" {coin} {side_display} ${price:.4f} {size:.4f} {peak_display} {pnl_sign}${pnl:.2f} ({final_pnl_pct:+.1f}%) """ if not closed_only: closed_trades_html = 'Aucun trade fermé' # Couleurs pour les stats pnl_color = "green" if trades_stats['total_pnl'] >= 0 else "red" pnl_sign = "+" if trades_stats['total_pnl'] >= 0 else "" wr_color = "green" if trades_stats['win_rate'] >= 50 else "orange" if trades_stats['win_rate'] >= 40 else "red" html = f"""

📊 Trading Report - {now}

💼 Balance & Marché

Balance Compte
${account_balance:,.2f}
BTC Prix
${market_analysis.get('btc_price', 0):,.2f}
BTC Variation 24h
{market_analysis.get('btc_change_24h', 0):+.2f}%
Tendance
{market_analysis.get('btc_trend', 'UNKNOWN')}

💰 Performance des 6 dernières heures

Total Trades
{trades_stats['total_trades']}
Win Rate
{trades_stats['win_rate']:.1f}%
PnL Total
{pnl_sign}${trades_stats['total_pnl']:.2f}
Trades Gagnants / Perdants
{trades_stats['winning_trades']} / {trades_stats['losing_trades']}

📋 Historique 6h

{closed_trades_html}
Symbole Side Prix Taille Peak PnL Final

🤖 Rapport généré automatiquement par Crypto Trading Bot

Prochain rapport dans 6 heures

""" return html def send_email(self, subject: str, html_content: str) -> bool: """Envoie l'email via la commande mail système""" try: import subprocess import tempfile # Créer un fichier temporaire pour le HTML with tempfile.NamedTemporaryFile(mode='w', suffix='.html', delete=False) as f: f.write(html_content) html_file = f.name # Utiliser la commande mail système avec From personnalisé cmd = [ 'mail', '-s', subject, '-a', f'Content-Type: text/html; charset=UTF-8', '-a', 'From: Bot Hyperliquid ', self.recipient_email ] with open(html_file, 'r') as f: result = subprocess.run( cmd, stdin=f, capture_output=True, text=True, timeout=30 ) # Nettoyer le fichier temporaire os.unlink(html_file) if result.returncode == 0: self.logger.info(f"✅ Email envoyé avec succès à {self.recipient_email}") return True else: self.logger.error(f"❌ Erreur mail: {result.stderr}") return False except Exception as e: self.logger.error(f"❌ Erreur envoi email: {e}") return False def generate_and_send_report(self): """Génère et envoie le rapport complet""" try: self.logger.info("📊 Génération du rapport de trading...") # Récupérer les données trades = self.get_recent_trades(hours=6) positions = self.get_current_positions() account_balance = self.get_account_balance() # Analyser trades_stats = self.analyze_trades(trades) market_analysis = self.get_market_analysis() # Générer le HTML html_content = self.format_html_report(trades_stats, market_analysis, positions, account_balance) # Envoyer subject = f"🤖 Trading Report - {datetime.now().strftime('%Y-%m-%d %H:%M')} - Balance: ${account_balance:.2f} - PnL 6h: ${trades_stats['total_pnl']:.2f}" success = self.send_email(subject, html_content) if success: self.last_report_time = time.time() self.logger.info("✅ Rapport envoyé avec succès") else: self.logger.error("❌ Échec envoi du rapport") return success except Exception as e: self.logger.error(f"❌ Erreur génération rapport: {e}") return False def run(self): """Boucle principale du service""" self.logger.info("🚀 Service de reporting démarré") self.logger.info(f"📧 Destinataire: {self.recipient_email}") self.logger.info(f"⏰ Intervalle: {self.report_interval / 3600}h") # Envoyer un premier rapport au démarrage self.logger.info("📨 Envoi du premier rapport...") self.generate_and_send_report() while True: try: # Attendre l'intervalle time.sleep(self.report_interval) # Générer et envoyer le rapport self.generate_and_send_report() except KeyboardInterrupt: self.logger.info("🛑 Arrêt du service de reporting") break except Exception as e: self.logger.error(f"❌ Erreur dans la boucle: {e}") time.sleep(60) # Attendre 1 minute avant de réessayer if __name__ == "__main__": # Créer et démarrer le service mailer = TradingReportMailer(recipient_email="binet.benjamin@gmail.com") mailer.run()