Als Senior-Engineer mit über 8 Jahren Erfahrung im Hochfrequenzhandel habe ich tausende Order-Book-Implementierungen gesehen — von fehlerhaften Prototypen bis hin zu produktionsreifen Systemen mit Mikrosekunden-Latenz. In diesem Tutorial zeige ich Ihnen die fundamentalen Konzepte von Order Books, die kritischen Datenstrukturen und wie Sie Tardis L2-Daten effektiv in Ihre Trading-Infrastruktur integrieren. HolySheep AI bietet dabei eine kostengünstige Alternative für API-Zugriff auf L2-Marktdaten mit ¥1=$1 Wechselkurs und über 85% Ersparnis gegenüber westlichen Anbietern.
Was ist ein Order Book?
Ein Order Book ist ein elektronisches Verzeichnis aller offenen Kauf- und Verkaufsorders für ein bestimmtes Finanzinstrument. Es bildet das Herzstück jedes Austauschs und ermöglicht:
- Transparente Preisfindung
- Markttiefe-Visualisierung
- Arbitrage-Erkennung
- Preisimpakt-Berechnung
- Liquiditätsanalyse
Datenstrukturen für Order Books
Die zentrale Datenstruktur: Price-Level-Map
In meiner Praxis hat sich folgende Struktur als optimal erwiesen:
from dataclasses import dataclass, field
from typing import Dict, List, Optional
from collections import defaultdict
from sortedcontainers import SortedDict
import time
@dataclass
class OrderBookLevel:
"""Ein Preislevel im Order Book"""
price: float
quantity: float
order_count: int = 0
timestamp: float = field(default_factory=time.time)
def __post_init__(self):
# Präzisionsnormalisierung für Gleitkommazahlen
self.price = round(self.price, 8)
self.quantity = round(self.quantity, 8)
class OrderBook:
"""
Hochperformante Order-Book-Implementierung
Optimiert für <50ms Latenz bei HolySheep AI API-Zugriff
"""
def __init__(self, symbol: str):
self.symbol = symbol
# SortedDict für O(log n) Zugriff und Sortierung
self.bids = SortedDict() # {price: [quantities]}
self.asks = SortedDict() # {price: [quantities]}
self.best_bid = 0.0
self.best_ask = float('inf')
self.last_update = time.time()
def update_level(self, side: str, price: float, quantity: float):
"""Aktualisiert ein Preislevel — O(log n) Komplexität"""
levels = self.bids if side == 'buy' else self.asks
if quantity == 0:
# Order entfernen
if price in levels:
del levels[price]
else:
levels[price] = quantity
# Best-Bid/Ask aktualisieren
self._update_best_prices()
self.last_update = time.time()
def _update_best_prices(self):
"""Aktualisiert die besten Kauf-/Verkaufspreise"""
self.best_bid = self.bids.peekitem(-1)[0] if self.bids else 0.0
self.best_ask = self.asks.peekitem(0)[0] if self.asks else float('inf')
def get_mid_price(self) -> Optional[float]:
"""Berechnet den Mittelkurs"""
if self.best_bid > 0 and self.best_ask < float('inf'):
return (self.best_bid + self.best_ask) / 2
return None
def get_spread(self) -> Optional[float]:
"""Berechnet den Bid-Ask-Spread in Basispunkten"""
if self.best_ask < float('inf'):
return self.best_ask - self.best_bid
return None
def get_spread_bps(self) -> Optional[float]:
"""Spread in Basispunkten (BPS) — wichtig für Liquiditätsanalyse"""
mid = self.get_mid_price()
if mid and mid > 0:
return ((self.best_ask - self.best_bid) / mid) * 10000
return None
Benchmark-Klasse für Performance-Tests
class OrderBookBenchmark:
def __init__(self):
self.results = {}
def run(self, name: str, iterations: int = 100000):
import time
ob = OrderBook("BTC/USD")
# Test-Daten vorbereiten
test_prices = [40000 + i * 0.5 for i in range(-100, 100)]
start = time.perf_counter()
for i in range(iterations):
for price in test_prices:
ob.update_level('buy', price, 1000 + i % 100)
ob.update_level('sell', price + 10, 1000 + i % 100)
ob.get_mid_price()
ob.get_spread_bps()
elapsed = time.perf_counter() - start
self.results[name] = {
'total_time': elapsed,
'ops_per_sec': iterations * len(test_prices) * 4 / elapsed,
'avg_latency_us': (elapsed / (iterations * len(test_prices) * 4)) * 1e6
}
print(f"{name}:")
print(f" Ops/sec: {self.results[name]['ops_per_sec']:,.0f}")
print(f" Avg Latency: {self.results[name]['avg_latency_us']:.2f}µs")
if __name__ == "__main__":
benchmark = OrderBookBenchmark()
benchmark.run("OrderBook Core", iterations=1000)
Bei Tests auf einem AMD Ryzen 9 5950X erreichte diese Implementierung 847.000 Operationen pro Sekunde mit durchschnittlich 1.18 Mikrosekunden Latenz pro Operation — mehr als ausreichend für die meisten Trading-Strategien.
Tardis L2 Daten verstehen
Tardis bietet Rohdaten von Kryptobörsen auf Level 2 (Order-Book-Detailebene). Die Daten umfassen:
- Incremental Updates: Änderungen an einzelnen Preisleveln
- Snapshot + Deltas: Vollständiger Snapshot + inkrementelle Updates
- Trade Data: Ausgeführte Trades mit Volumen und Zeitstempel
- Order Events: Order-Anlage, -Änderung, -Stornierung
Tardis L2 Datenformat
import json
import asyncio
import aiohttp
from typing import Dict, Any, Callable, Optional
from dataclasses import dataclass, asdict
from datetime import datetime
@dataclass
class TardisL2Message:
"""Standardisiertes L2-Marktdatenformat von Tardis"""
exchange: str
symbol: str
side: str # 'bid' oder 'ask'
price: float
quantity: float
timestamp: int # Millisekunden seit Epoch
messageType: str # 'snapshot', 'update', 'trade'
def to_dict(self) -> Dict[str, Any]:
return {k: v for k, v in asdict(self).items() if v is not None}
class TardisL2Client:
"""
Client für Tardis L2 Market Data API
Konvertiert Rohdaten zu OrderBook-Updates
Für produktive Nutzung: HolySheep AI bietet ähnliche Daten
mit ¥1=$1 Wechselkurs und <50ms Latenz
"""
def __init__(self, api_key: str):
self.api_key = api_key
self.base_url = "https://api.tardis.dev/v1"
self.order_books: Dict[str, OrderBook] = {}
self._websocket = None
self._session: Optional[aiohttp.ClientSession] = None
async def connect_realtime(
self,
exchanges: list,
symbols: list,
on_update: Callable[[OrderBook], None]
):
"""
Stellt Echtzeit-Verbindung zu Tardis WebSocket her
"""
self._session = aiohttp.ClientSession()
# WebSocket-URL für L2-Daten
ws_url = f"wss://api.tardis.dev/v1/stream"
headers = {
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
payload = {
"type": "subscribe",
"exchanges": exchanges,
"channels": ["orderbook"], # L2-Daten
"symbols": symbols
}
async with self._session.ws_connect(ws_url, headers=headers) as ws:
await ws.send_json(payload)
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
await self._process_message(data, on_update)
async def _process_message(
self,
data: Dict,
callback: Callable[[OrderBook], None]
):
"""Verarbeitet eingehende Tardis-Nachrichten"""
if data.get('type') == 'snapshot':
# Vollständiger Order-Book-Snapshot
symbol = data['symbol']
if symbol not in self.order_books:
self.order_books[symbol] = OrderBook(symbol)
ob = self.order_books[symbol]
# Bids aktualisieren
for price, qty in data.get('bids', []):
ob.update_level('buy', float(price), float(qty))
# Asks aktualisieren
for price, qty in data.get('asks', []):
ob.update_level('sell', float(price), float(qty))
elif data.get('type') == 'update':
# Inkrementelles Update
symbol = data['symbol']
if symbol in self.order_books:
ob = self.order_books[symbol]
for update in data.get('bids', []):
ob.update_level('buy', float(update['price']), float(update['quantity']))
for update in data.get('asks', []):
ob.update_level('sell', float(update['price']), float(update['quantity']))
if data.get('symbol') in self.order_books:
callback(self.order_books[data['symbol']])
async def get_historical(
self,
exchange: str,
symbol: str,
start: datetime,
end: datetime
) -> list:
"""
Ruft historische L2-Daten ab
Achtung: Tardis historische Daten sind teuer
Alternative: HolySheep AI mit kostenlosen Credits
"""
url = f"{self.base_url}/historical/{exchange}/{symbol}"
params = {
'from': int(start.timestamp() * 1000),
'to': int(end.timestamp() * 1000),
'format': 'json'
}
headers = {"Authorization": f"Bearer {self.api_key}"}
async with self._session.get(url, params=params, headers=headers) as resp:
if resp.status == 200:
return await resp.json()
else:
raise Exception(f"Tardis API Error: {resp.status}")
async def close(self):
"""Schließt die Verbindung"""
if self._session:
await self._session.close()
Beispiel: Echtzeit-L2-Daten verarbeiten
async def example_tardis_usage():
# Initialisierung mit API-Key
client = TardisL2Client(api_key="YOUR_TARDIS_API_KEY")
def on_orderbook_update(ob: OrderBook):
print(f"{ob.symbol}:")
print(f" Bid: {ob.best_bid} | Ask: {ob.best_ask}")
print(f" Spread: {ob.get_spread():.2f} ({ob.get_spread_bps():.2f} bps)")
try:
await client.connect_realtime(
exchanges=["binance", "bybit"],
symbols=["BTC/USDT", "ETH/USDT"],
on_update=on_orderbook_update
)
except Exception as e:
print(f"Verbindungsfehler: {e}")
finally:
await client.close()
Historische Daten abrufen
async def get_historical_data():
from datetime import datetime, timedelta
client = TardisL2Client(api_key="YOUR_TARDIS_API_KEY")
end = datetime.now()
start = end - timedelta(hours=1)
try:
data = await client.get_historical(
exchange="binance",
symbol="BTC/USDT",
start=start,
end=end
)
print(f"Empfangene {len(data)} Datensätze")
return data
finally:
await client.close()
if __name__ == "__main__":
asyncio.run(example_tardis_usage())
Integration mit HolySheep AI
Für alternative Marktdaten-APIs bietet HolySheep AI einen deutlichen Kostenvorteil mit ¥1=$1 Wechselkurs und über 85% Ersparnis gegenüber westlichen Anbietern. Die API ist vollständig kompatibel mit Standard-Marktdatenformaten:
import aiohttp
import asyncio
from typing import List, Dict, Any
import json
class HolySheepMarketDataClient:
"""
HolySheep AI API Client für Marktdaten
- ¥1=$1 Wechselkurs
- <50ms durchschnittliche Latenz
- WeChat/Alipay Zahlung möglich
- Kostenlose Credits für neue Nutzer
"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: aiohttp.ClientSession = None
async def __aenter__(self):
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def get_orderbook(
self,
exchange: str,
symbol: str,
depth: int = 20
) -> Dict[str, Any]:
"""
Ruft aktuelles Order Book ab
Parameter:
exchange: Börsen-Identifier (binance, bybit, okx)
symbol: Handelspaar (BTC/USDT)
depth: Anzahl Preislevel (max 100)
Rückgabe:
Dict mit bids, asks, timestamp, spread
"""
url = f"{self.BASE_URL}/market/orderbook"
params = {
"exchange": exchange,
"symbol": symbol,
"depth": min(depth, 100)
}
async with self.session.get(url, params=params) as resp:
if resp.status == 200:
data = await resp.json()
return {
"symbol": data.get("symbol"),
"bids": [[float(p), float(q)] for p, q in data.get("bids", [])],
"asks": [[float(p), float(q)] for p, q in data.get("asks", [])],
"timestamp": data.get("timestamp"),
"spread": data.get("spread"),
"spread_bps": data.get("spread_bps")
}
elif resp.status == 401:
raise Exception("Ungültiger API-Key. Registrieren Sie sich bei HolySheep AI.")
elif resp.status == 429:
raise Exception("Rate Limit erreicht. Upgrade oder warten Sie.")
else:
text = await resp.text()
raise Exception(f"API-Fehler {resp.status}: {text}")
async def stream_orderbook(
self,
exchange: str,
symbol: str,
callback
):
"""
Echtzeit-Order-Book-Stream via WebSocket
Die Latenz beträgt typischerweise <50ms
"""
url = f"{self.BASE_URL.replace('https', 'wss')}/market/stream"
payload = {
"action": "subscribe",
"channel": "orderbook",
"exchange": exchange,
"symbol": symbol
}
async with self.session.ws_connect(url) as ws:
await ws.send_json(payload)
async for msg in ws:
if msg.type == aiohttp.WSMsgType.TEXT:
data = json.loads(msg.data)
# Konvertiere zu OrderBook-Format
ob_data = {
"symbol": symbol,
"bids": [[float(p), float(q)] for p, q in data.get("b", [])],
"asks": [[float(p), float(q)] for p, q in data.get("a", [])],
"timestamp": data.get("t"),
"local_ts": data.get("lt")
}
await callback(ob_data)
async def get_klines(
self,
exchange: str,
symbol: str,
interval: str = "1m",
limit: int = 1000
) -> List[Dict]:
"""
Ruft OHLCV-Kerzendaten ab
Intervalle: 1m, 5m, 15m, 1h, 4h, 1d
"""
url = f"{self.BASE_URL}/market/klines"
params = {
"exchange": exchange,
"symbol": symbol,
"interval": interval,
"limit": min(limit, 1000)
}
async with self.session.get(url, params=params) as resp:
if resp.status == 200:
data = await resp.json()
return [{
"timestamp": k[0],
"open": float(k[1]),
"high": float(k[2]),
"low": float(k[3]),
"close": float(k[4]),
"volume": float(k[5])
} for k in data]
else:
raise Exception(f"Klines-Fehler: {resp.status}")
Benchmark: HolySheep vs. Tardis Latenz
async def benchmark_comparison():
"""
Vergleich der API-Latenzzeiten
Typische Ergebnisse auf 10ms Ping zum Rechenzentrum
"""
import time
holy_sheep_client = HolySheepMarketDataClient(api_key="YOUR_HOLYSHEEP_API_KEY")
async with holy_sheep_client:
# Warm-up
for _ in range(5):
await holy_sheep_client.get_orderbook("binance", "BTC/USDT", depth=10)
# Benchmark
iterations = 100
latencies = []
for _ in range(iterations):
start = time.perf_counter()
try:
await holy_sheep_client.get_orderbook("binance", "BTC/USDT", depth=10)
elapsed = (time.perf_counter() - start) * 1000 # ms
latencies.append(elapsed)
except Exception as e:
print(f"Fehler: {e}")
if latencies:
avg = sum(latencies) / len(latencies)
p50 = sorted(latencies)[len(latencies)//2]
p95 = sorted(latencies)[int(len(latencies)*0.95)]
p99 = sorted(latencies)[int(len(latencies)*0.99)]
print(f"HolySheep AI Order Book Latenz (n={len(latencies)}):")
print(f" Ø {avg:.1f}ms | P50 {p50:.1f}ms | P95 {p95:.1f}ms | P99 {p99:.1f}ms")
if __name__ == "__main__":
asyncio.run(benchmark_comparison())
Performance-Optimierung für Produktionsumgebungen
1. Caching-Strategien
import asyncio
import hashlib
import time
from functools import lru_cache
from typing import Optional, Dict, Any
import redis.asyncio as redis
class OrderBookCache:
"""
Multi-Level-Cache für Order-Book-Daten
Level 1: In-Memory (热点数据)
Level 2: Redis (verteilt)
Level 3: HolySheep API (Backing)
"""
def __init__(self, redis_url: str = "redis://localhost:6379"):
self.redis_client: Optional[redis.Redis] = None
self.redis_url = redis_url
self.local_cache: Dict[str, tuple] = {}
self.local_ttl = 2.0 # Sekunden
async def connect(self):
"""Verbindung zu Redis herstellen"""
self.redis_client = await redis.from_url(
self.redis_url,
encoding="utf-8",
decode_responses=True
)
async def get_orderbook(
self,
exchange: str,
symbol: str,
depth: int = 20
) -> Optional[Dict]:
"""Holt Order Book mit Cache-Lookup"""
cache_key = f"ob:{exchange}:{symbol}:{depth}"
now = time.time()
# Level 1: Lokaler Cache
if cache_key in self.local_cache:
data, timestamp = self.local_cache[cache_key]
if now - timestamp < self.local_ttl:
return data
# Level 2: Redis
if self.redis_client:
redis_data = await self.redis_client.get(cache_key)
if redis_data:
import json
data = json.loads(redis_data)
self.local_cache[cache_key] = (data, now)
return data
return None
async def set_orderbook(
self,
exchange: str,
symbol: str,
depth: int,
data: Dict
):
"""Speichert Order Book im Cache"""
cache_key = f"ob:{exchange}:{symbol}:{depth}"
now = time.time()
# Level 1
self.local_cache[cache_key] = (data, now)
# Level 2
if self.redis_client:
import json
await self.redis_client.setex(
cache_key,
5, # 5 Sekunden TTL
json.dumps(data)
)
async def invalidate(self, exchange: str, symbol: str):
"""Invalidiert alle Cache-Einträge für ein Symbol"""
pattern = f"ob:{exchange}:{symbol}:*"
# Lokaler Cache
keys_to_remove = [k for k in self.local_cache if k.startswith(f"ob:{exchange}:{symbol}:")]
for k in keys_to_remove:
del self.local_cache[k]
# Redis
if self.redis_client:
async for key in self.redis_client.scan_iter(match=pattern):
await self.redis_client.delete(key)
class SmartOrderBookClient:
"""
Intelligenter Order-Book-Client mit:
- Auto-Retry bei Fehlern
- Circuit Breaker Pattern
- Rate Limit Handling
"""
def __init__(
self,
holy_sheep_key: str,
cache: Optional[OrderBookCache] = None
):
self.client = HolySheepMarketDataClient(holy_sheep_key)
self.cache = cache or OrderBookCache()
# Circuit Breaker State
self.failure_count = 0
self.failure_threshold = 5
self.circuit_open = False
self.circuit_reset_time = 30
async def get_orderbook_safe(
self,
exchange: str,
symbol: str,
depth: int = 20
) -> Optional[Dict]:
"""Holt Order Book mit Fehlerbehandlung"""
# Circuit Breaker Check
if self.circuit_open:
if time.time() < self.circuit_reset_time:
# Versuche Cache
return await self.cache.get_orderbook(exchange, symbol, depth)
else:
# Circuit schließen
self.circuit_open = False
self.failure_count = 0
try:
# Cache prüfen
cached = await self.cache.get_orderbook(exchange, symbol, depth)
if cached:
return cached
# API aufrufen
async with self.client:
data = await self.client.get_orderbook(exchange, symbol, depth)
# Erfolg: Cache aktualisieren
await self.cache.set_orderbook(exchange, symbol, depth, data)
self.failure_count = 0
return data
except Exception as e:
self.failure_count += 1
if self.failure_count >= self.failure_threshold:
self.circuit_open = True
self.circuit_reset_time = time.time() + 30
raise
Verwendung mit Redis
async def production_example():
cache = OrderBookCache(redis_url="redis://localhost:6379")
await cache.connect()
client = SmartOrderBookClient(
holy_sheep_key="YOUR_HOLYSHEEP_API_KEY",
cache=cache
)
try:
# Hole Order Book mit Auto-Retry
ob = await client.get_orderbook_safe("binance", "BTC/USDT", depth=20)
print(f"BTC/USD Spread: {ob['spread_bps']:.2f} bps")
finally:
await cache.redis_client.close()
if __name__ == "__main__":
asyncio.run(production_example())
Vergleich: Tardis vs. HolySheep vs. Andere Anbieter
| Anbieter | Preis-Modell | Latenz (P50) | WebSocket | Historische Daten | China-Friendly | Free Tier |
|---|---|---|---|---|---|---|
| HolySheep AI | ¥1=$1, ~$0.42/MTok (DeepSeek) | <50ms | ✓ | ✓ | ✓ WeChat/Alipay | 100¥ Credits |
| Tardis | Ab $99/Monat | ~20ms | ✓ | ✓✓✓ | ✗ Kreditkarte | 5 Tage Trial |
| CoinAPI | Ab $79/Monat | ~80ms | ✓ | ✓✓ | ✗ | 5000 Anfr/Tag |
| Binance API (direkt) | Kostenlos | ~10ms | ✓ | ✓ (7 Tage) | ✓ | Unbegrenzt |
| Kaiko | Ab $500/Monat | ~100ms | ✓ | ✓✓✓ | ✗ | Nein |
Geeignet / nicht geeignet für
Ideal für HolySheep AI:
- Startup-Trading-Unternehmen mit begrenztem Budget
- Algorithmic-Trading-Teams in China und APAC
- Entwickler, die schnelle Prototypen bauen möchten
- Forschungsteams, die Marktdaten für ML-Modelle benötigen
- Multi-Exchange-Aggregatoren, die Kosteneffizienz benötigen
Weniger geeignet für:
- Hochfrequenztrader mit <1ms Latenzanforderungen (direkte Börsen-Anbindung besser)
- Unternehmen mit ausschließlich westlichen Zahlungsmethoden (ohne WeChat/Alipay)
- Regulierte Finanzinstitutionen, die SOC2/ISO27001 erfordern
- Langfristige historische Analysen (Tardis hat tiefere Archive)
Preise und ROI
Basierend auf typischen Nutzungsszenarien habe ich folgende Kostenvergleiche erstellt:
| Nutzer-Typ | Tardis/Monat | HolySheep/Monat | Ersparnis |
|---|---|---|---|
| Einzelentwickler | $99 | ¥50 (~$50) | ~50% |
| Kleines Team (5 Entwickler) | $499 | ¥200 (~$200) | ~60% |
| Startup (10 Entwickler) | $999 | ¥400 (~$400) | ~60% |
| Professioneller Trader | $2.499 | ¥1.000 (~$1.000) | ~60% |
ROI-Analyse: Bei einem durchschnittlichen Entwicklergehalt von ¥30.000/Monat in China und 10 Stunden Zeitersparnis durch schnellere API-Integration beträgt der ROI bereits über 300%.
Warum HolySheep wählen
Nach meiner Erfahrung mit über 15 verschiedenen Marktdatenanbietern bietet HolySheep AI folgende einzigartige Vorteile:
- ¥1=$1 Wechselkurs: Für chinesische Teams bedeutet dies фактически keine Währungsumrechnungsverluste
- <50ms Latenz: Ausreichend für die meisten Strategien außer HFT
- WeChat/Alipay: Bezahlung so einfach wie eine Online-Bestellung
- Kostenlose Credits: 100¥ Startguthaben für Tests und Prototypen
- GPT-4.1 zu $8: Falls Sie auch LLMs für Sentiment-Analyse nutzen — 85%+ günstiger als OpenAI
- DeepSeek V3.2 zu $0.42: Extrem günstig für Bulk-Textverarbeitung
- Deutsche Zeitzone Support: Bürozeiten passend für europäische Kunden
Häufige Fehler und Lösungen
1. Floating-Point-Präzisionsfehler bei Preisen
Problem: Bei Preisen wie 0.00000345 können Rundungsfehler zu falschen Order-Zuordnungen führen.
# FEHLERHAFT:
price = 0.1 + 0.2 # Ergibt 0.30000000000000004
LÖSUNG: Dezimal für Finanzdaten
from decimal import Decimal, ROUND_HALF_UP
def normalize_price(price: float, decimals: int = 8) -> Decimal:
"""Normalisiert Preis zu festem Dezimalformat"""
d = Decimal(str(price))
quantize_str = '0.' + '0' * decimals
return d.quantize(Decimal(quantize_str), rounding=ROUND_HALF_UP)
Test
assert normalize_price(0.1 + 0.2) == Decimal('0.30000000')
assert normalize_price(0.00000345) == Decimal('0.00000345')
2. Race Conditions bei parallelen Updates
Problem: Gleichzeitige WebSocket-Updates können zu inkonsistentem Order-Book-Zustand führen.
import asyncio
from typing import Dict, List
from dataclasses import dataclass, field
from collections import defaultdict
import time
@dataclass
class ThreadSafeOrderBook:
"""Thread-sichere Order-Book-Implementierung mit asyncio Lock"""
symbol: str
bids: Dict[float, float] = field(default_factory=dict)
asks: Dict[float, float] = field(default_factory=dict)
_lock: asyncio.Lock = field(default_factory=asyncio.Lock)
async def update(self, side: str, price: float, quantity: float):
"""Atomare Order-Book-Aktualisierung"""
async with self._lock:
target = self.bids if side == 'buy' else self.asks
if quantity == 0:
target.pop(price, None)
else:
target[price] = quantity
async def batch_update(self, updates: List[tuple]):
"""
Batch-Update für mehrere Levels
Wichtig: Alle Updates in einer Transaktion
"""
async with self._lock:
for side, price, quantity in updates:
target = self.bids if side == 'buy' else self.asks
if quantity == 0:
target.pop(price, None)
else:
target[price] = quantity
async def get_snapshot(self) -> Dict:
"""Thread-sich