Are you looking to take your cryptocurrency trading to the next level? Look no further than Bybit, the leading crypto derivatives exchange. With advanced trading tools, low fees, and a user-friendly platform, Bybit makes it easy to trade Bitcoin, Ethereum, and other popular cryptocurrencies. And if you sign up using our affiliate link and use the referral code CODEARMO, you'll receive exclusive benefits and bonuses up to $30,000 to help you get started. Don't miss out on this opportunity to join one of the fastest-growing communities in crypto trading. Sign up for Bybit today and start trading like a pro!
To proceed with the steps outlined in this article, it is necessary to create a webhook in the desired Discord chat to which signals can be posted. The process of creating a webhook is simple, and can be accomplished by following the steps below:
-
Click on the settings icon located on the chat in which you intend to create the bot.
-
Navigate to the 'Integrations' tab.
-
Click on 'Webhook', and a purple button labelled 'New Webhook' should appear.
-
Clicking this button will create the webhook, allowing you to name it according to your preferences and select the channel in which you would like to post.
- Finally, copy the webhook URL and paste it in to the python script below
import aiohttp
import asyncio
import pandas as pd
import datetime as dt
from pybit.unified_trading import HTTP
import numpy as np
from discord import SyncWebhook
url = 'https://discordapp.com/api/webhooks/your_webhook from discord goes here'
webhook = SyncWebhook.from_url(url)
bb = HTTP(testnet=False)
result = bb.get_tickers(
category="linear").get('result')['list']
tickers = [asset['symbol'] for asset in result if asset['symbol'].endswith('USDT')]
# remove stablecoin pairs
for ticker in tickers:
if ticker[:4] in ['USDC', 'BUSD']:
tickers.remove(ticker)
Completing the steps above gives us a total of 210 USDT perpetuals to analyze. However, using the code from this article will be unacceptably slow, if we need tickers for say an hourly timeframe. Therefore we will use the aiohttp package as can be seen from the import statements.
What is Aiohttp ?
aiohttp is a Python package that allows for the creation of asynchronous HTTP clients and servers. It provides a simple and efficient way to make HTTP requests and handle responses asynchronously, making it well-suited for building web services and APIs that can handle a large number of concurrent requests. The package is built on top of the Python asyncio library, which allows for non-blocking I/O operations and improves the performance and scalability of web applications.
This package is very useful and I encourage readers to read the full documentation here.
Choosing an Indicator
Well for the purposes of this article we are going to choose a simple moving average crossover. So the steps we need to complete are as follows:
1. Download all the data asynchronously from Bybit API.
2. Calculate the whether the indicator has provided a long or short signal.
3. Post the indicator to our discord chat.
Step 1
We need to create two functions to download the data , the first as shown below is the construct_url , this takes a a base url and allows us to pass symbol, category & interval as arguments. The next step, is to make a function that assigns these requests to an asyncio task. We then need to send these requests using the get_data function shown below, this will retrieve the data for each symbol in the list we defined previously.
def construct_url(category, symbol, interval):
url = f'https://api.bybit.com/v5/market/kline?category={category}&symbol={symbol}&interval={interval}'
return url
def add_tasks(session, category, tickers, interval):
'''
:param session: aiohttp session for async data retrieval
:param tickers: list of tickers generated from ticker list
:param interval: data interval e.g. 60 == 60mins
:return: tasks to be processed by get_data method
'''
tasks = []
for ticker in tickers:
url = construct_url(category=category, symbol=ticker, interval=interval)
tasks.append(asyncio.create_task(session.get(url, ssl=True)))
return tasks
async def get_data(category, tickers, interval):
results = []
async with aiohttp.ClientSession() as session:
tasks = add_tasks(session=session, category=category, tickers=tickers, interval=interval)
responses = await asyncio.gather(*tasks)
for response in responses:
results.append(await response.json())
return results
Now we can collect the data as shown below. It takes only 1 second to download hourly data for over 200 symbols, very cool!
import time
t = time.time()
results = asyncio.run(get_data(category='linear', tickers=tickers, interval=60))
print(time.time() -t)
print(len(results))
Step 2
Next we need to format the data, and calculate the indicator. One potential problem that immediately catches the eye, is the fact that we are calculating a 200 period moving average without checking whether 200 data points exist. Perhaps those that want to put this in to production should think about this and apply some logic to deal with potential issues arising from lack of data.
Note the get_ma_signal function below will return a integer 1 = 'long' , -1 = short and 0 = no signal.
def format_data(response):
'''
Parameters
----------
respone : dict
response from calling get_klines() method from pybit.
freq : str
DESCRIPTION.
interval that has been passed in to get_klines method, used in order
to round the datetimes, which will be mapped to pandas freq str
Returns
-------
dataframe of ohlc data with date as index
'''
data = response.get('list', None)
if not data:
return
data = pd.DataFrame(data,
columns=[
'timestamp',
'open',
'high',
'low',
'close',
'volume',
'turnover'
],
)
f = lambda x: dt.datetime.utcfromtimestamp(int(x) / 1000)
data.index = data.timestamp.apply(f)
return data[::-1].apply(pd.to_numeric)
def get_ma_signal_from_data(response):
data = response.get('result')
if not data:
return
df= format_data(data)
df['ma20'] = df.close.rolling(20).mean()
df['ma200'] = df.close.rolling(200).mean()
df['long_signal'] = np.where((df.ma20 > df.ma200) & (df.ma20.shift(1) < df.ma200.shift(1)), 1, 0)
df['short_signal'] =np.where((df.ma20 < df.ma200) & (df.ma20.shift(1) > df.ma200.shift(1)), -1, 0)
df['signal'] = df.long_signal + df.short_signal
return df.signal[-1:].values[0]
Note that if you wish to change the code above to a different indicator, you would simply change the get_ma_signal_from data to something that looks like the template given below.
def your_indicator(response):
data = response.get('result')
if not data:
return
df= format_data(data)
'''
Logic for your indicator goes here
'''
return 'your signal output'
Step 3
Now that we have functionality to pull the data and calculate the signal. The next step is to generate a message to send to discord, note that we loop through and append to a list of messages, due to the fact that there is a 2000 character limit to send messages through the webhook to discord.
def generate_discord_message(results, tickers):
message = """"""
messages = []
for sym, res in zip(tickers,results):
signal = get_ma_signal_from_data(res)
if len(message) > 1500:
messages.append(message)
message = """"""
if signal == 0:
print(f'{sym} has no signal currently')
elif signal == 1:
message += f'\n{sym} just had a moving average crossover BUY now \U0001F911'
elif signal == -1:
message += f'\n{sym} just had a moving average crossover SELL now \U0001F911'
messages.append(message)
return messages
messages = generate_discord_message(results=results, tickers=tickers)
for text in messages:
if text:
webhook.send(text)
else:
pass
As an example, the message below will be sent through to our discord server. For readers, that would like to put there on version of this in to production and have signed up through our referral link, feel free to ask questions in the discord about the best way to deploy this code to a server.
ZRXUSDT just had a moving average crossover SELL now 