能推数据了。

binance_order_flow.py

@@ -43,7 +43,7 @@ fill_probabilities = {}
 order_disappearances = {}
 order_executions = {}
 last_trade = {'price': None, 'qty': None, 'side': None}
-messages = queue.Queue()                        # 创建一个线程安全队列
+messages = queue.Queue()  # 创建一个线程安全队列
 
 
 def on_message_trade(_ws, message):
@@ -155,7 +155,6 @@ def calculate_fill_probabilities():
             fill_probabilities[price] = 0
 
 
-# Function to periodically log and save fill probabilities
 # Function to periodically log and save fill probabilities
 def log_and_save_fill_probabilities():
     global df_order_book, fill_probabilities
@@ -196,22 +195,10 @@ async def ws_inited(ws, path):
             logger.info("WebSocket 连接已关闭")
             break
 
-def start_websocket_server():
-    loop = asyncio.new_event_loop()
-    asyncio.set_event_loop(loop)
-    try:
-        loop.run_until_complete(
-            websockets.serve(ws_inited, "localhost", 6789)
-        )
-        logger.info("WebSocket 服务器启动在 ws://localhost:6789")  # 直接使用 print 输出
-        loop.run_forever()
-    except Exception as e:
-        logger.error(f"无法启动WebSocket服务器: {e}")  # 输出错误信息
-    finally:
-        loop.close()
-
-# 在新线程中启动 WebSocket 服务器
-threading.Thread(target=start_websocket_server, daemon=True).start()
+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()
@@ -219,5 +206,8 @@ def stop_all_threads():
     depth_thread.join()
     log_and_save_thread.join()
 
+# 启动 WebSocket 服务器
+asyncio.run(start_websocket_server())
+
 # 停止所有线程（在需要停止时调用）
-# stop_all_threads()
+# stop_all_threads()