""" Cross-exchange triangular latency arbitrage detector. Latency arb isn't just A vs B. It's A → B → C across three venues. If BTC/USDT is slow to update on Venue 1, but Venue 2 is fast on ETH/BTC and Venue 3 is fast on ETH/USDT, the slow leg creates a triangular arbitrage opportunity. Path: Buy BTC/USDT on slow venue → Sell ETH/BTC on fast venue → Sell ETH/USDT on fast venue Strategy: monitor 3-venue latency simultaneously, detect when one leg lags, compute implied arbitrage spread, and signal execution-ready opportunities. """ from __future__ import annotations from collections import deque from typing import Optional class TriangularLatencyArb: """Detect triangular arbitrage from cross-venue latency discrepancies. Monitors up to 3 venues with configurable latency estimates. When one venue lags on a specific pair, the triangle becomes profitable. Usage: arb = TriangularLatencyArb() arb.update_price("hl", "BTC-USDT", 64500, latency_ms=5) arb.update_price("hl", "ETH-BTC", 0.049, latency_ms=8) arb.update_price("hl", "ETH-USDT", 3160, latency_ms=6) opportunities = arb.detect() """ def __init__( self, min_spread_bps: float = 0.5, # minimum bps profit to signal max_latency_diff_ms: float = 200.0, # max venue latency gap to consider window: int = 50, ): self._min_spread = min_spread_bps self._max_latency = max_latency_diff_ms self._window = window # venue → pair → (price, latency_ms, timestamp) self._prices: dict[str, dict[str, deque[tuple[float, float, float]]]] = {} self._latencies: dict[str, float] = {} # venue → avg latency # ── Data feed ──────────────────────────────────────────── def update_price(self, venue: str, pair: str, price: float, latency_ms: float = 0): """Record a price snapshot from a venue with its latency.""" v = venue.lower() p = pair.upper().replace("/", "-") self._prices.setdefault(v, {}).setdefault( p, deque(maxlen=self._window) ).append((price, latency_ms, latency_ms)) # (price, latency, timestamp_simple) # Update venue latency estimate (EMA) old_lat = self._latencies.get(v, latency_ms) self._latencies[v] = old_lat * 0.9 + latency_ms * 0.1 def get_price(self, venue: str, pair: str) -> float | None: """Get latest price from a venue.""" dq = self._prices.get(venue.lower(), {}).get(pair.upper().replace("/", "-")) return dq[-1][0] if dq else None def get_latency(self, venue: str) -> float: return self._latencies.get(venue.lower(), 0) # ── Arbitrage detection ────────────────────────────────── def detect(self) -> dict: """Detect triangular arbitrage opportunities across venues. Triangle: BTC-USDT → ETH-BTC → ETH-USDT If any leg is on a slower venue, implied profit exists. Returns list of opportunities with profit, confidence, and execution plan. """ opportunities = [] pairs = ["BTC-USDT", "ETH-BTC", "ETH-USDT"] venues = list(self._prices.keys()) if len(venues) < 1: return {"opportunities": [], "venue_count": len(venues)} # Find latency gaps between venues if len(venues) >= 2: lat_gaps = self._latency_gaps() else: lat_gaps = {} # For each venue, compute implied cross-rate vs direct for v in venues: btc_usdt = self.get_price(v, "BTC-USDT") eth_btc = self.get_price(v, "ETH-BTC") eth_usdt = self.get_price(v, "ETH-USDT") if not (btc_usdt and eth_btc and eth_usdt): continue # Implied ETH-USDT from triangle: BTC-USDT × ETH-BTC implied_eth = btc_usdt * eth_btc implied_bps = (eth_usdt - implied_eth) / implied_eth * 10000 if abs(implied_bps) > self._min_spread: opportunities.append({ "venue": v, "type": "internal_triangular", "btc_usdt": btc_usdt, "eth_btc": eth_btc, "eth_usdt": eth_usdt, "implied_eth_usdt": round(implied_eth, 2), "spread_bps": round(implied_bps, 2), "direction": "sell_eth" if implied_bps > 0 else "buy_eth", "latency_ms": round(self._latencies.get(v, 0), 2), }) # Cross-venue: check if one venue's slow leg creates arb if len(venues) >= 2: for i, v1 in enumerate(venues): for v2 in venues[i + 1:]: opp = self._cross_venue_arb(v1, v2, pairs) if opp: opportunities.append(opp) return { "opportunities": opportunities[:10], "venue_count": len(venues), "latency_gaps": lat_gaps, "best_opportunity": max(opportunities, key=lambda o: abs(o["spread_bps"])) if opportunities else None, } def _cross_venue_arb(self, v1: str, v2: str, pairs: list[str]) -> dict | None: """Check if buying on slow venue, selling on fast venue is profitable.""" best_opp = None best_bps = 0 for pair in pairs: p1 = self.get_price(v1, pair) p2 = self.get_price(v2, pair) if not p1 or not p2: continue spread_bps = abs(p2 - p1) / p1 * 10000 lat_diff = abs(self.get_latency(v1) - self.get_latency(v2)) if spread_bps > self._min_spread and lat_diff > 10: opp = { "type": "cross_venue", "pair": pair, "slow_venue": v1 if self.get_latency(v1) > self.get_latency(v2) else v2, "fast_venue": v1 if self.get_latency(v1) < self.get_latency(v2) else v2, "buy_at": round(min(p1, p2), 2), "sell_at": round(max(p1, p2), 2), "spread_bps": round(spread_bps, 2), "latency_diff_ms": round(lat_diff, 2), } if spread_bps > best_bps: best_bps = spread_bps best_opp = opp return best_opp def _latency_gaps(self) -> dict: """Compute latency gaps between all venue pairs.""" venues = list(self._latencies.keys()) gaps = {} for i, v1 in enumerate(venues): for v2 in venues[i + 1:]: diff = abs(self._latencies.get(v1, 0) - self._latencies.get(v2, 0)) key = f"{v1}_{v2}" gaps[key] = round(diff, 2) return gaps def signal(self) -> dict: """Generate trading signal for cross-venue triangular arb.""" result = self.detect() opps = result.get("opportunities", []) if not opps: return {"action": "none", "reason": "no_opportunity"} best = opps[0] if best["type"] == "cross_venue" and abs(best["spread_bps"]) > self._min_spread * 2: return { "action": "arbitrage", "type": "cross_venue", "details": best, "confidence": min(1.0, abs(best["spread_bps"]) / (self._min_spread * 5)), } elif best["type"] == "internal_triangular" and abs(best["spread_bps"]) > self._min_spread * 3: return { "action": "arbitrage", "type": "triangular", "details": best, "confidence": min(1.0, abs(best["spread_bps"]) / (self._min_spread * 5)), } return {"action": "monitor", "reason": "spread_too_small", "best_bps": round(best["spread_bps"], 2)} def summary(self) -> dict: opps = self.detect() return { "venues": list(self._prices.keys()), "latencies": self._latencies, "opportunities_count": len(opps.get("opportunities", [])), "best_opportunity": opps.get("best_opportunity"), "signal": self.signal(), }