diff --git a/README.md b/README.md index 7a354b5..fa8675b 100644 --- a/README.md +++ b/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_` **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 diff --git a/live/main.py b/live/main.py index 06cf4f5..09aff4b 100644 --- a/live/main.py +++ b/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,6 +159,7 @@ 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)} " @@ -148,6 +167,7 @@ def main() -> None: 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() diff --git a/live/parsers.py b/live/parsers.py index b59840b..4d9a284 100644 --- a/live/parsers.py +++ b/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` / `@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, {}, + )] diff --git a/tables.py b/tables.py index 2d82032..c334ee6 100644 --- a/tables.py +++ b/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 """, }