""" HPLL-DataNode — Super Space Gộp HPLL-L4 (WebSocket Collector) + HPLL-Transformer (ETL Processor) vào 1 Space. Thu thập toàn bộ dữ liệu Hyperliquid real-time và push vào HF Dataset. """ import asyncio import websockets import json import pandas as pd import re import gc import os import sys import time import threading import requests from datetime import datetime from http.server import HTTPServer, BaseHTTPRequestHandler from huggingface_hub import HfApi, CommitOperationAdd, CommitOperationDelete, hf_hub_download # ═══════════════════════════════════════════════════════════════ # CONFIG # ═══════════════════════════════════════════════════════════════ HF_TOKEN = os.environ.get("HF_TOKEN") DATASET_ID = os.environ.get("HF_DATASET_ID", "gionuibk/hyperliquidL2Book-v2") DATA_DIR = "/data" CACHE_DIR = "/data/cache" SELF_URL = "https://gionuibk-hpll-datanode.hf.space" WS_URI = "wss://api.hyperliquid.xyz/ws" FLUSH_INTERVAL_S = 600 # flush raw data every 10 min (RAM -> SSD -> HF commit) ETL_INTERVAL_S = 3600 # run ETL every 60 min to reduce Hugging Face API rate limits os.makedirs(DATA_DIR, exist_ok=True) os.makedirs(CACHE_DIR, exist_ok=True) if not HF_TOKEN: print("ERROR: HF_TOKEN not set!", flush=True) sys.exit(1) api = HfApi(token=HF_TOKEN) # ═══════════════════════════════════════════════════════════════ # SHARED STATE # ═══════════════════════════════════════════════════════════════ BUFFER_LOCK = threading.Lock() GLOBAL_BUFFERS = { "l2book": [], "trades": [], "candles": [], "liquidations": [], "webdata": [], } STATS = { "start_time": time.time(), "last_ws_msg": time.time(), "l2book_msg": 0, "trades_msg": 0, "candles_msg": 0, "webdata_msg": 0, "liquidations_detected":0, "flush_count": 0, "etl_count": 0, "errors": 0, "ws_status": "Connecting", } def log(msg, tag="DataNode"): ts = datetime.now().strftime("%H:%M:%S") print(f"[{tag}][{ts}] {msg}", flush=True) # ═══════════════════════════════════════════════════════════════ # HEALTH SERVER # ═══════════════════════════════════════════════════════════════ class HealthHandler(BaseHTTPRequestHandler): def do_GET(self): uptime = int(time.time() - STATS["start_time"]) payload = { "service": "HPLL-DataNode", "version": "2.0-merged", "status": "running", "ws_status": STATS["ws_status"], "uptime_s": uptime, "l2book_msg": STATS["l2book_msg"], "trades_msg": STATS["trades_msg"], "candles_msg": STATS["candles_msg"], "webdata_msg": STATS["webdata_msg"], "liquidations":STATS["liquidations_detected"], "flush_count": STATS["flush_count"], "etl_count": STATS["etl_count"], "errors": STATS["errors"], } # Check if client wants JSON if self.path in ["/api", "/json"] or "application/json" in self.headers.get("Accept", ""): self.send_response(200) self.send_header("Content-Type", "application/json") self.end_headers() self.wfile.write(json.dumps(payload).encode()) return # Render styled HTML dashboard self.send_response(200) self.send_header("Content-Type", "text/html; charset=utf-8") self.end_headers() html = f""" Aetheris HPLL-DataNode

Aetheris HPLL-DataNode

WebSocket Data Collector running smoothly.

