#!/usr/bin/env python3 """ Time Pattern Analyzer - Détection de patterns horaires et journaliers Exploite les régularités temporelles du marché crypto """ import numpy as np from typing import Dict, List, Optional from datetime import datetime, timedelta import logging logger = logging.getLogger("TimePatternAnalyzer") class TimePatternAnalyzer: """Analyse des patterns temporels (heures, jours de la semaine)""" def __init__(self): # Statistiques des performances horaires (à enrichir avec l'historique) self.hourly_performance = {} # {hour: {avg_return, volatility, win_rate}} self.daily_performance = {} # {weekday: {avg_return, volatility, win_rate}} # Patterns connus (basés sur observations empiriques du marché crypto) self.known_patterns = { 'best_hours': [14, 15, 16, 17], # Heures UTC les plus actives (US markets) 'worst_hours': [2, 3, 4, 5], # Heures creuses 'best_days': [1, 2, 3], # Mardi, Mercredi, Jeudi (0=Lundi) 'worst_days': [5, 6], # Samedi, Dimanche } logger.info("✅ Time Pattern Analyzer initialisé") def analyze_time_patterns(self, symbol: str = None) -> Dict: """ Analyse les patterns temporels actuels Returns: Dict avec bonus/malus selon l'heure et le jour """ now = datetime.utcnow() hour = now.hour weekday = now.weekday() # 0=Lundi, 6=Dimanche analysis = { 'current_hour': hour, 'current_weekday': weekday, 'hour_score': 0, 'day_score': 0, 'combined_score': 0, 'is_favorable_time': False, 'recommendations': [] } # === ANALYSE HORAIRE === hour_score = self._score_hour(hour) analysis['hour_score'] = hour_score if hour in self.known_patterns['best_hours']: analysis['recommendations'].append(f"✅ Heure favorable ({hour}h UTC)") elif hour in self.known_patterns['worst_hours']: analysis['recommendations'].append(f"⚠️ Heure creuse ({hour}h UTC)") # === ANALYSE JOUR DE LA SEMAINE === day_score = self._score_weekday(weekday) analysis['day_score'] = day_score if weekday in self.known_patterns['best_days']: analysis['recommendations'].append(f"✅ Jour favorable ({self._weekday_name(weekday)})") elif weekday in self.known_patterns['worst_days']: analysis['recommendations'].append(f"⚠️ Weekend - volatilité réduite") # === PATTERNS SPÉCIAUX === # US Market Open (14h30 UTC = 9h30 EST) if 14 <= hour <= 15: hour_score += 5 analysis['recommendations'].append("📈 US Market Open - forte activité") # Asian Market Hours (23h-8h UTC) if hour >= 23 or hour <= 8: hour_score -= 3 analysis['recommendations'].append("🌏 Asian Hours - liquidité réduite") # Vendredi après-midi (fin de semaine) if weekday == 4 and hour >= 16: # Vendredi après 16h UTC day_score -= 5 analysis['recommendations'].append("⚠️ Fin de semaine - prudence") # Dimanche soir (début de semaine) if weekday == 6 and hour >= 20: # Dimanche soir day_score += 5 analysis['recommendations'].append("✅ Ouverture hebdomadaire - opportunités") # === SCORE COMBINÉ === combined_score = hour_score + day_score analysis['combined_score'] = combined_score analysis['is_favorable_time'] = combined_score > 0 return analysis def _score_hour(self, hour: int) -> float: """ Score l'heure actuelle (-10 à +10) Basé sur patterns empiriques du marché crypto: - Pics d'activité: 14h-17h UTC (US markets) - Creux: 2h-5h UTC (nuit US) """ # Heures de forte activité (US trading hours) if 14 <= hour <= 17: return 10 # Bonnes heures (Europe + début US) if 8 <= hour <= 13: return 5 # Heures moyennes (fin US) if 18 <= hour <= 21: return 0 # Heures creuses (Asia + nuit US) if hour >= 22 or hour <= 1: return -5 # Très creuses (nuit profonde) if 2 <= hour <= 5: return -10 return 0 def _score_weekday(self, weekday: int) -> float: """ Score le jour de la semaine (-10 à +10) 0=Lundi, 1=Mardi, 2=Mercredi, 3=Jeudi, 4=Vendredi, 5=Samedi, 6=Dimanche """ # Meilleurs jours (milieu de semaine) if weekday in [1, 2, 3]: # Mardi, Mercredi, Jeudi return 8 # Bon jour (Lundi) if weekday == 0: return 5 # Jour moyen (Vendredi) if weekday == 4: return 0 # Weekends (faible liquidité) if weekday in [5, 6]: # Samedi, Dimanche return -8 return 0 def _weekday_name(self, weekday: int) -> str: """Retourne le nom du jour""" days = ['Lundi', 'Mardi', 'Mercredi', 'Jeudi', 'Vendredi', 'Samedi', 'Dimanche'] return days[weekday] if 0 <= weekday <= 6 else 'Inconnu' def should_trade_now(self) -> bool: """ Détermine si c'est un bon moment pour trader Returns: True si conditions temporelles favorables """ analysis = self.analyze_time_patterns() return analysis['is_favorable_time'] def get_time_bonus(self) -> float: """ Retourne un bonus de score basé sur le timing (-10 à +10) Returns: Bonus à ajouter au score d'une opportunité """ analysis = self.analyze_time_patterns() return analysis['combined_score'] def learn_from_trade(self, symbol: str, entry_time: datetime, profit: float, duration_hours: float): """ Apprend des trades passés pour améliorer les statistiques horaires/journalières Args: symbol: Symbole tradé entry_time: Heure d'entrée profit: Profit réalisé (%) duration_hours: Durée du trade en heures """ hour = entry_time.hour weekday = entry_time.weekday() # Mise à jour stats horaires if hour not in self.hourly_performance: self.hourly_performance[hour] = { 'total_trades': 0, 'total_profit': 0, 'wins': 0 } stats = self.hourly_performance[hour] stats['total_trades'] += 1 stats['total_profit'] += profit stats['wins'] += 1 if profit > 0 else 0 stats['avg_return'] = stats['total_profit'] / stats['total_trades'] stats['win_rate'] = (stats['wins'] / stats['total_trades']) * 100 # Mise à jour stats journalières if weekday not in self.daily_performance: self.daily_performance[weekday] = { 'total_trades': 0, 'total_profit': 0, 'wins': 0 } daily_stats = self.daily_performance[weekday] daily_stats['total_trades'] += 1 daily_stats['total_profit'] += profit daily_stats['wins'] += 1 if profit > 0 else 0 daily_stats['avg_return'] = daily_stats['total_profit'] / daily_stats['total_trades'] daily_stats['win_rate'] = (daily_stats['wins'] / daily_stats['total_trades']) * 100 logger.debug(f"📊 Stats mises à jour - {symbol} à {hour}h, jour {weekday}") def get_best_trading_hours(self, min_trades: int = 10) -> List[int]: """ Retourne les meilleures heures pour trader (basé sur historique) Args: min_trades: Minimum de trades pour considérer l'heure Returns: Liste des heures (0-23) triées par performance """ if not self.hourly_performance: return self.known_patterns['best_hours'] valid_hours = [] for hour, stats in self.hourly_performance.items(): if stats['total_trades'] >= min_trades: score = stats['avg_return'] * stats['win_rate'] valid_hours.append((hour, score)) if not valid_hours: return self.known_patterns['best_hours'] # Trier par score décroissant valid_hours.sort(key=lambda x: x[1], reverse=True) return [hour for hour, _ in valid_hours[:5]] def get_best_trading_days(self, min_trades: int = 10) -> List[int]: """ Retourne les meilleurs jours pour trader Returns: Liste des jours (0-6) triés par performance """ if not self.daily_performance: return self.known_patterns['best_days'] valid_days = [] for day, stats in self.daily_performance.items(): if stats['total_trades'] >= min_trades: score = stats['avg_return'] * stats['win_rate'] valid_days.append((day, score)) if not valid_days: return self.known_patterns['best_days'] valid_days.sort(key=lambda x: x[1], reverse=True) return [day for day, _ in valid_days[:5]] def get_statistics_summary(self) -> Dict: """ Retourne un résumé des statistiques temporelles Returns: Dict avec stats globales """ summary = { 'total_hours_tracked': len(self.hourly_performance), 'total_days_tracked': len(self.daily_performance), 'best_hours': self.get_best_trading_hours(), 'best_days': self.get_best_trading_days(), 'hourly_stats': {}, 'daily_stats': {} } # Stats horaires détaillées for hour, stats in self.hourly_performance.items(): if stats['total_trades'] >= 5: summary['hourly_stats'][hour] = { 'trades': stats['total_trades'], 'avg_return': round(stats.get('avg_return', 0), 2), 'win_rate': round(stats.get('win_rate', 0), 1) } # Stats journalières détaillées for day, stats in self.daily_performance.items(): if stats['total_trades'] >= 5: summary['daily_stats'][self._weekday_name(day)] = { 'trades': stats['total_trades'], 'avg_return': round(stats.get('avg_return', 0), 2), 'win_rate': round(stats.get('win_rate', 0), 1) } return summary # Instance globale _time_pattern_analyzer = None def get_time_pattern_analyzer() -> TimePatternAnalyzer: """Retourne l'instance globale du time pattern analyzer""" global _time_pattern_analyzer if _time_pattern_analyzer is None: _time_pattern_analyzer = TimePatternAnalyzer() return _time_pattern_analyzer