Der Hyperliquid-Orderbuch-Replay ist eine der gefragtesten Funktionen für quantitativ orientierte Trader und Algo-Entwickler im Jahr 2026. Die historische Rekonstruktion von Limit-Orders, Trades und Liquidationsdaten ermöglicht die Entwicklung und das Backtesting von Trading-Strategien unter realen Marktbedingungen. Doch die etablierten Lösungen wie Tardis werden zunehmend durch kostengünstigere Alternativen ergänzt – insbesondere durch HolySheep AI mit seiner hochperformanten API-Infrastruktur.
Warum Orderbuch-Replay für Hyperliquid kritisch ist
Hyperliquid hat sich als eine der führenden Perpetual-Futures-Börsen positioniert. Das Orderbuch auf Level-2/Tick-Ebene enthält:
- Alle offenen Buy- und Sell-Orders mit Precisions-Time-Stamps
- Order-Modifikationen und -Löschungen (Cancels)
- Matching-Events mit Aggressor-Side-Indikator
- Liquidation-Events und Funding-Rate-Updates
Die vollständige Rekonstruktion erfordert mehr als 500 GB historischer Daten pro Monat – ein Volumen, das bei herkömmlichen Datenanbietern zu erheblichen Kosten führt.
Kostenvergleich: Tardis vs. HolySheep AI (2026)
| Kriterium | Tardis | HolySheep AI | Ersparnis |
|---|---|---|---|
| Hyperliquid Orderbuch (pro GB) | $0.89 | ¥0.12 (~$0.12) | ~87% |
| Trade-Daten (pro 1M Events) | $2.50 | ¥0.80 (~$0.80) | ~68% |
| Latenz (API-Response) | 120-180ms | <50ms | 3x schneller |
| Minimale monatliche Kosten | $49 | Kostenlose Credits | ∞ |
| Authentifizierung | API-Key | API-Key (kompatibel) | Identisch |
Geeignet / Nicht geeignet für
✅ Geeignet für:
- Algo-Trader, die Backtesting mit echten Hyperliquid-Orderbuchdaten durchführen
- Market-Maker, die Spread- und Depth-Analysen benötigen
- Forscher, die Liquidations-Ketten und Funding-Cycle-Dynamiken untersuchen
- Exchange-Aggregatoren, die Cross-Exchange-Arbitrage validieren
- Teams mit Budget-Beschränkungen, die <$200/Monat für Daten ausgeben
❌ Nicht geeignet für:
- Unternehmen, die Sub-Second-Real-Time-Feeds benötigen (hier sind dedizierte WebSocket-Streams besser)
- Nutzer, die ausschließlich CEX-Daten (Binance, Bybit) benötigen
- Projekte mit Compliance-Anforderungen, die regulatorische Datenzertifizierungen erfordern
HolySheep AI: Architektur für Orderbuch-Replay
HolySheep AI bietet eine REST-kompatible Schnittstelle, die sowohl mit Tardis-Endpoints als auch mit nativem Hyperliquid-Daten-Feed kompatibel ist. Die API unterstützt:
- Historisches Orderbuch-Snapshot (alle 100ms oder 1s rekonstruierbar)
- Incremental-Update-Stream für Echtzeit-Replay
- Aggregierte Trades mit Liquidation-Flags
- Funding-Rate-History mit 8-Stunden-Intervallen
Preise und ROI (2026)
Für ein mittelgroßes quantitatives Team mit 10 Millionen Token/Monat Verbrauch:
| Modell | Input-Kosten | Output-Kosten | Gesamt (10M Tok.) |
|---|---|---|---|
| GPT-4.1 | $40 | $160 | $200 |
| Claude Sonnet 4.5 | $75 | $225 | $300 |
| Gemini 2.5 Flash | $12.50 | $50 | $62.50 |
| DeepSeek V3.2 | $2.10 | $8.40 | $10.50 |
HolySheep-Vorteil: Mit WeChat/Alipay-Zahlung und ¥1=$1-Kurs erhalten Sie 85%+ Ersparnis gegenüber westlichen Anbietern. Die kostenlosen Credits ermöglichen den Einstieg ohne initiale Investition.
Implementierung: Orderbuch-Replay mit HolySheep AI
Beispiel 1: Abruf historischer Orderbuch-Snapshots
#!/usr/bin/env python3
"""
Hyperliquid Orderbuch-Replay via HolySheep AI API
Kompatibel mit Tardis-Endpunkt-Schema
"""
import requests
import json
from datetime import datetime, timedelta
class HyperliquidReplayClient:
"""Client für Orderbuch-Replay mit HolySheep AI"""
def __init__(self, api_key: str):
self.base_url = "https://api.holysheep.ai/v1"
self.headers = {
"Authorization": f"Bearer {api_key}",
"Content-Type": "application/json"
}
def get_orderbook_snapshot(
self,
market: str = "HYPE-PERP",
timestamp: int = None,
depth: int = 25
) -> dict:
"""
Ruft historischen Orderbuch-Snapshot ab.
Args:
market: Trading-Paar (z.B. HYPE-PERP)
timestamp: Unix-Timestamp in Millisekunden
depth: Anzahl Preisstufen pro Seite
Returns:
Dict mit bids/asks und Metadaten
"""
endpoint = f"{self.base_url}/hyperliquid/orderbook"
params = {
"market": market,
"depth": depth
}
if timestamp:
params["timestamp"] = timestamp
try:
response = requests.get(
endpoint,
headers=self.headers,
params=params,
timeout=30
)
response.raise_for_status()
return response.json()
except requests.exceptions.Timeout:
raise TimeoutError(
f"API-Timeout nach 30s bei timestamp={timestamp}"
)
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
raise RuntimeError(
"Rate-Limit erreicht. Retry-After-Header prüfen."
)
raise RuntimeError(f"HTTP {e.response.status_code}: {e}")
def get_trades(
self,
market: str = "HYPE-PERP",
start_time: int = None,
end_time: int = None,
limit: int = 1000
) -> list:
"""
Ruft historische Trades mit Liquidations-Flags ab.
"""
endpoint = f"{self.base_url}/hyperliquid/trades"
params = {
"market": market,
"limit": min(limit, 5000) # Hard limit
}
if start_time:
params["startTime"] = start_time
if end_time:
params["endTime"] = end_time
response = requests.get(
endpoint,
headers=self.headers,
params=params,
timeout=30
)
response.raise_for_status()
trades = response.json().get("data", [])
# Datenqualitätsprüfung
validated_trades = self._validate_trade_data(trades)
return validated_trades
def _validate_trade_data(self, trades: list) -> list:
"""
Validierung der Trade-Daten auf Konsistenz.
Prüft:
- Monoton steigende Timestamps
- Gültige Preise (positiv, im erwarteten Range)
- Liquidation-Flag-Konsistenz
"""
if not trades:
return []
validated = []
prev_timestamp = 0
for trade in trades:
# Timestamp-Monotonie prüfen
ts = trade.get("timestamp", 0)
if ts < prev_timestamp:
print(f"WARNUNG: Timestamp-Regression bei {ts}")
continue
# Preis-Plausibilität
price = float(trade.get("price", 0))
if price <= 0:
print(f"WARNUNG: Ungültiger Preis bei {trade.get('id')}")
continue
# Liquidation-Flag-Validierung
if trade.get("isLiquidation"):
size = float(trade.get("size", 0))
if size < 100: # Minimale Liquidation-Size
print(f"INFO: Kleine Liquidation {size} bei {ts}")
validated.append(trade)
prev_timestamp = ts
return validated
Nutzung
if __name__ == "__main__":
client = HyperliquidReplayClient(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
# Beispiel: Orderbuch vor grosser Liquidation
target_ts = int(
(datetime.now() - timedelta(hours=2)).timestamp() * 1000
)
orderbook = client.get_orderbook_snapshot(
market="HYPE-PERP",
timestamp=target_ts,
depth=50
)
print(f"Orderbuch vom {datetime.fromtimestamp(target_ts/1000)}")
print(f"Bids: {len(orderbook.get('bids', []))}")
print(f"Asks: {len(orderbook.get('asks', []))}")
Beispiel 2: Vollständiger Orderbuch-Replay-Loop mit Backtesting
#!/usr/bin/env python3
"""
Orderbuch-Replay Engine für Strategie-Backtesting
Implementiert vollständigen Market-Maker-Replay
"""
import json
import sqlite3
from typing import Generator, Tuple
from dataclasses import dataclass
from datetime import datetime
@dataclass
class OrderBookLevel:
"""Einzelne Orderbuch-Stufe"""
price: float
size: float
side: str # 'bid' oder 'ask'
order_id: str
timestamp: int
@dataclass
class Trade:
"""Trade-Event"""
id: str
price: float
size: float
side: str
is_liquidation: bool
timestamp: int
class OrderBookReplayEngine:
"""
Engine für sequentiellen Orderbuch-Replay.
Funktionsweise:
1. Lädt Snapshots in Intervallen
2. Wendet inkrementelle Updates an
3. Generiert Replay-Events für Backtesting
"""
def __init__(self, client, db_path: str = None):
self.client = client
self.current_bids = {} # price -> OrderBookLevel
self.current_asks = {}
self.db_path = db_path
self._init_db()
def _init_db(self):
"""Initialisiert lokale Cache-DB"""
if not self.db_path:
self.db_path = ":memory:"
conn = sqlite3.connect(self.db_path)
conn.execute("""
CREATE TABLE IF NOT EXISTS orderbook_cache (
timestamp INTEGER,
market TEXT,
side TEXT,
price REAL,
size REAL,
PRIMARY KEY (timestamp, market, side, price)
)
""")
conn.commit()
self.conn = conn
def replay(
self,
market: str,
start_time: int,
end_time: int,
snapshot_interval_ms: int = 1000
) -> Generator[Tuple[int, dict, list], None, None]:
"""
Generator für sequentiellen Replay.
Yields:
(timestamp, orderbook_snapshot, recent_trades)
"""
current_ts = start_time
while current_ts < end_time:
# Hole Snapshot
try:
snapshot = self.client.get_orderbook_snapshot(
market=market,
timestamp=current_ts,
depth=100
)
except Exception as e:
print(f"Fehler bei {current_ts}: {e}")
current_ts += snapshot_interval_ms
continue
# Hole Trades seit letztem Snapshot
trades = self.client.get_trades(
market=market,
start_time=current_ts - snapshot_interval_ms,
end_time=current_ts,
limit=5000
)
# Baue Orderbuch auf
orderbook = self._build_orderbook(snapshot)
# Qualitätsprüfung
if self._validate_orderbook_quality(orderbook):
yield current_ts, orderbook, trades
else:
print(f"WARNUNG: Qualitätsproblem bei {current_ts}")
current_ts += snapshot_interval_ms
def _build_orderbook(self, snapshot: dict) -> dict:
"""Konstruiert Orderbuch-Dict aus Snapshot"""
bids = {}
asks = {}
for bid in snapshot.get("bids", []):
price = float(bid["price"])
size = float(bid["size"])
if size > 0:
bids[price] = size
for ask in snapshot.get("asks", []):
price = float(ask["price"])
size = float(ask["size"])
if size > 0:
asks[price] = size
# Berechne abgeleitete Metriken
best_bid = max(bids.keys()) if bids else 0
best_ask = min(asks.keys()) if asks else float('inf')
spread = (best_ask - best_bid) / best_bid if best_bid > 0 else 0
return {
"bids": bids,
"asks": asks,
"best_bid": best_bid,
"best_ask": best_ask,
"spread_bps": spread * 10000,
"depth": sum(bids.values()) + sum(asks.values())
}
def _validate_orderbook_quality(self, orderbook: dict) -> bool:
"""
Validierung der Orderbuch-Qualität.
Checks:
- Spread im plausiblen Bereich (<5% für Perps)
- Bester Bid < Bester Ask
- Mindestanzahl an Stufen
- Gesamttiefe im plausiblen Bereich
"""
# Spread-Check
if orderbook["spread_bps"] > 500: # >5%
return False
# Bid < Ask
if orderbook["best_bid"] >= orderbook["best_ask"]:
return False
# Tiefe plausibel?
depth = orderbook["depth"]
if depth <= 0 or depth > 1_000_000_000: # Unplausibel
return False
return True
Backtesting-Integration
def run_simple_mm_backtest(engine: OrderBookReplayEngine):
"""
Einfacher Market-Maker Backtest.
Strategie:
- Platziere Bid 1 Tick unter bestem Bid
- Platziere Ask 1 Tick über bestem Ask
- Erfasse Spread als PnL
"""
market = "HYPE-PERP"
start = int((datetime.now().timestamp() - 3600) * 1000)
end = int(datetime.now().timestamp() * 1000)
tick_size = 0.01 # Anpassen je nach Markt
total_pnl = 0.0
trades = 0
for ts, ob, recent_trades in engine.replay(
market, start, end, snapshot_interval_ms=1000
):
if not ob["bids"] or not ob["asks"]:
continue
# Simulated MM-Position
bid_price = ob["best_bid"] - tick_size
ask_price = ob["best_ask"] + tick_size
# Zähle Trades im Spread
trades_in_period = len([
t for t in recent_trades
if bid_price <= t["price"] <= ask_price
])
trades += trades_in_period
# Spread-PnL (vereinfacht)
if trades_in_period > 0:
spread_pnl = trades_in_period * (ask_price - bid_price)
total_pnl += spread_pnl
print(f"Backtest Ergebnis:")
print(f" Trades: {trades}")
print(f" Spread-PnL: {total_pnl:.2f}")
Ausführung
if __name__ == "__main__":
from hyperliquid_replay import HyperliquidReplayClient
client = HyperliquidReplayClient(
api_key="YOUR_HOLYSHEEP_API_KEY"
)
engine = OrderBookReplayEngine(client)
run_simple_mm_backtest(engine)
Datenqualitätsprüfung: Validierung der Orderbuch-Daten
Die Qualität der historischen Orderbuch-Daten ist entscheidend für valide Backtesting-Ergebnisse. Folgende Prüfungen sind essentiell:
3-Stufen-Qualitätsvalidierung
#!/usr/bin/env python3
"""
Datenqualitäts-Validator für Hyperliquid Orderbuch-Replay
"""
import statistics
from typing import List, Dict, Tuple
class DataQualityValidator:
"""
Umfassende Qualitätsprüfung für Orderbuch-Daten.
Validierungsstufen:
1. Strukturelle Integrität (Schema-Check)
2. Statistische Plausibilität (Outlier-Detection)
3. Konsistenz-Validierung (Cross-Validation)
"""
def __init__(self, tolerance_bps: float = 50.0):
"""
Args:
tolerance_bps: Akzeptable Abweichung in Basispunkten
"""
self.tolerance_bps = tolerance_bps / 10000 # Zu Dezimal
def validate_orderbook_snapshot(
self,
snapshot: dict
) -> Tuple[bool, List[str]]:
"""
Validiert einen einzelnen Orderbuch-Snapshot.
Returns:
(is_valid, list_of_issues)
"""
issues = []
# 1. Struktur-Check
if "bids" not in snapshot or "asks" not in snapshot:
issues.append("FEHLER: Fehlende bids/asks keys")
return False, issues
# 2. Bid < Ask Check
best_bid = max([float(b["price"]) for b in snapshot["bids"]], default=0)
best_ask = min([float(a["price"]) for a in snapshot["asks"]], default=float('inf'))
if best_bid >= best_ask:
issues.append(
f"FEHLER: Bid ({best_bid}) >= Ask ({best_ask}) - Crossed Book"
)
# 3. Spread-Plausibilität
spread = (best_ask - best_bid) / best_bid if best_bid > 0 else 0
if spread > self.tolerance_bps:
issues.append(
f"WARNUNG: Spread {spread*10000:.1f}bps > {self.tolerance_bps*10000}bps"
)
# 4. Empty Price Check
empty_prices = [
b["price"] for b in snapshot["bids"]
if float(b["price"]) == 0
]
if empty_prices:
issues.append(f"WARNUNG: {len(empty_prices)} Bids mit Preis 0")
# 5. Negative Size Check
negative_sizes = sum(
1 for b in snapshot["bids"] + snapshot["asks"]
if float(b["size"]) < 0
)
if negative_sizes > 0:
issues.append(f"FEHLER: {negative_sizes} Einträge mit negativer Grösse")
return len([i for i in issues if i.startswith("FEHLER")]) == 0, issues
def validate_trade_sequence(
self,
trades: List[dict]
) -> Tuple[bool, Dict]:
"""
Validiert eine Sequenz von Trades auf Konsistenz.
"""
if not trades:
return True, {"status": "empty_sequence"}
issues = []
price_stats = []
size_stats = []
prev_ts = 0
for i, trade in enumerate(trades):
# Timestamp-Monotonie
ts = trade.get("timestamp", 0)
if ts < prev_ts:
issues.append(f"Trade {i}: Timestamp-Regression")
prev_ts = ts
# Preis-Validierung
price = float(trade.get("price", 0))
size = float(trade.get("size", 0))
if price <= 0:
issues.append(f"Trade {i}: Ungültiger Preis {price}")
else:
price_stats.append(price)
if size < 0:
issues.append(f"Trade {i}: Negative Size {size}")
else:
size_stats.append(size)
# Statistik-basierte Outlier-Detection
stats_result = {}
if price_stats:
mean_price = statistics.mean(price_stats)
stdev_price = statistics.stdev(price_stats) if len(price_stats) > 1 else 0
outlier_threshold = mean_price + 3 * stdev_price
outliers = [p for p in price_stats if abs(p - mean_price) > outlier_threshold]
if outliers:
issues.append(
f"WARNUNG: {len(outliers)} Preise ausserhalb 3σ (mean={mean_price:.4f})"
)
stats_result["price_mean"] = mean_price
stats_result["price_stdev"] = stdev_price
stats_result["outliers"] = len(outliers)
if size_stats:
stats_result["size_mean"] = statistics.mean(size_stats)
stats_result["size_max"] = max(size_stats)
# Liquidation-Size-Check
liquidations = [t for t in trades if t.get("isLiquidation")]
if liquidations:
liq_sizes = [float(t["size"]) for t in liquidations]
stats_result["liq_count"] = len(liquidations)
stats_result["liq_avg_size"] = statistics.mean(liq_sizes)
# Ungewöhnlich grosse Liquidationen
large_liqs = [s for s in liq_sizes if s > stats_result["size_mean"] * 10]
if large_liqs:
issues.append(
f"INFO: {len(large_liqs)} Liquidationen >10x Durchschnitt"
)
stats_result["total_trades"] = len(trades)
stats_result["issues"] = issues
return len([i for i in issues if i.startswith("FEHLER")]) == 0, stats_result
def generate_quality_report(
self,
orderbook_data: List[dict],
trade_data: List[dict]
) -> dict:
"""
Generiert vollständigen Qualitätsbericht.
"""
report = {
"summary": {
"total_snapshots": len(orderbook_data),
"total_trades": len(trade_data),
"overall_status": "PASS"
},
"snapshot_issues": [],
"trade_issues": []
}
# Validiere alle Snapshots
failed_snapshots = 0
for i, snapshot in enumerate(orderbook_data):
is_valid, issues = self.validate_orderbook_snapshot(snapshot)
if not is_valid:
failed_snapshots += 1
report["snapshot_issues"].append({
"index": i,
"timestamp": snapshot.get("timestamp"),
"issues": issues
})
report["summary"]["failed_snapshots"] = failed_snapshots
report["summary"]["snapshot_pass_rate"] = (
(len(orderbook_data) - failed_snapshots) / len(orderbook_data)
if orderbook_data else 1.0
)
# Validiere Trades
is_valid, trade_stats = self.validate_trade_sequence(trade_data)
report["trade_stats"] = trade_stats
if not is_valid:
report["summary"]["overall_status"] = "FAIL"
return report
Nutzung
if __name__ == "__main__":
validator = DataQualityValidator(tolerance_bps=100.0)
# Annahme: data enthält geladene Orderbuch-Snapshots und Trades
# report = validator.generate_quality_report(orderbooks, trades)
# print(json.dumps(report, indent=2))
print("DataQualityValidator bereit für Integration")
Häufige Fehler und Lösungen
Fehler 1: Rate-Limit-Erschöpfung bei grossen Replay-Jobs
Symptom: API-Aufrufe 返回 429 Too Many Requests nach ca. 100 Anfragen.
Lösung: Implementieren Sie exponentielles Backoff mit Batch-Request-Strategie:
# Rate-Limit Handler mit Exponential Backoff
import time
import requests
def safe_api_call(func, max_retries=5, base_delay=1.0):
"""Führt API-Call mit automatischer Retry-Logik aus."""
for attempt in range(max_retries):
try:
return func()
except requests.exceptions.HTTPError as e:
if e.response.status_code == 429:
# Rate-Limit: Exponentielles Backoff
retry_after = float(
e.response.headers.get("Retry-After", base_delay * 2 ** attempt)
)
print(f"Rate-Limit erreicht. Warte {retry_after:.1f}s...")
time.sleep(retry_after)
else:
raise
except requests.exceptions.Timeout:
if attempt < max_retries - 1:
delay = base_delay * 2 ** attempt
print(f"Timeout. Retry in {delay:.1f}s...")
time.sleep(delay)
else:
raise
raise RuntimeError(f"Max retries ({max_retries}) überschritten")
Fehler 2: Datenlücken bei Timestamps ausserhalb der Cache-Zeit
Symptom: Orderbuch-Snapshots 返回 leer oder mit fehlenden Preisstufen für ältere Timestamps.
Lösung: Validieren Sie den Timestamp-Bereich vor der Anfrage und füllen Sie Lücken mit Interpolation:
# Timestamp-Validierung und Interpolation
from datetime import datetime, timedelta
def ensure_valid_timestamp(ts: int) -> int:
"""Validiert und korrigiert Timestamps."""
now_ms = int(datetime.now().timestamp() * 1000)
max_age_ms = 30 * 24 * 3600 * 1000 # 30 Tage
if ts > now_ms:
print("WARNUNG: Timestamp in der Zukunft, korrigiere...")
return now_ms
if now_ms - ts > max_age_ms:
print(f"WARNUNG: Timestamp älter als 30 Tage: {ts}")
# Option: Nur teilweise laden oder interpolieren
return ts
def interpolate_orderbook(before: dict, after: dict, ratio: float) -> dict:
"""
Interpoliert Orderbuch zwischen zwei Snapshots.
Args:
before: Früherer Snapshot
after: Späterer Snapshot
ratio: Gewichtung (0.0 = 100% before, 1.0 = 100% after)
"""
interpolated = {"bids": [], "asks": []}
# Sammle alle Preise
all_bids = set(float(b["price"]) for b in before["bids"])
all_bids.update(float(b["price"]) for b in after["bids"])
all_asks = set(float(a["price"]) for a in before["asks"])
all_asks.update(float(a["price"]) for a in after["asks"])
# Interpoliere Bids
bid_dict = {float(b["price"]): float(b["size"]) for b in before["bids"]}
after_bid_dict = {float(b["price"]): float(b["size"]) for b in after["bids"]}
for price in sorted(all_bids, reverse=True)[:100]:
size_before = bid_dict.get(price, 0)
size_after = after_bid_dict.get(price, 0)
interpolated_size = size_before * (1 - ratio) + size_after * ratio
if interpolated_size > 0:
interpolated["bids"].append({
"price": str(price),
"size": str(interpolated_size)
})
# Analog für Asks
ask_dict = {float(a["price"]): float(a["size"]) for a in before["asks"]}
after_ask_dict = {float(a["price"]): float(a["size"]) for a in after["asks"]}
for price in sorted(all_asks)[:100]:
size_before = ask_dict.get(price, 0)
size_after = after_ask_dict.get(price, 0)
interpolated_size = size_before * (1 - ratio) + size_after * ratio
if interpolated_size > 0:
interpolated["asks"].append({
"price": str(price),
"size": str(interpolated_size)
})
return interpolated
Fehler 3: Falsche Liquidation-Flag-Interpretation
Symptom: Liquidation-Trades werden nicht korrekt identifiziert, was zu falschen Backtesting-Ergebnissen führt.
Lösung: Validieren Sie Liquidation-Events gegen Funding- und Price-Impact-Daten:
# Liquidation-Validierung
def validate_liquidation(trade: dict, context: dict) -> bool:
"""
Validiert, ob ein Trade tatsächlich eine Liquidation ist.
Checkpoints:
1. Trade-Grösse über Mindest-Schwelle
2. Preis ausserhalb normaler Spread-Parameter
3. Korrelation mit Funding-Event
"""
MIN_LIQ_SIZE = {
"HYPE-PERP": 100,
"BTC-PERP": 0.01,
"ETH-PERP": 0.1
}
market = context.get("market", "HYPE-PERP")
min_size = MIN_LIQ_SIZE.get(market, 10)
# 1. Size-Check
if float(trade.get("size", 0)) < min_size:
return False
# 2. Preis-Impact-Check
best_bid = context.get("best_bid", 0)
best_ask = context.get("best_ask", float('inf'))
price = float(trade.get("price", 0))
# Liquidation sollte ausserhalb des normalen Spreads liegen
max_normal_price = best_ask * 1.001
min_normal_price = best_bid * 0.999
if min_normal_price <= price <= max_normal_price:
# Preis innerhalb normalen Spreads -> wahrscheinlich keine Liquidation
return False
# 3. Side-Konsistenz
expected_side = "sell" if price > best_ask else "buy"
if trade.get("side") != expected_side:
return False
return True
def aggregate_liquidation_events(trades: list, window_ms: int = 1000) -> list:
"""
Aggregiert mehrere Liquidation-Events im selben Zeitfenster.
Dies kann auf eine.single Liquidation cascade hinweisen.
"""
if not trades:
return []
liq_trades = [t for t in trades if t.get("isLiquidation")]
aggregated = []
current_batch = []
batch_start = 0
for trade in liq_trades:
ts = trade.get("timestamp", 0)
if not current_batch:
batch_start = ts
if ts - batch_start <= window_ms:
current_batch.append(trade)
else:
if current_batch:
aggregated.append({
"type": "cascade",
"timestamp": batch_start,
"count": len(current_batch),
"total_size": sum(float(t["size"]) for t in current_batch),
"trades": current_batch
})
current_batch = [trade]
batch_start = ts
# Letzter Batch
if current_batch:
aggregated.append({
"type": "cascade" if len(current_batch) > 1 else "single",
"timestamp": batch_start,
"count": len(current_batch),
"total_size": sum(float(t["size"]) for t in current_batch),
"trades": current_batch
})
return aggregated
HolySheep AI: Konfiguration und Best Practices
Die HolySheep AI API bietet spezifische Features für Hyperliquid-Daten:
- Batch-Requests: Bis zu 100 Timestamps pro Request für effizientes Bulk-Loading
- Komprimierte Responses: gzip/brotli für reduzierte Bandbreite
- Webhook-Integration: Push-Benachrichtigungen bei neuen Liquidations
- Custom Endpoints: Unterstützung für Tardis-kompatible URL-Schemata