Как реализовать функционал отображения индикаторов в телеграм-боте
Индикаторы помогают трейдеру глубже анализировать рынок и принимать более эффективные решения в торговле. В этой статье разберем, как реализовать отображение самых популярных индикаторов на ценовом графике с помощью телеграм-бота — чтобы автоматизировать анализ, правильнее определять тренды и точнее отслеживать движение цен.
Определяем типы индикаторов
Мы будем использовать для анализа через телеграм-бот следующие индикаторы:
OHLC — индикатор OHLC. Показывает цены открытия, максимума, минимума и закрытия для каждого периода.
MA — индикатор Moving Average. Показывает среднее значение цены за выбранный промежуток времени.
VOL — индикатор Volume. Показывает объемы торгов — количество заключенных сделок за единицу времени.
BB — индикатор Bollinger Bands. Показывает отклонения цены на рынке относительно нормального торгового интервала в режиме реального времени.
RSI — индикатор Relative Strength Index. Показывает соотношение положительных и отрицательных изменений цены.
STC — индикатор Stochastic Oscillator. Показывает отношение между ценой закрытия и диапазоном «максимум-минимум» за период в виде процентов.
ZIGZAG — индикатор ZigZag. Показывает наиболее значимые высшие и низшие точки графика.
ALL — дашборд с 3 индикаторами. Показывает индикаторы Moving Average, Volume и Stochastic Oscillator.
Определяем инструменты и параметры для графиков
Для индикаторов определяем следующие параметры: инструмент, интервал, период, значение окна.
Инструменты
BTCUSDT – Биткоин/ Tether
ETHUSD — Эфириум/ Доллар
BTCPERP — Биткоин/ Perpetual Protocol
ETHPERP — Эфириум/ Perpetual Protocol
BTCUSD — Биткоин/ Доллар
XRPUSDT — XRP/ Tether
BNBUSDT — Binance Coin/ Tether
DOGEUSDT — Dogecoin/ Tether
GMTUSDT — GMT/ Tether
Интервалы
1, 3, 5, 15, 30, 60, 120, 240, 360, 720, D, W, M
Периоды
Любое числовое значение в количестве дней — например, интервал в 90 дней.
Значения окна
Количество дней: 1, 3, 5, 10, 30.
Также у индикаторов BB, RSI, STC, ALL можно добавить значение параметра «окно» через нижнее подчеркивание. Например: BB_1, RSI_5, STC_10, ALL_30
Реализация отображения индикаторов: подготовительный этап
Создаем токен для телеграм-бота:
TOKEN = os.environ.get(‘TOKEN’)
Данные получаем с помощью API от сервиса bybit.com:
PATH = ‘https://api.bybit.com/’
ENDPOINT = ‘/v5/market/kline’
URL = PATH + ENDPOINT
Добавляем список инструментов:
SYMBOLS = [‘BTCUSDT’, ‘ETHUSDT’, ‘BTCPERP’, ‘ETHPERP’,
‘BTCUSD’, ‘ETHUSD’, ‘BTCUSDT’,
‘XRPUSDT’, ‘BNBUSDT’, ‘DOGEUSDT’, ‘GMTUSDT’]
Добавляем список графиков:
CHARTS = [‘OHLC’, ‘MA’, ‘VOL’, ‘MPET’, ‘ZIGZAG’,
‘ALL_1’, ‘ALL_5’, ‘ALL_10’, ‘ALL_30’,
‘BB_1’, ‘BB_3’, ‘BB_5’, ‘BB_10’,
‘RSI_1’, ‘RSI_5’, ‘RSI_10’, ‘RSI_30’,
‘STC_1’, ‘STC_5’, ‘STC_10’, ‘STC_30’]
Добавляем список интервалов:
INTERVALS = [1, 3, 5, 15, 30, 60, 120,
240, 360, 720, ‘D’, ‘W’, ‘M’]
Добавляем словарь расшифровок и терминов:
INTERVAL_NAMES = {
1: ‘1 minute’,
3: ‘3 minutes’,
5: ‘5 minutes’,
15: ’15 minutes’,
30: ’30 minutes’,
60: ‘1 hour’,
120: ‘2 hours’,
240: ‘4 hours’,
360: ‘6 hours’,
720: ’12 hours’,
‘D’: ‘day’,
‘W’: ‘week’,
‘M’: ‘month’,
Создаем бота:
bot = telebot.TeleBot(TOKEN)
Создаем функцию для отправки инструкции по формированию запроса после запуска бота командой /start:
symbols_list = (‘BTCUSDT, ETHUSDT, BTCPERP, ‘
‘\n ETHPERP, BTCUSD, ETHUSD, ‘
‘\n BTCUSDT, XRPUSDT, BNBUSDT, ‘
‘\n DOGEUSDT, GMTUSDT’)
intervals_list = (‘1, 3, 5, 15, 30, 60, 120, ‘
‘\n 240, 360, 720, D, W, M’)
charts_list = (‘OHLC, MA, VOL, MPET, ZIGZAG, ‘
‘\n ALL_1, ALL_5, ALL_10, ALL_30, ‘
‘\n BB_1, BB_3, BB_5, BB_10, ‘
‘\n RSI_1, RSI_5, RSI_10, RSI_30, ‘
‘\n STC_1, STC_5, STC_10, STC_30’)
text = (f»Укажите:»
«\n 1. Инструмент»
f»\n {symbols_list}»
«\n\n 2. Интервал (в минутах)»
f»\n {intervals_list}»
«\n\n 3. Период (в днях).»
«\n\n 4. График:»
f»\n {charts_list}»
«\n\n Пример: BTCUSD 240 7 OHLC»)
bot.send_message(message.chat.id, text)
Создаем функцию для проверки наличия запроса со стороны пользователя:
request = message.text.split()
if request:
return True
else:
return False
Создаем функцию для выдачи результата на основе информации, запрошенной пользователем:
Получение содержания запроса пользователя:
request = message.text.split()
Текущее время для отображения в ответе пользователю:
date_now = dt.datetime.fromtimestamp(time.time()).strftime(‘%d.%m.%Y’)
Текст сообщения об ошибке, в случае если пользователь неверно ввел данные:
text_error = ‘Проверьте пожалуйста формат введенных данных.’
Проверка запроса на наличие требуемых 4 параметров:
# try:
if (len(request) == 4):
Определение инструмента из введенного запроса:
symbol = str(request[0])
Определение интервала из введенного запроса:
if request[1] in [‘D’, ‘W’, ‘M’]:
interval = str(request[1])
elif request[1] in [
‘1’, ‘3’, ‘5’, ’15’, ’30’, ’60’,
‘120’, ‘240’, ‘360’, ‘720’
]:
interval = int(request[1])
Определение периода из введенного запроса:
period = int(request[2])
Определение типа графика из введенного запроса:
chart_name = str(request[3])
Проверка валидности введенных данных в запросе для последующей генерации соответствующего ответа:
if ((symbol in SYMBOLS) & (interval in INTERVALS) &
(type(period) == int) & (chart_name in CHARTS)):
Получение данных на основе полученного инструмента и интервала:
data = bybit.get_data(URL, symbol, interval)
Определение начальной и конечной даты на основе полученного периода и фильтрации полученных данных:
end_date = dt.date.today()
start_date = end_date + dt.timedelta(days=-period)
Фильтрация полученных данных на основе параметров, полученных в запросе пользователя:
df = etl(data, start_date, end_date)
Определение паттерна «тренд» для полученных данных:
trend = ohlc.trend_pattern(df)
reversal = ohlc.reversal_pattern(df)
oid_fill = ohlc.void_fill_pattern(df)
Генерация шаблона отчета для пользователя на полученный от него запрос согласно указанным параметрам:
text_analytics = (f»Инструмент: {symbol}»
f»\nДата: {date_now}»
«\nВремя анализа: 00:01»
f»\nАнализ на: {period} days»
f»\nНаправление тренда: {trend}»
f»\nПаттерн разворот: {reversal}»
f»\nПаттерн пустота: {oid_fill}»
f»\nТочки входа определены на: «
f»{INTERVAL_NAMES[interval]}»)
Реализация отображения индикаторов
График OHLC
Функция для визуализации данных в виде графика OHLC.
«»»
chart = go.Candlestick(x=df[‘time’],
open=df[‘open’],
high=df[‘high’],
low=df[‘low’],
close=df[‘close’])
layout = go.Layout(title=go.layout.Title(text=f'{symbol}’),
xaxis_title=’Date’,
yaxis_title=’Price’,
xaxis_rangeslider_visible=False)
fig = go.Figure(data=[chart], layout=layout)
fig.write_image(‘images/chart_’ + ‘OHLC_’ +
symbol + ‘_’ +
str(interval) + ‘_’ +
str(period) + ‘.png’)
График Moving Averages
Функция для визуализации данных на основе метода скользящих средних.
«»»
dt_all = pd.date_range(start=list(df[‘time’])[0], end=list(df[‘time’])[-1])
dt_obs = [d.strftime(«%Y-%m-%d») for d in pd.to_datetime(df[‘time’])]
dt_breaks = [d for d in dt_all.strftime(«%Y-%m-%d»).tolist()
if d not in dt_obs]
df[‘MA20’] = df[‘close’].rolling(window=20).mean()
df[‘MA5’] = df[‘close’].rolling(window=5).mean()
fig = go.Figure()
fig.add_trace(go.Candlestick(x=df[‘time’],
open=df[‘open’],
high=df[‘high’],
low=df[‘low’],
close=df[‘close’],
showlegend=False))
fig.add_trace(go.Scatter(x=df[‘time’],
y=df[‘MA5’],
opacity=0.7,
line=dict(color=’blue’, width=2),
name=’MA 5′))
fig.add_trace(go.Scatter(x=df[‘time’],
y=df[‘MA20’],
opacity=0.7,
line=dict(color=’orange’, width=2),
name=’MA 20′))
fig.update_xaxes(rangebreaks=[dict(values=dt_breaks)])
fig.update_layout(xaxis_rangeslider_visible=False)
fig.update_layout(xaxis_title=’Date’)
fig.update_layout(yaxis_title=’Price’)
fig.update_layout(title=symbol)
fig.write_image(‘images/chart_’ + ‘MA_’
+ symbol + ‘_’
+ str(interval) + ‘_’
+ str(period) + ‘.png’)
График Bollinger Bands
Функция для визуализации данных в виде Bollinger Bands.
df[‘sma’] = df[‘close’].rolling(k).mean()
df[‘std’] = df[‘close’].rolling(k).std(ddof=0)
fig = make_subplots(
rows=1,
cols=1,
shared_xaxes=True,
subplot_titles=(symbol),
vertical_spacing=0.1,
row_width=[0.7]
)
fig.add_trace(go.Candlestick(x=df[‘time’],
open=df[‘open’],
high=df[‘high’],
low=df[‘low’],
close=df[‘close’], showlegend=False,
name=’candlestick’),
row=1,
col=1)
fig.add_trace(go.Scatter(x=df[‘time’],
y=df[‘sma’],
line_color=’black’,
name=’sma’),
row=1,
col=1)
fig.add_trace(go.Scatter(x=df[‘time’],
y=df[‘sma’] + (df[‘std’] * 2),
line_color=’gray’,
line={‘dash’: ‘dash’},
name=’upper band’,
opacity=0.5),
row=1,
col=1)
fig.add_trace(go.Scatter(x=df[‘time’],
y=df[‘sma’] — (df[‘std’] * 2),
line_color=’gray’,
line={‘dash’: ‘dash’},
fill=’tonexty’,
name=’lower band’,
opacity=0.5),
row=1,
col=1)
fig.update(layout_xaxis_rangeslider_visible=False)
fig.write_image(‘images/chart_’ + ‘BB_’
+ symbol + ‘_’
+ str(interval) + ‘_’
+ str(period) + ‘.png’)
График Volume
Функция для визуализации данных в виде volume графика.
colors = [‘green’ if row[‘open’] — row[‘close’] >= 0
else ‘red’ for index, row in df.iterrows()]
chart = go.Bar(x=df[‘time’],
y=df[‘volume’],
marker_color=colors
)
layout = go.Layout(title=go.layout.Title(text=f'{symbol}’),
xaxis_title=’Date’,
yaxis_title=’Price’,
xaxis_rangeslider_visible=False)
fig = go.Figure(data=[chart], layout=layout)
fig.write_image(‘images/chart_’ +
‘VOL_’ + symbol +
‘_’ + str(interval) +
‘_’ + str(period) + ‘.png’)
График RSI
Функция для визуализации данных в виде RSI графика.
delta = df[‘close’].diff()[1:]
pricesUp = delta.copy()
pricesDown = delta.copy()
pricesUp[pricesUp < 0] = 0
pricesDown[pricesDown > 0] = 0
rollUp = pricesUp.rolling(k).mean()
rollDown = pricesDown.abs().rolling(k).mean()
rs = rollUp / rollDown
rsi = 100.0 — (100.0 / (1.0 + rs))
df[‘RSI’] = rsi.fillna(0)
layout = go.Layout(title=go.layout.Title(text=f'{symbol}’),
xaxis_title=’Date’,
yaxis_title=’%’,
xaxis_rangeslider_visible=False)
chart = go.Scatter(x=df[‘time’], y=df[‘RSI’])
fig = go.Figure(data=[chart], layout=layout)
fig.write_image(‘images/chart_’ +
‘RSI_’ + symbol +
‘_’ + str(interval) +
‘_’ + str(period) + ‘.png’)
График Stochastic
Функция для визуализации данных в виде stochastic графика.
df = df.copy()
low_min = df[‘low’].rolling(window=k).min()
high_max = df[‘high’].rolling(window=k).max()
df[‘stoch_k’] = 100 * (df[‘close’] — low_min)/(high_max — low_min)
df[‘stoch_d’] = df[‘stoch_k’].rolling(window=d).mean()
layout = go.Layout(title=go.layout.Title(text=f'{symbol}’),
xaxis_title=’Date’,
yaxis_title=’%’,
xaxis_rangeslider_visible=False)
fig = go.Figure(layout=layout)
fig.add_trace(
go.Scatter(
x=df[‘time’], y=df[‘stoch_k’], name=’K stochastic’)
)
fig.add_trace(
go.Scatter(
x=df[‘time’], y=df[‘stoch_d’], name=’D stochastic’)
)
fig.write_image(‘images/chart_’ +
‘STC_’ + symbol +
‘_’ + str(interval) +
‘_’ + str(period) + ‘.png’)
График Market Profile External Timeframe
Функция для визуализации данных в виде графиков market profile и external timeframe.
dt_all = pd.date_range(start=df[‘time’].iloc[0], end=df[‘time’].iloc[-1])
dt_obs = [d.strftime(«%Y-%m-%d») for d in pd.to_datetime(df[‘time’])]
dt_breaks = [d for d in dt_all.strftime(«%Y-%m-%d»).tolist() if d not in dt_obs]
df[‘MA20’] = df[‘close’].rolling(window=20).mean()
df[‘MA5’] = df[‘close’].rolling(window=5).mean()
Подготовка данных для market profile.
num_intervals = 50 — количество интервалов (ценовых диапазонов)
prices = np.linspace(df[‘low’].min(), df[‘high’].max(), num_intervals) — ценовые диапазоны
tpo_counts = [] — подсчет TPO (Time Price Opportunity) для цен (смотрим, какие проторговки входят в определенные интервалы)
for i in range(1, len(prices)):
count = np.sum(((df[‘high’] >= prices[i — 1]) & (df[‘low’] <= prices[i])) |
((df[‘low’] <= prices[i]) & (df[‘high’] >= prices[i — 1])))
tpo_counts.append(count)
max_tpo = max(tpo_counts)
max_width = 0.08 — Высота столбцов Market Profile.
mp_shapes = [] — словарь для форм прямоугольников Market Profile
for idx, tpo_count in enumerate(tpo_counts):
# print(idx, prices[idx], prices[idx + 1])
shape = {
‘type’: ‘rect’,
‘xref’: «paper»,
‘yref’: «y»,
‘x0’: 0,
‘x1’: max_width * (tpo_count / max_tpo), — подсчет происходит как высота столбца, значение от 0 до 1
‘y0’: prices[idx],
‘y1’: prices[idx + 1],
‘fillcolor’: «grey»,
‘opacity’: 0.3
}
mp_shapes.append(shape)
Подготовка данных для external timeframe.
Группируем данные по неделям для external timeframe.
grouped = df.groupby(pd.Grouper(key=’time’, freq=’W’))
grouped_list = list(grouped)
et_shapes = []
for i in range(len(grouped_list)):
name, group = grouped_list[i]
if not group.empty:
color_ = ‘LightSalmon’ if group[‘open’].iloc[0] > group[‘close’].iloc[-1] else ‘LightBlue’
x1 = group[‘time’].max() if i == (len(grouped_list) — 1) else grouped_list[i+1][1][‘time’].min()
shape = {
‘type’: ‘rect’,
‘xref’: ‘x’,
‘yref’: ‘y’,
‘x0’: group[‘time’].iloc[0],
‘x1’: x1,
‘y0’: group[‘low’].min(),
‘y1’: group[‘high’].max(),
‘fillcolor’: color_,
‘opacity’: 0.5,
‘line’: {
‘width’: 1,
}
}
et_shapes.append(shape)
fig = go.Figure()
fig.add_trace(go.Candlestick(x=df[‘time’],
open=df[‘open’],
high=df[‘high’],
low=df[‘low’],
close=df[‘close’],
showlegend=False))
fig.add_trace(go.Scatter(x=df[‘time’],
y=df[‘MA5’],
opacity=0.7,
line=dict(color=’blue’, width=2),
name=’MA 5′))
fig.add_trace(go.Scatter(x=df[‘time’],
y=df[‘MA20’],
opacity=0.7,
line=dict(color=’orange’, width=2),
name=’MA 20′))
fig.update_xaxes(rangebreaks=[dict(values=dt_breaks)])
fig.update_layout(xaxis_rangeslider_visible=False)
fig.update_layout(xaxis_title=’Date’)
fig.update_layout(yaxis_title=’Price’)
Добавляем фигуры в список shapes и отрисовываем прямоугольники:
shapes = et_shapes[1:] + mp_shapes
fig.update_layout(shapes=shapes)
fig.write_image(‘images/chart_’ +
‘MPET_’ + symbol +
‘_’ + str(interval) +
‘_’ + str(period) + ‘.png’)
Дашборд из трех графиков
Функция для визуализации трех типов графиков.
All charts (MA, VOL, STC).
«»»
df[‘MA20’] = df[‘close’].rolling(window=20).mean()
df[‘MA5’] = df[‘close’].rolling(window=5).mean()
fig=go.Figure()
# add subplot properties when initializing fig variable
fig = plotly.subplots.make_subplots(rows=3, cols=1, shared_xaxes=True,
vertical_spacing=0.01,
row_heights=[0.5, 0.2, 0.2]
)
fig.add_trace(go.Candlestick(x=df[‘time’],
open=df[‘open’],
high=df[‘high’],
low=df[‘low’],
close=df[‘close’],
showlegend=False))
fig.add_trace(go.Scatter(x=df[‘time’],
y=df[‘MA5’],
opacity=0.7,
line=dict(color=’blue’, width=2),
name=’MA 5′))
fig.add_trace(go.Scatter(x=df[‘time’],
y=df[‘MA20’],
opacity=0.7,
line=dict(color=’orange’, width=2),
name=’MA 20′))
# volume
colors = [‘green’ if row[‘open’] — row[‘close’] >= 0
else ‘red’ for index, row in df.iterrows()]
fig.add_trace(go.Bar(x=df[‘time’],
y=df[‘volume’],
marker_color=colors
), row=2, col=1)
# stochastic
low_min = df[‘low’].rolling(window=k).min()
high_max = df[‘high’].rolling(window=k).max()
df[‘stoch_k’] = 100 * (df[‘close’] — low_min)/(high_max — low_min)
df[‘stoch_d’] = df[‘stoch_k’].rolling(window=d).mean()
fig.add_trace(
go.Scatter(
x=df[‘time’], y=df[‘stoch_k’], opacity=0.7,
line=dict(color=’red’, width=2), name=’K stochastic’),
row=3, col=1
)
fig.add_trace(
go.Scatter(
x=df[‘time’], y=df[‘stoch_d’], opacity=0.7,
line=dict(color=’blue’, width=2), name=’D stochastic’),
row=3, col=1
)
fig.update_layout(xaxis_rangeslider_visible=False)
fig.update_layout(yaxis_title=’Price’)
fig.update_layout(title=symbol)
fig.write_image(‘images/chart_’ +
‘ALL_’ + symbol +
‘_’ + str(interval) +
‘_’ + str(period) + ‘.png’)
График ZigZag
Функция для визуализации данных паттерна ZigZag.
«»»
l = len(df[‘high’].tolist()) — 1
df = df.sort_values(‘time’, ascending=False)
zigzags = indicators.zigzag(df[‘high’].tolist(), df[‘low’].tolist()) — получаем кортежи точек
updated_zigzags = [((-peak[0] + l, peak[1]), (-valley[0] + l, valley[1])) for peak, valley in zigzags] — получаем нужные индексы для кортежей
zigzags = updated_zigzags[::-1] — разворачиваем полученные данные для отрисовки в правильном порядке
fig = go.Figure(data=[go.Candlestick(x=df[‘time’],
open=df[‘open’],
high=df[‘high’],
low=df[‘low’],
close=df[‘close’])])
Создаем зигзаги и соединяем пробелы между найденными минимумами и максимумами:
for i in range(len(zigzags)):
peak = zigzags[i][0]
valley = zigzags[i][1]
peak_next = zigzags[i+1][0] if i + 1 < len(zigzags) else zigzags[i][1]
fig.add_trace(
go.Scatter(
x=[df[‘time’].iloc[peak[0]], df[‘time’].iloc[valley[0]]],
y=[peak[1], valley[1]],
mode=»lines+markers»,
line=dict(color=»blue»),
marker=dict(size=6, color=»blue»)
)
)
fig.add_trace(
go.Scatter(
x=[df[‘time’].iloc[valley[0]], df[‘time’].iloc[peak_next[0]]],
y=[valley[1], peak_next[1]],
mode=»lines+markers»,
line=dict(color=»blue»),
marker=dict(size=6, color=»blue»)
Получаем фракталы:
upper_fractals, lower_fractals = indicators.find_fractals(df)
Фракталы вверх:
for index, row in df[upper_fractals].iterrows():
fig.add_annotation(
x=row[‘time’],
y=row[‘high’],
text=»\u25B2″, # Ззнак стрелки вверх
font=dict(size=16, color=»#03C04A»),
ax=0,
ay=-5,
xanchor=»center»,
yanchor=»bottom»
Фракталы вниз:
for index, row in df[lower_fractals].iterrows():
fig.add_annotation(
x=row[‘time’],
y=row[‘low’],
text=»\u25BC», # знак стрелки вниз
font=dict(size=16, color=»#D91E18″),
ax=0,
ay=5,
xanchor=»center»,
yanchor=»top»
)
fig.update_layout(xaxis_rangeslider_visible=False)
fig.write_image(‘images/chart_’ +
‘ZIGZAG_’ + symbol +
‘_’ + str(interval) +
‘_’ + str(period) + ‘.png’)
Дополнительные функции
Отправка сообщения об ошибке в случае некорректного ввода данных:
bot.send_message(message.chat.id, text_error)
Запуск цикла для отслеживания ошибок при работе бота:
i = 0
while True:
i += 1
try:
Запуск бота:
bot.polling()
Определение текущего времени, для отслеживания ошибок:
time_now = dt.datetime.fromtimestamp(
time.time()).strftime(‘%d-%m-%Y %H:%M:%S’)
Вывод номера очередности, времени и текста ошибки:
print(i, time_now, err)