In der Welt des algorithmischen Handels und der quantitativen Finanzanalyse ist das historische Orderbook-Replay eine der faszinierendsten Techniken, die es ermöglicht, vergangene Marktbedingungen originalgetreu zu simulieren. In diesem umfassenden Tutorial zeige ich Ihnen Schritt für Schritt, wie Sie mit Python ein vollständiges System zur Orderbuch-Rekonstruktion und simulierten Orderausführung aufbauen.

Was Sie in diesem Tutorial lernen werden:

Was ist ein Orderbook und warum ist die Rekonstruktion wichtig?

Ein Orderbook (auch Auftragsbuch genannt) ist die digitale Darstellung aller aktiven Kauf- und Verkaufsaufträge für ein bestimmtes Handelspaar zu einem bestimmten Zeitpunkt. Stellen Sie es sich wie eine digitale Auktion vor:

Die Rekonstruktion historischer Orderbooks ermöglicht es Ihnen:

Grundlagen: Tardis-API und Datenformate verstehen

Bevor wir mit dem Code beginnen, müssen wir verstehen, wie Tardis.history (ein führender Anbieter für historische Marktdaten) seine Daten strukturiert. Die API liefert verschiedene Nachrichtentypen, die wir für die Orderbook-Rekonstruktion benötigen.

Die wichtigsten Nachrichtentypen

Python-Umgebung einrichten

Zunächst richten wir unsere Entwicklungsumgebung ein. Ich empfehle die Verwendung einer virtuellen Umgebung, um Abhängigkeiten sauber zu verwalten.

# Virtuelle Umgebung erstellen und aktivieren
python -m venv orderbook_env
source orderbook_env/bin/activate  # Linux/Mac

orderbook_env\Scripts\activate  # Windows



Notwendige Pakete installieren

pip install pandas numpy requests asyncio aiohttp
pip install holy-sheep-sdk  # HolySheep AI SDK


Überprüfen der Installation

python -c "import pandas; print('Pandas Version:', pandas.__version__)"
python -c "import numpy; print('NumPy Version:', numpy.__version__)"

HolySheep AI: Kostengünstige KI-Unterstützung für Ihre Analyse

Bevor wir tief in die Implementierung eintauchen, möchte ich Ihnen Jetzt registrieren bei HolySheep AI empfehlen. Diese Plattform bietet Zugang zu hochwertigen KI-Modellen zu einem Bruchteil der Kosten anderer Anbieter:

Mit der Integration von WeChat und Alipay ist die Bezahlung für chinesische Nutzer besonders einfach, und die Latenz liegt unter 50ms bei minimaler Antwortzeit.

Implementierung: Schritt für Schritt zum funktionierenden System

Schritt 1: Daten von Tardis.history abrufen

Zunächst benötigen wir historische Marktdaten. Tardis.history bietet eine umfangreiche API für den Zugriff auf historische Orderbook-Daten von über 50 Kryptowährungsbörsen.

import requests
import pandas as pd
from datetime import datetime, timedelta
import json


Tardis API Konfiguration

TARDIS_API_KEY = "your_tardis_api_key"
EXCHANGE = "binance"  # Unterstützte Börsen: binance, okex, bybit, etc.
SYMBOL = "BTC-USDT"
START_DATE = "2024-01-01"
END_DATE = "2024-01-02"

def fetch_tardis_data(exchange, symbol, start_date, end_date):
    """
    Ruft historische Marktdaten von Tardis.history ab.
    
    Args:
        exchange: Name der Börse (z.B. 'binance', 'okex')
        symbol: Handelspaar (z.B. 'BTC-USDT')
        start_date: Startdatum im Format 'YYYY-MM-DD'
        end_date: Enddatum im Format 'YYYY-MM-DD'
    
    Returns:
        DataFrame mit historischen Marktdaten
    """
    url = f"https://api.tardis.dev/v1/converters/{exchange}/{symbol}"
    
    params = {
        "from": start_date,
        "to": end_date,
        "format": "json",
        "types": "orderbook_snapshot,orderbook_update,trade"
    }
    
    headers = {
        "Authorization": f"Bearer {TARDIS_API_KEY}"
    }
    
    response = requests.get(url, params=params, headers=headers)
    response.raise_for_status()
    
    data = response.json()
    return pd.DataFrame(data)


