Бот необходим для поиска оптимальных пар для торговли, которые удовлетворяют установленным условиям.
В частности можно задать временной диапазон исследования, количество дней соответствующих минимальному проценту ежедневного отклонения курса в указанном временном периоде, удержание цены в определенном коридоре относительно среднесуточных значений. Оптимально использовать для установки нейтральных торговых роботов. Позволяет поднять PnL в несколько раз.
Заказано: ООО «Открытый институт». Используется в качестве учебного пособия для обучающихся экономического факультета.
Исходный код bybit.py
Python
from pybit.unified_trading import HTTP
import time
import telebot
import time
from aiogram.types import ReplyKeyboardRemove, \
ReplyKeyboardMarkup, KeyboardButton, \
InlineKeyboardMarkup, InlineKeyboardButton
import json
import requests
import json
import time
from operator import itemgetter
days=10
pod_days=5
z=2
f=10
bot = telebot.TeleBot('API_TOKEN', threaded=False)
zapom=[]
sborik=[]
def solve_new(days,pod_days,z,f):
time_now=int(str(time.time()).split('.')[0]+'000')
time_prev=time_now-(days*24*3600*1000)
print(time_now)
print(time_prev)
'''
time_now=1706400000000
time_prev=1706054400000
'''
spis=[]
file=open('resss.txt', encoding='utf-8')
for i in file:
spis.append(i.strip())
file.close()
'''
session = HTTP(testnet=True)
pu=session.get_tickers(category="linear")
print()
#input('csdcsd')
spis=[]
file=open('resss.txt','w', encoding='utf-8')
for to in pu['result']['list']:
spis.append(to['symbol'])
file.write(to['symbol']+'\n')
file.close()
'''
print(len(spis))
sche=0
slovar={}
if(days<=3):
for sym in spis[:]:
try:
sche=sche+1
slovar[sym]=[]
session = HTTP(testnet=True)
prom=session.get_kline(
category="linear",
symbol=sym,
interval='60',
start=time_prev,
end=time_now,
)
for re in prom['result']['list']:
slovar[sym].append([float(i) for i in re])
print(sche,prom)
except:
print('ERRORRR!!!!!!!')
else:
for sym in spis[:]:
try:
sche=sche+1
slovar[sym]=[]
session = HTTP(testnet=True)
prom=session.get_kline(
category="linear",
symbol=sym,
interval='D',
start=time_prev,
end=time_now,
)
for re in prom['result']['list']:
slovar[sym].append([float(i) for i in re])
print(sche,prom)
except:
print('ERRORRR!!!!!!!')
#input('dfvfd')
result=[]
for sym in slovar:
try:
max=0
min=1000000000000
count=0
sum=0
new_sum_otkl=0
print(sym)
for data in slovar[sym]:
print(data)
sum=sum+((data[1]+data[4])/2)
if(data[2]>max):
max=data[2]
if(data[3]<min):
min=data[3]
otkl_min=abs(((data[1]-data[3])/data[1])*100)
otkl_max=abs(((data[1]-data[2])/data[1])*100)
new_otlk=((data[2]-data[3])/data[3])*100
new_sum_otkl=new_sum_otkl+new_otlk
print(otkl_max,otkl_min)
if(otkl_min>=z or otkl_max>=z):
count=count+1
print('URAAAAA')
mean_price=sum/len(slovar[sym])
print(count)
print(mean_price)
print(min,max)
all_otkl_min=abs(((mean_price-min)/mean_price)*100)
all_otkl_max=abs(((mean_price-max)/max)*100)
print(all_otkl_min,all_otkl_max)
if(count>=pod_days and all_otkl_min<=f and all_otkl_max<=f):
print('!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!!1')
result.append([sym,max,min,count,all_otkl_min,all_otkl_max,new_sum_otkl])
except:
print('ERROR!!!!')
all_sym=[]
res_sym={}
sorted_list = sorted(result, key=itemgetter(6), reverse=True)
for res in sorted_list:
print('@@@@@', res)
all_sym.append(res[0])
res_sym[res[0]]=res[-1]
sorted_footballers_by_goals = sorted(res_sym.items(), key=lambda x:x[1], reverse=True)
converted_dict = dict(sorted_footballers_by_goals)
print(converted_dict)
print(len(sorted_list))
all_soob=[]
soob=''
podd=0
for con in converted_dict:
podd=podd+1
if(len(soob+str(podd)+'. '+con+' '+'('+str(round(converted_dict[con],1))+')\n')>4000):
all_soob.append(soob)
soob=''
soob=soob+str(podd)+'. '+con+' '+'('+str(round(converted_dict[con],1))+')\n'
all_soob.append(soob)
#soob='\n'.join(all_sym)
if(all_soob==['']):
all_soob=['Данных по данным фильтрам не найдено!']
print(all_soob)
return [all_soob,sorted_list]
@bot.message_handler(commands=['start'])
def start_command(message):
msg11=bot.send_message(message.chat.id, 'Для биржи байбит перебрать все пары и выдать список тех, которые удовлетворяют условиям: за Х прошедших дней Y дней суточное изменение было не менее Z%, но при этом общее отклонение от средней составило не более F%.')
msg11=bot.send_message(message.chat.id, 'Введите X,Y,Z,F в формате \n\nX\nY\nZ\nF')
bot.register_next_step_handler(msg11, telephone)
def telephone(message):
global zapom
try:
keyboard = InlineKeyboardMarkup()
b1 = InlineKeyboardButton(text='Получить данные', callback_data='real')
keyboard.add(b1)
zapom=message.text.split('\n')
zapom=[int(i.strip()) for i in zapom]
bot.send_message(message.chat.id, 'X: '+message.text.split('\n')[0]+'\nY: '+message.text.split('\n')[1]+'\nZ: '+message.text.split('\n')[2]+'\nF: '+message.text.split('\n')[3],reply_markup=[keyboard])
except:
bot.send_message(message.chat.id,'Ошибка! Ввести новые данные')
start_command(message)
@bot.callback_query_handler(func=lambda call: True)
def step2(call):
global zapom
global sborik
try:
if call.data == 'real':
bot.send_message(call.message.chat.id, 'Ожидаем...')
sooobs=solve_new(zapom[0],zapom[1],zapom[2],zapom[3])
if(sooobs[0]!=['Данных по данным фильтрам не найдено!']):
for text in sooobs[0]:
bot.send_message(call.message.chat.id, text)
time.sleep(1)
#start_command(call.message)
next_step(call.message,sooobs[1])
else:
bot.send_message(call.message.chat.id, 'Данных по данным фильтрам не найдено!')
start_command(call.message)
if call.data == 'da':
next_step(call.message,sborik)
if call.data == 'net':
start_command(call.message)
except:
bot.send_message(call.message.chat.id,'Ошибка! Ввести новые данные')
start_command(call.message)
def next_step(message,data_real):
global zapom
try:
sd=bot.send_message(message.chat.id, 'Введите номер пары, чтобы получить рекомендуемые параметры бота.')
bot.register_next_step_handler(sd, second_step,data_real)
except:
bot.send_message(message.chat.id,'Ошибка! Ввести новые данные')
start_command(message)
def second_step(message,data_real):
global zapom
try:
kapa=int(message.text.strip())
sd=bot.send_message(message.chat.id, 'Введите % отклонения коридора.')
bot.register_next_step_handler(sd, third_step,kapa,data_real)
except:
bot.send_message(message.chat.id,'Ошибка! Ввести новые данные')
start_command(message)
def third_step(message,kapa,data_real):
global zapom
global sborik
try:
#kapa=int(message.text.strip())
#sd=bot.send_message(message.chat.id, 'Введите % отклонения коридора.')
#bot.register_next_step_handler(sd, third_step,kapa,data_real)
print(message.text,kapa,data_real)
kapa=kapa-1
sborik=data_real
bot.send_message(message.chat.id,data_real[kapa][0]+'\n'+str(data_real[kapa][2]-((data_real[kapa][2]/100)*int(message.text.strip())))+' - '+str(data_real[kapa][1]+((data_real[kapa][1]/100)*int(message.text.strip()))))
keyboard = InlineKeyboardMarkup()
b1 = InlineKeyboardButton(text='Да', callback_data='da')
b2 = InlineKeyboardButton(text='Нет', callback_data='net')
keyboard.add(b1)
keyboard.add(b2)
bot.send_message(message.chat.id, 'Выдать параметры для еще одной пары ?',reply_markup=[keyboard])
#start_command(message)
except:
bot.send_message(message.chat.id,'Ошибка! Ввести новые данные')
start_command(message)
while(True):
try:
bot.polling()
except:
print('Error!!!!')