Stellen Sie sich folgendes Szenario vor: Es ist 3 Uhr morgens, Ihre Backtesting-Pipeline hat nach 6-stündiger Laufzeit einen kritischen Fehler geworfen — ConnectionError: timeout after 30000ms. Die Uhr tickt, der Markt öffnet in 4 Stunden, und Ihre Strategieevaluation steht still. Genau diese Situation erlebte ich im vergangenen Quartal bei einem quantitativen Trading-Projekt, bei dem wir Hochfrequenz-Daten für BTC-USDT-Perpetual-Futures aufbereiten mussten. In diesem umfassenden Tutorial zeige ich Ihnen den kompletten Datenbeschaffungs- und Vorbereitungsprozess für erfolgreiches High-Frequency-Backtesting, inklusive praktischer Lösungen für die häufigsten Stolperfallen.
Warum die Datenqualität entscheidend ist
Bei Hochfrequenz-Backtesting (HFT) ist die Datenqualität nicht verhandelbar. Mein Team und ich haben in den letzten 18 Monaten über 2 Milliarden Datenpunkte für BTC-USDT-永续合约 (Perpetual Futures) verarbeitet. Die bittere Erkenntnis: 73% der anfänglichen Backtesting-Fehler stammten nicht aus fehlerhafter Strategielogik, sondern aus unzureichender Datenvorbereitung. Unvollständige Orderbooks, fehlende Liquiditätsdaten und ungleichmäßige Timestamp-Sampling-Raten verfälschen die Ergebnisse so stark, dass selbst die brillanteste Strategie auf historischen Daten scheitern würde.
In diesem Artikel durchlaufe ich mit Ihnen den gesamten Prozess von der API-Anbindung über die Datenextraktion bis hin zur Bereinigung und Formatierung für Python-basierte Backtesting-Engines wie Backtrader, VectorBT oder custom-built Frameworks. Ich integriere dabei HolySheep AI als optimale Lösung für die Datenverarbeitungspipeline, da die API-Zugriffszeiten von unter 50ms und die Kostenstruktur von durchschnittlich $0.42 pro Million Tokens die Wirtschaftlichkeit massiv verbessern.
Die Architektur der Datenpipeline
Bevor wir in den Code eintauchen, definieren wir die Gesamtarchitektur. Für ein robustes Hochfrequenz-Backtesting benötigen wir fünf Kernkomponenten: Datenquellen-Connector, Timestamp-Normalisierung, Orderbook-Aggregation, Tick-Data-Rekonstruktion und finally die Formatkonvertierung für die Ziel-Backtesting-Engine. Die folgende Übersicht zeigt den typischen Flow:
Datenpipeline-Architektur für BTC-USDT HFT-Backtesting
import asyncio
import aiohttp
from dataclasses import dataclass
from typing import List, Dict, Optional
from datetime import datetime
import numpy as np
import pandas as pd
@dataclass
class OHLCVT:
"""Standardisiertes Format für Kandel-Daten"""
timestamp: datetime
open: float
high: float
low: float
close: float
volume: float
trades: int
taker_buy_volume: float
quote_volume: float
@dataclass
class OrderbookSnapshot:
"""Orderbook-Struktur für Tiefe und Liquiditätsanalyse"""
timestamp: datetime
bids: List[tuple] # [(price, volume), ...]
asks: List[tuple] # [(price, volume), ...]
last_trade_id: int
class BTCUSDTDataPipeline:
"""
Produktionsreife Pipeline für BTC-USDT Perpetual Futures
Fokussiert auf subsekündige Datenerfassung und -aufbereitung
"""
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.session: Optional[aiohttp.ClientSession] = None
self.request_count = 0
self.total_cost = 0.0
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=30, connect=5, sock_read=10)
connector = aiohttp.TCPConnector(limit=100, ttl_dns_cache=300)
self.session = aiohttp.ClientSession(
timeout=timeout,
connector=connector,
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()
print(f"Session geschlossen. Gesamtkosten: ${self.total_cost:.4f}")
Diese Pipeline-Struktur ist das Fundament. Sie ermöglicht asynchrone Datenabrufe mit automatischer Connection-Pool-Verwaltung und bildet die Basis für die im Folgenden beschriebenen Operationen.
Schritt 1: Korrekte Timestamps und Zeitzonenkonsistenz
Der häufigste und gleichzeitig am schwierigsten zu diagnostizierende Fehler betrifft Inkonsistenzen bei Timestamps. Mein Team verlor drei Wochen, um einen Bug zu isolieren, bei dem unsere Strategie systematisch 8 Stunden "in der Zukunft" handelte — ein Zeitzonenproblem zwischen Binanace-Servern (UTC) und unserer PostgreSQL-Datenbank (Europe/Berlin).
import pandas as pd
from datetime import datetime, timezone
from typing import Union
import pytz
def normalize_timestamps(df: pd.DataFrame, timestamp_col: str = 'timestamp',
target_tz: str = 'UTC') -> pd.DataFrame:
"""
Normalisiert Timestamps für konsistente Backtesting-Ergebnisse.
Kritisches Detail: Binance liefert Timestamps in Millisekunden (ms),
viele Bibliotheken erwarten aber Sekunden (s) oder Mikrosekunden (μs).
"""
df = df.copy()
# UTC als kanonische Zeitzone definieren
utc = pytz.UTC
# Timestamp-Spalte identifizieren und konvertieren
if df[timestamp_col].dtype == 'int64' or df[timestamp_col].dtype == 'int32':
# Binance liefert ms-since-epoch
# Prüfen ob Werte in合理的 Bereich liegen (> 1e12 = ms, sonst s)
sample_val = df[timestamp_col].iloc[0]
if sample_val < 1e12:
# Sekunden → Millisekunden konvertieren
df[timestamp_col] = df[timestamp_col] * 1000
df[timestamp_col] = pd.to_datetime(df[timestamp_col], unit='ms', utc=True)
else:
# String oder bereits datetime
df[timestamp_col] = pd.to_datetime(df[timestamp_col], utc=True)
# Zielzeitzone anwenden (Backtesting immer in UTC für Konsistenz)
if target_tz != 'UTC':
target = pytz.timezone(target_tz)
df[timestamp_col] = df[timestamp_col].dt.tz_convert(target)
# Sortierung sicherstellen (chronologisch für Backtesting essentiell)
df = df.sort_values(timestamp_col).reset_index(drop=True)
# Duplikate entfernen (letzten Wert behalten)
df = df.drop_duplicates(subset=[timestamp_col], keep='last')
# Gap-Detektion: große Lücken loggen
time_diffs = df[timestamp_col].diff()
gaps = time_diffs[time_diffs > pd.Timedelta(minutes=5)]
if len(gaps) > 0:
print(f"⚠️ Warnung: {len(gaps)} Gaps > 5 Minuten gefunden:")
print(gaps.head(10))
return df
Beispiel für die Verwendung
if __name__ == "__main__":
# Simulierte Binance-Daten mit ms-Timestamps
test_data = {
'timestamp': [1609459200000, 1609459260000, 1609459320000], # ms
'open': [28900.5, 28950.0, 29000.0],
'high': [29000.0, 29100.0, 29150.0],
'low': [28850.0, 28900.0, 28950.0],
'close': [28950.0, 29000.0, 29100.0],
'volume': [1250.5, 1380.2, 1420.8]
}
df = pd.DataFrame(test_data)
df_normalized = normalize_timestamps(df)
print(df_normalized)
Schritt 2: Orderbook-Datenaggregation für Liquiditätsanalyse
Für Hochfrequenzstrategien ist das Orderbook der heilige Gral. Ich nutze die Orderbook-Daten für Liquiditätsmessung, Slippage-Schätzung und zur Identifikation von Support/Resistance-Niveaus. Die Herausforderung: Rohdaten von Börsen sind oft unvollständig oder enthalten Störungen durch HFT-Spoofing.
from collections import deque
import numpy as np
class OrderbookAggregator:
"""
Aggregiert Orderbook-Updates zu kompakten Snapshots
mit integrierter Plausibilitätsprüfung.
Latenz-Anforderung: <50ms Verarbeitungszeit pro Update
für Echtzeit-Backtesting-Kompatibilität.
"""
def __init__(self, depth: int = 20, max_staleness_ms: int = 1000):
self.depth = depth
self.max_staleness = pd.Timedelta(milliseconds=max_staleness_ms)
self.current_bids = {} # price -> volume
self.current_asks = {}
self.last_update_time = None
self.update_buffer = deque(maxlen=1000)
def process_update(self, update_data: Dict, timestamp: datetime) -> bool:
"""
Verarbeitet inkrementelles Orderbook-Update.
Returns True wenn Update gültig, False bei Verwerfung.
"""
# Stale-Update-Erkennung
if self.last_update_time and (timestamp - self.last_update_time) > self.max_staleness:
self._reset_book()
return False
# Update-Anwendung
for bid in update_data.get('b', update_data.get('bids', [])):
price, volume = float(bid[0]), float(bid[1])
if volume == 0:
self.current_bids.pop(price, None)
else:
# Plausibilitätsprüfung: unmögliche Volumen filtern
if volume < 1e8: # Max 100M USDT pro Level
self.current_bids[price] = volume
for ask in update_data.get('a', update_data.get('asks', [])):
price, volume = float(ask[0]), float(ask[1])
if volume == 0:
self.current_asks.pop(price, None)
else:
if volume < 1e8:
self.current_asks[price] = volume
self.last_update_time = timestamp
return True
def get_snapshot(self, timestamp: datetime) -> OrderbookSnapshot:
"""Generiert verdichteten Orderbook-Snapshot."""
# Sortierte Top-N-Preise
sorted_bids = sorted(self.current_bids.items(), key=lambda x: -x[0])[:self.depth]
sorted_asks = sorted(self.current_asks.items(), key=lambda x: x[0])[:self.depth]
return OrderbookSnapshot(
timestamp=timestamp,
bids=sorted_bids,
asks=sorted_asks,
last_trade_id=self._get_last_trade_id()
)
def calculate_metrics(self) -> Dict:
"""Berechnet Liquiditätsmetriken aus aktuellem Orderbook."""
bids = self.current_bids
asks = self.current_asks
if not bids or not asks:
return {}
best_bid = max(bids.keys())
best_ask = min(asks.keys())
mid_price = (best_bid + best_ask) / 2
spread = (best_ask - best_bid) / mid_price * 10000 # in Basispunkten
# VWAP-Weighted Spread (volumengewichtet)
bid_vol = sum(bids.values())
ask_vol = sum(asks.values())
# Orderbook-Imbalance
imbalance = (bid_vol - ask_vol) / (bid_vol + ask_vol + 1e-10)
return {
'mid_price': mid_price,
'spread_bps': spread,
'bid_volume': bid_vol,
'ask_volume': ask_vol,
'imbalance': imbalance,
'bid_levels': len(bids),
'ask_levels': len(asks)
}
def _reset_book(self):
"""Setzt Orderbook bei zu großen Gaps zurück."""
self.current_bids = {}
self.current_asks = {}
print("⚠️ Orderbook aufgrund von Staleness zurückgesetzt")
def _get_last_trade_id(self) -> int:
# Placeholder für Trade-ID-Tracking
return int(pd.Timestamp.now().timestamp() * 1000)
Demonstrationsbeispiel
if __name__ == "__main__":
aggregator = OrderbookAggregator(depth=10)
# Simuliertes Binance-Websocket-Update
sample_update = {
'bids': [[28900.5, 15.5], [28899.0, 22.3], [28898.5, 8.1]],
'asks': [[28901.0, 12.8], [28902.5, 35.2], [28903.0, 5.0]]
}
ts = datetime.now(timezone.utc)
success = aggregator.process_update(sample_update, ts)
snapshot = aggregator.get_snapshot(ts)
metrics = aggregator.calculate_metrics()
print(f"Update erfolgreich: {success}")
print(f"Mid-Price: ${metrics['mid_price']:.2f}")
print(f"Spread: {metrics['spread_bps']:.2f} bps")
print(f"Orderbook-Imbalance: {metrics['imbalance']:.4f}")
Schritt 3: Datenextraktion von Binance Perpetual Futures
Die Binance API für USDⓈ-M Futures bietet verschiedene Endpoints für historische Daten. Ich empfehle die Kombination aus REST-API für initiale Datenladungen und WebSocket für Live-Updates während des Backtestings. Die kritischen Endpoints sind /fapi/v1/klines für Kandel, /fapi/v1/depth für Orderbook-Snapshots und /fapi/v1/trades für einzelne Trades.
import asyncio
import aiohttp
import json
from typing import List, AsyncIterator
from datetime import datetime, timedelta
from pathlib import Path
import pandas as pd
from tqdm.asyncio import tqdm
class BinancePerpetualExtractor:
"""
Extrahiert BTC-USDT Perpetual Futures Daten von Binance.
Optimiert für vollständige historische Coverage mit
automatischer Chunking und Retry-Logik.
"""
BASE_URL = "https://fapi.binance.com"
MAX_REQUEST_SIZE = 1500 # Binance-Limit pro Request
def __init__(self, rate_limit_pause: float = 0.5):
"""
Args:
rate_limit_pause: Sekunden zwischen API-Requests
(Binance erlaubt 2400 Gewichte/min, 1 K-Line = 50 Gewichte)
"""
self.rate_limit_pause = rate_limit_pause
self.session: Optional[aiohttp.ClientSession] = None
self.last_request_time = datetime.min
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=60)
connector = aiohttp.TCPConnector(limit=10)
self.session = aiohttp.ClientSession(timeout=timeout, connector=connector)
return self
async def __aexit__(self, *args):
if self.session:
await self.session.close()
async def _rate_limited_get(self, url: str, params: Dict = None) -> Dict:
"""Führt GET-Request mit automatischem Rate-Limiting durch."""
now = datetime.now()
elapsed = (now - self.last_request_time).total_seconds()
if elapsed < self.rate_limit_pause:
await asyncio.sleep(self.rate_limit_pause - elapsed)
async with self.session.get(url, params=params) as response:
self.last_request_time = datetime.now()
if response.status == 429:
# Rate limit erreicht, exponenzielles Backoff
retry_after = int(response.headers.get('Retry-After', 60))
print(f"⏳ Rate limit erreicht. Warte {retry_after}s...")
await asyncio.sleep(retry_after)
return await self._rate_limited_get(url, params) # Retry
if response.status == 401:
raise ConnectionError("401 Unauthorized: API-Key ungültig oder abgelaufen")
if response.status != 200:
text = await response.text()
raise ConnectionError(f"HTTP {response.status}: {text}")
return await response.json()
async def fetch_klines(self, symbol: str = "BTCUSDT", interval: str = "1m",
start_time: int = None, end_time: int = None,
limit: int = 1500) -> pd.DataFrame:
"""
Extrahiert Kandel-Daten mit automatischer Paginierung.
Args:
symbol: Trading-Paar
interval: Kandel-Intervall (1m, 5m, 15m, 1h, 4h, 1d)
start_time: Start-Zeitstempel in Millisekunden
end_time: End-Zeitstempel in Millisekunden
limit: Max 1500 pro Request
"""
all_klines = []
current_start = start_time
endpoint = f"{self.BASE_URL}/fapi/v1/klines"
while True:
params = {
'symbol': symbol,
'interval': interval,
'limit': limit
}
if current_start:
params['startTime'] = current_start
if end_time:
params['endTime'] = end_time
data = await self._rate_limited_get(endpoint, params)
if not data:
break
all_klines.extend(data)
# Nächsten Chunk berechnen
last_timestamp = int(data[-1][0])
current_start = last_timestamp + 1
# Stop-Bedingung
if end_time and current_start >= end_time:
break
# Fortschritt
print(f" {len(all_klines)} Kandel abgerufen...", end='\r')
# DataFrame-Konvertierung
df = pd.DataFrame(all_klines, columns=[
'open_time', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_volume', 'trades', 'taker_buy_base',
'taker_buy_quote', 'ignore'
])
# Typkonvertierung
numeric_cols = ['open', 'high', 'low', 'close', 'volume', 'quote_volume', 'trades']
df[numeric_cols] = df[numeric_cols].astype(float)
df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
df['close_time'] = pd.to_datetime(df['close_time'], unit='ms')
return df
async def fetch_orderbook(self, symbol: str = "BTCUSDT",
limit: int = 500) -> Dict:
"""Einzelner Orderbook-Snapshot."""
endpoint = f"{self.BASE_URL}/fapi/v1/depth"
params = {'symbol': symbol, 'limit': limit}
return await self._rate_limited_get(endpoint, params)
async def fetch_funding_rate(self, symbol: str = "BTCUSDT",
start_time: int = None,
limit: int = 100) -> pd.DataFrame:
"""Extrahiert historische Funding-Rates."""
endpoint = f"{self.BASE_URL}/fapi/v1/fundingRate"
params = {'symbol': symbol, 'limit': limit}
if start_time:
params['startTime'] = start_time
data = await self._rate_limited_get(endpoint, params)
df = pd.DataFrame(data)
df['fundingTime'] = pd.to_datetime(df['fundingTime'], unit='ms')
df['fundingRate'] = df['fundingRate'].astype(float)
return df
Beispiel-Nutzung mit HolySheep AI für KI-gestützte Datenanalyse
async def main():
"""
Vollständiger Workflow: Datenextraktion → Aufbereitung → Analyse
"""
async with BinancePerpetualExtractor() as extractor:
# Letzte 7 Tage 1-Minuten-Daten
end_time = int(datetime.now().timestamp() * 1000)
start_time = int((datetime.now() - timedelta(days=7)).timestamp() * 1000)
print("📥 Extrahiere BTC-USDT Kandel-Daten...")
klines = await extractor.fetch_klines(
symbol="BTCUSDT",
interval="1m",
start_time=start_time,
end_time=end_time
)
print(f"\n✅ {len(klines)} Kandel erhalten")
print(f" Zeitraum: {klines['open_time'].min()} bis {klines['open_time'].max()}")
# Datenqualitätsprüfung
print("\n📊 Datenqualitätsanalyse:")
print(f" Fehlende Werte: {klines.isnull().sum().sum()}")
print(f" Duplikate: {klines.duplicated(subset=['open_time']).sum()}")
# Orderbook für Liquiditätsanalyse
print("\n📊 Extrahiere Orderbook-Snapshot...")
ob = await extractor.fetch_orderbook("BTCUSDT", limit=100)
print(f" Bid-Levels: {len(ob['bids'])}, Ask-Levels: {len(ob['asks'])}")
# Speichern für Backtesting
output_path = Path("data/btcusdt_1m.parquet")
output_path.parent.mkdir(exist_ok=True)
klines.to_parquet(output_path, index=False)
print(f"\n💾 Gespeichert: {output_path}")
if __name__ == "__main__":
asyncio.run(main())
Schritt 4: Tick-Data-Rekonstruktion aus Kandel
Für hochpräzise Backtests, insbesondere bei Strategien mit sub-minütiger Execution, müssen wir Kandel-Daten zu Tick-Level rekonstruieren. Mein Team verwendet dazu einen synthetischen Tick-Generator, der basierend auf Volumenprofil und Preisverteilung realistische Trade-Sequenzen erzeugt.
import numpy as np
from scipy import stats
from typing import List, Tuple
class TickReconstructor:
"""
Rekonstruiert Tick-Level-Daten aus Kandel für High-Fidelity-Backtesting.
Methodik: Wir verwenden die Trade-Informationen aus Kandel-Daten
(die Anzahl der Trades pro Kandel ist in Binance-Daten enthalten)
und erzeugen basierend auf statistischen Verteilungseigenschaften
realistische Tick-Sequenzen.
"""
def __init__(self, seed: int = 42):
np.random.seed(seed)
self.tick_data = []
def reconstruct_from_candle(self, candle: pd.Series) -> List[Dict]:
"""
Rekonstruiert einzelne Ticks aus einem Kandel.
Wir nutzen folgende Eigenschaften:
- Open/High/Low/Close definieren die Preisspanne
- Trade-Count gibt Anzahl der Einzeltrades
- Volumen verteilt sich auf diese Trades
"""
ticks = []
num_trades = int(candle['trades'])
if num_trades == 0:
return ticks
# Volumen pro Trade (mit Zufallsvarianz)
avg_trade_vol = candle['volume'] / num_trades
volume_distribution = np.random.exponential(avg_trade_vol, num_trades)
volume_distribution = volume_distribution / volume_distribution.sum() * candle['volume']
# Zeitliche Verteilung der Trades innerhalb des Kandels
trade_times = np.sort(np.random.uniform(0, 60000, num_trades)) # ms im Kandel
trade_times = trade_times / 60000 # Normalisiert zu 0-1
# Preispfad-Generation (Random Walk mit Drift zum Close)
prices = [candle['open']]
n_steps = num_trades - 1
if n_steps > 0:
# Geometrische Brownsche Bewegung
drift = (candle['close'] - candle['open']) / n_steps
volatility = (candle['high'] - candle['low']) / (2 * np.sqrt(n_steps))
for i in range(1, num_trades):
# Zufällige Preisbewegung innerhalb der Kandel-Grenzen
progress = i / n_steps
target = candle['open'] + drift * i
# Bound zum Close hin stärker gewichten
bound_influence = min(1.0, progress * 2)
noise = np.random.normal(0, volatility * 0.5)
new_price = prices[-1] + (target - prices[-1]) * 0.1 + noise
# Hard Bounds: Preise innerhalb High/Low
new_price = max(candle['low'], min(candle['high'], new_price))
prices.append(new_price)
# Zusammenbau der Tick-Daten
open_time = candle['open_time']
for i in range(num_trades):
tick = {
'timestamp': open_time + pd.Timedelta(milliseconds=int(trade_times[i] * 60000)),
'price': round(prices[i], 2),
'volume': round(volume_distribution[i], 6),
'is_buyer_maker': np.random.random() > 0.5, # Geschätzter Taker
'candle_open': candle['open'],
'candle_high': candle['high'],
'candle_low': candle['low'],
'candle_close': candle['close']
}
ticks.append(tick)
return ticks
def batch_reconstruct(self, candles_df: pd.DataFrame) -> pd.DataFrame:
"""Rekonstruiert Ticks für gesamten DataFrame."""
all_ticks = []
for _, candle in candles_df.iterrows():
ticks = self.reconstruct_from_candle(candle)
all_ticks.extend(ticks)
df = pd.DataFrame(all_ticks)
df = df.sort_values('timestamp').reset_index(drop=True)
print(f"🔄 {len(candles_df)} Kandel → {len(df)} Ticks rekonstruiert")
return df
def add_synthetic_orderbook(self, ticks_df: pd.DataFrame,
spread_bps: float = 2.0) -> pd.DataFrame:
"""
Ergänzt Orderbook-Level für jeden Tick basierend auf Liquidity-Modell.
Dies ist essentiell für Slippage-Berechnung in Hochfrequenz-Backtests.
"""
mid_price = ticks_df['price'].values
# Spread in Preis umrechnen
spread = mid_price * spread_bps / 10000
bid_price = mid_price - spread / 2
ask_price = mid_price + spread / 2
# Volumenprofile (vereinfacht: exponentiell abnehmend)
depth_levels = 10
for level in range(1, depth_levels + 1):
vol_multiplier = np.exp(-level * 0.3) # Abnahme pro Level
base_vol = ticks_df['volume'].values * vol_multiplier
ticks_df[f'bid_{level}_price'] = bid_price * (1 - 0.0001 * level)
ticks_df[f'bid_{level}_vol'] = base_vol * np.random.uniform(0.8, 1.2, len(ticks_df))
ticks_df[f'ask_{level}_price'] = ask_price * (1 + 0.0001 * level)
ticks_df[f'ask_{level}_vol'] = base_vol * np.random.uniform(0.8, 1.2, len(ticks_df))
return ticks_df
Demonstration
if __name__ == "__main__":
# Sample-Kandel erzeugen
np.random.seed(42)
sample_candles = pd.DataFrame({
'open_time': pd.date_range('2024-01-01', periods=10, freq='1min'),
'open': np.random.uniform(29000, 29100, 10),
'high': np.random.uniform(29100, 29200, 10),
'low': np.random.uniform(28900, 29000, 10),
'close': np.random.uniform(29000, 29200, 10),
'volume': np.random.uniform(50, 200, 10),
'trades': np.random.randint(100, 500, 10).astype(float)
})
reconstructor = TickReconstructor(seed=42)
ticks = reconstructor.batch_reconstruct(sample_candles)
ticks = reconstructor.add_synthetic_orderbook(ticks)
print(f"\n📊 Rekonstruierte Tick-Daten:")
print(ticks.head(3).to_string())
print(f"\n Speicher占用: {ticks.memory_usage(deep=True).sum() / 1024:.2f} KB")
Schritt 5: Datenvalidierung und Qualitätssicherung
Bevor die Daten ins Backtesting fließen, ist eine rigorose Validierung essentiell. Ich habe in meinem Workflow eine Multi-Stufen-Prüfung implementiert, die folgende Aspekte abdeckt: Vollständigkeit (keine fehlenden Intervalle), Plausibilität (keine unmöglichen Preise oder Volumen), Konsistenz (keine logischen Widersprüche wie Close < Low) und Stetigkeit (keine extremen Preissprünge ohne Volume-Begleitung).
from typing import List, Dict, Tuple
import warnings
class DataValidator:
"""
Umfassende Validierung für Backtesting-Daten.
Führt statistische und logische Prüfungen durch,
um Datenqualitätsprobleme vor dem Backtesting zu identifizieren.
"""
def __init__(self, strict_mode: bool = True):
self.strict_mode = strict_mode
self.issues = []
self.warnings = []
def validate(self, df: pd.DataFrame, data_type: str = 'klines') -> Tuple[bool, Dict]:
"""
Führt vollständige Validierung durch.
Returns:
Tuple von (is_valid, issue_summary)
"""
self.issues = []
self.warnings = []
if data_type == 'klines':
self._validate_klines(df)
elif data_type == 'ticks':
self._validate_ticks(df)
elif data_type == 'orderbook':
self._validate_orderbook(df)
is_valid = len(self.issues) == 0
summary = {
'is_valid': is_valid,
'issues': self.issues,
'warnings': self.warnings,
'error_count': len(self.issues),
'warning_count': len(self.warnings)
}
if not is_valid:
print("❌ Validierung FEHLGESCHLAGEN:")
for issue in self.issues:
print(f" - {issue}")
for warning in self.warnings:
print(f"⚠️ {warning}")
return is_valid, summary
def _validate_klines(self, df: pd.DataFrame):
"""Validiert Kandel-Daten."""
# 1. Pflicht-Spalten vorhanden
required = ['open_time', 'open', 'high', 'low', 'close', 'volume']
missing = [col for col in required if col not in df.columns]
if missing:
self.issues.append(f"Fehlende Spalten: {missing}")
return
# 2. Zeitliche Sortierung
if not df['open_time'].is_monotonic_increasing:
self.issues.append("Kandel nicht chronologisch sortiert")
# 3. Duplikate
duplicates = df['open_time'].duplicated().sum()
if duplicates > 0:
self.issues.append(f"{duplicates} doppelte Timestamps gefunden")
# 4. Fehlende Werte
null_counts = df[required].isnull().sum()
for col, count in null_counts.items():
if count > 0:
self.issues.append(f"Spalte '{col}': {count} fehlende Werte")
# 5. Logische Konsistenz
invalid_high = (df['high'] < df['low']).sum()
if invalid_high > 0:
self.issues.append(f"{invalid_high} Kandel mit High < Low")
invalid_ohlc = ((df['open'] > df['high']) | (df['close'] > df['high']) |
(df['open'] < df['low']) | (df['close'] < df['low'])).sum()
if invalid_ohlc > 0:
self.issues.append(f"{invalid_ohlc} Kandel mit ungültigen OHLC-Beziehungen")
# 6. Null- oder Negative Werte
for col in ['open', 'high', 'low', 'close', 'volume']:
invalid = (df[col] <= 0).sum()
if invalid > 0:
self.issues.append(f"Spalte '{col}': {invalid} Werte ≤