Skip to content

Instantly share code, notes, and snippets.

Embed
What would you like to do?
import asyncio
import websockets
import json
import time
import hmac
def get_args_secret(_api_key, _api_secrete):
expires = str(int(round(time.time())+5000))+"000"
_val = 'GET/realtime' + expires
signature = str(hmac.new(bytes(_api_secrete, "utf-8"), bytes(_val, "utf-8"), digestmod="sha256").hexdigest())
auth = {}
auth["op"] = "auth"
auth["args"] = [_api_key, expires, signature]
args_secret = json.dumps(auth)
return args_secret
async def my_loop_WebSocket_bybit():
async with websockets.client.Connect("wss://stream.bybit.com/realtime") as websocket:
await websocket.send(get_args_secret(api_key, api_secret)); # secret
print("Connected to bybit WebSocket with secret key");
await websocket.send('{"op":"subscribe","args":["trade.BTCUSD"]}');
data_rcv_response = await websocket.recv();
print("response for subscribe req. : " + data_rcv_response);
while True:
...
...
...
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment