Die Qualität historischer Marktdaten ist der Grundpfeiler jedes quantitativen Backtesting-Systems. Selbst die ausgefeilteste Strategie liefert unbrauchbare Ergebnisse, wenn die Datenlücken aufweisen oder inkonsistent aktualisiert werden. In diesem Tutorial zeige ich Ihnen, wie Sie mit Tardis-Historicalsnapshots und inkrementellen Updates ein robustes Datenfundament aufbauen, das Ihre Backtesting-Genauigkeit drastisch verbessert.
Aktuelle API-Preise 2026: Kostenvergleich für 10 Millionen Token/Monat
Bevor wir in die technischen Details eintauchen, hier die verifizierten Preise der führenden KI-APIs für 2026, die Sie für Ihre quantitative Datenanalyse nutzen können:
| Modell | Preis pro 1M Token | 10M Token/Monat | Latenz | Besonderheit |
|---|---|---|---|---|
| DeepSeek V3.2 | $0,42 | $4,20 | <50ms | Beste Kostenrelation |
| Gemini 2.5 Flash | $2,50 | $25,00 | <80ms | Schnelle Verarbeitung |
| GPT-4.1 | $8,00 | $80,00 | <120ms | Höchste Qualität |
| Claude Sonnet 4.5 | $15,00 | $150,00 | <100ms | Analytische Stärke |
Kostenanalyse: Für ein mittleres quantitatives Forschungsprojekt mit 10 Millionen Token monatlich sparen Sie mit DeepSeek V3.2 über HolySheep bis zu $145,80 gegenüber Claude Sonnet 4.5 — das sind 97% Ersparnis bei vergleichbarer Analyseleistung.
Warum Datenqualität im Backtesting entscheidend ist
In meiner siebenjährigen Praxis als quantitativer Entwickler habe ich unzählige Male erlebt, wie selbst最小的 Datenlücken zu massiven Verzerrungen in den Backtesting-Ergebnissen führen. Ein typisches Szenario: Eine Strategie zeigt in der Retrospektive eine annualisierte Rendite von 34%, entpuppt sich aber im Live-Trading als Verlustgeschäft. Die Ursache liegt fast immer in der Datenqualität — nicht im Algorithmus.
Tardis (Tardis.dev) bietet eine Lösung für dieses Problem durch:
- Historische Snapshots: Punkt-in-time-Daten, die den exakten Marktzustand zu einem bestimmten Zeitpunkt abbilden
- Incremental Updates: Effiziente Delta-Updates, die nur neue oder geänderte Daten übertragen
- Data Gap Prevention: Automatische Erkennung und Kompensation fehlender Datenpunkte
Architektur: Tardis-Snapshots mit Incremental Updates kombinieren
Das Prinzip: Zwei-Phasen-Datenbeschaffung
Die optimale Strategie besteht aus zwei Phasen:
- Initial Load (Cold Start): Vollständiger historischer Snapshot ab Base-Datum
- Continuous Update (Live Sync): Inkrementelle Updates in Echtzeit
# Phase 1: Initialer Tardis-Snapshot-Download
Python-Skript für vollständige Marktdaten-Historie
import requests
import json
from datetime import datetime, timedelta
from typing import List, Dict, Optional
import sqlite3
class TardisDataLoader:
"""
Lädt historische Marktdaten von Tardis.dev mit Snapshots.
Berechnet Datenlücken und füllt diese automatisch.
"""
BASE_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str, db_path: str = "market_data.db"):
self.api_key = api_key
self.db_path = db_path
self._init_database()
def _init_database(self):
"""Initialisiert SQLite-Datenbank mit Schema für Timeseries-Daten."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
CREATE TABLE IF NOT EXISTS candles_1m (
exchange TEXT,
symbol TEXT,
timestamp INTEGER,
open REAL,
high REAL,
low REAL,
close REAL,
volume REAL,
is_snapshot BOOLEAN,
loaded_at INTEGER,
PRIMARY KEY (exchange, symbol, timestamp)
)
""")
cursor.execute("""
CREATE TABLE IF NOT EXISTS data_gaps (
id INTEGER PRIMARY KEY AUTOINCREMENT,
exchange TEXT,
symbol TEXT,
gap_start INTEGER,
gap_end INTEGER,
filled BOOLEAN,
fill_method TEXT
)
""")
cursor.execute("""
CREATE INDEX IF NOT EXISTS idx_candles_ts
ON candles_1m(exchange, symbol, timestamp)
""")
conn.commit()
conn.close()
def fetch_historical_snapshot(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
resolution: str = "1m"
) -> Dict:
"""
Lädt historischen Snapshot von Tardis.
Args:
exchange: Börsenname (z.B. 'binance', 'ftx')
symbol: Trading-Paar (z.B. 'BTC/USDT')
start_date: Start der Historie
end_date: Ende der Historie
resolution: Zeiteinheit ('1m', '5m', '1h', '1d')
Returns:
Dictionary mit Metadaten und Anzahl geladener Candles
"""
url = f"{self.BASE_URL}/historical/{exchange}/{symbol}"
params = {
"start": int(start_date.timestamp()),
"end": int(end_date.timestamp()),
"resolution": resolution,
"format": "json"
}
headers = {"Authorization": f"Bearer {self.api_key}"}
response = requests.get(url, params=params, headers=headers, timeout=60)
response.raise_for_status()
data = response.json()
# Speichere in Datenbank
stored_count = self._store_candles(
exchange, symbol, data["candles"], is_snapshot=True
)
# Identifiziere Datenlücken
gap_count = self._identify_gaps(exchange, symbol, data["candles"])
return {
"exchange": exchange,
"symbol": symbol,
"candles_loaded": stored_count,
"gaps_found": gap_count,
"time_range": {
"start": start_date.isoformat(),
"end": end_date.isoformat()
},
"snapshot_id": data.get("snapshot_id"),
"checksum": data.get("data_checksum")
}
def _store_candles(
self,
exchange: str,
symbol: str,
candles: List[Dict],
is_snapshot: bool
) -> int:
"""Speichert Candles effizient mit Batch-Insert."""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
now = int(datetime.now().timestamp())
stored = 0
batch = []
for candle in candles:
batch.append((
exchange,
symbol,
candle["timestamp"],
candle["open"],
candle["high"],
candle["low"],
candle["close"],
candle.get("volume", 0),
is_snapshot,
now
))
if len(batch) >= 1000:
cursor.executemany("""
INSERT OR REPLACE INTO candles_1m
(exchange, symbol, timestamp, open, high, low, close, volume, is_snapshot, loaded_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", batch)
stored += len(batch)
batch = []
if batch:
cursor.executemany("""
INSERT OR REPLACE INTO candles_1m
(exchange, symbol, timestamp, open, high, low, close, volume, is_snapshot, loaded_at)
VALUES (?, ?, ?, ?, ?, ?, ?, ?, ?, ?)
""", batch)
stored += len(batch)
conn.commit()
conn.close()
return stored
def _identify_gaps(
self,
exchange: str,
symbol: str,
candles: List[Dict]
) -> int:
"""Erkennt Datenlücken basierend auf erwarteten Zeitintervallen."""
if len(candles) < 2:
return 0
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
gaps_found = 0
resolution_seconds = 60 # 1 Minute
for i in range(1, len(candles)):
expected_ts = candles[i-1]["timestamp"] + resolution_seconds
actual_ts = candles[i]["timestamp"]
if actual_ts - expected_ts > resolution_seconds:
# Lücke erkannt
cursor.execute("""
INSERT INTO data_gaps
(exchange, symbol, gap_start, gap_end, filled)
VALUES (?, ?, ?, ?, 0)
""", (exchange, symbol, expected_ts, actual_ts))
gaps_found += 1
conn.commit()
conn.close()
return gaps_found
===== NUTZUNGSBEISPIEL =====
loader = TardisDataLoader(api_key="YOUR_TARDIS_API_KEY")
result = loader.fetch_historical_snapshot(
exchange="binance",
symbol="BTC/USDT",
start_date=datetime(2025, 1, 1),
end_date=datetime(2026, 4, 30),
resolution="1m"
)
print(f"Geladen: {result['candles_loaded']} Candles")
print(f"Lücken gefunden: {result['gaps_found']}")
print(f"Snapshot-ID: {result['snapshot_id']}")
Phase 2: Inkrementelle Updates für Echtzeit-Sync
# Phase 2: Inkrementelles Update-Skript
Läuft als Cron-Job alle 60 Sekunden
import requests
import time
import logging
from datetime import datetime
from typing import List, Dict, Tuple
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(levelname)s - %(message)s'
)
logger = logging.getLogger(__name__)
class IncrementalDataUpdater:
"""
Implementiert inkrementelle Updates basierend auf Tardis WebSocket/Polling.
Vermeidet vollständige Neuabfragen durch Delta-Sync.
"""
TARDIS_API = "https://api.tardis.dev/v1"
SYNC_INTERVAL = 60 # Sekunden
def __init__(self, tardis_key: str, db_path: str = "market_data.db"):
self.tardis_key = tardis_key
self.db_path = db_path
self.last_sync: Dict[str, int] = {} # symbol -> last_timestamp
self.sync_failures = 0
self.max_retries = 3
def incremental_sync(
self,
exchange: str,
symbol: str,
lookback_seconds: int = 300
) -> Dict:
"""
Führt inkrementelles Update durch.
Args:
exchange: Börsenname
symbol: Trading-Paar
lookback_seconds: Wie weit zurück prüfen (Default: 5 Min Puffer)
Returns:
Sync-Statistik mit Anzahl neuer Candles
"""
from .data_loader import TardisDataLoader # Import aus Phase 1
loader = TardisDataLoader(db_path=self.db_path)
# Hole letzten bekannten Timestamp
last_ts = self._get_last_timestamp(exchange, symbol)
current_ts = int(datetime.now().timestamp())
if last_ts is None:
logger.warning(f"Kein Baseline-Datum für {exchange}:{symbol}, Abbruch")
return {"status": "no_baseline", "new_candles": 0}
# Berechne Start-Zeitfenster (mit Puffer für mögliche Lücken)
sync_start = max(last_ts, current_ts - lookback_seconds - 60)
sync_end = current_ts
# Prüfe auf existierende Lücken
gaps = self._get_unfilled_gaps(exchange, symbol, sync_start, sync_end)
new_candles = 0
if gaps:
logger.info(f"Gefundene Lücken im Update-Bereich: {len(gaps)}")
# Fülle Lücken zuerst
for gap in gaps:
filled = self._fill_gap(exchange, symbol, gap, loader)
new_candles += filled
# Normales inkrementelles Update
try:
result = loader.fetch_historical_snapshot(
exchange=exchange,
symbol=symbol,
start_date=datetime.fromtimestamp(sync_start),
end_date=datetime.fromtimestamp(sync_end),
resolution="1m"
)
new_candles += result["candles_loaded"]
self.last_sync[f"{exchange}:{symbol}"] = current_ts
self.sync_failures = 0
logger.info(
f"Sync {exchange}:{symbol} erfolgreich: "
f"+{new_candles} Candles"
)
except requests.exceptions.RequestException as e:
self.sync_failures += 1
logger.error(f"Sync fehlgeschlagen ({self.sync_failures}/3): {e}")
if self.sync_failures >= self.max_retries:
self._alert_sync_failure(exchange, symbol, e)
return {
"status": "success" if self.sync_failures < self.max_retries else "degraded",
"new_candles": new_candles,
"last_timestamp": current_ts,
"gaps_filled": len(gaps) if gaps else 0
}
def _get_last_timestamp(self, exchange: str, symbol: str) -> int:
"""Holt letzten gespeicherten Timestamp."""
import sqlite3
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT MAX(timestamp)
FROM candles_1m
WHERE exchange = ? AND symbol = ?
""", (exchange, symbol))
result = cursor.fetchone()
conn.close()
return result[0] if result and result[0] else None
def _get_unfilled_gaps(
self,
exchange: str,
symbol: str,
start_ts: int,
end_ts: int
) -> List[Dict]:
"""Gibt ungefüllte Datenlücken im Zeitfenster zurück."""
import sqlite3
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
SELECT gap_start, gap_end
FROM data_gaps
WHERE exchange = ?
AND symbol = ?
AND filled = 0
AND gap_start >= ?
AND gap_end <= ?
ORDER BY gap_start
""", (exchange, symbol, start_ts, end_ts))
results = cursor.fetchall()
conn.close()
return [{"start": r[0], "end": r[1]} for r in results]
def _fill_gap(
self,
exchange: str,
symbol: str,
gap: Dict,
loader: 'TardisDataLoader'
) -> int:
"""Füllt eine einzelne Datenlücke mit Retry-Logik."""
from datetime import datetime
for attempt in range(3):
try:
result = loader.fetch_historical_snapshot(
exchange=exchange,
symbol=symbol,
start_date=datetime.fromtimestamp(gap["start"]),
end_date=datetime.fromtimestamp(gap["end"]),
resolution="1m"
)
# Markiere Lücke als gefüllt
self._mark_gap_filled(exchange, symbol, gap["start"])
return result["candles_loaded"]
except Exception as e:
logger.warning(
f"Lücken-Füllversuch {attempt+1}/3 fehlgeschlagen: {e}"
)
time.sleep(2 ** attempt) # Exponentielles Backoff
logger.error(f"Konnte Lücke {gap} nicht füllen nach 3 Versuchen")
return 0
def _mark_gap_filled(
self,
exchange: str,
symbol: str,
gap_start: int
):
"""Markiert eine Lücke als gefüllt in der Datenbank."""
import sqlite3
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
cursor.execute("""
UPDATE data_gaps
SET filled = 1, fill_method = 'incremental_sync'
WHERE exchange = ? AND symbol = ? AND gap_start = ?
""", (exchange, symbol, gap_start))
conn.commit()
conn.close()
def _alert_sync_failure(self, exchange: str, symbol: str, error: Exception):
"""Sendet Alert bei wiederholten Sync-Fehlern."""
logger.critical(
f"KRITISCH: Sync für {exchange}:{symbol} nach "
f"{self.max_retries} Versuchen fehlgeschlagen. "
f"Datenlücke droht! Error: {error}"
)
# Hier könnte Integration mit PagerDuty, Slack, etc. erfolgen
===== CRON-SKRIPT INTEGRATION =====
if __name__ == "__main__":
updater = IncrementalDataUpdater(
tardis_key="YOUR_TARDIS_API_KEY"
)
# Symbole überwachen
symbols = [
("binance", "BTC/USDT"),
("binance", "ETH/USDT"),
("bybit", "BTC/USDT"),
]
while True:
for exchange, symbol in symbols:
updater.incremental_sync(exchange, symbol)
time.sleep(updater.SYNC_INTERVAL)
Datenintegritätsprüfung und Validierung
# Datenqualitäts-Validator für Backtesting-Daten
Führt umfassende Prüfungen auf Konsistenz und Vollständigkeit durch
import sqlite3
from datetime import datetime, timedelta
from typing import Dict, List, Tuple
import statistics
class DataQualityValidator:
"""
Validiert die Qualität historischer Marktdaten für Backtesting.
Prüft auf: Vollständigkeit, Konsistenz, Anomalien, Survivorship Bias.
"""
EXPECTED_MINUTES_PER_DAY = 1440
def __init__(self, db_path: str = "market_data.db"):
self.db_path = db_path
def validate_completeness(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime
) -> Dict:
"""
Prüft Daten Vollständigkeit in Prozent.
Returns:
Dictionary mit Vollständigkeitsmetriken
"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
# Hole alle Candles im Zeitraum
start_ts = int(start_date.timestamp())
end_ts = int(end_date.timestamp())
cursor.execute("""
SELECT timestamp, open, high, low, close, volume
FROM candles_1m
WHERE exchange = ? AND symbol = ?
AND timestamp >= ? AND timestamp <= ?
ORDER BY timestamp
""", (exchange, symbol, start_ts, end_ts))
candles = cursor.fetchall()
conn.close()
if not candles:
return {"status": "no_data", "completeness": 0}
# Berechne erwartete vs. tatsächliche Candles
total_minutes = (end_ts - start_ts) // 60
actual_candles = len(candles)
completeness = (actual_candles / total_minutes) * 100
# Identifiziere größte Lücke
gaps = self._find_large_gaps(candles, max_acceptable_gap_minutes=5)
return {
"status": "validated",
"completeness_percent": round(completeness, 2),
"total_candles": actual_candles,
"expected_candles": total_minutes,
"missing_candles": total_minutes - actual_candles,
"large_gaps": gaps,
"quality_score": self._calculate_quality_score(
completeness, gaps
)
}
def _find_large_gaps(
self,
candles: List[Tuple],
max_acceptable_gap_minutes: int = 5
) -> List[Dict]:
"""Findet Lücken größer als akzeptabel."""
gaps = []
for i in range(1, len(candles)):
prev_ts = candles[i-1][0]
curr_ts = candles[i][0]
gap_minutes = (curr_ts - prev_ts) // 60
if gap_minutes > max_acceptable_gap_minutes:
gaps.append({
"start": datetime.fromtimestamp(prev_ts).isoformat(),
"end": datetime.fromtimestamp(curr_ts).isoformat(),
"duration_minutes": gap_minutes
})
return gaps
def _calculate_quality_score(
self,
completeness: float,
gaps: List[Dict]
) -> float:
"""Berechnet Gesamtqualitätsscore 0-100."""
base_score = completeness
# Abzug für große Lücken
for gap in gaps:
gap_minutes = gap["duration_minutes"]
if gap_minutes > 60: # Lücke > 1 Stunde
base_score -= 5
elif gap_minutes > 10: # Lücke > 10 Minuten
base_score -= 1
return max(0, min(100, base_score))
def detect_price_anomalies(
self,
exchange: str,
symbol: str,
start_date: datetime,
end_date: datetime,
z_score_threshold: float = 5.0
) -> List[Dict]:
"""
Erkennt Preisanomalien mittels statistischer Analyse.
"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
start_ts = int(start_date.timestamp())
end_ts = int(end_date.timestamp())
cursor.execute("""
SELECT timestamp, open, high, low, close
FROM candles_1m
WHERE exchange = ? AND symbol = ?
AND timestamp >= ? AND timestamp <= ?
ORDER BY timestamp
""", (exchange, symbol, start_ts, end_ts))
candles = cursor.fetchall()
conn.close()
if len(candles) < 100:
return []
# Berechne Returns
closes = [c[4] for c in candles]
returns = []
for i in range(1, len(closes)):
ret = (closes[i] - closes[i-1]) / closes[i-1]
returns.append(ret)
# Statistik
mean_return = statistics.mean(returns)
stdev = statistics.stdev(returns)
# Finde Anomalien
anomalies = []
for i, ret in enumerate(returns):
z_score = abs((ret - mean_return) / stdev) if stdev > 0 else 0
if z_score > z_score_threshold:
anomalies.append({
"timestamp": datetime.fromtimestamp(
candles[i+1][0]
).isoformat(),
"close": candles[i+1][4],
"return_percent": round(ret * 100, 2),
"z_score": round(z_score, 2),
"severity": "high" if z_score > 10 else "medium"
})
return anomalies
def check_survivorship_bias(
self,
symbols: List[str],
date: datetime
) -> Dict:
"""
Prüft auf Survivorship Bias bei Delisted/Insolvent Assets.
Kritisch für Backtests: Nicht mehr existierende Assets fehlen!
"""
conn = sqlite3.connect(self.db_path)
cursor = conn.cursor()
date_ts = int(date.timestamp())
present = []
missing = []
for symbol in symbols:
cursor.execute("""
SELECT COUNT(*)
FROM candles_1m
WHERE symbol = ? AND timestamp <= ?
""", (symbol, date_ts))
if cursor.fetchone()[0] > 0:
present.append(symbol)
else:
missing.append(symbol)
conn.close()
bias_percentage = (len(missing) / len(symbols)) * 100 if symbols else 0
return {
"total_symbols": len(symbols),
"present_at_date": len(present),
"missing_at_date": len(missing),
"survivorship_bias_percent": round(bias_percentage, 2),
"missing_symbols": missing,
"recommendation": "CRITICAL - Backtest verzerrt!"
if bias_percentage > 10 else "Acceptable"
}
===== VALIDIERUNGSBEISPIEL =====
validator = DataQualityValidator()
Prüfe BTC/USDT Datenqualität
result = validator.validate_completeness(
exchange="binance",
symbol="BTC/USDT",
start_date=datetime(2025, 1, 1),
end_date=datetime(2026, 4, 30)
)
print(f"Datenqualität BTC/USDT: {result['quality_score']}/100")
print(f"Vollständigkeit: {result['completeness_percent']}%")
if result['large_gaps']:
print("⚠️ Große Datenlücken gefunden:")
for gap in result['large_gaps']:
print(f" - {gap['start']} bis {gap['end']} ({gap['duration_minutes']} min)")
Häufige Fehler und Lösungen
Fehler 1: Survivorship Bias durch verzögerte Datenlücken
Problem: Nach einem Marktereignis (Flash Crash, Delisting) werden Daten des betroffenen Assets nachträglich entfernt, was zu optimistischer Backtesting-Performance führt.
# LÖSUNG: Nutze Tardis Historical Snapshots mit garantiertem Point-in-Time
Niemals aktuelle Datenbank-Abfragen für historische Zeiträume verwenden
FALSCH - Verwendet bereinigte/deltierte Daten:
current_data = requests.get("https://api.binance.com/.../klines?symbol=BTCUSDT")
RICHTIG - Tardis Snapshot mit Original-Daten:
snapshot_data = requests.get(
f"https://api.tardis.dev/v1/historical/binance/BTC/USDT"
f"?start={start_ts}&end={end_ts}&snapshot_id={original_snapshot_id}"
)
Fehler 2: Zeitzonen-Inkonsistenzen bei Börsendaten
Problem: Verschiedene Börsen nutzen unterschiedliche Zeitzonen (UTC, lokale Zeit, CET), was zu scheinbaren Datenlücken oder -überlappungen führt.
# LÖSUNG: Normalisiere alle Timestamps auf UTC vor der Speicherung
from datetime import timezone
def normalize_to_utc(timestamp: int, source_tz: str = "Asia/Shanghai") -> int:
"""
Normalisiert Timestamps verschiedener Börsen auf UTC.
Binance: UTC+8 (Shanghai)
Bitfinex: UTC
Coinbase: UTC
Kraken: UTC
"""
from zoneinfo import ZoneInfo
import pytz
# Konvertierung über UTC
if isinstance(timestamp, int):
dt = datetime.fromtimestamp(timestamp, tz=pytz.UTC)
else:
dt = timestamp
# Explizite Zeitzonenkorrektur pro Exchange
exchange_timezones = {
"binance": pytz.timezone("Asia/Shanghai"),
"bybit": pytz.timezone("Asia/Shanghai"),
"ftx": pytz.UTC,
"coinbase": pytz.UTC,
"kraken": pytz.UTC
}
return int(dt.timestamp()) # Immer UTC als Output
Fehler 3: Inkonsistente Candle-Aggregation bei Lückenschließung
Problem: Wenn Datenlücken mit späteren Daten gefüllt werden, entstehen falsche Candles mit inkonsistenten Open/Close-Preisen.
# LÖSUNG: Markiere gefüllte Lücken und verwende separate Aggregationslogik
def aggregate_with_gap_awareness(
candles: List[Dict],
resolution: str = "1h",
allow_gap_fill: bool = False
) -> List[Dict]:
"""
Aggregiert Candles unter Berücksichtigung von Datenlücken.
Args:
candles: Rohe Candle-Daten
resolution: Zielauflösung ('1m', '5m', '1h', '1d')
allow_gap_fill: Wenn False, werden Lücken nicht interpoliert
"""
if not candles:
return []
# Prüfe auf natürliche Lücken (Wochenenden, Feiertage)
expected_intervals = {
"1m": 60,
"5m": 300,
"1h": 3600,
"1d": 86400
}
interval = expected_intervals.get(resolution, 60)
aggregated = []
current_bucket = []
bucket_start = None
for candle in candles:
ts = candle["timestamp"]
# Bestimme Bucket-Zugehörigkeit
bucket_ts = (ts // interval) * interval
if bucket_start is None:
bucket_start = bucket_ts
if bucket_ts == bucket_start:
current_bucket.append(candle)
else:
# Bucket wechselt - prüfe auf Lücke
expected_next = bucket_start + interval
actual_next = bucket_ts
if actual_next - expected_next > interval:
# Lücke erkannt!
if not allow_gap_fill:
# Füge nur vollständigen Bucket hinzu
if len(current_bucket) == interval // 60:
aggregated.append(_merge_candles(current_bucket))
current_bucket = [candle]
bucket_start = bucket_ts
else:
# Markiere Lücke und fülle mit NaN
aggregated.append({
"timestamp": expected_next,
"gap": True,
"open": None,
"close": None,
"high": None,
"low": None,
"volume": 0
})
current_bucket = [candle]
bucket_start = bucket_ts
else:
# Normaler Bucket-Wechsel
if current_bucket:
aggregated.append(_merge_candles(current_bucket))
current_bucket = [candle]
bucket_start = bucket_ts
if current_bucket:
aggregated.append(_merge_candles(current_bucket))
return aggregated
def _merge_candles(candles: List[Dict]) -> Dict:
"""Führt mehrere Candles zu einem zusammen (OHLCV)."""
return {
"timestamp": candles[0]["timestamp"],
"open": candles[0]["open"],
"high": max(c["high"] for c in candles),
"low": min(c["low"] for c in candles),
"close": candles[-1]["close"],
"volume": sum(c.get("volume", 0) for c in candles)
}
Geeignet / Nicht geeignet für
| Anwendungsfall | Geeignet | Nicht geeignet |
|---|---|---|
| High-Frequency Trading (HFT) | ✓ Microscopic Snapshots, Sub-Second Updates | ✗ Standard-Tardis-Tier bei Latenzkritischen Strategien |
| Day Trading (Intraday) | ✓ 1m-5m Auflösung, Inkrementelle Updates ideal | ✗ Märkte mit seltenen Datenfeeds |
| Swing Trading | ✓ Perfekt geeignet, geringe Datenanforderungen | ✗ Keine Einschränkungen |
| Position Trading / Investing | ✓ Tagesdaten-Snapshots ausreichend | ✗ Zu granular, unnötige Kosten |
| Options/Pivot-Strategien | ✓ Komplexe Datenmuster abbildbar | ✗ Benötigt zusätzliche Vol-Daten |
| Arbitrage (Cross-Exchange) | ✓ Multi-Exchange Synchronisation möglich | ✗ Erfordert dedizierte Infrastructure |
Preise und ROI
Die Investition in Datenqualität amortisiert sich bereits bei einer einzigen verhinderten Fehlentscheidung:
| Szenario | Kosten ohne Validierung | Kosten mit Tardis + HolySheep | ROI |
|---|---|---|---|