Einleitung: Mein Weg zum algorithmischen Market Making
Als ich 2024 begann, mich intensiv mit Kryptowährungshandel zu beschäftigen, stand ich vor einer fundamentalen Frage: Wie können manuelle Trader mit automated Market Makern konkurrieren? Die Antwort fand ich in der Kombination aus OKX Market Maker API und intelligenten KI-gestützten Hedging-Strategien. In diesem Tutorial teile ich meine Erfahrungen aus über 18 Monaten Praxisbetrieb – inklusive der Fehler, die mich Tausende Dollar kosteten, bevor ich ein stabiles System entwickelte.
Der konkrete Anwendungsfall: Ich betreibe einen Market-Making-Bot für mehrere ERC-20 Token-Paare auf OKX mit einem Startingkapital von 50.000 USD. Mein Ziel war eine monatliche Rendite von 3-5% bei maximal 2% Drawdown. Der Schlüssel zum Erfolg liegt in der präzisen Abstimmung zwischen Order-Placement, automatischem Hedging und KI-gestützter Sentiment-Analyse.
Was ist die OKX Market Maker API?
Die OKX Market Maker API ist eine RESTful- und WebSocket-basierte Schnittstelle, die es Entwicklern ermöglicht, automatisierte Handelsstrategien zu implementieren. Im Gegensatz zur Standard-Trading-API bietet sie spezielle Features für Market Maker:
- Rabattprogramme: Market Maker erhalten reduzierte Trading-Fees (bis zu 0,02% Maker-Rabatt)
- Priorisierte Order-Ausführung: Schnellere Order-Book-Platzierung bei hoher Volatilität
- WebSocket-Feeds: Echtzeit-Updates für Orderbook, Trades und Preisbewegungen
- Algo-Orders: Erweiterte Order-Typen wie Iceberg, TWAP und Stop-Loss-Kombinationen
Architektur des automatisierten Hedging-Systems
Ein robustes Market-Making-System besteht aus vier Kernkomponenten:
1. OKX API Gateway
Die Kommunikation mit der OKX Exchange erfolgt über HTTPS-REST-Endpunkte. Die Basis-URL für alle API-Anfragen ist:
# OKX API Endpunkte
OKX_REST_BASE = "https://www.okx.com"
OKX_WS_URL = "wss://ws.okx.com:8443/ws/v5/public"
Authentifizierung
API_KEY = "your_okx_api_key"
SECRET_KEY = "your_okx_secret_key"
PASSPHRASE = "your_passphrase"
USE_SANDBOX = False # Auf True für Testnet setzen2. Orderbook-Manager
Der Orderbook-Manager puffert Preisdaten und berechnet Spread-Statistiken in Echtzeit:
import asyncio
import json
from datetime import datetime
from typing import Dict, List, Optional
import aiohttp
import hmac
import base64
import hashlib
class OKXOrderbookManager:
def __init__(self, symbol: str, depth: int = 25):
self.symbol = symbol.upper()
self.depth = depth
self.orderbook_bids = [] # [(price, size), ...]
self.orderbook_asks = []
self.last_update = None
self.spread_bps = 0.0
self.mid_price = 0.0
async def initialize(self):
"""Lädt initialen Orderbook-Status"""
endpoint = f"/api/v5/market/books?instId={self.symbol}&sz={self.depth}"
async with aiohttp.ClientSession() as session:
async with session.get(f"{OKX_REST_BASE}{endpoint}") as resp:
data = await resp.json()
if data.get("code") == "0":
books = data["data"][0]
self.orderbook_bids = [(float(b[0]), float(b[1])) for b in books["bids"]]
self.orderbook_asks = [(float(a[0]), float(a[1])) for a in books["asks"]]
self._calculate_spread()
def _calculate_spread(self):
"""Berechnet aktuellen Spread in Basispunkten"""
if self.orderbook_bids and self.orderbook_asks:
best_bid = self.orderbook_bids[0][0]
best_ask = self.orderbook_asks[0][0]
self.mid_price = (best_bid + best_ask) / 2
self.spread_bps = ((best_ask - best_bid) / self.mid_price) * 10000
def get_optimal_bid_price(self, target_spread_bps: float = 5.0) -> float:
"""Berechnet optimale Bid-Price für Market Making"""
base_price = self.mid_price
offset = base_price * (target_spread_bps / 10000)
return round(base_price - offset, self._get_precision())
def get_optimal_ask_price(self, target_spread_bps: float = 5.0) -> float:
"""Berechnet optimale Ask-Price für Market Making"""
base_price = self.mid_price
offset = base_price * (target_spread_bps / 10000)
return round(base_price + offset, self._get_precision())
def _get_precision(self) -> int:
"""Bestimmt Dezimalpräzision basierend auf Preisbereich"""
if self.mid_price >= 10000:
return 2
elif self.mid_price >= 100:
return 3
else:
return 5
Beispiel: Orderbook für BTC-USDT initialisieren
async def main():
obm = OKXOrderbookManager("BTC-USDT-SWAP")
await obm.initialize()
print(f"Mid-Price: ${obm.mid_price:,.2f}")
print(f"Spread: {obm.spread_bps:.2f} bps")
print(f"Optimal Bid: ${obm.get_optimal_bid_price(5.0):,.2f}")
print(f"Optimal Ask: ${obm.get_optimal_ask_price(5.0):,.2f}")
asyncio.run(main())3. Hedging-Engine mit KI-Sentiment-Analyse
Der kritischste Teil: Die automatische Absicherung offener Positionen. Hier integriere ich HolySheep AI für die Sentiment-Analyse von Nachrichten und Social Media:
import requests
from typing import Tuple, Optional
from dataclasses import dataclass
from enum import Enum
class HedgingStrategy(Enum):
FULL_HEDGE = "full" # 100% Absicherung
PARTIAL_HEDGE = "partial" # 50% Absicherung
DELTA_NEUTRAL = "delta" # Dynamisch basierend auf Delta
TRAILING_HEDGE = "trailing" # Trailing-Stop-ähnlich
@dataclass
class HedgePosition:
symbol: str
quantity: float
entry_price: float
current_price: float
pnl_pct: float
hedge_recommended: bool
hedge_ratio: float
class AIHedgingEngine:
"""
KI-gestützte Hedging-Engine mit HolySheep AI Integration
"""
def __init__(self, holysheep_api_key: str):
self.holysheep_base = "https://api.holysheep.ai/v1"
self.api_key = holysheep_api_key
self.strategy = HedgingStrategy.DELTA_NEUTRAL
def analyze_market_sentiment(self, news_headlines: List[str]) -> dict:
"""
Analysiert Marktsentiment für Hedge-Entscheidungen
Nutzt DeepSeek V3.2 für kosteneffiziente Sentiment-Analyse
"""
prompt = f"""Analysiere das folgende Marktsentiment für Trading-Hedging:
Nachrichten: {json.dumps(news_headlines, ensure_ascii=False)}
Antworte im JSON-Format:
{{"sentiment": "bullish/bearish/neutral", "confidence": 0.0-1.0,
"recommended_hedge_ratio": 0.0-1.0, "risk_factors": [...]}}"""
response = requests.post(
f"{self.holysheep_base}/chat/completions",
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
},
json={
"model": "deepseek-v3.2",
"messages": [{"role": "user", "content": prompt}],
"temperature": 0.3,
"max_tokens": 500
},
timeout=5 # Timeout für schnelle Entscheidungen
)
if response.status_code == 200:
result = response.json()
content = result["choices"][0]["message"]["content"]
return json.loads(content)
return {"sentiment": "neutral", "confidence": 0.5, "recommended_hedge_ratio": 0.5}
def calculate_hedge_parameters(
self,
position: HedgePosition,
sentiment: dict,
market_volatility: float
) -> Tuple[float, float, HedgingStrategy]:
"""
Berechnet optimale Hedge-Parameter basierend auf:
1. Position-Delta
2. KI-Sentiment
3. Markvolatilität
"""
base_hedge_ratio = sentiment.get("recommended_hedge_ratio", 0.5)
# Volatilitätsanpassung
if market_volatility > 0.05: # >5% tägliche Volatilität
adjusted_ratio = min(1.0, base_hedge_ratio * 1.5)
strategy = HedgingStrategy.FULL_HEDGE
elif market_volatility > 0.02: # >2% tägliche Volatilität
adjusted_ratio = min(1.0, base_hedge_ratio * 1.2)
strategy = HedgingStrategy.PARTIAL_HEDGE
else:
adjusted_ratio = base_hedge_ratio
strategy = HedgingStrategy.DELTA_NEUTRAL
# Hedge-Quantity und Stop-Loss
hedge_quantity = position.quantity * adjusted_ratio
stop_loss_pct = 0.02 + (market_volatility * 2) # Stop-Loss basierend auf Volatilität
return hedge_quantity, stop_loss_pct, strategy
def execute_hedge_order(
self,
position: HedgePosition,
hedge_quantity: float,
use_perpetual: bool = True
) -> dict:
"""
Platziert Hedge-Order (typischerweise auf Perpetual-Futures)
"""
hedge_symbol = position.symbol.replace("-SWAP", "-USDT-SWAP") if use_perpetual else position.symbol
# Bei Long-Position: Short-Hedge platzieren
if position.pnl_pct >= 0:
order_side = "sell"
order_type = "limit"
else:
order_side = "buy"
order_type = "stop"
hedge_price = self._calculate_hedge_price(position.current_price, order_side)
return {
"hedge_symbol": hedge_symbol,
"side": order_side,
"type": order_type,
"quantity": hedge_quantity,
"price": hedge_price,
"status": "ready_to_submit"
}
def _calculate_hedge_price(self, current_price: float, side: str) -> float:
"""Berechnet Hedge-Preis mit leichtem Discount für schnelle Ausführung"""
if side == "sell":
return round(current_price * 0.998, 4) # 0.2% Discount
else:
return round(current_price * 1.002, 4) # 0.2% Premium
Praxisbeispiel: Hedge-Entscheidung für BTC-Position
if __name__ == "__main__":
holysheep = AIHedgingEngine("YOUR_HOLYSHEEP_API_KEY")
# Simulierte Position
btc_position = HedgePosition(
symbol="BTC-USDT-SWAP",
quantity=0.5,
entry_price=67500.00,
current_price=68200.00,
pnl_pct=1.04,
hedge_recommended=False,
hedge_ratio=0.0
)
# News-Sentiment analysieren
headlines = [
"Bitcoin ETF verzeichnet Rekordzuflüsse von $500M",
"Fed signalisiert mögliche Zinssenkung 2026",
"Technische Analyse: BTC testet Widerstand bei $69.000"
]
sentiment = holysheep.analyze_market_sentiment(headlines)
print(f"Sentiment: {sentiment['sentiment']} (Confidence: {sentiment['confidence']:.1%})")
print(f"Recommended Hedge Ratio: {sentiment['recommended_hedge_ratio']:.1%}")
# Hedge-Parameter berechnen
hedge_qty, stop_loss, strategy = holysheep.calculate_hedge_parameters(
btc_position, sentiment, market_volatility=0.035
)
print(f"Strategy: {strategy.value}")
print(f"Hedge Quantity: {hedge_qty:.4f} BTC")
print(f"Stop-Loss: {stop_loss:.2%}")4. Risikomanagement-System
Das Risikomanagement ist das Herzstück jedes Market-Making-Systems:
from dataclasses import dataclass, field
from typing import Dict, List
from datetime import datetime, timedelta
import statistics
@dataclass
class RiskLimits:
max_position_size: float = 1.0 # Max Position in BTC
max_daily_loss: float = 0.02 # Max 2% Daily Loss
max_drawdown: float = 0.05 # Max 5% Drawdown
min_spread_bps: float = 3.0 # Min Spread in bps
max_spread_bps: float = 50.0 # Max Spread in bps
max_orders_per_minute: int = 60 # Rate-Limit
min_balance_usdt: float = 5000.0 # Minimum Wallet-Balance
@dataclass
class TradingStats:
daily_pnl: float = 0.0
daily_trades: int = 0
win_rate: float = 0.0
avg_spread_captured: float = 0.0
peak_balance: float = 0.0
current_drawdown: float = 0.0
trades_history: List[dict] = field(default_factory=list)
class RiskManager:
def __init__(self, limits: RiskLimits, initial_balance: float):
self.limits = limits
self.initial_balance = initial_balance
self.current_balance = initial_balance
self.stats = TradingStats()
self.stats.peak_balance = initial_balance
self.order_timestamps = []
def can_place_order(
self,
symbol: str,
side: str,
quantity: float,
price: float
) -> Tuple[bool, str]:
"""
Validiert Order gegen alle Risikolimits
Return: (allowed: bool, reason: str)
"""
# 1. Rate-Limit Prüfung
now = datetime.now()
self.order_timestamps = [
ts for ts in self.order_timestamps
if now - ts < timedelta(minutes=1)
]
if len(self.order_timestamps) >= self.limits.max_orders_per_minute:
return False, "Rate-Limit überschritten"
# 2. Balance-Prüfung
order_value = quantity * price
if side == "buy" and order_value > self.current_balance * 0.9:
return False, f"Unzureichendes Balance ({order_value:.2f} > {self.current_balance * 0.9:.2f})"
# 3. Position-Size-Prüfung
if quantity > self.limits.max_position_size:
return False, f"Position zu groß ({quantity:.4f} > {self.limits.max_position_size})"
# 4. Drawdown-Prüfung
if self.stats.current_drawdown > self.limits.max_drawdown:
return False, f"Max Drawdown erreicht ({self.stats.current_drawdown:.2%})"
# 5. Balance-Minimum
if self.current_balance < self.limits.min_balance_usdt:
return False, f"Balance unter Minimum ({self.current_balance:.2f})"
return True, "OK"
def update_stats(self, trade: dict):
"""Aktualisiert Trading-Statistiken nach jedem Trade"""
self.stats.daily_trades += 1
self.stats.trades_history.append({
**trade,
"timestamp": datetime.now().isoformat()
})
# P&L Update
if trade["side"] == "buy" and trade.get("realized_pnl"):
self.stats.daily_pnl -= trade["realized_pnl"]
elif trade["side"] == "sell" and trade.get("realized_pnl"):
self.stats.daily_pnl += trade["realized_pnl"]
# Balance Update
if trade.get("pnl"):
self.current_balance += trade["pnl"]
# Peak und Drawdown
if self.current_balance > self.stats.peak_balance:
self.stats.peak_balance = self.current_balance
self.stats.current_drawdown = (
(self.stats.peak_balance - self.current_balance) / self.stats.peak_balance
)
# Spread-Kalculation
if trade.get("spread_bps"):
spreads = [t.get("spread_bps", 0) for t in self.stats.trades_history[-100:]]
self.stats.avg_spread_captured = statistics.mean(spreads)
# Order-Timestamp speichern
self.order_timestamps.append(datetime.now())
def check_emergency_stop(self) -> Tuple[bool, str]:
"""
Prüft Emergency-Stop-Bedingungen
"""
# Daily Loss Stop
daily_loss_pct = abs(self.stats.daily_pnl) / self.initial_balance
if daily_loss_pct > self.limits.max_daily_loss:
return True, f"Daily Loss Limit erreicht: {daily_loss_pct:.2%}"
# Extreme Drawdown
if self.stats.current_drawdown > self.limits.max_drawdown * 1.5:
return True, f"Kritischer Drawdown: {self.stats.current_drawdown:.2%}"
# Volatilitäts-Stop
if len(self.stats.trades_history) >= 10:
recent_pnls = [t.get("pnl", 0) for t in self.stats.trades_history[-10:]]
volatility = statistics.stdev(recent_pnls) if len(recent_pnls) > 1 else 0
if volatility > self.initial_balance * 0.01: # >1% Volatilität
return True, f"Hohe Volatilität erkannt: ${volatility:.2f}"
return False, ""
def get_risk_report(self) -> dict:
"""Generiert täglichen Risiko-Bericht"""
return {
"balance": self.current_balance,
"daily_pnl": self.stats.daily_pnl,
"daily_pnl_pct": self.stats.daily_pnl / self.initial_balance,
"current_drawdown": self.stats.current_drawdown,
"peak_balance": self.stats.peak_balance,
"trades_today": self.stats.daily_trades,
"avg_spread_captured_bps": self.stats.avg_spread_captured,
"risk_score": self._calculate_risk_score()
}
def _calculate_risk_score(self) -> float:
"""Berechnet Risk Score (0-100, höher = riskant)"""
score = 0
# Drawdown-Beitrag (max 40 Punkte)
score += min(40, self.stats.current_drawdown * 800)
# Daily Loss (max 30 Punkte)
daily_loss_pct = abs(self.stats.daily_pnl) / self.initial_balance
score += min(30, daily_loss_pct * 1500)
# Positions-Konzentration (max 30 Punkte)
# Vereinfacht: nimmt an dass 1 Position = volles Risiko
score += 15 # Basis-Risiko
return min(100, score)
Nutzung
limits = RiskLimits(
max_position_size=0.5,
max_daily_loss=0.015, # 1.5%
max_drawdown=0.04, # 4%
min_spread_bps=4.0,
max_orders_per_minute=50
)
risk_mgr = RiskManager(limits, initial_balance=50000.0)
Test: Order validieren
allowed, reason = risk_mgr.can_place_order(
symbol="BTC-USDT-SWAP",
side="buy",
quantity=0.1,
price=68000.0
)
print(f"Order erlaubt: {allowed}, Grund: {reason}")Risikokontrolle: Die fünf Säulen
Meine Erfahrung zeigt, dass erfolgreiches Market Making auf fünf Risikokategorien basiert:
| Risikokategorie | Beschreibung | Max. Exposure | Stop-Loss |
|---|---|---|---|
| Position Risk | Ungesicherte Token-Exposition | 5% des Kapitals | 2% Drawdown |
| Spread Risk | Zu enge Spreads bei Volatilität | Dynamisch | 5 bps Minimum |
| Execution Risk | Slippage bei Order-Ausführung | 0.5% pro Trade | Market-Order-Stop |
| Liquidity Risk | Unfähigkeit Position zu schließen | 10% des Orderbooks | Depth < 1 BTC |
| Counterparty Risk | Exchange-Ausfall oder API-Probleme | N/A | Auto-Disconnect |
Integration mit KI: Sentiment-getriebenes Market Making
Der größte Vorteil des KI-gestützten Ansatzes liegt in der Sentiment-Analyse. Mein Bot analysiert:
- Twitter/X-Trends für das spezifische Token
- On-Chain-Metriken (Exchange-Flows, Wallet-Bewegungen)
- Nachrichten-Sentiment über HolySheep AI
- Options-Marktdaten für Volatilitäts-Signale
Die HolySheep AI Integration bietet dabei entscheidende Vorteile:
- Kosten: DeepSeek V3.2 kostet nur $0.42/MToken (85%+ günstiger als OpenAI GPT-4.1)
- Latenz: <50ms Response-Zeit für Echtzeit-Entscheidungen
- Flexibilität: Unterstützung für GPT-4.1, Claude Sonnet 4.5, Gemini 2.5 Flash
- Zahlung: WeChat Pay und Alipay für asiatische Nutzer
Häufige Fehler und Lösungen
Fehler 1: Race Conditions bei gleichzeitigen Orders
Symptom: Doppelte Orders, inkonsistente Positions-Updates, "Insufficient balance" trotz korrekter Berechnung.
Ursache: Asynchroner Code ohne proper locking bei mehreren WebSocket-Feeds.
# FEHLERHAFT - Race Condition
class BrokenOrderManager:
def __init__(self):
self.positions = {}
async def place_order(self, symbol, side, qty):
# Race: Position wird nicht atomar geprüft
if self.positions.get(symbol, 0) + qty > MAX_POSITION:
return None
# Hier kann zwischen Prüfung und Ausführung eine andere Order eingereiht werden
result = await self.okx.place_order(symbol, side, qty)
self.positions[symbol] += qty if side == "buy" else -qty
return result
LÖSUNG - Thread-Safe mit Lock
import asyncio
from contextlib import asynccontextmanager
class ThreadSafeOrderManager:
def __init__(self):
self.positions = {}
self._lock = asyncio.Lock()
self._pending_orders = set()
@asynccontextmanager
async def atomic_position_update(self, symbol: str, qty: float, side: str):
"""Atomare Position-Updates mit Locking"""
async with self._lock:
order_id = f"{symbol}_{side}_{qty}_{asyncio.get_event_loop().time()}"
self._pending_orders.add(order_id)
try:
# Position prüfen
current_pos = self.positions.get(symbol, 0)
new_pos = current_pos + qty if side == "buy" else current_pos - qty
if abs(new_pos) > MAX_POSITION:
raise ValueError(f"Position Limit überschritten: {new_pos}")
# Position aktualisieren
self.positions[symbol] = new_pos
yield order_id
finally:
self._pending_orders.discard(order_id)
async def place_order_safe(self, symbol: str, side: str, qty: float, price: float):
"""Sichere Order-Platzierung mit Atomic Update"""
try:
async with self.atomic_position_update(symbol, qty, side) as order_id:
result = await self.okx.place_order(
symbol=symbol,
side=side,
quantity=qty,
price=price
)
print(f"Order {order_id} erfolgreich: {result}")
return result
except ValueError as e:
print(f"Order abgelehnt: {e}")
return None
except Exception as e:
# Rollback bei Fehler
print(f"Order fehlgeschlagen: {e}, Rollback wird durchgeführt")
await self._sync_positions()
return NoneFehler 2: WebSocket Reconnection Storms
Symptom: Massiver Nachrichten-Backlog, Orders basierend auf veralteten Preisen, Memory-Leak.
Ursache: Exponentielles Backoff ohne Jitter, keine Heartbeat-Überwachung.
import asyncio
import random
from typing import Optional
class RobustWebSocketManager:
def __init__(
self,
url: str,
on_message: callable,
max_reconnect_attempts: int = 10,
heartbeat_interval: int = 30
):
self.url = url
self.on_message = on_message
self.max_attempts = max_reconnect_attempts
self.heartbeat_interval = heartbeat_interval
self.ws: Optional[WebSocket] = None
self.last_heartbeat = None
self.reconnect_count = 0
self.is_running = False
# Message Buffer für Out-of-Order Messages
self.message_buffer = {}
self.last_sequence = {}
async def connect(self):
"""Verbindung mit exponential Backoff + Jitter"""
self.is_running = True
base_delay = 1
while self.is_running and self.reconnect_count < self.max_attempts:
try:
async with aiohttp.ClientSession() as session:
self.ws = await session.ws_connect(
self.url,
heartbeat=self.heartbeat_interval
)
self.last_heartbeat = asyncio.get_event_loop().time()
self.reconnect_count = 0
print(f"WebSocket verbunden: {self.url}")
# Heartbeat-Task starten
heartbeat_task = asyncio.create_task(self._heartbeat_monitor())
# Message-Handler
async for msg in self.ws:
if msg.type == aiohttp.WSMsgType.TEXT:
await self._handle_message(msg.data)
elif msg.type == aiohttp.WSMsgType.PING:
await self.ws.ping()
elif msg.type == aiohttp.WSMsgType.ERROR:
print(f"WebSocket Fehler: {msg.data}")
break
heartbeat_task.cancel()
except aiohttp.ClientError as e:
self.reconnect_count += 1
# Exponential Backoff mit Jitter
delay = min(60, base_delay * (2 ** self.reconnect_count))
delay *= (0.5 + random.random()) # Jitter: 50%-150%
print(f"Reconnect in {delay:.1f}s (Attempt {self.reconnect_count})")
await asyncio.sleep(delay)
async def _handle_message(self, data: str):
"""Verarbeitet Nachrichten mit Sequenz-Nummer-Tracking"""
try:
msg = json.loads(data)
# Sequenz-Nummer für Out-of-Order-Detection
if "sequence" in msg:
symbol = msg.get("arg", {}).get("channel", "unknown")
seq = msg["sequence"]
if symbol in self.last_sequence:
if seq <= self.last_sequence[symbol]:
print(f"Alte Nachricht verworfen: {seq} < {self.last_sequence[symbol]}")
return
self.last_sequence[symbol] = seq
await self.on_message(msg)
except json.JSONDecodeError as e:
print(f"JSON Parse Fehler: {e}")
async def _heartbeat_monitor(self):
"""Überwacht Heartbeat und triggert Reconnect bei Timeout"""
while True:
await asyncio.sleep(self.heartbeat_interval)
if self.ws and self.ws.closed:
print("Heartbeat: WebSocket geschlossen")
break
time_since_heartbeat = asyncio.get_event_loop().time() - self.last_heartbeat
if time_since_heartbeat > self.heartbeat_interval * 3:
print(f"Heartbeat-Timeout erkannt: {time_since_heartbeat:.1f}s")
await self.ws.close()
breakFehler 3: Fehlende Slippage-Kontrolle bei hoher Volatilität
Symptom: Orders werden zu extremen Preisen ausgeführt, plötzliche Verluste.
Ursache: Keine Validierung des Market-Impact vor Order-Platzierung.
import asyncio
from typing import Tuple
class SlippageController:
def __init__(
self,
max_slippage_bps: float = 20.0, # Max 20 bps = 0.2%
volatility_threshold: float = 0.01, # 1% Volatilität
min_orderbook_depth: float = 0.5 # Min 0.5 BTC im Orderbook
):
self.max_slippage_bps = max_slippage_bps
self.volatility_threshold = volatility_threshold
self.min_depth = min_orderbook_depth
self.orderbook_cache = {}
async def validate_order_price(
self,
symbol: str,
side: str,
requested_price: float,
orderbook: dict
) -> Tuple[bool, float, str]:
"""
Validiert Order-Preis gegen Slippage-Limits
Return: (valid, adjusted_price, reason)
"""
# Depth-Prüfung
if side == "buy":
available_depth = sum(
float(q) for p, q in orderbook.get("asks", [])[:5]
if float(p) <= requested_price * 1.01
)
else:
available_depth = sum(
float(q) for p, q in orderbook.get("bids", [])[:5]
if float(p) >= requested_price * 0.99
)
if available_depth < self.min_depth:
return False, 0, f"Unzureichende Orderbook-Tiefe: {available_depth:.4f}"
# Besten Preis ermitteln
if side == "buy":
best_price = float(orderbook["asks"][0][0])
else:
best_price = float(orderbook["bids"][0][0])
# Slippage berechnen
slippage_bps = abs(requested_price - best_price) / best_price * 10000
if slippage_bps > self.max_slippage_bps:
# Try to adjust to max allowed price
max_allowed_price = best_price * (1 + self.max_slippage_bps / 10000) if side == "buy" \
else best_price * (1 - self.max_slippage_bps / 10000)
if