"""Liquidation alert bot (Python, standard library + the blockmetric SDK). Alerts on big liquidations and liquidation cascades as they happen, and warns once per cluster when the price comes close to a large liquidation cluster of the heatmap. Events arrive the way your plan allows: BOT_MODE=stream live stream (API Pro), reconnects with Last-Event-ID BOT_MODE=webhook signed webhooks pushed to this process (API Dev, API Pro), verifies Blockmetric-Signature BOT_MODE=poll polls new liquidations (every key, also the free one) BOT_MODE=auto stream, falling back to poll without API Pro (default) BLOCKMETRIC_API_KEY=bm_live_... BLOCKMETRIC_BASE_URL=https:///api/v1 python bot.py """ import json import os import signal import sys import threading import urllib.request from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer from typing import Any, Callable, Dict, List, Mapping, Optional from blockmetric import OPERATIONS, Blockmetric, BlockmetricError, construct_event, heatmap_clusters def config(env: Mapping[str, str] = os.environ) -> Dict[str, Any]: def num(name: str, default: float) -> float: v = env.get(name) return float(v) if v not in (None, "") else default return { "mode": env.get("BOT_MODE", "auto"), "coins": [c.strip() for c in env.get("BOT_COINS", "BTC,ETH").split(",") if c.strip()], "kinds": env.get("BOT_KINDS", "big_liquidation,liq_cascade,whale_liquidated"), "min_severity": int(num("BOT_MIN_SEVERITY", 2)), "min_usd": num("BOT_MIN_USD", 250_000), "cluster_min_usd": num("BOT_CLUSTER_MIN_USD", 5_000_000), "cluster_distance": num("BOT_CLUSTER_DISTANCE", 0.01), "cluster_minutes": num("BOT_CLUSTER_MINUTES", 5), "poll_seconds": num("POLL_SECONDS", 15), "port": int(num("PORT", 3000)), "hook_path": env.get("HOOK_PATH", "/hook"), "secret": env.get("BLOCKMETRIC_WEBHOOK_SECRET"), "alert_url": env.get("ALERT_WEBHOOK_URL") or None, "exit_after": int(num("BOT_EXIT_AFTER", 0)), } def usd(n: Optional[float]) -> str: if n is None: return "-" a = abs(n) return f"${a / 1e9:.2f}B" if a >= 1e9 else f"${a / 1e6:.2f}M" if a >= 1e6 else f"${a / 1e3:.1f}k" if a >= 1e3 else f"${a:.0f}" def format_event(ev: Mapping[str, Any]) -> str: d = ev["data"] return f"[{d['kind']}] sev {d['severity']}{' ' + d['coin'] if d.get('coin') else ''}: {d['title']}" def format_liquidation(liq: Mapping[str, Any]) -> str: return f"[liquidation] {liq['coin']} {liq.get('side') or ''} {usd(liq.get('usd'))} on {liq['venue']} at {liq.get('price')} ({liq['ts']})" def near_clusters(heatmap: Mapping[str, Any], min_usd: float, max_distance: float) -> List[Dict[str, Any]]: """Clusters of the newest heatmap column that are large and close to the mark.""" c = heatmap_clusters(heatmap, top=10) if not c or not c["mark"]: return [] return [x for x in c["above"] + c["below"] if x["usd"] >= min_usd and x["distance"] is not None and abs(x["distance"]) <= max_distance] class Bot: def __init__(self, cfg: Dict[str, Any], client: Optional[Blockmetric] = None, out: Callable[[str], None] = print) -> None: self.cfg = cfg self.client = client or Blockmetric(user_agent="liquidation-alert-bot/1.0") self.out = out self.stop = threading.Event() self.coins = {c.upper() for c in cfg["coins"]} self._seen: set = set() self._sent = 0 self._lock = threading.Lock() # ---- alerts --------------------------------------------------------------------------------------------------- def alert(self, key: str, text: str) -> None: with self._lock: if key in self._seen or self.stop.is_set(): return self._seen.add(key) self.out(text) if self.cfg["alert_url"]: # Slack-compatible body; any endpoint that takes JSON works. body = json.dumps({"text": text, "content": text}).encode() req = urllib.request.Request(self.cfg["alert_url"], data=body, headers={"Content-Type": "application/json"}, method="POST") try: urllib.request.urlopen(req, timeout=10).close() # noqa: S310 - URL from your own config except OSError as e: self.out(f"alert delivery failed: {e}") self._sent += 1 if self.cfg["exit_after"] and self._sent >= self.cfg["exit_after"]: self.stop.set() def on_event(self, ev: Mapping[str, Any]) -> None: coin = ev["data"].get("coin") if coin and self.coins and coin.upper() not in self.coins: return self.alert(ev["id"], format_event(ev)) # ---- heatmap clusters ----------------------------------------------------------------------------------------- def check_clusters(self) -> None: for coin in self.cfg["coins"]: try: data = self.client.heatmap(coin, tf="15m", hours=6)["data"] except BlockmetricError as e: self.out(f"heatmap {coin}: {e}") continue for x in near_clusters(data, self.cfg["cluster_min_usd"], self.cfg["cluster_distance"]): side = "shorts" if x["side"] == "short" else "longs" self.alert(f"cluster:{coin}:{x['price']}", f"[cluster] {data['coin']}: {usd(x['usd'])} of {side} liquidate at {x['price']} ({x['distance'] * 100:.2f} % from the mark)") def _cluster_loop(self) -> None: while not self.stop.is_set(): self.check_clusters() if self.stop.wait(self.cfg["cluster_minutes"] * 60): return # ---- event sources -------------------------------------------------------------------------------------------- def poll(self) -> None: params: Dict[str, Any] = {"limit": 200} if "minUsd" in OPERATIONS["liquidations"]["query"]: params["min_usd"] = self.cfg["min_usd"] # filtered on the server where the API knows minUsd self.out(f"polling new liquidations >= {usd(self.cfg['min_usd'])} every {self.cfg['poll_seconds']:g}s") on_error = lambda e: self.out(f"poll failed: {e}") # noqa: E731 for liq in self.client.follow("liquidations", interval=self.cfg["poll_seconds"], stop_event=self.stop, on_error=on_error, **params): if (liq.get("usd") or 0) < self.cfg["min_usd"] or (self.coins and liq["coin"].upper() not in self.coins): continue self.alert(f"liq:{liq['id']}", format_liquidation(liq)) def stream(self) -> None: for ev in self.client.stream( kinds=self.cfg["kinds"], coin=self.cfg["coins"], min_severity=self.cfg["min_severity"], stop_event=self.stop, on_ready=lambda _: self.out("connected to the live stream"), on_reconnect=lambda r: self.out(f"reconnecting in {r['delay']:.0f}s ({r['reason']})"), ): self.on_event(ev) def webhook(self) -> None: secret = self.cfg["secret"] if not secret: raise SystemExit("BOT_MODE=webhook needs BLOCKMETRIC_WEBHOOK_SECRET (whsec_...) - the signing secret of your endpoint on /account/api.") bot = self kinds = set(self.cfg["kinds"].split(",")) class Hook(BaseHTTPRequestHandler): def log_message(self, *args: Any) -> None: # quiet pass def do_POST(self) -> None: # noqa: N802 if self.path.split("?")[0] != bot.cfg["hook_path"]: self.send_response(404) self.end_headers() return raw = self.rfile.read(int(self.headers.get("Content-Length", 0))) # verify these exact bytes try: ev = construct_event(raw, self.headers.get("Blockmetric-Signature"), secret) except BlockmetricError: self.send_response(400) self.end_headers() return self.send_response(204) # answer first (within 10 s), then work self.end_headers() if ev["type"] == "test" or (ev["type"] in kinds and ev["data"]["severity"] >= bot.cfg["min_severity"]): bot.on_event(ev) server = ThreadingHTTPServer(("", self.cfg["port"]), Hook) self.out(f"webhook receiver listening on http://localhost:{server.server_address[1]}{self.cfg['hook_path']}") thread = threading.Thread(target=server.serve_forever, kwargs={"poll_interval": 0.2}, daemon=True) thread.start() self.stop.wait() server.shutdown() server.server_close() def run(self) -> None: if self.cfg["cluster_minutes"] > 0: threading.Thread(target=self._cluster_loop, daemon=True).start() mode = self.cfg["mode"] if mode == "webhook": self.webhook() elif mode == "poll": self.poll() elif mode == "stream": self.stream() else: try: self.stream() except BlockmetricError as e: if e.code not in ("plan_required", "scope_required"): raise self.out("the live stream needs API Pro, falling back to polling") self.poll() def main() -> int: bot = Bot(config()) signal.signal(signal.SIGINT, lambda *_: bot.stop.set()) try: bot.run() except BlockmetricError as e: print(f"error: {e}" + (f" - see {e.doc_url}" if e.doc_url else ""), file=sys.stderr) return 1 return 0 if __name__ == "__main__": sys.exit(main())