#!/usr/bin/env python3 """ liquidation_levels.py - Real-time Liquidation Level Estimator ============================================================== Estimates where liquidation clusters are based on current open interest, funding rates, and price levels. No historical data download needed. Uses Bybit API to estimate liquidation zones and provides: - Liquidation cluster levels (above and below current price) - Liquidation intensity score (-1 to +1) - Suggested TP/SL adjustments based on liquidation zones Integration: call get_liquidation_signal(exchange, symbol) from pro_trader.py """ import logging import numpy as np logger = logging.getLogger("AHAD QUANT") def estimate_liquidation_levels(exchange, symbol, current_price=None): """ Estimate liquidation cluster levels from open interest and leverage data. Uses the principle that most retail traders use 5x-25x leverage, so liquidation prices cluster at predictable distances from entry. Args: exchange: ccxt exchange instance symbol: e.g. "BTC/USDT:USDT" current_price: current market price (fetched if None) Returns: dict with: liq_above: list of (price, intensity) for liquidation levels above liq_below: list of (price, intensity) for liquidation levels below nearest_liq_above: nearest liquidation cluster above current price nearest_liq_below: nearest liquidation cluster below current price liq_bias: -1 to +1 (positive = more longs to liquidate below) """ try: if current_price is None: ticker = exchange.fetch_ticker(symbol) current_price = float(ticker["last"]) # Fetch open interest if available oi_long = 0 oi_short = 0 try: # Bybit long/short ratio coin = symbol.split("/")[0] import requests r = requests.get( f"https://api.bybit.com/v5/market/account-ratio", params={"category": "linear", "symbol": f"{coin}USDT", "period": "1h", "limit": 1}, timeout=5 ) if r.status_code == 200: data = r.json().get("result", {}).get("list", []) if data: buy_ratio = float(data[0].get("buyRatio", 0.5)) sell_ratio = float(data[0].get("sellRatio", 0.5)) oi_long = buy_ratio oi_short = sell_ratio except Exception: oi_long = 0.5 oi_short = 0.5 # Common leverage levels used by retail (5x, 10x, 20x, 25x, 50x) leverages = [5, 10, 20, 25, 50] # For LONG positions, liquidation = entry * (1 - 1/leverage) # For SHORT positions, liquidation = entry * (1 + 1/leverage) liq_below = [] # long liquidations (below current price) liq_above = [] # short liquidations (above current price) for lev in leverages: # Where longs opened near current price get liquidated liq_price_long = current_price * (1 - 0.9 / lev) # 90% of margin = liq distance_pct = (current_price - liq_price_long) / current_price * 100 # Intensity based on how common this leverage is intensity = _leverage_popularity(lev) * oi_long liq_below.append((round(liq_price_long, 6), round(intensity, 3), f"{lev}x")) # Where shorts opened near current price get liquidated liq_price_short = current_price * (1 + 0.9 / lev) intensity_short = _leverage_popularity(lev) * oi_short liq_above.append((round(liq_price_short, 6), round(intensity_short, 3), f"{lev}x")) # Sort by distance from current price liq_below.sort(key=lambda x: -x[0]) # closest first liq_above.sort(key=lambda x: x[0]) # closest first # Nearest clusters nearest_below = liq_below[0][0] if liq_below else current_price * 0.95 nearest_above = liq_above[0][0] if liq_above else current_price * 1.05 # Bias: positive = more longs to liquidate (bearish pressure) total_below = sum(x[1] for x in liq_below) total_above = sum(x[1] for x in liq_above) total = total_below + total_above liq_bias = (total_below - total_above) / total if total > 0 else 0.0 return { "liq_above": liq_above, "liq_below": liq_below, "nearest_liq_above": nearest_above, "nearest_liq_below": nearest_below, "liq_bias": round(liq_bias, 4), "long_ratio": round(oi_long, 4), "short_ratio": round(oi_short, 4), "current_price": current_price, } except Exception as e: logger.debug(f"[LIQ] Failed for {symbol}: {e}") return None def _leverage_popularity(leverage): """Estimate how popular each leverage level is among retail traders.""" # Based on Bybit/Binance data: most use 5-10x popularity = { 5: 0.30, # 30% of traders 10: 0.35, # 35% most popular 20: 0.20, # 20% 25: 0.10, # 10% 50: 0.05, # 5% degens } return popularity.get(leverage, 0.1) def get_liquidation_signal(exchange, symbol, side="LONG"): """ Get a liquidation-based trading signal. Args: exchange: ccxt instance symbol: trading pair side: "LONG" or "SHORT" - the side we want to trade Returns: dict with: score: -1 to +1 (positive = favorable for the given side) nearest_target: price level where liquidation cascade helps us nearest_danger: price level where liquidation cascade hurts us adjust_tp: suggested TP adjustment (closer to liq cluster) adjust_sl: suggested SL adjustment (away from liq cluster) """ levels = estimate_liquidation_levels(exchange, symbol) if not levels: return {"score": 0, "nearest_target": None, "nearest_danger": None} price = levels["current_price"] bias = levels["liq_bias"] if side == "LONG": # For LONG: we want short liquidations above (cascade up = good) # and we fear long liquidations below (cascade down = bad) score = -bias # negative bias = more shorts to squeeze = good for long nearest_target = levels["nearest_liq_above"] nearest_danger = levels["nearest_liq_below"] else: # For SHORT: we want long liquidations below (cascade down = good) # and we fear short liquidations above (cascade up = bad) score = bias # positive bias = more longs to liquidate = good for short nearest_target = levels["nearest_liq_below"] nearest_danger = levels["nearest_liq_above"] # TP adjustment: put TP just before the cascade target (take profit before bounce) target_dist = abs(nearest_target - price) adjust_tp = target_dist * 0.9 # 90% of distance to liq cluster # SL adjustment: put SL beyond the danger zone (don't get caught in cascade) danger_dist = abs(nearest_danger - price) adjust_sl = danger_dist * 0.5 # SL at 50% of distance to danger cluster return { "score": round(score, 4), "nearest_target": round(nearest_target, 6), "nearest_danger": round(nearest_danger, 6), "adjust_tp": round(adjust_tp, 6), "adjust_sl": round(adjust_sl, 6), "long_ratio": levels["long_ratio"], "short_ratio": levels["short_ratio"], } if __name__ == "__main__": print("[LIQ] Liquidation Levels - smoke test") import ccxt ex = ccxt.bybit() for coin in ["BTC", "ETH", "SOL"]: sym = f"{coin}/USDT:USDT" levels = estimate_liquidation_levels(ex, sym) if levels: print(f"\n {coin}: price=${levels['current_price']}") print(f" Long/Short ratio: {levels['long_ratio']}/{levels['short_ratio']}") print(f" Liq bias: {levels['liq_bias']} ({'bearish' if levels['liq_bias'] > 0 else 'bullish'})") print(f" Nearest liq below: ${levels['nearest_liq_below']:.2f}") print(f" Nearest liq above: ${levels['nearest_liq_above']:.2f}") sig = get_liquidation_signal(ex, sym, "SHORT") print(f" SHORT signal: score={sig['score']}, target=${sig['nearest_target']:.2f}") print("\n[LIQ] Smoke test passed")