Onde o Smart Money liquida a sardinhada: Prática com Mapa de Liquidação
O mapa de calor de liquidações é o raio-X da esperança alheia. Enquanto o trader varejo coloca o stop loss certinho pela análise técnica, logo atrás da máxima ou mínima local, o grande player enxerga esse acúmulo de pontos no mapa como combustível pronto para mover o preço.
Para o market maker, é fisicamente impossível montar ou fechar uma posição de milhões de dólares em um mercado ilíquido. Ele precisa de liquidez. E essa liquidez vem direto do margin call dos outros.
A armadilha para os vendidos: Anatomia do Short Squeeze
Pensa comigo: o ativo tá despencando há três dias. A massa enxerga fraqueza, entra vendida com alavancagem de 20x a 50x e soca o stop loss (ou segura a margem até a liquidação) bem atrás da resistência local mais próxima — tipo uma figura redonda ou o topo de uma consolidação.
O que faz o grande player que precisa montar uma posição comprada pesada ou desovar o seu estoque?
- Pressionar o preço: O preço sobe devagarzinho, sem causar pânico nos vendidos, mas colando na zona de liquidação do mapa.
- Violinada impulsiva: Em um único puxão violento, o preço sobe 1.5% a 2% acima da resistência.
- Reação em cadeia: Nesse exato momento, os primeiros stops são acionados e as posições hiperalavancadas liquidadas. A corretora entra comprando a mercado automaticamente para fechar esses shorts.
- Explosão de volume: Essa demanda de ordens a mercado gerada pelos margin calls bate de frente com as ordens limite de venda do grande player, que já tinha deixado tudo armado esperando justamente essa liquidez de pânico.
O varejo pensa: «Rompimento! Reversão de tendência, vou entrar comprado com tudo!». Compra no topo do FOMO e do impulso, pra logo em seguida o grande player virar a mão e o preço cair igual uma pedra, buscando agora os stops dos compradores.
Por que o Order Book «derrete»
Quando dispara a liquidação de um player alavancado em 20x com $500.000, a corretora joga uma ordem a mercado para zerar a posição. Se não houver ordens limite de contraparte suficientes no livro naquele preço, o valor rasga os níveis.
- Absorção das ordens limite: O motor de liquidação começa a varrer o livro. Se no nível de $60.000 tinha só 5 Bitcoins e a liquidação é de 20 Bitcoins, o preço rompe e continua andando até achar volume.
- Efeito dominó: Cada nível rompido ativa novos stop losses e novas liquidações de quem estava menos alavancado (tipo a galera de 10x).
- Repique violento (Dead Cat Bounce): Assim que a cascata queima toda a liquidez do movimento, o livro fica «vazio». Nesse momento, o market maker, que só esperava esse esgotamento, entra no melhor preço possível com um bloco gigante de ordens limite, revertendo o preço instantaneamente.
Setups Operacionais: Como extrair dinheiro das zonas de liquidação
Não tente operar o momento exato da liquidação no escuro — você vai ser atropelado pelo spread e pelo slippage. O segredo é operar a reação do mercado depois que o incêndio apagar.
Setup #1: Caça aos Stops (Liquidity Sweep)
- Condições: Você identifica no mapa de calor um muro denso de liquidações de short ou long a menos de 1% de distância do preço atual.
- O que fazer: Esperar a violinada no nível. Não entre na operação durante o rompimento seco.
- Execução: O preço dá aquela esticada violenta além do nível, o volume no gráfico de 1 minuto vai para a lua e o Open Interest (OI) despenca (a galera foi stopada/liquidada). Assim que o candle fechar deixando um pavio longo de volta para dentro do nível, abra a posição no sentido oposto.
- Stop Loss: Posicionado estritamente atrás da ponta do pavio que coletou a liquidez. Se o preço voltar a rasgar aquele nível, significa que foi um rompimento verdadeiro, não uma falsa captura de liquidez. Mantenha seu risco sob controle rígido.
Automação do Monitoramento: Script para capturar anomalias
Para não ter que ficar mofando na frente do monitor 24/7 esperando a cascata acontecer, use o script abaixo. Ele monitora o fluxo de liquidações da exchange em tempo real e joga no console quando rola um volume atípico de fechamentos forçados.
O código foi feito em Python puro, sem dependências pesadas, roda super leve e não cai se a conexão com a API oscilar.
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."
)Como integrar isso no seu workflow:
- Ajuste o gatilho (`threshold_usd`) de acordo com a sua banca e o par negociado. Para altcoins grandes e majors ($BTC, $ETH), um filtro a partir de $500.000 pega apenas as liquidações relevantes de players grandes.
- Quando o script apitar, mude para o gráfico do EXMON e analise os clusters. Se uma grande liquidação rolou em cima de um suporte ou resistência forte, esse é o seu gatilho principal para preparar um trade contra a tendência.
Por que o Mapa Mentir para Traders Solo e Como Lê-lo em Conjunto com as Funding Rates
O erro mais comum de quem está começando é olhar para o mapa de liquidação no vácuo. Uma grande concentração de barras vermelhas ou verdes no gráfico não garante que o preço vá reverter. O mapa mostra onde está o combustível potencial, mas não diz para qual lado o motor vai acelerar.
Para diferenciar um falso violino (fakeout) de uma varredura de liquidez (sweep) real e violenta, você precisa cruzar a taxa de financiamento (Funding Rate) com o interesse aberto (Open Interest):
- Funding Extremamente Positivo + Open Interest em Alta: A sardinhada está entupindo o mercado de longs alavancados. Todo mundo crente na alta. Esse é o cenário perfeito para um "long squeeze" brutal liquidando todo mundo para baixo. O mapa de liquidação na parte inferior estará sobrecarregado.
- Funding Profundamente Negativo + Open Interest em Alta: O mercado está entupido de shorts. Qualquer impulso local para cima desencadeia um "short squeeze" em cadeia, porque quem está vendido a seco é obrigado a comprar a qualquer preço para fechar a posição.
Se você vê um cluster massivo de liquidações de short no mapa e a Funding Rate continua sangrando no negativo por várias horas seguidas, a probabilidade de uma lapada para cima é gigante. O player grande não vai deixar essa liquidez dadeira passar batida.
Setup #2: Rompimento Real na Varredura de Liquidez (Continuação de Tendência)
Nem todo expurgo de liquidação termina com um retorno rápido em V. Às vezes o combustível acumulado é tão insano que o preço não faz apenas um "pavio de liquidação", mas engata uma tendência impulsiva violenta que dura horas.
Como identificar um rompimento verdadeiro da zona de liquidação:
- Ausência de Retração: O preço rasga a zona cheia de stops, mas em vez de deixar uma sombra longa, o vela de 1 hora fecha com corpo cheio cravado acima (ou abaixo) do nível.
- Comportamento do CVD (Cumulative Volume Delta): O delta de compras/vendas agressivas dispara na vertical junto com o preço. Isso indica que não são apenas as ordens automáticas de liquidação da corretora comprando, mas também os grandes compradores a mercado entrando rasgando no movimento.
- Open Interest Após a Violada: Depois que o expurgo de liquidações passa, o OI não despenca para o zero (ou cai bem pouco e volta a subir imediatamente). Esse é o sinal definitivo de que novas mãos fortes estão assumindo o controle na liquidez liberada, e não apenas fechando posições antigas.
Neste caso, a entrada é feita no retest da zona rompida de cima para baixo (ou de baixo para cima), com um stop curto e bem ajustado atrás da linha do nível.
Regras Rígidas de Gestão de Risco ao Caçar Liquidações
Operar em zonas de liquidação é andar em um campo minado. Erre o timing por dois segundos ou entre sem stop, e você vira apenas mais um ponto de dado no mapa de calor de outra pessoa.
- Esqueça Alavancagem Acima de 5x–10x: Quando você opera contra impulsos ultra-voláteis na caça de stops, o mercado pode te dar um slippage de 1% a 2% em milissegundos. Alavancagem alta vai liquidar a sua conta antes mesmo da sua ordem de stop-loss ser processada no order book.
- Stop-Loss é Sagrado: Se o preço rompeu o nível de liquidação e foi embora sem você, não faça preço médio (martingale). Aceite a perda fixa (1-2% da banca), reconheça que o mercado estava certo e espere a próxima zona de setup limpa.
- Ignore Clusters Pequenos: Não reaja a cada oscilaçãozinha de preço no livro de ordens. Player grande só caça dinheiro grande. Foque apenas nos clusters cujo volume seja dezenas de vezes maior que o volume médio por minuto do ativo.
Trading não é adivinhar para onde a vela verde vai apontar. É cálculo matemático de probabilidades e a habilidade fria de usar o erro dos outros a seu favor. Analise o mapa de liquidez, pense como um market maker e nunca, em hipótese alguma, deixe uma posição desprotegida.