Beispielaufruf

try:
    df = fetch_tardis_data(EXCHANGE, SYMBOL, START_DATE, END_DATE)
    print(f"Erfolgreich {len(df)} Datensätze abgerufen")
    print(df.head())
except requests.exceptions.RequestException as e:
    print(f"API-Fehler: {e}")

Schritt 2: Orderbook-Klasse implementieren

Jetzt implementieren wir die Kernklasse für die Orderbook-Verwaltung. Diese Klasse wird alle Operationen zur Manipulation des Orderbooks durchführen.

from sortedcontainers import SortedDict
from dataclasses import dataclass, field
from typing import Dict, List, Tuple, Optional
from decimal import Decimal
import heapq

@dataclass
class Order:
    """Repräsentiert eine einzelne Order im Orderbook."""
    price: float
    quantity: float
    order_id: str
    side: str  # 'bid' oder 'ask'
    
    def __lt__(self, other):
        """Comparison für Heap-Operationen."""
        if self.side == 'bid':
            return self.price > other.price  # Max-Heap für Bids
        else:
            return self.price < other.price  # Min-Heap für Asks

class Orderbook:
    """
    Implementiert ein vollständiges Orderbook mit Unterstützung für
    Limit- und Market-Orders, verschiedene Ordertypen und 
    simulierte Orderausführung.
    """
    
    def __init__(self, symbol: str):
        self.symbol = symbol
        self.bids = SortedDict()  # Preis -> Liste von Orders
        self.asks = SortedDict()   # Preis -> Liste von Orders
        self.trades = []           # Historie aller Trades
        self.sequence = 0         # Sequenznummer für Updates
        
    def apply_snapshot(self, snapshot: Dict):
        """
        Wendet einen vollständigen Orderbook-Snapshot an.
        
        Args:
            snapshot: Dictionary mit 'bids' und 'asks' Listen
                     Format: [[price, quantity], ...]
        """
        self.bids.clear()
        self.asks.clear()
        
        # Bids verarbeiten (absteigend sortiert)
        for price, quantity in snapshot.get('bids', []):
            if quantity > 0:
                self.bids[price] = quantity
                
        # Asks verarbeiten (aufsteigend sortiert)
        for price, quantity in snapshot.get('asks', []):
            if quantity > 0:
                self.asks[price] = quantity
                
        self.sequence += 1
        
    def apply_update(self, update: Dict):
        """
        Wendet eine partielle Orderbook-Änderung (Delta) an.
        
        Args:
            update: Dictionary mit Änderungen
                   Format: {'bids': [[price, quantity], ...], 'asks': [...]}
        """
        # Bids aktualisieren
        for price, quantity in update.get('bids', []):
            if quantity == 0:
                self.bids.pop(price, None)
            else:
                self.bids[price] = quantity
                
        # Asks aktualisieren
        for price, quantity in update.get('asks', []):
            if quantity == 0:
                self.asks.pop(price, None)
            else:
                self.asks[price] = quantity
                
        self.sequence += 1
        
    def add_order(self, order: Order) -> List[Tuple]:
        """
        Fügt eine neue Order hinzu und führt sie gegen bestehende Orders aus.
        
        Args:
            order: Die hinzuzufügende Order
            
        Returns:
            Liste von ausgeführten Trades als Tuple (price, quantity, side)
        """
        executed_trades = []
        
        if order.side == 'bid':
            # Kauforder matcht gegen Asks (niedrigste zuerst)
            while order.quantity > 0 and order.price >= self.get_best_ask():
                best_ask_price = self.get_best_ask()
                if best_ask_price is None:
                    break
                    
                available_qty = self.asks.get(best_ask_price, 0)
                fill_qty = min(order.quantity, available_qty)
                
                # Trade ausführen
                trade = (best_ask_price, fill_qty, 'buy')
                executed_trades.append(trade)
                self.trades.append(trade)
                
                # Orderbook aktualisieren
                if fill_qty == available_qty:
                    del self.asks[best_ask_price]
                else:
                    self.asks[best_ask_price] = available_qty - fill_qty
                    
                order.quantity -= fill_qty
                
            # Restliche Order als Limit-Order hinzufügen
            if order.quantity > 0:
                current_qty = self.bids.get(order.price, 0)
                self.bids[order.price] = current_qty + order.quantity
                
        else:  # ask (Verkauf)
            # Verkaufsorder matcht gegen Bids (höchsten zuerst)
            while order.quantity > 0 and order.price <= self.get_best_bid():
                best_bid_price = self.get_best_bid()
                if best_bid_price is None:
                    break
                    
                available_qty = self.bids.get(best_bid_price, 0)
                fill_qty = min(order.quantity, available_qty)
                
                # Trade ausführen
                trade = (best_bid_price, fill_qty, 'sell')
                executed_trades.append(trade)
                self.trades.append(trade)
                
                # Orderbook aktualisieren
                if fill_qty == available_qty:
                    del self.bids[best_bid_price]
                else:
                    self.bids[best_bid_price] = available_qty - fill_qty
                    
                order.quantity -= fill_qty
                
            # Restliche Order als Limit-Order hinzufügen
            if order.quantity > 0:
                current_qty = self.asks.get(order.price, 0)
                self.asks[order.price] = current_qty + order.quantity
                
        return executed_trades
    
    def get_best_bid(self) -> Optional[float]:
        """Gibt den höchsten Gebotspreis zurück."""
        if self.bids:
            return self.bids.keys()[-1]  # SortedDict ist aufsteigend
        return None
    
    def get_best_ask(self) -> Optional[float]:
        """Gibt den niedrigsten Verkaufspreis zurück."""
        if self.asks:
            return self.asks.keys()[0]
        return None
    
    def get_spread(self) -> Optional[float]:
        """Berechnet den aktuellen Spread."""
        best_bid = self.get_best_bid()
        best_ask = self.get_best_ask()
        if best_bid and best_ask:
            return best_ask - best_bid
        return None
    
    def get_mid_price(self) -> Optional[float]:
        """Berechnet den mittleren Preis (Mid-Price)."""
        best_bid = self.get_best_bid()
        best_ask = self.get_best_ask()
        if best_bid and best_ask:
            return (best_bid + best_ask) / 2
        return None
    
    def get_depth(self, levels: int = 10) -> Dict:
        """
        Gibt die Markttiefe für eine bestimmte Anzahl von Preisstufen zurück.
        
        Args:
            levels: Anzahl der Preisstufen
            
        Returns:
            Dictionary mit 'bids' und 'asks' Listen
        """
        bids = [(price, qty) for price, qty in list(self.bids.items())[-levels:]]
        asks = [(price, qty) for price, qty in list(self.asks.items())[:levels]]
        
        return {'bids': bids, 'asks': asks}
    
    def visualize(self) -> str:
        """
        Erstellt eine textuelle Visualisierung des Orderbooks.
        
        Returns:
            String-Darstellung des Orderbooks
        """
        output = [f"\n{'='*60}"]
        output.append(f"Orderbook für {self.symbol} (Sequence: {self.sequence})")
        output.append(f"Best Bid: {self.get_best_bid()} | Best Ask: {self.get_best_ask()}")
        output.append(f"Spread: {self.get_spread():.2f} ({self.get_spread()/self.get_mid_price()*100:.3f}%)")
        output.append(f"{'='*60}")
        
        # Top 5 Bids
        output.append("\n📈 BIDS (Kaufaufträge):")
        output.append("-" * 40)
        top_bids = list(self.bids.items())[-5:][::-1]
        for price, qty in top_bids:
            bar = "█" * int(qty / 0.5)  # Visuelle Darstellung
            output.append(f"  {price:>10.2f} | {qty:>10.4f} {bar}")
            
        # Top 5 Asks
        output.append("\n📉 ASKS (Verkaufsaufträge):")
        output.append("-" * 40)
        top_asks = list(self.asks.items())[:5]
        for price, qty in top_asks:
            bar = "█" * int(qty / 0.5)
            output.append(f"  {price:>10.2f} | {qty:>10.4f} {bar}")
            
        return "\n".join(output)


