import json import websocket import ssl import threading import os import time import datetime from decimal import Decimal # 自動取得目前 .py 檔案所在的目錄 current_directory = os.path.dirname(os.path.realpath(__file__)) # JSON 檔案的完整路徑 json_file_path = os.path.join(current_directory, 'arbitrage_pairs.json') # 每次呼叫都會重新讀入最新的 JSON 資料 def load_arbitrage_pairs(): with open(json_file_path, 'r', encoding='utf-8') as f: return json.load(f) # 回傳所有幣種清單 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 WebSocket URL OKX_PUBLIC_WS_URL = "wss://ws.okx.com:8443/ws/v5/public" # 通用的 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(2) 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_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價格與成交量 # 當現貨賣1 < 合約買1時，表示有套利機會 # spot　swap # ask ask # bid bid sea_mkt = search_market(coin) funding_apy = Decimal(sea_mkt["funding_apy"]) # 將 funding_apy 轉換為浮動數字類型 IN_threshold = 1 OUT_threshold = 1 if funding_apy > 50: IN_threshold -= 0.5 OUT_threshold += 2 elif funding_apy > 0: pass # 維持原本的門檻 elif funding_apy > -20: IN_threshold += 1 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}") print(f"➡️ 合約高出現貨 {spread}%") print(sea_mkt["funding_apy"]) # 執行套利進場邏輯 # 出場條件：現貨買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}") print(f"➡️ 現貨高出合約 {spread}%") print(sea_mkt["funding_apy"]) # 執行套利出場邏輯 # 在現貨市場購買，並在合約市場做空 # execute_arbitrage(spot_ask, swap_bid) # 實作套利交易操作 # def execute_arbitrage(spot_price, swap_price): # current_time = datetime.datetime.now().strftime('%Y-%m-%d %H:%M:%S')# 獲取當前時間並格式化 # print(f"公：{current_time}：執行套利：以現貨價格 {spot_price} 買入，並以合約價格 {swap_price} 賣空") # 實際的套利交易操作，例如使用API執行交易等 # 每隔 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_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_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) # 創建並啟動 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 連線 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() # 啟動打印容器的執行緒 # print_thread = threading.Thread(target=print_price_container, daemon=True) # print_thread.start()