""" Tick-size and lot-size regime exploitation. Exchanges dynamically adjust minimum price increments (tick size) and minimum order sizes (lot size) based on asset price. When an asset crosses these thresholds, spreads widen or narrow instantly. Hyperliquid tick sizes (example — actual values from exchange): BTC: 0.1 USD tick (<$100K), 0.5 USD tick ($100K-$1M) — approximate ETH: 0.01 USD tick SOL: 0.001 USD tick Strategy: when price crosses a tick-size boundary, be the first to adjust your quoting engine. Competitors quoting at old tick sizes will be either non-competitive (too wide) or illegal (too tight). Capture the spread gap. """ from __future__ import annotations from collections import deque from typing import Optional # Hyperliquid tick size schedule (example values — confirm from exchange docs) TICK_SCHEDULE = { "BTC": [ (float("-inf"), 100000, 0.1), (100000, float("inf"), 0.5), ], "ETH": [ (float("-inf"), 10000, 0.01), (10000, float("inf"), 0.05), ], "SOL": [ (float("-inf"), 1000, 0.001), (1000, float("inf"), 0.005), ], # Default for unknown coins "DEFAULT": [ (float("-inf"), float("inf"), 0.01), ], } class TickRegimeMonitor: """Monitor and exploit tick-size regime changes. Tracks when an asset price approaches a tick-size boundary and generates signals to adjust quoting before competitors. """ def __init__( self, approach_threshold_pct: float = 1.0, # 1% from boundary history_window: int = 100, ): self._approach_threshold = approach_threshold_pct / 100 self._prices: dict[str, deque[float]] = {} self._current_tick: dict[str, float] = {} self._approaching_boundary: dict[str, dict] = {} self._window = history_window # ── Data feed ──────────────────────────────────────────── def get_tick_size(self, coin: str, price: float) -> float: """Get current tick size for a coin at given price.""" schedule = TICK_SCHEDULE.get(coin.upper(), TICK_SCHEDULE["DEFAULT"]) for lo, hi, tick in schedule: if lo <= price < hi: return tick return schedule[-1][2] # fallback def update_price(self, coin: str, price: float): """Feed a new price observation.""" c = coin.upper() self._prices.setdefault(c, deque(maxlen=self._window)).append(price) self._current_tick[c] = self.get_tick_size(c, price) # Check if approaching boundary schedule = TICK_SCHEDULE.get(c, TICK_SCHEDULE["DEFAULT"]) for lo, hi, tick in schedule: if lo <= price < hi: # Check approach to upper boundary if hi != float("inf"): dist_up = (hi - price) / price if 0 < dist_up < self._approach_threshold: next_tick = self.get_tick_size(c, hi) if next_tick != tick: self._approaching_boundary[c] = { "direction": "up", "current_tick": tick, "next_tick": next_tick, "boundary_price": hi, "distance_pct": round(dist_up * 100, 2), "crosses_in_bars": self._estimate_cross_bars(c, hi), } return # Check approach to lower boundary if lo != float("-inf"): dist_down = (price - lo) / price if 0 < dist_down < self._approach_threshold: prev_tick = self.get_tick_size(c, lo - 1) if prev_tick != tick: self._approaching_boundary[c] = { "direction": "down", "current_tick": tick, "next_tick": prev_tick, "boundary_price": lo, "distance_pct": round(dist_down * 100, 2), "crosses_in_bars": self._estimate_cross_bars(c, lo), } return self._approaching_boundary.pop(c, None) def _estimate_cross_bars(self, coin: str, boundary_price: float) -> int: """Estimate how many bars until price crosses the boundary.""" prices = list(self._prices.get(coin, [])) if len(prices) < 5: return -1 recent = prices[-min(20, len(prices)):] if len(recent) < 5: return -1 # Simple linear trend estimate xs = list(range(len(recent))) n = len(xs) sum_x = sum(xs) sum_y = sum(recent) sum_xy = sum(x * y for x, y in zip(xs, recent)) sum_x2 = sum(x * x for x in xs) slope = (n * sum_xy - sum_x * sum_y) / (n * sum_x2 - sum_x * sum_x) if (n * sum_x2 - sum_x * sum_x) != 0 else 0 last_price = recent[-1] if abs(slope) < 1e-10: return -1 bars_to_cross = int((boundary_price - last_price) / slope) return max(0, bars_to_cross) if bars_to_cross > 0 else -1 # ── Signal ────────────────────────────────────────────── def signal(self, coin: str) -> dict: """Generate tick-size regime signal. Returns action to take before competitors adjust. """ c = coin.upper() boundary = self._approaching_boundary.get(c) current_tick = self._current_tick.get(c, 0.01) if not boundary: return { "action": "none", "current_tick": current_tick, "reason": "no_regime_change", } tick_delta = boundary["next_tick"] - boundary["current_tick"] bars = boundary["crosses_in_bars"] if tick_delta > 0: # Tick size INCREASING → spread will widen → HFTs will post wider quotes # Be the first to widen to the new tick action = "widen_quotes" reason = f"Tick increasing {boundary['current_tick']}→{boundary['next_tick']} " reason += f"({boundary['distance_pct']}% away" + (f", ~{bars} bars" if bars > 0 else "") + ")" else: # Tick size DECREASING → spread will narrow → HFTs will tighten # Be the first to tighten to the new tick action = "tighten_quotes" reason = f"Tick decreasing {boundary['current_tick']}→{boundary['next_tick']} " reason += f"({boundary['distance_pct']}% away" + (f", ~{bars} bars" if bars > 0 else "") + ")" return { "action": action, "current_tick": boundary["current_tick"], "next_tick": boundary["next_tick"], "tick_delta": round(tick_delta, 5), "boundary_price": boundary["boundary_price"], "distance_pct": boundary["distance_pct"], "estimated_bars_to_cross": bars, "reason": reason, "urgent": boundary["distance_pct"] < 0.5, # within 0.5% = urgent } def summary(self) -> dict: return {coin: self.signal(coin) for coin in self._prices}