Beispiel: Orderbook instanziieren und testen

ob = Orderbook("BTC-USDT")


Simulierten Snapshot setzen

test_snapshot = {
    'bids': [
        [50000.0, 1.5],
        [49999.5, 2.0],
        [49999.0, 3.0],
        [49998.0, 1.0],
        [49997.0, 0.5]
    ],
    'asks': [
        [50001.0, 1.0],
        [50001.5, 2.5],
        [50002.0, 1.5],
        [50002.5, 3.0],
        [50003.0, 2.0]
    ]
}

ob.apply_snapshot(test_snapshot)
print(ob.visualize())

Schritt 3: Replay-Engine für historische Daten

Jetzt implementieren wir die Replay-Engine, die historische Daten sequenziell abspielt und dabei alle Orderbook-Änderungen und Trades korrekt verarbeitet.

import heapq
from dataclasses import dataclass, field
from typing import Iterator, Callable, Optional, List
from datetime import datetime
from enum import Enum

class MessageType(Enum):
    SNAPSHOT = "snapshot"
    UPDATE = "update"
    TRADE = "trade"

@dataclass(order=True)
class MarketMessage:
    """Eine Nachricht im Datenstream mit Zeitstempel."""
    timestamp: datetime = field(compare=True)
    message_type: MessageType = field(compare=False)
    data: dict = field(compare=False)
    exchange: str = field(compare=False)
    symbol: str = field(compare=False)
    
