feat: forced-liquidation stream (!forceOrder@arr) → crypto.liquidations

README.md

@@ -1,13 +1,13 @@
 # tick-crawler — Binance tick + market data → ClickHouse (`crypto`)
 
 Kéo dữ liệu Binance (**USDM futures** + **spot**) cho HFT vào warehouse ClickHouse dùng chung
-(`crypto`). Điền **toàn bộ 9 bảng**: 5 bảng tick mới + 4 bảng cũ (`symbol_info`, `ohlcv`,
+(`crypto`). Điền **toàn bộ 10 bảng**: 6 bảng tick mới + 4 bảng cũ (`symbol_info`, `ohlcv`,
 `open_interest`, `funding_rate`) — bằng `CREATE … IF NOT EXISTS` và `populate`, **không bao
 giờ `ALTER`** bảng cũ.
 
 - **Backfill (quá khứ)** — `data.binance.vision`: futures `trades`/`bookDepth`/`metrics`; spot `trades`.
 - **Live (real-time)** — WebSocket order book + `@aggTrade` + `@kline` (mọi khung, nến đã đóng)
-  + REST open-interest/funding-rate + REST long/short metrics.
+  + `!forceOrder@arr` (thanh lý toàn thị trường) + REST open-interest/funding-rate + REST long/short metrics.
 
 > Dependency manager: **Poetry**, virtualenv **in-project** (`.venv/`). Chạy **cùng máy với
 > ClickHouse** (endpoint tự nhận `127.0.0.1`). Tune cho **8 CPU** dùng chung. Trades chạy
@@ -33,7 +33,7 @@ cd HFT/research/data/tick-crawler
 poetry install                       # tạo .venv/ + cài deps
 cp .env.example .env                 # điền CH_PASSWORD
 
-poetry run python migrate.py --recreate   # tạo 5 bảng tick mới (kèm CODEC nén)
+poetry run python migrate.py --recreate   # tạo 6 bảng tick mới (kèm CODEC nén)
 poetry run python symbol_info.py          # điền metadata symbol_info (futures universe)
 MARKET_TYPE=spot poetry run python symbol_info.py   # metadata cho spot (cặp lấy từ pairs.yaml)
 
@@ -81,13 +81,14 @@ Cả hai `network_mode: host` (tới ClickHouse `127.0.0.1`). `crawler-spot` ove
 Auto-pick `CH_HOST`: `127.0.0.1` → `192.168.122.226` (NAT) → `100.115.36.121` (Tailscale);
 trên datalake VM chọn `127.0.0.1` ngay. Client HTTP 8124 (`clickhouse-connect`). Creds trong `.env`.
 
-## Bảng (`migrate.py` tạo 5 bảng mới; `symbol_info.py`/live điền 4 bảng cũ)
+## Bảng (`migrate.py` tạo 6 bảng mới; `symbol_info.py`/live điền 4 bảng cũ)
 | Bảng | Loại | Nguồn |
 |---|---|---|
 | `trades` | mới | Vision `trades` + live WS `@aggTrade` (`/market`); `extra.src`=`vision`/`ws_agg`/`rest_agg` |
 | `book_depth` | mới | Vision `bookDepth` (±1..5% mỗi 5s) |
 | `book_snapshot_l2` | mới | live WS `@depthN@100ms` (L2 top-N, mặc định 20) |
 | `futures_metrics` | mới | Vision `metrics` + live REST `/futures/data/*` (long/short ratios, 5m) |
+| `liquidations` | mới | live WS `!forceOrder@arr` (thanh lý cưỡng bức toàn thị trường, **futures-only**, route `/market`) |
 | `ingest_state` | mới | registry idempotent cho backfill |
 | **`symbol_info`** | cũ | `exchangeInfo` (tick/step/precision/notional/status…) — 1 dòng/symbol |
 | **`ohlcv`** | cũ | live WS `@kline_<iv>` **mọi khung, chỉ nến đã đóng** (`is_final=1`) |
@@ -109,7 +110,9 @@ Giảm ~50–60% so với LZ4 mặc định (vd `trade_id` từ 2.0x lên ~rất
 - **OI + funding** (futures): 1 REST poller mỗi `OI_FUNDING_SECS` (premiumIndex all-symbols 1 call + openInterest/symbol).
 - **Metrics** (futures): 1 REST poller `/futures/data/*` mỗi 5 phút.
 
-Cờ tắt từng phần: `--no-orderbook --no-trades --no-klines --no-oi-funding --no-metrics`.
+- **Liquidations** (futures): 1 WS `!forceOrder@arr` (route `/market`) — 1 connection phủ toàn thị trường, ~1–2 event/s, bỏ qua universe (cascade chéo symbol chính là tín hiệu).
+
+Cờ tắt từng phần: `--no-orderbook --no-trades --no-klines --no-oi-funding --no-metrics --no-liquidations`.
 
 ### Xoay vòng WebSocket 24h (không mất data)
 `live/wsmanager.py` dùng **make-before-break**: mở connection thay thế ~15s trước mốc 24h, chạy

live/main.py

@@ -115,6 +115,23 @@ def _run_klines(symbols: list[str], seconds: int) -> None:
         proxy=config.WS_PROXY, base=config.WS_MARKET_BASE))
 
 
