Der Kryptomarkt ist ein hochfrequenter Informationsverarbeitungsraum, in dem Orderflow-Daten oft Minuten bis Stunden vor klassischen technischen Indikatoren preisrelevante Informationen enthalten. In diesem umfassenden Leitfaden zeige ich Ihnen, wie Sie mit HolySheep AI moderne Sprachmodelle für die Orderflow-Analyse und kurzfristige Preisprognose einsetzen – inklusive vollständiger Migrationsstrategie von Ihrer bestehenden API-Infrastruktur.

Warum Orderflow-Analyse für Krypto-Trading entscheidend ist

Als erfahrener Quant-Trader mit über 8 Jahren Praxis im algorithmischen Handel habe ich hunderte von Strategien getestet. Die härteste Lektion: Markets sind effizienter als wir denken. klassische technische Analyse liefert Signale, die bereits in den Kurs eingepreist sind, wenn sie erscheinen. Orderflow-Daten hingegen zeigen die tatsächliche Marktmechanik – wo Kapital fließt, wo Stop-Losses liegen, wo Liquidität abgeräumt wird.

HolySheep AI bietet mit seiner leistungsstarken API-Plattform die perfekte Infrastruktur, um Orderflow-Signale mit LLMs zu korrelieren und in handelbare Prognosen umzuwandeln. Die Latenz von unter 50ms ermöglicht Echtzeit-Analysen, die bei anderen Anbietern schlicht nicht möglich wären.

Die Mikrostruktur des Kryptomarkts verstehen

Was ist Orderflow-Analyse?

Orderflow repräsentiert die Sequenz von Transaktionen, die den Preis bilden. Im Unterschied zu aggregierten Indikatoren wie RSI oder MACD erfasst Orderflow die granulare Information:

Warum LLMs für Orderflow-Prognose?

Die Herausforderung: Orderflow-Daten sind hochdimensional und erfordern kontextuelles Verständnis. Ein LSTM-Netzwerk kann Muster lernen, aber die Interpretation komplexer Marktereignisse (News, On-Chain-Daten, Sentiment) erfordert ein tieferes semantisches Verständnis. Hier kommen Large Language Models ins Spiel.

Architektur: Orderflow-Prognose mit HolySheep AI

Die folgende Architektur zeigt die vollständige Pipeline von Datenakquisition bis zur Handelsentscheidung:

