#!/usr/bin/env python # -*- coding: utf-8 -*- """ 策略B · 清算冲击可视化看板 v3 HTML模板在 templates/dashboard.html 访问: http://localhost:5050 """ import sys, os, json, time, ssl, threading from collections import deque from datetime import datetime import websocket sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) import config from flask import Flask, jsonify, send_file from flask_cors import CORS sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'exchange')) from liquidation import LiquidationTracker SYMBOL, PORT = "BTC-USDT-SWAP", 5050 P1_ENTER, P1_WATCH, ADV_ENTER = 0.80, 0.55, 1.0 # ── 全局数据 ── class Store: def __init__(self): self.mid=0.0; self.bids={}; self.asks={} self.bid1p=0.0; self.ask1p=0.0; self.bid01p=0.0; self.ask01p=0.0; self.imb=0.0 self.t200=0; self.avg5=0; self.lt=0.0; self.dr=0.0; self.buy_ratio=0.5; self.vol1s=0 self.p1=0.0; self.adv=0.0; self.iliq=0.0; self.wall=0.0; self.level='等待' self.liq_events=deque(maxlen=50) self.trades=deque(maxlen=200) self.db=[]; self.da=[] # 缓存的p1/adv历史（前端不需要，保持兼容） self.p1h=deque(maxlen=100); self.advh=deque(maxlen=100); self.tls=deque(maxlen=100) self.wall_snapshots=deque(maxlen=30) self.fake_wall_score=0.0 self._debug_books = False # disable debug def dict(self): return {'mid':round(self.mid,4),'imb':round(self.imb,3), 'p1':round(self.p1,3),'adv':round(self.adv,3), 'level':self.level,'iliq':round(self.iliq,1),'wall':round(self.wall,1), 'lt':round(self.lt,3),'dr':round(self.dr,3),'buy_ratio':round(self.buy_ratio,3), 'bid1p':round(self.bid1p,1),'ask1p':round(self.ask1p,1), 'db':self.db[-30:],'da':self.da[-30:], 'le':[{'tm':e['tm'],'s':e['s'],'sz':e['sz'],'px':e['px'],'v':e['v']} for e in self.liq_events][-15:], 'p1h':list(self.p1h),'advh':list(self.advh),'tls':list(self.tls), 'fake_wall':round(self.fake_wall_score,3)} S=Store(); LT=LiquidationTracker(window_sec=10) def cg_loop(): while True: try: LT.poll_coinglass("BTCUSDT") except: pass time.sleep(60) threading.Thread(target=cg_loop, daemon=True).start() def calc(): try: if not S.bids or not S.asks: return bb=max(float(p) for p in S.bids); ba=min(float(p) for p in S.asks) except Exception as e: print('calc error:', str(e)[:80]); return S.mid=(bb+ba)/2; m=S.mid if m==0: return S.bid1p=sum(v for p,v in S.bids.items() if float(p)>=m*0.99) S.ask1p=sum(v for p,v in S.asks.items() if float(p)<=m*1.01) S.bid01p=sum(v for p,v in S.bids.items() if float(p)>=m*0.998) S.ask01p=sum(v for p,v in S.asks.items() if float(p)<=m*1.002) t=S.bid1p+S.ask1p; S.imb=(S.bid1p-S.ask1p)/t if t>0 else 0 b=sorted([(float(p),v) for p,v in S.bids.items()], reverse=True) a=sorted([(float(p),v) for p,v in S.asks.items()]) new_db = [] cb = 0 for p, v in b: if p < m * 0.97: break cb += v new_db.append({'p': p, 'v': v, 'c': cb}) new_da = [] ca = 0 for p, v in a: if p > m * 1.03: break ca += v new_da.append({'p': p, 'v': v, 'c': ca}) S.db = new_db S.da = new_da now=time.time_ns(); c2=now-200_000_000; c1=now-1_000_000_000; c5=now-5_000_000_000 h=S.trades v2=sum(t['s'] for t in h if t['t']*1e6>c2) v1=sum(t['s'] for t in h if t['t']*1e6>c1) v5=[t['s'] for t in h if t['t']*1e6>c5] a5=sum(v5)/len(v5) if v5 else 0 b1=sum(t['s'] for t in h if t['t']*1e6>c1 and t['d']=='buy') s1=sum(t['s'] for t in h if t['t']*1e6>c1 and t['d']=='sell') r2=[t for t in h if t['t']*1e6>c2] S.t200=v2; S.avg5=a5; S.vol1s=v1 S.buy_ratio=b1/(b1+s1) if (b1+s1)>0 else 0.5 S.lt=min(sum(t['s'] for t in r2 if t['s']>a5*2)/max(a5,0.001),2.0)/2.0 if a5>0 else 0 S.dr=min(v2/max(a5,0.001),2.0)/2.0 if a5>0 else 0 p1=max(0.0,min(1.0,0.4*S.imb+0.3*S.lt+0.3*S.dr)) iq=LT.I_liq(mid_price=m) wl=S.bid01p if S.liq_events: ls=S.liq_events[-1]['s']; wl=S.bid01p if ls=='buy' else S.ask01p adv=iq/max(wl,0.001) S.p1=p1; S.adv=adv; S.iliq=iq; S.wall=wl # --- 假墙检测 --- S.wall_snapshots.append((time.time(), wl)) # 移除5秒前的快照 cutoff = time.time() - 5.0 while S.wall_snapshots and S.wall_snapshots[0][0] < cutoff: S.wall_snapshots.popleft() ws = [w for _, w in S.wall_snapshots] if len(ws) >= 2: avg_wall = sum(ws) / len(ws) max_wall = max(ws) # 异常大墙（>3倍均值）+ 快速消失 = 疑似假墙 S.fake_wall_score = max(0.0, (max_wall / max(avg_wall, 0.01) - 1) * 0.5) # 墙厚度波动过大 -> 假墙可能性高 if avg_wall > 0: cv = (sum((w - avg_wall)**2 for w in ws) / len(ws))**0.5 / avg_wall S.fake_wall_score = max(S.fake_wall_score, cv * 0.7) else: S.fake_wall_score = 0.0 # 假墙惩罚：降低adv有效值 effective_adv = adv * (1.0 - min(S.fake_wall_score, 1.0) * 0.5) # --- if p1>=P1_ENTER and effective_adv>=ADV_ENTER: S.level='入场' elif p1>=P1_WATCH: S.level='观察' elif S.fake_wall_score > 0.6: S.level = '假墙' else: S.level='等待' S.p1h.append(p1); S.advh.append(min(effective_adv,5)); S.tls.append(datetime.now().strftime('%H:%M:%S')) def on_msg(ws, raw): try: d=json.loads(raw) if isinstance(raw,str) else raw except: return if not isinstance(d,dict): return try: ch=d.get('arg',{}).get('channel','') if ch=='books' and d.get('data'): if d.get('action')=='snapshot': S.bids={}; S.asks={} for b in d['data']: if isinstance(b,dict): for i in (b.get('bids') or []): if isinstance(i,(list,tuple)) and len(i)>=2: p,s=str(i[0]),float(i[1]) if s==0: S.bids.pop(p,None) else: S.bids[p]=s for i in (b.get('asks') or []): if isinstance(i,(list,tuple)) and len(i)>=2: p,s=str(i[0]),float(i[1]) if s==0: S.asks.pop(p,None) else: S.asks[p]=s if S._debug_books: print('[BOOKS] sn=%s dlen=%d' % (d.get('action','?'), len(d['data']))) calc() if S._debug_books: S._debug_books = False elif ch=='trades' and d.get('data'): for t in d['data']: if isinstance(t,dict): S.trades.append({'t':int(t.get('ts',0)),'d':str(t.get('side','')),'s':float(t.get('sz',0))}) calc() elif ch=='liquidation-orders' and d.get('data'): for ld in d['data']: if isinstance(ld,dict): sd=str(ld.get('side','')); sz=float(ld.get('sz',0)); px=float(ld.get('px',0)); ts=int(ld.get('ts',0)) LT.add_ws_raw(SYMBOL, sd, sz, px, ts, ld) S.liq_events.append({'tm':datetime.fromtimestamp(ts/1000).strftime('%H:%M:%S'),'s':sd,'sz':sz,'px':px,'v':sz*px}) except Exception as e: import traceback print('WS error:', str(e)[:100]) traceback.print_exc() def on_open(ws): print(f"WS connected -> {SYMBOL}") ws.send(json.dumps({"op":"subscribe","args":[ {"channel":"books","instId":SYMBOL}, {"channel":"trades","instId":SYMBOL}, {"channel":"liquidation-orders","instId":SYMBOL}, ]})) # ── Flask ── BASE = os.path.dirname(__file__) app=Flask(__name__); CORS(app) @app.route('/') def idx(): return send_file(os.path.join(BASE, 'templates', 'dashboard.html')) @app.route('/api/s') def api_s(): return jsonify(S.dict()) # ── 主入口 ── if __name__ == '__main__': proxy=config.PROXIES.get('http://','') if config.PROXIES else '' def ws_thread(): ws=websocket.WebSocketApp("wss://ws.okx.com:8443/ws/v5/public", header={"User-Agent":"Mozilla/5.0"}, on_open=on_open,on_message=on_msg, on_error=lambda ws,e:print(f"WS error:{e}"), on_close=lambda *a:print("WS closed")) if proxy: ws.run_forever(http_proxy_host="127.0.0.1",http_proxy_port=7897, sslopt={"cert_reqs":ssl.CERT_NONE},ping_interval=20) else: ws.run_forever(sslopt={"cert_reqs":ssl.CERT_NONE},ping_interval=20) threading.Thread(target=ws_thread, daemon=True).start() print(f"Dashboard: http://localhost:{PORT}") app.run(host='0.0.0.0', port=PORT, debug=False, use_reloader=False)