+def _run_liquidations(seconds: int) -> None:
+    """All-market forced liquidations `!forceOrder@arr` -> crypto.liquidations (futures only).
+
+    One connection on the /market route covers the WHOLE futures market (cheap,
+    ~1-2 events/s), so it ignores the per-symbol universe on purpose — cross-symbol
+    liquidation cascades are themselves the signal. The Public route returns nothing
+    for this stream.
+    """
+    print(f"[liquidations pid={os.getpid()}] !forceOrder@arr (all-market)", flush=True)
+    asyncio.run(_ws_loop(
+        ["_"], seconds,                              # one dummy symbol -> a single stream
+        streams_for=lambda _s: ["!forceOrder@arr"],
+        table="liquidations", cols=tables.LIQUIDATIONS_COLS,
+        build=lambda d, st: parsers.liquidation_rows(d, now_utc()),
+        proxy=config.WS_PROXY, base=config.WS_MARKET_BASE))
+
+
 def _chunk(xs: list, n: int) -> list[list]:
     n = max(1, min(n, len(xs)))
     return [xs[i::n] for i in range(n)]
@@ -131,6 +148,7 @@ def main() -> None:
     ap.add_argument("--no-metrics", action="store_true")
     ap.add_argument("--no-oi-funding", action="store_true")
     ap.add_argument("--no-orderbook", action="store_true")
+    ap.add_argument("--no-liquidations", action="store_true")
     ap.add_argument("--rest-trades", action="store_true",
                     help="collect trades via REST pagination instead of WS @aggTrade")
     args = ap.parse_args()
@@ -141,13 +159,15 @@ def main() -> None:
     ws_trades = config.WS_TRADES and not args.rest_trades and not args.no_trades
     run_metrics = not args.no_metrics and not is_spot          # long/short ratios: futures only
     run_oi_funding = not args.no_oi_funding and not is_spot    # OI + funding: futures only
+    run_liquidations = not args.no_liquidations and not is_spot  # forced liquidations: futures only
     ob_groups = [g for g in _chunk(syms, args.groups) if g]
     kl_groups = [g for g in _chunk(syms, args.kline_groups) if g]
     print(f"market={config.MARKET_TYPE}  host={os.environ['CH_HOST']}  symbols={len(syms)}  "
           f"ob_groups={len(ob_groups)}  kline_groups={len(kl_groups) if not args.no_klines else 0}"
           f"x{len(config.KLINE_INTERVALS)}iv  "
           f"trades={'ws' if ws_trades else ('rest' if not args.no_trades else 'off')}  "
           f"oi_funding={run_oi_funding}  metrics={run_metrics}  "
+          f"liquidations={run_liquidations}  "
           f"seconds={args.seconds or 'forever'}", flush=True)
 
     procs: list[mp.Process] = []
