जहाँ Smart Money रिटेल को काटता है: Liquidation Map पर काम करने की Practical Strategy
Liquidation Heatmap समझो दूसरों के सपनों का X-ray है। जब रिटेल ट्रेडर किसी लोकल high या low के पीछे Technical Analysis के हिसाब से अपना Stop-Loss लगाता है, तब Whale या बड़ा प्लेयर Heatmap पर उन सभी SL पॉइंट्स के गुच्छे को अपने मूव के लिए रेडी-मेड ईंधन (Fuel) की तरह देखता है।
Market Maker के पास पतले (Thin) मार्केट में करोड़ों डॉलर की पोजीशन बनाने या काटने की जगह ही नहीं होती। उसे चाहिए Liquidity, और यह Liquidity उसे बाकी दुनिया के Margin Calls से मिलती है।
Shorters के लिए Trap: Short Squeeze का असली Mechanism
सिचुएशन इमेजिन करो: मार्केट तीन दिन से लगातार गिर रहा है। रिटेल पब्लिक मार्केट को वीक मानकर 20x–50x Leverage के साथ Short ठोक देती है और पास के रेजिस्टेंस (जैसे राउंड नंबर या कंसोलिडेशन के टॉप) पर Stop-Loss लगा देती है या Liquidation तक मार्जिन होल्ड करती है।
अब उस Big Player को सोचो जिसे भारी Long पोजीशन बनानी है या अपने बैग्स खाली करने हैं, वह क्या करेगा?
- Price Compression (प्राइस को ऊपर दबाना): प्राइस धीरे-धीरे ऊपर खिसकेगा ताकि Shorters पैनिक न करें, लेकिन यह सीधा Heatmap के Liquidation Zone के पास पहुँच जाता है।
- Impulsive Spike (एक तगड़ा स्पाइक): एक ही झटके में प्राइस को रेजिस्टेंस लेवल से 1.5–2% ऊपर खींच दिया जाता है।
- Chain Reaction (डोमिनो इफेक्ट): इस पॉइंट पर शुरुआती SL ट्रिगर होते हैं और हाई-लेवरेज वाली पोजीशंस लिक्विडेट होने लगती हैं। एक्सचेंज इन शॉर्त्स को बंद करने के लिए मार्केट पर ऑटोमैटिक Buy ऑर्डर मारने लगता है।
- Volume Explosion (वॉल्यूम का धमाका): एक्सचेंज के मार्जिन कॉल्स से आने वाले Market Buyers सीधे उस बड़े प्लेयर के Limit Sell ऑर्डर्स से टकराते हैं, जिसने पैनिक बाइंग की इस लिक्विडिटी में अपने ऑर्डर्स पहले से बिछा रखे थे।
रिटेल ट्रेडर सोचता है: "Breakout आ गया! ट्रेंड पलट गया, चलो Long मारते हैं!" वह मूव के एकदम टॉप पर पैनिक में बाय करता है, और वहीं से बड़ा प्लेयर गेम पलट देता है—प्राइस पत्थर की तरह नीचे गिरता है और इस बार Long्स के स्टॉप-लॉस उड़ाता है।
Order Book में 'धमाका' क्यों होता है?
जब $500,000 की 20x लेवरेज वाली पोजीशन लिक्विडेट होती है, तो एक्सचेंज उसे क्लोज करने के लिए तुरंत Market Order ठोकता है। अगर Order Book में उल्टी तरफ (Opposite Orders) पर्याप्त Limit Bids नहीं हैं, तो प्राइस लेवल्स को चीरता हुआ निकल जाता है।
- Limit Orders का सफाया: Liquidation Engine ऑर्डर बुक को खाली करना शुरू करता है। अगर $60,000 के लेवल पर 5 BTC रखे हैं और 20 BTC लिक्विडेट हो रहे हैं, तो प्राइस लेवल को तोड़कर तब तक आगे भागेगा जब तक उसे पूरा वॉल्यूम न मिल जाए।
- Domino Effect: हर अगला टूटा हुआ लेवल नए Stop-Losses और कम लेवरेज (जैसे 10x वाले) वालों की लिक्विडेशन ट्रिगर कर देता है।
- Dead Cat Bounce (रिवर्सल स्पाइक): जैसे ही यह कैस्केड पूरे इंपल्स में सारी लिक्विडिटी को भस्म कर देता है, ऑर्डर बुक एकदम खाली हो जाती है। ठीक उसी वक्त Market Maker, जो इस मौके की ताक में था, ज़बरदस्त Limit Orders के साथ सबसे बेस्ट प्राइस पर एंट्री मारता है और प्राइस तुरंत वी-शेप में पलट जाता है।
Trading Setups: Liquidation Zones से पैसा कैसे निकालें
लिक्विडेशन के उस पागलपन भरे मोमेंट में आँख बंद करके ट्रेड में मत कूदो—स्लिपेज और स्प्रेड तुम्हें कच्चा चबा जाएँगे। जब यह आग ठंडी हो जाए, तब मार्केट के Reaction पर खेलना चाहिए।
Setup #1: Stop Hunt पर ट्रेड (Liquidation Grab)
- Conditions: Heatmap पर Longs या Shorts का एक मोटा क्लस्टर दिख रहा हो, जहाँ तक पहुँचने में प्राइस को 1% से भी कम का मूव चाहिए।
- Action: लेवल टूटने/स्पाइक मारने का वेट करो। ब्रेकआउट होते ही तुरंत एंट्री मत लो।
- Execution: प्राइस तेज़ी से लेवल के बाहर जाता है, 1-मिनट चार्ट पर वॉल्यूम आसमान छूने लगता है, और Open Interest (OI) तेज़ी से गिरता है (सबका सफाया हो गया)। जैसे ही कैंडल वापस लेवल के अंदर आकर लंबी विक (Wick/Shadow) बनाकर क्लोज हो—उल्टी डायरेक्शन में ट्रेड ठोक दो।
- Stop-Loss: SL उस स्पाइक वाली कैंडल के विक (Extreme Shadow) के ठीक बाहर रहेगा। अगर प्राइस दोबारा उस लेवल को पार करता है, तो इसका मतलब यह असली Breakout था, Fakeout नहीं। रिस्क को हमेशा टाइट और कंट्रोल में रखो।
Automation Script: Anomaly Detection Script
मार्केट में लिक्विडेशन कैस्केड का इंतज़ार करते हुए 24 घंटे स्क्रीन के सामने न बैठना पड़े, इसके लिए नीचे दिए गए स्क्रिप्ट का यूज़ करो। यह रियल-टाइम में एक्सचेंज के लिक्विडेशन डेटा को ट्रैक करता है और बड़े ऑर्डर्स मिलने पर कंसोल में अलर्ट देता है।
कोड एकदम शुद्ध Python में लिखा गया है बिना किसी फालतू लाइब्रेरी के, यह एकदम स्टेबल चलता है और एक्सचेंज से कनेक्शन टूटने पर भी क्रैश नहीं होता।
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."
)इसे अपने Workflow में कैसे यूज़ करें:
- अपने कैपिटल और एसेट के हिसाब से `threshold_usd` सेट करो। $BTC या $ETH जैसे बड़े कॉइन्स के लिए $500,000+ का थ्रेशोल्ड असली बड़े Stop Hunts को पकड़ेगा।
- जैसे ही स्क्रिप्ट अलर्ट दे, तुरंत चार्ट्स पर स्विच करो और Clusters चेक करो। अगर कोई बड़ी लिक्विडेशन किसी तगड़े Support/Resistance लेवल पर हुई है—तो समझ जाओ काउंटर-ट्रेंड (Reversal) ट्रेड सेट करने का यही तुम्हारा मेन सिग्नल है।
लिक्विडेशन मैप सोलो ट्रेडर्स को धोखा क्यों देता है और Funding Rates के साथ इसे कैसे पढ़ें
नये ट्रेडर्स की सबसे बड़ी गलती यह है कि वे लिक्विडेशन हीटमैप को अकेले (in isolation) देखते हैं। चार्ट पर लाल या हरी पट्टियों का भारी जमावड़ा इस बात की गारंटी नहीं है कि मार्केट वहीं से रिवर्स हो जाएगा। मैप आपको सिर्फ संभावित ईंधन (liquidity fuel) दिखाता है, लेकिन यह नहीं बताता कि मेन इंजन बाजार को किस दिशा में ले जाएगा।
एक फेक-आउट (fakeout) और एक असली बड़े स्वीप (liquidity sweep) के बीच का अंतर समझने के लिए आपको फंडिंग रेट (Funding Rate) और ओपन इंटरेस्ट (Open Interest) को मिलाकर देखना होगा:
- बेहद पॉजिटिव Funding + तेजी से बढ़ता Open Interest: रिटेल क्राउड धड़ाधड़ हाई लेवरेज के साथ लॉन्ग पोजीशन भर रहा है। हर कोई बुलिश है। यह लॉन्ग ट्रेडर्स के कत्लेआम (long squeeze) के लिए सबसे मुफीद माहौल बनाता है। नीचे का लिक्विडेशन मैप भारी भरा होगा।
- काफी नेगेटिव Funding + तेजी से बढ़ता Open Interest: मार्केट में शॉर्ट करने वालों का तांता लगा हुआ है। ऊपर की तरफ एक छोटा सा भी स्पाइक एक चेन-रिएक्शन शॉर्ट स्क्वीज़ (short squeeze) को ट्रिगर कर देगा, क्योंकि शॉर्टर्स को अपनी पोजीशन कवर करने के लिए पैनिक बाय करना पड़ेगा।
अगर आपको मैप पर शॉर्ट लिक्विडेशन का एक बड़ा क्लस्टर दिखता है और फंडिंग रेट लगातार कई घंटों से गहरे माइनस में अटका हुआ है—तो ऊपर की तरफ एक वायलेंट मूव की संभावना सबसे ज्यादा होती है। बड़े प्लेयर्स (whales) इतनी आसानी से टेबल पर पड़ी लिक्विडिटी को नहीं छोड़ते।
सेटअप #2: असली लिक्विडिटी स्वीप पर ब्रेकआउट (ट्रेड कंटिन्यूएशन)
हर लिक्विडेशन स्वीप तुरंत V-शेप रिजेक्शन या विक (wick) बनाकर खत्म नहीं होता। कभी-कभी जमा हुआ फ्यूल इतना ज्यादा होता है कि कीमत सिर्फ स्टॉप्स हंट करके वापस नहीं लौटती, बल्कि अगले कुछ घंटों के लिए एक तगड़े मोमेंटम ट्रेंड में निकल जाती है।
लिक्विडेशन जोन के असली ब्रेकआउट को कैसे पहचानें:
- कोई रिट्रेसमेंट नहीं: प्राइस स्टॉप क्लस्टर के लेवल को चीरते हुए निकल जाता है, लेकिन एक लंबी विक छोड़ने के बजाय 1-घंटे की कैंडल उस लेवल के पार मजबूती से क्लोज होती है।
- CVD (Cumulative Volume Delta) का उछाल: एग्रेसिव मार्केट बाय/सेल डेल्टा कीमत के साथ-साथ वर्टिकली रॉकेट की तरह ऊपर भागता है। इसका मतलब है कि सिर्फ एक्सचेंज के ऑटोमैटिक लिक्विडेशन ही बाय नहीं कर रहे, बल्कि बड़े मार्केट बायर्स भी ब्रेकआउट में कूद चुके हैं।
- स्वीप के बाद Open Interest का बर्ताव: लिक्विडेशन का शुरुआती झटका खत्म होने के बाद भी OI जीरो नहीं होता (या हल्का सा डिप लेकर तुरंत वापस बढ़ने लगता है)। यह इस बात का पक्का संकेत है कि केवल पुराने डूबे हुए ट्रेड बंद नहीं हुए हैं, बल्कि नए स्मार्ट मनी ने नई पोजीशंस बनाई हैं।
ऐसी स्थिति में एंट्री टूटे हुए लेवल के री-टेस्ट पर ली जाती है, और स्टॉप लॉस को उस स्ट्रक्चर के ठीक पीछे एकदम टाइट रखा जाता है।
लिक्विडिटी हंटिंग के दौरान रिस्क मैनेजमेंट के कड़े नियम
लिक्विडेशन जोन्स में ट्रेडिंग करना बारूदी सुरंगों पर चलने जैसा है। दो सेकंड की टाइमिंग मिस्टेक या बिना स्टॉप लॉस के एंट्री—और आप खुद किसी और के हीटमैप का डेटा प्वाइंट बन जाएंगे।
- 5x–10x से ज्यादा लेवरेज भूल जाइए: जब आप स्टॉप हंट वाले हाई-वोलेटिलिटी स्पाइक्स को ट्रेड कर रहे हों, तो मार्केट मिलीसेकंड्स में 1–2% का स्लिपेज (slippage) दे सकता है। आपका स्टॉप लॉस आर्डर बुक में ट्रिगर होने से पहले ही हाई लेवरेज आपके मार्जिन का सूप बना देगा।
- स्टॉप लॉस पत्थर की लकीर है: अगर प्राइस लिक्विडेशन लेवल को तोड़कर आपके बिना आगे निकल जाए—तो एवरेज डाउन (averaging down) कभी मत कीजिए। अपने पोर्टफोलियो का एक फिक्स्ड हिस्सा (1-2% मैक्स) खोना स्वीकार कीजिए, मानिए कि मार्केट सही था और अगले साफ सेटअप का इंतजार कीजिए।
- छोटे-मोटे क्लस्टर्स को इग्नोर करें: आर्डर बुक के हर छोटे-मोटे शोर पर रिएक्ट मत कीजिए। बड़े प्लेयर्स सिर्फ मोटी लिक्विडिटी का शिकार करते हैं। केवल उन क्लस्टर्स पर ध्यान दें जिनका वॉल्यूम उस एसेट के नॉर्मल 1-मिनट वॉल्यूम से दसियों गुना बड़ा हो।
ट्रेडिंग का मतलब अगली हरी कैंडल की दिशा का तुक्का लगाना नहीं है। यह सिर्फ प्रोबेबिलिटी का गणित है और ठंडे दिमाग से दूसरों की गलतियों का फायदा उठाना है। लिक्विडिटी मैप को एनालाइज कीजिए, मार्केट मेकर की तरह सोचिए और अपनी पोजीशन को कभी भी बिना प्रोटेक्शन के मत छोड़िए।
