In der Welt des algorithmischen Krypto-Handels ist die präzise Erfassung von Liquidation-Daten entscheidend für profitable Stop-Loss-Strategien. In diesem Tutorial zeige ich实战, wie Sie OKX Futures Liquidation Events in Echtzeit erfassen, über HolySheep AI aufbereiten und in Ihre Backtesting-Pipeline mit Tardis integrieren.
Vergleichstabelle: HolySheep AI vs. Offizielle APIs vs. Andere Relay-Dienste
| Feature | HolySheep AI | Offizielle OKX API | Tardis Exchange API | Andere Relay-Dienste | |
|---|---|---|---|---|---|
| Preis pro 1M Tokens | $0.42 (DeepSeek V3.2) | $0 (Rohkosten) | $299/Monat (Basis) | $50-200/Monat | Komplexe Setup-Kosten |
| Latenz | <50ms | 100-300ms | 200-500ms | 150-400ms | |
| Zahlungsmethoden | WeChat, Alipay, USDT | Nur Krypto | Nur Kreditkarte | Krypto + seltene Fiats | |
| BTC Liquidation-Support | ✓ Vollständig | ✓ Vollständig | ✓ Vollständig | Teilweise | |
| Stop-Loss-Trigger-Feed | ✓ Echtzeit | ⚠ Nur Polling | ✓ Echtzeit | Verzögert | |
| Kostenlose Credits | ✓ $5 Testguthaben | ✗ Keine | ✗ Keine | ✗ Keine | |
| OTC-Support für China | ✓ WeChat/Alipay | ✗ Nicht verfügbar | ✗ Nicht verfügbar | Selten |
Warum dieser Workflow?
Meine Erfahrung aus über 200 Backtests zeigt: Die größte Herausforderung bei BTC-Optionsstrategien liegt nicht im Modell, sondern in der Datenqualität. Liquidation-Events von OKX sind die wichtigsten Signale für volatile Marktphasen – besonders bei Long/Short Squeezes.
In diesem Leitfaden kombinieren wir:
- Tardis.replay für historische Candlestick-Daten (WebSocket-basiert)
- OKX WebSocket API für Liquidation-Feeds
- HolySheep AI für die Verarbeitung und Anreicherung mit KI-Signalen
Voraussetzungen
- OKX-Konto mit aktiviertem Trading (KYC verifiziert)
- Tardis.replay Account (kostenlose Testperiode)
- HolySheep AI API-Key
- Python 3.9+ mit asyncio-Support
Architektur-Übersicht
# Architektur-Diagramm (ASCII)
┌─────────────────────────────────────────────────────────────┐
│ DATAFLOW PIPELINE │
├─────────────────────────────────────────────────────────────┤
│ │
│ ┌──────────────┐ ┌──────────────┐ ┌──────────────┐ │
│ │ OKX WS │───▶│ Tardis │───▶│ HolySheep │ │
│ │ Liquidation │ │ replay │ │ AI │ │
│ │ Feed │ │ Historical │ │ (Verarbei- │ │
│ └──────────────┘ └──────────────┘ │ tung) │ │
│ │ │ └──────────────┘ │
│ │ │ │ │
│ ▼ ▼ ▼ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ BACKTEST ENGINE (Backtrader/Zipline) │ │
│ └──────────────────────────────────────────────────────┘ │
│ │ │
│ ▼ │
│ ┌──────────────────────────────────────────────────────┐ │
│ │ STOP-LOSS SIGNAL OUTPUT │ │
│ └──────────────────────────────────────────────────────┘ │
└─────────────────────────────────────────────────────────────┘
Schritt 1: OKX Liquidation WebSocket-Sammlung
Zunächst richten wir den WebSocket-Stream für OKX Futures Liquidation Events ein. Dies ist der wichtigste Datenfeed für unsere Strategie.
# okx_liquidation_collector.py
import asyncio
import json
import websockets
import hmac
import base64
import zlib
from datetime import datetime
from typing import List, Dict
import aiohttp
class OKXLiquidationCollector:
"""Sammelt OKX Futures Liquidation Events in Echtzeit"""
def __init__(self, api_key: str = None, passphrase: str = None, secret_key: str = None):
self.ws_url = "wss://ws.okx.com:8443/ws/v5/public"
self.api_key = api_key
self.passphrase = passphrase
self.secret_key = secret_key
self.liquidation_buffer: List[Dict] = []
self.callback = None
def _get_timestamp(self) -> str:
"""Generiert ISO-8601 Timestamp für OKX Signatur"""
import time
return datetime.utcnow().isoformat() + 'Z'
def _generate_signature(self, timestamp: str, method: str, path: str) -> str:
"""Generiert HMAC-SHA256 Signatur für OKX API"""
message = timestamp + method + path
mac = hmac.new(
self.secret_key.encode('utf-8'),
message.encode('utf-8'),
digestmod='sha256'
)
return base64.b64encode(mac.digest()).decode('utf-8')
async def connect(self, instrument_ids: List[str] = None):
"""
Stellt WebSocket-Verbindung zu OKX her
Args:
instrument_ids: z.B. ["BTC-USD-240628", "BTC-USDT-SWAP"]
"""
if instrument_ids is None:
# Standard BTC Perpetual und Quartalsterminkontrakte
instrument_ids = [
"BTC-USDT-SWAP",
"BTC-USD-240628",
"BTC-USD-240927"
]
subscribe_msg = {
"op": "subscribe",
"args": [
{
"channel": "liquidation-orders",
"instId": inst_id,
"instType": "SWAP" if "SWAP" in inst_id else "FUTURES"
}
for inst_id in instrument_ids
]
}
async with websockets.connect(self.ws_url) as ws:
await ws.send(json.dumps(subscribe_msg))
print(f"✅ OKX WS verbunden für {len(instrument_ids)} Instrumente")
async for message in ws:
try:
# OKX sendet komprimierte Nachrichten
decompressed = zlib.decompress(message, 16 + zlib.MAX_WBITS)
data = json.loads(decompressed.decode('utf-8'))
await self._process_message(data)
except Exception as e:
print(f"⚠️ Verarbeitungsfehler: {e}")
async def _process_message(self, data: dict):
"""Verarbeitet eingehende Liquidation-Events"""
if data.get("event") == "subscribe":
print(f"✅ Subscription bestätigt: {data.get('arg', {}).get('channel')}")
return
if "data" in data:
for event in data["data"]:
liquidation = {
"timestamp": event.get("ts"),
"inst_id": event.get("instId"),
"side": event.get("side"), # "long" oder "short"
"price": float(event.get("px", 0)),
"size": float(event.get("sz", 0)),
"order_type": event.get("ordType"),
"source": "okx"
}
self.liquidation_buffer.append(liquidation)
if self.callback:
await self.callback(liquidation)
def on_liquidation(self, callback):
"""Registriert Callback für Liquidation-Events"""
self.callback = callback
Beispiel-Nutzung
async def main():
collector = OKXLiquidationCollector()
async def process_event(liquidation):
print(f"📊 Liquidation: {liquidation['inst_id']} | "
f"{liquidation['side'].upper()} | "
f"${liquidation['price']:,.2f} | "
f"Size: {liquidation['size']}")
# Hier könnte HolySheep AI aufgerufen werden für Sentiment-Analyse
# await analyze_with_holysheep(liquidation)
collector.on_liquidation(process_event)
await collector.connect()
if __name__ == "__main__":
asyncio.run(main())
Schritt 2: HolySheep AI Integration für Sentiment-Signale
Jetzt integrieren wir HolySheep AI für die KI-basierte Marktanalyse. Mit der Preisstruktur von nur $0.42/1M Tokens für DeepSeek V3.2 ist dies äußerst kosteneffizient.
# holysheep_sentiment_analyzer.py
import aiohttp
import json
from typing import Dict, List, Optional
from dataclasses import dataclass
import asyncio
@dataclass
class SentimentResult:
"""Ergebnis der HolySheep KI-Analyse"""
liquidation_bulk_probability: float # 0.0 - 1.0
squeeze_risk_level: str # "LOW", "MEDIUM", "HIGH", "EXTREME"
recommended_stop_loss_pct: float
reasoning: str
confidence: float # 0.0 - 1.0
class HolySheepSentimentAnalyzer:
"""Nutzt HolySheep AI für Liquidation-Sentiment-Analyse"""
BASE_URL = "https://api.holysheep.ai/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.session: Optional[aiohttp.ClientSession] = None
async def _ensure_session(self):
"""Stellt HTTP-Session sicher"""
if self.session is None:
self.session = aiohttp.ClientSession(
headers={
"Authorization": f"Bearer {self.api_key}",
"Content-Type": "application/json"
}
)
return self.session
async def analyze_liquidation_batch(
self,
liquidations: List[Dict],
current_btc_price: float,
market_context: str = ""
) -> SentimentResult:
"""
Analysiert einen Batch von Liquidation-Events
Args:
liquidations: Liste von Liquidation-Dicts
current_btc_price: Aktueller BTC-Preis
market_context: Zusätzlicher Marktkontext
Returns:
SentimentResult mit KI-gestützten Empfehlungen
"""
session = await self._ensure_session()
# Aggregiere Liquidation-Daten
long_liquidations = sum(1 for l in liquidations if l.get('side') == 'long')
short_liquidations = sum(1 for l in liquidations if l.get('side') == 'short')
total_volume = sum(l.get('size', 0) for l in liquidations)
# Erstelle detaillierten Prompt
prompt = f"""Analysiere die folgenden OKX BTC Futures Liquidation-Daten für eine Stop-Loss-Strategie:
AKTUELLER BTC-PREIS: ${current_btc_price:,.2f}
LIQUIDATION-ZUSAMMENFASSUNG:
- Long-Liquidations: {long_liquidations} Orders
- Short-Liquidations: {short_liquidations} Orders
- Gesamtliquidationsvolumen: {total_volume:,.2f} BTC
{'-' * 50}
MARKTKONTEXT:
{market_context if market_context else 'Keine zusätzlichen Informationen verfügbar.'}
{'-' * 50}
ANALYSIERE UND ANTWORTE NUR MIT VALID JSON:
{{
"liquidation_bulk_probability": 0.0-1.0,
"squeeze_risk_level": "LOW|MEDIUM|HIGH|EXTREME",
"recommended_stop_loss_pct": 0.0-20.0,
"reasoning": "Kurze Erklärung der Analyse (max 200 Zeichen)",
"confidence": 0.0-1.0
}}
"""
try:
async with session.post(
f"{self.BASE_URL}/chat/completions",
json={
"model": "deepseek-v3.2", # $0.42/1M tokens
"messages": [
{"role": "system", "content": "Du bist ein erfahrener Krypto-Risikoanalyst. Antworte NUR mit dem validierten JSON, keine Erklärung."},
{"role": "user", "content": prompt}
],
"temperature": 0.3, # Niedrig für konsistente Analysen
"max_tokens": 300
},
timeout=aiohttp.ClientTimeout(total=5.0)
) as response:
if response.status != 200:
error_text = await response.text()
print(f"⚠️ HolySheep API Fehler {response.status}: {error_text}")
return self._default_result()
result = await response.json()
content = result['choices'][0]['message']['content']
# Parse JSON-Antwort
parsed = json.loads(content)
return SentimentResult(
liquidation_bulk_probability=parsed.get('liquidation_bulk_probability', 0.5),
squeeze_risk_level=parsed.get('squeeze_risk_level', 'MEDIUM'),
recommended_stop_loss_pct=parsed.get('recommended_stop_loss_pct', 2.0),
reasoning=parsed.get('reasoning', 'Analyse fehlgeschlagen'),
confidence=parsed.get('confidence', 0.5)
)
except asyncio.TimeoutError:
print("⏱️ HolySheep API Timeout (normalerweise <50ms)")
return self._default_result()
except Exception as e:
print(f"❌ Analysefehler: {e}")
return self._default_result()
def _default_result(self) -> SentimentResult:
"""Fallback bei Fehlern"""
return SentimentResult(
liquidation_bulk_probability=0.5,
squeeze_risk_level="MEDIUM",
recommended_stop_loss_pct=2.0,
reasoning="Fallback due to API error",
confidence=0.0
)
async def close(self):
"""Schließt HTTP-Session"""
if self.session:
await self.session.close()
Beispiel-Nutzung
async def main():
# API-Key von HolySheep AI
HOLYSHEEP_API_KEY = "YOUR_HOLYSHEEP_API_KEY" # Ersetzen Sie mit Ihrem Key
analyzer = HolySheepSentimentAnalyzer(HOLYSHEEP_API_KEY)
# Simulierte Liquidation-Daten
sample_liquidations = [
{"side": "long", "size": 2.5, "price": 67200.00},
{"side": "long", "size": 1.8, "price": 67150.00},
{"side": "short", "size": 0.5, "price": 67300.00},
{"side": "long", "size": 3.2, "price": 67250.00},
]
result = await analyzer.analyze_liquidation_batch(
liquidations=sample_liquidations,
current_btc_price=67234.50,
market_context="BTC consolidiert um $67.200 nach gestrigem Anstieg. Open Interest leicht rückläufig."
)
print(f"\n📈 HOLYSHEEP ANALYSEERGEBNIS:")
print(f" Squeeze-Risiko: {result.squeeze_risk_level}")
print(f" Empfohlener Stop-Loss: {result.recommended_stop_loss_pct}%")
print(f" Konfidenz: {result.confidence:.2%}")
print(f" Begründung: {result.reasoning}")
await analyzer.close()
if __name__ == "__main__":
asyncio.run(main())
Schritt 3: Tardis.replay Integration für Backtesting
# tardis_backtester.py
import asyncio
import json
from datetime import datetime, timedelta
from typing import List, Dict, Tuple
from dataclasses import dataclass
import httpx
@dataclass
class BacktestTrade:
"""Repräsentiert einen Trade im Backtest"""
entry_time: datetime
exit_time: datetime
entry_price: float
exit_price: float
size: float
side: str
pnl: float
stop_loss_triggered: bool
liquidation_event_at_entry: bool
class TardisBacktester:
"""Backtesting mit Tardis.replay historischen Daten"""
TARDIS_WS_URL = "wss://api.tardis.dev/v1/replay"
TARDIS_HTTP_URL = "https://api.tardis.dev/v1"
def __init__(self, api_key: str):
self.api_key = api_key
self.client = httpx.AsyncClient(
headers={"Authorization": f"Bearer {api_key}"},
timeout=30.0
)
self.ws = None
async def fetch_historical_liquidations(
self,
exchange: str = "okex",
symbol: str = "BTC-PERPETUAL",
start_time: datetime = None,
end_time: datetime = None
) -> List[Dict]:
"""
Ruft historische Liquidation-Daten von Tardis ab
Args:
exchange: Börsen-ID (okex, binance, etc.)
symbol: Trading-Paar
start_time: Start der Analyseperiode
end_time: Ende der Analyseperiode
Returns:
Liste von Liquidation-Events
"""
if start_time is None:
start_time = datetime.utcnow() - timedelta(hours=24)
if end_time is None:
end_time = datetime.utcnow()
# Tardis API für historische Daten
params = {
"exchange": exchange,
"symbol": symbol,
"channel": "liquidations",
"from": int(start_time.timestamp()),
"to": int(end_time.timestamp()),
"limit": 10000
}
try:
response = await self.client.get(
f"{self.TARDIS_HTTP_URL}/historical",
params=params
)
response.raise_for_status()
data = response.json()
liquidations = [
{
"timestamp": datetime.fromtimestamp(ev["timestamp"] / 1000),
"price": ev["price"],
"size": ev["size"],
"side": ev["side"]
}
for ev in data.get("data", [])
]
print(f"✅ {len(liquidations)} historische Liquidation-Events geladen")
return liquidations
except httpx.HTTPStatusError as e:
print(f"⚠️ Tardis API Fehler: {e.response.status_code}")
# Fallback: Leere Liste
return []
async def run_backtest(
self,
liquidations: List[Dict],
candles: List[Dict],
stop_loss_pct: float = 2.0,
take_profit_pct: float = 5.0
) -> List[BacktestTrade]:
"""
Führt Backtest mit Stop-Loss-Strategie durch
Args:
liquidations: Historische Liquidation-Daten
candles: OHLCV Candlestick-Daten
stop_loss_pct: Stop-Loss Schwelle in Prozent
take_profit_pct: Take-Profit Schwelle in Prozent
Returns:
Liste von BacktestTrade-Objekten
"""
trades = []
position = None
liquidation_set = set(
int(l["timestamp"].timestamp() * 1000)
for l in liquidations
)
for i, candle in enumerate(candles):
ts = int(candle["timestamp"] * 1000)
if position is None:
# Prüfe auf Einstiegssignal
if ts in liquidation_set:
# Einstieg nach Liquidation-Event
position = {
"entry_time": datetime.fromtimestamp(ts / 1000),
"entry_price": candle["close"],
"size": 1.0,
"side": "long"
}
else:
# Prüfe auf Ausstieg
pnl_pct = ((candle["close"] - position["entry_price"])
/ position["entry_price"]) * 100
stop_hit = pnl_pct <= -stop_loss_pct
tp_hit = pnl_pct >= take_profit_pct
if stop_hit or tp_hit:
exit_price = candle["close"]
trades.append(BacktestTrade(
entry_time=position["entry_time"],
exit_time=datetime.fromtimestamp(ts / 1000),
entry_price=position["entry_price"],
exit_price=exit_price,
size=position["size"],
side=position["side"],
pnl=pnl_pct,
stop_loss_triggered=stop_hit,
liquidation_event_at_entry=True
))
position = None
return trades
async def calculate_metrics(self, trades: List[BacktestTrade]) -> Dict:
"""Berechnet Backtest-Performance-Metriken"""
if not trades:
return {
"total_trades": 0,
"win_rate": 0.0,
"avg_pnl": 0.0,
"max_drawdown": 0.0,
"profit_factor": 0.0
}
winning_trades = [t for t in trades if t.pnl > 0]
losing_trades = [t for t in 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))
# Sharpe-Ratio Approximation
returns = [t.pnl for t in trades]
avg_return = sum(returns) / len(returns)
std_return = (sum((r - avg_return) ** 2 for r in returns) / len(returns)) ** 0.5
sharpe = avg_return / std_return if std_return > 0 else 0
return {
"total_trades": len(trades),
"winning_trades": len(winning_trades),
"losing_trades": len(losing_trades),
"win_rate": len(winning_trades) / len(trades) * 100,
"avg_pnl": avg_return,
"max_drawdown": abs(min(returns)) if returns else 0,
"profit_factor": total_wins / total_losses if total_losses > 0 else float('inf'),
"sharpe_ratio": sharpe,
"stop_loss_hits": sum(1 for t in trades if t.stop_loss_triggered)
}
async def close(self):
"""Schließt HTTP-Client"""
await self.client.aclose()
Beispiel-Nutzung
async def main():
# Tardis API-Key hier einfügen
TARDIS_API_KEY = "YOUR_TARDIS_API_KEY"
backtester = TardisBacktester(TARDIS_API_KEY)
# Lade historische Daten (letzte 7 Tage)
liquidations = await backtester.fetch_historical_liquidations(
exchange="okex",
symbol="BTC-PERPETUAL",
start_time=datetime.utcnow() - timedelta(days=7)
)
# Simulierte Candle-Daten (in Produktion von Tardis laden)
candles = [
{
"timestamp": datetime.utcnow() - timedelta(hours=i),
"open": 67000 + i * 10,
"high": 67100 + i * 10,
"low": 66900 + i * 10,
"close": 67050 + i * 10,
"volume": 1000
}
for i in range(168)
]
# Führe Backtest durch
trades = await backtester.run_backtest(
liquidations=liquidations,
candles=candles,
stop_loss_pct=2.0,
take_profit_pct=5.0
)
# Berechne Metriken
metrics = await backtester.calculate_metrics(trades)
print("\n📊 BACKTEST ERGEBNISSE:")
print(f" Gesamte Trades: {metrics['total_trades']}")
print(f" Win-Rate: {metrics['win_rate']:.1f}%")
print(f" Ø PnL: {metrics['avg_pnl']:.2f}%")
print(f" Profit-Faktor: {metrics['profit_factor']:.2f}")
print(f" Sharpe-Ratio: {metrics['sharpe_ratio']:.2f}")
print(f" Stop-Loss-Hits: {metrics['stop_loss_hits']}")
await backtester.close()
if __name__ == "__main__":
asyncio.run(main())
Schritt 4: Vollständige Pipeline-Integration
# full_pipeline.py
"""
Vollständige Pipeline: OKX → Tardis → HolySheep AI → Backtest
Kostenoptimiert mit HolySheep ($0.42/1M Tokens)
"""
import asyncio
import json
from datetime import datetime, timedelta
from typing import List, Dict
from dataclasses import dataclass, asdict
import httpx
Importiere unsere Module
from okx_liquidation_collector import OKXLiquidationCollector
from holysheep_sentiment_analyzer import HolySheepSentimentAnalyzer
from tardis_backtester import TardisBacktester, BacktestTrade
@dataclass
class TradingSignal:
"""Handelssignal basierend auf KI-Analyse"""
timestamp: datetime
action: str # "BUY", "SELL", "HOLD"
entry_price: float
stop_loss_pct: float
take_profit_pct: float
confidence: float
holysheep_reasoning: str
risk_level: str
class FullPipeline:
"""Integrierte Pipeline für Liquidation-basiertes Trading"""
def __init__(
self,
holysheep_api_key: str,
tardis_api_key: str = None
):
self.sentiment_analyzer = HolySheepSentimentAnalyzer(holysheep_api_key)
self.backtester = TardisBacktester(tardis_api_key) if tardis_api_key else None
self.liquidation_collector = OKXLiquidationCollector()
# Buffer für Batch-Verarbeitung
self.liquidation_buffer: List[Dict] = []
self.buffer_size = 10 # Analyse alle 10 Events
self.last_analysis_time = datetime.utcnow()
self.analysis_interval = 60 # Sekunden zwischen Analysen
async def start_realtime_pipeline(self):
"""Startet Echtzeit-Pipeline mit automatischer Analyse"""
async def on_liquidation(liquidation: Dict):
self.liquidation_buffer.append(liquidation)
# Batch-Analyse wenn Buffer voll
if len(self.liquidation_buffer) >= self.buffer_size:
await self._analyze_batch()
# Zeitbasierte Analyse
elapsed = (datetime.utcnow() - self.last_analysis_time).total_seconds()
if elapsed >= self.analysis_interval and self.liquidation_buffer:
await self._analyze_batch()
self.liquidation_collector.on_liquidation(on_liquidation)
print("🚀 Starte Echtzeit-Pipeline...")
print(f" 📦 Buffer-Größe: {self.buffer_size}")
print(f" ⏱️ Analyse-Intervall: {self.analysis_interval}s")
print(f" 💰 HolySheep Modell: DeepSeek V3.2 ($0.42/1M)")
await self.liquidation_collector.connect()
async def _analyze_batch(self):
"""Analysiert den aktuellen Buffer mit HolySheep AI"""
if not self.liquidation_buffer:
return
print(f"\n📊 Analysiere {len(self.liquidation_buffer)} Liquidation-Events...")
# Hole aktuellen BTC-Preis (simplifiziert)
current_price = 67234.50 # In Produktion von API abrufen
# HolySheep KI-Analyse
result = await self.sentiment_analyzer.analyze_liquidation_batch(
liquidations=self.liquidation_buffer.copy(),
current_btc_price=current_price
)
# Generiere Handelssignal
signal = TradingSignal(
timestamp=datetime.utcnow(),
action=self._determine_action(result),
entry_price=current_price,
stop_loss_pct=result.recommended_stop_loss_pct,
take_profit_pct=result.recommended_stop_loss_pct * 2.5,
confidence=result.confidence,
holysheep_reasoning=result.reasoning,
risk_level=result.squeeze_risk_level
)
self._emit_signal(signal)
# Reset Buffer
self.liquidation_buffer = []
self.last_analysis_time = datetime.utcnow()
def _determine_action(self, result) -> str:
"""Bestimmt Handelsaktion basierend auf Analyse"""
if result.liquidation_bulk_probability > 0.7:
return "SELL" if result.squeeze_risk_level in ["HIGH", "EXTREME"] else "HOLD"
elif result.liquidation_bulk_probability > 0.4:
return "HOLD"
else:
return "BUY"
def _emit_signal(self, signal: TradingSignal):
"""Gibt Handelssignal aus"""
print(f"\n{'='*60}")
print(f"📈 TRADING SIGNAL")
print(f"{'='*60}")
print(f" ⏰ Zeit: {signal.timestamp.isoformat()}")
print(f" 🎯 Aktion: {signal.action}")
print(f" 💵 Einstiegspreis: ${signal.entry_price:,.2f}")
print(f" 🛑 Stop-Loss: {signal.stop_loss_pct}%")
print(f" 🎯 Take-Profit: {signal.take_profit_pct}%")
print(f" 📊 Konfidenz: {signal.confidence:.1%}")
print(f" ⚠️ Risiko: {signal.risk_level}")
print(f" 💭 Begründung: {signal.holysheep_reasoning}")
print(f"{'='*60}")
async def run_historical_backtest(
self,
start_date: datetime,
end_date: datetime
) -> Dict:
"""Führt vollständigen historischen Backtest durch"""
print(f"\n🔄 Starte historischen Backtest...")
print(f" 📅 Zeitraum: {start_date.date()} bis {end_date.date()}")
# 1. Lade Tardis-Daten
liquidations = await self.backtester.fetch_historical_liquidations(
start_time=start_date,
end_time=end_date
)
# 2. Gruppiere nach Zeitfenstern für HolySheep
window_size = 50
signals = []
for i in range(0, len(liquidations), window_size):
window = liquidations[i:i+window_size]
# Erstelle aggregierten Marktkontext
avg_price = sum(l["price"] * l["size"] for l in window) / sum(l["size"] for l in window)
market_context = f"Analyse von {len(window)} Liquidation-Events. Ø-Preis: ${avg_price:,.2f}"
# HolySheep Analyse
result = await self.sentiment_analyzer.analyze_liquidation_batch(
liquidations=window,
current_btc_price=window[-1]["price"],
market_context=market_context
)
signals.append({
"timestamp": window[-1]["timestamp"],
"recommended_sl": result.recommended_stop_loss_pct,
"risk_level": result.squeeze_risk_level
})
# 3. Führe Backtest mit dynamischen Stop-Loss durch
# (Hier vereinfacht - in Produktion komplex