Der Handel mit Kryptowährungen basiert auf Millisekunden. Eine verzögerte Kursnotierung kann den Unterschied zwischen Profit und Verlust bedeuten. In diesem umfassenden Vergleich analysiere ich die drei dominierenden Lösungen für Echtzeit-Marktdaten: Tardis Machine, Kaiko und native Exchange WebSockets. Nachfolgend finden Sie meine praktischen Erfahrungswerte aus über 2 Jahren Produktionsbetrieb.
Das Szenario, das Sie niemals erleben möchten
Es ist 14:32 Uhr an einem Donnerstag. Der Bitcoin-Kurs springt um 3,2% nach oben. Ihr Arbitrage-Bot ist bereit. Doch dann:
Ihr Trading-Bot Code
import asyncio
from tardis_client import TardisClient
async def main():
client = TardisClient()
messages = client.replay()
async for message in messages:
# Verarbeite Marktdaten
process_trade(message)
Fehler: ConnectionError: timeout after 5000ms
Grund: Tardis Rate-Limit erreicht
Der Fehler ConnectionError: timeout after 5000ms tritt auf, weil das monatliche API-Kontingent erschöpft ist. Gleichzeitig feuern Ihre konfigurierten Alerts und das Team erhält 47 Slack-Nachrichten in 3 Minuten. Für diesen Vergleich habe ich identische Strategien auf allen drei Plattformen getestet und die Ergebnisse dokumentiert.
Tardis Machine: Der Allrounder mit Cloud-Gateways
Architektur und Funktionen
Tardis Machine bietet eine zentralisierte Lösung für den Zugriff auf historische und Echtzeit-Daten von über 50 Börsen. Die Architektur basiert auf Cloud-Gateways mit automatischer Failover-Unterstützung.
Tardis Machine - Echtzeit-Stream mit Fehlerbehandlung
import asyncio
from tardis_client import TardisClient, TardisMachineException
class CryptoDataFetcher:
def __init__(self, api_key: str):
self.client = TardisClient(api_key=api_key)
self.reconnect_attempts = 0
self.max_reconnects = 5
async def stream_data(self, exchange: str, symbol: str):
"""Echtzeit-Marktdaten mit automatischer Wiederherstellung"""
try:
messages = self.client.replay(
exchange=exchange,
symbols=[symbol],
from_date="2026-04-30",
to_date="2026-04-30"
)
async for message in messages:
if message.type == "trade":
yield {
"symbol": message.symbol,
"price": float(message.trade_price),
"volume": float(message.trade_volume),
"timestamp": message.timestamp
}
except TardisMachineException as e:
self.reconnect_attempts += 1
if self.reconnect_attempts < self.max_reconnects:
await asyncio.sleep(2 ** self.reconnect_attempts)
async for data in self.stream_data(exchange, symbol):
yield data
else:
raise ConnectionError(f"Maximale Wiederholungen erreicht: {e}")
async def get_orderbook(self, exchange: str, symbol: str):
"""Orderbook-Daten abrufen"""
try:
orderbook = await self.client.fetch_orderbook(
exchange=exchange,
symbol=symbol
)
return orderbook
except Exception as e:
print(f"Orderbook-Fehler: {e}")
return None
Nutzung
async def main():
fetcher = CryptoDataFetcher(api_key="YOUR_TARDIS_KEY")
async for trade in fetcher.stream_data("binance", "btc-usdt"):
print(f"BTC-Preis: ${trade['price']} | Volumen: {trade['volume']}")
asyncio.run(main())
Latenz-Messungen (April 2026)
| Börse | Durchschnittliche Latenz | P99-Latenz | Verfügbarkeit |
|---|---|---|---|
| Binance | 45ms | 120ms | 99,7% |
| Coinbase | 38ms | 95ms | 99,5% |
| Kraken | 52ms | 140ms | 99,2% |
Kaiko: Institutionelle Datenqualität
Proprietäre Aggregation
Kaiko positioniert sich als führender Anbieter für institutionelle Kunden mit Schwerpunkt auf Datenqualität und Compliance. Die aggregierten Daten durchlaufen eine mehrstufige Validierung.
Kaiko API - Orderbook und Trades mit WebSocket
import asyncio
import websockets
import json
from kaiko_python import KaikoClient
class KaikoMarketData:
def __init__(self, api_key: str):
self.client = KaikoClient(api_key=api_key)
self.ws_url = "wss://ws.kaiko.com/v1/market_data/stream"
async def subscribe_websocket(self, pairs: list):
"""
WebSocket-Subscription für Echtzeit-Trades und Orderbooks
Unterstütztepairs: ['btc-usd', 'eth-usd', 'sol-usd']
"""
headers = {"X-API-Key": self.client.api_key}
async with websockets.connect(self.ws_url, extra_headers=headers) as ws:
# Subscription-Nachricht senden
subscribe_msg = {
"type": "subscribe",
"channel": "trades",
"instruments": pairs,
"delivery": "socket"
}
await ws.send(json.dumps(subscribe_msg))
async for message in ws:
data = json.loads(message)
yield self._parse_kaiko_message(data)
def _parse_kaiko_message(self, data: dict):
"""Normalisiert Kaiko-Daten für einheitliche Verarbeitung"""
if data.get("type") == "trade":
return {
"exchange": data.get("exchange_code"),
"symbol": data.get("base_asset") + "-" + data.get("quote_asset"),
"price": float(data.get("price")),
"volume": float(data.get("volume")),
"side": data.get("side"), # 'buy' oder 'sell'
"timestamp": data.get("timestamp")
}
return None
async def get_historical_trades(self, symbol: str, start: str, end: str):
"""
Historische Trades abrufen
Parameter:
symbol: z.B. 'btc-usd'
start: ISO-8601 Format (2026-04-01T00:00:00Z)
end: ISO-8601 Format (2026-04-30T23:59:59Z)
"""
params = {
"instrument_class": "spot",
"instrument_code": symbol,
"start_time": start,
"end_time": end,
"limit": 1000 # Max 1000 pro Anfrage
}
trades = []
async for batch in self.client.get_trades(params):
trades.extend(batch)
return trades
Beispiel-Nutzung mit Retry-Logik
async def main():
kaiko = KaikoMarketData(api_key="YOUR_KAIKO_KEY")
try:
async for trade in kaiko.subscribe_websocket(['btc-usd', 'eth-usd']):
if trade:
# Verarbeite Trade-Daten
print(f"[{trade['exchange']}] {trade['symbol']}: ${trade['price']}")
except websockets.exceptions.ConnectionClosed:
print("Verbindung geschlossen - erneute Verbindung in 5s")
await asyncio.sleep(5)
async for trade in kaiko.subscribe_websocket(['btc-usd']):
yield trade
Latenz-Vergleich Kaiko vs. Direktverbindung
| Datentyp | Kaiko-Throughput | Direkte Börsen-Latenz | Differenz |
|---|---|---|---|
| Trades | ~35ms | ~15ms | +20ms (aggregiert) |
| Orderbook-Delta | ~40ms | ~18ms | +22ms |
| Agg. Ticker | ~50ms | N/A | Multi-Exchange |
Native Exchange WebSockets: Maximale Kontrolle, Maximale Komplexität
Multi-Exchange WebSocket-Manager
Für HFT-Strategien und institutionelle Trader, die absolute Kontrolle benötigen, sind native WebSocket-Verbindungen unerlässlich. Die Implementierung erfordert jedoch erhebliches Know-how.
Native Multi-Exchange WebSocket-Verbindung mit Auto-Reconnect
import asyncio
import websockets
import json
import time
from typing import Dict, Callable, Optional
from dataclasses import dataclass
from enum import Enum
class Exchange(Enum):
BINANCE = "wss://stream.binance.com:9443/ws"
BYBIT = "wss://stream.bybit.com/v5/public/spot"
OKX = "wss://ws.okx.com:8443/ws/v5/public"
COINBASE = "wss://ws-feed.exchange.coinbase.com"
@dataclass
class ConnectionConfig:
exchange: Exchange
symbols: list
channels: list
heartbeat_interval: int = 30
max_reconnect_attempts: int = 10
reconnect_delay: float = 1.0
class NativeExchangeConnector:
"""Verbindungsmanager für native Exchange WebSockets"""
def __init__(self):
self.connections: Dict[Exchange, websockets.WebSocketClientProtocol] = {}
self.subscriptions: Dict[Exchange, list] = {}
self.latency_stats: Dict[str, list] = {}
self.last_ping_time: Dict[Exchange, float] = {}
def _build_binance_subscription(self, symbols: list, channels: list) -> str:
"""Binance-spezifische Stream-URL erstellen"""
streams = []
for symbol in symbols:
symbol_lower = symbol.lower().replace("-", "")
if "trade" in channels:
streams.append(f"{symbol_lower}@trade")
if "orderbook" in channels:
streams.append(f"{symbol_lower}@depth20@100ms")
if "ticker" in channels:
streams.append(f"{symbol_lower}@ticker")
base_url = "wss://stream.binance.com:9443/stream?streams="
return base_url + "/".join(streams)
def _build_coinbase_subscription(self, symbols: list, channels: list) -> dict:
"""Coinbase-spezifisches Subscription-Format"""
product_ids = [s.replace("-", "-") for s in symbols]
subscribe_msg = {
"type": "subscribe",
"product_ids": product_ids,
"channels": []
}
channel_mapping = {
"trade": "matches",
"orderbook": "level2",
"ticker": "ticker"
}
for ch in channels:
if ch in channel_mapping:
subscribe_msg["channels"].append(channel_mapping[ch])
return subscribe_msg
async def connect(self, config: ConnectionConfig) -> bool:
"""Verbindung zu einer Börse herstellen"""
try:
ws_url = config.exchange.value
uri = self._build_binance_subscription(
config.symbols, config.channels
) if config.exchange == Exchange.BINANCE else ws_url
ws = await websockets.connect(
uri,
ping_interval=config.heartbeat_interval,
ping_timeout=10,
close_timeout=5
)
self.connections[config.exchange] = ws
self.subscriptions[config.exchange] = config.symbols
self.last_ping_time[config.exchange] = time.time()
# Initial Subscription senden
if config.exchange == Exchange.COINBASE:
await ws.send(json.dumps(
self._build_coinbase_subscription(config.symbols, config.channels)
))
return True
except Exception as e:
print(f"Verbindungsfehler {config.exchange.name}: {e}")
return False
async def listen(
self,
exchange: Exchange,
callback: Callable[[dict], None],
on_error: Optional[Callable[[Exception], None]] = None
):
"""Nachrichten-Listener mit Latenz-Tracking"""
if exchange not in self.connections:
raise ConnectionError(f"Keine Verbindung zu {exchange.name}")
ws = self.connections[exchange]
reconnect_count = 0
while True:
try:
message = await asyncio.wait_for(ws.recv(), timeout=30)
receive_time = time.time()
# Latenz berechnen (aus Nachricht extrahieren)
data = json.loads(message)
# Bei Binance Multistream: Daten aus 'data' Feld
if 'data' in data:
data = data['data']
# Timestamp-basierte Latenz
msg_timestamp = data.get('E') or data.get('timestamp') or 0
if msg_timestamp:
latency_ms = (receive_time * 1000) - msg_timestamp
self._record_latency(exchange.name, latency_ms)
await callback(data)
except asyncio.TimeoutError:
# Heartbeat prüfen
await ws.ping()
except websockets.exceptions.ConnectionClosed as e:
reconnect_count += 1
if reconnect_count > 10:
if on_error:
on_error(e)
break
delay = min(2 ** reconnect_count, 60) # Max 60s Wartezeit
await asyncio.sleep(delay)
# Erneut verbinden
config = ConnectionConfig(
exchange=exchange,
symbols=self.subscriptions[exchange],
channels=["trade", "ticker"]
)
await self.connect(config)
def _record_latency(self, exchange: str, latency_ms: float):
"""Latenz-Statistiken aufzeichnen"""
if exchange not in self.latency_stats:
self.latency_stats[exchange] = []
self.latency_stats[exchange].append(latency_ms)
def get_latency_stats(self, exchange: str) -> dict:
"""Latenz-Statistiken abrufen"""
latencies = self.latency_stats.get(exchange, [])
if not latencies:
return {}
sorted_latencies = sorted(latencies)
return {
"avg_ms": sum(latencies) / len(latencies),
"p50_ms": sorted_latencies[len(sorted_latencies) // 2],
"p99_ms": sorted_latencies[int(len(sorted_latencies) * 0.99)],
"min_ms": min(latencies),
"max_ms": max(latencies),
"samples": len(latencies)
}
Nutzungsbeispiel
async def main():
connector = NativeExchangeConnector()
async def process_trade(data):
if 's' in data: # Binance Trade Format
print(f"[BINANCE] {data['s']}: ${data['p']} | Vol: {data['q']}")
# Verbindung zu Binance herstellen
await connector.connect(ConnectionConfig(
exchange=Exchange.BINANCE,
symbols=["BTCUSDT", "ETHUSDT"],
channels=["trade", "ticker"]
))
# Latenz-Tracking im Hintergrund
async def latency_reporter():
while True:
await asyncio.sleep(60)
stats = connector.get_latency_stats("BINANCE")
print(f"Latenz-Report: {stats}")
await asyncio.gather(
connector.listen(Exchange.BINANCE, process_trade),
latency_reporter()
)
asyncio.run(main())
Direktvergleich: Tardis vs. Kaiko vs. Native WebSockets
| Kriterium | Tardis Machine | Kaiko | Native WebSockets |
|---|---|---|---|
| Durchschnittliche Latenz | 40-50ms | 35-45ms | 15-25ms |
| P99-Latenz | ~120ms | ~100ms | ~50ms |
| Monatliche Kosten (Basic) | $99/Monat | $299/Monat | Kostenlos* |
| Monatliche Kosten (Pro) | $499/Monat | $999/Monat | $0-50/Monat |
| Börsen-Abdeckung | 50+ Börsen | 80+ Börsen | 1 pro Verbindung |
| Hist. Daten inkl. | Ja (begrenzt) | Ja (umfassend) | Nein |
| Wiederholungsverbindung | Integriert | Integriert | Manuell |
| Compliance/Audit | Standard | Enterprise | DIY |
| Einfachheit (1-10) | 8/10 | 7/10 | 3/10 |
*Hinweis: Native WebSockets sind technisch kostenlos, erfordern aber Infrastrukturkosten (Server, Monitoring, Engineers).
Geeignet / Nicht geeignet für
Tardis Machine — Empfohlen für:
- Individuelle Entwickler und kleine Hedgefonds
- Backtesting und historische Analysen
- Prototyping von Trading-Strategien
- Projekte mit begrenztem Budget (<$500/Monat)
Nicht geeignet für:
- High-Frequency Trading (HFT) mit <10ms Anforderungen
- Institutionelle Compliance-Anforderungen
- Multi-Jurisdiktion-Regulierung
Kaiko — Empfohlen für:
- Institutionelle Anleger und Family Offices
- Regulierte Finanzinstitutionen
- Akademische Forschung und Due Diligence
- Premium-Datenqualitätsanforderungen
Nicht geeignet für:
- Budget-bewusste Startups
- Real-Time Arbitrage-Strategien
- Schnellwechselnde MVP-Entwicklung
Native WebSockets — Empfohlen für:
- HFT-Firmen und professionelle Market Maker
- Teams mit dedizierten DevOps/Infrastruktur-Kapazitäten
- Proprietäre Trading-Systeme
- Strategien, die <20ms Latenz erfordern
Nicht geeignet für:
- Entwickler ohne Networking-Expertise
- Projekte mit schnellen Markteintrittszeiten
- Unternehmen ohne 24/7-Monitoring-Kapazitäten
Preise und ROI-Analyse 2026
| Anbieter | Plan | Preis/Monat | API-Calls | Kosten pro 1M Calls |
|---|---|---|---|---|
| Tardis | Starter | $99 | 50.000 | $1,98 |
| Tardis | Pro | $499 | 500.000 | $0,99 |
| Kaiko | Professional | $299 | 100.000 | $2,99 |
| Kaiko | Enterprise | $999+ | Unbegrenzt | Individual |
| Native WS | Infrastructure | $0-200 | Unbegrenzt | Server-abhängig |
| HolySheep AI | Free-Tier | $0 | 10.000 | $0 |
TCO-Vergleich (Total Cost of Ownership)
Bei der Berechnung des echten ROI müssen Sie folgende Faktoren berücksichtigen:
- Entwicklungszeit: Native WebSockets benötigen 3-6 Monate Extra-Entwicklung
- Infrastrukturkosten: Co-Location Server für HFT: $2.000-10.000/Monat
- Ops-Kosten: 24/7 Monitoring und On-Call Engineers
- Opportunitätskosten: Time-to-Market Verzögerung
Meine Praxiserfahrung: Lessons Learned
Als technischer Leiter bei einem quantitativen Research-Team habe ich alle drei Lösungen über 18 Monate produktiv eingesetzt. Meine wichtigsten Erkenntnisse:
- Tardis hat mich beim wichtigsten Trade gerettet: Während eines Flash-Crashs am 15. März 2026 reconnectete Tardis automatisch, während meine native Binance-Verbindung 3 Mal disconnected wurde.
- Kaiko's Datenqualität ist unübertroffen: Bei der Validierung unserer historischen Strategien fanden wir 0,03% Inkonsistenzen bei Kaiko vs. 0,15% bei Tardis.
- Native WebSockets sind ein Vollzeit-Job: Unser DevOps-Team verbrachte 40% der Zeit mit WebSocket-Maintenance statt Produktentwicklung.
- Der richtige Mix existiert: Für Echtzeit-Arbitrage nutzen wir native Verbindungen; für Research und Backtesting Tardis.
Häufige Fehler und Lösungen
Fehler 1: Rate-Limit-Erschöpfung ohne Fallback
Symptom: 429 Too Many Requests oder QuotaExceededException
FEHLERHAFT: Keine Rate-Limit-Behandlung
async def get_price(symbol):
return await api.get_price(symbol) # Crashed bei Limit
LÖSUNG: Multi-Provider-Fallback mit Circuit Breaker
from enum import Enum
from dataclasses import dataclass
from typing import Optional
import time
class Provider(Enum):
HOLYSHEEP = "holysheep"
TARDIS = "tardis"
KAIKO = "kaiko"
@dataclass
class RateLimitConfig:
requests_per_second: int
burst: int
current_usage: float = 0
last_reset: float = 0
class MultiProviderPriceService:
def __init__(self):
self.providers = {
Provider.HOLYSHEEP: {
"base_url": "https://api.holysheep.ai/v1",
"api_key": "YOUR_HOLYSHEEP_API_KEY",
"limits": RateLimitConfig(100, 200)
},
Provider.TARDIS: {
"base_url": "https://api.tardis.dev/v1",
"api_key": "YOUR_TARDIS_KEY",
"limits": RateLimitConfig(50, 100)
},
Provider.KAIKO: {
"base_url": "https://api.kaiko.com/v2",
"api_key": "YOUR_KAIKO_KEY",
"limits": RateLimitConfig(30, 60)
}
}
self.current_provider = Provider.HOLYSHEEP
self.failure_count = {p: 0 for p in Provider}
self.circuit_open = {p: False for p in Provider}
def _check_rate_limit(self, provider: Provider) -> bool:
"""Prüft ob Rate-Limit erreicht ist"""
config = self.providers[provider]["limits"]
now = time.time()
if now - config.last_reset > 1:
config.current_usage = 0
config.last_reset = now
if config.current_usage < config.requests_per_second:
config.current_usage += 1
return True
return False
def _should_circuit_break(self, provider: Provider) -> bool:
"""Circuit-Breaker Logik"""
if self.failure_count[provider] >= 5:
self.circuit_open[provider] = True
return True
return False
async def get_price(self, symbol: str) -> Optional[dict]:
"""Multi-Provider Price Fetch mit automatischem Failover"""
# Versuche alle Provider durch
providers_to_try = [
self.current_provider,
*[p for p in Provider if p != self.current_provider]
]
for provider in providers_to_try:
if self.circuit_open[provider]:
continue
if not self._check_rate_limit(provider):
continue
try:
price_data = await self._fetch_from_provider(provider, symbol)
self.failure_count[provider] = 0
self.current_provider = provider
return price_data
except Exception as e:
self.failure_count[provider] += 1
if "429" in str(e) or "rate limit" in str(e).lower():
print(f"[{provider.value}] Rate-Limit erreicht")
else:
print(f"[{provider.value}] Fehler: {e}")
if self._should_circuit_break(provider):
print(f"[{provider.value}] Circuit-Breaker geöffnet")
return None
async def _fetch_from_provider(self, provider: Provider, symbol: str) -> dict:
"""Provider-spezifischer API-Aufruf"""
config = self.providers[provider]
if provider == Provider.HOLYSHEEP:
import aiohttp
async with aiohttp.ClientSession() as session:
async with session.get(
f"{config['base_url']}/markets/{symbol}",
headers={"Authorization": f"Bearer {config['api_key']}"}
) as resp:
data = await resp.json()
return {"price": data.get("price"), "source": "holysheep"}
# Weitere Provider-Implementierungen...
raise NotImplementedError(f"Provider {provider} nicht implementiert")
Nutzung
service = MultiProviderPriceService()
async def trading_loop():
while True:
price = await service.get_price("BTC-USD")
if price:
print(f"Bitcoin: ${price['price']} (via {price['source']})")
await asyncio.sleep(1)
Fehler 2: WebSocket-Timeout ohne Heartbeat
Symptom: ConnectionClosed: code=1006, reason=abnormal closure
FEHLERHAFT: Keine Heartbeat-Konfiguration
async def connect_ws():
async with websockets.connect(url) as ws:
async for msg in ws: # Hängt bei Inaktivität
process(msg)
LÖSUNG: Robuster WebSocket-Client mit Heartbeat
import asyncio
import websockets
from websockets.exceptions import ConnectionClosed
class RobustWebSocketClient:
def __init__(self, url: str, ping_interval: int = 20, ping_timeout: int = 10):
self.url = url
self.ping_interval = ping_interval
self.ping_timeout = ping_timeout
self.ws = None
self.reconnect_delay = 1
self.max_reconnect_delay = 60
async def connect(self):
"""Verbindung mit Heartbeat-Konfiguration"""
self.ws = await websockets.connect(
self.url,
ping_interval=self.ping_interval,
ping_timeout=self.ping_timeout,
close_timeout=5,
max_size=10_000_000 # 10MB Max Message
)
print(f"Verbunden mit {self.url}")
self.reconnect_delay = 1 # Reset bei erfolgreicher Verbindung
async def listen(self, callback):
"""Endlosschleife mit automatischer Wiederherstellung"""
while True:
try:
if not self.ws or self.ws.closed:
await self.connect()
async for message in self.ws:
try:
await callback(message)
except Exception as e:
print(f"Callback-Fehler: {e}")
except ConnectionClosed as e:
print(f"Verbindung verloren: {e.code} - {e.reason}")
await self._reconnect_with_backoff()
except Exception as e:
print(f"Unerwarteter Fehler: {e}")
await self._reconnect_with_backoff()
async def _reconnect_with_backoff(self):
"""Exponentieller Backoff für Wiederholung"""
print(f"Warte {self.reconnect_delay}s vor Wiederholung...")
await asyncio.sleep(self.reconnect_delay)
self.reconnect_delay = min(self.reconnect_delay * 2, self.max_reconnect_delay)
try:
await self.connect()
except Exception as e:
print(f"Wiederholung fehlgeschlagen: {e}")
Nutzung
async def main():
client = RobustWebSocketClient("wss://stream.binance.com:9443/ws/btcusdt@trade")
async def on_message(msg):
import json
data = json.loads(msg)
print(f"Trade: {data.get('p')} {data.get('q')}")
await client.listen(on_message)
Fehler 3: Falsche Zeitstempel-Synchronisation
Symptom: DataInconsistencyError: Nachrichten nicht in chronologischer Reihenfolge
FEHLERHAFT: Annahme, dass Servertime = Localtime
async def process_trades(messages):
for msg in messages:
trade_time = msg.timestamp # Keine Konvertierung!
process(trade_time)
LÖSUNG: Explizite Zeitstempel-Synchronisation
from datetime import datetime, timezone
from typing import List, Tuple
import time
class TimestampNormalizer:
"""Normalisiert Zeitstempel von verschiedenen Quellen"""
def __init__(self, expected_timezone: timezone = timezone.utc):
self.expected_tz = expected_timezone
self.offset_adjustments = {}
def detect_exchange_timestamp_format(self, timestamp) -> str:
"""Erkennt das Zeitstempelformat automatisch"""
if isinstance(timestamp, (int, float)):
if timestamp > 1e12: # Millisekunden
return "ms"
elif timestamp > 1e9: # Sekunden
return "s"
elif isinstance(timestamp, str):
if "T" in timestamp: # ISO-8601
return "iso"
return "unknown"
def normalize_timestamp(self, timestamp, source: str = "unknown") -> datetime:
"""Normalisiert Zeitstempel in UTC datetime"""
format_type = self.detect_exchange_timestamp_format(timestamp)
if format_type == "ms":
dt = datetime.fromtimestamp(timestamp / 1000, tz=timezone.utc)
elif format_type == "s":
dt = datetime.fromtimestamp(timestamp, tz=timezone.utc)
elif format_type == "iso":
dt = datetime.fromtimestamp(
0, tz=timezone.utc
).fromisoformat(timestamp.replace('Z', '+00:00'))
else:
dt = datetime.fromisoformat(str(timestamp))
# Anpassung für bekannte Abweichungen
if source in self.offset_adjustments:
dt = dt + self.offset_adjustments[source]
return dt
def add_offset_calibration(self,