In der Welt des algorithmischen Handels ist der Zugang zu hochqualitativen Marktdaten der Schlüssel zum Erfolg. Als erfahrener Market Maker habe ich jahrelang mit verschiedenen Datenanbietern gearbeitet und bin kürzlich auf eine besonders effiziente Lösung gestoßen: HolySheep AI. In diesem Tutorial zeige ich Ihnen, wie Sie Tardis AscendEX Tick-Daten in Ihre做市strategie (Market-Making-Strategie) integrieren und dabei gleichzeitig die盘口滑点 (Orderbook-Slippages) modellieren und die订单撮合 (Order-Matching)回放 praktisch umsetzen.
Kostenanalyse: LLM-Preise für Trading-Bots 2026
Bevor wir in die technische Implementierung einsteigen, möchte ich Ihnen die aktuellen Kosten für die Nutzung von Large Language Models in Ihren Trading-Bots vorstellen. Diese Zahlen sind entscheidend für Ihre ROI-Berechnung:
| Modell | Preis pro Million Token | Kosten für 10M Token/Monat | Relative Kosten |
|---|---|---|---|
| GPT-4.1 | $8,00 | $80,00 | 19× teurer als DeepSeek |
| Claude Sonnet 4.5 | $15,00 | $150,00 | 36× teurer als DeepSeek |
| Gemini 2.5 Flash | $2,50 | $25,00 | 6× teurer als DeepSeek |
| DeepSeek V3.2 | $0,42 | $4,20 | Benchmark |
Einsparpotenzial: Wenn Sie bisher GPT-4.1 für Ihre Trading-Logik verwendet haben und auf DeepSeek V3.2 umsteigen, sparen Sie bei 10 Millionen Token pro Monat stolze $75,80 — das sind 95% weniger Kosten bei gleicher Funktionalität für viele Trading-Anwendungen.
Voraussetzungen und Setup
- Python 3.9+ mit
pip - HolySheep AI API-Key (erhalten Sie hier Ihr kostenloses Startguthaben)
- Tardis.dev Konto für AscendEX raw data feed
- Grundlegendes Verständnis von Orderbook-Modellen
1. HolySheep AI Client-Setup für Trading-Anwendungen
Der erste Schritt ist die Konfiguration des HolySheep AI Clients. Mit ¥1 = $1 Wechselkurs und Unterstützung für WeChat/Alipay ist HolySheep besonders für asiatische Trader attraktiv. Die durchschnittliche Latenz unter 50ms macht es ideal für zeitkritische Market-Making-Anwendungen.
# holysheep_trading_client.py
import requests
import time
import json
from typing import Dict, List, Optional
from dataclasses import dataclass
from enum import Enum
class OrderSide(Enum):
BUY = "buy"
SELL = "sell"
@dataclass
class TradingSignal:
symbol: str
side: OrderSide
price: float
quantity: float
confidence: float
reasoning: str
class HolySheepTradingClient:
"""Client für HolySheep AI Trading-Integration mit DeepSeek V3.2"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session = requests.Session()
self.session.headers.update({
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
})
# Latenz-Metriken für Performance-Monitoring
self.latencies = []
self.total_tokens_used = 0
def analyze_market_conditions(
self,
orderbook: Dict,
recent_trades: List[Dict],
volatility: float
) -> TradingSignal:
"""Analysiert Marktbedingungen für Market-Making-Entscheidungen"""
# Prompt für DeepSeek V3.2 - optimiert für Trading
prompt = f"""Du bist ein erfahrener Market Maker für Krypto-Börsen.
Analysiere die folgenden Marktdaten und generiere ein Market-Making-Signal:
Orderbook-Daten:
Bid-Side: {json.dumps(orderbook['bids'][:5])}
Ask-Side: {json.dumps(orderbook['asks'][:5])}
Letzte Trades:
{json.dumps(recent_trades[:10])}
Volatilität: {volatility:.4f}
Gib eine JSON-Antwort mit:
- action: "bid", "ask", "both", oder "wait"
- price_adjustment: prozentuale Preisanpassung
- confidence: Konfidenzwert 0-1
- reasoning: kurze Begründung
Antworte NUR mit validem JSON."""
start_time = time.time()
try:
response = self.session.post(
f"{self.BASE_URL}/chat/completions",
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3, # Niedrig für konsistente Trading-Entscheidungen
"max_tokens": 500
},
timeout=5 # 5 Sekunden Timeout für Trading-Anwendungen
)
latency_ms = (time.time() - start_time) * 1000
self.latencies.append(latency_ms)
if response.status_code == 200:
data = response.json()
self.total_tokens_used += data.get('usage', {}).get('total_tokens', 0)
content = data['choices'][0]['message']['content']
# JSON parsen aus der Antwort
return self._parse_trading_signal(content, orderbook)
else:
print(f"API-Fehler: {response.status_code} - {response.text}")
return self._fallback_signal(orderbook)
except requests.exceptions.Timeout:
print("Timeout: HolySheep AI Antwort zu langsam")
return self._fallback_signal(orderbook)
except Exception as e:
print(f"Unerwarteter Fehler: {e}")
return self._fallback_signal(orderbook)
def _parse_trading_signal(self, content: str, orderbook: Dict) -> TradingSignal:
"""Parst das Trading-Signal aus der API-Antwort"""
try:
# Versuche JSON zu extrahieren
json_str = content.strip()
if json_str.startswith("```"):
json_str = json_str.split("```")[1]
if json_str.startswith("json"):
json_str = json_str[4:]
signal_data = json.loads(json_str)
mid_price = (float(orderbook['bids'][0][0]) + float(orderbook['asks'][0][0])) / 2
return TradingSignal(
symbol="ASCENDEX",
side=OrderSide.BUY if signal_data.get('action') == 'bid' else OrderSide.SELL,
price=mid_price * (1 + signal_data.get('price_adjustment', 0) / 100),
quantity=signal_data.get('quantity', 0.001),
confidence=signal_data.get('confidence', 0.5),
reasoning=signal_data.get('reasoning', 'Fallback')
)
except Exception as e:
print(f"Parsing-Fehler: {e}")
return self._fallback_signal(orderbook)
def _fallback_signal(self, orderbook: Dict) -> TradingSignal:
"""Fallback-Signal bei API-Fehlern - für resiliente Systeme"""
mid_price = (float(orderbook['bids'][0][0]) + float(orderbook['asks'][0][0])) / 2
return TradingSignal(
symbol="ASCENDEX",
side=OrderSide.BUY,
price=mid_price * 0.999,
quantity=0.001,
confidence=0.1,
reasoning="Fallback wegen API-Fehler"
)
def get_performance_stats(self) -> Dict:
"""Gibt Performance-Statistiken zurück"""
avg_latency = sum(self.latencies) / len(self.latencies) if self.latencies else 0
return {
"avg_latency_ms": round(avg_latency, 2),
"min_latency_ms": round(min(self.latencies), 2) if self.latencies else 0,
"max_latency_ms": round(max(self.latencies), 2) if self.latencies else 0,
"total_requests": len(self.latencies),
"total_tokens": self.total_tokens_used,
"estimated_cost_DeepSeek": round(self.total_tokens_used / 1_000_000 * 0.42, 4)
}
Initialisierung
client = HolySheepTradingClient(api_key="YOUR_HOLYSHEEP_API_KEY")
print("✅ HolySheep AI Trading Client initialisiert")2. Tardis AscendEX Tick-Daten-Integration
Jetzt integrieren wir den Tardis AscendEX Feed für Echtzeit-Tick-Daten. Tardis liefert die rohen Exchange-WebSocket-Streams mit allen Orderbuch-Updates und Trades.
# tardis_ascendex_client.py
import asyncio
import json
import zlib
import struct
from typing import Dict, List, Callable, Optional
from dataclasses import dataclass, field
from collections import defaultdict
import time
@dataclass
class OrderbookLevel:
price: float
quantity: float
@dataclass
class Orderbook:
symbol: str
bids: List[OrderbookLevel] = field(default_factory=list)
asks: List[OrderbookLevel] = field(default_factory=list)
last_update: float = field(default_factory=time.time)
def get_spread(self) -> float:
if self.bids and self.asks:
return self.asks[0].price - self.bids[0].price
return 0.0
def get_mid_price(self) -> float:
if self.bids and self.asks:
return (self.bids[0].price + self.asks[0].price) / 2
return 0.0
def get_depth(self, levels: int = 10) -> Dict:
"""Berechnet die Markttiefe für Volumensanalyse"""
bid_volume = sum(l.quantity for l in self.bids[:levels])
ask_volume = sum(l.quantity for l in self.asks[:levels])
return {
"bid_volume": bid_volume,
"ask_volume": ask_volume,
"imbalance": (bid_volume - ask_volume) / (bid_volume + ask_volume) if (bid_volume + ask_volume) > 0 else 0,
"bid_volume_ma5": bid_volume, # Simplified für Demo
"ask_volume_ma5": ask_volume
}
@dataclass
class Trade:
symbol: str
side: str # "buy" oder "sell"
price: float
quantity: float
timestamp: int
trade_id: str
class TardisAscendEXClient:
"""Client für Tardis AscendEX WebSocket-Feed mit Orderbook-Rekonstruktion"""
ASCENDEX_WS_URL = "wss://tardis-ascendex.exchange:9443"
def __init__(self, tardis_token: str):
self.tardis_token = tardis_token
self.orderbooks: Dict[str, Orderbook] = {}
self.recent_trades: Dict[str, List[Trade]] = defaultdict(list)
self.callbacks: List[Callable] = []
self.volatility: Dict[str, float] = {}
async def connect(self, symbols: List[str]):
"""Verbindet zum Tardis AscendEX WebSocket-Feed"""
import websockets
# Tardis authentication
auth_msg = {
"type": "auth",
"token": self.tardis_token
}
subscribe_msg = {
"type": "subscribe",
"channel": "raw",
"exchange": "ascendex",
"symbols": symbols
}
try:
async with websockets.connect(self.ASCENDEX_WS_URL) as ws:
await ws.send(json.dumps(auth_msg))
auth_response = await ws.recv()
print(f"Auth-Antwort: {auth_response}")
await ws.send(json.dumps(subscribe_msg))
print(f"✅ Subscribed zu: {symbols}")
async for message in ws:
await self._handle_message(message)
except Exception as e:
print(f"Verbindungsfehler: {e}")
raise
async def _handle_message(self, message: bytes):
"""Verarbeitet eingehende Tardis-Nachrichten"""
try:
# Dekomprimiere falls nötig
if message[0:2] == b'\x1f\x8b':
message = zlib.decompress(message)
data = json.loads(message)
# Verarbeite basierend auf Message-Typ
msg_type = data.get("type", "")
if msg_type == "data":
await self._process_data(data)
elif msg_type == "batched-data":
for item in data.get("data", []):
await self._process_data(item)
except Exception as e:
print(f"Nachrichtenverarbeitungsfehler: {e}")
async def _process_data(self, data: Dict):
"""Verarbeitet einzelne Daten-Updates"""
channel = data.get("channel", "")
symbol = data.get("symbol", "")
if channel == "raw":
# Orderbook-Update verarbeiten
if "book" in data:
self._update_orderbook(symbol, data["book"])
# Trade-Update verarbeiten
elif "trade" in data:
self._process_trade(symbol, data["trade"])
# Notify callbacks (z.B. für Trading-Engine)
for callback in self.callbacks:
await callback(symbol, self.orderbooks.get(symbol), self.recent_trades.get(symbol, []))
def _update_orderbook(self, symbol: str, book_data: Dict):
"""Rekonstruiert Orderbook aus Incremental-Updates"""
if symbol not in self.orderbooks:
self.orderbooks[symbol] = Orderbook(symbol=symbol)
ob = self.orderbooks[symbol]
# Parse bids und asks
for side, levels in [("bid", book_data.get("bid", [])),
("ask", book_data.get("ask", []))]:
for price_str, qty_str in levels:
price = float(price_str)
qty = float(qty_str)
level_list = ob.bids if side == "bid" else ob.asks
# Finde oder erstelle Level
found = False
for level in level_list:
if abs(level.price - price) < 1e-8:
if qty == 0:
level_list.remove(level)
else:
level.quantity = qty
found = True
break
if not found and qty > 0:
level_list.append(OrderbookLevel(price=price, quantity=qty))
# Sortiere: Bids absteigend, Asks aufsteigend
if side == "bid":
level_list.sort(key=lambda x: -x.price)
else:
level_list.sort(key=lambda x: x.price)
ob.last_update = time.time()
def _process_trade(self, symbol: str, trade_data: Dict):
"""Verarbeitet Trade-Daten und aktualisiert Recent-Trades"""
trade = Trade(
symbol=symbol,
side=trade_data.get("side", "buy"),
price=float(trade_data["p"]),
quantity=float(trade_data["q"]),
timestamp=trade_data.get("ts", 0),
trade_id=trade_data.get("id", "")
)
self.recent_trades[symbol].append(trade)
# Behalte nur die letzten 100 Trades
self.recent_trades[symbol] = self.recent_trades[symbol][-100:]
# Berechne rolling Volatilität
self._calculate_volatility(symbol)
def _calculate_volatility(self, symbol: str):
"""Berechnet rolling Volatilität aus recent trades"""
trades = self.recent_trades.get(symbol, [])
if len(trades) < 2:
self.volatility[symbol] = 0.0
return
prices = [t.price for t in trades[-20:]]
returns = [(prices[i] - prices[i-1]) / prices[i-1] for i in range(1, len(prices))]
if returns:
mean = sum(returns) / len(returns)
variance = sum((r - mean) ** 2 for r in returns) / len(returns)
self.volatility[symbol] = variance ** 0.5
def register_callback(self, callback: Callable):
"""Registriert einen Callback für Orderbook-Updates"""
self.callbacks.append(callback)
Async Main für Testing
async def main():
client = TardisAscendEXClient(tardis_token="YOUR_TARDIS_TOKEN")
async def on_update(symbol, orderbook, trades):
if orderbook:
print(f"[{symbol}] Spread: {orderbook.get_spread():.6f}, "
f"Mid: {orderbook.get_mid_price():.6f}, "
f"Depth: {orderbook.get_depth()}")
client.register_callback(on_update)
# Starte Verbindung
await client.connect(["BTC/USDT"])
if __name__ == "__main__":
asyncio.run(main())3. Orderbook-Slippages-Modellierung
Die盘口滑点 (Orderbook Slippage) ist kritisch für die Bewertung von Market-Making-Strategien. Mein Modell berücksichtigt die Tiefe des Orderbooks und die zu erwartenden Kosten beim Durchlaufen mehrerer Level.
# slippage_model.py
import math
from typing import Dict, List, Tuple, Optional
from dataclasses import dataclass
@dataclass
class SlippageEstimate:
"""Ergebnis der Slippage-Schätzung"""
expected_cost_bps: float # Kosten in Basispunkten
worst_case_cost_bps: float # Worst-Case Kosten
probability_of_fill: float # Wahrscheinlichkeit der Ausführung
effective_spread: float # Effektiver Spread
market_impact: float # Marktauswirkungs-Schätzung
class OrderbookSlippageModel:
"""Modelliert die Slippage-Kosten für Orderausführung"""
def __init__(self, fee_tier: float = 0.001):
"""
Args:
fee_tier: Börsengebühren-Tier (z.B. 0.001 = 0.1% Makergebühr)
"""
self.fee_tier = fee_tier
def estimate_slippage(
self,
orderbook: 'Orderbook',
order_side: str, # "buy" oder "sell"
order_quantity: float,
time_horizon_ms: int = 100
) -> SlippageEstimate:
"""
Schätzt Slippage-Kosten für eine gegebene Order
Args:
orderbook: Aktueller Orderbook-Zustand
order_side: "buy" (wir kaufen) oder "sell" (wir verkaufen)
order_quantity: Menge der Order
time_horizon_ms: Zeitrahmen für Ausführung
Returns:
SlippageEstimate mit detaillierten Kostenanalysen
"""
if order_side == "buy":
levels = orderbook.asks # Wir heben Ask-Seite auf
else:
levels = orderbook.bids # Wir drücken Bid-Seite herunter
if not levels:
return SlippageEstimate(
expected_cost_bps=0,
worst_case_cost_bps=0,
probability_of_fill=0,
effective_spread=0,
market_impact=0
)
# Berechne Füllungswahrscheinlichkeit basierend auf Depth
cumulative_volume = 0
execution_price = 0
volumes = []
for level in levels:
cumulative_volume += level.quantity
execution_price += level.price * level.quantity
volumes.append(level.quantity)
if cumulative_volume >= order_quantity:
break
if cumulative_volume < order_quantity:
# Order passt nicht komplett in Orderbook
return SlippageEstimate(
expected_cost_bps=10000, # 100% Slippage
worst_case_cost_bps=20000,
probability_of_fill=cumulative_volume / order_quantity,
effective_spread=0,
market_impact=1.0
)
# Volume Weighted Average Price (VWAP)
vwap = execution_price / order_quantity
mid_price = orderbook.get_mid_price()
# Slippage in Basispunkten (1 BP = 0.01%)
slippage_bps = abs(vwap - mid_price) / mid_price * 10000
# Worst-Case: Slippage + Spread
spread_bps = orderbook.get_spread() / mid_price * 10000
worst_case_bps = slippage_bps + spread_bps
# Effektiver Spread (was wir wirklich zahlen)
effective_spread = vwap - mid_price if order_side == "buy" else mid_price - vwap
# Marktauswirkungs-Schätzung (vereinfachtes Almgren-Chriss Modell)
market_impact = self._estimate_market_impact(
order_quantity=order_quantity,
orderbook_volume=cumulative_volume,
volatility=0.02 # 2% Volatilität angenommen
)
# Wahrscheinlichkeit der vollständigen Ausführung
fill_prob = min(1.0, cumulative_volume / order_quantity)
return SlippageEstimate(
expected_cost_bps=round(slippage_bps, 2),
worst_case_cost_bps=round(worst_case_bps, 2),
probability_of_fill=round(fill_prob, 4),
effective_spread=round(effective_spread, 8),
market_impact=round(market_impact, 4)
)
def _estimate_market_impact(
self,
order_quantity: float,
orderbook_volume: float,
volatility: float
) -> float:
"""
Vereinfachtes Almgren-Chriss Marktauswirkungsmodell
Impact = sigma * (Q / ADV)^0.6 * eta
Args:
order_quantity: Ordergröße
orderbook_volume: Verfügbares Volumen im Orderbook
volatility: Geschätzte Volatilität
Returns:
Marktauswirkung als Bruchteil (z.B. 0.01 = 1%)
"""
participation_rate = order_quantity / orderbook_volume if orderbook_volume > 0 else 1.0
# Vereinfachte Formel
eta = 0.1 # Marktauswirkungsparameter
impact = volatility * (participation_rate ** 0.6) * eta
return impact
def calculate_pnl_with_slippage(
self,
spread_bps: float,
your_bid_cost_bps: float,
your_ask_cost_bps: float,
orderbook_depth: int = 5,
fill_probability: float = 0.95
) -> Dict[str, float]:
"""
Berechnet PnL einer Market-Making-Strategie unter Berücksichtigung von Slippage
Args:
spread_bps: Strategie-Spread in Basispunkten
your_bid_cost_bps: Ihre durchschnittliche Bid-Slippage
your_ask_cost_bps: Ihre durchschnittliche Ask-Slippage
orderbook_depth: Durchschnittliche Orderbook-Tiefe
fill_probability: Wahrscheinlichkeit dass Order ausgeführt wird
Returns:
Dictionary mit PnL-Metriken
"""
# Brutto-Gewinn pro Trade
gross_profit_per_trade = spread_bps / 10000
# Netto-Gewinn nach Slippage
net_profit_per_trade = gross_profit_per_trade - (
(your_bid_cost_bps + your_ask_cost_bps) / 2 / 10000
)
# Adjustiert für Füllungswahrscheinlichkeit
expected_profit_per_cycle = net_profit_per_trade * fill_probability
# Anzahl Zyklen pro Tag (vereinfacht)
cycles_per_day = orderbook_depth * 24 # Annahme
# Täglicher erwarteter Gewinn
daily_expected_pnl = expected_profit_per_cycle * cycles_per_day
# Break-even Spread (minimale Spread die profitabel ist)
break_even_spread = (
(your_bid_cost_bps + your_ask_cost_bps) / 2 / fill_probability
)
return {
"gross_profit_per_trade": round(gross_profit_per_trade, 6),
"net_profit_per_trade": round(net_profit_per_trade, 6),
"expected_daily_pnl_pct": round(daily_expected_pnl * 100, 4),
"break_even_spread_bps": round(break_even_spread, 2),
"risk_adjusted_roi": round(net_profit_per_trade / (your_bid_cost_bps + your_ask_cost_bps) * 100, 2)
}
Beispiel-Berechnung
if __name__ == "__main__":
from tardis_ascendex_client import Orderbook, OrderbookLevel
# Simulierter Orderbook
ob = Orderbook(symbol="BTC/USDT")
ob.bids = [
OrderbookLevel(50000, 1.5),
OrderbookLevel(49999, 2.0),
OrderbookLevel(49998, 3.0),
OrderbookLevel(49997, 5.0),
OrderbookLevel(49996, 8.0),
]
ob.asks = [
OrderbookLevel(50001, 1.5),
OrderbookLevel(50002, 2.0),
OrderbookLevel(50003, 3.0),
OrderbookLevel(50004, 5.0),
OrderbookLevel(50005, 8.0),
]
model = OrderbookSlippageModel()
# Schätze Slippage für 2 BTC Order
slip = model.estimate_slippage(ob, "buy", 2.0)
print(f"Slippage-Schätzung für 2 BTC Kauf:")
print(f" Erwartete Kosten: {slip.expected_cost_bps:.2f} BPS")
print(f" Worst-Case: {slip.worst_case_cost_bps:.2f} BPS")
print(f" Füllungswahrscheinlichkeit: {slip.probability_of_fill*100:.1f}%")
# PnL-Berechnung
pnl = model.calculate_pnl_with_slippage(
spread_bps=10, # 0.1% Spread
your_bid_cost_bps=5,
your_ask_cost_bps=5
)
print(f"\nPnL-Analyse:")
print(f" Break-Even Spread: {pnl['break_even_spread_bps']:.2f} BPS")4. Order-Matching-Replay-System
Für das Backtesting und die Validierung Ihrer Strategie ist ein Order-Matching-Replay unerlässlich. Mein System kann historische Tick-Daten akkurat wiedergeben und die Auswirkungen Ihrer Orders simulieren.
# order_matching_replay.py
import json
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from collections import deque
import bisect
@dataclass
class Order:
order_id: str
symbol: str
side: str # "buy" oder "sell"
price: float
quantity: float
filled_quantity: float = 0
timestamp: int = 0
status: str = "pending" # pending, filled, cancelled, partial
@property
def remaining_quantity(self) -> float:
return self.quantity - self.filled_quantity
@dataclass
class ReplayResult:
"""Ergebnis einer Replay-Simulation"""
total_trades: int
total_volume: float
avg_slippage_bps: float
max_slippage_bps: float
pnl: float
fill_rate: float
order_log: List[Dict] = field(default_factory=list)
class OrderMatchingReplay:
"""Simuliert Order-Ausführung basierend auf historischen Tick-Daten"""
def __init__(self, db_path: str = "trading_replay.db"):
self.db_path = db_path
self.active_orders: Dict[str, Order] = {}
self.trade_log: List[Dict] = []
self._init_database()
def _init_database(self):
"""Initialisiert SQLite-Datenbank für Replay-Daten"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS replay_orders (
order_id TEXT PRIMARY KEY,
symbol TEXT,
side TEXT,
price REAL,
quantity REAL,
filled_quantity REAL,
timestamp INTEGER,
status TEXT,
replay_time INTEGER
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS replay_trades (
id INTEGER PRIMARY KEY AUTOINCREMENT,
order_id TEXT,
symbol TEXT,
side TEXT,
price REAL,
quantity REAL,
slippage_bps REAL,
timestamp INTEGER,
FOREIGN KEY(order_id) REFERENCES replay_orders(order_id)
)
""")
conn.commit()
conn.close()
def replay_tick_sequence(
self,
ticks: List[Dict],
strategy_func, # Callback der Orders generiert
initial_capital: float = 10000.0
) -> ReplayResult:
"""
Replayed eine Sequenz von Ticks und simuliert Order-Ausführung
Args:
ticks: Liste von Tick-Daten im Format {'type': 'trade'/'book', ...}
strategy_func: Funktion die Orders generiert basierend auf aktuellem State
initial_capital: Startkapital für PnL-Berechnung
Returns:
ReplayResult mit detaillierten Statistiken
"""
capital = initial_capital
position = 0.0
all_slippages = []
filled_orders = 0
total_orders = 0
# Orderbook-State
bids = [] # [(price, quantity), ...]
asks = []
for tick in ticks:
tick_type = tick.get('type')
if tick_type == 'book':
# Update Orderbook-State
bids = [(float(p), float(q)) for p, q in tick.get('bid', [])]
asks = [(float(p), float(q)) for p, q in tick.get('ask', [])]
elif tick_type == 'trade':
trade_price = float(tick['p'])
trade_qty = float(tick['q'])
trade_side = tick.get('side', 'buy')
trade_ts = tick.get('ts', 0)
# Prüfe aktive Orders gegen Trade
self._match_orders(
trade_price=trade_price,
trade_qty=trade_qty,
trade_side=trade_side,
trade_ts=trade_ts,
bids=bids,
asks=asks,
slippages=all_slippages
)
# Generiere neue Orders basierend auf Strategie
new_orders = strategy_func(bids, asks, capital, position)
for order in new_orders:
self._submit_order(order, tick.get('ts', 0))
total_orders += 1
# Aktualisiere Kapital und Position
for order_id, order in list(self.active_orders.items()):
if order.status == 'filled':
fill_price = order.price
if order.side == 'buy':
cost = fill_price * order.filled_quantity
capital -= cost
position += order.filled_quantity
else:
revenue = fill_price * order.filled_quantity
capital += revenue
position -= order.filled_quantity
filled_orders += 1
# Berechne finale PnL
final_value = capital + position * (bids[0][0] if bids else 0)
pnl = final_value - initial_capital
return ReplayResult(
total_trades=len(self.trade_log),
total_volume=sum(t['quantity'] for t in self.trade_log),
avg_slippage_bps=sum(all_slippages) / len(all_slippages) if all_slippages else 0,
max_slippage_bps=max(all_slippages) if all_sl