# pip install websockets msgpack zstandard psutil import asyncio import json import random import os import ssl import time import threading import websockets import msgpack import zstandard as zstd import psutil from datetime import datetime # 1. 初始化客户端解压引擎 dctx = zstd.ZstdDecompressor() # 2. 配置 WebSocket 服务端地址 domain = "zhunData.cn" WS_URI = f"wss://{domain}/ws/" AUTH_KEY = "你的key" # 3. 重连配置 RECONNECT_MIN = 5 # 最小重连等待time（秒） RECONNECT_MAX = 10 # 最大重连等待time（秒） # 4. 客户端内存日志文件路径 CLIENT_MEMORY_LOG_FILE = "client_memory_monitor.log" CLIENT_TICK_SIZE_LOG_FILE = "client_tick_data_size.log" _recv_size_lock = threading.Lock() _recv_size_hourly_bytes = 0 _recv_size_trading_bytes = 0 _current_hour = -1 _trading_day = None async def memory_monitor(): """ 异步内存监控协程：每隔 60 秒将客户端内存占用写入日志文件 """ process = psutil.Process(os.getpid()) while True: await asyncio.sleep(60) try: mem_mb = process.memory_info().rss / 1024 / 1024 log_time = datetime.now().strftime('%Y-%m-%d %H:%M:%S') log_msg = f"[{log_time}] Client RSS Memory: {mem_mb:.2f} MB\n" # 追加写入到日志文件 with open(CLIENT_MEMORY_LOG_FILE, "a", encoding="utf-8") as f: f.write(log_msg) except Exception as e: print(f"[Client Monitor Error] {e}") async def tick_data_size_monitor(): global _recv_size_hourly_bytes, _recv_size_trading_bytes, _current_hour, _trading_day while True: await asyncio.sleep(60) try: now = datetime.now() hour = now.hour today = now.date() with _recv_size_lock: hourly_bytes = _recv_size_hourly_bytes trading_bytes = _recv_size_trading_bytes log_time = now.strftime('%Y-%m-%d %H:%M:%S') if hour != _current_hour and _current_hour != -1: with _recv_size_lock: _recv_size_hourly_bytes = 0 hourly_mb = hourly_bytes / 1024 / 1024 log_msg = f"[{log_time}] 上一h Tick 数据接收量: {hourly_mb:.2f} MB ({hourly_bytes} bytes)\n" with open(CLIENT_TICK_SIZE_LOG_FILE, "a", encoding="utf-8") as f: f.write(log_msg) _current_hour = hour if 9 <= hour <= 15: if today != _trading_day: if _trading_day is not None: with _recv_size_lock: _recv_size_trading_bytes = 0 trading_mb = trading_bytes / 1024 / 1024 log_msg = f"[{log_time}] 上一交易日(9-15点) Tick 数据接收量: {trading_mb:.2f} MB ({trading_bytes} bytes)\n" with open(CLIENT_TICK_SIZE_LOG_FILE, "a", encoding="utf-8") as f: f.write(log_msg) _trading_day = today else: if _trading_day is not None and today != _trading_day: with _recv_size_lock: _recv_size_trading_bytes = 0 trading_mb = trading_bytes / 1024 / 1024 log_msg = f"[{log_time}] 上一交易日(9-15点) Tick 数据接收量: {trading_mb:.2f} MB ({trading_bytes} bytes)\n" with open(CLIENT_TICK_SIZE_LOG_FILE, "a", encoding="utf-8") as f: f.write(log_msg) _trading_day = None except Exception as e: print(f"[Client Tick Size Monitor Error] {e}") async def subscribe_client(ws): """ 封装订阅逻辑，方便在每次重连成功后重新执行 """ # 订阅Market quote sub_quote_msg = { "action": "subscribe", "channel": "market_quote", "symbols": [ "tick.stock.*", # 订阅所有股票 # "tick.index.*", # 订阅深市所有指数 # "tick.fund.*", # 订阅所有基金 # "tick.bond.*", # "tick.fund.sz.159227", # "tick.stock.sz.000001", # "tick.stock.sh.600744", ] } await ws.send(json.dumps(sub_quote_msg)) response = await ws.recv() print(f"[*] Server confirmed: {response}") # # 订阅涨停提示 # sub_alert_msg = { # "action": "subscribe", # "channel": "limit_alert", # "symbols": ["stock.*"] # 订阅所有股票的Limit-up alert # } # await ws.send(json.dumps(sub_alert_msg)) # response = await ws.recv() # print(f"[*] Server confirmed: {response}") def _handle_text_message(text_data: dict): msg_type = text_data.get("type", "unknown") if msg_type == "key_expiring_soon": print(f"⚠️ [续订提醒] Key expiring soon | days_remaining: {text_data.get('days_remaining')} | expired_at: {text_data.get('expired_at')}") elif msg_type == "key_expired_warning": print(f"⚠️ [过期警告] Key expired | grace_remaining: {text_data.get('grace_hours_remaining')}h | expired_at: {text_data.get('expired_at')}") elif msg_type == "key_expired_disconnect": print(f"🚫 [断开连接] Key expired timeout, server disconnecting | {text_data.get('msg')}") elif msg_type == "subscribe_ok": print(f"✅ [Subscribe OK] channel: {text_data.get('channel')} | symbol: {text_data.get('symbols')}") elif msg_type == "subscribe_rejected": print(f"❌ [Subscribe rejected] {text_data.get('msg')}") elif msg_type == "error": print(f"❌ [Server error] {text_data.get('msg')}") else: print(f"📩 [Text message] {text_data}") def _handle_binary_message(tick_data: dict): msg_type = tick_data.get("type", "unknown") code = tick_data.get("key", "N/A") data = tick_data.get("data", "N/A") if msg_type == "limit_alert": print(f"🚨 [Limit-up alert] symbol: {code} | 数据: {tick_data}") elif msg_type == "market_quote": tick_time = data.get('time') datetime_str = None if tick_time: datetime_str = datetime.fromtimestamp(tick_time / 1000) print(code) # if code == 'tick.stock.sz.000001': # print(f"📈 [Market quote] symbol: {code} | time: {datetime_str} {data}") else: pass async def receive_data(ws): """ 封装数据接收与解压逻辑 """ global _recv_size_hourly_bytes, _recv_size_trading_bytes async for message in ws: try: if isinstance(message, str): _handle_text_message(json.loads(message)) continue # print(f"[Debug] Received binary message length: {len(message)} bytes, {message}") msg_size = len(message) raw_bytes = dctx.decompress(message) result = msgpack.unpackb(raw_bytes, raw=False) if isinstance(result, list): for item in result: _handle_binary_message(item) else: _handle_binary_message(result) with _recv_size_lock: _recv_size_hourly_bytes += msg_size now_hour = datetime.now().hour if 9 <= now_hour <= 15: _recv_size_trading_bytes += msg_size except Exception as e: print(f"[Error] 解压或解析数据failed: {e}") async def run_client(): """带自动重连机制的主客户端循环""" monitor_task = asyncio.create_task(memory_monitor()) size_monitor_task = asyncio.create_task(tick_data_size_monitor()) while True: try: print(f"[*] Connecting to WebSocket server: {WS_URI} ...") # 通过 URL 参数传递鉴权 Key ws_uri = f"{WS_URI}?key={AUTH_KEY}" _ssl_ctx = ssl.SSLContext(ssl.PROTOCOL_TLS_CLIENT) _ssl_ctx.check_hostname = False _ssl_ctx.verify_mode = ssl.CERT_NONE async with websockets.connect(ws_uri, ssl=_ssl_ctx) as ws: print("[+] Connected!") await subscribe_client(ws) await receive_data(ws) except ( websockets.ConnectionClosed, ConnectionRefusedError, OSError, asyncio.TimeoutError ) as e: print(f"[!] Connection closed or error: {e}") if hasattr(e, 'code') and e.code in (4003, 4001, 4009): print(f"[!] Auth rejected (code={e.code}), not reconnecting") return wait_time = random.randint(RECONNECT_MIN, RECONNECT_MAX) print(f"[*] 将在 {wait_time} s before reconnecting...") await asyncio.sleep(wait_time) except Exception as e: print(f"[Fatal Error] Unknown error: {e}") await asyncio.sleep(RECONNECT_MAX) if __name__ == "__main__": try: # 【核心新增】每次启动客户端时，清空旧的日志文件，方便观察本次运行的数据 if os.path.exists(CLIENT_MEMORY_LOG_FILE): os.remove(CLIENT_MEMORY_LOG_FILE) print(f"[*] Cleared old client memory log: {CLIENT_MEMORY_LOG_FILE}") if os.path.exists(CLIENT_TICK_SIZE_LOG_FILE): os.remove(CLIENT_TICK_SIZE_LOG_FILE) print(f"[*] Cleared old client tick size log: {CLIENT_TICK_SIZE_LOG_FILE}") asyncio.run(run_client()) except KeyboardInterrupt: print("\n[*] Client exited manually.")