class OrderbookReplayEngine:
    """
    Engine für das Replay historischer Orderbook-Daten.
    
    Features:
    - Sequenzielle Verarbeitung von Snapshots und Updates
    - Zeitlich korrekte Orderausführung
    - Callbacks für Trades, Orderbook-Änderungen
    - Geschwindigkeitskontrolle (Echtzeit vs. beschleunigt)
    """
    
    def __init__(self, symbol: str, on_trade: Optional[Callable] = None,
                 on_orderbook_change: Optional[Callable] = None):
        self.symbol = symbol
        self.orderbook = Orderbook(symbol)
        self.on_trade = on_trade
        self.on_orderbook_change = on_orderbook_change
        self.trades = []
        self.orderbook_states = []
        
    def process_messages(self, messages: Iterator[MarketMessage]):
        """
        Verarbeitet einen Stream von Marktnachrichten.
        
        Args:
            messages: Iterator von MarketMessage-Objekten
        """
        for message in messages:
            self._process_single_message(message)
            
    def _process_single_message(self, message: MarketMessage):
        """Verarbeitet eine einzelne Marktnachricht."""
        if message.message_type == MessageType.SNAPSHOT:
            self.orderbook.apply_snapshot(message.data)
            
        elif message.message_type == MessageType.UPDATE:
            self.orderbook.apply_update(message.data)
            
        elif message.message_type == MessageType.TRADE:
            trade_data = {
                'timestamp': message.timestamp,
                'price': message.data.get('price'),
                'quantity': message.data.get('quantity'),
                'side': message.data.get('side'),
                'trade_id': message.data.get('id')
            }
            self.trades.append(trade_data)
            
            if self.on_trade:
                self.on_trade(trade_data)
                
        # Callback für Orderbook-Änderungen
        if self.on_orderbook_change:
            state = {
                'timestamp': message.timestamp,
                'orderbook': self.orderbook.get_depth(20)
            }
            self.orderbook_states.append(state)
            self.on_orderbook_change(state)
            
    def replay_with_speed(self, messages: Iterator[MarketMessage], 
                          speed: float = 1.0, max_messages: Optional[int] = None):
        """
        Replay mit kontrollierter Geschwindigkeit.
        
        Args:
            messages: Iterator von Marktnachrichten
            speed: Geschwindigkeitsfaktor (1.0 = Echtzeit, 2.0 = doppelt so schnell)
            max_messages: Maximale Anzahl zu verarbeitender Nachrichten
        """
        import time
        
        count = 0
        last_timestamp = None
        
        for message in messages:
            # Geschwindigkeit kontrollieren
            if last_timestamp and speed > 0:
                elapsed = (message.timestamp - last_timestamp).total_seconds()
                sleep_time = elapsed / speed
                if sleep_time > 0:
                    time.sleep(min(sleep_time, 1.0))  # Max 1 Sekunde warten
                    
            self._process_single_message(message)
            last_timestamp = message.timestamp
            count += 1
            
            if max_messages and count >= max_messages:
                break
                
        print(f"Replay abgeschlossen: {count} Nachrichten verarbeitet")
        
    def analyze_spread(self) -> dict:
        """
        Analysiert den Spread-Verlauf während des Replays.
        
        Returns:
            Dictionary mit Spread-Statistiken
        """
        spreads = []
        mid_prices = []
        
        for state in self.orderbook_states:
            ob = Orderbook(self.symbol)
            ob.apply_snapshot(state['orderbook'])
            spreads.append(ob.get_spread())
            mid_prices.append(ob.get_mid_price())
            
        return {
            'avg_spread': sum(spreads) / len(spreads) if spreads else 0,
            'max_spread': max(spreads) if spreads else 0,
            'min_spread': min(spreads) if spreads else 0,
            'avg_mid_price': sum(mid_prices) / len(mid_prices) if mid_prices else 0,
            'total_trades': len(self.trades)
        }
        
    def detect_large_trades(self, threshold: float) -> List[dict]:
        """
        Erkennt große Trades (Whale Trades).
        
        Args:
            threshold: Mindestvolumen für "große" Trades
            
        Returns:
            Liste der großen Trades
        """
        return [t for t in self.trades if t['quantity'] >= threshold]


