Бот позволяет проанализировать историю торгов за определенное количество дней, а также эмулировать работу торгового робота с определенным диапазоном гридов. Это позволяет спрогнозировать наиболее эффективный шаг сетки для данной конкретной торговой пары.
Заказано: ООО «Открытый институт». Используется в качестве учебного пособия для обучающихся экономического факультета.
Исходный код grid_bybit.py
Python
from pybit.unified_trading import HTTP
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
zapom=[]
bot = telebot.TeleBot('API_TOKEN', threaded=False)
def spis_time(time_now,time_prev):
res_time=[]
pro=time_prev
while(True):
if((pro+600*60000)>time_now):
res_time.append([pro,time_now])
break
else:
res_time.append([pro,pro+600*60000])
pro=pro+600*60000
return res_time
def create_list(a,b,dgrid):
spis=[]
spis.append(a)
for i in range(dgrid):
spis.append(a+(i+1)*((b-a)/dgrid))
return spis
def create_dgid(x,y,z):
spis=[]
spis.append(x)
if(x>=y):
aaaa=1/0
while(True):
if((x+z)>=y):
spis.append(y)
break
else:
spis.append(x+z)
x=x+z
return spis
def count_proh(min,max,diap):
pos_min=0
pos_max=0
for i in range(len(diap)-1):
if(min>=diap[i] and min<diap[i+1]):
pos_min=i+1
if(max>diap[i] and max<=diap[i+1]):
pos_max=i+1
if(min>diap[-1]):
pos_min=len(diap)
if(max>diap[-1]):
pos_max=len(diap)
return [pos_max-pos_min,pos_min,pos_max]
def res_count(sym,times_list,a,b,dgrid_niz,dgrid_verh,dgrid_shag,plecho):
prev_times=[]
summury=[]
for time_dia in times_list:
session = HTTP(testnet=True)
prom=session.get_kline(
category="linear",
symbol=sym,
interval=1,
start=time_dia[0],
end=time_dia[1],
limit=1000
)
print(len(prom['result']['list']))
for para in prom['result']['list']:
para=[float(i) for i in para]
if(para[0] not in prev_times):
prev_times.append(para[0])
summury.append(para)
all_grids=create_dgid(dgrid_niz,dgrid_verh,dgrid_shag)
#print(all_grids)
allres_grids={}
for dgrid in all_grids:
result=0
setka=create_list(a,b,dgrid)
#print(dgrid,setka)
for para in summury:
tek_result=count_proh(para[3],para[2],setka)
result=result+tek_result[0]
intr=(((b-a)/dgrid)*result)*plecho
allres_grids[dgrid]=[intr,result]
sorted_footballers_by_goals = sorted(allres_grids.items(), key=lambda x:x[1][0], reverse=True)
converted_dict = dict(sorted_footballers_by_goals)
print(converted_dict)
'''
max_data=0
max_dgrid=0
for i in allres_grids:
if(allres_grids[i][0]>max_data):
max_data=allres_grids[i][0]
max_dgrid=i
print(i,allres_grids[i][1],allres_grids[i][0])
'''
#return [max_dgrid,str(round(max_data,2)),allres_grids[max_dgrid][1]]
return converted_dict
@bot.message_handler(commands=['start'])
def start_command(message):
msg11=bot.send_message(message.chat.id, 'Введите запрос в формате:\n\nПара\nКоличество дней\nНиз диапазона-Верх диапазона\nДиапазон гридов\nШаг грида\nПлечо')
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, 'Пара: '+message.text.split('\n')[0]+'\nКоличество дней: '+message.text.split('\n')[1]+'\nНиз диапазона-Верх диапазона: '+message.text.split('\n')[2]+'\nДиапазон гридов: '+message.text.split('\n')[3]+'\nШаг грида: '+message.text.split('\n')[4]+'\nПлечо: '+message.text.split('\n')[5],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
try:
if call.data == 'real':
bot.send_message(call.message.chat.id, 'Ожидаем...')
spis=[]
file=open('coins.txt', encoding='utf-8')
for i in file:
spis.append(i.strip())
file.close()
zapom[1]=int(zapom[1])
a=float(zapom[2].split('-')[0].strip())
b=float(zapom[2].split('-')[1].strip())
dgrid_niz=int(zapom[3].split('-')[0].strip())
dgrid_verh=int(zapom[3].split('-')[1].strip())
dgrid_shag=int(zapom[4].strip())
plecho=int(zapom[5].strip())
if(zapom[0].upper().strip() not in spis):
bot.send_message(call.message.chat.id, 'Данной пары не найдено!\nВыбирайте пары из данного файла.')
bot.send_document(call.message.chat.id,open('coins.txt', 'rb'))
start_command(call.message)
else:
time_now=int(str(time.time()).split('.')[0]+'000')
time_prev=time_now-(zapom[1]*24*3600*1000)
puu=spis_time(time_now,time_prev)
sooobs=res_count(zapom[0].upper().strip(),puu,a,b,dgrid_niz,dgrid_verh,dgrid_shag,plecho)
res_soob='Лучшие результаты:\n'
sche=0
for cu in sooobs:
sche=sche+1
if(sche>10):
break
try:
res_soob=res_soob+'='+str(cu)+', $'+str(sooobs[cu][1])+', '+str(round(sooobs[cu][0],2))+'%'+'\n'
except:
pass
#bot.send_message(call.message.chat.id,'Лучший результат дало %d сеток - %s'%(sooobs[0],sooobs[1])+' %.\nКоличество пересечений: '+str(sooobs[2]))
bot.send_message(call.message.chat.id,res_soob)
bot.send_message(call.message.chat.id,'Текущий коридор составляет '+str(round(((b - a)/(a/100)),2))+'%, что потребует при плече '+str(plecho)+' подушки безопасности в '+str(round((((b - a)/(a/100))/(100/plecho))/20,2))+'х')
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)
while(True):
try:
bot.polling()
except:
print('Error!!!!')