import json import websocket import ssl import threading import time import hmac import hashlib import base64 import sys import os # 動態添加相對路徑到 sys.path sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../conf'))) # 引入 conf.py import conf # OKX 公共 WebSocket URL OKX_PUBLIC_WS_URL = "wss://ws.okx.com:8443/ws/v5/public" # OKX 私有 WebSocket URL OKX_PRIVATE_WS_URL = "wss://ws.okx.com:8443/ws/v5/private" # API 认证信息 API_KEY = conf.apiKeyOKX6 # 直接使用字符串 SECRET_KEY = conf.secretOKX6 # 直接使用字符串 PASSPHRASE = conf.passphrase # 假設在 conf.py 中也有定義 passphrase symbols = [ 'FIL', 'BTC', 'ETH', 'XRP', 'ATOM', 'BCH', 'BSV', 'EOS', 'ETC', 'LTC', 'TRX', 'ADA', 'KNC', 'LINK', 'NEO', 'DOGE', 'ALGO', 'COMP', 'IOST', 'IOTA', 'ONT', 'QTUM', 'THETA', 'XLM', 'XTZ', 'MKR', 'SNX', 'ZRX', 'DOT', 'BAT', 'SUSHI', 'YFI', 'CRV', 'UMA', 'BAND', 'JST', 'KSM', 'TRB', 'BAL', 'LRC', 'RSR', 'STORJ', 'ZIL', 'UNI', 'AVAX', 'FLM', 'AAVE', 'CVC', 'GRT', 'NEAR', 'BNT', 'EGLD', '1INCH', 'SOL', 'BADGER', 'MASK', 'CFX', 'CHZ', 'MANA', 'ALPHA', 'PERP', 'SAND', 'CRO', 'LPT', 'RVN', 'XCH', 'SHIB', 'CSPR', 'ICP', 'MINA', 'AXS', 'YGG', 'SLP', 'AGLD', 'DYDX', 'CELO', 'GALA', 'KISHU', 'GODS', 'ENS', 'IMX', 'PEOPLE', 'NFT', 'BICO', 'LOOKS', 'API3', 'APE', 'GMT', 'LUNA', 'OP', 'LUNC', 'SWEAT', 'ETHW', 'APT', 'USTC', 'BNB', 'TON', 'LDO', 'GMX', 'MAGIC', 'CORE', 'AR', 'WOO', 'BLUR', 'FLOKI', 'STX', 'ARB', 'RDNT', 'PEPE', 'AIDOGE', 'SUI', 'CETUS', 'ORDI', 'WLD', 'HBAR', 'VRA', 'BIGTIME', 'ORBS', 'WAXP', 'GAS', 'TIA', 'MEME', 'FLOW', 'PYTH', 'SSV', 'INJ', 'AUCTION', 'ID', 'RON', 'CTC', 'TURBO', 'ACE', 'SATS', 'METIS', 'BONK', 'JTO', 'MOVR', 'NMR', 'JOE', 'FXS', 'LQTY', 'LSK', 'JUP', 'ZETA', 'OM', 'ONE', 'STRK', 'ACH', 'T', 'ICX', 'DGB', 'ENJ', 'VELO', 'RAY', 'AEVO', 'ETHFI', 'W', 'TNSR', 'FOXY', 'MEW', 'WIF', 'PRCL', 'MERL', 'ZENT', 'GLM', 'NOT', 'ULTI', 'ATH', 'ZK', 'ZRO', 'CVX', 'UXLINK', 'MAX', 'ONDO', 'RENDER', 'BOME', 'DOGS', 'POPCAT', 'NEIROETH', 'BRETT', 'CAT', 'TAO', 'CATI', 'POL', 'HMSTR', 'EIGEN', 'MOODENG', 'NEIRO', 'PUFFER', 'SCR', 'GOAT', 'X', 'GRASS', 'PNUT', 'ACT', 'DOG', 'DEGEN', 'SLERF', 'ARKM', 'OL', 'MEMEFI', 'SWELL', 'MORPHO', 'MAJOR', 'ZKJ', 'MOVE', 'ME', 'VIRTUAL', 'SUNDOG', 'VANA', 'PENGU', 'FARTCOIN', 'AI16Z', 'GRIFFAIN', 'ZEREBRO', 'AIXBT', 'ALCH', 'BIO', 'SONIC', 'DUCK', 'SOLV', 'NC', 'TRUMP', 'S', 'J', 'ANIME', 'ARC', 'SWARMS', 'COOKIE', 'VINE', 'AVAAI', 'BUZZ', 'PIPPIN', 'BERA', 'LAYER', 'IP', 'PI', 'KAITO', 'HYPE', 'SHELL', 'GPS' ] # 订阅公共频道配置 subscribe_message = { "op": "subscribe", "args": [{"channel": "bbo-tbt", "instId": f"{symbol}-USDT-SWAP"} for symbol in symbols] } # 生成 OKX API 认证签名 def generate_signature(timestamp, method, request_path, body): message = f"{timestamp}{method}{request_path}{body}" mac = hmac.new(SECRET_KEY.encode(), message.encode(), hashlib.sha256) return base64.b64encode(mac.digest()).decode() # WebSocket连接成功时的回调函数（公共频道） def on_public_open(ws): print("公：WebSocket 连接已开启，开始订阅...") ws.send(json.dumps(subscribe_message)) # WebSocket接收到消息时的回调函数（公共频道） def on_public_message(ws, message): data = json.loads(message) if "data" in data and data["data"]: orderbook = data["data"][0] symbol = data["arg"]["instId"] ask_price, ask_size = float(orderbook["asks"][0][0]), float(orderbook["asks"][0][1]) bid_price, bid_size = float(orderbook["bids"][0][0]), float(orderbook["bids"][0][1]) ask_value = ask_price * ask_size bid_value = bid_price * bid_size spread = ask_price / bid_price - 1 # 設定門檻條件 threshold = 0.01 # 0.2% 價差門檻 min_value = 100 # 交易價值門檻 if spread > threshold and ask_value > min_value and bid_value > min_value: spread_percen = spread * 100 print(f"公：交易對: {symbol}") print(f"公：賣盤: 價格={ask_price}, 數量={ask_size}, 價值={ask_value}") print(f"公：買盤: 價格={bid_price}, 數量={bid_size}, 價值={bid_value}") print(f"公：價差: {spread} USDT ({spread_percen:.6f}%)") # 當條件達到，將信號傳送給私有頻道進行下單 if spread > threshold and ask_value > min_value and bid_value > min_value: place_order(wsp, symbol, side="buy") # 觸發買單 place_order(wsp, symbol, side="sell") # 觸發賣單 # WebSocket连接关闭时的回调函数（公共频道） def on_public_close(ws, close_status_code, close_msg): print("公：WebSocket 连接已关闭") # WebSocket错误时的回调函数（公共频道） def on_public_error(ws, error): print(f"公：WebSocket发生错误: {error}") # 送出交易請求（私有頻道） def place_order(ws, symbol, side): print(f"發送{side}訂單... for {symbol}") # 動態建立訂單訊息 order_message = { "id": "test123456789", # 生成唯一訂單 ID（最多 16 個字元） "op": "order", "args": [ { "instId": f"{symbol}", # 使用動態的 symbol 生成交易對 "tdMode": "cross", # 逐倉模式 "side": side, # 買入或賣出 "ordType": "market", # 市價單 "sz": "1", # 購買/賣出 1USDT 等值的 CSPR 合約 } ] } # 發送訂單訊息 ws.send(json.dumps(order_message)) # 發送訂單 # WebSocket 连接成功时的回调函数（私有频道） def on_private_open(wsp): print("私：WebSocket 私有頻道已開啟，開始認證...") timestamp = str(int(time.time())) signature = generate_signature(timestamp, "GET", "/users/self/verify", "") auth_message = { "op": "login", "args": [ { "apiKey": API_KEY, "passphrase": PASSPHRASE, "timestamp": timestamp, "sign": signature } ] } wsp.send(json.dumps(auth_message)) # 启动心跳线程 threading.Thread(target=send_heartbeat, args=(wsp,), daemon=True).start() # WebSocket接收到消息时的回调函数（私有频道） def on_private_message(wsp, message): print("私：收到的訊息:", message) # data = json.loads(message) if message == "pong": print("私：收到 pong 訊息，連線正常") return # 不需要解析 pong 訊息 if not message.strip(): # 避免解析空字串 print("⚠️ 收到空訊息，忽略") return try: data = json.loads(message) except json.JSONDecodeError: print(f"⚠️ 無法解析訊息: {message}") return if "event" in data and data["event"] == "login": print("私：認證成功...") # WebSocket连接关闭时的回调函数（私有频道） def on_private_close(wsp, close_status_code, close_msg): print("私：WebSocket連線已關閉，準備重連...") reconnect_ws() # 觸發重連 # WebSocket错误时的回调函数（私有频道） def on_private_error(wsp, error): print(f"私：WebSocket發生錯誤: {error}") reconnect_ws() # 觸發重連 # 重新連接 WebSocket def reconnect_ws(): time.sleep(5) # 休息5秒，避免頻繁請求 print("正在重新連接 WebSocket...") start_private_ws() # 重新啟動 WebSocket 連線 # 心跳保持連線 def send_heartbeat(wsp): while True: try: if wsp.sock and wsp.sock.connected: wsp.send("ping") print("私：發送心跳 ping") time.sleep(20) except Exception as e: print(f"私：心跳錯誤: {e}") break # 創建並啟動公共 WebSocket 連線 def start_public_ws(): ws = websocket.WebSocketApp( OKX_PUBLIC_WS_URL, on_open=on_public_open, on_message=on_public_message, on_close=on_public_close, on_error=on_public_error ) ws.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) # 創建並啟動私有 WebSocket 連線 def start_private_ws(): wsp = websocket.WebSocketApp( OKX_PRIVATE_WS_URL, on_open=on_private_open, on_message=on_private_message, on_close=on_private_close, on_error=on_private_error ) wsp.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) return wsp # 让它返回 WebSocket 连接 if __name__ == "__main__": # 啟動公共 WebSocket 线程 public_ws_thread = threading.Thread(target=start_public_ws, daemon=True) public_ws_thread.start() # 啟動私有 WebSocket 线程 # private_ws_thread = threading.Thread(target=start_private_ws, daemon=True) # private_ws_thread.start() # 启动 WebSocket 线程 wsp = websocket.WebSocketApp( OKX_PRIVATE_WS_URL, on_open=on_private_open, on_message=on_private_message, on_close=on_private_close, on_error=on_private_error ) ws_thread = threading.Thread(target=wsp.run_forever, kwargs={"sslopt": {"cert_reqs": ssl.CERT_NONE}}) ws_thread.start() # # 确保 WebSocket 连接完成 # time.sleep(5) # # 发送买入订单 # place_order(wsp, "CSPR", "sell") # 觸發賣單 # 主线程保持运行 while True: time.sleep(1)