Place-Order-Trading-Bot/signal_cache.py

#!/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!")