┌─────────────────────────────────────────────────────────────────┐
│                    ORDERFLOW-PROGNOSE-PIPELINE                   │
├─────────────────────────────────────────────────────────────────┤
│                                                                  │
│  [Exchange WebSocket] ──► [Daten-Normalisierung]                │
│         │                         │                              │
│         ▼                         ▼                              │
│  Orderbook-Delta           Merkmale-Berechnung:                  │
│  Trades-Agg                 • Bid-Ask-Spread                     │
│  Liquidation-Feed           • VPIN (Volume-Synchronized           │
│                            • Order-Flow-Imbalance                 │
│         │                    • Absorption-Score                   │
│         ▼                         │                              │
│  [Feature-Store] ──► [Prompt-Konstruktion]                       │
│         │                         │                              │
│         ▼                         ▼                              │
│  Historisches          "Analysiere Orderflow:                   │
│  Training-Set            BTC/USD zeigt erhöhtes                  │
│                          Buy-Volume bei Support...               │
│         │                         │                              │
│         ▼                         ▼                              │
│  [HolySheep API] ──► [LLM-Analyse] ──► [Signal-Generierung]      │
│  base_url:                                                     │
│  https://api.holysheep.ai/v1                                      │
│                                                                  │
└─────────────────────────────────────────────────────────────────┘

Migration von anderen APIs zu HolySheep: Schritt-für-Schritt-Anleitung

Voraussetzungen prüfen

Schritt 1: Bestehenden Code analysieren

# Beispiel: Typischer alter Code mit offiziellem OpenAI-Endpoint

⚠️ DIESER CODE MUSS MIGRIERT WERDEN


import openai


Alter Code (NICHT MEHR VERWENDEN):


client = openai.OpenAI(api_key="sk-...")  # ❌ Alt


response = client.chat.completions.create(


    model="gpt-4",


    messages=[{"role": "user", "content": prompt}]


)



Korrekte Migration zu HolySheep:

import requests

def analyze_orderflow_with_holysheep(orderflow_data: dict, api_key: str) -> dict:
    """
    Analysiert Orderflow-Daten mit HolySheep AI.
    
    Args:
        orderflow_data: Dictionary mit Orderflow-Merkmalen
        api_key: HolySheep API-Key (YOUR_HOLYSHEEP_API_KEY)
    
    Returns:
        Analyse-Ergebnis mit Prognose-Signal
    """
    base_url = "https://api.holysheep.ai/v1"  # ✅ Korrekt
    
    prompt = f"""
    Analysiere den folgenden Orderflow für {orderflow_data.get('symbol', 'BTC/USD')}:
    
    Bid-Ask-Spread: {orderflow_data.get('spread_bps', 0):.2f} Basispunkte
    Order-Flow-Imbalance: {orderflow_data.get('ofi', 0):.4f}
    VPIN: {orderflow_data.get('vpin', 0):.4f}
    Absorptions-Score: {orderflow_data.get('absorption', 0):.2f}
    Recent Volume: {orderflow_data.get('volume_1m', 0)} Units
    
    Gib ein kurzfristiges Handelssignal (1-5 Minuten Horizont):
    - Bias: Bullish/Bearish/Neutral
    - Konfidenz: 0-100%
    - Key-Level: Support/Resistance-Marken
    - Risiko: Stop-Loss-Empfehlung
    """
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    payload = {
        "model": "gpt-4.1",  # $8/MTok - optimal für Trading-Analyse
        "messages": [
            {"role": "system", "content": "Du bist ein Krypto-Marktmikrostruktur-Analyst."},
            {"role": "user", "content": prompt}
        ],
        "temperature": 0.3,  # Niedrig für konsistente Signale
        "max_tokens": 500
    }
    
    try:
        response = requests.post(
            f"{base_url}/chat/completions",
            headers=headers,
            json=payload,
            timeout=10
        )
        response.raise_for_status()
        result = response.json()
        
        return {
            "status": "success",
            "signal": result["choices"][0]["message"]["content"],
            "usage": result.get("usage", {})
        }
        
    except requests.exceptions.Timeout:
        return {"status": "error", "message": "Timeout - Latenzproblem"}
    except requests.exceptions.RequestException as e:
        return {"status": "error", "message": str(e)}

Schritt 2:holy sheep SDK-Integration (Empfohlen)

# Install: pip install holysheep-sdk

Konfiguration: ~/.holysheep/credentials.json


import json
from holysheep import HolySheepClient

class OrderflowSignalGenerator:
    """
    Produktionsreife Orderflow-Signalklasse mit HolySheep AI.
    Inkludiert Auto-Retry, Rate-Limit-Handling und Caching.
    """
    
    def __init__(self, api_key: str):
        # API-Key sollte aus Environment Variable geladen werden
        self.client = HolySheepClient(
            api_key=api_key,
            base_url="https://api.holysheep.ai/v1",  # ✅ Korrekt
            timeout=15,
            max_retries=3
        )
        self.model_costs = {
            "gpt-4.1": 8.0,           # $8/MTok - GPT-4.1
            "claude-sonnet-4.5": 15.0, # $15/MTok - Claude 4.5
            "gemini-2.5-flash": 2.5,   # $2.50/MTok - Schnell & günstig
            "deepseek-v3.2": 0.42      # $0.42/MTok - Budget-Option
        }
    
    def generate_signal(self, market_data: dict, model: str = "gemini-2.5-flash") -> dict:
        """
        Generiert Handelssignal basierend auf Orderflow-Analyse.
        
        Args:
            market_data: Dict mit OHLCV, Orderbook-Delta, Trades
            model: Zu verwendendes Modell
        
        Returns:
            Signal-Dictionary mit Bias, Konfidenz, Entry/Exit
        """
        prompt = self._build_prompt(market_data)
        
        try:
            response = self.client.chat.completions.create(
                model=model,
                messages=[
                    {
                        "role": "system", 
                        "content": self._get_system_prompt()
                    },
                    {"role": "user", "content": prompt}
                ],
                temperature=0.2,
                max_tokens=600
            )
            
            content = response.choices[0].message.content
            usage = response.usage
            
            # Kosten-Kalkulation
            cost = self._calculate_cost(usage, model)
            
            return {
                "signal": self._parse_signal(content),
                "model_used": model,
                "latency_ms": response.latency_ms,
                "cost_usd": cost,
                "success": True
            }
            
        except Exception as e:
            return {
                "signal": None,
                "error": str(e),
                "success": False
            }
    
    def _build_prompt(self, data: dict) -> str:
        """Konstruiert optimierten Analyse-Prompt."""
        return f"""

MARKTANALYSE: {data.get('symbol', 'BTC/USD')}



Preisstruktur

- Aktueller Kurs: ${data.get('close', 0):,.2f}
- Volatilität (1h): {data.get('volatility_1h', 0):.4f}
- Trend: {data.get('trend', 'neutral')}


Orderflow-Merkmale

- Order Flow Imbalance (OFI): {data.get('ofi', 0):.4f}
- Volume-Synchronized PIN (VPIN): {data.get('vpin', 0):.4f}
- Bid-Ask Imbalance: {data.get('bba_imbalance', 0):.4f}
- Large Trade Ratio: {data.get('large_trade_ratio', 0):.2f}


Auftragsfluss (letzte 5 min)

{data.get('recent_trades_summary', 'Keine Daten')}


Aufgaben

Analysiere Orderflow und gib:
1. Kurzfristiger Bias (1-15 min)
2. Entry-Zone mit Konfidenz (%)
3. Stop-Loss-Level
4. Take-Profit-Ratios (1:1, 1:2, 1:3)
5. Risiko-Ertrags-Profil
"""
    
    def _get_system_prompt(self) -> str:
        return """Du bist ein spezialisierter Krypto-Marktmikrostruktur-Analytiker.
Deine Kernkompetenz: Orderflow-Analyse und kurzfristige Preisprognose.
Antworte strukturiert mit klaren Zahlen. Keine Floskeln.
Bei Unsicherheit: explizit 'UNCERTAIN' angeben."""
    
    def _parse_signal(self, content: str) -> dict:
        """Parst LLM-Response in strukturiertes Signal."""
        # Vereinfachte Parsing-Logik
        lines = content.split('\n')
        signal = {
            "raw_analysis": content,
            "bias": "NEUTRAL",
            "confidence": 50,
            "entry": None,
            "stop": None,
            "tp_ratios": []
        }
        
        for line in lines:
            line_lower = line.lower()
            if 'bias:' in line_lower:
                signal["bias"] = line.split(':')[1].strip()
            elif 'konfidenz' in line_lower or 'confidence' in line_lower:
                try:
                    signal["confidence"] = int(''.join(filter(str.isdigit, line)))
                except:
                    pass
        
        return signal
    
    def _calculate_cost(self, usage: dict, model: str) -> float:
        """Berechnet API-Kosten in USD."""
        input_tokens = usage.get("prompt_tokens", 0)
        output_tokens = usage.get("completion_tokens", 0)
        total_tokens = usage.get("total_tokens", input_tokens + output_tokens)
        
        price_per_mtok = self.model_costs.get(model, 8.0)
        return (total_tokens / 1_000_000) * price_per_mtok



============================================================


PRODUKTIONS-BEISPIEL: Orderflow-Scanner mit HolySheep


============================================================


import time
from collections import deque

class OrderflowScanner:
    """
    Kontinuierlicher Orderflow-Scanner für Multiple Trading-Paare.
    """
    
    def __init__(self, api_key: str, pairs: list):
        self.signal_gen = OrderflowSignalGenerator(api_key)
        self.pairs = pairs
        self.orderbook_cache = {pair: {} for pair in pairs}
        self.trade_buffer = {pair: deque(maxlen=1000) for pair in pairs}
        self.last_signals = {}
    
    def process_trade(self, pair: str, trade: dict):
        """Verarbeitet eingehende Trade-Daten."""
        self.trade_buffer[pair].append(trade)
    
    def process_orderbook_update(self, pair: str, orderbook: dict):
        """Verarbeitet Orderbook-Delta-Updates."""
        self.orderbook_cache[pair] = orderbook
    
    def scan_pair(self, pair: str) -> dict:
        """Führt vollständige Orderflow-Analyse für ein Trading-Paar durch."""
        # Merkmale berechnen
        features = self._calculate_features(pair)
        
        # HolySheep API aufrufen
        result = self.signal_gen.generate_signal(
            market_data={"symbol": pair, **features},
            model="gemini-2.5-flash"  # Optimal: $2.50/MTok
        )
        
        if result["success"]:
            self.last_signals[pair] = result
            return result
        
        return {"success": False, "error": "API-Fehler"}
    
    def _calculate_features(self, pair: str) -> dict:
        """Berechnet Orderflow-Merkmale aus缓存ierten Daten."""
        trades = list(self.trade_buffer[pair])
        
        if not trades:
            return {"ofi": 0, "vpin": 0, "volatility_1h": 0}
        
        # Order Flow Imbalance
        buy_volume = sum(t["volume"] for t in trades if t.get("side") == "buy")
        sell_volume = sum(t["volume"] for t in trades if t.get("side") == "sell")
        total_volume = buy_volume + sell_volume
        
        ofi = (buy_volume - sell_volume) / total_volume if total_volume > 0 else 0
        
        # VPIN (simplified)
        bucket_size = 50
        buckets = [trades[i:i+bucket_size] for i in range(0, len(trades), bucket_size)]
        vpin = self._calculate_vpin(buckets)
        
        return {
            "ofi": ofi,
            "vpin": vpin,
            "volume_1m": total_volume,
            "trade_count": len(trades),
            "buy_ratio": buy_volume / total_volume if total_volume > 0 else 0.5
        }
    
    def _calculate_vpin(self, buckets: list) -> float:
        """Berechnet Volume-Synchronized Probability of Informed Trading."""
        if len(buckets) < 2:
            return 0.5
        
        volumes = [sum(t["volume"] for t in bucket) for bucket in buckets]
        buy_volumes = [sum(t["volume"] for t in bucket if t.get("side") == "buy") 
                       for bucket in buckets]
        
        vpin = sum(abs(v - bv) for v, bv in zip(volumes, buy_volumes)) / sum(volumes)
        return vpin if vpin > 0 else 0.5



============================================================


NUTZUNGSBEISPIEL


============================================================


if __name__ == "__main__":
    # API-Key aus Environment oder sicherer Quelle laden
    API_KEY = "YOUR_HOLYSHEEP_API_KEY"  # Ersetzen Sie mit echtem Key
    
    # Scanner initialisieren
    scanner = OrderflowScanner(
        api_key=API_KEY,
        pairs=["BTC/USDT", "ETH/USDT", "SOL/USDT"]
    )
    
    # Beispiel: Simulation einer Orderflow-Situation
    demo_data = {
        "BTC/USDT": {
            "ofi": 0.35,
            "vpin": 0.62,
            "bba_imbalance": 0.28,
            "large_trade_ratio": 0.15,
            "recent_trades_summary": "Erhöhtes Buy-Volume bei $67.200 Support. 3 Large-Buys über 5 BTC."
        }
    }
    
    # Signal generieren
    signal_gen = OrderflowSignalGenerator(API_KEY)
    result = signal_gen.generate_signal(
        market_data={"symbol": "BTC/USDT", **demo_data["BTC/USDT"]},
        model="gemini-2.5-flash"  # $2.50/MTok
    )
    
    print(f"Signal für BTC/USDT:")
    print(f"Status: {result['success']}")
    if result['success']:
        print(f"Modell: {result['model_used']}")
        print(f"Latenz: {result['latency_ms']}ms")
        print(f"Kosten: ${result['cost_usd']:.6f}")
        print(f"Signal: {json.dumps(result['signal'], indent=2)}")

Geeignet / nicht geeignet für

Geeignet für Nicht geeignet für
Quant-Trading-Teams mit Orderflow-Erfahrung Manuelle Trader ohne Programmierkenntnisse
HFT-Firmen mit Latenz-Anforderungen <100ms Langfristige Positionstrader (Swing/Position)
Algorithmic Trading Shops mit skalierbarem Bedarf Einzelhändler mit <$10k Kontostand
Research-Teams für Marktmikrostruktur-Studien Personen ohne Verständnis von Risikomanagement
Crypto-Fonds mit institutionellen Anforderungen Regulierte Finanzinstitutionen (Compliance-Einschränkungen)

Preise und ROI

Modell Preis pro MTok Use Case Sparen vs. OpenAI
DeepSeek V3.2 $0.42 Batch-Analyse, Backtesting, Research 92% günstiger
Gemini 2.5 Flash $2.50 Echtzeit-Signale, Produktion (Empfohlen) 75% günstiger
GPT-4.1 $8.00 Hochpräzise Analyse, komplexe Muster 20% günstiger
Claude Sonnet 4.5 $15.00 Edge-Fälle, nuancierte Marktanalyse 25% günstiger

ROI-Analyse für Orderflow-Trading

Basierend auf meinen Tests mit einem Orderflow-Signal-Generator:

Der Wechsel zu HolySheep amortisiert sich ab dem ersten produktiven Tag.

Warum HolySheep wählen

Häufige Fehler und Lösungen

Fehler 1: Timeout ohne Retry-Logik

# ❌ FEHLERHAFT: Kein Error-Handling
response = requests.post(url, json=payload)
result = response.json()  # Crashed bei Timeout!


✅ LÖSUNG: Robustes Retry-Pattern mit Exponential Backoff


import time
import requests
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry

def robust_api_call(base_url: str, payload: dict, api_key: str, max_retries: int = 3) -> dict:
    """
    Führt API-Call mit automatischer Retry-Logik durch.
    
    Strategy: Exponential Backoff
    - Retry 1: 1 Sekunde warten
    - Retry 2: 2 Sekunden warten  
    - Retry 3: 4 Sekunden warten
    """
    
    session = requests.Session()
    
    # Retry-Adapter konfigurieren
    retry_strategy = Retry(
        total=max_retries,
        backoff_factor=1,
        status_forcelist=[429, 500, 502, 503, 504],
        allowed_methods=["POST"]
    )
    
    adapter = HTTPAdapter(max_retries=retry_strategy)
    session.mount("https://", adapter)
    
    headers = {
        "Authorization": f"Bearer {api_key}",
        "Content-Type": "application/json"
    }
    
    last_error = None
    
    for attempt in range(max_retries + 1):
        try:
            response = session.post(
                f"{base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=(10, 30)  # (connect_timeout, read_timeout)
            )
            
            if response.status_code == 200:
                return {"success": True, "data": response.json()}
            elif response.status_code == 429:
                # Rate-Limit erreicht
                retry_after = int(response.headers.get("Retry-After", 5))
                print(f"Rate-Limit erreicht. Warte {retry_after}s...")
                time.sleep(retry_after)
                continue
            else:
                return {
                    "success": False,
                    "error": f"HTTP {response.status_code}",
                    "data": response.text
                }
                
        except requests.exceptions.Timeout:
            last_error = "Timeout nach 30s"
            print(f"Versuch {attempt + 1}/{max_retries + 1} fehlgeschlagen: Timeout")
            
        except requests.exceptions.ConnectionError as e:
            last_error = f"Verbindungsfehler: {str(e)}"
            print(f"Versuch {attempt + 1}/{max_retries + 1} fehlgeschlagen: Connection Error")
            
        except Exception as e:
            last_error = f"Unerwarteter Fehler: {str(e)}"
            print(f"Versuch {attempt + 1}/{max_retries + 1} fehlgeschlagen: {e}")
        
        # Exponential Backoff nur bei wiederholbaren Fehlern
        if attempt < max_retries:
            wait_time = 2 ** attempt
            print(f"Warzte {wait_time}s vor Retry...")
            time.sleep(wait_time)
    
    return {
        "success": False,
        "error": last_error,
        "retries": max_retries
    }



Nutzung:

result = robust_api_call(
    base_url="https://api.holysheep.ai/v1",
    payload=payload,
    api_key="YOUR_HOLYSHEEP_API_KEY",
    max_retries=3
)

Fehler 2: Falsches Modell für Echtzeit-Trading

# ❌ FEHLERHAFT: teures Modell für einfache Signale
response = client.chat.completions.create(
    model="claude-sonnet-4.5",  # $15/MTok - Verschwendung!
    messages=[...],
    temperature=0.7,
    max_tokens=2000
)


✅ LÖSUNG: Modell-Selection basierend auf Use-Case


class ModelSelector:
    """Optimiert Modellauswahl basierend auf Anwendungsfall und Budget."""
    
    MODELS = {
        "realtime_signal": {
            "primary": "gemini-2.5-flash",    # $2.50/MTok
            "fallback": "deepseek-v3.2",      # $0.42/MTok
            "max_latency_ms": 2000
        },
        "deep_analysis": {
            "primary": "gpt-4.1",             # $8/MTok
            "fallback": "gemini-2.5-flash",
            "max_latency_ms": 10000
        },
        "backtesting": {
            "primary": "deepseek-v3.2",       # $0.42/MTok - MAXIMAL BILLIG
            "fallback": None,
            "max_latency_ms": 5000
        },
        "sentiment_check": {
            "primary": "gemini-2.5-flash",
            "fallback": "deepseek-v3.2",
            "max_latency_ms": 3000
        }
    }
    
    @classmethod
    def get_model(cls, use_case: str, prefer_speed: bool = True) -> str:
        """
        Gibt optimales Modell für Anwendungsfall zurück.
        
        Args:
            use_case: Einer der definierten Use-Cases
            prefer_speed: Wenn True, wähle schnellstes Modell
        
        Returns:
            Modellname
        """
        config = cls.MODELS.get(use_case, cls.MODELS["realtime_signal"])
        
        if prefer_speed:
            return config["primary"]
        else:
            # Für maximale Kostenersparnis
            return config.get("fallback", config["primary"])
    
    @classmethod
    def estimate_cost(cls, use_case: str, input_tokens: int, output_tokens: int) -> float:
        """Schätzt Kosten für API-Call."""
        model = cls.get_model(use_case)
        
        costs = {
            "gpt-4.1": 8.0,
            "claude-sonnet-4.5": 15.0,
            "gemini-2.5-flash": 2.5,
            "deepseek-v3.2": 0.42
        }
        
        price = costs.get(model, 8.0)
        total_tokens = input_tokens + output_tokens
        
        return (total_tokens / 1_000_000) * price



Nutzung im Orderflow-Scanner:



Für Echtzeit-Signal (schnell + günstig):

signal_model = ModelSelector.get_model("realtime_signal", prefer_speed=True)

→ "gemini-2.5-flash"



Für Batch-Backtest (maximale Ersparnis):

backtest_model = ModelSelector.get_model("backtesting", prefer_speed=False)

→ "deepseek-v3.2"



Kosten schätzen:

cost = ModelSelector.estimate_cost("realtime_signal", 800, 200)
print(f"Geschätzte Kosten pro Signal: ${cost:.6f}")

Fehler 3: Verlorene Signale durch fehlendes Caching

# ❌ FEHLERHAFT: Keine Deduplizierung, teure wiederholte Calls
def get_signal(pair: str, timeframe: str):
    # Jeder Aufruf -> neuer API-Call = Geldverschwendung
    return call_holysheep_api(pair, timeframe)


Bei 1000 Anfragen/sec für gleiches Pair = 1000× API-Kosten!



✅ LÖSUNG: Intelligentes Cache-System mit Invalidierung


from datetime import datetime, timedelta
from typing import Optional
import hashlib
import json

class OrderflowSignalCache:
    """
    Memory-Cache für Orderflow-Signale mit TTL-basierter Invalidierung.
    Verhindert doppelte API-Calls für identische Anfragen.
    """
    
    def __init__(self, default_ttl_seconds: int = 30):
        self._cache = {}
        self._timestamps = {}
        self.default_ttl = default_ttl_seconds
        
        # Statistic
        self.hits = 0
        self.misses = 0
    
    def _make_key(self, pair: str, features: dict) -> str:
        """Erstellt eindeutigen Cache-Key aus Request-Parametern."""
        # Nur relevante Merkmale für Hashing
        cache_data = {
            "pair": pair,
            "ofi_bucket": round(features.get("ofi", 0), 2),  # Runden für weniger Variation
            "vpin_bucket": round(features.get("vpin", 0), 2),
            "trend": features.get("trend", "neutral")
        }
        serialized = json.dumps(cache_data, sort_keys=True)
        return hashlib.sha256(serialized.encode()).hexdigest()[:16]
    
    def get(self, pair: str, features: dict) -> Optional[dict]:
        """
        Gibt gecachtes Signal zurück, falls vorhanden und valid.
        
        Returns:
            Cached Signal oder None
        """
        key = self._make_key(pair, features)
        
        if key in self._cache:
            # TTL-Prüfung
            age = (datetime.now() - self._timestamps[key]).total_seconds()
            
            if age < self.default_ttl:
                self.hits += 1
                return self._cache[key]
            else:
                # Expired - aufräumen
                del self._cache[key]
                del self._timestamps[key]
        
        self.misses += 1
        return None
    
    def set(self, pair: str, features: dict, signal: dict, ttl: Optional[int] = None):
        """Speichert Signal im Cache."""
        key = self._make_key(pair, features)
        self._cache[key] = signal
        self._timestamps[key] = datetime.now()
    
    def get_stats(self) -> dict:
        """Gibt Cache-Statistiken zurück."""
        total = self.hits + self.misses
        hit_rate = (self.hits / total * 100) if total > 0 else 0
        
        return {
            "hits": self.hits,
            "misses": self.misses,
            "hit_rate":

Verwandte Ressourcen
Verwandte Artikel