import json import websocket import ssl import threading import time import datetime 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.apiKeyOKX0 # 直接使用字符串 # SECRET_KEY = conf.secretOKX0 # 直接使用字符串 API_KEY = conf.apiKeyOKX6 # 直接使用字符串 SECRET_KEY = conf.secretOKX6 # 直接使用字符串 PASSPHRASE = conf.passphrase # 假設在 conf.py 中也有定義 passphrase # 生成 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 send_heartbeat(ws, name="WebSocket"): while True: try: if ws.sock and ws.sock.connected: current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') ws.send("ping") print(f"{name}：{current_time}：發送心跳 ping") time.sleep(60) except Exception as e: print(f"{name}：心跳錯誤: {e}") break # 重新連接 WebSocket def reconnect_ws(name, start_function): current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') print(f"{name}：{current_time} reconnect_ws 函式 正在重新連線 WebSocket...") time.sleep(5) # 等待 5 秒鐘後重新連線 start_function() # 呼叫對應的連線啟動函式 # 送出交易請求（私有頻道） def place_order(ws_private, 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_private.send(json.dumps(order_message)) # 發送訂單 # 每隔 5 秒打印 balance_and_position def print_balance_and_position(): while True: time.sleep(2) # 每隔 5 秒打印一次 print("目前的價格容器 (balance_and_position)：",json.dumps(balance_and_position, indent=4)) # 全域字典，用來儲存帳戶資訊 account_data = {} # 處理帳戶資訊的函式 def parse_account_data(msg): global account_data if not msg.get("data"): return # 預設只處理第一筆資料 data = msg["data"][0] # 提取基本資料 account_data["adjEq"] = data.get("adjEq") account_data["borrowFroz"] = data.get("borrowFroz") account_data["imr"] = data.get("imr") account_data["isoEq"] = data.get("isoEq") account_data["mgnRatio"] = data.get("mgnRatio") account_data["mmr"] = data.get("mmr") account_data["notionalUsd"] = data.get("notionalUsd") account_data["notionalUsdForBorrow"] = data.get("notionalUsdForBorrow") account_data["notionalUsdForFutures"] = data.get("notionalUsdForFutures") account_data["notionalUsdForOption"] = data.get("notionalUsdForOption") account_data["notionalUsdForSwap"] = data.get("notionalUsdForSwap") account_data["ordFroz"] = data.get("ordFroz") account_data["totalEq"] = data.get("totalEq") account_data["upl"] = data.get("upl") account_data["uTime"] = data.get("uTime") # 處理 details 中 USDT 的部分 usdt_info = next((item for item in data.get("details", []) if item.get("ccy") == "USDT"), None) if usdt_info: account_data["USDT"] = { "availBal": usdt_info.get("availBal"), "availEq": usdt_info.get("availEq"), "borrowFroz": usdt_info.get("borrowFroz"), "cashBal": usdt_info.get("cashBal"), "coinUsdPrice": usdt_info.get("coinUsdPrice"), "collateralEnabled": usdt_info.get("collateralEnabled"), "crossLiab": usdt_info.get("crossLiab"), "disEq": usdt_info.get("disEq"), "eq": usdt_info.get("eq"), "eqUsd": usdt_info.get("eqUsd"), } # 外部初始化字典 balance_and_position = {} def parse_balance_and_position(msg): """ 解析 balance_and_position 資訊，提取現貨和期貨的持倉資料 """ global balance_and_position # 使用外部字典 # 擷取現貨（balData）資訊 for item in msg['data'][0]['balData']: coin = item['ccy'] # 幣種 cash_bal = item['cashBal'] # 現貨餘額 # 如果該幣種不存在，則初始化 if coin not in balance_and_position: balance_and_position[coin] = {'spot_pos': 0, 'swap_pos': 0} balance_and_position[coin]['spot_pos'] = cash_bal # 更新現貨持倉 # 擷取期貨（posData）資訊 for item in msg['data'][0]['posData']: coin = item['instId'].split('-')[0] # 從 instId 擷取幣種，去掉 '-USDT-SWAP' pos = item['pos'] # 期貨持倉數量 # 如果該幣種不存在，則初始化 if coin not in balance_and_position: balance_and_position[coin] = {'spot_pos': 0, 'swap_pos': 0} balance_and_position[coin]['swap_pos'] = pos # 更新期貨持倉 # 返回更新後的結果 return balance_and_position def on_private_message(ws_private, message): try: # PONG 回覆（心跳回應） if message == "pong": print("💓 私：PONG 心跳回應") return # 將文字訊息轉為 JSON 格式 msg = json.loads(message) # 登入成功確認 if msg.get("event") == "login": if msg.get("code") == "0": print("🔐 私：登入成功") # 訂閱持倉頻道 subscribe_message = { "op": "subscribe", "args": [ {"channel": "account"}, # 訂閱帳戶頻道 {"channel": "balance_and_position"}, # {"channel": "positions", "instType": "SWAP"}, # 訂閱合約持倉頻道並指定 instType: "SWAP" # {"channel": "liquidation-warning", "instType": "SWAP"} # 訂閱合約持倉頻道並指定 instType: "SWAP" ] } ws_private.send(json.dumps(subscribe_message)) else: print(f"❌ 私：登入失敗，錯誤訊息：{msg}") return # 訂閱成功確認 if msg.get("event") == "subscribe": channel = msg.get("arg", {}).get("channel") print(f"📡 私：已訂閱頻道 {channel}") return # 特殊 event，例如連線數顯示 if msg.get("event") == "channel-conn-count": print(f"📩 私：頻道連線數資訊：{msg}") return # 處理帳戶資料 if msg.get("arg", {}).get("channel") == "account" and "data" in msg: print("📊 收到帳戶資訊：", msg) parse_account_data(msg) print("✅ 已更新 account_data:",json.dumps(account_data, indent=4)) return # 📊 處理 balance_and_position 資訊 if msg.get("arg", {}).get("channel") == "balance_and_position" and "data" in msg: print("📊 收到現貨&持倉訊息：", msg) parse_balance_and_position(msg) print(f"✅ 解析後的資料：balance_and_position",json.dumps(balance_and_position, indent=4)) return # 📈 處理 positions 資訊（如合約持倉） if msg.get("arg", {}).get("channel") == "positions" and "data" in msg: print("📈 收到合約持倉訊息：", msg) return # 清算警告訊息處理 if msg.get("arg", {}).get("channel") == "liquidation-warning" and "data" in msg: print("⚠️ 清算警告訊息：", msg) # 可以在這裡根據實際需要進行進一步的處理，例如解析數據或顯示警告訊息。 return # 其他未知類型訊息 print("❓ 私：收到未處理訊息：", msg) except json.JSONDecodeError: print(f"❗ 私：訊息解碼失敗：{message}") except Exception as e: print(f"🔥 私：處理訊息時發生錯誤：{e}") # WebSocket 连接成功时的回调函数（私有频道） def on_private_open(ws_private): 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 } ] } ws_private.send(json.dumps(auth_message)) # 發送訂閱請求 # 啟動心跳（推薦放這） threading.Thread(target=send_heartbeat, args=(ws_private, "私"), daemon=True).start() # WebSocket连接关闭时的回调函数（私有频道） def on_private_close(ws_private, close_status_code, close_msg): print("私：WebSocket 連線關閉") reconnect_ws("私", start_private_ws) # WebSocket错误时的回调函数（私有频道） def on_private_error(ws_private, error): print(f"私：WebSocket 發生錯誤: {error}") reconnect_ws("私", start_private_ws) # 創建並啟動私有 WebSocket 連線 def start_private_ws(): ws_private = 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_private.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) # 啟動 WebSocket 連線 if __name__ == "__main__": # 初始化全局變數 public_ws_thread = None private_ws_thread = None # 創建並啟動公頻道 WebSocket 連線 # public_ws_thread = threading.Thread(target=start_ws_public) # public_ws_thread.start() # 創建並啟動私頻道 WebSocket 連線 private_ws_thread = threading.Thread(target=start_private_ws) private_ws_thread.start() # # 发送买入订单 # place_order(ws_private, "CSPR", "sell") # 觸發賣單 # 啟動打印容器的執行緒 print_thread = threading.Thread(target=print_balance_and_position, daemon=True) print_thread.start()