import json import websocket import ssl import threading import os import time import datetime from decimal import Decimal, ROUND_DOWN import hmac import hashlib import base64 import sys sys.path.append(os.path.abspath(os.path.join(os.path.dirname(__file__), '../../conf'))) import conf import psutil def display_current_process_usage(interval=1): # 取得當前 Python 進程的 PID pid = os.getpid() process = psutil.Process(pid) while True: # 取得當前 Python 進程的 CPU 使用率 cpu_usage = process.cpu_percent(interval=0.5) # 取得當前 Python 進程的記憶體使用情況 memory_info = process.memory_info() memory_usage = memory_info.rss / (1024 * 1024) # 轉換為 MB # 顯示當前 Python 進程的 CPU 和記憶體使用情況 print(f"當前 Python 進程 CPU 使用率: {cpu_usage}%") print(f"當前 Python 進程記憶體使用量: {memory_usage:.2f} MB") print('-' * 40) # 設定每秒更新一次 time.sleep(interval) # 全局變數，用於存儲私有頻道的 WebSocket 物件 ws_private = None OKX_PUBLIC_WS_URL = "wss://ws.okx.com:8443/ws/v5/public" OKX_PRIVATE_WS_URL = "wss://ws.okx.com:8443/ws/v5/private" API_KEY = conf.apiKeyOKX6 # 直接使用字符串 SECRET_KEY = conf.secretOKX6 # 直接使用字符串 PASSPHRASE = conf.passphrase # 假設在 conf.py 中也有定義 passphrase current_directory = os.path.dirname(os.path.realpath(__file__))# 自動取得目前 .py 檔案所在的目錄 json_file_path = os.path.join(current_directory, 'arbitrage_pairs.json')# JSON 檔案的完整路徑 # 創建並啟動 WebSocket 連線 def start_ws_public(): ws_public = websocket.WebSocketApp( OKX_PUBLIC_WS_URL, on_open=on_open_public, on_message=on_message_public, on_close=on_close_public, on_error=on_error_public ) ws_public.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) # 創建並啟動私有 WebSocket 連線 def start_ws_private(): ws_private = websocket.WebSocketApp( OKX_PRIVATE_WS_URL, on_open=on_open_private, on_message=on_message_private, on_close=on_close_private, on_error=on_error_private ) ws_private.run_forever(sslopt={"cert_reqs": ssl.CERT_NONE}) # 全局變數來保存載入的 JSON 資料 cached_data = None last_load_time = 0 # 上次讀取檔案的時間 # 讀取 JSON 資料的函數，15 分鐘重載一次 def load_arbitrage_pairs(): global cached_data, last_load_time # 檢查是否已經超過 15 分鐘 current_time = time.time() # 獲取當前時間（秒數） if current_time - last_load_time > 15 * 60: # 15 分鐘 with open(json_file_path, 'r', encoding='utf-8') as f: cached_data = json.load(f) # 讀取並更新資料 last_load_time = current_time # 更新最後讀取時間 print("JSON data reloaded.") return cached_data # 回傳所有幣種清單 def get_arbitrage_coins(): arbitrage_pairs = load_arbitrage_pairs() return [item["coin"] for item in arbitrage_pairs] # 查找特定幣種的市場資料 def search_market(coin): arbitrage_pairs = load_arbitrage_pairs() for item in arbitrage_pairs: if item['coin'] == coin: return item return None # 生成 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(30) 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() # 呼叫對應的連線啟動函式 # 儲存每個幣種的價格資料 price_container = {} # 更新價格資料 def update_price(coin, spot_bid, spot_b_vol, spot_ask, spot_a_vol, swap_bid, swap_b_vol, swap_ask, swap_a_vol): price_container[coin] = { "spot_bid": spot_bid, # 現貨市場的買1價格 "spot_b_vol": spot_b_vol, # 現貨市場的買1價格 "spot_ask": spot_ask, # 現貨市場的賣1價格 "spot_a_vol": spot_a_vol, # 現貨市場的賣1價格 "swap_bid": swap_bid, # 合約市場的買1價格 "swap_b_vol": swap_b_vol, # 合約市場的買1價格 "swap_ask": swap_ask, # 合約市場的賣1價格 "swap_a_vol": swap_a_vol # 合約市場的賣1價格 } # WebSocket 连接成功时的回调函数（私有频道） def on_open_private(ws): global ws_private # 引用全局變數 ws_private ws_private = ws # 把 WebSocket 物件賦值給全局變數 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() def on_message_private(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_close_private(ws_private, close_status_code, close_msg): print("私：WebSocket 連線關閉") reconnect_ws("私", start_ws_private) # WebSocket错误时的回调函数（私有频道） def on_error_private(ws_private, error): print(f"私：WebSocket 發生錯誤: {error}") reconnect_ws("私", start_ws_private) def place_order(ws_private, order_type, coin, side, sz): if ws_private is None: print("WebSocket private 尚未初始化或尚未連接.") print(f"發送 {side} 訂單... for {coin} ({order_type})") # 根據訂單類型動態設置 instId 和 tgtCcy if order_type == "SPOT": inst_id = f"{coin}-USDT" # 現貨交易對 tgt_ccy = 'base_ccy' # 現貨交易需要設定 tgtCcy elif order_type == "SWAP": inst_id = f"{coin}-USDT-SWAP" # 合約交易對 tgt_ccy = None # 合約交易不需要 tgtCcy else: print("不支援的訂單類型") return # 動態建立訂單訊息 order_message = { "id": "test123456789", # 生成唯一訂單 ID（最多 16 個字元） "op": "order", "args": [ { "instId": inst_id, # 使用動態的 symbol 生成交易對 "tdMode": "cross", # 全倉模式 "side": side, # 買入或賣出 "ordType": "market", # 市價單 # 如果是 SPOT 訂單，設置 tgtCcy "tgtCcy": tgt_ccy if tgt_ccy else None, "sz": str(sz), # 購買/賣出數量，轉為字串以適應 API 格式 } ] } #base_ccy: 交易货币 ；quote_ccy：计价货币，仅适用于币币市价订单，默认买单为quote_ccy，卖单为base_ccy # 發送訂單訊息 ws_private.send(json.dumps(order_message)) # 發送訂單 # WebSocket 連線成功時的回調函式 def on_open_public(ws_public): print("公：WebSocket連線已開啟，開始訂閱...") # coins = [ # # 'KISHU' # 'PI' # ] coins = get_arbitrage_coins() # 訂閱頻道配置 subscribe_message = { "op": "subscribe", "args": [ {"channel": "bbo-tbt", "instId": f"{coin}-USDT-SWAP"} for coin in coins ] + [ {"channel": "bbo-tbt", "instId": f"{coin}-USDT"} for coin in coins ] } ws_public.send(json.dumps(subscribe_message)) # 發送訂閱請求 # 啟動心跳執行緒 threading.Thread(target=send_heartbeat, args=(ws_public, "公"), daemon=True).start() # WebSocket 接收到訊息時的回調函式 def on_message_public(ws_public, message): # print(message) if message == "pong": print("💓 公：PONG") return # 不需要解析 pong 訊息 data = json.loads(message) if 'event' in data: return # 不需要解析 pong 訊息 # if 'data' in data: # 這裡處理正常情況下的數據 # print("收到的數據:", data) # 確認是否是訂閱的市場資料 if "arg" in data and "channel" in data["arg"] and "bbo-tbt" in data["arg"]["channel"]: inst_id = data["arg"]["instId"] coin = inst_id.split("-")[0] # 取得幣種 sea_mkt = search_market(coin) # print(sea_mkt["sw-ctVal"]) # 初始化本地變數（如果已經存在資料就從中取出） current_price = price_container.get(coin, { "spot_bid": None, "spot_b_vol": None, "spot_ask": None, "spot_a_vol": None, "swap_bid": None, "swap_b_vol": None, "swap_ask": None, "swap_a_vol": None }) # 判斷是否是合約市場或現貨市場 if inst_id.endswith("USDT-SWAP"): current_price["swap_bid"] = str(Decimal(data["data"][0]["bids"][0][0])) current_price["swap_b_vol"] = str(Decimal(data["data"][0]["bids"][0][0]) * Decimal(data["data"][0]["bids"][0][1]) * Decimal(sea_mkt["sw-ctVal"])) current_price["swap_ask"] = str(Decimal(data["data"][0]["asks"][0][0])) current_price["swap_a_vol"] = str(Decimal(data["data"][0]["asks"][0][0]) * Decimal(data["data"][0]["asks"][0][1]) * Decimal(sea_mkt["sw-ctVal"])) elif inst_id.endswith("USDT"): # 現貨市場部分 current_price["spot_bid"] = str(Decimal(data["data"][0]["bids"][0][0])) current_price["spot_b_vol"] = str(Decimal(data["data"][0]["bids"][0][0]) * Decimal(data["data"][0]["bids"][0][1])) # 現貨市場買1價格的成交量 current_price["spot_ask"] = str(Decimal(data["data"][0]["asks"][0][0])) current_price["spot_a_vol"] = str(Decimal(data["data"][0]["asks"][0][0]) * Decimal(data["data"][0]["asks"][0][1])) # 現貨市場賣1價格的成交量 # 立刻更新價格容器 update_price(coin, current_price["spot_bid"], current_price["spot_b_vol"], current_price["spot_ask"], current_price["spot_a_vol"], current_price["swap_bid"], current_price["swap_b_vol"], current_price["swap_ask"], current_price["swap_a_vol"]) # 如果所有價格都齊全，執行套利檢查 if all(value is not None for value in current_price.values()): check_arbitrage_opportunity(coin) # 檢查套利機會 def check_arbitrage_opportunity(coin): # 確保資料存在 if coin not in price_container: return prices = price_container[coin] # 轉換價格為 Decimal，確保精確度 swap_bid, swap_b_vol = Decimal(prices["swap_bid"]), Decimal(prices["swap_b_vol"]) # 合約市場買1價格與成交量 swap_ask, swap_a_vol = Decimal(prices["swap_ask"]), Decimal(prices["swap_a_vol"]) # 合約市場賣1價格與成交量 spot_bid, spot_b_vol = Decimal(prices["spot_bid"]), Decimal(prices["spot_b_vol"]) # 現貨市場買1價格與成交量 spot_ask, spot_a_vol = Decimal(prices["spot_ask"]), Decimal(prices["spot_a_vol"]) # 現貨市場賣1價格與成交量 # coin = "PI" pos = get_position(coin, balance_and_position) ##載入全部持倉資料 sea_mkt = search_market(coin) ##載入最小下單 合約面值 精度 等資料 spot_position = Decimal(pos.get("spot_pos", "0")) swap_position = Decimal(pos.get("swap_pos", "0")) sp_minSz = Decimal(sea_mkt.get("sp-minSz", "0")) sp_lotSz = Decimal(sea_mkt.get("sp-lotSz", "0")) sw_minSz = Decimal(sea_mkt.get("sw-minSz", "0")) sw_lotSz = Decimal(sea_mkt.get("sw-lotSz", "0")) sw_ctVal = Decimal(sea_mkt.get("sw-ctVal", "0")) funding_apy = Decimal(sea_mkt.get("funding_apy", "0")) spot_position_vol = spot_position * swap_bid #現貨持倉價值 swap_position_vol = swap_position * swap_bid #合約持倉價值 # print(f"幣種: {coin}") # print(swap_bid) # print(spot_position_vol) # print(swap_position_vol) # print(spot_position) # print(swap_position) # print(f"現貨最小下單量 sp-minSz: {sp_minSz}") # print(f"現貨最小跳動 sp-lotSz: {sp_lotSz}") # print(f"合約最小下單量 sw-minSz: {sw_minSz}") # print(f"合約最小跳動 sw-lotSz: {sw_lotSz}") # print(f"合約面額 sw-ctVal: {sw_ctVal}") # print(f"資金費年化 funding_apy: {funding_apy}%") min_spot_value = swap_bid * sp_minSz min_swap_value = swap_bid * sw_minSz * sw_ctVal if min_swap_value > min_spot_value: spot_order_amount = max(min_swap_value / swap_bid, sp_minSz) swap_order_amount = sw_minSz else: spot_order_amount = sp_minSz swap_order_amount = max(min_spot_value / swap_bid / sw_ctVal, sw_minSz) # 如果小於 sw_minSz，則設為 sw_minSz # 四捨五入到符合最小跳動量的倍數（向下取整，避免超過餘額或錯誤） def adjust_to_step(value, step): return (value // step) * step # 調整精度 spot_order_amount = adjust_to_step(spot_order_amount, sp_lotSz).quantize(sp_lotSz, rounding=ROUND_DOWN) swap_order_amount = adjust_to_step(swap_order_amount, sw_lotSz).quantize(sw_lotSz, rounding=ROUND_DOWN) # print(f"現貨最小下單金額: {min_spot_value:.4f} USDT") # print(f"合約最小下單金額: {min_swap_value:.4f} USDT") # print(f"建議現貨下單數量: {spot_order_amount}") # print(f"建議合約下單張數: {swap_order_amount}") IN_threshold = 1.0 OUT_threshold = 1.0 # 把 funding_apy 轉換為 float funding_apy = float(funding_apy) if funding_apy > 50: IN_threshold -= 0.2 OUT_threshold += 2 elif 0 <= funding_apy < 50: pass # 維持原本的門檻 elif -20 <= funding_apy < 0: IN_threshold += 1 OUT_threshold -= 0.1 elif -50 <= funding_apy < -20: IN_threshold += 2 OUT_threshold -= 0.2 else: IN_threshold += 10 # 基本上不進入 OUT_threshold -= 0.5 min_value = 100 # 成交價值門檻 # 進場條件：合約買1 > 現貨賣1 → 買現貨，空合約 if swap_bid > spot_ask: spread = round((swap_bid / spot_ask - 1) * 100, 2) if spread > IN_threshold and swap_b_vol > min_value and spot_a_vol > min_value: current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') print(f"{current_time} ✅進場套利✅：買現貨 {coin}@{spot_ask}，空合約 {coin}@{swap_bid}，🌈資費 {funding_apy}，量{spot_a_vol},{swap_b_vol}") print(f"➡️ 合約高出現貨 {spread}%➡️ ➡️ 進場域值 {IN_threshold}") # 執行套利進場邏輯 if spot_position_vol < 10 and swap_position_vol > -10: place_order(ws_private, "SPOT", coin, "buy", spot_order_amount) place_order(ws_private, "SWAP", coin, "sell", swap_order_amount) # 出場條件：現貨買1 > 合約賣1 → 賣現貨，多合約 elif spot_bid > swap_ask: spread = round((spot_bid / swap_ask - 1) * 100, 2) if spread > OUT_threshold and spot_b_vol > min_value and swap_a_vol > min_value: current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S') print(f"{current_time} ⭕出場套利⭕：賣現貨 {coin}@{spot_bid}，多合約 {coin}@{swap_ask}，🌈資費 {funding_apy}，量{spot_b_vol},{swap_a_vol}") print(f"➡️ 現貨高出合約 {spread}%➡️ ➡️ 出場域值 {OUT_threshold}") # 執行套利出場邏輯 # 條件：有空單且現貨也有部位，且數量都夠大，才執行套利出場 if ( swap_position < 0 and abs(swap_position) >= swap_order_amount and spot_position > 0 and spot_position >= spot_order_amount ): print(f"符合套利出場條件，準備對 {coin} 平倉") # 平空單（買入平倉） place_order(ws_private, "SWAP", coin, "buy", swap_order_amount) # 賣出現貨 place_order(ws_private, "SPOT", coin, "sell", spot_order_amount) # 每隔 5 秒打印 price_container def print_price_container(): while True: time.sleep(1) # 每隔 5 秒打印一次 print("目前的價格容器 (price_container)：") print(json.dumps(price_container, indent=4)) # WebSocket 連線關閉時的回調函式 def on_close_public(ws_public, close_status_code, close_msg): print("公：WebSocket on_close_public 連線關閉") reconnect_ws("公", start_ws_public) # WebSocket 錯誤時的回調函式 def on_error_public(ws_public, error): print(f"公：WebSocket on_error_public 發生錯誤: {error}") # reconnect_ws("公", start_ws_public) # 主线程保持运行 # while True: # time.sleep(1) # 每隔 5 秒打印 balance_and_position def print_balance_and_position(): while True: time.sleep(60) # 每隔 5 秒打印一次 print("現貨&合約持倉：",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 get_position(coin, data): if coin in data: return data[coin] else: return { "spot_pos": "0", "swap_pos": "0" } # 啟動 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() private_ws_thread = threading.Thread(target=start_ws_private) private_ws_thread.start() # 啟動打印 現貨合約持倉 ((私)) print_thread = threading.Thread(target=print_balance_and_position, daemon=True) print_thread.start() # 啟動打印即時價量 關鍵數據 但沒啥好看的 只是看他有沒有工作而已 # print_thread = threading.Thread(target=print_price_container, daemon=True) # print_thread.start() # 启动 WebSocket 线程 # ws_private = websocket.WebSocketApp( # OKX_PRIVATE_WS_URL, # on_open=on_open_private, # on_message=on_message_private, # on_close=on_close_private, # on_error=on_error_private # ) # private_ws_thread = threading.Thread(target=ws_private.run_forever, kwargs={"sslopt": {"cert_reqs": ssl.CERT_NONE}}) # private_ws_thread.start() # time.sleep(5) # # place_order(ws_private, "SPOT", "CSPR", "buy",1) # place_order(ws_private, "SWAP", "CSPR", "buy", 1) # 開始顯示當前 Python 進程的使用情況 display_current_process_usage() # 主线程保持运行 while True: time.sleep(1)