@@ -164,6 +184,8 @@ def main() -> None:
         procs.append(mp.Process(target=poll_oi_funding.run, args=(syms, args.seconds)))
     if run_metrics:
         procs.append(mp.Process(target=poll_metrics.run, args=(syms, args.seconds)))
+    if run_liquidations:                               # all-market !forceOrder@arr, /market route
+        procs.append(mp.Process(target=_run_liquidations, args=(args.seconds,)))
 
     for p in procs:
         p.start()

live/parsers.py

@@ -65,3 +65,19 @@ def kline_rows(data: dict, ingested: datetime) -> list[tuple]:
         dec(k["v"]), dec(k["q"]), int(k["n"]), dec(k["V"]), dec(k["Q"]),
         1, ms_to_dt(data["E"]), ingested, {},
     )]
+
+
+def liquidation_rows(data: dict, ingested: datetime) -> list[tuple]:
+    """`!forceOrder@arr` / `<symbol>@forceOrder` payload -> crypto.liquidations rows.
+
+    Futures-only forced-liquidation feed (delivered on the /market route, same as
+    @aggTrade — the Public route returns nothing). `o['S']` is the liquidation
+    order side: SELL = a LONG was force-closed, BUY = a SHORT was force-closed.
+    """
+    o = data["o"]
+    return [(
+        EXC, MKT, o["s"], o["S"].lower(), o["o"], o["f"],
+        dec(o["q"]), dec(o["p"]), dec(o["ap"]), o["X"],
+        dec(o["l"]), dec(o["z"]),
+        ms_to_dt(o["T"]), ms_to_dt(data["E"]), ingested, {},
+    )]

tables.py

@@ -31,6 +31,11 @@
 INGEST_STATE_COLS = [
     "dataset", "symbol", "date", "status", "rows", "bytes", "sha256", "updated_at",
 ]
+LIQUIDATIONS_COLS = [
+    "exchange", "market_type", "symbol", "side", "order_type", "time_in_force",
+    "orig_qty", "price", "avg_price", "status", "last_filled_qty", "filled_qty",
+    "trade_ts", "source_ts", "ingested_at", "extra",
+]
 # ---- insert column orders for the PRE-EXISTING tables we now populate ----
 OHLCV_COLS = [
     "exchange", "market_type", "symbol", "interval", "ts_open", "open", "high", "low",
@@ -153,5 +158,30 @@
 ENGINE = ReplacingMergeTree(updated_at)
 ORDER BY (dataset, symbol, date)
 SETTINGS index_granularity = 8192
+""",
+    "liquidations": """
+CREATE TABLE IF NOT EXISTS crypto.liquidations
+(
+    exchange        LowCardinality(String),
+    market_type     Enum8('spot'=1,'um'=2,'cm'=3),
+    symbol          LowCardinality(String),
+    side            Enum8('buy'=1,'sell'=2),
+    order_type      LowCardinality(String),
+    time_in_force   LowCardinality(String),
+    orig_qty        Decimal(38,18)    CODEC(ZSTD(1)),
+    price           Decimal(38,18)    CODEC(ZSTD(1)),
+    avg_price       Decimal(38,18)    CODEC(ZSTD(1)),
+    status          LowCardinality(String),
+    last_filled_qty Decimal(38,18)    CODEC(ZSTD(1)),
+    filled_qty      Decimal(38,18)    CODEC(ZSTD(1)),
+    trade_ts        DateTime64(3,'UTC') CODEC(DoubleDelta, ZSTD(1)),
+    source_ts       DateTime64(3,'UTC') CODEC(DoubleDelta, ZSTD(1)),
+    ingested_at     DateTime64(3,'UTC') CODEC(DoubleDelta, ZSTD(1)),
+    extra           Map(String,String)
+)
+ENGINE = ReplacingMergeTree(ingested_at)
+PARTITION BY toYYYYMMDD(trade_ts)
+ORDER BY (exchange, market_type, symbol, trade_ts, side, price, orig_qty)
+SETTINGS index_granularity = 8192
 """,
 }