#!/usr/bin/env python3 """ SIGNAL CACHE - Speichert Signal-Daten für ML Training Wenn ein Trade geöffnet wird, werden die Signal-Informationen hier gespeichert. Wenn der Trade geschlossen wird, können die Daten abgerufen und zum demo_tracker geloggt werden. VERWENDUNG: from signal_cache import signal_cache # Beim Trade-Open: signal_cache.store(ticket_id, signal_info, enhanced_score, hybrid_score) # Beim Trade-Close: cached = signal_cache.get(ticket_id) if cached: demo_tracker.log_trade(..., base_confidence=cached['base_confidence'], enhanced_score=cached['enhanced_score'], hybrid_score=cached['hybrid_score'], signal_quality=cached['signal_quality'] ) """ import json import os import logging from datetime import datetime from typing import Dict, Optional, Any logger = logging.getLogger(__name__) # ========================================== # CONFIGURATION # ========================================== CACHE_FILE = "signal_cache.json" MAX_CACHE_AGE_HOURS = 48 # Alte Einträge nach 48h löschen # ========================================== # SIGNAL CACHE CLASS # ========================================== class SignalCache: """ Cache für Trade-Signal-Daten Persistiert auf Disk für Restart-Sicherheit """ def __init__(self, cache_file: str = CACHE_FILE): self.cache_file = cache_file self.cache: Dict[str, Dict] = {} self._load_cache() logger.info(f"SignalCache initialized with {len(self.cache)} entries") def _load_cache(self): """Lädt Cache von Disk""" if os.path.exists(self.cache_file): try: with open(self.cache_file, 'r') as f: self.cache = json.load(f) self._cleanup_old_entries() except Exception as e: logger.warning(f"Could not load signal cache: {e}") self.cache = {} def _save_cache(self): """Speichert Cache auf Disk""" try: with open(self.cache_file, 'w') as f: json.dump(self.cache, f, indent=2, default=str) except Exception as e: logger.error(f"Could not save signal cache: {e}") def _cleanup_old_entries(self): """Entfernt Einträge älter als MAX_CACHE_AGE_HOURS""" now = datetime.now() to_remove = [] for ticket, data in self.cache.items(): try: timestamp = datetime.fromisoformat(data.get('timestamp', '')) age_hours = (now - timestamp).total_seconds() / 3600 if age_hours > MAX_CACHE_AGE_HOURS: to_remove.append(ticket) except: to_remove.append(ticket) for ticket in to_remove: del self.cache[ticket] if to_remove: logger.info(f"Cleaned up {len(to_remove)} old cache entries") self._save_cache() def store(self, ticket: int, signal_info: Dict[str, Any], enhanced_score: float = 0.0, hybrid_score: float = 0.0, lot_multiplier: float = 1.0) -> bool: """ Speichert Signal-Daten für einen Trade Args: ticket: MT5 Order Ticket ID signal_info: Dictionary mit Signal-Analyse Daten enhanced_score: Enhanced Signal Score (0-100) hybrid_score: Hybrid Score (0-100) lot_multiplier: Equity Curve Lot Multiplier Returns: True wenn erfolgreich gespeichert """ try: # Extrahiere relevante Daten aus signal_info cache_entry = { 'ticket': ticket, 'timestamp': datetime.now().isoformat(), # Signal Analysis 'base_confidence': signal_info.get('confidence', 0), 'adaptive_threshold': signal_info.get('adaptive_threshold', 0), 'signal_quality': signal_info.get('signal_quality', 'unknown'), 'entry_signal': signal_info.get('entry_signal', 0), # Market Regime 'market_regime': signal_info.get('market_regime', 'unknown'), 'regime_strength': signal_info.get('regime_strength', 0), 'risk_adjusted_strength': signal_info.get('risk_adjusted_strength', 0), # Enhanced Scoring 'enhanced_score': enhanced_score, 'hybrid_score': hybrid_score, # Equity Curve 'lot_multiplier': lot_multiplier, # Session 'session': signal_info.get('session', 'unknown'), } self.cache[str(ticket)] = cache_entry self._save_cache() logger.info(f"Cached signal for ticket {ticket}: conf={cache_entry['base_confidence']:.1f}%, quality={cache_entry['signal_quality']}") return True except Exception as e: logger.error(f"Failed to cache signal for ticket {ticket}: {e}") return False def get(self, ticket: int) -> Optional[Dict]: """ Holt gecachte Signal-Daten für einen Trade Args: ticket: MT5 Order Ticket ID Returns: Dict mit Signal-Daten oder None """ return self.cache.get(str(ticket)) def remove(self, ticket: int) -> bool: """ Entfernt einen Eintrag nach Verwendung Args: ticket: MT5 Order Ticket ID Returns: True wenn erfolgreich entfernt """ ticket_str = str(ticket) if ticket_str in self.cache: del self.cache[ticket_str] self._save_cache() return True return False def get_and_remove(self, ticket: int) -> Optional[Dict]: """ Holt und entfernt Signal-Daten (für Trade-Close) Args: ticket: MT5 Order Ticket ID Returns: Dict mit Signal-Daten oder None """ data = self.get(ticket) if data: self.remove(ticket) return data def get_stats(self) -> Dict: """Gibt Cache-Statistiken zurück""" return { 'total_entries': len(self.cache), 'cache_file': self.cache_file, 'oldest_entry': min( (datetime.fromisoformat(d.get('timestamp', datetime.now().isoformat())) for d in self.cache.values()), default=None ), 'tickets': list(self.cache.keys()) } def print_status(self): """Zeigt Cache-Status an""" stats = self.get_stats() print("\n" + "=" * 50) print("SIGNAL CACHE STATUS") print("=" * 50) print(f" Cached Trades: {stats['total_entries']}") print(f" Cache File: {stats['cache_file']}") if stats['tickets']: print(f" Tickets: {', '.join(stats['tickets'][:5])}") if len(stats['tickets']) > 5: print(f" ... and {len(stats['tickets']) - 5} more") print("=" * 50) # ========================================== # GLOBAL INSTANCE # ========================================== signal_cache = SignalCache() # ========================================== # HELPER FUNCTIONS # ========================================== def store_trade_signal(ticket: int, signal_info: Dict, enhanced_score: float = 0.0, hybrid_score: float = 0.0, lot_multiplier: float = 1.0) -> bool: """Convenience function to store signal data""" return signal_cache.store(ticket, signal_info, enhanced_score, hybrid_score, lot_multiplier) def get_trade_signal(ticket: int) -> Optional[Dict]: """Convenience function to get and remove signal data""" return signal_cache.get_and_remove(ticket) def get_cache_status() -> str: """Returns formatted cache status""" stats = signal_cache.get_stats() return f"SignalCache: {stats['total_entries']} entries cached" # ========================================== # STANDALONE TEST # ========================================== if __name__ == "__main__": print("Testing Signal Cache...") # Test data test_signal = { 'confidence': 85.5, 'adaptive_threshold': 70, 'signal_quality': 'excellent', 'entry_signal': 1, 'market_regime': 'trending', 'regime_strength': 65.0, 'risk_adjusted_strength': 12500.0, 'session': 'asian' } # Store signal_cache.store(12345, test_signal, enhanced_score=78.5, hybrid_score=82.0) # Retrieve cached = signal_cache.get(12345) print(f"\nCached data: {json.dumps(cached, indent=2)}") # Status signal_cache.print_status() # Cleanup signal_cache.remove(12345) print("\nTest completed!")