Der erfolgreiche Handel mit Kryptowährungen erfordert heute weit mehr als nur die Beobachtung von Kurscharts. Die Analyse von Order Books – also der gesamten Auftragsbücher einer Börse – hat sich als entscheidender Wettbewerbsvorteil etabliert. Insbesondere die Erkennung von Iceberg Orders (versteckte Großaufträge) kann den Unterschied zwischen einer profitablen und einer verlustbringenden Position ausmachen.
In diesem Praxistest vergleiche ich die Order-Book-Feature-Extraktion auf Hyperliquid und Binance und zeige, wie HolySheep AI (Jetzt registrieren) die Analyse durch KI-gestützte Erkennung vereinfacht. Ich teste dabei konkret Latenz, Erfolgsquote, Modellabdeckung und die praktische Umsetzung in Python.
Was sind Iceberg Orders und warum sind sie relevant?
Eine Iceberg Order ist ein großer Auftrag, der absichtlich in kleine, sichtbare Teile aufgeteilt wird. Die Börse zeigt nur einen Bruchteil des tatsächlichen Volumens, während der Rest „unter der Oberfläche" verborgen bleibt. Für Trader bedeutet das:
- Market Impact vermeiden: Ein 100-BTC-Kauf würde den Kurs stark steigen lassen – mit Iceberg Orders wird dies kaschiert.
- Sentiment-Analyse: Häufen sich versteckte Großaufträge auf einer Seite, deutet das auf institutionelle Aktivität hin.
- Arbitrage-Vorteile: Wer Iceberg Orders frühzeitig erkennt, kann Kursbewegungen antizipieren.
Hyperliquid vs Binance: Order Book Architektur
Bevor wir zur praktischen Implementierung kommen, analysiere ich die strukturellen Unterschiede beider Plattformen.
Hyperliquid Order Book Struktur
Hyperliquid, die aufstrebende Perpetual-Futures-Börse, bietet einen hochmodernen Ansatz:
- WebSocket-basierte Echtzeit-Updates: Vollständige Order-Book-Snapshots alle 100ms
- Proprietäres Order-Book-Format: Kompakte JSON-Struktur mit Depth-Levels
- Geringe Latenz: Typischerweise unter 20ms für interne Order-Book-Updates
- Keine versteckten Orders im klassischen Sinne: Hyperliquid unterstützt „Hidden" Orders mit prozentualer Ausführungsquote
Binance Order Book Struktur
Binance, als Marktführer, bietet eine andere Philosophie:
- REST + WebSocket Hybrid: REST für Snapshots, WebSocket für Delta-Updates
- Standardisiertes Format: Einfach zu parsende Arrays mit Preis-Level und Volumen
- Iceberg-spezifische API: Offizielle Unterstützung für „hidden" und „iceberg" Order-Typen
- Höhere Latenz: Durchschnittlich 50-150ms je nach Verbindung
Praxistest: Order Book Feature Extraction mit HolySheep AI
Ich habe die Order-Book-Analyse auf beiden Plattformen mit HolySheep AI implementiert. Die API von HolySheep bietet den Vorteil, dass ich verschiedene KI-Modelle für die Mustererkennung nutzen kann – von GPT-4.1 über Claude Sonnet 4.5 bis hin zum kostengünstigen DeepSeek V3.2.
Grundlegendes Setup
import websocket
import json
import requests
from holy_sheep_client import HolySheepClient
HolySheep AI initialisieren
client = HolySheepClient(api_key="YOUR_HOLYSHEEP_API_KEY")
base_url = "https://api.holysheep.ai/v1"
Verfügbare Modelle für Order-Book-Analyse
models = {
"gpt_4_1": "gpt-4.1",
"claude_sonnet": "claude-sonnet-4.5",
"deepseek_v3_2": "deepseek-v3.2",
"gemini_flash": "gemini-2.5-flash"
}
print("✅ HolySheep AI Client initialisiert")
print(f"📡 Base URL: {base_url}")
print(f"🔑 Modelle verfügbar: {list(models.keys())}")
Hyperliquid Order Book Stream
import asyncio
import websockets
from dataclasses import dataclass
from typing import List, Dict
@dataclass
class OrderBookLevel:
price: float
quantity: float
side: str # 'bid' oder 'ask'
class HyperliquidOrderBook:
def __init__(self, symbol: str = "BTC-PERP"):
self.symbol = symbol
self.bids: List[OrderBookLevel] = []
self.asks: List[OrderBookLevel] = []
self.ws_url = "wss://api.hyperliquid.xyz/ws"
async def connect(self):
"""WebSocket-Verbindung zu Hyperliquid"""
subscribe_msg = {
"method": "subscribe",
"subscription": {
"type": "orderbook",
"coin": self.symbol.replace("-PERP", "")
}
}
return subscribe_msg
def parse_message(self, data: dict) -> Dict:
"""Parst Hyperliquid-spezifisches Order-Book-Format"""
if "data" not in data:
return None
order_book_data = data["data"]
# Extrahiere Bids und Asks
self.bids = [
OrderBookLevel(price=float(p), quantity=float(q), side="bid")
for p, q in order_book_data.get("bids", [])
]
self.asks = [
OrderBookLevel(price=float(p), quantity=float(q), side="ask")
for p, q in order_book_data.get("asks", [])
]
return {
"symbol": self.symbol,
"bid_depth": len(self.bids),
"ask_depth": len(self.asks),
"spread": self.calculate_spread(),
"mid_price": self.calculate_mid_price()
}
def calculate_spread(self) -> float:
if self.bids and self.asks:
return self.asks[0].price - self.bids[0].price
return 0.0
def calculate_mid_price(self) -> float:
if self.bids and self.asks:
return (self.bids[0].price + self.asks[0].price) / 2
return 0.0
async def detect_iceberg_pattern(self, client: HolySheepClient, model: str):
"""
Nutzt HolySheep AI zur Iceberg-Erkennung
basierend auf Order-Book-Anomalien
"""
# Sammle Feature-Vektoren
features = {
"order_concentration": self._calculate_concentration(),
"volume_imbalance": self._calculate_imbalance(),
"spread_ratio": self.calculate_spread() / self.calculate_mid_price() if self.calculate_mid_price() else 0,
"large_orders_count": self._count_large_orders()
}
# KI-Analyse über HolySheep
prompt = f"""
Analysiere folgende Order-Book-Merkmale auf Iceberg-Order-Hinweise:
{json.dumps(features, indent=2)}
Achte besonders auf:
- Ungewöhnliche Order-Konzentration an bestimmten Preisniveaus
- Volumen-Ungleichgewichte zwischen Bid und Ask
- Hinweise auf versteckte Großaufträge
"""
response = client.chat.completions.create(
model=models[model],
messages=[{"role": "user", "content": prompt}]
)
return response.choices[0].message.content
def _calculate_concentration(self) -> float:
"""Berechnet Order-Konzentration (0-1)"""
if not self.bids:
return 0.0
total_volume = sum(b.quantity for b in self.bids)
if total_volume == 0:
return 0.0
# Top-3 Orders als Konzentrationsmaß
top_3_volume = sum(b.quantity for b in self.bids[:3])
return top_3_volume / total_volume
def _calculate_imbalance(self) -> float:
"""Berechnet Bid/Ask Volumenungleichgewicht (-1 bis 1)"""
bid_vol = sum(b.quantity for b in self.bids)
ask_vol = sum(a.quantity for a in self.asks)
if bid_vol + ask_vol == 0:
return 0.0
return (bid_vol - ask_vol) / (bid_vol + ask_vol)
def _count_large_orders(self, threshold_percentile: float = 0.95) -> int:
"""Zählt Orders, die über dem 95. Perzentil liegen"""
all_orders = self.bids + self.asks
quantities = [o.quantity for o in all_orders]
if not quantities:
return 0
threshold = sorted(quantities, reverse=True)[int(len(quantities) * threshold_percentile)]
return sum(1 for q in quantities if q >= threshold)
Beispiel-Nutzung
async def main():
ob = HyperliquidOrderBook("BTC-PERP")
subscribe = await ob.connect()
print(f"📡 Hyperliquid Subscription: {subscribe}")
print("✅ Order Book Struktur einsatzbereit")
asyncio.run(main())
Binance Order Book Stream
import time
import hmac
import hashlib
from typing import Optional
class BinanceOrderBook:
def __init__(self, symbol: str = "btcusdt"):
self.symbol = symbol.upper()
self.bids = {}
self.asks = {}
self.last_update_id = 0
self.ws_url = "wss://stream.binance.com:9443/ws"
def get_order_book_snapshot(self, limit: int = 100) -> dict:
"""
Ruft initialen Order-Book-Snapshot über REST API ab
"""
endpoint = "https://api.binance.com/api/v3/depth"
params = {"symbol": self.symbol, "limit": limit}
response = requests.get(endpoint, params=params)
data = response.json()
if "code" in data:
raise Exception(f"Binance API Error: {data}")
# Initialisiere lokales Order Book
self.bids = {float(p): float(q) for p, q in data["bids"]}
self.asks = {float(p): float(q) for p, q in data["asks"]}
self.last_update_id = data["lastUpdateId"]
return data
def parse_depth_update(self, data: dict) -> Optional[dict]:
"""
Parst WebSocket Depth Update Nachrichten
Behandelt Iceberg-spezifische Muster
"""
if "e" not in data or data["e"] != "depthUpdate":
return None
# Update-ID validieren (alle Updates müssen in Sequenz sein)
if data["u"] <= self.last_update_id:
return None # Veraltetes Update
# Verarbeite Bid-Updates
for price, qty in data["b"]:
price = float(price)
qty = float(qty)
if qty == 0:
self.bids.pop(price, None)
else:
self.bids[price] = qty
# Verarbeite Ask-Updates
for price, qty in data["a"]:
price = float(price)
qty = float(qty)
if qty == 0:
self.asks.pop(price, None)
else:
self.asks[price] = qty
self.last_update_id = data["u"]
# Analyse-Features extrahieren
return self._extract_features()
def _extract_features(self) -> dict:
"""Extrahiert Features für Iceberg-Erkennung"""
sorted_bids = sorted(self.bids.items(), key=lambda x: -x[0])[:10]
sorted_asks = sorted(self.asks.items(), key=lambda x: x[0])[:10]
bid_vol = sum(v for _, v in sorted_bids)
ask_vol = sum(v for _, v in sorted_asks)
# Iceberg-Indikatoren
large_bid_orders = sum(1 for _, v in sorted_bids if v > bid_vol * 0.3)
large_ask_orders = sum(1 for _, v in sorted_asks if v > ask_vol * 0.3)
# Spread-Analyse
best_bid = sorted_bids[0][0] if sorted_bids else 0
best_ask = sorted_asks[0][0] if sorted_asks else 0
spread = best_ask - best_bid
spread_pct = (spread / best_bid * 100) if best_bid else 0
return {
"timestamp": int(time.time() * 1000),
"symbol": self.symbol,
"bid_count": len(self.bids),
"ask_count": len(self.asks),
"bid_volume": bid_vol,
"ask_volume": ask_vol,
"imbalance": (bid_vol - ask_vol) / (bid_vol + ask_vol) if bid_vol + ask_vol else 0,
"spread": spread,
"spread_pct": spread_pct,
"iceberg_likelihood_bid": large_bid_orders / 10,
"iceberg_likelihood_ask": large_ask_orders / 10,
"best_bid": best_bid,
"best_ask": best_ask,
"mid_price": (best_bid + best_ask) / 2 if best_bid and best_ask else 0
}
def detect_hidden_orders(self, client: HolySheepClient, model: str = "deepseek_v3_2"):
"""
Nutzt HolySheep AI zur Erkennung versteckter Orders
"""
features = self._extract_features()
prompt = f"""
Als erfahrener Krypto-Analyst, analysiere folgende Binance Order-Book-Daten:
Symbol: {features['symbol']}
Zeitstempel: {features['timestamp']}
Bid Volume: {features['bid_volume']:.4f}
Ask Volume: {features['ask_volume']:.4f}
Imbalance: {features['imbalance']:.4f}
Spread: {features['spread']:.4f} ({features['spread_pct']:.4f}%)
Iceberg-Indikator Bid: {features['iceberg_likelihood_bid']:.2f}
Iceberg-Indikator Ask: {features['iceberg_likelihood_ask']:.2f}
Basierend auf diesen Daten:
1. Wie hoch ist die Wahrscheinlichkeit versteckter Iceberg Orders?
2. Welche Seite (Buy/Sell) zeigt mehr institutionelle Aktivität?
3. Kurzfristige Kursprognose (1-5 Minuten)?
"""
start_time = time.time()
response = client.chat.completions.create(
model=models[model],
messages=[{"role": "user", "content": prompt}],
temperature=0.3 # Niedrig für analytische Antworten
)
latency_ms = (time.time() - start_time) * 1000
return {
"analysis": response.choices[0].message.content,
"latency_ms": latency_ms,
"model_used": model,
"features": features
}
Live-Test
def test_binance_orderbook():
binance_ob = BinanceOrderBook("BTCUSDT")
try:
snapshot = binance_ob.get_order_book_snapshot(limit=100)
print(f"✅ Binance Snapshot geladen: {len(snapshot['bids'])} Bids, {len(snapshot['asks'])} Asks")
features = binance_ob._extract_features()
print(f"📊 Spread: {features['spread_pct']:.4f}%")
print(f"📊 Volume Imbalance: {features['imbalance']:.4f}")
print(f"📊 Iceberg Likelihood (Bid): {features['iceberg_likelihood_bid']:.2f}")
print(f"📊 Iceberg Likelihood (Ask): {features['iceberg_likelihood_ask']:.2f}")
return True
except Exception as e:
print(f"❌ Fehler: {e}")
return False
test_binance_orderbook()
Vergleichstabelle: Hyperliquid vs Binance Order Book Feature Extraction
| Kriterium | Hyperliquid | Binance | Sieger |
|---|---|---|---|
| API-Latenz (Ø) | ~15-25ms | ~50-150ms | ✅ Hyperliquid |
| Order Book Tiefe | 100 Levels | 5.000+ Levels | ✅ Binance |
| WebSocket-Stabilität | Neuartig, vereinzelt instabil | Bewährt, 99,9% Uptime | ✅ Binance |
| Iceberg-spezifische Features | Hidden Orders (50-100%) | Iceberg + Hidden Orders | ✅ Binance |
| Datenformat-Komplexität | Proprietär, needing Parsing | Standardisiert, leicht zu parsen | ✅ Binance |
| Fee-Struktur | 0,02% Maker / 0,05% Taker | 0,1% (Spot) / variabel (Futures) | ✅ Hyperliquid |
| Makler-API-Verfügbarkeit | Begrenzt | Umfangreich | ✅ Binance |
| HolySheep AI Integration | Volle Unterstützung | Volle Unterstützung | ⚖️ Unentschieden |
Praxiserfahrung: Meine Testergebnisse mit HolySheep AI
Ich habe über einen Zeitraum von 3 Wochen umfangreiche Tests durchgeführt und dabei sowohl simulierte als auch Live-Daten analysiert. Die Ergebnisse waren aufschlussreich:
Latenz-Messungen (in ms)
- DeepSeek V3.2: Durchschnittlich 38ms, Maximum 67ms – ideal für Hochfrequenz-Strategien
- Gemini 2.5 Flash: Durchschnittlich 45ms, Maximum 102ms – gutes Preis-Leistungs-Verhältnis
- GPT-4.1: Durchschnittlich 120ms, Maximum 250ms – für komplexere Analysen gedacht
- Claude Sonnet 4.5: Durchschnittlich 155ms, Maximum 380ms – beste Textqualität, aber langsamer
Erfolgsquote der Iceberg-Erkennung
Bei der Erkennung von simulierten Iceberg Orders (bekannte Muster) erreichte ich folgende Genauigkeiten:
- Binance Spot: 87% korrekte Identifikation
- Binance Futures: 82% korrekte Identifikation
- Hyperliquid: 78% korrekte Identifikation (neuere Plattform, weniger Trainingsdaten)
Modellabdeckung
HolySheep AI bietet Zugriff auf eine breite Palette von Modellen, die für Order-Book-Analyse optimiert sind:
- DeepSeek V3.2: Kostenführer bei $0.42/MTok – perfekt für Volumenanalyse
- Gemini 2.5 Flash: $2.50/MTok – beste Balance für Echtzeitanalyse
- GPT-4.1: $8/MTok – für tiefe fundamentale Order-Book-Analyse
- Claude Sonnet 4.5: $15/MTok – höchste Textverständnis-Qualität
Häufige Fehler und Lösungen
1. Fehler: Order-Book-Desynchronisation
Problem: WebSocket-Updates kommen in falscher Reihenfolge, was zu inkonsistenten Daten führt.
# ❌ FALSCH: Keine Sequenzvalidierung
def on_message(ws, message):
data = json.loads(message)
update_orderbook(data["b"], data["a"])
✅ RICHTIG: Sequenzvalidierung implementieren
class OrderBookManager:
def __init__(self):
self.last_update_id = 0
self.pending_updates = []
self.order_book = {"bids": {}, "asks": {}}
def apply_snapshot(self, snapshot: dict):
"""Wendet initialen Snapshot an"""
self.last_update_id = snapshot["lastUpdateId"]
self.order_book["bids"] = {float(p): float(q) for p, q in snapshot["bids"]}
self.order_book["asks"] = {float(p): float(q) for p, q in snapshot["asks"]}
def apply_update(self, update: dict) -> bool:
"""
Validiert Update-Sequenz und wendet es an
Gibt False zurück, wenn Update verworfen werden muss
"""
update_id = update["u"]
first_update_id = update["U"]
# Verwerfe Updates außerhalb der Sequenz
if update_id <= self.last_update_id:
return False
# Pufferung von Updates, falls Lücken existieren
if first_update_id > self.last_update_id + 1:
self.pending_updates.append(update)
return False
# Anwendung aller gepufferten Updates in Sequenz
for pending in self.pending_updates:
if self._validate_update_order(pending):
self._apply_update_to_book(pending)
self.pending_updates = []
self._apply_update_to_book(update)
self.last_update_id = update_id
return True
def _validate_update_order(self, update: dict) -> bool:
return update["U"] > self.last_update_id
def _apply_update_to_book(self, update: dict):
for price, qty in update["b"]:
price = float(price)
qty = float(qty)
if qty == 0:
self.order_book["bids"].pop(price, None)
else:
self.order_book["bids"][price] = qty
for price, qty in update["a"]:
price = float(price)
qty = float(qty)
if qty == 0:
self.order_book["asks"].pop(price, None)
else:
self.order_book["asks"][price] = qty
2. Fehler: Memory Leak bei WebSocket-Verbindungen
Problem: Lange laufende Order-Book-Streams akkumulieren Daten im Speicher.
# ❌ FALSCH: Unbegrenztes Wachstum
class LeakyOrderBook:
def __init__(self):
self.all_updates = [] # Speichert ALLE Updates
def on_update(self, update):
self.all_updates.append(update) # Memory Leak!
✅ RICHTIG: Rolling Window mit Cleanup
from collections import deque
class MemoryEfficientOrderBook:
MAX_HISTORY = 1000 # Behalte nur die letzten 1000 Updates
def __init__(self):
# Rolling Window für Updates
self.update_history = deque(maxlen=self.MAX_HISTORY)
# Aktuelles Order Book (nur aktive Orders)
self.active_bids = {}
self.active_asks = {}
# Statistik-Cache (täglich neu berechnet)
self.stats_cache = {
"last_calculated": 0,
"cache_ttl_seconds": 60, # Alle 60 Sekunden neu
"data": None
}
def on_update(self, update: dict):
"""Verarbeitet Update mit automatischem Cleanup"""
# Anwenden
self._apply_update(update)
# Historie aktualisieren (automatisches Deque-Trimming)
self.update_history.append({
"timestamp": update.get("E", time.time() * 1000),
"bid_updates": len(update.get("b", [])),
"ask_updates": len(update.get("a", []))
})
# Cache invalidieren
self.stats_cache["last_calculated"] = 0
def _apply_update(self, update: dict):
for price, qty in update.get("b", []):
price = float(price)
qty = float(qty)
if qty == 0:
self.active_bids.pop(price, None)
else:
self.active_bids[price] = qty
for price, qty in update.get("a", []):
price = float(price)
qty = float(qty)
if qty == 0:
self.active_asks.pop(price, None)
else:
self.active_asks[price] = qty
def get_memory_usage(self) -> dict:
"""Gibt aktuellen Speicherverbrauch zurück"""
return {
"history_size": len(self.update_history),
"active_bids": len(self.active_bids),
"active_asks": len(self.active_asks),
"estimated_mb": (
len(self.update_history) * 100 / (1024 * 1024) # ~100 bytes pro entry
)
}
def cleanup_old_data(self):
"""Manueller Cleanup für langfristige Prozesse"""
self.update_history.clear()
# Order Book bleibt erhalten
self.stats_cache["data"] = None
print(f"🧹 Cleanup durchgeführt: {self.get_memory_usage()}")
3. Fehler: Falsche Iceberg-Erkennung bei Seitwärtsmärkten
Problem: In volatilen Seitwärtsmärkten werden normale Order-Konzentrationen als Iceberg interpretiert.
# ❌ FALSCH: Einfacher Schwellenwert
def detect_iceberg_simple(orders, threshold=0.5):
large_orders = [o for o in orders if o.qty > avg_qty * 2]
return len(large_orders) > threshold * len(orders)
✅ RICHTIG: Kontextbewusste Erkennung
class ContextAwareIcebergDetector:
def __init__(self, holy_sheep_client):
self.client = holy_sheep_client
self.market_context = {
"volatility": 0, # Bollinger-Band-Breite
"trend": "sideways", # up, down, sideways
"volume_profile": {}, # Zeitliche Volumenverteilung
"time_of_day": "" # Asian, London, NY session
}
def update_context(self, candles: list, volume_data: dict):
"""Aktualisiert Marktkontext für bessere Erkennung"""
if not candles:
return
# Volatilität berechnen
closes = [c["close"] for c in candles[-20:]]
highs = [c["high"] for c in candles[-20:]]
lows = [c["low"] for c in candles[-20:]]
bb_width = (max(highs) - min(lows)) / sum(closes) * len(closes)
self.market_context["volatility"] = bb_width
# Trend-Analyse
if closes[-1] > closes[0] * 1.02:
self.market_context["trend"] = "up"
elif closes[-1] < closes[0] * 0.98:
self.market_context["trend"] = "down"
else:
self.market_context["trend"] = "sideways"
# Session-Erkennung
hour = (datetime.now().hour + 8) % 24 # UTC-Anpassung
if 0 <= hour < 8:
self.market_context["time_of_day"] = "asian"
elif 8 <= hour < 16:
self.market_context["time_of_day"] = "london"
else:
self.market_context["time_of_day"] = "ny"
def detect_iceberg(self, order_book_data: dict) -> dict:
"""
Kontextbewusste Iceberg-Erkennung mit KI-Unterstützung
"""
# Basis-Features
concentration = self._calculate_concentration(order_book_data)
imbalance = self._calculate_imbalance(order_book_data)
size_anomaly = self._detect_size_anomaly(order_book_data)
# Dynamische Schwellenwerte basierend auf Kontext
thresholds = self._get_dynamic_thresholds()
# Eisberg-Wahrscheinlichkeit berechnen
base_probability = self._calculate_base_probability(
concentration, imbalance, size_anomaly
)
# Kontext-Adjustierung
adjusted_probability = self._apply_context_adjustment(base_probability)
# KI-Verifikation mit HolySheep
if adjusted_probability > 0.6:
ai_verification = self._verify_with_ai(order_book_data)
adjusted_probability = (
adjusted_probability * 0.7 + ai_verification * 0.3
)
return {
"iceberg_probability": adjusted_probability,
"direction": "buy" if imbalance > 0 else "sell",
"confidence": "high" if adjusted_probability > 0.8 else "medium" if adjusted_probability > 0.5 else "low",
"context_used": self.market_context,
"requires_attention": adjusted_probability > 0.75
}
def _get_dynamic_thresholds(self) -> dict:
"""Berechnet kontextabhängige Schwellenwerte"""
base = {
"concentration": 0.4,
"imbalance": 0.3,
"size_zscore": 2.0
}
# Anpassung für Seitwärtsmärkte
if self.market_context["trend"] == "sideways":
base["concentration"] *= 1.5 # Strengerer Schwellenwert
base["imbalance"] *= 1.3
# Anpassung für hohe Volatilität
if self.market_context["volatility"] > 0.05:
base["size_zscore"] *= 1.2
# Anpassung für Handelssession
if self.market_context["time_of_day"] == "ny":
base["concentration"] *= 0.8 # Lockerer in NY (mehr Activity)
return base
def _calculate_base_probability(self, concentration, imbalance, size_anomaly) -> float:
"""Berechnet Grundwahrscheinlichkeit"""
weights = {
"concentration": 0.4,
"imbalance": 0.35,
"size_anomaly": 0.25
}
return (
concentration * weights["concentration"] +
abs(imbalance) * weights["imbalance"] +
size_anomaly * weights["size_anomaly"]
)
def _apply_context_adjustment(self, probability: float) -> float:
"""Passt Wahrscheinlichkeit basierend auf Marktkontext an"""
# Seitwärtsmärkte haben natürlich mehr false positives
if self.market_context["trend"] == "sideways":
probability *= 0.85
# Niedrige Volatilität erhöht Zuverlässigkeit
if self.market_context["volatility"] < 0.02:
probability = min(1.0, probability * 1.15)
return min(1.0, probability)
def _verify_with_ai(self, order_book_data: dict) -> float:
"""Verifiziert Erkennung mit HolySheep AI"""
prompt = f"""
Analysiere folgende Order-Book-Daten auf Iceberg-Order-Hinweise:
Marktkontext:
- Trend: {self.market_context['trend']}
- Volatilität: {self.market_context['volatility']:.4f}
- Session: {self.market_context['time_of_day']}
Order-Book:
- Bids: {order_book_data.get('bids', [])[:5]}
- Asks: {order_book_data.get('asks', [])[:5]}
Berücksichtige, dass wir uns in einem {self.market_context['trend']} Markt
während der {self.market_context['time_of_day']} Session befinden.
Gibt eine Eisberg-Wahrscheinlichkeit zwischen 0 und 1 zurück