我正在使用Binance的API。我正在連接他們的API，試圖評估Binance在其平臺上是否有資產清單。下面是資產清單：
assets = ['tribe', 'pax']
我通過在SOCKET鏈接中插入資產的名稱將其傳遞給他們的API：
SOCKET = f "wss://stream.binance.com:9443/ws/{asset}usdt@ticker"
如果on_message被呼叫，我知道資產確實存在于他們的網站上，因為那時我已經完成了與他們的API的一致連接，它將繼續列印訊息，除非我關閉連接（我這樣做）。然而，如果在N時間內沒有收到訊息，我就知道他們沒有我正在尋找的資產。在這種情況下，Binance確實有tribe，但沒有pax。如果n次后資產不在他們的網站上，我想關閉連接，我如何做到這一點？
import ssl
import websocket
def on_open（ws）。
print('connection: successful')
def on_close(ws, *args) 。
print('connection: lost')
print("---------------------------------------------------")
ws.close()
def on_message（ws, message）。
print("message received")
print()
ws.close()
def on_error（ws, message）。
print(message)
print()
assets = ['tribe', 'pax']
for assets in assets。
print(asset)
SOCKET = f "wss://stream.binance.com:9443/ws/{asset}usdt@ticker"
ws = websocket.WebSocketApp(SOCKET, on_open=on_open, on_close=on_close, on_message=on_message,
on_error=on_error)
ws.run_forever(sslopt={"cert_reqs"/span>: ssl.CERT_NONE})
我曾嘗試寫道：
if on_message == False:
ws.close()
然而，這并不奏效，因為據我所知，on_message甚至沒有被呼叫。
這里是Binance的API檔案。 https://github.com/binance/binance-spot-api-docs/blob/master/web-socket-streams.md#klinecandlestick-streams 這里是websocket-客戶端檔案。 https://websocket-client.readthedocs.io/en/latest/examples.html
uj5u.com熱心網友回復：
試試這個：-
import websocket
import ssl
import time
from threading import Thread
class Binance（）。
def __init__(self, asset, timeout=5) 。
self.url = f'wss://stream.binance.com:9443/ws/{asset}usdt@ticker'
self.ws = None
self.mr = False[/span]。
self.timeout = timeout
def start（self）。
self.ws = websocket.WebSocketApp(self.url, on_message=self.on_message)
Thread(target=self.monitor).start()
self.ws.run_forever(sslopt={"cert_reqs"/span>: ssl.CERT_NONE})
def on_message（self, ws, message）。
self.mr = True。
print(message)
def stop（self）。
if self.ws:
self.ws.close()
self.ws = None.
def monitor（self）。
while True:
time.sleep(self.timeout)
if not self.mr:
自我.停止()
break
self.mr = False[/span]。
def runner（asset）。
Binance(asset).start()
for asset in ['pax', 'trib']:
Thread(target=runner, args=(asset,)).start()
uj5u.com熱心網友回復：
這很粗糙，但它可以作業
。
import ssl
import websocket
import requests
import json
import pprint
def on_open（ws）。
print('connection: successful')
def on_close(ws, *args) 。
print('connection: lost')
print()
ws.close()
def on_message（ws, message）。
json_message = json.load(message)
pprint.pprint(json_message)
ws.close()
def on_error（ws, message）。
print("error"/span>)
print()
assets = ['TRIBE'/span>, 'PAX'/span>, 'OGN'/span>, 'RAI'/span>, 'QNT'/span>, 'BTC'/span>]
for assets in assets。
url = f "https://www.binance.com/en/trade/{asset}_USDT"
soup = requests.get(url).text
if "https://bin.bnbstatic.com/static/images/electron/404-error.png" in soup:
print(f"{asset} not on coinbase"/span>)
else:
print(asset, "on coinbase")
SOCKET = f "wss://stream.binance.com:9443/ws/{asset.lower()}usdt@ticker"
print(SOCKET)
ws = websocket.WebSocketApp(SOCKET, on_open=on_open, on_close=on_close, on_message=on_message, on_error=on_error)
ws.run_forever(sslopt={"cert_reqs"/span>: ssl.CERT_NONE})
print("------------------------------")
print()
此外，這并沒有使用你所想的方法，但它確實像你想的那樣作業（我想）。
轉載請註明出處，本文鏈接：https://www.uj5u.com/shujuku/332633.html
標籤：
下一篇：MatlabMEX-函式的副作用