[网络协议]python监控Hoo交易所websocket示例代码并附解密函数

import websocket
import json


import os
os.environ["http_proxy"] = "http://127.0.0.1:7890"
os.environ["https_proxy"] = "http://127.0.0.1:7890"
import base64
import time
import random
from Crypto.Cipher import AES
from Crypto.Util.Padding import pad

def get_ran():
    n = ["0", "1", "2", "3", "4", "5", "6", "7", "8", "9", "a", "b", "c", "d", "e", "f", "g", "h", "i", "j", "k", "l", "m",
         "n", "o", "p", "q", "r", "s", "t", "u", "v", "w", "x", "y", "z", "A", "B", "C", "D", "E", "F", "G", "H", "I", "J",
         "K", "L", "M", "N", "O", "P", "Q", "R", "S", "T", "U", "V", "W", "X", "Y", "Z"]
    a = ''
    for _ in range(13):
        i = round(random.random()*(len(n)-1))
        a+=n[i]
    return a
def encrypt_mobile(key,text):
    key = key.encode('utf-8')
    mode = AES.MODE_CBC
    cryptos = AES.new(key, mode,IV='develop-zhenjing'.encode())
    cipher_text = cryptos.encrypt(pad(text.encode(),8))
    return base64.b64encode(cipher_text).decode()
def get_login_hoo():
    rn = get_ran()
    key = 'hoo'+rn
    t_ = str(int(time.time()))
    text = "{\"nonce\":\""+rn+"\",\"ts\":"+t_+"}"
    t_ = str(int(time.time()))
    s = '{"op":"loginws","data":"'+encrypt_mobile(key,text)+'","token":"","nonce":"'+rn+'","ts":'+t_+'}'
    print(s)
    return s
def on_message(ws, message):
    # print(message)
    # print(type(message))
    if 'login success' in message:
        # ws.send('{"op":"sub","topic":"quote:BTC-USDT"}')
        ws.send('{"op":"sub","topic":"kline:15Min:BTC-USDT"}')
    else:
        try:
            dict1 = json.loads(message)
            if dict1.get('ticks'):
                print(dict1.get('ticks')[0].get('timestamp'),dict1.get('ticks')[0].get('close'))
         
        except Exception as e:
            print(str(e))
def on_error(ws, error):
    print(error)
def on_close(ws, close_status_code, close_msg):
    print("### closed ###")
def on_open(ws):
    print("Opened connection")
    ws.send(get_login_hoo())
if __name__ == "__main__":
    while True:
        try:
            ws = websocket.WebSocketApp("wss://ws.hoowss.com/wsf",
                                        keep_running=True,

                                      on_open=on_open,
                                      on_message=on_message,
                                      on_error=on_error,
                                      on_close=on_close)

            ws.run_forever()  # Set dispatcher to automatic reconnection
        except:
            pass

有可能需要翻墙代理
os.environ["http_proxy"] = "http://127.0.0.1:7890"
os.environ["https_proxy"] = "http://127.0.0.1:7890"

不需要就注释掉，或者改成自己的代理

websocket连通后要发送登录指令，登陆有加密，此代码包含登录解密，默认可直接运行

创作打卡挑战赛

赢取流量/现金/CSDN周边激励大奖