Callback-Funktionen für das Replay

def on_trade_callback(trade):
    """Wird bei jedem Trade während des Replays aufgerufen."""
    print(f"Trade ausgeführt: {trade['price']} @ {trade['quantity']}")

def on_orderbook_change_callback(state):
    """Wird bei jeder Orderbook-Änderung aufgerufen."""
    pass  # Für Performance-Tests kann hier Logging deaktiviert werden


Engine instanziieren

replay_engine = OrderbookReplayEngine(
    symbol="BTC-USDT",
    on_trade=on_trade_callback,
    on_orderbook_change=on_orderbook_change_callback
)

Schritt 4: Integration mit HolySheep AI für KI-gestützte Analyse

Ein besonderes Feature dieses Systems ist die Integration mit HolySheep AI für KI-gestützte Marktanalyse. Sie können die leistungsstarken Sprachmodelle nutzen, um automatisch Handelssignale zu generieren oder Muster in den Orderbook-Daten zu erkennen.

import requests
import json
from typing import Dict, List, Optional

class HolySheepAIIntegration:
    """
    Integration mit HolySheep AI für KI-gestützte Orderbook-Analyse.
    
    Nutzt fortschrittliche Sprachmodelle zur:
    - Erkennung von Handelsmustern
    - Vorhersage von Preisbewegungen
    - Sentiment-Analyse des Marktes
    """
    
    def __init__(self, api_key: str):
        self.api_key = api_key
        self.base_url = "https://api.holysheep.ai/v1"
        self.model = "deepseek-v3.2"  # Kostengünstigste Option: $0.42/MTok
        
    def analyze_orderbook(self, orderbook_state: Dict, 
                          market_context: str) -> Dict:
        """
        Analysiert den aktuellen Orderbook-Zustand mit KI.
        
        Args:
            orderbook_state: Aktueller Orderbook-Inhalt
            market_context: Zusätzlicher Marktkontext
            
        Returns:
            KI-Analyseergebnis mit Handelssignalen
        """
        prompt = f"""
        Analysiere den folgenden Orderbook-Zustand für {orderbook_state.get('symbol', 'BTC-USDT')}:
        
        Top 5 Bids (Kaufaufträge):
        {json.dumps(orderbook_state.get('bids', [])[:5], indent=2)}
        
        Top 5 Asks (Verkaufsaufträge):
        {json.dumps(orderbook_state.get('asks', [])[:5], indent=2)}
        
        Marktkontext: {market_context}
        
        Bitte analysiere:
        1. Aktuelle Marktsituation (bullish/bearish/neutral)
        2. Unterstützungs- und Widerstandsniveaus
        3. Wahrscheinliche kurzfristige Preisbewegung
        4. Risikobewertung
        
        Antworte im JSON-Format mit den Feldern:
        - sentiment: String (bullish/bearish/neutral)
        - support_level: Float
        - resistance_level: Float
        - predicted_direction: String (up/down/sideways)
        - confidence: Float (0-1)
        - recommendation: String (buy/sell/hold)
        """
        
        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": "Du bist ein erfahrener Krypto-Marktanalyst."},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.3,
            "max_tokens": 500
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            
            result = response.json()
            content = result['choices'][0]['message']['content']
            
            # Parse JSON aus der Antwort
            analysis = json.loads(content)
            return analysis
            
        except requests.exceptions.RequestException as e:
            return {"error": str(e), "fallback": "Manual analysis required"}
            
    def generate_trading_signals(self, replay_stats: Dict) -> str:
        """
        Generiert Handelssignale basierend auf Replay-Statistiken.
        
        Args:
            replay_stats: Statistiken aus dem Orderbook-Replay
            
        Returns:
            Zusammenfassung der Handelssignale
        """
        prompt = f"""
        Basierend auf folgenden Replay-Statistiken für Bitcoin:
        
        Durchschnittlicher Spread: ${replay_stats.get('avg_spread', 0):.2f}
        Maximale Spread: ${replay_stats.get('max_spread', 0):.2f}
        Anzahl Trades: {replay_stats.get('total_trades', 0)}
        Durchschnittspreis: ${replay_stats.get('avg_mid_price', 0):.2f}
        
        Generiere eine kurze Handelszusammenfassung mit:
        1. Marktcharakteristik (volatil/stabil)
        2. Empfohlene Strategie
        3. Risikohinweise
        """
        
        payload = {
            "model": self.model,
            "messages": [
                {"role": "system", "content": "Du bist ein quantitativer Handelsberater."},
                {"role": "user", "content": prompt}
            ],
            "temperature": 0.5,
            "max_tokens": 300
        }
        
        headers = {
            "Authorization": f"Bearer {self.api_key}",
            "Content-Type": "application/json"
        }
        
        try:
            response = requests.post(
                f"{self.base_url}/chat/completions",
                headers=headers,
                json=payload,
                timeout=30
            )
            response.raise_for_status()
            
            result = response.json()
            return result['choices'][0]['message']['content']
            
        except requests.exceptions.RequestException as e:
            return f"Fehler bei der Signalanalyse: {str(e)}"


