Spaces:
Running
Running
| """ | |
| 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"""<!DOCTYPE html> | |
| <html> | |
| <head> | |
| <title>Aetheris HPLL-DataNode</title> | |
| <meta name="viewport" content="width=device-width, initial-scale=1.0"> | |
| <style> | |
| body {{ | |
| background: #0f172a; | |
| color: #f8fafc; | |
| font-family: -apple-system, BlinkMacSystemFont, "Segoe UI", Roboto, Helvetica, Arial, sans-serif; | |
| margin: 0; | |
| padding: 2rem; | |
| display: flex; | |
| justify-content: center; | |
| align-items: center; | |
| min-height: 80vh; | |
| }} | |
| .card {{ | |
| background: rgba(30, 41, 59, 0.7); | |
| border: 1px solid rgba(255, 255, 255, 0.1); | |
| backdrop-filter: blur(16px); | |
| border-radius: 16px; | |
| padding: 2rem; | |
| width: 100%; | |
| max-width: 500px; | |
| box-shadow: 0 10px 25px -5px rgba(0, 0, 0, 0.3), 0 8px 10px -6px rgba(0, 0, 0, 0.3); | |
| }} | |
| h1 {{ | |
| font-size: 1.5rem; | |
| margin-top: 0; | |
| margin-bottom: 0.5rem; | |
| display: flex; | |
| align-items: center; | |
| gap: 0.5rem; | |
| }} | |
| .status-badge {{ | |
| display: inline-block; | |
| width: 10px; | |
| height: 10px; | |
| border-radius: 50%; | |
| background: #22c55e; | |
| box-shadow: 0 0 8px #22c55e; | |
| }} | |
| .metrics-grid {{ | |
| display: grid; | |
| grid-template-columns: 1fr 1fr; | |
| gap: 1rem; | |
| margin: 1.5rem 0; | |
| }} | |
| .metric {{ | |
| background: rgba(15, 23, 42, 0.5); | |
| padding: 1rem; | |
| border-radius: 8px; | |
| border: 1px solid rgba(255, 255, 255, 0.05); | |
| }} | |
| .metric-label {{ | |
| font-size: 0.75rem; | |
| color: #94a3b8; | |
| text-transform: uppercase; | |
| }} | |
| .metric-value {{ | |
| font-size: 1.25rem; | |
| font-weight: 700; | |
| color: #38bdf8; | |
| margin-top: 0.25rem; | |
| }} | |
| .footer {{ | |
| font-size: 0.75rem; | |
| color: #64748b; | |
| text-align: center; | |
| margin-top: 1.5rem; | |
| }} | |
| </style> | |
| </head> | |
| <body> | |
| <div class="card"> | |
| <h1> | |
| <span class="status-badge"></span> | |
| Aetheris HPLL-DataNode | |
| </h1> | |
| <p style="color: #94a3b8; font-size: 0.875rem; margin-bottom: 1.5rem;">WebSocket Data Collector running smoothly.</p> | |
| <div class="metrics-grid"> | |
| <div class="metric"> | |
| <div class="metric-label">Uptime</div> | |
| <div class="metric-value">{uptime//3600}h {(uptime%3600)//60}m</div> | |
| </div> | |
| <div class="metric"> | |
| <div class="metric-label">WS Connection</div> | |
| <div class="metric-value" style="color: #22c55e;">{payload['ws_status']}</div> | |
| </div> | |
| <div class="metric"> | |
| <div class="metric-label">L2 Book Messages</div> | |
| <div class="metric-value">{payload['l2book_msg']:,}</div> | |
| </div> | |
| <div class="metric"> | |
| <div class="metric-label">Trades Messages</div> | |
| <div class="metric-value">{payload['trades_msg']:,}</div> | |
| </div> | |
| <div class="metric"> | |
| <div class="metric-label">Candles Messages</div> | |
| <div class="metric-value">{payload['candles_msg']:,}</div> | |
| </div> | |
| <div class="metric"> | |
| <div class="metric-label">Dataset Flushes</div> | |
| <div class="metric-value">{payload['flush_count']}</div> | |
| </div> | |
| </div> | |
| <div class="footer"> | |
| Dataset Target: gionuibk/hyperliquidL2Book-v2 | |
| </div> | |
| </div> | |
| </body> | |
| </html>""" | |
| 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 | |