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: Controls how fast "relevance" drops off with distance. # 100 means an order $100 away has ~36% weight. 50 is tighter (scalping), 200 is wider (swing). DECAY_LAMBDA = 100.0 # IMPACT_SENSITIVITY: Converts the weighted volume score into Price Impact ($). # Multiplier for the Square Root Law. 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} "liq_history": [], # Liquidity Trend history: {t, v} "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. Input: diff_x: List of distances from mid ($). diff_y: List of CUMULATIVE Net Liquidity (Bids - Asks). """ 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] # Unpack cumulative volume to get marginal volume at this step marginal_vol = cum_vol - prev_vol prev_vol = cum_vol # Apply Exponential Decay (Spatial Damping) # Orders close to spread (dist=0) have weight 1.0 # Orders far away decay towards 0.0 weight = math.exp(-dist / DECAY_LAMBDA) weighted_imbalance += marginal_vol * weight # 2. Calculate Market Impact (Square Root Law) # Impact is not linear; it follows a square root function of volume. 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/Resistance Scans) # We still use the raw curve to find "Walls" 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] # Resistance: Net Liquidity flips from + to - (Buyer exhaustion / Seller Wall) if prev_val > 0 and curr_val < 0 and resistance_level is None: resistance_level = current_mid + dist # Support: Net Liquidity flips from - to + (Seller exhaustion / Buyer Wall) 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 # Sending the decay-weighted score } def process_market_data(): if not market_state['ready']: return {"error": "Initializing..."} mid = market_state['current_mid'] # Snapshot Top 300 for Depth Chart 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) # Calculate Net Liquidity Curve (Depth) # We interpolate to ensure bids and asks are compared at the exact same distances diff_x, diff_y = [], [] if d_b_x and d_a_x: max_dist = min(d_b_x[-1], d_a_x[-1]) # Resolution: 100 steps across the available depth step_size = max_dist / 100 steps = [i * step_size for i in range(1, 101)] for s in steps: # Find cumulative bid vol 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 # Find cumulative ask vol 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) # Cumulative Net Imbalance analysis = analyze_structure(diff_x, diff_y, mid) # Store Liquidity Trend for history now = time.time() if analysis: # Update Trend History if needed (throttle slightly to match graph res) if not market_state['liq_history'] or (now - market_state['liq_history'][-1]['t'] > 0.5): market_state['liq_history'].append({'t': now, 'v': analysis['net_score']}) if len(market_state['liq_history']) > HISTORY_LENGTH: market_state['liq_history'].pop(0) return { "mid": mid, "history": market_state['history'], # Price History "liq_history": market_state['liq_history'], # Net Liq History "diff": { "x": diff_x, "y": diff_y }, # Depth Snapshot "analysis": analysis } # --- HTML Frontend --- HTML_PAGE = f""" HFT Liquidity Dashboard | {SYMBOL_KRAKEN}
ESTABLISHING UPLINK...
Connecting to WebSocket Stream...
BTC/USD Price Action---
HFT Weighted Imbalance (Decay {DECAY_LAMBDA})0.0
Net Liquidity StructureRange: $100
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