ftdt-quant-lab/live/monitors/cross_venue.py

"""
Cross-venue lead-lag monitor.

Detects when one exchange leads another in price discovery.
Used for informational purposes only in Phase 4 — no auto-trading.
"""

from __future__ import annotations

from collections import deque
from typing import Optional

import numpy as np


class CrossVenueMonitor:
    """Monitor price lead-lag relationships between exchanges.

    Tracks mid prices across venues and computes cross-correlation
    and lead-lag structure. Can detect when Hyperliquid follows
    Binance or vice versa.

    Usage:
        monitor = CrossVenueMonitor(pairs=[("hl", "binance")], window=100)
        monitor.update("hl", "BTC", 50000.0)
        monitor.update("binance", "BTC", 50000.5)
        result = monitor.lead_lag("BTC")
    """

    def __init__(
        self,
        pairs: list[tuple[str, str]] | None = None,
        window: int = 100,
        max_lag: int = 10,
    ):
        self._window = window
        self._max_lag = max_lag
        self._pairs = pairs or [("hl", "binance"), ("hl", "bybit"), ("hl", "okx")]

        # venue → coin → deque of mid prices
        self._prices: dict[str, dict[str, deque[float]]] = {}
        self._timestamps: dict[str, dict[str, deque[float]]] = {}

    def update(self, venue: str, coin: str, price: float, timestamp: float):
        """Record a mid price observation from a venue."""
        v = venue.lower()
        c = coin.upper()
        self._prices.setdefault(v, {}).setdefault(c, deque(maxlen=self._window)).append(price)
        self._timestamps.setdefault(v, {}).setdefault(c, deque(maxlen=self._window)).append(timestamp)

    def lead_lag(self, coin: str, venue_a: str = "hl", venue_b: str = "binance") -> dict | None:
        """Determine which venue leads by cross-correlation at various lags.

        Returns {leading_venue: str, max_correlation: float, lag: int}
        Negative lag = venue_a leads, positive lag = venue_b leads.
        """
        prices_a = list(self._prices.get(venue_a.lower(), {}).get(coin.upper(), []))
        prices_b = list(self._prices.get(venue_b.lower(), {}).get(coin.upper(), []))

        min_len = min(len(prices_a), len(prices_b))
        if min_len < self._max_lag + 2:
            return None

        a = np.array(prices_a[-min_len:])
        b = np.array(prices_b[-min_len:])

        best_corr = -1.0
        best_lag = 0

        for lag in range(-self._max_lag, self._max_lag + 1):
            if lag < 0:
                corr = np.corrcoef(a[-lag:], b[:lag])[0, 1] if lag < 0 else 0
            elif lag > 0:
                corr = np.corrcoef(a[:min_len - lag], b[lag:])[0, 1]
            else:
                corr = np.corrcoef(a, b)[0, 1]

            if not np.isnan(corr) and abs(corr) > abs(best_corr):
                best_corr = float(corr)
                best_lag = lag

        return {
            "leading_venue": venue_a if best_lag < 0 else venue_b if best_lag > 0 else "none",
            "correlation": round(best_corr, 4),
            "lag": best_lag,
            "samples": min_len,
        }

    def spot_premium(self, coin: str, venue: str = "hl", spot_venue: str = "binance") -> dict | None:
        """Compute the premium of venue over spot (basis proxy)."""
        v = self._prices.get(venue.lower(), {}).get(coin.upper())
        sv = self._prices.get(spot_venue.lower(), {}).get(coin.upper())
        if not v or not sv:
            return None

        perp = v[-1]
        spot = sv[-1]
        basis_bps = (perp - spot) / spot * 10000 if spot > 0 else 0

        return {
            "venue": venue,
            "spot_venue": spot_venue,
            "perp_price": perp,
            "spot_price": spot,
            "basis_bps": round(basis_bps, 2),
        }

    def summary(self, coin: str) -> dict:
        """Summary for a given coin across all venues."""
        result = {}
        for va, vb in self._pairs:
            ll = self.lead_lag(coin, va, vb)
            if ll:
                result[f"{va}_{vb}"] = ll

        premium = self.spot_premium(coin)
        if premium:
            result["premium"] = premium

        return result