Uptime
{uptime//3600}h {(uptime%3600)//60}m
WS Connection
{payload['ws_status']}
L2 Book Messages
{payload['l2book_msg']:,}
Trades Messages
{payload['trades_msg']:,}
Candles Messages
{payload['candles_msg']:,}
Dataset Flushes
{payload['flush_count']}
Dataset Target: gionuibk/hyperliquidL2Book-v2
""" self.wfile.write(html.encode('utf-8')) def log_message(self, *args): pass def start_health_server(): server = HTTPServer(("0.0.0.0", 7860), HealthHandler) log("Health server started on :7860") server.serve_forever() threading.Thread(target=start_health_server, daemon=True).start() # ═══════════════════════════════════════════════════════════════ # SELF KEEPALIVE (ping every 90s để HF không sleep Space) # ═══════════════════════════════════════════════════════════════ def start_keepalive(): log("Keepalive started (90s interval)") while True: time.sleep(90) try: r = requests.get(SELF_URL, timeout=15) log(f"🏓 Self-ping: HTTP {r.status_code}") except Exception as e: log(f"⚠️ Self-ping fail: {e}") threading.Thread(target=start_keepalive, daemon=True).start() # ═══════════════════════════════════════════════════════════════ # WATCHDOG — restart container nếu không có data 3 phút # ═══════════════════════════════════════════════════════════════ def start_watchdog(): log("Watchdog started (timeout=180s)") while True: time.sleep(60) silence = time.time() - STATS["last_ws_msg"] if silence > 180: log(f"CRITICAL: No WS data for {int(silence)}s — exiting to force restart") sys.exit(1) uptime = int(time.time() - STATS["start_time"]) log(f"❤️ Uptime: {uptime//3600}h{(uptime%3600)//60}m | WS: {STATS['ws_status']} | " f"L2:{STATS['l2book_msg']} T:{STATS['trades_msg']} C:{STATS['candles_msg']}") threading.Thread(target=start_watchdog, daemon=True).start() # ═══════════════════════════════════════════════════════════════ # MODULE 1 — COLLECTOR (WebSocket → Buffer) # ═══════════════════════════════════════════════════════════════ def get_all_coins(): try: resp = requests.post( "https://api.hyperliquid.xyz/info", headers={"Content-Type": "application/json"}, json={"type": "meta"}, timeout=15 ) perps = [a["name"] for a in resp.json().get("universe", [])] log(f"Universe: {len(perps)} perp coins") return perps except Exception as e: log(f"Failed to fetch coin list: {e} — using fallback") return ["BTC", "ETH", "SOL", "DOGE", "AVAX", "XRP", "LINK", "ARB", "OP", "SEI"] async def drain_messages(ws, timeout=0.05): drained = 0 while True: try: msg_txt = await asyncio.wait_for(ws.recv(), timeout=timeout) STATS["last_ws_msg"] = time.time() _process_message(msg_txt) drained += 1 except asyncio.TimeoutError: break return drained def _process_message(msg_txt): try: msg = json.loads(msg_txt) channel = msg.get("channel") data = msg.get("data") if not data: return now = datetime.now() coin = None buf = None if channel == "l2Book": coin = data.get("coin") buf = "l2book" STATS["l2book_msg"] += 1 elif channel == "trades": if isinstance(data, list) and data: coin = data[0].get("coin") # Detect liquidations for t in data: if t.get("liquidation") or "liq" in str(t.get("users", [])).lower(): with BUFFER_LOCK: GLOBAL_BUFFERS["liquidations"].append( {"timestamp": now, "coin": coin, "payload": msg_txt} ) STATS["liquidations_detected"] += 1 break buf = "trades" STATS["trades_msg"] += 1 elif channel == "candle": coin = data.get("s") buf = "candles" STATS["candles_msg"] += 1 elif channel == "webData2": coin = "ALL" buf = "webdata" STATS["webdata_msg"] += 1 if buf and coin: with BUFFER_LOCK: GLOBAL_BUFFERS[buf].append( {"timestamp": now, "coin": coin, "payload": msg_txt} ) except Exception: pass async def subscribe_batch(ws, coins): await ws.send(json.dumps({ "method": "subscribe", "subscription": {"type": "webData2", "user": "0x0000000000000000000000000000000000000000"} })) CHUNK = 10 for i in range(0, len(coins), CHUNK): batch = coins[i:i+CHUNK] for coin in batch: await ws.send(json.dumps({"method": "subscribe", "subscription": {"type": "l2Book", "coin": coin}})) await ws.send(json.dumps({"method": "subscribe", "subscription": {"type": "trades", "coin": coin}})) await ws.send(json.dumps({"method": "subscribe", "subscription": {"type": "candle", "coin": coin, "interval": "1m"}})) drained = await drain_messages(ws) log(f"Subscribed {min(i+CHUNK, len(coins))}/{len(coins)} coins (drained {drained})") await asyncio.sleep(0.3) async def collector_loop(): while True: try: coins = get_all_coins() STATS["ws_status"] = "Connecting" async with websockets.connect(WS_URI, ping_interval=30, ping_timeout=60, max_size=None) as ws: STATS["ws_status"] = "Connected" await subscribe_batch(ws, coins) log(f"✅ Collecting from {len(coins)} coins...") while True: msg_txt = await ws.recv() STATS["last_ws_msg"] = time.time() _process_message(msg_txt) except Exception as e: STATS["ws_status"] = "Disconnected" STATS["errors"] += 1 log(f"WS error: {e} — reconnecting in 5s") await asyncio.sleep(5) # ═══════════════════════════════════════════════════════════════ # MODULE 2 — RAW FLUSH (Buffer → HF Dataset raw parquet) # ═══════════════════════════════════════════════════════════════ def _make_commit_op(data, dtype): if not data: return None try: ts = int(datetime.now().timestamp()) now = datetime.now() year = now.strftime("%Y") month = now.strftime("%m") day = now.strftime("%d") folder = f"l4_{dtype}/year={year}/month={month}/day={day}" fname = f"l4_{dtype}_{ts}.parquet" local = os.path.join(DATA_DIR, fname) pd.DataFrame(data).to_parquet(local, compression="snappy", index=False) with open(local, "rb") as f: content = f.read() os.remove(local) return CommitOperationAdd(path_in_repo=f"data/{folder}/{fname}", path_or_fileobj=content) except Exception as e: log(f"Prepare op failed ({dtype}): {e}") STATS["errors"] += 1 return None async def flush_loop(): """Flush raw buffers to HF dataset every FLUSH_INTERVAL_S seconds.""" while True: await asyncio.sleep(FLUSH_INTERVAL_S) ops = [] labels = [] with BUFFER_LOCK: for dtype, buf in GLOBAL_BUFFERS.items(): if buf: chunk = list(buf) buf.clear() op = _make_commit_op(chunk, dtype) if op: ops.append(op) labels.append(f"{dtype}:{len(chunk)}") if ops: STATS["flush_count"] += 1 log(f"═══ FLUSH #{STATS['flush_count']} | {', '.join(labels)}") try: await asyncio.to_thread( api.create_commit, repo_id=DATASET_ID, repo_type="dataset", operations=ops, commit_message=f"Auto: Raw flush #{STATS['flush_count']}" ) log("Flush committed ✅") except Exception as e: log(f"Flush commit failed: {e}") if "429" in str(e): await asyncio.sleep(600) # ═══════════════════════════════════════════════════════════════ # MODULE 3 — ETL (Raw parquet → Parsed parquet) # ═══════════════════════════════════════════════════════════════ def _etl_channel(channel, raw_folder, target_folder, all_files): ops = [] raw_files = sorted([f for f in all_files if f.startswith(f"data/{raw_folder}/") and f.endswith(".parquet")]) if not raw_files: return ops # Watermark done = [f for f in all_files if f.startswith(f"data/{target_folder}/") and f.endswith(".parquet")] pat = re.compile(rf"{channel}_(\d+)_(\d+)\.parquet") max_ts = 0.0 for f in done: m = pat.match(f.split("/")[-1]) if m: t = int(m.group(2)) if t > max_ts: max_ts = float(t) new_files = [] for f in raw_files: try: ts = int(f.split("/")[-1].replace(".parquet","").split("_")[-1]) if ts > max_ts: new_files.append(f) except: continue if not new_files: return ops batch = new_files[:3] log(f"ETL [{channel}]: {len(new_files)} new files, processing {len(batch)}", "ETL") records = [] for fp in batch: local = None try: local = hf_hub_download(repo_id=DATASET_ID, filename=fp, repo_type="dataset", local_dir=CACHE_DIR, token=HF_TOKEN) df = pd.read_parquet(local) for _, row in df.iterrows(): try: payload = json.loads(row["payload"]) d = payload.get("data") if channel == "l2book" and d: coin = d.get("coin") ts = d.get("time") lvls = d.get("levels") if lvls and len(lvls) == 2: rec = {"coin": coin, "timestamp": datetime.fromtimestamp(ts/1000) if ts else row["timestamp"], "server_time": ts} for i in range(20): bids, asks = lvls[0], lvls[1] rec[f"bid_px_{i+1}"] = float(bids[i]["px"]) if i < len(bids) else 0.0 rec[f"bid_sz_{i+1}"] = float(bids[i]["sz"]) if i < len(bids) else 0.0 rec[f"ask_px_{i+1}"] = float(asks[i]["px"]) if i < len(asks) else 0.0 rec[f"ask_sz_{i+1}"] = float(asks[i]["sz"]) if i < len(asks) else 0.0 records.append(rec) elif channel == "trades" and isinstance(d, list): for t in d: t["timestamp"] = datetime.fromtimestamp(t.get("time", 0)/1000) records.append(t) elif channel == "candles" and d: records.append({ "coin": d.get("s"), "timestamp": datetime.fromtimestamp(d.get("t",0)/1000), "open": float(d.get("o",0)), "high": float(d.get("h",0)), "low": float(d.get("l",0)), "close": float(d.get("c",0)), "volume": float(d.get("v",0)), "num_trades": d.get("n",0) }) except: continue del df; gc.collect() except Exception as e: log(f"ETL file error {fp}: {e}", "ETL") finally: if local and os.path.exists(local): os.remove(local) if records: tdf = pd.DataFrame(records) mn_ts = int(tdf["timestamp"].min().timestamp()) mx_ts = int(tdf["timestamp"].max().timestamp()) now = datetime.now() year = now.strftime("%Y") month = now.strftime("%m") day = now.strftime("%d") folder = f"{target_folder}/year={year}/month={month}/day={day}" fname = f"{channel}_{mn_ts}_{mx_ts}.parquet" lpath = os.path.join(CACHE_DIR, fname) tdf.to_parquet(lpath, index=False, compression="snappy") with open(lpath, "rb") as f: ops.append(CommitOperationAdd(path_in_repo=f"data/{folder}/{fname}", path_or_fileobj=f.read())) os.remove(lpath) del records, tdf; gc.collect() return ops async def etl_loop(): """Run ETL every ETL_INTERVAL_S seconds.""" await asyncio.sleep(60) # Wait 1 min after start before first ETL while True: try: log("Starting ETL cycle...", "ETL") all_files = await asyncio.to_thread( lambda: list(api.list_repo_files(repo_id=DATASET_ID, repo_type="dataset")) ) all_ops = [] for ch, raw_f, tgt_f in [ ("l2book", "l4_l2book", "l2book"), ("trades", "l4_trades", "trades"), ("candles", "l4_candles", "candles"), ]: ops = await asyncio.to_thread(_etl_channel, ch, raw_f, tgt_f, all_files) all_ops.extend(ops) gc.collect() if all_ops: STATS["etl_count"] += 1 log(f"ETL commit: {len(all_ops)} ops", "ETL") await asyncio.to_thread( api.create_commit, repo_id=DATASET_ID, repo_type="dataset", operations=all_ops, commit_message=f"ETL: Batch #{STATS['etl_count']}" ) log("ETL committed ✅", "ETL") else: log("ETL: No new data.", "ETL") except Exception as e: log(f"ETL cycle error: {e}", "ETL") if "429" in str(e): await asyncio.sleep(600) await asyncio.sleep(ETL_INTERVAL_S) # ═══════════════════════════════════════════════════════════════ # MAIN # ═══════════════════════════════════════════════════════════════ async def main(): log("🚀 HPLL-DataNode v2.0 (Collector only) starting...") threading.Thread(target=start_watchdog, daemon=True).start() await asyncio.gather( collector_loop(), flush_loop(), ) if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: pass