Donde el «Smart Money» liquida al retail: Guía práctica para operar con mapas de calor
El mapa de calor de liquidaciones (Liquidation Heatmap) es básicamente una radiografía de las esperanzas ajenas. Mientras el trader retail pone su stop loss usando análisis técnico clásico por encima o por debajo del último máximo o mínimo local, la ballena o institucional ve en el mapa esa acumulación de puntos como la gasolina perfecta para mover el precio a su favor.
Un market maker no tiene forma física de armar o cerrar una posición de millones de dólares en un mercado sin liquidez. Necesita volumen. Y esa liquidez se la regalan los margin calls de la masa.
La trampa para los shorts: Anatomía de un Short Squeeze
Imagínate este escenario: el activo lleva tres días cayendo en picada. El retail ve debilidad, mete shorts apalancados a 20x–50x y clava sus stop losses (o deja su margen al límite de la liquidación) justo detrás de la resistencia local más cercana, digamos en un nivel redondo o en la parte alta de un rango de consolidación.
¿Qué hace el pez gordo cuando necesita cargar un long gigante o liquidar sus inventarios?
- Compresión del precio: Empuja el precio despacio hacia arriba, sin levantar alarmas entre los shorteros, pero acercándolo peligrosamente a la zona de liquidación en el mapa.
- Gatillazo/Pંકzo impulsivo: Con un movimiento seco y violento, disparan el precio entre un 1.5% y un 2% por encima de la resistencia.
- Reacción en cadena: Ahí saltan los primeros stops y las posiciones súper apalancadas se van a cero. El exchange ejecuta compras a mercado automáticamente para liquidar esas posiciones en short.
- Explosión de volumen: Esas compras a mercado generadas por los margin calls chocan de frente con las órdenes limit de venta que la ballena ya tenía colocadas con anticipación, absorbiendo toda esa demanda desesperada.
El trader retail piensa: «¡Ruptura! ¡Cambio de tendencia, me subo al long!». Compra justo en el pico de FOMO del impulso, momento en el cual la ballena se da la vuelta, vende todo y el precio cae como una piedra, yendo ahora por los stops de los longs.
Por qué se «rompe» el order book
Cuando salta la liquidación de una posición grande de $500,000 apalancada a 20x, el exchange mete una orden a mercado para cerrarla sí o sí. Si en el order book no hay suficientes órdenes limit en el sentido contrario a esos precios, el precio atraviesa los niveles sin frenos.
- Absorción de órdenes limit: El motor de liquidación empieza a barrer el libro. Si a nivel de $60,000 solo había 5 Bitcoin colgados en el libro y hay que liquidar 20 Bitcoin, el precio barre ese nivel y sigue de largo hasta encontrar la liquidez necesaria.
- Efecto dominó: Cada nivel barrido activa nuevos stop losses y liquidaciones de traders que venían más tranquilos con menos apalancamiento (por ejemplo, los de 10x).
- Rebote de gato muerto (Dead Cat Bounce): En cuanto la cascada quema toda la liquidez del movimiento, el order book queda "vacío". Es ahí cuando el market maker, que estaba esperando que se agotara el movimiento, entra a un precio privilegiado con bloques gigantes de órdenes limit, haciendo girar el precio de golpe.
Setups operativos: Cómo sacarle jugo a las zonas de liquidación
Ni se te ocurra intentar operar a ciegas en el segundo exacto de la liquidación —el spread y el slippage te van a masacrar. La clave está en operar la reacción del mercado una vez que el incendio ya se apagó.
Setup #1: Caza de liquidez (Stop Hunting)
- Condiciones: En el mapa de calor ves un bloque denso de shorts o longs a menos de un 1% de distancia del precio actual.
- Qué hacer: Esperar el mechazo/barrido del nivel. Jamás entres en el breakout inicial.
- Ejecución: El precio se dispara violentamente pasando el nivel, el volumen en la vela de 1 minuto se va al cielo y el Open Interest (OI) cae en picado (limpiaron a todo el mundo). En cuanto la vela cierra dejando una mecha larga y regresa por debajo/encima del nivel, metes la operación en dirección contraria.
- Stop Loss: Lo pones de forma estricta justo por encima/debajo del extremo de esa mecha impulsiva que barrió las liquidaciones. Si el precio vuelve a romper ese máximo/mínimo, significa que no fue un falso breakout sino un rompimiento real. Tu riesgo tiene que estar ultra controlado.
Automatizando el rastreo: Script para detectar anomalías
Para que no tengas que estar pegado a la pantalla 24/7 esperando a que ocurra una cascada de liquidaciones, puedes usar este script. Rastrea el flow de liquidaciones del exchange en tiempo real e imprime en consola los volúmenes anómalos de cierres forzados.
Está desarrollado en Python puro sin dependencias raras, es súper estable y se recupera solo si se cae la conexión con el exchange.
import json
import time
import threading
import logging
from collections import deque, OrderedDict
from statistics import median
from typing import Dict, Tuple
import requests
import websocket
from requests.adapters import HTTPAdapter
from urllib3.util.retry import Retry
class ProductionMarketEngine:
"""
Production-grade Binance Futures liquidation / market-state engine.
Data sources:
- Binance Futures forceOrder WebSocket
- Binance Futures ticker WebSocket
- Binance Futures Open Interest REST
Main analysis window:
5 minutes
Features:
- Rolling liquidation window
- Calendar-aligned 5-minute liquidation buckets
- TTL event deduplication
- Real market price feed
- Exact 5-minute OI delta
- Liquidation imbalance
- Liquidation intensity
- Price momentum
- Volatility
- Market-state classification
- Signal scoring
- WebSocket reconnect with exponential backoff
- HTTP retries for Open Interest
- Graceful shutdown
- Thread-safe state
- Runtime health monitoring
"""
def __init__(
self,
symbol: str = "btcusdt",
window_seconds: int = 300,
oi_poll_seconds: int = 10,
baseline_buckets: int = 288,
min_baseline_buckets: int = 12,
):
self.symbol = symbol.lower()
self.symbol_upper = symbol.upper()
self.window_seconds = window_seconds
self.oi_poll_seconds = oi_poll_seconds
# 288 × 5 minutes = 24 hours.
self.baseline_buckets = baseline_buckets
# Minimum completed buckets before baseline becomes reliable.
self.min_baseline_buckets = min_baseline_buckets
# --------------------------------------------------------------
# Thread synchronization
# --------------------------------------------------------------
self.lock = threading.RLock()
# Event used for interruptible thread shutdown / waiting.
self.stop_event = threading.Event()
# --------------------------------------------------------------
# Runtime state
# --------------------------------------------------------------
self.is_running = False
self.ws_combined = None
self.ws_thread = None
self.oi_thread = None
self.maintenance_thread = None
# --------------------------------------------------------------
# Liquidation rolling window
# --------------------------------------------------------------
self.events = deque()
# event_id -> received timestamp
self.seen_ids = OrderedDict()
self.dedup_ttl_seconds = max(
window_seconds * 2,
600,
)
# --------------------------------------------------------------
# Real market price history
# --------------------------------------------------------------
self.price_history = deque()
self.current_price = 0.0
self.last_price_timestamp = 0.0
# --------------------------------------------------------------
# Open Interest
# --------------------------------------------------------------
self.oi_history = deque(
maxlen=180
)
self.current_oi = 0.0
self.last_oi_timestamp = 0.0
# --------------------------------------------------------------
# Calendar-aligned liquidation buckets
# --------------------------------------------------------------
# {
# bucket_timestamp: liquidation_notional
# }
self.bucket_accum = OrderedDict()
# --------------------------------------------------------------
# WebSocket state
# --------------------------------------------------------------
self.ws_connected = False
self.last_ws_message = 0.0
self.ws_reconnects = 0
# --------------------------------------------------------------
# Diagnostics
# --------------------------------------------------------------
self.total_liquidation_events = 0
self.invalid_liquidation_events = 0
# --------------------------------------------------------------
# Logger
# --------------------------------------------------------------
self.logger = logging.getLogger(
f"ProductionMarketEngine.{self.symbol_upper}"
)
# ==================================================================
# TIME / BUCKET HELPERS
# ==================================================================
@staticmethod
def _bucket_timestamp(timestamp: float) -> int:
"""
Convert timestamp to calendar-aligned 5-minute bucket.
"""
return int(timestamp // 300) * 300
# ==================================================================
# CLEANUP
# ==================================================================
def _clean_ttl_cache(self, now: float) -> None:
"""
Remove expired liquidation event IDs.
"""
cutoff = now - self.dedup_ttl_seconds
while self.seen_ids:
first_key, first_timestamp = next(
iter(self.seen_ids.items())
)
if first_timestamp < cutoff:
self.seen_ids.popitem(
last=False
)
else:
break
def _clean_old_events(self, now: float) -> None:
"""
Keep liquidation and price data only inside
the rolling analysis window.
"""
cutoff = now - self.window_seconds
while self.events:
if self.events[0]["timestamp"] < cutoff:
self.events.popleft()
else:
break
while self.price_history:
if self.price_history[0]["timestamp"] < cutoff:
self.price_history.popleft()
else:
break
def _clean_old_buckets(self, now: float) -> None:
"""
Remove old liquidation buckets.
This method is called from the dedicated maintenance thread,
so bucket cleanup does NOT depend on new liquidation events.
"""
current_bucket = self._bucket_timestamp(now)
minimum_bucket = (
current_bucket
- (self.baseline_buckets + 2) * 300
)
while self.bucket_accum:
first_bucket = next(
iter(self.bucket_accum)
)
if first_bucket < minimum_bucket:
self.bucket_accum.popitem(
last=False
)
else:
break
# ==================================================================
# LIQUIDATION INGESTION
# ==================================================================
def _process_liquidation(
self,
order_data: dict,
) -> None:
"""
Process Binance Futures forceOrder event.
"""
if not isinstance(
order_data,
dict,
):
return
now = time.time()
raw_side = str(
order_data.get(
"S",
"",
)
).upper()
# --------------------------------------------------------------
# Validate side before touching deduplication cache.
# --------------------------------------------------------------
if raw_side == "SELL":
side = "LONG_LIQ"
elif raw_side == "BUY":
side = "SHORT_LIQ"
else:
with self.lock:
self.invalid_liquidation_events += 1
return
try:
execution_timestamp_ms = int(
order_data.get(
"T",
int(now * 1000),
)
)
event_timestamp = (
execution_timestamp_ms
/ 1000.0
)
price = float(
order_data.get(
"ap",
order_data.get(
"p",
0,
),
)
)
qty = float(
order_data.get(
"q",
0,
)
)
order_id = str(
order_data.get(
"i",
"",
)
)
except (
TypeError,
ValueError,
):
with self.lock:
self.invalid_liquidation_events += 1
self.logger.warning(
"Invalid liquidation event received."
)
return
if price <= 0 or qty <= 0:
with self.lock:
self.invalid_liquidation_events += 1
return
notional = price * qty
if notional <= 0:
return
# --------------------------------------------------------------
# Event ID
#
# Prefer Binance order ID when available.
# If unavailable, use deterministic composite fallback.
# --------------------------------------------------------------
if order_id:
event_id = (
f"{execution_timestamp_ms}_"
f"{order_id}"
)
else:
event_id = (
f"{execution_timestamp_ms}_"
f"{side}_"
f"{price:.12f}_"
f"{qty:.12f}"
)
event = {
"id": event_id,
"timestamp": event_timestamp,
"side": side,
"price": price,
"qty": qty,
"notional": notional,
}
with self.lock:
self._clean_ttl_cache(
now
)
if event_id in self.seen_ids:
return
self.seen_ids[event_id] = now
self.events.append(
event
)
# Calendar-aligned bucket.
bucket = self._bucket_timestamp(
event_timestamp
)
if bucket not in self.bucket_accum:
self.bucket_accum[bucket] = 0.0
self.bucket_accum[bucket] += (
notional
)
self.total_liquidation_events += 1
# ==================================================================
# MARKET PRICE INGESTION
# ==================================================================
def _process_ticker(
self,
ticker_data: dict,
) -> None:
"""
Process Binance ticker event.
"""
try:
price = float(
ticker_data.get(
"c",
0,
)
)
except (
TypeError,
ValueError,
):
return
if price <= 0:
return
now = time.time()
with self.lock:
self.current_price = price
self.last_price_timestamp = now
self.price_history.append(
{
"timestamp": now,
"price": price,
}
)
# ==================================================================
# OPEN INTEREST HTTP SESSION
# ==================================================================
@staticmethod
def _create_http_session() -> requests.Session:
"""
Create requests Session with connection pooling
and retry policy.
"""
session = requests.Session()
retry = Retry(
total=4,
connect=4,
read=4,
status=4,
backoff_factor=0.5,
status_forcelist=(
429,
500,
502,
503,
504,
),
allowed_methods=frozenset(
[
"GET",
]
),
respect_retry_after_header=True,
)
adapter = HTTPAdapter(
max_retries=retry,
pool_connections=4,
pool_maxsize=4,
)
session.mount(
"https://",
adapter,
)
session.headers.update(
{
"User-Agent":
"ProductionMarketEngine/1.0"
}
)
return session
# ==================================================================
# OPEN INTEREST POLLER
# ==================================================================
def _poll_open_interest(
self,
) -> None:
"""
Poll Binance Futures Open Interest.
"""
url = (
"https://fapi.binance.com"
"/fapi/v1/openInterest"
)
session = (
self._create_http_session()
)
try:
while not self.stop_event.is_set():
try:
response = session.get(
url,
params={
"symbol":
self.symbol_upper
},
timeout=5,
)
response.raise_for_status()
data = response.json()
oi = float(
data.get(
"openInterest",
0,
)
)
if oi <= 0:
raise ValueError(
"Invalid Open Interest."
)
now = time.time()
with self.lock:
self.current_oi = oi
self.last_oi_timestamp = now
self.oi_history.append(
{
"timestamp": now,
"oi": oi,
}
)
except requests.RequestException as exc:
self.logger.warning(
"Open Interest request failed: %s",
exc,
)
except (
ValueError,
TypeError,
) as exc:
self.logger.warning(
"Invalid Open Interest response: %s",
exc,
)
except Exception:
self.logger.exception(
"Unexpected Open Interest error."
)
# Interruptible wait.
self.stop_event.wait(
self.oi_poll_seconds
)
finally:
session.close()
# ==================================================================
# MAINTENANCE THREAD
# ==================================================================
def _maintenance_loop(
self,
) -> None:
"""
Periodic memory/state cleanup.
Important:
cleanup is independent of liquidation activity.
"""
while not self.stop_event.wait(10):
now = time.time()
with self.lock:
self._clean_ttl_cache(
now
)
self._clean_old_events(
now
)
self._clean_old_buckets(
now
)
# ==================================================================
# OI 5-MINUTE DELTA
# ==================================================================
def _get_oi_change_5m(
self,
now: float,
) -> Tuple[float, bool]:
"""
Calculate actual 5-minute Open Interest change.
Returns:
change_pct
data_ready
"""
if len(self.oi_history) < 2:
return 0.0, False
cutoff = (
now
- self.window_seconds
)
baseline_sample = None
for sample in self.oi_history:
if sample["timestamp"] <= cutoff:
baseline_sample = sample
else:
break
# Not enough historical data yet.
if baseline_sample is None:
oldest = self.oi_history[0]
if (
now
- oldest["timestamp"]
< self.window_seconds * 0.8
):
return 0.0, False
baseline_sample = oldest
latest_sample = (
self.oi_history[-1]
)
first_oi = baseline_sample["oi"]
last_oi = latest_sample["oi"]
if first_oi <= 0:
return 0.0, False
change_pct = (
(
last_oi
- first_oi
)
/ first_oi
) * 100.0
return change_pct, True
# ==================================================================
# PRICE METRICS
# ==================================================================
def _get_price_metrics(
self,
) -> Tuple[
float,
float,
float,
bool,
]:
"""
Returns:
current price
5m price change
5m high-low volatility
ready
"""
if len(
self.price_history
) < 2:
return (
0.0,
0.0,
0.0,
False,
)
first_price = (
self.price_history[0]["price"]
)
last_price = (
self.price_history[-1]["price"]
)
if first_price <= 0:
return (
0.0,
0.0,
0.0,
False,
)
price_change_pct = (
(
last_price
- first_price
)
/ first_price
) * 100.0
prices = [
item["price"]
for item in self.price_history
]
high_price = max(prices)
low_price = min(prices)
if low_price > 0:
volatility_pct = (
(
high_price
- low_price
)
/ low_price
) * 100.0
else:
volatility_pct = 0.0
return (
last_price,
price_change_pct,
volatility_pct,
True,
)
# ==================================================================
# LIQUIDATION METRICS
# ==================================================================
def _get_liquidation_metrics(
self,
) -> Tuple[
float,
float,
float,
float,
]:
"""
Returns:
long liquidation
short liquidation
total liquidation
imbalance
"""
long_liq = 0.0
short_liq = 0.0
for event in self.events:
if event["side"] == "LONG_LIQ":
long_liq += event["notional"]
elif event["side"] == "SHORT_LIQ":
short_liq += event["notional"]
total_liq = (
long_liq
+ short_liq
)
if total_liq > 0:
imbalance = (
(
long_liq
- short_liq
)
/ total_liq
)
else:
imbalance = 0.0
return (
long_liq,
short_liq,
total_liq,
imbalance,
)
# ==================================================================
# BASELINE
# ==================================================================
def _get_baseline(
self,
) -> Tuple[
float,
int,
bool,
]:
"""
Calculate median liquidation volume
from completed calendar-aligned 5m buckets.
Current incomplete bucket is excluded.
"""
now = time.time()
current_bucket = (
self._bucket_timestamp(
now
)
)
completed = []
for (
bucket_timestamp,
volume,
) in self.bucket_accum.items():
if (
bucket_timestamp
< current_bucket
):
completed.append(
volume
)
if not completed:
return (
0.0,
0,
False,
)
completed = completed[
-self.baseline_buckets:
]
baseline = median(
completed
)
reliable = (
len(completed)
>= self.min_baseline_buckets
)
return (
float(baseline),
len(completed),
reliable,
)
# ==================================================================
# UTILITIES
# ==================================================================
@staticmethod
def _clamp(
value: float,
minimum: float,
maximum: float,
) -> float:
return max(
minimum,
min(
maximum,
value,
),
)
# ==================================================================
# SIGNAL SCORES
# ==================================================================
def _calculate_scores(
self,
intensity: float,
imbalance: float,
price_change_pct: float,
oi_change_pct: float,
oi_ready: bool,
) -> Dict[str, float]:
"""
Calculate normalized signal-strength scores.
Scores are signal strengths, not probabilities.
"""
liquidation_pressure = (
self._clamp(
(
intensity
/ 5.0
) * 100.0,
0.0,
100.0,
)
)
imbalance_score = (
abs(imbalance)
* 100.0
)
price_impulse = (
self._clamp(
(
abs(
price_change_pct
)
/ 3.0
)
* 100.0,
0.0,
100.0,
)
)
if oi_ready:
oi_contraction = (
self._clamp(
(
abs(
min(
oi_change_pct,
0.0,
)
)
/ 3.0
)
* 100.0,
0.0,
100.0,
)
)
else:
oi_contraction = 0.0
return {
"liquidation_pressure":
liquidation_pressure,
"imbalance":
imbalance_score,
"price_impulse":
price_impulse,
"oi_contraction":
oi_contraction,
}
# ==================================================================
# MARKET ANALYSIS
# ==================================================================
def analyze_market(
self,
) -> Tuple[str, Dict]:
"""
Main market-state classifier.
"""
now = time.time()
with self.lock:
self._clean_old_events(
now
)
self._clean_ttl_cache(
now
)
self._clean_old_buckets(
now
)
(
last_price,
price_change_pct,
volatility_pct,
price_ready,
) = self._get_price_metrics()
if not price_ready:
return (
"INITIALIZING_PRICE_FEED",
{},
)
(
oi_change_pct,
oi_ready,
) = self._get_oi_change_5m(
now
)
(
long_liq,
short_liq,
total_liq,
imbalance,
) = self._get_liquidation_metrics()
(
baseline_med,
baseline_count,
baseline_ready,
) = self._get_baseline()
# ----------------------------------------------------------
# No liquidation activity.
# ----------------------------------------------------------
if total_liq <= 0:
metrics = {
"price":
last_price,
"price_change_5m":
price_change_pct,
"volatility_5m":
volatility_pct,
"oi_change_5m":
oi_change_pct,
"oi_ready":
oi_ready,
"total_liq_usd":
0.0,
"long_liq_usd":
0.0,
"short_liq_usd":
0.0,
"imbalance":
0.0,
"intensity":
0.0,
"baseline_med":
baseline_med,
"baseline_buckets":
baseline_count,
"baseline_ready":
baseline_ready,
"scores": {
"liquidation_pressure":
0.0,
"imbalance":
0.0,
"price_impulse":
self._clamp(
(
abs(
price_change_pct
)
/ 3.0
)
* 100.0,
0.0,
100.0,
),
"oi_contraction":
0.0,
},
"confidence":
0.0,
}
if (
abs(
price_change_pct
) < 0.3
):
return (
"NO_LIQUIDATION_ACTIVITY",
metrics,
)
return (
"PRICE_MOVEMENT_NO_LIQUIDATION",
metrics,
)
# ----------------------------------------------------------
# Intensity.
# ----------------------------------------------------------
if (
baseline_ready
and baseline_med > 0
):
intensity = (
total_liq
/ baseline_med
)
else:
intensity = 0.0
# ----------------------------------------------------------
# Scores.
# ----------------------------------------------------------
scores = (
self._calculate_scores(
intensity=intensity,
imbalance=imbalance,
price_change_pct=price_change_pct,
oi_change_pct=oi_change_pct,
oi_ready=oi_ready,
)
)
# ----------------------------------------------------------
# Confidence.
# ----------------------------------------------------------
confidence_components = [
scores[
"liquidation_pressure"
],
scores[
"imbalance"
],
scores[
"price_impulse"
],
]
if oi_ready:
confidence_components.append(
scores[
"oi_contraction"
]
)
confidence = (
sum(
confidence_components
)
/ len(
confidence_components
)
)
metrics = {
"price":
last_price,
"price_change_5m":
price_change_pct,
"volatility_5m":
volatility_pct,
"oi_change_5m":
oi_change_pct,
"oi_ready":
oi_ready,
"total_liq_usd":
total_liq,
"long_liq_usd":
long_liq,
"short_liq_usd":
short_liq,
"imbalance":
imbalance,
"intensity":
intensity,
"baseline_med":
baseline_med,
"baseline_buckets":
baseline_count,
"baseline_ready":
baseline_ready,
"scores":
scores,
"confidence":
confidence,
}
# ==========================================================
# CASCADES
# ==========================================================
if (
baseline_ready
and intensity >= 3.0
and imbalance >= 0.60
and price_change_pct <= -2.0
):
return (
"CASCADING_DOWNSELL",
metrics,
)
if (
baseline_ready
and intensity >= 3.0
and imbalance <= -0.60
and price_change_pct >= 2.0
):
return (
"CASCADING_PUMP",
metrics,
)
# ==========================================================
# LONG CAPITULATION
# ==========================================================
if (
baseline_ready
and oi_ready
and intensity >= 2.0
and imbalance >= 0.50
and price_change_pct <= -0.50
and oi_change_pct <= -0.50
):
return (
"LONG_CAPITULATION",
metrics,
)
# ==========================================================
# SHORT SQUEEZE EXHAUSTION
# ==========================================================
if (
baseline_ready
and oi_ready
and intensity >= 2.0
and imbalance <= -0.50
and price_change_pct >= 0.50
and oi_change_pct <= -0.50
):
return (
"SHORT_SQUEEZE_EXHAUSTION",
metrics,
)
# ==========================================================
# DELEVERAGING
# ==========================================================
if (
oi_ready
and imbalance >= 0.50
and oi_change_pct <= -0.30
):
return (
"LONG_DELEVERAGING",
metrics,
)
if (
oi_ready
and imbalance <= -0.50
and oi_change_pct <= -0.30
):
return (
"SHORT_DELEVERAGING",
metrics,
)
# ==========================================================
# EXTREME LIQUIDATION SPIKES
# ==========================================================
if (
baseline_ready
and intensity >= 2.5
and imbalance >= 0.60
):
return (
"EXTREME_LONG_LIQUIDATION_SPIKE",
metrics,
)
if (
baseline_ready
and intensity >= 2.5
and imbalance <= -0.60
):
return (
"EXTREME_SHORT_LIQUIDATION_SPIKE",
metrics,
)
# ==========================================================
# ELEVATED LIQUIDATION
# ==========================================================
if imbalance >= 0.40:
return (
"ELEVATED_LONG_LIQUIDATION",
metrics,
)
if imbalance <= -0.40:
return (
"ELEVATED_SHORT_LIQUIDATION",
metrics,
)
# ==========================================================
# BASELINE WARMUP
# ==========================================================
if not baseline_ready:
return (
"WARMING_BASELINE",
metrics,
)
if intensity >= 1.5:
return (
"BALANCED_HIGH_LIQUIDATION_VOLUME",
metrics,
)
return (
"BALANCED",
metrics,
)
# ==================================================================
# WEBSOCKET CREATION
# ==================================================================
def _create_websocket(self):
"""
Create Binance combined WebSocket.
"""
combined_stream = (
"wss://fstream.binance.com/stream"
f"?streams="
f"{self.symbol}@forceOrder/"
f"{self.symbol}@ticker"
)
def on_open(ws):
with self.lock:
self.ws_connected = True
self.logger.info(
"WebSocket connected: %s",
self.symbol_upper,
)
def on_message(
ws,
message,
):
self.last_ws_message = (
time.time()
)
try:
payload = json.loads(
message
)
stream = payload.get(
"stream",
"",
)
data = payload.get(
"data",
{},
)
if (
"forceOrder"
in stream
):
self._process_liquidation(
data.get(
"o",
{},
)
)
elif (
"ticker"
in stream
):
self._process_ticker(
data
)
except json.JSONDecodeError:
self.logger.warning(
"Invalid WebSocket JSON."
)
except Exception:
self.logger.exception(
"Unexpected WebSocket message error."
)
def on_error(
ws,
error,
):
with self.lock:
self.ws_connected = False
self.logger.warning(
"WebSocket error: %s",
error,
)
def on_close(
ws,
close_status_code,
close_msg,
):
with self.lock:
self.ws_connected = False
self.logger.warning(
"WebSocket closed: "
"code=%s message=%s",
close_status_code,
close_msg,
)
return websocket.WebSocketApp(
combined_stream,
on_open=on_open,
on_message=on_message,
on_error=on_error,
on_close=on_close,
)
# ==================================================================
# WEBSOCKET LOOP
# ==================================================================
def _websocket_loop(
self,
) -> None:
"""
Persistent WebSocket loop.
Uses exponential reconnect backoff capped at 60 seconds.
Shutdown is interruptible through stop_event.
"""
backoff = 1
while not self.stop_event.is_set():
try:
self.ws_combined = (
self._create_websocket()
)
self.ws_reconnects += 1
self.logger.info(
"Starting WebSocket "
"connection attempt #%d.",
self.ws_reconnects,
)
self.ws_combined.run_forever(
ping_interval=30,
ping_timeout=10,
skip_utf8_validation=False,
)
# If run_forever returns normally after
# an established connection, reset backoff.
if self.is_running:
backoff = 1
except Exception as exc:
self.logger.error(
"WebSocket run_forever crashed: %s",
exc,
)
finally:
with self.lock:
self.ws_connected = False
if self.stop_event.is_set():
break
self.logger.info(
"Reconnecting WebSocket in %d seconds...",
backoff,
)
# Interruptible exponential backoff.
if self.stop_event.wait(
backoff
):
break
backoff = min(
backoff * 2,
60,
)
self.logger.info(
"WebSocket loop stopped."
)
# ==================================================================
# START
# ==================================================================
def start(
self,
) -> None:
"""
Start all background threads.
"""
with self.lock:
if self.is_running:
self.logger.warning(
"Engine is already running."
)
return
self.is_running = True
self.stop_event.clear()
# --------------------------------------------------------------
# Open Interest thread
# --------------------------------------------------------------
self.oi_thread = threading.Thread(
target=self._poll_open_interest,
name=(
f"OI_Poll_"
f"{self.symbol_upper}"
),
daemon=True,
)
self.oi_thread.start()
# --------------------------------------------------------------
# WebSocket thread
# --------------------------------------------------------------
self.ws_thread = threading.Thread(
target=self._websocket_loop,
name=(
f"WS_Loop_"
f"{self.symbol_upper}"
),
daemon=True,
)
self.ws_thread.start()
# --------------------------------------------------------------
# Maintenance thread
# --------------------------------------------------------------
self.maintenance_thread = (
threading.Thread(
target=self._maintenance_loop,
name=(
f"Maintenance_"
f"{self.symbol_upper}"
),
daemon=True,
)
)
self.maintenance_thread.start()
self.logger.info(
"ProductionMarketEngine started "
"successfully: %s",
self.symbol_upper,
)
# ==================================================================
# STOP
# ==================================================================
def stop(
self,
) -> None:
"""
Graceful and interruptible shutdown.
Order:
1. Signal all workers to stop.
2. Close WebSocket.
3. Wait for threads.
4. Reset runtime state.
"""
with self.lock:
if not self.is_running:
return
self.logger.info(
"Stopping ProductionMarketEngine..."
)
self.is_running = False
self.stop_event.set()
self.ws_connected = False
# --------------------------------------------------------------
# Explicitly close WebSocket.
# This unblocks run_forever().
# --------------------------------------------------------------
ws = self.ws_combined
if ws is not None:
try:
ws.close()
except Exception as exc:
self.logger.debug(
"Error closing WebSocket: %s",
exc,
)
# --------------------------------------------------------------
# Join worker threads.
# --------------------------------------------------------------
current_thread = (
threading.current_thread()
)
threads = [
self.ws_thread,
self.oi_thread,
self.maintenance_thread,
]
for thread in threads:
if (
thread is not None
and thread.is_alive()
and thread is not current_thread
):
thread.join(
timeout=5.0
)
with self.lock:
self.ws_combined = None
self.ws_thread = None
self.oi_thread = None
self.maintenance_thread = None
self.logger.info(
"ProductionMarketEngine stopped."
)
# ==================================================================
# HEALTH
# ==================================================================
def health(
self,
) -> Dict:
"""
Runtime health information.
"""
now = time.time()
with self.lock:
ws_age = (
now
- self.last_ws_message
if self.last_ws_message > 0
else None
)
oi_age = (
now
- self.last_oi_timestamp
if self.last_oi_timestamp > 0
else None
)
price_age = (
now
- self.last_price_timestamp
if self.last_price_timestamp > 0
else None
)
return {
"running":
self.is_running,
"symbol":
self.symbol_upper,
"websocket_connected":
self.ws_connected,
"websocket_last_message_age":
ws_age,
"websocket_reconnects":
self.ws_reconnects,
"price":
self.current_price,
"price_age":
price_age,
"open_interest":
self.current_oi,
"oi_age":
oi_age,
"liquidation_events":
self.total_liquidation_events,
"invalid_liquidation_events":
self.invalid_liquidation_events,
"dedup_cache_size":
len(
self.seen_ids
),
"rolling_liquidation_events":
len(
self.events
),
"price_history_size":
len(
self.price_history
),
"oi_history_size":
len(
self.oi_history
),
"baseline_buckets":
len(
self.bucket_accum
),
}
# ======================================================================
# CONSOLE OUTPUT
# ======================================================================
def print_market_state(
state: str,
metrics: Dict,
) -> None:
"""
Human-readable realtime market state.
"""
if not metrics:
print(
f"\r[{time.strftime('%H:%M:%S')}] "
f"Инициализация потоков...",
end="",
flush=True,
)
return
scores = metrics.get(
"scores",
{},
)
print(
f"\n[{time.strftime('%H:%M:%S')}] "
f"СТАТУС: >>> {state} <<<"
)
print(
f" ├─ Цена BTC: "
f"${metrics['price']:,.2f}"
)
print(
f" ├─ Цена 5M: "
f"{metrics['price_change_5m']:+.2f}%"
)
print(
f" ├─ Волатильность 5M: "
f"{metrics['volatility_5m']:.2f}%"
)
oi_status = (
"READY"
if metrics["oi_ready"]
else "WARMING"
)
print(
f" ├─ Open Interest 5M: "
f"{metrics['oi_change_5m']:+.2f}% "
f"[{oi_status}]"
)
print(
f" ├─ Ликвидации 5M: "
f"${metrics['total_liq_usd']:,.0f}"
)
print(
f" │ ├─ Long: "
f"${metrics['long_liq_usd']:,.0f}"
)
print(
f" │ └─ Short: "
f"${metrics['short_liq_usd']:,.0f}"
)
print(
f" ├─ Imbalance: "
f"{metrics['imbalance']:+.3f}"
)
print(
f" ├─ Intensity: "
f"{metrics['intensity']:.2f}x"
)
print(
f" ├─ Baseline median: "
f"${metrics['baseline_med']:,.0f}"
)
print(
f" ├─ Baseline buckets: "
f"{metrics['baseline_buckets']}"
f"{' [READY]' if metrics['baseline_ready'] else ' [WARMING]'}"
)
print(
f" ├─ Liquidation score: "
f"{scores.get('liquidation_pressure', 0):.0f}/100"
)
print(
f" ├─ Imbalance score: "
f"{scores.get('imbalance', 0):.0f}/100"
)
print(
f" ├─ Price impulse: "
f"{scores.get('price_impulse', 0):.0f}/100"
)
print(
f" ├─ OI contraction: "
f"{scores.get('oi_contraction', 0):.0f}/100"
)
print(
f" └─ Confidence: "
f"{metrics.get('confidence', 0):.0f}/100"
)
# ======================================================================
# LOGGING
# ======================================================================
def configure_logging() -> None:
logging.basicConfig(
level=logging.INFO,
format=(
"%(asctime)s | "
"%(levelname)s | "
"%(name)s | "
"%(message)s"
),
datefmt="%Y-%m-%d %H:%M:%S",
)
# ======================================================================
# APPLICATION
# ======================================================================
if __name__ == "__main__":
configure_logging()
engine = ProductionMarketEngine(
symbol="btcusdt",
window_seconds=300,
oi_poll_seconds=10,
baseline_buckets=288,
min_baseline_buckets=12,
)
print(
"Starting Production Market Engine..."
)
print(
"Streams:"
)
print(
" - Binance Futures forceOrder"
)
print(
" - Binance Futures ticker"
)
print(
" - Binance Futures Open Interest"
)
print(
"Analysis window: 5 minutes"
)
print(
"Baseline: 288 calendar-aligned 5M buckets"
)
print(
"Baseline warm-up: "
f"{engine.min_baseline_buckets} buckets"
)
engine.start()
try:
while True:
state, metrics = (
engine.analyze_market()
)
print_market_state(
state,
metrics,
)
time.sleep(10)
except KeyboardInterrupt:
print(
"\n\nStopping engine..."
)
finally:
engine.stop()
print(
"Engine stopped."
)Cómo integrar esto en tu workflow diario:
- Ajusta el umbral (`threshold_usd`) según el tamaño de tu cuenta y el activo que estés operando. Para las cryptos de alta capitalización ($BTC, $ETH), un threshold a partir de $500,000 filtra el ruido y te muestra dónde saltaron los stops pesados de verdad.
- En cuanto brinque una alerta del script, cámbiate al gráfico de EXMON y revisa los clusters. Si ves una liquidación grande justo sobre una zona clave de soporte o resistencia, esa es tu señal principal para empezar a preparar un trade contratendencia.
Por qué el mapa engaña a los traders minoristas y cómo leerlo junto con los Funding Rates
El error más común de un novato es mirar el mapa de liquidaciones de forma aislada. Una pila densa de barras rojas o verdes en el gráfico no te garantiza una reversión. El mapa solo te muestra combustible potencial, pero no te dice hacia dónde acelerará el motor del mercado.
Para diferenciar un mechazo falso (fakeout) de una barrida de liquidez real y agresiva, tienes que cruzar la tasa de financiación (Funding Rate) con el interés abierto (Open Interest):
- Funding extremadamente positivo + subida del Open Interest: La masa está metiendo longs sobreapalancados a lo loco. Todos juran que esto se va a la luna. Es el escenario perfecto para una barrida profunda que liquide a los longs hacia abajo. El mapa de liquidaciones por la parte inferior va a estar saturado.
- Funding profundamente negativo + subida del Open Interest: El mercado está atiborrado de shorts. Cualquier impulso alcista mínimo desata un short squeeze en cadena, porque los shorts prestados tienen que recompraarse sí o sí.
Si ves un cluster masivo de liquidaciones de shorts en el mapa y el Funding Rate lleva varias horas en negativo profundo, la probabilidad de un reventón alcista brutal es altísima. Un pez gordo no va a dejar pasar la oportunidad de ir a buscar esa liquidez.
Setup #2: Rompimiento real en una barrida de liquidez (Continuación de tendencia)
No toda cacería de liquidaciones termina en un rebotazo en V. A veces el volumen de combustible acumulado es tan bestia que el precio no solo deja un mechazo de liquidación, sino que entra en una tendencia impulsiva que dura varias horas.
Cómo identificar un breakout verdadero en una zona de liquidación:
- Ausencia de retroceso: El precio rompe el nivel donde se acumulan los stops, pero en lugar de dejar una mecha larga, consolida por encima o por debajo del nivel con una vela de 1 hora de cuerpo completo.
- Comportamiento del CVD (Cumulative Volume Delta): El delta de compras/ventas agresivas se dispara en vertical junto con el precio. Esto significa que no solo están comprando las liquidaciones forzosas del exchange, sino que los compradores a mercado con billeteras grandes se están sumando con todo al movimiento.
- Open Interest tras el mechazo: Una vez ejecutadas las liquidaciones, el OI no se desploma a cero (o cae levemente y vuelve a subir de inmediato). Esta es la señal clara de que está entrando dinero fresco con posiciones grandes a tomar el relevo, y no simplemente gente cerrando deudas viejas.
En este caso, la entrada se hace en el retesteo de la zona rota, con un stop-loss estricto pegado justo detrás del nivel.
Reglas de oro de gestión de riesgo para cazar liquidaciones
Operar en zonas de liquidación es caminar por un campo minado. Te equivocas de timing por un par de segundos o entras sin stop, y pasas a ser el punto de datos en el mapa de calor de alguien más.
- Olvídate de apalancamientos mayores a 5x–10x: Cuando operas contra impulsos volátiles buscando barridas de stops, el mercado te puede meter un slippage de 1% a 2% en milisegundos. Un apalancamiento alto te va a quemar la cuenta antes de que tu propio stop-loss toque la orderbook.
- El stop-loss es sagrado: Si el precio rompió el nivel de liquidación y siguió de largo sin ti, no promedies a la baja. Acepta la pérdida fija de tu cuenta (1-2%), dale la razón al mercado y espera a que se forme el siguiente setup.
- Ignora los clusters pequeños: No reacciones a cada fluctuación pequeña en la orderbook. Las ballenas solo van tras el dinero pesado. Enfócate únicamente en los clusters cuyo volumen supere por decenas de veces el volumen promedio por minuto del activo.
Hacer trading no es adivinar para dónde va la siguiente vela verde. Es un cálculo matemático de probabilidades y aprovechar fríamente los errores de los demás a tu favor. Analiza el mapa de liquidez, piensa como un market maker y nunca dejes una posición desprotegida.
