HFT Liquidity Dashboard | {SYMBOL

import asyncio import json import logging import time import bisect import math from aiohttp import web import websockets # --- Configuration --- SYMBOL_KRAKEN = "BTC/USD" PORT = 7860 HISTORY_LENGTH = 300 BROADCAST_RATE = 0.1 # 10Hz updates # --- HFT Damping Configuration --- DECAY_LAMBDA = 100.0 IMPACT_SENSITIVITY = 0.5 # --- Logging --- logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s') # --- In-Memory State --- market_state = { "bids": {}, "asks": {}, "history": [], # Price history: {t, p} "current_mid": 0.0, "prev_mid": 0.0, "ready": False } connected_clients = set() # --- AI Logic Helper (HFT Version) --- def analyze_structure(diff_x, diff_y, current_mid): """ Applies HFT Spatial Decay and Square Root Market Impact models. """ if not diff_y or len(diff_y) < 5: return None weighted_imbalance = 0.0 prev_vol = 0.0 # 1. Calculate Spatial Weighted Imbalance for i in range(len(diff_x)): dist = diff_x[i] cum_vol = diff_y[i] # diff_y is the Net (Bid - Ask) marginal_vol = cum_vol - prev_vol prev_vol = cum_vol weight = math.exp(-dist / DECAY_LAMBDA) weighted_imbalance += marginal_vol * weight # 2. Market Impact if weighted_imbalance != 0: impact = math.sqrt(abs(weighted_imbalance)) * IMPACT_SENSITIVITY if weighted_imbalance < 0: impact = -impact else: impact = 0.0 projected_price = current_mid + impact # 3. Structural Reversals support_level = None resistance_level = None scan_limit = len(diff_y) // 2 for i in range(1, scan_limit): prev_val = diff_y[i-1] curr_val = diff_y[i] dist = diff_x[i] if prev_val > 0 and curr_val < 0 and resistance_level is None: resistance_level = current_mid + dist if prev_val < 0 and curr_val > 0 and support_level is None: support_level = current_mid - dist return { "projected": projected_price, "support": support_level, "resistance": resistance_level, "net_score": weighted_imbalance } def process_market_data(): if not market_state['ready']: return {"error": "Initializing..."} mid = market_state['current_mid'] # Snapshot Top 300 raw_bids = sorted(market_state['bids'].items(), key=lambda x: -x[0])[:300] raw_asks = sorted(market_state['asks'].items(), key=lambda x: x[0])[:300] # Calculate Cumulative Volume d_b_x, d_b_y, cum = [], [], 0 for p, q in raw_bids: d = mid - p if d >= 0: cum += q d_b_x.append(d); d_b_y.append(cum) d_a_x, d_a_y, cum = [], [], 0 for p, q in raw_asks: d = p - mid if d >= 0: cum += q d_a_x.append(d); d_a_y.append(cum) # Interpolate for Charts diff_x, diff_y = [], [] chart_bids, chart_asks = [], [] # Separated arrays for the raw depth chart if d_b_x and d_a_x: max_dist = min(d_b_x[-1], d_a_x[-1]) step_size = max_dist / 100 steps = [i * step_size for i in range(1, 101)] for s in steps: # Interpolate Bid Volume at distance s idx_b = bisect.bisect_right(d_b_x, s) vol_b = d_b_y[idx_b-1] if idx_b > 0 else 0 # Interpolate Ask Volume at distance s idx_a = bisect.bisect_right(d_a_x, s) vol_a = d_a_y[idx_a-1] if idx_a > 0 else 0 diff_x.append(s) diff_y.append(vol_b - vol_a) # Net chart_bids.append(vol_b) # Raw Bid chart_asks.append(vol_a) # Raw Ask analysis = analyze_structure(diff_x, diff_y, mid) return { "mid": mid, "history": market_state['history'], "depth_x": diff_x, "depth_net": diff_y, "depth_bids": chart_bids, "depth_asks": chart_asks, "analysis": analysis } # --- HTML Frontend --- HTML_PAGE = f""" HFT Liquidity Dashboard | {SYMBOL_KRAKEN}

ESTABLISHING UPLINK...

Connecting to WebSocket Stream...

BTC/USD Price Action---

Market Depth (Bids vs Asks)

Net Delta (Bids - Asks)

HFT ANALYTICS ENGINE

WEIGHTED IMBALANCE SCORE 0

MARKET STRUCTURE

RESIST:---

SUPPORT:---

IMPACT PROJECTION ---

> ALGO LOGS

""" async def kraken_worker(): global market_state while True: try: async with websockets.connect("wss://ws.kraken.com/v2") as ws: logging.info(f"🔌 Connected to Kraken ({SYMBOL_KRAKEN})") await ws.send(json.dumps({ "method": "subscribe", "params": {"channel": "book", "symbol": [SYMBOL_KRAKEN], "depth": 500} })) async for message in ws: payload = json.loads(message) channel = payload.get("channel") data = payload.get("data", []) if channel == "book": for item in data: for bid in item.get('bids', []): q, p = float(bid['qty']), float(bid['price']) if q == 0: market_state['bids'].pop(p, None) else: market_state['bids'][p] = q for ask in item.get('asks', []): q, p = float(ask['qty']), float(ask['price']) if q == 0: market_state['asks'].pop(p, None) else: market_state['asks'][p] = q if market_state['bids'] and market_state['asks']: best_bid = max(market_state['bids'].keys()) best_ask = min(market_state['asks'].keys()) mid = (best_bid + best_ask) / 2 market_state['prev_mid'] = market_state['current_mid'] market_state['current_mid'] = mid market_state['ready'] = True now = time.time() if not market_state['history'] or (now - market_state['history'][-1]['t'] > 0.5): market_state['history'].append({'t': now, 'p': mid}) if len(market_state['history']) > HISTORY_LENGTH: market_state['history'].pop(0) except Exception as e: logging.warning(f"⚠️ Reconnecting: {e}") await asyncio.sleep(3) async def broadcast_worker(): while True: if connected_clients and market_state['ready']: payload = process_market_data() msg = json.dumps(payload) for ws in list(connected_clients): try: await ws.send_str(msg) except: pass await asyncio.sleep(BROADCAST_RATE) async def websocket_handler(request): ws = web.WebSocketResponse() await ws.prepare(request) connected_clients.add(ws) try: async for msg in ws: pass finally: connected_clients.remove(ws) return ws async def handle_index(request): return web.Response(text=HTML_PAGE, content_type='text/html') async def start_background(app): app['kraken_task'] = asyncio.create_task(kraken_worker()) app['broadcast_task'] = asyncio.create_task(broadcast_worker()) async def cleanup_background(app): app['kraken_task'].cancel() app['broadcast_task'].cancel() try: await app['kraken_task']; await app['broadcast_task'] except: pass async def main(): app = web.Application() app.router.add_get('/', handle_index) app.router.add_get('/ws', websocket_handler) app.on_startup.append(start_background) app.on_cleanup.append(cleanup_background) runner = web.AppRunner(app) await runner.setup() site = web.TCPSite(runner, '0.0.0.0', PORT) await site.start() print(f"🚀 AI Dashboard: http://localhost:{PORT}") await asyncio.Event().wait() if __name__ == "__main__": try: asyncio.run(main()) except KeyboardInterrupt: pass