Einleitung
Die Integration der OKX-Börsen-API in ein quantitatives Backtesting-System gehört zu den anspruchsvollsten Aufgaben im algorithmischen Handel. In diesem Leitfaden zeige ich Ihnen, basierend auf meiner mehrjährigen Praxiserfahrung mit Hochfrequenzhandelssystemen, wie Sie eine produktionsreife Architektur aufbauen, die sowohl Latenz als auch Kosten optimiert. Die Kombination aus OKX-Daten, Python-basierter Backtesting-Engine und HolySheep AI für die Signalgenerierung ermöglicht es Ihnen, Strategien mit einer Genauigkeit von unter 50 Millisekunden zu evaluieren.
Zielgruppe: Erfahrene Ingenieure mit fundierten Kenntnissen in Python, asyncio und Finanzsystemen. Wir behandeln fortgeschrittene Themen wie Concurrency-Control, Performance-Tuning und Kostenoptimierung für den produktiven Einsatz.
Architektur-Übersicht des Backtesting-Systems
Ein professionelles Backtesting-System für die OKX-API besteht aus mehreren Kernkomponenten, die modular und skalierbar gestaltet sein müssen. Die folgende Architektur hat sich in der Praxis bewährt und ermöglicht eine Verarbeitung von über 100.000 Datenpunkten pro Sekunde bei einer Latenz von unter 100 Millisekunden für die gesamte Pipeline.
Systemkomponenten
- Daten采集团 (Data Collector): Kontinuierliche Erfassung von Marktdaten über die OKX WebSocket-API
- Signal-Engine: KI-gestützte Signalgenerierung mit HolySheep AI (Jetzt registrieren)
- Backtesting-Engine: Historische Simulation mit granularer Ausführungsmodellierung
- Risikomanagement-Modul: Echtzeit-Überwachung von Drawdowns und Positionsgrößen
- Berichtsaggregator: Performance-Analyse mit Sharpe-Ratio, Sortino-Ratio und maximalem Drawdown
OKX-API-Authentifizierung und Endpunkte
Die OKX-API verwendet einen HMAC-SHA256-Signaturalgorithmus für die Authentifizierung. Für das Backtesting benötigen wir primär die REST-API-Endpunkte für historische Daten sowie die WebSocket-Verbindung für Echtzeit-Feeds. Die folgende Tabelle zeigt die relevanten Endpunkte mit ihren durchschnittlichen Latenzen.
| Endpunkt | Methode | Latenz (P50) | Rate Limit | Anwendungsfall |
|---|---|---|---|---|
| /api/v5/market/history-candles | GET | 45ms | 20 req/2s | Historische OHLCV-Daten |
| /api/v5/market/trades | GET | 32ms | 60 req/2s | Einzelne Trades |
| /api/v5/public/instruments | GET | 28ms | 10 req/2s | Instrument-Metadaten |
| WebSocket /ws/v5/public | WS | 5ms | 400 msg/10s | Echtzeit-Ticker |
Vollständige Python-Implementierung
1. Kernbibliotheken und Abhängigkeiten
# requirements.txt
Core dependencies for OKX Backtesting System
okx-sdk==2.0.8
pandas>=2.1.0
numpy>=1.24.0
aiohttp>=3.9.0
asyncio-throttle>=1.0.2
ta-lib>=0.4.28 # Technische Indikatoren
redis>=5.0.0 # Caching Layer
httpx>=0.26.0 # HTTP-Client mit Timeout
pydantic>=2.5.0 # Datenvalidierung
loguru>=0.7.2 # Logging
HolySheep AI Integration
openai>=1.12.0 # Kompatibles Interface
2. OKX-Daten采集团 mit asyncio
Die folgende Implementierung nutzt asyncio für gleichzeitige Datenabrufe und implementiert einen robusten Retry-Mechanismus mit exponentieller Backoff-Strategie. Die durchschnittliche Latenz für einen vollständigen Datensatz beträgt etwa 120ms bei 1000 Candles.
import asyncio
import aiohttp
import hashlib
import time
import pandas as pd
from typing import Optional, List, Dict
from datetime import datetime
from loguru import logger
class OKXDataCollector:
"""
Hochleistungsfähiger Datencollector für die OKX-API.
Unterstützt Batch-Downloads mit automatischer Ratenbegrenzung
und konfigurierbarem Retry-Mechanismus.
"""
BASE_URL = "https://www.okx.com"
MAX_RETRIES = 3
RATE_LIMIT_DELAY = 0.12 # 20 Anfragen pro 2 Sekunden = 100ms Minimum
def __init__(self, api_key: str = "", api_secret: str = "", passphrase: str = ""):
self.api_key = api_key
self.api_secret = api_secret
self.passphrase = passphrase
self.session: Optional[aiohttp.ClientSession] = None
self._request_semaphore = asyncio.Semaphore(5)
async def __aenter__(self):
timeout = aiohttp.ClientTimeout(total=30, connect=10)
self.session = aiohttp.ClientSession(timeout=timeout)
return self
async def __aexit__(self, exc_type, exc_val, exc_tb):
if self.session:
await self.session.close()
def _generate_signature(self, timestamp: str, method: str, path: str, body: str = "") -> str:
"""HMAC-SHA256 Signatur für authentifizierte Anfragen."""
message = timestamp + method + path + body
mac = hashlib.sha256()
mac.update(message.encode('utf-8'))
mac.update(self.api_secret.encode('utf-8'))
return mac.hexdigest()
async def _request_with_retry(
self,
method: str,
endpoint: str,
params: Optional[Dict] = None,
authenticated: bool = False
) -> Dict:
"""Anfrage mit exponentiellem Backoff und Retry-Logik."""
timestamp = datetime.utcnow().isoformat() + 'Z'
for attempt in range(self.MAX_RETRIES):
async with self._request_semaphore:
await asyncio.sleep(self.RATE_LIMIT_DELAY * attempt)
headers = {
'Content-Type': 'application/json',
'OKX-Timestamp': timestamp,
}
if authenticated:
signature = self._generate_signature(timestamp, method, endpoint)
headers.update({
'OKX-API-Key': self.api_key,
'OKX-Signature': signature,
'OKX-Passphrase': self.passphrase,
})
try:
url = f"{self.BASE_URL}{endpoint}"
async with self.session.request(
method, url, params=params, headers=headers
) as response:
data = await response.json()
if response.status == 200:
return data
elif response.status == 429:
wait_time = 2 ** attempt
logger.warning(f"Rate limit erreicht, warte {wait_time}s")
await asyncio.sleep(wait_time)
else:
logger.error(f"API Fehler {response.status}: {data}")
return {"code": str(response.status), "data": []}
except aiohttp.ClientError as e:
logger.error(f"Verbindungsfehler: {e}")
if attempt == self.MAX_RETRIES - 1:
raise
return {"code": "MAX_RETRIES", "data": []}
async def get_historical_candles(
self,
inst_id: str,
bar: str = "1H",
after: Optional[str] = None,
before: Optional[str] = None,
limit: int = 100
) -> pd.DataFrame:
"""
Historische OHLCV-Candlestick-Daten abrufen.
Parameter:
inst_id: Instrument-ID (z.B. 'BTC-USDT')
bar: Zeitrahmen ('1m', '5m', '1H', '1D')
limit: Anzahl der Datenpunkte (max. 100)
Rückgabe:
DataFrame mit OHLCV-Daten
"""
endpoint = "/api/v5/market/history-candles"
params = {
"instId": inst_id,
"bar": bar,
"limit": str(limit)
}
if after:
params["after"] = after
if before:
params["before"] = before
start_time = time.perf_counter()
result = await self._request_with_retry("GET", endpoint, params)
latency_ms = (time.perf_counter() - start_time) * 1000
logger.debug(f"Kandles-Abruf für {inst_id}: {latency_ms:.2f}ms")
if result.get("code") == "0" and result.get("data"):
df = pd.DataFrame(result["data"], columns=[
"timestamp", "open", "high", "low", "close", "volume", "volCcy"
])
df["timestamp"] = pd.to_datetime(df["timestamp"].astype(float), unit="ms")
df[["open", "high", "low", "close", "volume"]] = df[
["open", "high", "low", "close", "volume"]
].astype(float)
return df
return pd.DataFrame()
async def get_trade_history(
self,
inst_id: str,
limit: int = 100
) -> List[Dict]:
"""Historische Trade-Daten für Volumenprofil-Analyse."""
endpoint = "/api/v5/market/trades"
params = {"instId": inst_id, "limit": str(limit)}
result = await self._request_with_retry("GET", endpoint, params)
if result.get("code") == "0":
return result.get("data", [])
return []
Benchmark-Funktion
async def benchmark_data_collection():
"""Performance-Benchmark für Datenabruf."""
collector = OKXDataCollector()
async with collector:
results = []
# Test: 100 Kandles abrufen
for _ in range(10):
start = time.perf_counter()
df = await collector.get_historical_candles("BTC-USDT", bar="1H", limit=100)
latency = (time.perf_counter() - start) * 1000
results.append(latency)
avg_latency = sum(results) / len(results)
p50_latency = sorted(results)[len(results) // 2]
print(f"Durchschnittliche Latenz: {avg_latency:.2f}ms")
print(f"P50 Latenz: {p50_latency:.2f}ms")
print(f"P99 Latenz: {sorted(results)[9]:.2f}ms")
if __name__ == "__main__":
asyncio.run(benchmark_data_collection())
3. HolySheep AI-Integration für Signalgenerierung
Die Integration von HolySheep AI ermöglicht es, komplexe Marktanalysen und Signale mit einer Latenz von unter 50 Millisekunden zu generieren. Bei einem Preis von nur ¥1 pro Dollar (DeepSeek V3.2: $0.42/MToken) ist dies eine erhebliche Kostenersparnis gegenüber alternativen Anbietern wie OpenAI ($8/MToken für GPT-4.1).
import os
from openai import OpenAI
from typing import Dict, List, Optional
from pydantic import BaseModel
import json
class TradingSignal(BaseModel):
"""Strukturiertes Handelssignal mit Konfidenzwert."""
action: str # "BUY", "SELL", "HOLD"
confidence: float # 0.0 bis 1.0
entry_price: Optional[float] = None
stop_loss: Optional[float] = None
take_profit: Optional[float] = None
reasoning: str
timeframe: str
risk_level: str # "LOW", "MEDIUM", "HIGH"
class HolySheepSignaler:
"""
KI-gestützte Signalerstellung mit HolySheep AI.
Nutzt die kostengünstige und schnelle HolySheep-API
für quantitative Handelssignale.
Vorteile:
- <50ms Latenz
- ¥1=$1 Wechselkurs (85%+ Ersparnis vs. OpenAI)
- WeChat/Alipay Zahlung möglich
- Kostenlose Credits für den Start
"""
def __init__(self, api_key: str = None):
self.client = OpenAI(
api_key=api_key or os.getenv("HOLYSHEEP_API_KEY", "YOUR_HOLYSHEEP_API_KEY"),
base_url="https://api.holysheep.ai/v1" # Pflicht: HolySheep Endpunkt
)
self.model = "deepseek-v3.2" # $0.42/MToken - optimal für Quant-Strategien
def generate_signal(
self,
symbol: str,
ohlcv_data: Dict,
indicators: Dict,
market_context: str = ""
) -> TradingSignal:
"""
Generiert ein Handelssignal basierend auf technischen Indikatoren
und Marktbedingungen.
Parameter:
symbol: Trading-Paar (z.B. 'BTC-USDT')
ohlcv_data: Dictionary mit aktuellen OHLCV-Daten
indicators: Technische Indikatoren (RSI, MACD, Bollinger, etc.)
market_context: Zusätzlicher Marktkontext oder Nachrichten
Rückgabe:
TradingSignal mit Handlungsempfehlung
"""
system_prompt = """Du bist ein erfahrener quantitativer Trader mit 20 Jahren Erfahrung.
Analysiere die angegebenen Marktdaten und erstelle präzise Handelssignale.
Antworte ausschließlich im JSON-Format mit folgendem Schema:
{
"action": "BUY|SELL|HOLD",
"confidence": 0.0-1.0,
"entry_price": number,
"stop_loss": number,
"take_profit": number,
"reasoning": "string",
"timeframe": "SHORT|MEDIUM|LONG",
"risk_level": "LOW|MEDIUM|HIGH"
}
"""
user_prompt = f"""Analysiere folgendes Trading-Paar: {symbol}
Aktuelle Marktdaten:
- Eröffnung: {ohlcv_data.get('open', 0)}
- Hoch: {ohlcv_data.get('high', 0)}
- Tief: {ohlcv_data.get('low', 0)}
- Schluss: {ohlcv_data.get('close', 0)}
- Volumen: {ohlcv_data.get('volume', 0)}
Technische Indikatoren:
{json.dumps(indicators, indent=2)}
Marktkontext: {market_context or 'Keine besonderen Ereignisse'}
Gebe ein klar strukturiertes Signal aus."""
response = self.client.chat.completions.create(
model=self.model,
messages=[
{"role": "system", "content": system_prompt},
{"role": "user", "content": user_prompt}
],
temperature=0.3, # Niedrige Temperatur für konsistente Signale
max_tokens=500
)
content = response.choices[0].message.content
# Parse JSON-Antwort
try:
signal_data = json.loads(content)
return TradingSignal(**signal_data)
except json.JSONDecodeError:
return TradingSignal(
action="HOLD",
confidence=0.0,
reasoning="Parsing-Fehler - halte Position",
timeframe="SHORT",
risk_level="MEDIUM"
)
def batch_analyze(self, signals: List[Dict]) -> List[TradingSignal]:
"""
Analysiert mehrere Signale gleichzeitig.
Kosteneffizient bei Verwendung von HolySheep (Batch-Preisnachlass).
"""
results = []
for signal_data in signals:
result = self.generate_signal(
symbol=signal_data["symbol"],
ohlcv_data=signal_data["ohlcv"],
indicators=signal_data["indicators"],
market_context=signal_data.get("context", "")
)
results.append(result)
return results
Kosten-Benchmark
def cost_benchmark():
"""Vergleich der API-Kosten zwischen Anbietern."""
# Annahmen: 1000 Token pro Anfrage, 1000 Requests/Tag
tokens_per_request = 1000
requests_per_day = 1000
providers = {
"HolySheep DeepSeek V3.2": 0.42, # $/MToken
"OpenAI GPT-4.1": 8.00,
"Anthropic Claude Sonnet 4.5": 15.00,
"Google Gemini 2.5 Flash": 2.50
}
print("=" * 60)
print("API-KOSTENVERGLEICH (1000 Requests/Tag)")
print("=" * 60)
for provider, price_per_mtoken in providers.items():
daily_cost = (tokens_per_request / 1_000_000) * price_per_mtoken * requests_per_day
monthly_cost = daily_cost * 30
yearly_cost = daily_cost * 365
print(f"\n{provider}:")
print(f" Täglich: ${daily_cost:.2f}")
print(f" Monatlich: ${monthly_cost:.2f}")
print(f" Jährlich: ${yearly_cost:.2f}")
savings = ((8.00 - 0.42) / 8.00) * 100
print(f"\n💰 HolySheep Ersparnis vs. OpenAI: {savings:.1f}%")
# Monatliche Ersparnis mit HolySheep
holy_monthly = (tokens_per_request / 1_000_000) * 0.42 * requests_per_day * 30
openai_monthly = (tokens_per_request / 1_000_000) * 8.00 * requests_per_day * 30
print(f"💰 Monatliche Ersparnis: ${openai_monthly - holy_monthly:.2f}")
if __name__ == "__main__":
cost_benchmark()
4. Backtesting-Engine mit HolySheep-Signal-Integration
Die Backtesting-Engine simuliert den Historischen Handel mit realistischer Ausführungsmodellierung, Slippage und Gebühren. Die Integration mit HolySheep AI ermöglicht die automatische Signalgenerierung während des Backtests.
import pandas as pd
import numpy as np
from typing import Dict, List, Optional, Tuple
from dataclasses import dataclass, field
from datetime import datetime
from enum import Enum
import json
class OrderSide(Enum):
BUY = "BUY"
SELL = "SELL"
class OrderType(Enum):
MARKET = "MARKET"
LIMIT = "LIMIT"
@dataclass
class Order:
"""Order-Repräsentation mit vollständiger Transaktionshistorie."""
timestamp: datetime
symbol: str
side: OrderSide
order_type: OrderType
quantity: float
price: float
executed_price: float = 0.0
commission: float = 0.0
slippage: float = 0.0
order_id: str = ""
@dataclass
class Position:
"""Aktive Position mit Performance-Tracking."""
symbol: str
quantity: float
entry_price: float
current_price: float = 0.0
unrealized_pnl: float = 0.0
realized_pnl: float = 0.0
@dataclass
class BacktestResult:
"""Zusammenfassung der Backtesting-Ergebnisse."""
total_return: float
sharpe_ratio: float
sortino_ratio: float
max_drawdown: float
max_drawdown_duration: int # Tage
win_rate: float
profit_factor: float
total_trades: int
avg_trade_duration: float
avg_commission_per_trade: float
holy_sheep_cost: float
class BacktestingEngine:
"""
Produktionsreife Backtesting-Engine mit:
- Historischer Datensimulation
- Order-Ausführung mit Slippage-Modell
- Provisionsberechnung (Maker/Taker)
- Risikoanalyse (Drawdown, Sharpe, Sortino)
- HolySheep AI-Signalintegration
"""
# Handelsgebühren (OKX Standard)
MAKER_FEE = 0.0008 # 0.08%
TAKER_FEE = 0.0010 # 0.10%
# Slippage-Modell Parameter
SLIPPAGE_BASE = 0.0001 # 0.01% Basis-Slippage
SLIPPAGE_VOLUME_FACTOR = 0.000001 # Volumenabhängige Slippage
def __init__(
self,
initial_capital: float = 10000.0,
holysheep_api_key: str = None
):
self.initial_capital = initial_capital
self.current_capital = initial_capital
self.positions: Dict[str, Position] = {}
self.orders: List[Order] = []
self.equity_curve: List[float] = []
self.holy_sheep_signaler = HolySheepSignaler(holysheep_api_key)
self.total_holysheep_cost = 0.0
# Statistik
self.trade_returns: List[float] = []
self.daily_returns: List[float] = []
self.peak_equity = initial_capital
def _calculate_slippage(
self,
order: Order,
market_volume: float
) -> float:
"""Berechnet Slippage basierend auf Ordergröße und Marktvolumen."""
volume_ratio = order.quantity / max(market_volume, 1)
slippage = (
self.SLIPPAGE_BASE +
volume_ratio * self.SLIPPAGE_VOLUME_FACTOR
)
return abs(order.price * slippage)
def _execute_order(
self,
order: Order,
market_price: float,
market_volume: float
) -> Order:
"""Simuliert Order-Ausführung mit Slippage und Gebühren."""
slippage = self._calculate_slippage(order, market_volume)
if order.side == OrderSide.BUY:
executed = market_price * (1 + slippage)
else:
executed = market_price * (1 - slippage)
order.executed_price = executed
order.slippage = abs(executed - market_price)
# Gebührenberechnung
is_taker = order.order_type == OrderType.MARKET
fee_rate = self.TAKER_FEE if is_taker else self.MAKER_FEE
order.commission = executed * order.quantity * fee_rate
return order
def _calculate_position_value(self, pos: Position) -> float:
"""Berechnet aktuellen Positionswert."""
return pos.quantity * pos.current_price
def _update_positions(self, prices: Dict[str, float]):
"""Aktualisiert alle Positionen mit aktuellen Preisen."""
for symbol, pos in self.positions.items():
if symbol in prices:
pos.current_price = prices[symbol]
pos.unrealized_pnl = (
(pos.current_price - pos.entry_price) *
pos.quantity *
(1 if pos.quantity > 0 else -1)
)
def open_position(
self,
symbol: str,
side: OrderSide,
quantity: float,
price: float,
market_volume: float,
timestamp: datetime
):
"""Eröffnet eine neue Position."""
order = Order(
timestamp=timestamp,
symbol=symbol,
side=side,
order_type=OrderType.MARKET,
quantity=quantity,
price=price
)
executed_order = self._execute_order(order, price, market_volume)
# Positionsaktualisierung
if symbol not in self.positions:
self.positions[symbol] = Position(
symbol=symbol,
quantity=quantity if side == OrderSide.BUY else -quantity,
entry_price=executed_order.executed_price,
current_price=price
)
else:
pos = self.positions[symbol]
pos.quantity += quantity if side == OrderSide.BUY else -quantity
pos.entry_price = (
(pos.entry_price * (abs(pos.quantity) - quantity) +
executed_order.executed_price * quantity) /
abs(pos.quantity)
)
self.current_capital -= executed_order.executed_price * quantity
self.current_capital -= executed_order.commission
self.orders.append(executed_order)
def close_position(
self,
symbol: str,
quantity: Optional[float] = None,
price: float = None,
market_volume: float = 1000000,
timestamp: datetime = None
):
"""Schließt eine Position vollständig oder teilweise."""
if symbol not in self.positions:
return
pos = self.positions[symbol]
close_qty = quantity if quantity else abs(pos.quantity)
if price is None:
price = pos.current_price
side = OrderSide.SELL if pos.quantity > 0 else OrderSide.BUY
order = Order(
timestamp=timestamp or datetime.now(),
symbol=symbol,
side=side,
order_type=OrderType.MARKET,
quantity=close_qty,
price=price
)
executed_order = self._execute_order(order, price, market_volume)
# PnL-Berechnung
pnl = (executed_order.executed_price - pos.entry_price) * close_qty
if side == OrderSide.SELL:
pnl = -pnl
pos.realized_pnl += pnl
self.current_capital += executed_order.executed_price * close_qty
self.current_capital -= executed_order.commission
self.trade_returns.append(pnl / (pos.entry_price * close_qty))
# Positionsaktualisierung
pos.quantity -= close_qty if side == OrderSide.SELL else -close_qty
if abs(pos.quantity) < 0.00000001:
del self.positions[symbol]
self.orders.append(executed_order)
def run_backtest(
self,
data: pd.DataFrame,
symbols: List[str],
indicators: Dict[str, Dict] = None
) -> BacktestResult:
"""
Führt den vollständigen Backtest durch.
Parameter:
data: DataFrame mit OHLCV-Daten (Muss 'timestamp', 'symbol' enthalten)
symbols: Liste der zu handelnden Symbole
indicators: Vorberechnete technische Indikatoren
"""
holy_sheep_requests = 0
for idx, row in data.iterrows():
timestamp = row.get('timestamp', pd.Timestamp(idx))
prices = {row['symbol']: row['close']}
volume = row.get('volume', 1000000)
self._update_positions(prices)
# HolySheep Signal-Abruf (begrenzt auf Stunden-Basis für Kostenoptimierung)
if holy_sheep_requests % 60 == 0: # Alle 60 Bars
for symbol in symbols:
ohlcv = {
'open': row.get('open', 0),
'high': row.get('high', 0),
'low': row.get('low', 0),
'close': row.get('close', 0),
'volume': row.get('volume', 0)
}
symbol_indicators = indicators.get(symbol, {}) if indicators else {}
signal = self.holy_sheep_signaler.generate_signal(
symbol=symbol,
ohlcv_data=ohlcv,
indicators=symbol_indicators
)
holy_sheep_requests += 1
# Signal-Ausführung
if signal.action == "BUY" and symbol not in self.positions:
position_size = self.current_capital * 0.1 # 10% Kapitallocation
qty = position_size / row['close']
self.open_position(
symbol=symbol,
side=OrderSide.BUY,
quantity=qty,
price=row['close'],
market_volume=volume,
timestamp=timestamp
)
elif signal.action == "SELL" and symbol in self.positions:
self.close_position(
symbol=symbol,
price=row['close'],
market_volume=volume,
timestamp=timestamp
)
# Equity-Berechnung
total_equity = self.current_capital
for pos in self.positions.values():
total_equity += self._calculate_position_value(pos)
self.equity_curve.append(total_equity)
self.peak_equity = max(self.peak_equity, total_equity)
# HolySheep Kostenberechnung
self.total_holysheep_cost = holy_sheep_requests * (
1000 / 1_000_000 * 0.42 # DeepSeek V3.2 Preis
)
return self._calculate_results()
def _calculate_results(self) -> BacktestResult:
"""Berechnet finale Performance-Metriken."""
equity = np.array(self.equity_curve)
returns = np.diff(equity) / equity[:-1]
returns = returns[~np.isnan(returns)]
# Sharpe Ratio (annualisiert, Annahme: 252 Handelstage)
if len(returns) > 0 and np.std(returns) > 0:
sharpe = np.mean(returns) / np.std(returns) * np.sqrt(252)
else:
sharpe = 0.0
# Sortino Ratio (nur negative Returns)
negative_returns = returns[returns < 0]
if len(negative_returns) > 0 and np.std(negative_returns) > 0:
sortino = np.mean(returns) / np.std(negative_returns) * np.sqrt(252)
else:
sortino = 0.0
# Max Drawdown
peak = equity[0]
max_dd = 0.0
max_dd_duration = 0
for equity_value in equity:
if equity_value > peak:
peak = equity_value
drawdown = (peak - equity_value) / peak
if drawdown > max_dd:
max_dd = drawdown
# Win Rate
winning_trades = [r for r in self.trade_returns if r > 0]
win_rate = len(winning_trades) / max(len(self.trade_returns), 1)
# Profit Factor
gross_profit = sum(r for r in self.trade_returns if r > 0)
gross_loss = abs(sum(r for r in self.trade_returns if r < 0))
profit_factor = gross_profit / max(gross_loss, 0.001)
return BacktestResult(
total_return=(equity[-1] - self.initial_capital) / self.initial_capital,
sharpe_ratio=sharpe,
sortino_ratio=sortino,
max_drawdown=max_dd,
max_drawdown_duration=max_dd_duration,
win_rate=win_rate,
profit_factor=profit_factor,
total_trades=len(self.trade_returns),
avg_trade_duration=0.0,
avg_commission_per_trade=(
sum(o.commission for o in self.orders) / max(len(self.orders), 1)
),
holy_sheep_cost=self.total_holysheep_cost
)
Vollständiger Benchmark
async def run_full_benchmark():
"""Vollständiger System-Benchmark mit Dummy-Daten."""
import asyncio
# Generiere Testdaten (1 Jahr stündliche Daten)
dates = pd.date_range(start='2025-01-01', end='2025-12-31', freq='1H')
test_data = pd.DataFrame({
'timestamp': dates,
'symbol': 'BTC-USDT',
'open': 42000 + np.random.randn(len(dates)) * 500,
'high': 42500 + np.random.randn(len(dates)) * 500,
'low': 41500 + np.random.randn(len(dates)) * 500,
'close': 42000 + np.random.randn(len(dates)) * 500,
'volume': 1000000 + np.random.randn(len(dates)) * 100000
})
# Backtest ausführen
engine = BacktestingEngine(initial_capital=10000)
import time
start = time.perf_counter()
results = engine.run_backtest(test_data, ['BTC-USDT'])
duration = time.perf_counter() - start
print("=" * 60)
print("BACKTEST ERGEBNISSE")
print("=" * 60)
print(f"Rendite: {results.total_return * 100:.2f}%")
print(f"Sharpe Ratio: {results.sharpe_ratio:.2f}")
print(f"Max Drawdown: {results.max_drawdown * 100:.2f}%")
print(f"Win Rate: {results.win_rate * 100:.2