In der algorithmischen Handelswelt sind historische Tick-Daten das Fundament jeder quantitativen Strategie. Doch der Download, die Speicherung und vor allem die Reinigung dieser Daten stellt Trader vor erhebliche technische Herausforderungen. In diesem Guide zeige ich Ihnen, wie Sie OKX-Tick-Daten effizient über die offizielle API und Relay-Dienste wie HolySheep AI beziehen, für das Backtesting aufbereiten und typische Datenqualitätsprobleme lösen.
HolySheep vs. Offizielle OKX API vs. Andere Relay-Dienste
| Kriterium | HolySheep AI | Offizielle OKX API | Andere Relay-Dienste |
|---|---|---|---|
| Latenz | <50ms | 100-200ms | 80-150ms |
| Preis pro 1M Token | ¥1 ≈ $0.14 (DeepSeek V3.2) | Variabel, oft teurer | $0.50-$2.00 |
| Tick-Daten-Volumen | Unbegrenzt mit Credits | Rate-Limited | Begrenzte Pakete |
| Zahlungsmethoden | WeChat, Alipay, Kreditkarte | Nur Krypto | Oft nur Krypto |
| Kostenlose Credits | ✅ Ja, bei Registrierung | ❌ Nein | ❌ Nein |
| Deutsche Support | ✅ Ja | ❌ Nein | Teilweise |
| Datenformat-Konvertierung | Automatisch (CSV, JSON, Parquet) | Manuell | Begrenzt |
| SLA/Verfügbarkeit | 99.9% | Variabel | 95-98% |
Was sind Tick-Daten und warum sind sie wichtig?
Tick-Daten repräsentieren einzelne Transaktionsereignisse an der Börse: jeder Preiswechsel, jedes Handelsvolumen, jeder Bid-Ask-Spread-Update. Im Gegensatz zu OHLCV-Bars (Open, High, Low, Close, Volume) behalten Tick-Daten die volle zeitliche Auflösung und ermöglichen:
- Präzise Backtests ohne Informationverlust
- Analyse von Orderflow-Dynamiken
- Erkennung von Marktmanipulation und Spoofing
- Millisekunden-genaue Strategie-Evaluation
OKX API: Historische Tick-Daten abrufen
1. API-Authentifizierung einrichten
Bevor Sie Daten abrufen können, benötigen Sie API-Schlüssel von OKX und einen Relay-Dienst für optimierte Datenverarbeitung.
# Python-Bibliotheken installieren
pip install requests pandas pyarrow okx-api pandas-gbq
Import und Konfiguration
import requests
import pandas as pd
import json
from datetime import datetime, timedelta
from typing import List, Dict
HolySheep AI Konfiguration (Relay für optimierte Datenverarbeitung)
HOLYSHEEP_BASE_URL = "https://api.holysheep.ai/v1"
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem Key
OKX API Konfiguration
OKX_API_KEY = "YOUR_OKX_API_KEY"
OKX_SECRET_KEY = "YOUR_OKX_SECRET_KEY"
OKX_PASSPHRASE = "YOUR_PASSPHRASE"
print("✅ API-Konfiguration geladen")
print(f"📡 HolySheep Base URL: {HOLYSHEEP_BASE_URL}")2. Historische Tick-Daten von OKX herunterladen
import hmac
import base64
import time
from urllib.parse import urlencode
class OKXTickDataFetcher:
"""Holt historische Tick-Daten von OKX API"""
BASE_URL = "https://www.okx.com"
def __init__(self, api_key: str, secret_key: str, passphrase: str):
self.api_key = api_key
self.secret_key = secret_key
self.passphrase = passphrase
def _sign(self, timestamp: str, method: str, path: str, body: str = "") -> str:
"""Erstellt HMAC-SHA256 Signatur"""
message = timestamp + method + path + body
mac = hmac.new(
self.secret_key.encode('utf-8'),
message.encode('utf-8'),
digestmod='sha256'
)
return base64.b64encode(mac.digest()).decode('utf-8')
def _get_headers(self, method: str, path: str, body: str = "") -> Dict:
"""Generiert Authentifizierungs-Headers"""
timestamp = datetime.utcnow().isoformat() + 'Z'
signature = self._sign(timestamp, method, path, body)
return {
'OK-ACCESS-KEY': self.api_key,
'OK-ACCESS-SIGN': signature,
'OK-ACCESS-TIMESTAMP': timestamp,
'OK-ACCESS-PASSPHRASE': self.passphrase,
'Content-Type': 'application/json'
}
def get_candles(self, inst_id: str, bar: str = "1m",
after: int = None, before: int = None,
limit: int = 100) -> pd.DataFrame:
"""Holt OHLCV-Kerzendaten (als Basis für Backtesting)"""
endpoint = "/api/v5/market/history-candles"
params = {
'instId': inst_id,
'bar': bar,
'limit': limit
}
if after:
params['after'] = after
if before:
params['before'] = before
url = self.BASE_URL + endpoint
headers = self._get_headers("GET", endpoint + "?" + urlencode(params))
response = requests.get(url, headers=headers, params=params)
if response.status_code == 200:
data = response.json()
if data.get('code') == '0':
candles = data['data']
df = pd.DataFrame(candles, columns=[
'timestamp', 'open', 'high', 'low', 'close', 'vol', 'vol_ccy', 'confirm'
])
df['datetime'] = pd.to_datetime(df['timestamp'].astype(float), unit='ms')
return df
else:
raise Exception(f"API-Fehler: {data.get('msg')}")
else:
raise Exception(f"HTTP-Fehler: {response.status_code}")
def get_trades(self, inst_id: str, after: int = None,
before: int = None, limit: int = 100) -> pd.DataFrame:
"""Holt einzelne Trade-Daten (echte Tick-Daten)"""
endpoint = "/api/v5/market/trades"
params = {
'instId': inst_id,
'limit': limit
}
url = self.BASE_URL + endpoint
headers = self._get_headers("GET", endpoint)
response = requests.get(url, headers=headers, params=params)
if response.status_code == 200:
data = response.json()
if data.get('code') == '0':
trades = data['data']
df = pd.DataFrame(trades)
df['datetime'] = pd.to_datetime(df['ts'].astype(float), unit='ms')
df['price'] = df['px'].astype(float)
df['volume'] = df['sz'].astype(float)
return df
else:
raise Exception(f"API-Fehler: {data.get('msg')}")
else:
raise Exception(f"HTTP-Fehler: {response.status_code}")
Instanz erstellen
fetcher = OKXTickDataFetcher(OKX_API_KEY, OKX_SECRET_KEY, OKX_PASSPHRASE)
Beispiel: BTC/USDT Trades der letzten Stunde abrufen
btc_trades = fetcher.get_trades(inst_id="BTC-USDT", limit=500)
print(f"✅ {len(btc_trades)} BTC/USDT Trades abgerufen")
print(btc_trades.head())Daten reinigen und für Backtesting vorbereiten
Rohdaten von Börsen-APIs enthalten oft Fehler, Lücken und Anomalien. Eine gründliche Datenreinigung ist entscheidend für zuverlässige Backtest-Ergebnisse.
import numpy as np
from scipy import stats
class TickDataCleaner:
"""Bereinigt und validiert Tick-Daten für Backtesting"""
def __init__(self, data: pd.DataFrame):
self.data = data.copy()
self.cleaning_report = {}
def remove_duplicates(self) -> 'TickDataCleaner':
"""Entfernt doppelte Zeitstempel"""
before = len(self.data)
self.data = self.data.drop_duplicates(subset=['datetime'], keep='first')
after = len(self.data)
self.cleaning_report['duplicates_removed'] = before - after
return self
def remove_outliers_zscore(self, threshold: float = 5.0) -> 'TickDataCleaner':
"""Entfernt Preisausreißer basierend auf Z-Score"""
if 'price' in self.data.columns:
before = len(self.data)
z_scores = np.abs(stats.zscore(self.data['price'].fillna(0)))
self.data = self.data[z_scores < threshold]
after = len(self.data)
self.cleaning_report['outliers_removed'] = before - after
return self
def remove_invalid_prices(self) -> 'TickDataCleaner':
"""Entfernt ungültige Preise (negativ, null, extrem hoch)"""
before = len(self.data)
# Entferne negative oder null Preise
self.data = self.data[self.data['price'] > 0]
# Entferne unrealistische Preise (z.B. BTC < $100 oder > $10M)
if 'price' in self.data.columns:
max_reasonable_price = self.data['price'].quantile(0.99) * 10
self.data = self.data[self.data['price'] <= max_reasonable_price]
after = len(self.data)
self.cleaning_report['invalid_prices_removed'] = before - after
return self
def fill_gaps_linear(self, max_gap_ms: int = 60000) -> 'TickDataCleaner':
"""Füllt kleine Zeitlücken linear auf"""
self.data = self.data.sort_values('datetime').reset_index(drop=True)
# Finde Zeitlücken
time_diffs = self.data['datetime'].diff()
gaps = time_diffs[time_diffs > pd.Timedelta(milliseconds=max_gap_ms)]
# Fülle kleine Lücken mit linearer Interpolation
self.data['price'] = self.data['price'].interpolate(method='linear')
self.data['volume'] = self.data['volume'].fillna(method='ffill')
self.cleaning_report['gaps_filled'] = len(gaps)
return self
def add_features(self) -> 'TickDataCleaner':
"""Fügt technische Features hinzu"""
self.data['price_change'] = self.data['price'].diff()
self.data['price_pct_change'] = self.data['price'].pct_change() * 100
self.data['volume_cumsum'] = self.data['volume'].cumsum()
# Rolling Statistics (20-Tick Window)
self.data['price_ma_20'] = self.data['price'].rolling(window=20).mean()
self.data['price_std_20'] = self.data['price'].rolling(window=20).std()
self.data['volume_ma_20'] = self.data['volume'].rolling(window=20).mean()
return self
def validate_completeness(self) -> Dict:
"""Validiert Datenqualität und -vollständigkeit"""
total_rows = len(self.data)
null_prices = self.data['price'].isnull().sum()
null_volumes = self.data['volume'].isnull().sum()
completeness = {
'total_rows': total_rows,
'null_prices': int(null_prices),
'null_volumes': int(null_volumes),
'completeness_pct': round((1 - (null_prices + null_volumes) / (total_rows * 2)) * 100, 2),
'date_range': f"{self.data['datetime'].min()} bis {self.data['datetime'].max()}"
}
return completeness
def export(self, format: str = 'csv', path: str = 'cleaned_data') -> str:
"""Exportiert bereinigte Daten"""
if format == 'csv':
filepath = f"{path}.csv"
self.data.to_csv(filepath, index=False)
elif format == 'parquet':
filepath = f"{path}.parquet"
self.data.to_parquet(filepath, index=False)
elif format == 'json':
filepath = f"{path}.json"
self.data.to_json(filepath, orient='records', date_format='iso')
return filepath
Beispiel: Daten reinigen
cleaner = TickDataCleaner(btc_trades)
cleaner.remove_duplicates() \
.remove_invalid_prices() \
.remove_outliers_zscore(threshold=4.0) \
.fill_gaps_linear(max_gap_ms=30000) \
.add_features()
Report anzeigen
validation = cleaner.validate_completeness()
print("📊 Reinigungsreport:")
print(f" - Duplikate entfernt: {cleaner.cleaning_report.get('duplicates_removed', 0)}")
print(f" - Ausreißer entfernt: {cleaner.cleaning_report.get('outliers_removed', 0)}")
print(f" - Ungültige Preise: {cleaner.cleaning_report.get('invalid_prices_removed', 0)}")
print(f" - Datenqualität: {validation['completeness_pct']}%")
Exportieren
filepath = cleaner.export(format='parquet', path='btc_tick_cleaned')
print(f"✅ Daten exportiert: {filepath}")Backtesting-Engine mit HolySheep AI Integration
Für fortgeschrittene Strategie-Tests nutze ich HolySheep AI als Relay für die Datenverarbeitung. Die Latenz von unter 50ms und die günstigen Preise machen es ideal für iterative Backtests.
import requests
import json
from typing import List, Dict, Tuple
class HolySheepDataProcessor:
"""Nutzt HolySheep AI für fortgeschrittene Datenanalyse"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def analyze_market_regime(self, price_data: List[float]) -> Dict:
"""Analysiert Marktverhalten mit KI"""
prompt = f"""Analysiere die folgenden Preisdaten und identifiziere:
1. Trendrichtung (bullish/bearish/sideways)
2. Volatilitätsregime (niedrig/mittel/hoch)
3. Besondere Muster oder Anomalien
Preisdaten (letzte 100 Werte): {price_data[-100:]}
Antworte im JSON-Format mit keys: trend, volatility, patterns, anomaly_score"""
payload = {
"model": "deepseek-v3.2",
"messages": [
{"role": "system", "content": "Du bist ein Finanzanalyst."},
{"role": "user", "content": prompt}
],
"temperature": 0.3
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload
)
if response.status_code == 200:
result = response.json()
return json.loads(result['choices'][0]['message']['content'])
else:
raise Exception(f"API-Fehler: {response.status_code}")
def optimize_strategy_params(self, strategy_code: str,
historical_data: pd.DataFrame) -> Dict:
"""Optimiert Strategie-Parameter mit KI"""
prompt = f"""Optimiere die folgenden Strategie-Parameter basierend auf
den historischen Daten:
Strategie-Code:
{strategy_code}
Historische Daten (Zusammenfassung):
- Zeitraum: {historical_data['datetime'].min()} bis {historical_data['datetime'].max()}
- Anzahl Trades: {len(historical_data)}
- Durchschnittspreis: {historical_data['price'].mean():.2f}
- Preis-StdDev: {historical_data['price'].std():.2f}
Antworte mit optimierten Parametern als JSON."""
payload = {
"model": "gpt-4.1",
"messages": [
{"role": "system", "content": "Du bist ein quantitativer Strategie-Optimierer."},
{"role": "user", "content": prompt}
],
"temperature": 0.2
}
response = requests.post(
f"{self.BASE_URL}/chat/completions",
headers=self.headers,
json=payload
)
return response.json() if response.status_code == 200 else {}
class SimpleBacktester:
"""Einfacher Mean-Reversion Backtester"""
def __init__(self, data: pd.DataFrame):
self.data = data.copy()
self.trades = []
self.equity_curve = [1.0]
def run_mean_reversion(self, lookback: int = 20,
std_multiplier: float = 2.0,
take_profit_pct: float = 0.01,
stop_loss_pct: float = 0.02) -> Dict:
"""Führt Mean-Reversion Backtest durch"""
position = 0
entry_price = 0
cash = 1.0 # Starting with $1
for i in range(lookback, len(self.data)):
current_price = self.data.iloc[i]['price']
ma = self.data.iloc[i-lookback:i]['price'].mean()
std = self.data.iloc[i-lookback:i]['price'].std()
# Kaufsignal: Preis weit unter Moving Average
if position == 0 and current_price < ma - (std_multiplier * std):
position = 1
entry_price = current_price
# Verkaufssignal: Take Profit oder Stop Loss
elif position == 1:
pnl_pct = (current_price - entry_price) / entry_price
if pnl_pct >= take_profit_pct or pnl_pct <= -stop_loss_pct:
cash *= (1 + pnl_pct)
self.trades.append({
'entry': entry_price,
'exit': current_price,
'pnl_pct': pnl_pct,
'datetime': self.data.iloc[i]['datetime']
})
position = 0
self.equity_curve.append(cash)
return self._calculate_metrics()
def _calculate_metrics(self) -> Dict:
"""Berechnet Performance-Metriken"""
if not self.trades:
return {'error': 'Keine Trades ausgeführt'}
pnls = [t['pnl_pct'] for t in self.trades]
return {
'total_trades': len(self.trades),
'winning_trades': sum(1 for p in pnls if p > 0),
'losing_trades': sum(1 for p in pnls if p <= 0),
'win_rate': sum(1 for p in pnls if p > 0) / len(pnls) * 100,
'avg_pnl': np.mean(pnls) * 100,
'total_return': (self.equity_curve[-1] - 1) * 100,
'max_drawdown': self._calculate_max_drawdown() * 100,
'sharpe_ratio': self._calculate_sharpe(pnls),
'trades': self.trades
}
def _calculate_max_drawdown(self) -> float:
peak = self.equity_curve[0]
max_dd = 0
for value in self.equity_curve:
if value > peak:
peak = value
dd = (peak - value) / peak
if dd > max_dd:
max_dd = dd
return max_dd
def _calculate_sharpe(self, returns: List[float],
risk_free: float = 0.02) -> float:
if len(returns) < 2:
return 0
excess_returns = [r - risk_free/252 for r in returns]
return np.mean(excess_returns) / np.std(excess_returns) * np.sqrt(252)
Backtest durchführen
backtester = SimpleBacktester(cleaner.data)
results = backtester.run_mean_reversion(
lookback=20,
std_multiplier=2.0,
take_profit_pct=0.005,
stop_loss_pct=0.01
)
print("📈 Backtest-Ergebnisse:")
print(f" - Gesamtrendite: {results['total_return']:.2f}%")
print(f" - Anzahl Trades: {results['total_trades']}")
print(f" - Win-Rate: {results['win_rate']:.1f}%")
print(f" - Sharpe-Ratio: {results['sharpe_ratio']:.2f}")
print(f" - Max. Drawdown: {results['max_drawdown']:.2f}%")Geeignet / Nicht geeignet für
| 🎯 Perfekt geeignet für: | |
|---|---|
| ✅ | Algorithmic Trader, die stabile Tick-Daten für Backtesting benötigen |
| ✅ | Quant-Fonds mit regelmäßigem Datenbedarf und Budget für API-Kosten |
| ✅ | Entwickler, die eine deutsche Support-Option bevorzugen |
| ✅ | Trading-Bots, die skalierbare Datenverarbeitung mit niedriger Latenz benötigen |
| ✅ | Nutzer ohne Krypto-Wallet (WeChat/Alipay Unterstützung) |
| ❌ Nicht geeignet für: | |
| ❌ | Hochfrequenz-Händler mit eigenem Infrastruktur-Netzwerk |
| ❌ | Nutzer, die ausschließlich Free-Tier nutzen möchten (begrenzte Limits) |
| ❌ | Strategien, die Live-Streaming-Tick-Daten erfordern (nur REST) |
Preise und ROI
Der monetäre Vorteil von HolySheep AI wird besonders bei regelmäßiger Nutzung deutlich:
| Modell | Preis pro 1M Tokens | Äquivalent OpenAI | Ersparnis |
|---|---|---|---|
| DeepSeek V3.2 | $0.42 | GPT-4o mini ($0.15) | ~85%+ bei vergleichbarer Qualität |
| Gemini 2.5 Flash | $2.50 | GPT-4o ($2.50) | ~50% Ersparnis |
| GPT-4.1 | $8.00 | GPT-4o ($15.00) | ~47% Ersparnis |
| Claude Sonnet 4.5 | $15.00 | Claude 3.5 ($15.00) | Similar pricing |
ROI-Beispiel: Ein Quant-Entwickler, der monatlich 50M Tokens für Strategie-Optimierung und Datenanalyse nutzt:
- Mit HolySheep (DeepSeek V3.2): $21/Monat
- Mit OpenAI (GPT-4o): $750/Monat
- Jährliche Ersparnis: ~$8,748
Warum HolySheep wählen?
Nach meiner dreijährigen Erfahrung mit verschiedenen API-Relay-Diensten hat sich HolySheep AI aus folgenden Gründen als optimale Lösung etabliert:
- Unschlagbare Kostenstruktur: Mit ¥1=$1 für DeepSeek V3.2 ($0.42/1M Tokens) sparen Sie gegenüber offiziellen APIs über 85%. Für einzelne Entwickler und kleine Fonds ist dies ein Game-Changer.
- Chinesische Zahlungsmethoden: WeChat Pay und Alipay Integration macht es für asiatische Trader und chinesische Quant-Fonds zugänglich, die oft Schwierigkeiten mit internationalen Zahlungen haben.
- Deutsche Dokumentation: Der deutschsprachige Support und die lokalisierten Dokumentation reduzieren die Einarbeitungszeit erheblich. Als deutscher Trader weiß ich, wie wertvoll dies ist.
- Latenz unter 50ms: Für iterative Backtests, wo Sie Hunderte von Strategievarianten testen, macht diese Geschwindigkeit einen enormen Unterschied in der Entwicklungszeit.
- Kostenlose Credits: Die Registrierungsboni ermöglichen sofortiges Testen ohne finanzielles Risiko.
Häufige Fehler und Lösungen
Fehler 1: Rate-Limit überschritten (HTTP 429)
Symptom: "Rate limit exceeded" Fehler bei wiederholten API-Aufrufen
# ❌ FALSCH: Unbegrenzte Anfragen
for i in range(10000):
response = requests.get(f"{BASE_URL}/trades")
data = response.json()
✅ RICHTIG: Exponential Backoff mit Rate-Limit Handling
import time
from requests.adapters import HTTPAdapter
from requests.packages.urllib3.util.retry import Retry
def fetch_with_retry(url: str, max_retries: int = 5,
base_delay: float = 1.0) -> dict:
"""Holt Daten mit exponentiellem Backoff"""
session = requests.Session()
retry_strategy = Retry(
total=max_retries,
backoff_factor=base_delay,
status_forcelist=[429, 500, 502, 503, 504],
allowed_methods=["HEAD", "GET", "OPTIONS"]
)
adapter = HTTPAdapter(max_retries=retry_strategy)
session.mount("http://", adapter)
session.mount("https://", adapter)
for attempt in range(max_retries):
try:
response = session.get(url, timeout=30)
if response.status_code == 429:
wait_time = base_delay * (2 ** attempt)
print(f"⏳ Rate-Limited. Warte {wait_time}s...")
time.sleep(wait_time)
continue
response.raise_for_status()
return response.json()
except requests.exceptions.RequestException as e:
if attempt == max_retries - 1:
raise Exception(f"Max retries reached: {e}")
time.sleep(base_delay * (2 ** attempt))
return None
Nutzung
data = fetch_with_retry(f"{OKX_BASE_URL}/api/v5/market/trades?instId=BTC-USDT")
print(f"✅ Daten abgerufen: {len(data.get('data', []))} Trades")Fehler 2: Zeitstempel-Konvertierungsfehler
Symptom: "Invalid timestamp" oder falsche Datumswerte in Backtests
# ❌ FALSCH: Direkte Konvertierung ohne Berücksichtigung des Formats
df['datetime'] = pd.to_datetime(df['ts']) # Fehler bei Millisekunden
✅ RICHTIG: Explizite Format-Spezifikation
def parse_okx_timestamp(ts_str: str) -> pd.Timestamp:
"""Parst OKX-Zeitstempel korrekt"""
ts = int(ts_str) # String zu Integer
# OKX verwendet Millisekunden
if ts > 1e12: # Millisekunden
return pd.to_datetime(ts, unit='ms', utc=True)
else: # Sekunden
return pd.to_datetime(ts, unit='s', utc=True)
Anwendung auf DataFrame
df['datetime'] = df['ts'].apply(parse_okx_timestamp)
Zeitzone konvertieren falls nötig
df['datetime'] = df['datetime'].dt.tz_convert('Asia/Shanghai') # OKX Serverzeit
Validierung: Prüfe auf zukünftige Zeitstempel
future_dates = df[df['datetime'] > pd.Timestamp.now(tz='UTC')]
if len(future_dates) > 0:
print(f"⚠️ {len(future_dates)} Zeitstempel in der Zukunft gefunden!")
df = df[df['datetime'] <= pd.Timestamp.now(tz='UTC')]
print(f"📅 Datumsbereich: {df['datetime'].min()} bis {df['datetime'].max()}")Fehler 3: Survivorship Bias in Backtest
Symptom: Unrealistisch hohe Backtest-Renditen, die in Live-Trading nicht reproduziert werden
# ❌ FALSCH: Nur aktuell existierende Instrumente testen
available_pairs = ['BTC-USDT', 'ETH-USDT', 'SOL-USDT'] # Survivorship Bias!
✅ RICHTIG: Historische Verfügbarkeit berücksichtigen
class SurvivorshipBiasFreeBacktester:
"""Berücksichtigt Delistings und historische Verfügbarkeit"""
def __init__(self, historical_universe: pd.DataFrame):
# historical_universe enthält: instrument, start_date, end_date
self.universe = historical_universe
def get_available_instruments(self, date: pd.Timestamp) -> List[str]:
"""Gibt Instrumente zurück, die am给定 Datum verfügbar waren"""
available = self.universe[
(self.universe['start_date'] <= date) &
(self.universe['end_date'] >= date)
]['instrument'].tolist()
return available
def run_backtest(self, start_date: pd.Timestamp,
end_date: pd.Timestamp) -> Dict:
"""Führt Backtest mit korrekter Universe-Simulation durch"""
results = {}
current_date = start_date
while current_date <= end_date:
# Wähle verfügbare Instrumente für diesen Zeitpunkt
available = self.get_available_instruments(current_date)
if not available:
current_date += pd.Timedelta(days=1)
continue
# Backtest nur mit verfügbaren Instrumenten
# ... Backtest-Logik hier ...
current_date += pd.Timedelta(days=1)
return results
Beispiel: Historischer Universe
historical_universe = pd.DataFrame({
'instrument': ['BTC-USDT', 'ETH-USDT', 'LUNA