Beispiel-Nutzung


api_key = "YOUR_HOLYSHEEP_API_KEY"  # Ersetzen Sie mit Ihrem echten Key


holy_sheep = HolySheepAIIntegration(api_key)

# 

orderbook_sample = {


    'symbol': 'BTC-USDT',


    'bids': [[50000, 1.5], [49999, 2.0]],


    'asks': [[50001, 1.0], [50002, 2.5]]


}

# 

analysis = holy_sheep.analyze_orderbook(


    orderbook_sample,


    market_context="Nachricht von positiven ETF-Zulassungen"


)


print(analysis)

Komplettes Anwendungsbeispiel: End-to-End Orderbook-Replay

Hier ist ein vollständiges, ausführbares Beispiel, das alle Komponenten zusammenführt.

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

KONFIGURATION


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


CONFIG = {
    # Tardis API (ersetzen Sie mit Ihrem echten API-Key)
    'tardis_api_key': 'YOUR_TARDIS_API_KEY',
    'exchange': 'binance',
    'symbol': 'BTC-USDT',
    
    # HolySheep AI API
    'holysheep_api_key': 'YOUR_HOLYSHEEP_API_KEY',
    
    # Replay-Parameter
    'start_date': '2024-01-15T00:00:00Z',
    'end_date': '2024-01-15T01:00:00Z',  # 1 Stunde Replay
    
    # Handelsparameter
    'initial_balance': 10000.0,  # USDT
    'position_size': 0.1,  # BTC pro Trade
}


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


