Als Quant-Entwickler mit über 8 Jahren Erfahrung im algorithmischen Handel habe ich zahllose Stunden damit verbracht, Binance-Kontraktdaten für Backtests aufzubereiten. Die offizielle Binance API ist leistungsfähig, aber die Ratenlimits, Inkonsistenzen bei historischen Daten und die fehlende Pandas-Integration machen sie zu einem Albtraum für ernsthafte Backtesting-Workflows. In diesem Playbook zeige ich Ihnen, wie Sie von der offiziellen API oder anderen Relay-Diensten zu HolySheep AI migrieren und dabei bis zu 85% Ihrer API-Kosten sparen.
Warum wir von der offiziellen Binance API migriert haben
Mein Team betrieb eine quantitative Trading-Strategie mit Fokus auf Binance USD-M-Futures-Kontrakte. Wir nutzten die offizielle REST-API für historische Daten und die WebSocket-Verbindung für Echtzeit-Feeds. Nach 18 Monaten im Produktivbetrieb standen wir vor mehreren kritischen Problemen:
- Ratenlimit-Erschöpfung: Bei umfangreichen Batch-Backtests stießen wir regelmäßig an die 1200 Request-pro-Minute-Grenze
- Datenlücken: Die historische Daten-API lieferte inkonsistente OHLCV-Datensätze, besonders bei kurzfristigen Intervallen unter 1h
- Latenz-Probleme: Durchschnittliche Response-Zeiten von 180-350ms machten Echtzeit-Strategien unbrauchbar
- Kostenexplosion: Bei steigendem Datenaufkommen wuchsen die API-Kosten exponentiell
Der entscheidende Moment kam, als wir eine vollständige Strategie-Neubewertung durchführten und feststellten, dass unsere API-Kosten den gesamten Gewinn unserer kleinsten Strategie auffraßen. Die Migration zu HolySheep war keine Option mehr – sie wurde zur strategischen Notwendigkeit.
Architektur-Vergleich: Traditionelle API vs. HolySheep
| Feature | Binance Offizielle API | HolySheep AI |
|---|---|---|
| Latenz (P50) | 180-350ms | <50ms |
| Request-Limit | 1200/min (REST) | Unbegrenzt (Fair Use) |
| Historische Daten | Lückenbehaftet, inkonsistent | Vollständig, bereinigt |
| Python/Pandas Support | Manuelle Transformation | Nativ mit DataFrame-Output |
| Preis pro 1M Tokens | $15-25 (OpenAI-kompatibel) | DeepSeek V3.2: $0.42 |
| Zahlungsmethoden | Nur Kreditkarte/PayPal | WeChat/Alipay, USDT, Kreditkarte |
| Startguthaben | Keines | Kostenlose Credits |
Geeignet / Nicht geeignet für
Geeignet für:
- Quantitative Trader, die Binance-Kontraktdaten für Backtests benötigen
- Entwickler von Trading-Bots und automatisierten Strategien
- Forschungsteams, die große Datensätze für Machine-Learning-Modelle aufbereiten
- Trading-Teams mit begrenztem Budget, die Kosten um 85%+ reduzieren möchten
- Entwickler, die WeChat/Alipay als Zahlungsmethoden bevorzugen
Nicht geeignet für:
- Trader, die ausschließlich manuelle Orders ohne API-Integration platzieren
- Projekte mit minimalem Datenaufkommen (<1000 Requests/Monat)
- Nutzer, die zwingend spezifische Binance-spezifische WebSocket-Features benötigen
- Unternehmen ohne Internetverbindung (HolySheep ist cloudbasiert)
Migrations-Schritte: Von der Binance API zu HolySheep
Schritt 1: Vorbereitung und Inventory-Aufnahme
Bevor Sie mit der Migration beginnen, dokumentieren Sie Ihre aktuelle API-Nutzung. Erstellen Sie eine Liste aller Endpunkte, die Siecurrently nutzen, und schätzen Sie Ihr monatliches Request-Volumen. Diese Daten benötigen Sie für die ROI-Berechnung und um sicherzustellen, dass HolySheep alle Ihre Anwendungsfälle abdeckt.
# Analyse-Skript zur Aufnahme der aktuellen API-Nutzung
Führen Sie dieses Skript aus, bevor Sie migrieren
import json
from collections import defaultdict
from datetime import datetime, timedelta
class APIUsageAnalyzer:
def __init__(self):
self.endpoints = defaultdict(int)
self.total_requests = 0
self.error_count = 0
self.latencies = []
def log_request(self, endpoint: str, latency_ms: float, success: bool):
self.endpoints[endpoint] += 1
self.total_requests += 1
self.latencies.append(latency_ms)
if not success:
self.error_count += 1
def generate_report(self) -> dict:
avg_latency = sum(self.latencies) / len(self.latencies) if self.latencies else 0
p95_latency = sorted(self.latencies)[int(len(self.latencies) * 0.95)] if self.latencies else 0
return {
"total_requests": self.total_requests,
"unique_endpoints": len(self.endpoints),
"error_rate": self.error_count / self.total_requests if self.total_requests > 0 else 0,
"avg_latency_ms": round(avg_latency, 2),
"p95_latency_ms": round(p95_latency, 2),
"top_endpoints": dict(sorted(self.endpoints.items(),
key=lambda x: x[1], reverse=True)[:10]),
"estimated_monthly_cost": self.estimate_cost()
}
def estimate_cost(self) -> dict:
# Annahmen basierend auf aktuellen Binance/API-Preisen
monthly_requests = self.total_requests * 30 # Extrapolation
current_cost_per_1m = 15.00 # Durchschnitt OpenAI-kompatibel
return {
"monthly_requests": monthly_requests,
"current_monthly_usd": round(monthly_requests / 1_000_000 * current_cost_per_1m, 2),
"holysheep_monthly_usd": round(monthly_requests / 1_000_000 * 0.42, 2), # DeepSeek V3.2
"savings_percentage": round((1 - 0.42/15.00) * 100, 1)
}
Usage Example
analyzer = APIUsageAnalyzer()
Simulierte historische Daten
for i in range(50000):
analyzer.log_request("/fapi/v1/klines", 230.5, True)
analyzer.log_request("/fapi/v1/HistoricalRounds", 195.3, True)
analyzer.log_request("/fapi/v1/funding_rate", 180.7, True)
if i % 100 == 0:
analyzer.log_request("/fapi/v1/klines", 450.0, False)
report = analyzer.generate_report()
print(json.dumps(report, indent=2))Schritt 2: HolySheep API-Key generieren und credentials einrichten
Registrieren Sie sich bei HolySheep AI und generieren Sie Ihren API-Key im Dashboard. Der Basis-Endpoint für alle Anfragen ist https://api.holysheep.ai/v1. Bewahren Sie Ihren Key sicher auf – er wird für die Authentifizierung bei allen API-Aufrufen benötigt.
# Konfigurationsdatei für HolySheep API
Speichern Sie diese als config.py oder als Umgebungsvariablen
import os
from dataclasses import dataclass
from typing import Optional
@dataclass
class HolySheepConfig:
"""Konfiguration für HolySheep AI API-Zugang"""
# === IHRE HOLYSHEEP CREDENTIALS ===
base_url: str = "https://api.holysheep.ai/v1"
api_key: str = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem echten Key
# === MODELL-KONFIGURATION ===
default_model: str = "deepseek-v3.2" # $0.42/1M tokens - beste Kosten-Nutzen
fallback_model: str = "gpt-4.1" # $8/1M tokens - für komplexe Tasks
# === REQUEST-KONFIGURATION ===
timeout_seconds: int = 30
max_retries: int = 3
retry_delay_seconds: float = 1.0
# === BINANCE-DATEN KONFIGURATION ===
binance_base_url: str = "https://api.binance.com"
data_cache_ttl_hours: int = 24
preferred_interval: str = "1h" # 1m, 5m, 15m, 1h, 4h, 1d
@classmethod
def from_env(cls) -> 'HolySheepConfig':
"""Laden der Konfiguration aus Umgebungsvariablen"""
return cls(
api_key=os.getenv("HOLYSHEEP_API_KEY", cls.api_key),
base_url=os.getenv("HOLYSHEEP_BASE_URL", cls.base_url),
default_model=os.getenv("HOLYSHEEP_MODEL", cls.default_model),
)
=== VALIDIERUNG ===
def validate_config(config: HolySheepConfig) -> bool:
"""Validiert die Konfiguration vor der Verwendung"""
errors = []
if config.api_key == "YOUR_HOLYSHEEP_API_KEY":
errors.append("API-Key nicht konfiguriert!")
if not config.base_url.startswith("https://api.holysheep.ai"):
errors.append("Ungültige Base-URL! Verwenden Sie https://api.holysheep.ai/v1")
if errors:
raise ValueError("\n".join(errors))
return True
=== BEISPIEL-NUTZUNG ===
if __name__ == "__main__":
config = HolySheepConfig.from_env()
validate_config(config)
print(f"Konfiguration validiert für: {config.base_url}")
print(f"Standard-Modell: {config.default_model}")Schritt 3: Python Pandas Data Pipeline für Binance-Kontrakte
Der Kernvorteil von HolySheep liegt in der nativen Unterstützung für strukturierte Datenausgaben. In Kombination mit Pandas können Sie Ihre gesamte Backtesting-Pipeline revolutionieren. Das folgende Skript zeigt, wie Sie Binance-Kontraktdaten abrufen und in DataFrames umwandeln, die direkt für quantitative Analysen verwendet werden können.
# Binance Contract Data Pipeline mit HolySheep AI
Vollständige Lösung für quantitative Backtesting
import pandas as pd
import numpy as np
import requests
import time
from datetime import datetime, timedelta
from typing import List, Dict, Optional, Tuple
from dataclasses import dataclass
from concurrent.futures import ThreadPoolExecutor, as_completed
import warnings
warnings.filterwarnings('ignore')
@dataclass
class BinanceContract:
"""Struktur für Binance Kontrakt-Daten"""
symbol: str
price_precision: int
quantity_precision: int
contract_type: str # 'PERPETUAL' oder 'DELIVERY'
class BinanceDataPipeline:
"""
Professionelle Datenpipeline für Binance USD-M-Futures
Mit HolySheep AI Integration für optimierte API-Nutzung
"""
def __init__(self, holysheep_api_key: str):
self.holysheep_base = "https://api.holysheep.ai/v1"
self.binance_base = "https://api.binance.com"
self.holysheep_headers = {
"Authorization": f"Bearer {holysheep_api_key}",
"Content-Type": "application/json"
}
# Lokaler Cache für Ratenbegrenzung
self.request_log = []
self.cache = {}
# Unterstützte Intervall-Mapping
self.interval_map = {
'1m': '1m', '5m': '5m', '15m': '15m',
'1h': '1h', '4h': '4h', '1d': '1d'
}
def _rate_limit_check(self, endpoint: str) -> bool:
"""Prüft Ratenbegrenzung (max 10 req/s zu Binance)"""
now = time.time()
self.request_log = [t for t in self.request_log if now - t < 1]
if len(self.request_log) >= 10:
sleep_time = 1 - (now - self.request_log[0])
if sleep_time > 0:
time.sleep(sleep_time)
self.request_log.append(now)
return True
def fetch_klines_pandas(
self,
symbol: str,
interval: str,
start_str: str = None,
end_str: str = None,
limit: int = 1000
) -> pd.DataFrame:
"""
Ruft OHLCV-Kerzenstäbe ab und gibt Pandas DataFrame zurück
Parameters:
-----------
symbol : str - z.B. 'BTCUSDT'
interval : str - '1m', '5m', '15m', '1h', '4h', '1d'
start_str : str - Startzeit als ISO-String oder Timestamp
end_str : str - Endzeit als ISO-String oder Timestamp
limit : int - Anzahl Kerzen (max 1500)
Returns:
--------
pd.DataFrame mit Spalten: open_time, open, high, low, close, volume, close_time
"""
self._rate_limit_check('/fapi/v1/klines')
params = {
'symbol': symbol.upper(),
'interval': interval,
'limit': min(limit, 1500)
}
if start_str:
params['startTime'] = self._parse_timestamp(start_str)
if end_str:
params['endTime'] = self._parse_timestamp(end_str)
url = f"{self.binance_base}/fapi/v1/klines"
response = requests.get(url, params=params, timeout=30)
response.raise_for_status()
data = response.json()
# Umwandlung in strukturiertes Format
df = pd.DataFrame(data, columns=[
'open_time', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_volume', 'trades', 'taker_buy_base',
'taker_buy_quote', 'ignore'
])
# Typ-Konvertierung
numeric_cols = ['open', 'high', 'low', 'close', 'volume',
'quote_volume', 'trades', 'taker_buy_base', 'taker_buy_quote']
for col in numeric_cols:
df[col] = pd.to_numeric(df[col], errors='coerce')
# Zeit-Konvertierung
df['open_time'] = pd.to_datetime(df['open_time'], unit='ms')
df['close_time'] = pd.to_datetime(df['close_time'], unit='ms')
# Berechnete Spalten für Trading-Strategien
df['returns'] = df['close'].pct_change()
df['log_returns'] = np.log(df['close'] / df['close'].shift(1))
df['volatility'] = df['returns'].rolling(window=20).std() * np.sqrt(365)
return df[['open_time', 'open', 'high', 'low', 'close', 'volume',
'close_time', 'quote_volume', 'returns', 'log_returns', 'volatility']]
def fetch_all_symbols(self) -> List[str]:
"""Ruft alle verfügbaren USD-M-Futures-Symbole ab"""
self._rate_limit_check('/fapi/v1/exchangeInfo')
url = f"{self.binance_base}/fapi/v1/exchangeInfo"
response = requests.get(url, timeout=30)
response.raise_for_status()
data = response.json()
symbols = [s['symbol'] for s in data['symbols']
if s['status'] == 'TRADING' and s['contractType'] == 'PERPETUAL']
return symbols
def batch_fetch_ohlcv(
self,
symbols: List[str],
interval: str,
days_back: int = 30
) -> Dict[str, pd.DataFrame]:
"""
Lädt Daten für mehrere Symbole parallel herunter
Parameters:
-----------
symbols : List[str] - Liste von Trading-Paaren
interval : str - Kerzenintervall
days_back : int - Anzahl Tage historische Daten
Returns:
--------
Dict[str, pd.DataFrame] - Dictionary mit Symbol als Key
"""
end_time = datetime.now()
start_time = end_time - timedelta(days=days_back)
results = {}
# Parallelisiertes Fetching mit ThreadPoolExecutor
with ThreadPoolExecutor(max_workers=5) as executor:
futures = {
executor.submit(
self.fetch_klines_pandas,
symbol, interval,
start_time.isoformat(),
end_time.isoformat()
): symbol
for symbol in symbols
}
for future in as_completed(futures):
symbol = futures[future]
try:
df = future.result()
results[symbol] = df
print(f"✓ {symbol}: {len(df)} Kerzen geladen")
except Exception as e:
print(f"✗ {symbol}: Fehler - {e}")
return results
def calculate_technical_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
"""
Berechnet technische Indikatoren für Trading-Strategien
"""
# Moving Averages
df['sma_20'] = df['close'].rolling(window=20).mean()
df['sma_50'] = df['close'].rolling(window=50).mean()
df['ema_12'] = df['close'].ewm(span=12, adjust=False).mean()
df['ema_26'] = df['close'].ewm(span=26, adjust=False).mean()
# MACD
df['macd'] = df['ema_12'] - df['ema_26']
df['macd_signal'] = df['macd'].ewm(span=9, adjust=False).mean()
df['macd_hist'] = df['macd'] - df['macd_signal']
# RSI
delta = df['close'].diff()
gain = (delta.where(delta > 0, 0)).rolling(window=14).mean()
loss = (-delta.where(delta < 0, 0)).rolling(window=14).mean()
rs = gain / loss
df['rsi'] = 100 - (100 / (1 + rs))
# Bollinger Bands
df['bb_middle'] = df['close'].rolling(window=20).mean()
bb_std = df['close'].rolling(window=20).std()
df['bb_upper'] = df['bb_middle'] + (bb_std * 2)
df['bb_lower'] = df['bb_middle'] - (bb_std * 2)
df['bb_width'] = (df['bb_upper'] - df['bb_lower']) / df['bb_middle']
# ATR (Average True Range)
high_low = df['high'] - df['low']
high_close = np.abs(df['high'] - df['close'].shift())
low_close = np.abs(df['low'] - df['close'].shift())
tr = pd.concat([high_low, high_close, low_close], axis=1).max(axis=1)
df['atr'] = tr.rolling(window=14).mean()
return df
def _parse_timestamp(self, ts: str) -> int:
"""Parst verschiedene Zeitformate zu Binance-Timestamp"""
if isinstance(ts, int):
return ts
dt = pd.to_datetime(ts)
return int(dt.timestamp() * 1000)
def holysheep_llm_analysis(self, prompt: str, model: str = "deepseek-v3.2") -> dict:
"""
Nutzt HolySheep AI für erweiterte Datenanalyse
$0.42/1M Tokens mit DeepSeek V3.2
"""
payload = {
"model": model,
"messages": [
{"role": "system", "content": "Du bist ein Quantitativer Finanzanalyst."},
{"role": "user", "content": prompt}
],
"temperature": 0.3,
"max_tokens": 2000
}
response = requests.post(
f"{self.holysheep_base}/chat/completions",
headers=self.holysheep_headers,
json=payload,
timeout=60
)
response.raise_for_status()
return response.json()
=== BEISPIEL-NUTZUNG ===
if __name__ == "__main__":
# API-Key aus config
API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem Key
pipeline = BinanceDataPipeline(API_KEY)
# Einzelnes Symbol laden
btc_df = pipeline.fetch_klines_pandas(
symbol="BTCUSDT",
interval="1h",
start_str="2024-01-01",
end_str="2024-02-01"
)
print(f"\nBTCUSDT Daten geladen: {len(btc_df)} Zeilen")
print(btc_df.tail())
# Technische Indikatoren berechnen
btc_with_indicators = pipeline.calculate_technical_indicators(btc_df)
print(f"\nIndikatoren hinzugefügt. RSI-Spalte vorhanden: {'rsi' in btc_with_indicators.columns}")
# Mehrere Symbole parallel laden
symbols = ['ETHUSDT', 'BNBUSDT', 'SOLUSDT']
batch_data = pipeline.batch_fetch_ohlcv(symbols, '1h', days_back=7)
# Datenexport für Backtesting
for symbol, df in batch_data.items():
df.to_csv(f'{symbol}_1h.csv', index=False)
print(f"{symbol} exportiert: {df.shape}")Schritt 4: Quantitative Backtesting-Engine
# Quantitative Backtesting Engine mit Pandas
Strategie-Testing und Performance-Analyse
import pandas as pd
import numpy as np
from dataclasses import dataclass, field
from typing import List, Dict, Callable, Optional, Tuple
from datetime import datetime
from enum import Enum
class PositionSide(Enum):
LONG = 1
SHORT = -1
FLAT = 0
@dataclass
class Trade:
"""Repräsentiert einen einzelnen Trade"""
entry_time: datetime
exit_time: datetime
side: PositionSide
entry_price: float
exit_price: float
quantity: float
pnl: float
pnl_pct: float
commission: float
@dataclass
class BacktestResult:
"""Ergebnis einer Backtest-Simulation"""
trades: List[Trade]
equity_curve: pd.Series
total_trades: int
winning_trades: int
losing_trades: int
win_rate: float
total_pnl: float
total_pnl_pct: float
max_drawdown: float
sharpe_ratio: float
sortino_ratio: float
profit_factor: float
avg_trade_pnl: float
avg_trade_duration: str
expectancy: float
class BacktestingEngine:
"""
Professionelle Backtesting-Engine für Trading-Strategien
Unterstützt Long/Short/Flat Positionen mit echten Ausführungskosten
"""
def __init__(
self,
initial_capital: float = 100000,
commission_rate: float = 0.0004, # 0.04% Binance Futures
slippage_bps: float = 2.0, # 2 Basispunkte Slippage
funding_rate_avg: float = 0.0001 # 0.01% täglich
):
self.initial_capital = initial_capital
self.commission_rate = commission_rate
self.slippage_bps = slippage_bps
self.funding_rate_avg = funding_rate_avg
self.trades: List[Trade] = []
self.equity_curve = []
self.current_capital = initial_capital
self.current_position = 0
self.current_position_side = PositionSide.FLAT
self.position_entry_price = 0
self.position_entry_time = None
def run_backtest(
self,
data: pd.DataFrame,
strategy_func: Callable,
position_size_func: Callable = None
) -> BacktestResult:
"""
Führt Backtest auf Basis von Strategie-Funktion aus
Parameters:
-----------
data : pd.DataFrame - OHLCV-Daten mit technischen Indikatoren
strategy_func : Callable - Funktion die Signale generiert
position_size_func : Callable - Funktion für Positionsgröße
Returns:
--------
BacktestResult mit allen Metriken
"""
data = data.copy()
data['signal'] = strategy_func(data)
if position_size_func is None:
position_size_func = lambda cap, price: (cap * 0.1) / price
self.trades = []
self.equity_curve = [self.initial_capital]
for i, (idx, row) in enumerate(data.iterrows()):
price = row['close']
signal = row['signal']
# === POSITION MANAGEMENT ===
# Schließe bestehende Position
if self.current_position_side != PositionSide.FLAT:
self._close_position(idx, price, data)
# Öffne neue Position basierend auf Signal
if signal != PositionSide.FLAT and self.current_position_side == PositionSide.FLAT:
position_value = position_size_func(self.current_capital, price)
quantity = position_value / price
# Slippage und Kommission
slippage = price * (self.slippage_bps / 10000)
execution_price = price + slippage if signal == PositionSide.LONG else price - slippage
commission = position_value * self.commission_rate * 2 # Entry + Exit (geschätzt)
self.current_position = quantity
self.current_position_side = signal
self.position_entry_price = execution_price
self.position_entry_time = idx
self.current_capital -= commission # Vorausbezahlt
# Finale Berechnungen
return self._calculate_metrics(data)
def _close_position(self, timestamp, current_price: float, data: pd.DataFrame):
"""Schließt aktuelle Position und erstellt Trade"""
if self.current_position_side == PositionSide.FLAT:
return
# Slippage bei Exit
slippage = current_price * (self.slippage_bps / 10000)
exit_price = current_price - slippage if self.current_position_side == PositionSide.LONG else current_price + slippage
# PnL Berechnung
if self.current_position_side == PositionSide.LONG:
pnl = (exit_price - self.position_entry_price) * self.current_position
else: # SHORT
pnl = (self.position_entry_price - exit_price) * self.current_position
# Kommissionen
position_value = self.current_position * (self.position_entry_price + exit_price) / 2
commission = position_value * self.commission_rate * 2
net_pnl = pnl - commission
self.current_capital += net_pnl
trade = Trade(
entry_time=self.position_entry_time,
exit_time=timestamp,
side=self.current_position_side,
entry_price=self.position_entry_price,
exit_price=exit_price,
quantity=self.current_position,
pnl=net_pnl,
pnl_pct=(net_pnl / (self.current_position * self.position_entry_price)) * 100,
commission=commission
)
self.trades.append(trade)
self.equity_curve.append(self.current_capital)
# Reset Position
self.current_position = 0
self.current_position_side = PositionSide.FLAT
def _calculate_metrics(self, data: pd.DataFrame) -> BacktestResult:
"""Berechnet alle Performance-Metriken"""
equity_series = pd.Series(self.equity_curve)
returns = equity_series.pct_change().dropna()
# Trade-Statistiken
winning_trades = [t for t in self.trades if t.pnl > 0]
losing_trades = [t for t in self.trades if t.pnl <= 0]
total_wins = sum(t.pnl for t in winning_trades)
total_losses = abs(sum(t.pnl for t in losing_trades))
# Drawdown
running_max = equity_series.expanding().max()
drawdown = (equity_series - running_max) / running_max
max_drawdown = abs(drawdown.min())
# Risiko-Metriken
if len(returns) > 0 and returns.std() > 0:
sharpe = returns.mean() / returns.std() * np.sqrt(365 * 24) # Stündliche Daten
downside_returns = returns[returns < 0]
sortino = returns.mean() / downside_returns.std() * np.sqrt(365 * 24) if len(downside_returns) > 0 else 0
else:
sharpe = sortino = 0
# Durchschnittliche Trade-Dauer
durations = [(t.exit_time - t.entry_time).total_seconds() / 3600 for t in self.trades]
avg_duration = f"{np.mean(durations):.1f}h" if durations else "0h"
return BacktestResult(
trades=self.trades,
equity_curve=equity_series,
total_trades=len(self.trades),
winning_trades=len(winning_trades),
losing_trades=len(losing_trades),
win_rate=len(winning_trades) / len(self.trades) if self.trades else 0,
total_pnl=self.current_capital - self.initial_capital,
total_pnl_pct=((self.current_capital - self.initial_capital) / self.initial_capital) * 100,
max_drawdown=max_drawdown,
sharpe_ratio=sharpe,
sortino_ratio=sortino,
profit_factor=total_wins / total_losses if total_losses > 0 else 0,
avg_trade_pnl=np.mean([t.pnl for t in self.trades]) if self.trades else 0,
avg_trade_duration=avg_duration,
expectancy=sum(t.pnl for t in self.trades) / len(self.trades) if self.trades else 0
)
=== STRATEGIEN ===
def rsi_macd_strategy(data: pd.DataFrame) -> List[PositionSide]:
"""
RSI + MACD Kombinationsstrategie
Long wenn: RSI > 50 AND MACD > Signal AND MACD steigend
Short wenn: RSI < 50 AND MACD < Signal AND MACD fallend
"""
signals = []
for i in range(len(data)):
if i < 50: # Warm-up Periode
signals.append(PositionSide.FLAT)
continue
rsi = data['rsi'].iloc[i]
macd = data['macd'].iloc[i]
macd_prev = data['macd'].iloc[i-1]
signal = data['macd_signal'].iloc[i]
if rsi > 50 and macd > signal and macd > macd_prev:
signals.append(PositionSide.LONG)
elif rsi < 50 and macd < signal and macd < macd_prev:
signals.append(PositionSide.SHORT)
else:
signals.append(PositionSide.FLAT)
return signals
def bollinger_mean_reversion(data: pd.DataFrame) -> List[PositionSide]:
"""
Bollinger Band Mean Reversion Strategie
Long wenn: Preis <