影刀RPA WebSocket实时推送：订阅行情与事件通知

title: “影刀RPA WebSocket实时推送订阅行情与事件通知”date: 2026-07-01author: 林焱影刀RPA WebSocket实时推送订阅行情与事件通知股票行情、交易所推送、IoT设备状态……这类数据不是你去查而是它来推用的是WebSocket长连接协议。影刀可以维持WebSocket连接实时接收推送数据第一时间触发响应。什么情况用什么适合WebSocket的场景实时行情监控股票、期货、加密货币价格订单系统推送新订单到来立即处理设备状态监控传感器数据实时上报即时通讯数据采集聊天室、客服对话不适合的场景数据更新频率很低每小时一次定时轮询就够服务端不支持WebSocket只有REST接口拼多多店群自动化上架方案怎么做方法1连接WebSocket接收实时数据【影刀操作】新建流程「WebSocket实时监控」添加【Python】指令importasyncioimportwebsocketsimportjsonimporttime received_messages[]should_stopFalseasyncdefsubscribe_market_data():订阅某交易所行情WebSocketuriwss://stream.example.com/ws/tickersubscribe_msg{method:SUBSCRIBE,params:[btcusdtticker,ethusdtticker],id:1}whilenotshould_stop:try:asyncwithwebsockets.connect(uri,ping_interval20,ping_timeout10)asws:# 发送订阅消息awaitws.send(json.dumps(subscribe_msg))print(WebSocket连接成功开始接收数据)whilenotshould_stop:try:msgawaitasyncio.wait_for(ws.recv(),timeout30)datajson.loads(msg)# 过滤心跳和订阅确认消息ifresultindata:continue# 处理行情数据ifdata.get(e)24hrTicker:symboldata[s]pricefloat(data[c])change_pctfloat(data[P])print(f{symbol}:{price}({change_pct:.2f}%))received_messages.append({symbol:symbol,price:price,change_pct:change_pct,timestamp:time.time()})# 价格预警ifsymbolBTCUSDTandprice70000:yda.set_variable(alert_triggered,True)yda.set_variable(alert_message,fBTC突破70000当前价格{price})exceptasyncio.TimeoutError:print(接收超时发送心跳包)awaitws.ping()exceptwebsockets.exceptions.ConnectionClosedase:print(f连接断开{e}5秒后重连)awaitasyncio.sleep(5)exceptExceptionase:print(f发生错误{e}5秒后重连)awaitasyncio.sleep(5)# 运行30分钟asyncdefmain():globalshould_stop taskasyncio.create_task(subscribe_market_data())awaitasyncio.sleep(1800)# 运行1800秒should_stopTrueawaittask asyncio.run(main())importjson yda.set_variable(ws_data,json.dumps(received_messages[-100:],ensure_asciiFalse))# 保存最近100条print(f共接收{len(received_messages)}条消息)方法2根据WebSocket数据触发操作接收到特定条件的数据后触发后续业务流程。【影刀操作】在WebSocket监听流程后添加【条件判断】指令判断条件变量alert_triggered等于 True如果为真添加【Python】指令发送企微通知importrequests alert_msgyda.get_variable(alert_message)webhook_urlhttps://qyapi.weixin.qq.com/cgi-bin/webhook/send?keyxxxpayload{msgtype:text,text:{content:f【价格预警】{alert_msg}}}requests.post(webhook_url,jsonpayload)print(企微告警已发送)方法3WebSocket服务端影刀做推送服务影刀主动推送数据给其他系统。【影刀操作】importasyncioimportwebsocketsimportjson connected_clientsset()asyncdefhandler(websocket,path):connected_clients.add(websocket)print(f新客户端连接当前连接数{len(connected_clients)})try:awaitwebsocket.wait_closed()finally:connected_clients.discard(websocket)asyncdefbroadcast(message):广播消息给所有连接的客户端ifconnected_clients:awaitasyncio.gather(*[ws.send(message)forwsinconnected_clients])asyncdefmain():# 启动WebSocket服务器asyncwithwebsockets.serve(handler,0.0.0.0,8765):print(WebSocket服务器启动端口8765)# 模拟定时推送数据foriinrange(100):datajson.dumps({type:update,value:i,ts:asyncio.get_event_loop().time()})awaitbroadcast(data)awaitasyncio.sleep(1)asyncio.run(main())有什么坑坑1连接频繁断开服务端有心跳超时机制不定期发心跳包连接会被踢掉。解决方法设置ping_interval20让websockets库自动维持心跳同时在外层加重连循环。坑2消息堆积处理不过来TEMU店群如何管理运营行情推送很快每秒几十条Python处理速度跟不上消息队列积压。解决方法用asyncio.Queue做缓冲接收和处理分离queueasyncio.Queue(maxsize1000)# 接收协程只往队列里放awaitqueue.put(msg)# 处理协程从队列取msgawaitqueue.get()坑3连接数据乱序网络不稳定时消息可能乱序到达。解决方法消息里加序列号处理前先排序。坑4服务端需要握手鉴权很多WebSocket服务需要在连接URL里带token或者先发鉴权消息。解决方法参考服务端文档连接时在URL里带?tokenxxx或者连接后第一条消息发鉴权包。总结WebSocket实时数据处理的关键在三点断线重连机制、心跳保活、消息处理不阻塞接收。在影刀里用asyncio处理WebSocket是标准方案把接收和处理逻辑分开系统就能稳定跑下去。