Gdzie „Smart Money” goli ulicę: Praktyka handlu z mapą likwidacji
Mapa cieplna likwidacji to RTG cudzych nadziei. Podczas gdy detaliczny trader ustawia stop-lossa z analizy technicznej tuż za najbliższym lokalnym szczytem lub dołkiem, duży gracz widzi na mapie zagęszczenie tych punktów jako gotowe paliwo do wywołania ruchu.
Market maker nie ma fizycznie możliwości zbudowania ani zamknięcia pozycji wartej miliony dolarów na płytkim rynku. Potrzebuje płynności. A tę płynność zapewniają mu cudze margin calla.
Pułapka na szortujących: Mechanika short squeeze'a
Wyobraź sobie sytuację: aktywo spada od trzech dni. Tłum widzi słabość, wchodzi w szorty z dźwignią 20x–50x i stawia stop-lossy (albo trzyma margines aż do likwidacji) tuż za najbliższym oporem — powiedzmy na okrągłym poziomie lub przy górnym ograniczeniu konsolidacji.
Co robi grubas, który musi załadować potężną pozycję long albo zrzucić swój zapas?
- Dociskanie ceny: Cena powoli pełznie w górę, nie wywołując jeszcze paniki u szortujących, ale podchodzi pod samą strefę likwidacji na mapie.
- Impulsywne wybicie: Jednym gwałtownym strzałem wyciągają cenę o 1.5–2% ponad poziom oporu.
- Reakcja łańcuchowa: W tym momencie strzelają pierwsze stopy, a pozycje z wysoką dźwignią idą do kasacji. Giełda automatycznie kupuje aktywo z rynku, żeby zamknąć te szorty.
- Eksplozja wolumenu: Rynkowe zlecenia kupna z giełdowych margin calli wpadają wprost na zlecenia limit sprzedającego grubasa, który wystawił je wcześniej, czekając na ten paniczny popyt ze strony szortów.
Trader detaliczny myśli: „Wybicie! Trend się odwrócił, ładuję się w longa!”. Kupuje na samym szczycie impulsu, po czym duży gracz obraca pozycję i cena leci w dół jak kamień, zbierając dla odmiany stopy z longów.
Dlaczego karnet zleceń „wybucha”
Gdy wpada likwidacja pozycji za 500,000 $ na dźwigni 20x, giełda wystawia zlecenie rynkowe na jej zamknięcie. Jeśli w arkuszu na odpowiednich poziomach cenowych nie ma przeciwstawnych zleceń limit, cena po prostu przestrzeliwuje kolejne poziomy.
- Zjadanie limitów: Silnik likwidacyjny zaczyna walić w arkusz. Jeśli na poziomie 60,000 $ stało 5 Bitcoinów w arkuszu, a do zlikwidowania jest 20 Bitcoinów, cena przebija ten poziom i leci dalej, aż znajdzie wolumen.
- Efekt domina: Każdy kolejny przebity poziom aktywuje nowe stop-lossy i likwidacje z niższą dźwignią (na przykład u tych, którzy siedzieli na 10x).
- Szybkie odbicie (V-bottom / Dead Cat Bounce): Gdy tylko kaskada wypali całą płynność na impulsie, arkusz staje się pusty. W tym momencie market maker, który czekał na to wyczerpanie, wchodzi po najlepszej cenie z ogromnym wolumenem zleceń limit, błyskawicznie zawracając cenę.
Setupy transakcyjne: Jak wyciągać kasę ze stref likwidacji
Nie próbuj łapać samego momentu likwidacji na ślepo — spread i poślizg cenowy (slippage) po prostu cię rozniosą. Trzeba grać pod reakcję rynku, gdy ten pożar już zgaśnie.
Setup nr 1: Gra pod czyszczenie płynności (Polowanie na stopy)
- Warunki: Na mapie likwidacji widać gęstą chmurę szortów lub longów, do której cenie zostało mniej niż procent ruchu.
- Co robić: Czekać na przebicie poziomu. Nie wchodzić w pozycję na samym wybiciu.
- Egzekucja: Cena gwałtownie wylatuje poza poziom, wolumen na wykresie 1-minutowym wywala w kosmos, a Open Interest (OI) drastycznie spada (wszyscy zgoleni). Gdy tylko świeca zamyka się z długim knotem wracającym pod poziom — otwierasz pozycję w przeciwnym kierunku.
- Stop-loss: Ustawiany rygorystycznie za ekstremum tego impulsu (za knotem, który zebrał likwidacje). Jeśli cena znowu przebije ten poziom — znaczy to, że to było prawdziwe wybicie, a nie fałszywka pod płynność. Ryzyko musisz trzymać na krótkiej smyczy.
Automatyzacja monitoringu: Skrypt do wyłapywania anomalii
Żeby nie ślęczeć przed monitorem 24/7 w oczekiwaniu na kaskadę, użyj poniższego skryptu. Śledzi on strumień likwidacji z giełdy w czasie rzeczywistym i wypluwa w konsoli anomalne wolumeny wymuszonych zamknięć.
Kod jest napisany w czystym Pythonie bez zbędnych bibliotek, działa stabilnie i nie wywala się przy zerwaniu połączenia z giełdą.
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."
)Jak wdrożyć to do swojego workflow:
- Ustaw próg (threshold_usd) pod swój depozyt i grany walor. Dla dużych coinów ($BTC, $ETH) próg od 500,000 $ wyłapuje już konkretne, grube likwidacje.
- Gdy skrypt wypluje alert, przełącz się na wykres EXMON i sprawdź klastry. Jeśli duża likwidacja poszła na silnym poziomie wsparcia lub oporu — to twój główny sygnał do przygotowania pozycji pod rozegranie kontry.
Dlaczego mapa kłamie pojedynczym traderom i jak czytać ją w połączeniu z Funding Rate
Najczęstszy błąd nowicjusza to patrzenie na mapę likwidacji w próżni. Gęste skupisko czerwonych lub zielonych pasków na wykresie wcale nie gwarantuje odwrócenia trendu. Mapa pokazuje potencjalne paliwo, ale nie mówi, w którą stronę pojedzie cała maszyna.
Aby odróżnić fałszywe wybicie od realnego, potężnego ruchu, musisz patrzeć na stawkę finansowania (Funding Rate) oraz otwarte pozycje (Open Interest):
- Skrajnie dodatni Funding + wzrost Open Interest: Tłum masowo ładuje się w longi z dużą dźwignią. Wszyscy wierzą w wzrosty. To idealne środowisko do brutalnego czyszczenia longów w dół. Mapa likwidacji na dole będzie wręcz przeładowana.
- Głęboko ujemny Funding + рост Open Interest: Rynek jest zapchany shortami. Jakikolwiek lokalny impuls w górę odpala reakcję łańcuchową i short squeeze, bo pożyczone pozycje krótkie trzeba natychmiast odkupić z rynku.
Jeśli widzisz na mapie potężny klaster likwidacji shortów, a Funding Rate utrzymuje się mocno pod kreską przez kilka godzin z rzędu — prawdopodobieństwo potężnego wybicia w górę rośnie do maksimum. Gruby gracz nie przepuści okazji, żeby zgarnąć taką płynność.
Setup #2: Wybicie poziomu przy realnym ruchu (Kontynuacja trendu)
Nie każde zgarnięcie likwidacji kończy się gwałtownym powrotem i odbiciem. Czasami ilość nagromadzonego paliwa jest tak duża, że cena nie robi tylko szybkiego knota-likwidacji, ale wchodzi w silny, impulsywny trend na kilka godzin.
Jak rozpoznać prawdziwe wybicie strefy likwidacji:
- Brak cofki: Cena przebija poziom skupiska stop-lossów, ale zamiast zostawić długi cień, zamyka się świecą godzinową mocno powyżej/poniżej poziomu.
- Zachowanie CVD (Cumulative Volume Delta): Delta agresywnych zakupów/sprzedaży pionowo szybuje w górę razem z ceną. Oznacza to, że rynek jest skupowany nie tylko przez automatyczne likwidacje giełdowe, ale do ruchu dołączają też duzi gracze rynkowi.
- Open Interest po wybiciu: Po przejściu fali likwidacji OI nie spada do zera (albo spada nieznacznie, a potem znowu rośnie). To jasny sygnał, że na zwolnione miejsce wchodzą nowe, duże pozycje, a nie tylko zamykają się stare długi.
W takim scenariuszu wchodzimy na reteście przebitej strefy (od góry do dołu lub od dołu do góry), ze sztywnym stopem tuż za granicą tego poziomu.
Twarde zasady risk managementu przy polowaniu na likwidacje
Handel w strefach likwidacji to chodzenie po polu minowym. Pomylisz się z timingiem o dwie sekundy albo wejdziesz bez stopa — i sam staniesz się częścią czyjejś mapy ciepła.
- Zapomnij o dźwigni powyżej 5x–10x: Kiedy grasz pod zmienne impulsy przy wycinaniu stopów, rynek może dać poślizg (slippage) o 1–2% w ułamku sekundy. Wysoka dźwignia zabije Twój depozyt zanim giełda w ogóle zrealizuje Twój własny stop-loss.
- Stop-loss to świętość: Jeśli cena przebiła poziom likwidacji i idzie dalej bez Ciebie — nie uśredniaj. Tnij stały procent depozytu (1-2%), przyznaj rynkowi rację i czekaj na kolejną czystą strefę.
- Ignoruj małe klastry: Nie reaguj na szum i małe plunięcia ceny w arkuszu. Duzi gracze polują tylko na grubą kasę. Patrz na klastry, których wolumen wielokrotnie przewyższa średni minutowy wolumen obrotu na danym aktywie.
Trading to nie zgadywanie, w którą stronę poleci zielona świeca. To chłodna kalkulacja prawdopodobieństwa i bezwzględne wykorzystywanie cudzych błędów na swoją korzyść. Analizuj mapę płynności, myśl jak market maker i nigdy nie zostawiaj pozycji bez ochrony.
