Abdoul Ahad Binizi
209 lines
8.2 KiB
Python
209 lines
8.2 KiB
Python
#!/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")
|