SIMULATION EINES TRADE-EXECUTORS


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


class TradeExecutor:
    """
    Führt simulierte Trades während des Replays aus
    und verfolgt die Performance.
    """
    
    def __init__(self, initial_balance: float):
        self.balance = initial_balance
        self.position = 0.0  # BTC
        self.trades_log = []
        self.equity_curve = []
        
    def execute_buy(self, price: float, quantity: float, timestamp: datetime):
        """Simuliert einen Kauf."""
        cost = price * quantity
        if cost <= self.balance:
            self.balance -= cost
            self.position += quantity
            
            trade_record = {
                'timestamp': timestamp,
                'side': 'buy',
                'price': price,
                'quantity': quantity,
                'cost': cost,
                'balance': self.balance,
                'position': self.position
            }
            self.trades_log.append(trade_record)
            return True
        return False
        
    def execute_sell(self, price: float, quantity: float, timestamp: datetime):
        """Simuliert einen Verkauf."""
        if quantity <= self.position:
            revenue = price * quantity
            self.balance += revenue
            self.position -= quantity
            
            trade_record = {
                'timestamp': timestamp,
                'side': 'sell',
                'price': price,
                'quantity': quantity,
                'revenue': revenue,
                'balance': self.balance,
                'position': self.position
            }
            self.trades_log.append(trade_record)
            return True
        return False
        
    def get_equity(self, current_price: float) -> float:
        """Berechnet den aktuellen Gesamtwert."""
        return self.balance + (self.position * current_price)
        
    def get_performance_summary(self) -> Dict:
        """Erstellt eine Performance-Zusammenfassung."""
        if not self.trades_log:
            return {"error": "No trades executed"}
            
        buys = [t for t in self.trades_log if t['side'] == 'buy']
        sells = [t for t in self.trades_log if t['side'] == 'sell']
        
        total_buys = sum(t['cost'] for t in buys)
        total_sells = sum(t['revenue'] for t in sells)
        
        return {
            'total_trades': len(self.trades_log),
            'total_buys': len(buys),
            'total_sells': len(sells),
            'total_invested': total_buys,
            'total_revenue': total_sells,
            'net_pnl': total_sells - total_buys,
            'final_balance': self.balance,
            'final_position': self.position,
            'win_rate': len([t for i, t in enumerate(self.trades_log) 
                           if i > 0 and t['side'] == 'sell' 
                           and t['revenue'] > self.trades_log[i-1]['cost']]) / max(len(sells), 1)
        }


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


HAUPTFUNKTION


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


def run_complete_replay_simulation():
    """
    Führt die vollständige Replay-Simulation durch.
    
    Diese Funktion:
    1. Initialisiert alle Kom

Verwandte Ressourcen
Verwandte Artikel