#!/usr/bin/env python # -*- coding: utf-8 -*- """ 策略B自动化测试 — 清算数据采集 + p1信号检测 持续运行，收集LABUSDT清算/订单簿数据，每5秒输出一次状态。 不下单。 """ import sys, os, json, time, ssl, threading from datetime import datetime from collections import deque import websocket sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..')) import config sys.path.insert(0, os.path.join(os.path.dirname(__file__), '..', '..', 'exchange')) from liquidation import LiquidationTracker, CoinGlassAPI SYMBOL = "BTC-USDT-SWAP" P1_ENTER = 0.80 P1_WATCH = 0.55 ADV_ENTER = 1.0 # BTC 挂单墙更厚，降低 Adv 阈值 print_lock = threading.Lock() def safe_print(s): with print_lock: print(s) # ── 订单簿 ── class OrderBook: def __init__(self): self.bids = {} self.asks = {} self.mid = 0.0 self._bid_1p = 0.0; self._ask_1p = 0.0 self._bid_01p = 0.0; self._ask_01p = 0.0 def apply(self, data): if data.get('action') == 'snapshot': self.bids.clear(); self.asks.clear() for b in data.get('bids',[]): p,sz,*_=b; sf=float(sz) if sf==0: self.bids.pop(p,None) else: self.bids[p]=sf for a in data.get('asks',[]): p,sz,*_=a; sf=float(sz) if sf==0: self.asks.pop(p,None) else: self.asks[p]=sf self._calc() def _calc(self): if not self.bids or not self.asks: return bb = max(float(p) for p in self.bids) ba = min(float(p) for p in self.asks) self.mid = (bb+ba)/2.0 if self.mid==0: return b1=self.mid*0.99; a1=self.mid*1.01 b01=self.mid*0.998; a01=self.mid*1.002 self._bid_1p=sum(v for p,v in self.bids.items() if float(p)>=b1) self._ask_1p=sum(v for p,v in self.asks.items() if float(p)<=a1) self._bid_01p=sum(v for p,v in self.bids.items() if float(p)>=b01) self._ask_01p=sum(v for p,v in self.asks.items() if float(p)<=a01) @property def imb(self): t=self._bid_1p+self._ask_1p return (self._bid_1p-self._ask_1p)/t if t>0 else 0 @property def wall_bid(self): return self._bid_01p @property def wall_ask(self): return self._ask_01p # ── 成交追踪 ── class TradeTracker: def __init__(self): self._200=deque(maxlen=60); self._5s=deque(maxlen=600) self._side200=deque(maxlen=60) def add(self, side, sz, px, ns): self._200.append((ns,sz)); self._5s.append((ns,sz)) self._side200.append((ns,side)) now=ns while self._200 and self._200[0][0]0 else 0 def large_taker(self): a=self.avg_5s() if a==0: return 0.0 return min(sum(s for _,s in self._200 if s>a*3)/a, 2.0)/2.0 def drain(self): a=self.avg_5s() if a==0: return 0.0 return min(self.vol_200()/max(a,0.001),2.0)/2.0 @property def buy_ratio(self): if not self._side200: return 0.5 b=sum(1 for _,s in self._side200 if s=='buy') return b/len(self._side200) @property def taker_vol_1s(self): """最近1s成交量""" now=time.time_ns() return sum(s for ts,s in self._200 if ts>now-1_000_000_000) # ── 全局状态 ── ob = OrderBook() trk = TradeTracker() liq_tracker = LiquidationTracker(window_sec=10) # Coinglass定时轮询 def cg_poller(): while True: try: liq_tracker.poll_coinglass("LABUSDT") except: pass time.sleep(60) threading.Thread(target=cg_poller, daemon=True).start() output_interval = 5.0 last_output = 0 def compute_p1(): return 0.4*ob.imb + 0.3*trk.large_taker() + 0.3*trk.drain() def compute_adv(): mid = ob.mid if mid==0: return 0,0,0 i_liq = liq_tracker.I_liq(mid_price=mid) wall = ob.wall_bid recent = liq_tracker.recent_events(3.0) if recent: last_side = recent[-1].side wall = ob.wall_bid if last_side=='buy' else ob.wall_ask adv = i_liq / max(wall, 0.001) return adv, i_liq, wall def print_status(): global last_output now = time.time() if now - last_output < output_interval: return last_output = now mid = ob.mid if mid==0: return p1 = compute_p1() p1 = max(0.0, min(1.0, p1)) adv, i_liq, wall = compute_adv() liq_sum = liq_tracker.summary(5.0) # 信号等级 if p1>=P1_ENTER and adv>=ADV_ENTER: level = "🔥🔥 入场" elif p1>=P1_WATCH: level = "👀 观察" elif adv<0.8: level = "⚠️ 强度弱" else: level = " 等待" ts = datetime.now().strftime('%H:%M:%S') status = (f"\n{'='*60}\n" f" {ts} | {SYMBOL} | {level}\n" f"{'='*60}\n" f" p1={p1:.3f} (imb={ob.imb:.3f} lt={trk.large_taker():.3f} drain={trk.drain():.3f})\n" f" Adv={adv:.3f} | I_liq=${i_liq:.1f}/s | 墙={wall:.1f}\n" f" Mid=${mid:.4f} | B墙={ob.wall_bid:.1f} A墙={ob.wall_ask:.1f}\n" f" 买比={trk.buy_ratio:.2f} | 成交1s={trk.taker_vol_1s:.1f}\n" f" 清算({liq_sum['total_events']}笔最近5s) | 方向={liq_sum['net_dir']:+.3f} " f" B=${liq_sum['buy_vol']:.0f} S=${liq_sum['sell_vol']:.0f}") safe_print(status) # ── WS回调 ── def on_msg(ws, raw): try: d = json.loads(raw) if isinstance(raw, str) else raw except: return if not isinstance(d, dict): return arg = d.get('arg',{}) ch = arg.get('channel','') if ch=='books' and d.get('data'): for b in d['data']: ob.apply(b) elif ch=='trades' and d.get('data'): for t in d['data']: liq_tracker.poll_okx_rest(SYMBOL) trk.add(t.get('side',''), float(t.get('sz',0)), float(t.get('px',0)), int(t.get('ts',0))*1_000_000) elif ch=='liquidation-orders' and d.get('data'): for ld in d['data']: side = ld.get('side','') sz = float(ld.get('sz',0)) px = float(ld.get('px',0)) ts = int(ld.get('ts',0)) liq_tracker.add_ws_raw(SYMBOL, side, sz, px, ts, ld) safe_print(f" 💀 清算 {side:>4s} | {sz:.2f}张 @ ${px:.4f} | ${sz*px:.0f}") print_status() def on_open(ws): safe_print(f" 🟢 WS已连接 → 订阅 {SYMBOL}") sub = json.dumps({ "op": "subscribe", "args": [ {"channel":"books","instId":SYMBOL}, {"channel":"trades","instId":SYMBOL}, {"channel":"liquidation-orders","instId":SYMBOL}, ] }) ws.send(sub) def main(): safe_print(f"📊 策略B清算数据采集 — {SYMBOL}") safe_print(f" 阈值: p1入场>{P1_ENTER} 观察>{P1_WATCH} Adv入场>{ADV_ENTER}") safe_print(f" 数据源: OKX WS(实时清算) + Coinglass(60s轮询)") safe_print(f" 连接中...") proxy = config.PROXIES.get('http://','') if config.PROXIES else '' 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: safe_print(f" 🔴 WS错误: {e}"), on_close=lambda *a: safe_print(f" 🔴 WS关闭") ) try: 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) except KeyboardInterrupt: safe_print("\n 测试结束") if __name__ == '__main__': main()