import asyncio import queue import threading import time import websocket import json import pandas as pd import warnings import logging import colorlog import websockets # 忽略 FutureWarning warnings.simplefilter(action='ignore', category=FutureWarning) # 配置日志 handler = colorlog.StreamHandler() handler.setFormatter(colorlog.ColoredFormatter( "%(log_color)s%(asctime)s - %(name)s - %(levelname)s \n %(message)s", datefmt=None, reset=True, log_colors={ 'DEBUG': 'cyan', 'INFO': 'blue', 'WARNING': 'yellow', 'ERROR': 'red', 'CRITICAL': 'bold_red', } )) logger = logging.getLogger("market_monitor") logger.setLevel(logging.INFO) logger.addHandler(handler) # Binance WebSocket API URL SOCKET_TRADE = "wss://stream.binance.com:9443/ws/btcusdt@trade" SOCKET_DEPTH = "wss://stream.binance.com:9443/ws/btcusdt@depth20@100ms" # Initialize the DataFrame df_trades = pd.DataFrame(columns=['price', 'qty', 'timestamp']) df_order_book = pd.DataFrame(columns=['bid_price', 'bid_qty', 'ask_price', 'ask_qty']) previous_order_book = None fill_probabilities = {} order_disappearances = {} order_executions = {} last_trade = {'price': None, 'qty': None, 'side': None} messages = queue.Queue() # 创建一个线程安全队列 def on_message_trade(_ws, message): global df_trades, order_executions, last_trade json_message = json.loads(message) trade = { 'price': float(json_message['p']), 'qty': float(json_message['q']), 'timestamp': pd.to_datetime(json_message['T'], unit='ms'), 'side': 'buy' if json_message['m'] else 'sell' # 'm' indicates whether the buyer is the market maker } trade_df = pd.DataFrame([trade]) if not trade_df.empty and not trade_df.isna().all().all(): df_trades = pd.concat([df_trades, trade_df], ignore_index=True) # 记录每个价格的实际成交总量 price = trade['price'] last_trade = {'price': price, 'qty': trade['qty'], 'side': trade['side']} if price not in order_executions: order_executions[price] = 0 order_executions[price] += trade['qty'] calculate_fill_probabilities() def on_message_depth(_ws, message): global df_order_book, order_disappearances, previous_order_book json_message = json.loads(message) bids = json_message['bids'][:10] # Top 10 bids asks = json_message['asks'][:10] # Top 10 asks order_book = { 'bid_price': [float(bid[0]) for bid in bids], 'bid_qty': [float(bid[1]) for bid in bids], 'ask_price': [float(ask[0]) for ask in asks], 'ask_qty': [float(ask[1]) for ask in asks] } df_order_book = pd.DataFrame([order_book]) if previous_order_book is not None: # 计算订单消失量 for level in range(10): bid_price = df_order_book['bid_price'].iloc[0][level] ask_price = df_order_book['ask_price'].iloc[0][level] bid_qty = df_order_book['bid_qty'].iloc[0][level] ask_qty = df_order_book['ask_qty'].iloc[0][level] prev_bid_qty = previous_order_book['bid_qty'].iloc[0][level] prev_ask_qty = previous_order_book['ask_qty'].iloc[0][level] # 计算bid订单消失量 if bid_price not in order_disappearances: order_disappearances[bid_price] = 0 if prev_bid_qty > bid_qty: order_disappearances[bid_price] += (prev_bid_qty - bid_qty) # 计算ask订单消失量 if ask_price not in order_disappearances: order_disappearances[ask_price] = 0 if prev_ask_qty > ask_qty: order_disappearances[ask_price] += (prev_ask_qty - ask_qty) previous_order_book = df_order_book calculate_fill_probabilities() def on_error(_ws, error): logger.error(error) def on_open(_ws): print("### opened ###") # Create a WebSocket app ws_trade = websocket.WebSocketApp(SOCKET_TRADE, on_message=on_message_trade, on_error=on_error) ws_depth = websocket.WebSocketApp(SOCKET_DEPTH, on_message=on_message_depth, on_error=on_error) # 定义要传递给 run_forever 的参数 http_proxy_host = "127.0.0.1" http_proxy_port = 7890 proxy_type = "http" # Run the WebSocket with proxy settings trade_thread = threading.Thread(target=ws_trade.run_forever, kwargs={ 'http_proxy_host': http_proxy_host, 'http_proxy_port': http_proxy_port, 'proxy_type': proxy_type }) depth_thread = threading.Thread(target=ws_depth.run_forever, kwargs={ 'http_proxy_host': http_proxy_host, 'http_proxy_port': http_proxy_port, 'proxy_type': proxy_type }) trade_thread.start() depth_thread.start() stop_event = threading.Event() # Function to calculate fill probabilities def calculate_fill_probabilities(): global order_executions, order_disappearances, fill_probabilities fill_probabilities = {} for price in order_disappearances: if price in order_executions: disappearances = order_disappearances[price] executions = order_executions[price] # 确保成交概率不大于1 fill_probabilities[price] = min(executions / disappearances, 1) if disappearances > 0 else 0 else: fill_probabilities[price] = 0 # Function to periodically log and save fill probabilities def log_and_save_fill_probabilities(): global df_order_book, fill_probabilities while not stop_event.is_set(): if fill_probabilities and not df_order_book.empty: asks = [[price, fill_probabilities[price]] for price in df_order_book['ask_price'].iloc[0] if price in fill_probabilities] bids = [[price, fill_probabilities[price]] for price in df_order_book['bid_price'].iloc[0] if price in fill_probabilities] last_price = last_trade['price'] last_qty = last_trade['qty'] side = last_trade['side'] data = { "asks": asks, "bids": bids, "last_price": last_price, "last_qty": last_qty, "side": side, "time": int(time.time() * 1000) } messages.put(data) # logger.info("Market Snapshot:\n%s", json.dumps(data)) stop_event.wait(0.1) # 每5秒更新一次 # 启动定期保存和打印线程 log_and_save_thread = threading.Thread(target=log_and_save_fill_probabilities) log_and_save_thread.start() async def ws_inited(ws, path): logger.info("客户端已连接上，ws://localhost:6789初始化完毕") while True: message = await asyncio.get_running_loop().run_in_executor(None, messages.get) # 检查连接是否仍然开放 if ws.open: message_data = json.dumps(message) # 将字典序列化为JSON字符串 await ws.send(message_data) else: logger.info("WebSocket 连接已关闭") break async def start_websocket_server(): server = await websockets.serve(ws_inited, "localhost", 6789) logger.info("WebSocket 服务器启动在 ws://localhost:6789") await server.wait_closed() def stop_all_threads(): stop_event.set() trade_thread.join() depth_thread.join() log_and_save_thread.join() # 启动 WebSocket 服务器 asyncio.run(start_websocket_server()) # 停止所有线程（在需要停止时调用） # stop_all_threads()