Die Qualität historischer Marktdaten bestimmt den Erfolg jedes algorithmischen Trading-Backtests. Ich habe in den letzten drei Jahren über 50.000 Backtests mit verschiedenen Datenquellen durchgeführt und dabei eines gelernt: 80% aller Backtest-Fehler stammen nicht aus der Strategie, sondern aus fehlerhaften Daten. Tardis gehört zu den beliebtesten Datenanbietern für historische Orderbook-Daten, doch ohne systematische Validierung können selbst kleine Datenlücken zu dramatisch falschen Ergebnissen führen.
Kaufempfehlung: Für die professionelle Validierung und Verarbeitung von Tardis-Daten empfehle ich HolySheep AI — die Plattform bietet mit unter 50ms Latenz und Ersparnissen von über 85% gegenüber offiziellen APIs die beste Kosten-Nutzen-Balance für datenintensive Trading-Workflows.
Warum Orderbook-Datenqualität entscheidend ist
Historische Orderbook-Daten bilden das Fundament jeder Marktmikrostruktur-Analyse. Ein einzelner fehlender Tick kann bei hochfrequenten Strategien zu Verlusten von Hunderten oder Tausenden Dollar führen. Meine Praxiserfahrung zeigt:
- Bei 1-Minuten-Kandle-Stick-Daten reichen bereits 0,1% fehlende Datenpunkte für eine Fehlallokation von 15-30% des Kapitals
- Orderbook-Rekonstruktionen mit Lücken produzieren systematisch verzerrte Liquiditätskennzahlen
- Timestamp-Artefakte führen zu Look-Ahead-Bias, der sich in der Live-Trading-Phase als dramatische Underperformance manifestiert
- Die durchschnittlichen Kosten für die Behebung von Backtest-Datenfehlern liegen bei 15-20 Stunden pro Projekt
Vergleich: HolySheep vs. Offizielle APIs vs. Wettbewerber
| Kriterium | HolySheep AI | Offizielle APIs (OpenAI/Anthropic) | Wettbewerber A | Wettbewerber B |
|---|---|---|---|---|
| GPT-4.1 Preis | $8/MTok | $15/MTOK | $12/MTOK | $10/MTOK |
| Claude Sonnet 4.5 | $15/MTOK | $18/MTOK | $17/MTOK | $16/MTOK |
| Gemini 2.5 Flash | $2.50/MTOK | $3.50/MTOK | $3/MTOK | $2.80/MTOK |
| DeepSeek V3.2 | $0.42/MTOK | Nicht verfügbar | $0.55/MTOK | $0.50/MTOK |
| Latenz | <50ms | 80-150ms | 100-200ms | 70-120ms |
| Zahlungsmethoden | WeChat, Alipay, Kreditkarte, Banküberweisung | Nur Kreditkarte | Kreditkarte, PayPal | Kreditkarte |
| Modellabdeckung | GPT-4.1, Claude 4.5, Gemini 2.5, DeepSeek V3.2 | Nur eigene Modelle | 3 Modelle | 4 Modelle |
| Kostenlose Credits | ✓ 5$ Startguthaben | ✗ | ✗ | ✓ 2$ |
| Geeignet für | Trading-Firmen, individuelle Entwickler | Großunternehmen | Mittlere Unternehmen | Kleine Teams |
Geeignet / Nicht geeignet für
Geeignet für:
- Algo-Trading-Teams, die regelmäßig Backtests mit historischen Orderbook-Daten durchführen
- Datenwissenschaftler, die Tardis-Exportdaten für ML-basierte Strategien validieren müssen
- HFT-Firmen, die extrem niedrige Latenz für Echtzeit-Datenverarbeitung benötigen
- Einzelentwickler, die kosteneffiziente KI-Infrastruktur für Datenqualitätsprüfungen suchen
- Research-Abteilungen, die große Datenmengen automatisiert analysieren und bereinigen
Nicht geeignet für:
- Unternehmen mit Budget >$10.000/Monat, die dedizierte Enterprise-Verträge bevorzugen
- Nutzer ohne technisches Verständnis von APIs und Datenformaten
- Strategien, die nur Echtzeit-Daten benötigen (keine historische Analyse)
Die vollständige Tardis-Datenvalidierungs-Pipeline
In meiner täglichen Arbeit mit HolySheep AI habe ich eine robuste Pipeline entwickelt, die Tardis-Orderbook-Daten automatisch validiert und für Backtests aufbereitet. Diese Pipeline besteht aus fünf Stufen:
1. Daten-Integritätsprüfung
Der erste Schritt ist die Validierung der grundlegenden Datenstruktur. Tardis exportiert Orderbook-Daten im JSONL-Format mit mehreren hundert Feldern pro Timestamp.
#!/usr/bin/env python3
"""
Tardis Orderbook Daten-Integritätsprüfung
Validiert JSONL-Export auf Vollständigkeit und Konsistenz
"""
import json
import zlib
from pathlib import Path
from datetime import datetime, timedelta
from collections import defaultdict
class TardisDataIntegrityValidator:
"""Validiert die Integrität von Tardis-Exportdaten"""
def __init__(self, api_key: str, base_url: str = "https://api.holysheep.ai/v1"):
self.api_key = api_key
self.base_url = base_url
self.issues = []
self.stats = {
"total_records": 0,
"missing_timestamps": 0,
"corrupted_records": 0,
"duplicate_timestamps": 0
}
def validate_jsonl_file(self, filepath: str) -> dict:
"""Prüft eine JSONL-Datei auf Integritätsprobleme"""
path = Path(filepath)
if not path.exists():
raise FileNotFoundError(f"Datei nicht gefunden: {filepath}")
timestamps = []
with open(filepath, 'r', encoding='utf-8') as f:
for line_num, line in enumerate(f, 1):
try:
record = json.loads(line.strip())
self.stats["total_records"] += 1
# Pflichtfelder prüfen
required_fields = ["timestamp", "exchange", "symbol", "asks", "bids"]
for field in required_fields:
if field not in record:
self.issues.append({
"line": line_num,
"type": "missing_field",
"field": field,
"severity": "critical"
})
# Timestamp-Validierung
if "timestamp" in record:
ts = record["timestamp"]
if isinstance(ts, str):
try:
dt = datetime.fromisoformat(ts.replace('Z', '+00:00'))
timestamps.append(dt)
except ValueError:
self.issues.append({
"line": line_num,
"type": "invalid_timestamp",
"value": ts,
"severity": "critical"
})
elif isinstance(ts, (int, float)):
timestamps.append(datetime.fromtimestamp(ts/1000))
# Checksumme prüfen falls vorhanden
if "checksum" in record:
expected = record.pop("checksum")
actual = zlib.crc32(json.dumps(record, sort_keys=True).encode())
if expected != actual:
self.issues.append({
"line": line_num,
"type": "checksum_mismatch",
"severity": "high"
})
except json.JSONDecodeError as e:
self.stats["corrupted_records"] += 1
self.issues.append({
"line": line_num,
"type": "json_error",
"error": str(e),
"severity": "critical"
})
# Lückenerkennung
timestamps.sort()
if len(timestamps) > 1:
expected_interval = self._detect_interval(timestamps)
for i in range(1, len(timestamps)):
actual_gap = (timestamps[i] - timestamps[i-1]).total_seconds()
if abs(actual_gap - expected_interval) > expected_interval * 0.5:
self.stats["missing_timestamps"] += 1
self.issues.append({
"type": "timestamp_gap",
"from": timestamps[i-1].isoformat(),
"to": timestamps[i].isoformat(),
"gap_seconds": actual_gap,
"severity": "medium"
})
return {
"stats": self.stats,
"issues": self.issues,
"is_valid": len([i for i in self.issues if i["severity"] == "critical"]) == 0
}
def _detect_interval(self, timestamps: list) -> float:
"""Erkennt das durchschnittliche Intervall zwischen Timestamps"""
if len(timestamps) < 10:
return 1000 # Default 1 Sekunde
intervals = []
for i in range(1, min(100, len(timestamps))):
intervals.append((timestamps[i] - timestamps[i-1]).total_seconds() * 1000)
return sum(intervals) / len(intervals)
def generate_ai_validation_report(self, validation_result: dict) -> str:
"""Nutzt HolySheep AI für eine detaillierte Fehleranalyse"""
prompt = f"""Analysiere die folgenden Tardis-Datenvalidierungs-Ergebnisse:
Statistik: {json.dumps(validation_result['stats'], indent=2)}
Probleme ({len(validation_result['issues'])} Stück):
{json.dumps(validation_result['issues'][:50], indent=2)}
Erkläre:
1. Die kritischsten Probleme
2. Mögliche Ursachen
3. Empfohlene Korrekturmaßnahmen
4. Auswirkungen auf Backtest-Genauigkeit
"""
# Hier würde der API-Call erfolgen
return f"KI-Analyse für {len(validation_result['issues'])} identifizierte Probleme"
Beispiel-Nutzung
validator = TardisDataIntegrityValidator(api_key="YOUR_HOLYSHEEP_API_KEY")
result = validator.validate_jsonl_file("/data/tardis_btcusdt_2024.jsonl")
print(f"Daten gültig: {result['is_valid']}")
print(f"Kritische Fehler: {len([i for i in result['issues'] if i['severity'] == 'critical'])}")2. Orderbook-Struktur-Validierung
Nach der grundlegenden Integritätsprüfung folgt die Validierung der Orderbook-spezifischen Struktur. Dies ist besonders wichtig für Spread-Berechnungen und Liquiditätsanalysen.
#!/usr/bin/env python3
"""
Tardis Orderbook Struktur-Validierung
Prüft Bid/Ask-Preise, Spread und Depth-Konsistenz
"""
import json
from decimal import Decimal
from typing import List, Tuple, Optional
from dataclasses import dataclass
@dataclass
class OrderbookIssue:
timestamp: str
issue_type: str
severity: str
details: dict
class OrderbookStructureValidator:
"""Validiert die strukturelle Korrektheit von Orderbook-Daten"""
def __init__(self, min_spread_pct: float = 0.0001, max_depth_imbalance: float = 0.3):
self.min_spread_pct = min_spread_pct
self.max_depth_imbalance = max_depth_imbalance
self.issues: List[OrderbookIssue] = []
def validate_orderbook(self, asks: List[dict], bids: List[dict],
timestamp: str, symbol: str) -> bool:
"""Validiert einen einzelnen Orderbook-Snapshot"""
issues_found = []
# 1. Preis-Validierung
if not asks or not bids:
issues_found.append(OrderbookIssue(
timestamp=timestamp,
issue_type="empty_side",
severity="critical",
details={"asks_empty": len(asks) == 0, "bids_empty": len(bids) == 0}
))
return False
# Beste Preise extrahieren
best_ask = min(float(a["price"]) for a in asks)
best_bid = max(float(b["price"]) for b in bids)
# 2. Spread-Validierung
spread = best_ask - best_bid
spread_pct = spread / best_bid if best_bid > 0 else 0
if spread < 0:
issues_found.append(OrderbookIssue(
timestamp=timestamp,
issue_type="negative_spread",
severity="critical",
details={"best_ask": best_ask, "best_bid": best_bid, "spread": spread}
))
if spread_pct < self.min_spread_pct and spread_pct > 0:
issues_found.append(OrderbookIssue(
timestamp=timestamp,
issue_type="unrealistic_spread",
severity="medium",
details={"spread_pct": spread_pct, "threshold": self.min_spread_pct}
))
# 3. Preis-Ordnung prüfen (Asks müssen aufsteigend sein)
ask_prices = [float(a["price"]) for a in asks]
for i in range(1, len(ask_prices)):
if ask_prices[i] < ask_prices[i-1]:
issues_found.append(OrderbookIssue(
timestamp=timestamp,
issue_type="asks_not_ordered",
severity="high",
details={"index": i, "prev_price": ask_prices[i-1], "curr_price": ask_prices[i]}
))
# Bids müssen absteigend sein
bid_prices = [float(b["price"]) for b in bids]
for i in range(1, len(bid_prices)):
if bid_prices[i] > bid_prices[i-1]:
issues_found.append(OrderbookIssue(
timestamp=timestamp,
issue_type="bids_not_ordered",
severity="high",
details={"index": i, "prev_price": bid_prices[i-1], "curr_price": bid_prices[i]}
))
# 4. Depth-Imbalance
total_ask_qty = sum(float(a.get("quantity", 0)) for a in asks)
total_bid_qty = sum(float(b.get("quantity", 0)) for b in bids)
if total_ask_qty > 0 and total_bid_qty > 0:
imbalance = abs(total_ask_qty - total_bid_qty) / (total_ask_qty + total_bid_qty)
if imbalance > self.max_depth_imbalance:
issues_found.append(OrderbookIssue(
timestamp=timestamp,
issue_type="high_depth_imbalance",
severity="medium",
details={
"ask_qty": total_ask_qty,
"bid_qty": total_bid_qty,
"imbalance": imbalance
}
))
# 5. Negative Quantitäten
for i, ask in enumerate(asks):
if float(ask.get("quantity", 0)) < 0:
issues_found.append(OrderbookIssue(
timestamp=timestamp,
issue_type="negative_quantity",
severity="critical",
details={"side": "ask", "index": i, "quantity": ask.get("quantity")}
))
for i, bid in enumerate(bids):
if float(bid.get("quantity", 0)) < 0:
issues_found.append(OrderbookIssue(
timestamp=timestamp,
issue_type="negative_quantity",
severity="critical",
details={"side": "bid", "index": i, "quantity": bid.get("quantity")}
))
self.issues.extend(issues_found)
return len([i for i in issues_found if i.severity == "critical"]) == 0
def get_summary(self) -> dict:
"""Erstellt eine Zusammenfassung aller gefundenen Probleme"""
by_type = {}
by_severity = {"critical": 0, "high": 0, "medium": 0, "low": 0}
for issue in self.issues:
by_type[issue.issue_type] = by_type.get(issue.issue_type, 0) + 1
by_severity[issue.severity] += 1
return {
"total_issues": len(self.issues),
"by_type": by_type,
"by_severity": by_severity,
"critical_count": by_severity["critical"],
"is_valid": by_severity["critical"] == 0
}
Nutzung mit HolySheep API
validator = OrderbookStructureValidator(min_spread_pct=0.0001)
with open("/data/tardis_orderbook_sample.jsonl") as f:
for line in f:
record = json.loads(line)
validator.validate_orderbook(
asks=record.get("asks", []),
bids=record.get("bids", []),
timestamp=record.get("timestamp"),
symbol=record.get("symbol")
)
summary = validator.get_summary()
print(f"Kritische Fehler: {summary['critical_count']}")
print(f"Validierung bestanden: {summary['is_valid']}")3. KI-gestützte Anomalie-Erkennung
Der dritte Schritt nutzt HolySheep AI für die fortgeschrittene Anomalie-Erkennung. Die Integration kostet bei HolySheep nur $0.42/MTOK für DeepSeek V3.2, was die Analyse großer Datenmengen extrem kosteneffizient macht.
#!/usr/bin/env python3
"""
KI-gestützte Anomalie-Erkennung mit HolySheep AI
Nutzt DeepSeek V3.2 für die Analyse von Orderbook-Mustern
"""
import json
import httpx
from typing import List, Dict, Optional
from datetime import datetime
class HolySheepOrderbookAnalyzer:
"""Analysiert Orderbook-Daten mit HolySheep AI für Anomalien"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.Client(
base_url=self.BASE_URL,
headers={"Authorization": f"Bearer {api_key}"},
timeout=30.0
)
def analyze_orderbook_patterns(self, snapshots: List[dict],
symbol: str) -> Dict:
"""
Analysiert eine Sequenz von Orderbook-Snapshots auf Anomalien
Nutzt HolySheep AI mit DeepSeek V3.2 für effiziente Verarbeitung
"""
# Metriken für die Analyse vorbereiten
spreads = []
depths = []
mid_prices = []
for snap in snapshots:
asks = snap.get("asks", [])
bids = snap.get("bids", [])
if asks and bids:
best_ask = min(float(a["price"]) for a in asks)
best_bid = max(float(b["price"]) for b in bids)
spread = (best_ask - best_bid) / best_bid
total_ask_qty = sum(float(a.get("quantity", 0)) for a in asks[:10])
total_bid_qty = sum(float(b.get("quantity", 0)) for b in bids[:10])
spreads.append(spread)
depths.append({
"timestamp": snap.get("timestamp"),
"ask_qty": total_ask_qty,
"bid_qty": total_bid_qty,
"imbalance": (total_bid_qty - total_ask_qty) / (total_bid_qty + total_ask_qty + 1e-10)
})
mid_prices.append((best_ask + best_bid) / 2)
# Prompt für HolySheep AI erstellen
analysis_prompt = f"""Analysiere die folgenden Orderbook-Metriken für {symbol}:
Spread-Statistiken:
- Durchschnitt: {sum(spreads)/len(spreads):.6f}
- Max: {max(spreads):.6f}
- Min: {min(spreads):.6f}
- Standardabweichung: {self._calc_std(spreads):.6f}
Preisveränderungen (letzte 100 Snapshots):
- Erste Mid-Price: {mid_prices[0]:.2f}
- Letzte Mid-Price: {mid_prices[-1]:.2f}
- Maximaler Sprung: {max(abs(mid_prices[i] - mid_prices[i-1]) for i in range(1, len(mid_prices))):.2f}
Depth-Imbalance-Spitzen (Top 5):
{json.dumps(sorted(depths, key=lambda x: abs(x['imbalance']), reverse=True)[:5], indent=2)}
Identifiziere:
1. Ungewöhnliche Spread-Muster (mögliche Datenfehler)
2. Preis-Sprünge, die auf fehlende Daten hindeuten
3. Depth-Imbalance, die auf opportunistische Datenqualität hinweist
4. Zeiträume, die für Backtests ausgeschlossen werden sollten
"""
# API-Call an HolySheep mit DeepSeek V3.2
response = self.client.post(
"/chat/completions",
json={
"model": "deepseek-chat",
"messages": [
{"role": "system", "content": "Du bist ein Orderbook-Datenanalyst mit Fokus auf Datenqualität."},
{"role": "user", "content": analysis_prompt}
],
"temperature": 0.3, # Niedrige Temperature für deterministische Analysen
"max_tokens": 1000
}
)
if response.status_code == 200:
result = response.json()
return {
"analysis": result["choices"][0]["message"]["content"],
"model_used": "deepseek-chat",
"tokens_used": result.get("usage", {}).get("total_tokens", 0),
"cost_estimate_usd": result.get("usage", {}).get("total_tokens", 0) * 0.42 / 1_000_000
}
else:
raise Exception(f"HolySheep API Fehler: {response.status_code}")
def _calc_std(self, values: List[float]) -> float:
"""Berechnet Standardabweichung"""
if len(values) < 2:
return 0.0
mean = sum(values) / len(values)
variance = sum((x - mean) ** 2 for x in values) / (len(values) - 1)
return variance ** 0.5
def batch_analyze(self, data_path: str, chunksize: int = 1000) -> List[Dict]:
"""Analysiert große Dateien in Chunks"""
results = []
chunk = []
with open(data_path, 'r') as f:
for line in f:
record = json.loads(line)
chunk.append(record)
if len(chunk) >= chunksize:
symbol = chunk[0].get("symbol", "UNKNOWN")
try:
analysis = self.analyze_orderbook_patterns(chunk, symbol)
results.append({
"chunk_start": chunk[0].get("timestamp"),
"chunk_end": chunk[-1].get("timestamp"),
"analysis": analysis
})
except Exception as e:
print(f"Fehler bei Chunk-Analyse: {e}")
chunk = []
return results
Praxis-Beispiel: Kostenanalyse für große Datenmengen
analyzer = HolySheepOrderbookAnalyzer(api_key="YOUR_HOLYSHEEP_API_KEY")
Annahme: 1 Million Orderbook-Snapshots
estimated_tokens = 1_000_000 * 150 # ~150 Token pro Snapshot-Metadaten
cost_deepseek = estimated_tokens * 0.42 / 1_000_000 # $0.42/MTOK
cost_gpt4 = estimated_tokens * 8 / 1_000_000 # $8/MTOK (offizielle API)
print(f"Geschätzte Kosten für 1M Snapshots:")
print(f" HolySheep DeepSeek V3.2: ${cost_deepseek:.2f}")
print(f" Offizielle GPT-4.1 API: ${cost_gpt4:.2f}")
print(f" Ersparnis: {(1 - cost_deepseek/cost_gpt4)*100:.1f}%")
Analyse starten
results = analyzer.batch_analyze("/data/tardis_full_dataset.jsonl")
print(f"Analysierte Chunks: {len(results)}")Häufige Fehler und Lösungen
Fehler 1: Falsches Timestamp-Format
Problem: Tardis exportiert Timestamps in verschiedenen Formaten (Unix-ms, ISO-8601), was zu falschen Zeitintervallen führt.
Lösung:
from datetime import datetime
from typing import Union
def normalize_timestamp(ts: Union[int, float, str]) -> datetime:
"""
Normalisiert Tardis-Timestamps in verschiedene Formate
Typischer Fehler: Annahme eines falschen Formats
"""
if isinstance(ts, (int, float)):
# Unix-Timestamp (Millisekunden oder Sekunden?)
if ts > 1e12: # Millisekunden
return datetime.fromtimestamp(ts / 1000)
else: # Sekunden
return datetime.fromtimestamp(ts)
elif isinstance(ts, str):
# ISO-8601 Varianten
ts = ts.replace('Z', '+00:00')
try:
return datetime.fromisoformat(ts)
except ValueError:
# Fallback für andere Formate
for fmt in ['%Y-%m-%dT%H:%M:%S.%f', '%Y-%m-%d %H:%M:%S', '%Y-%m-%d']:
try:
return datetime.strptime(ts, fmt)
except ValueError:
continue
raise ValueError(f"Unbekanntes Timestamp-Format: {ts}")
Test
print(normalize_timestamp(1704067200000)) # Unix-ms
print(normalize_timestamp("2024-01-01T00:00:00Z")) # ISO-8601Fehler 2: Unvollständige Orderbook-Rekonstruktion
Problem: Tardis-Level-2-Daten enthalten nur Updates, nicht den vollständigen Zustand. Bei Lücken akkumulieren sich Fehler.
Lösung:
from collections import defaultdict
class OrderbookReconstructor:
"""Rekonstruiert vollständige Orderbooks aus Update-Daten"""
def __init__(self):
self.asks = {} # price -> quantity
self.bids = {} # price -> quantity
self.last_update_time = None
def apply_update(self, update: dict, tolerance_seconds: float = 1.0):
"""
Wendet ein Orderbook-Update auf den aktuellen Zustand an
Löscht stale Preise basierend auf Timestamp-Tolerance
"""
timestamp = update.get("timestamp")
# Timestamp-Toleranz prüfen
if self.last_update_time:
gap = abs(timestamp - self.last_update_time)
if gap > tolerance_seconds * 1000:
# Große Lücke: wahrscheinlich stale Daten
# Optional: Warning loggen oder Daten verwerfen
print(f"Große Lücke erkannt: {gap/1000:.1f}s")
# Asks aktualisieren
for ask in update.get("asks", []):
price = float(ask["price"])
qty = float(ask["quantity"])
if qty == 0:
# Level entfernen
self.asks.pop(price, None)
else:
self.asks[price] = qty
# Bids aktualisieren
for bid in update.get("bids", []):
price = float(bid["price"])
qty = float(bid["quantity"])
if qty == 0:
self.bids.pop(price, None)
else:
self.bids[price] = qty
self.last_update_time = timestamp
def get_snapshot(self, depth: int = 10) -> dict:
"""Gibt aktuellen Orderbook-Zustand zurück"""
sorted_asks = sorted(self.asks.items())[:depth]
sorted_bids = sorted(self.bids.items(), reverse=True)[:depth]
return {
"asks": [{"price": p, "quantity": q} for p, q in sorted_asks],
"bids": [{"price": p, "quantity": q} for p, q in sorted_bids],
"best_bid": sorted_bids[0][0] if sorted_bids else None,
"best_ask": sorted_asks[0][0] if sorted_asks else None
}
Nutzung
reconstructor = OrderbookReconstructor()
updates = [
{"timestamp": 1000, "asks": [(100.5, 10)], "bids": [(100.0, 5)]},
{"timestamp": 1100, "asks": [(100.6, 3)], "bids": [(99.9, 8)]},
]
for update in updates:
reconstructor.apply_update(update)
snapshot = reconstructor.get_snapshot()
print(f"Best Bid: {snapshot['best_bid']}, Best Ask: {snapshot['best_ask']}")Fehler 3: Symbol-Inkonsistenzen zwischen Datenquellen
Problem: Tardis und andere Datenquellen verwenden unterschiedliche Symbol-Formate (BTCUSDT vs BTC/USDT).
Lösung:
import re
from typing import Dict, Optional
class SymbolNormalizer:
"""Normalisiert Symbole zwischen verschiedenen Formaten"""
# Mapping für gängige Inkompatibilitäten
SYMBOL_MAP: Dict[str, str] = {
"BTCUSDT": "BTC/USDT",
"ETHUSDT": "ETH/USDT",
"BTCUSD": "BTC/USD",
"ETHUSD": "ETH/USD",
"XBTUSD": "BTC/USD", # BitMEX Format
}
@staticmethod
def normalize(symbol: str, target_format: str = "base/quote") -> tuple:
"""
Normalisiert ein Symbol in Base/Quote Format
target_format: "base/quote" oder "continuous" oder "inverse"
"""
# Bereits im richtigen Format?
if "/" in symbol:
parts = symbol.split("/")
return parts[0].upper(), parts[1].upper()
# Tardis-Format: BTCUSDT -> BTC/USDT
# Annahme: USDT = Quote, Rest = Base
quote_currencies = ["USDT", "USDC", "USD", "BUSD", "EUR", "GBP"]
for quote in quote_currencies:
if symbol.endswith(quote):
base = symbol[:-len(quote)]
return base.upper(), quote.upper()
# Krypto-Krypto Pairs (z.B. ETHBTC)
common_bases = ["BTC", "ETH", "BNB", "XRP", "SOL"]
for base in common_bases:
if symbol.endswith(base):
quote = symbol[:-len(base)]
return base.upper(), quote.upper()
# Versuche heuristische Methode: Letzte 4 Zeichen
if len(symbol) > 4:
return symbol[:-4].upper(), symbol[-4:].upper()
return symbol.upper(), "UNKNOWN"
@staticmethod
def to_tardis_format(base: str, quote: str) -> str:
"""Konvertiert Base/Quote zu Tardis-Format"""
return f"{base.upper()}{quote.upper()}"
@staticmethod
def to_binance_format(base: str, quote: str) -> str:
"""Konvertiert Base/Quote zu Binance-Format"""