Привет, Habr! В этой статье я хочу поделиться своим проектом — Telegram-ботом, который автоматизирует торговлю на бирже Bybit на основе сигналов из специализированного канала. Бот парсит сообщения из Telegram-канала @TokenSplashBybit, извлекает информацию о предстоящих "token splash" (это события, когда новые токены добавляются на биржу с возможностью получения airdrop), и открывает длинные позиции (лонги) в момент результата. Почему лонги? Потому что token splash на Bybit часто сопровождаются airdrop-вознаграждениями для держателей позиций, многие трейдеры начинают шортить подобные позиции - тем более учитывая, что часто на токены существуют много разных мероприятий, например, binance alpha и прочие. Толпа почти никогда не зарабатывает - так подобных трейдеров почти всегда отвозят наверх, ликвидируя и собирая стопы, что делает стратегию прибыльной в долгосрочной перспективе. Я не даю финансовых советов — это просто технический проект для энтузиастов автоматизации и криптотрейдинга.
Я собрал небольшую статистику вручную на основе исторических данных, чтобы показать потенциальную работоспособность подхода. Конечно, прошлые результаты не гарантируют будущих, и торговля всегда связана с рисками. Но всё же работаю с этим кодом уже не один месяц, и результат действительно соответствует ожиданиям. Давайте разберёмся по порядку: от идеи до полного кода с объяснениями.
Что такое Token Splash на Bybit и почему это выгодно?
Token Splash — это событие на Bybit, когда новый токен добавляется в листинг, и биржа раздаёт airdrop (бесплатные токены) участникам. Часто это связано с фьючерсами на USDT, где открытие позиции в лонг позволяет получить долю от airdrop. Канал @TokenSplashBybit публикует анонсы с токенами и датой "result" (моментом, когда токен становится доступным для торговли).
Логика стратегии проста:
Парсим канал на новые анонсы.
Планируем открытие позиции на дату result (используя scheduler).
Открываем лонг на 70% от расчётного объёма с рыночным ордером.
Устанавливаем тейк-профиты (TP1 на +3%, TP2 на +6%) и стоп-лосс (SL на -2%). Можно изменять тейки и стопы, но пока что меня устраивают результаты с такими переменными - тем более что они обусловлены анализом результатов всех этих мероприятий.
Это позволяет захватить рост цены после выхода из позиций шортистов, что происходит после получения аирдропа.
Почему автоматизация? Ручная торговля требует постоянного мониторинга, а бот делает всё сам: от парсинга до размещения ордеров через API Bybit.
Статистика: Подтверждение работоспособности
Чтобы показать, что подход имеет смысл, я вручную собрал данные по нескольким token splash за последний период. Вот часть таблицы с примерами:
Выводы:
В 70% случаев цена растёт на 3–10% в первые минуты.
Редко (15%) — безоткатное падение. Это ловит стоп-лосс.
Значит, стратегия простая: Входим в лонг → ставим TP на +3% и +6% → ставим SL на -2% . Это оптимальные параметры, при которых мы получим наилучший винрейт.
Логика бота: Шаг за шагом
Бот состоит из нескольких частей:
Telegram-бот (Telebot): Интерфейс для пользователей. Позволяет включать/выключать бота, настраивать API-ключи Bybit, плечо (leverage) и маржу (margin). Данные хранятся в JSON-файлах.
Парсинг канала (Telethon): Асинхронный клиент для Telegram. Парсит сообщения по regex, извлекает токен и дату result. Если дата в будущем — планирует задачу.
Планировщик (APScheduler): Запускает функцию notify_all_enabled_users в момент result, которая вызывает long_token для каждого активного пользователя.
Торговля (Pybit): В long_token:
Получает цену и параметры инструмента.
Рассчитывает объём: qty = (leverage * margin) / price, корректирует по шагу (qtyStep).
Проверяет баланс USDT.
Размещает ордера: Market Buy на 70%, Limit Sell TP1/TP2, StopMarket SL.
Обработка ошибок: Логирование, уведомления в Telegram при неудачах.
Бот работает в многопоточном режиме: Telebot и Telethon в отдельных потоках, scheduler в фоне.
Объяснение кода
Давайте разберём ключевые части кода.
Импорты и настройки
import re
from datetime import datetime
import pytz
from telethon import TelegramClient, events
from apscheduler.schedulers.background import BackgroundScheduler
from apscheduler.triggers.date import DateTrigger
from pybit.unified_trading import HTTP
import json
import os
import telebot
from decimal import Decimal
import asyncio
import threading
import time
import logging
# Настройка логирования
logging.basicConfig(
level=logging.DEBUG,
format='%(asctime)s - %(levelname)s - %(message)s',
)
TOKEN = ""
bot = telebot.TeleBot(TOKEN)
API_ID =
API_HASH = ''
SESSION_NAME = "my_session"
CHANNEL = '@TokenSplashBybit'
# Регулярка для парсинга токена и Result даты
POST_REGEX = r'^(?P<token>\w+)\n.*?Result (?P<result_date>\d{2}\.\d{2}\.\d{4} \d{2}:\d{2}) UTC'
TRADE_AMOUNT = 1000 # USDT (unused; margin from settings)
TP1_PCT = 0.03
TP2_PCT = 0.06
STOP_LOSS_PCT = 0.02
user_states = {}
USER_DATA_PATH = "user_data_tg"
os.makedirs(USER_DATA_PATH, exist_ok=True)telebot — создаёт бота в Telegram (кнопки, сообщения).
telethon — читает канал (как "глаза" бота).
pybit — торгует на Bybit (открывает позиции).
apscheduler — запускает торговлю в нужное время.
Регулярное выражение (regex) — это "фильтр", который находит в тексте:
TOKEN
...
Result 15.10.2025 14:30 UTC→ и вытаскивает TOKEN и 14:30.
Хранение данных пользователей
# GET USER + SAVE USER INFO
def get_user_file(user_id):
return os.path.join(USER_DATA_PATH, f"{user_id}.json")
def load_user_data(user_id):
try:
with open(get_user_file(user_id), 'r') as f:
return json.load(f)
except:
return {}
def save_user_data(user_id, data):
try:
with open(get_user_file(user_id), 'w') as f:
json.dump(data, f)
except Exception as e:
logging.error(f"Failed to save user data for {user_id}: {e}")
def get_all_user_ids():
user_ids = []
for filename in os.listdir(USER_DATA_PATH):
if filename.endswith(".json"):
try:
user_id = int(filename.split(".")[0])
user_ids.append(user_id)
except:
continue
return user_ids
def notify_all_enabled_users(token):
user_ids = get_all_user_ids()
for uid in user_ids:
data = load_user_data(uid)
if data.get("bot_enabled"):
long_token(uid, token_symbol=token)Данные (API key, secret, leverage, margin, bot_enabled) хранятся в JSON по user_id.
notify_all_enabled_users вызывает торговлю для активных пользователей.
Каждый пользователь — это папка с файлом 123456.json вида:
{
"api_key": "abc123",
"secret": "xyz789",
"leverage": "10",
"margin": "20",
"bot_enabled": true
}leverage — плечо (10x = 10).
margin — сколько USDT тратить на сделку.
bot_enabled — включён ли бот.
Бот может работать для многих людей одновременно.
Меню и обработчики Telebot
def get_menu_markup(user_id):
data = load_user_data(user_id)
bot_enabled = data.get("bot_enabled", False)
state_button = " Бот включен" if bot_enabled else " Бот выключен"
markup = telebot.types.ReplyKeyboardMarkup(resize_keyboard=True)
markup.row(state_button, "ℹ️ Информация о боте")
markup.row("⚙️ Настройки")
return markup
settings_markup = telebot.types.ReplyKeyboardMarkup(resize_keyboard=True)
settings_markup.row("API Key", "secret")
settings_markup.row("Leverage", "Margin")
settings_markup.row("Назад")
# Вспомогательные функции для безопасного преобразования чисел
def safe_float(value, default=0.0):
"""Безопасное преобразование в float"""
if value is None:
return default
try:
if isinstance(value, str) and value.strip() == '':
return default
return float(value)
except (ValueError, TypeError):
return default
def safe_int(value, default=0):
"""Безопасное преобразование в int"""
if value is None:
return default
try:
if isinstance(value, str) and value.strip() == '':
return default
return int(float(value))
except (ValueError, TypeError):
return default
# MATH
def count_decimal_places(number: float) -> int:
s = f"{number:.16f}".rstrip('0')
if '.' in s:
return len(s.split('.')[1])
else:
return 0
# КОМАНДЫ ДЛЯ БОТА
@bot.message_handler(func=lambda msg: msg.text in [" Бот включен", " Бот выключен"])
def toggle_bot_state(message):
user_id = message.chat.id
data = load_user_data(user_id)
current_state = data.get("bot_enabled", False)
data["bot_enabled"] = not current_state
save_user_data(user_id, data)
status = "включен" if data["bot_enabled"] else "выключен"
bot.send_message(user_id, f"Бот теперь {status}.", reply_markup=get_menu_markup(user_id))
@bot.message_handler(func=lambda msg: msg.text == "ℹ️ Информация о боте")
def bot_info(message):
user_id = message.chat.id
info_text = (
"Это торговый бот для биржи Bybit, торгующий в лонг на токен сплэшах.\n"
"Вы можете включать и выключать бота, а также настраивать параметры в разделе Настройки. Для грамотной работы бота необходимо обязательно указать всё.\n"
"Если нужна помощь — обращайтесь @perpetual_god."
)
bot.send_message(user_id, info_text, reply_markup=get_menu_markup(user_id))
@bot.message_handler(commands=['start'])
def send_welcome(message):
user_id = message.chat.id
bot.send_message(user_id, "Добро пожаловать! Выберите действие:", reply_markup=get_menu_markup(user_id))
@bot.message_handler(func=lambda msg: msg.text in ["⚙️ Настройки"])
def handle_settings_menu(message):
user_id = message.chat.id
bot.send_message(user_id, "Выберите параметр:", reply_markup=settings_markup)
@bot.message_handler(func=lambda msg: msg.text in ["API Key", "secret", "Leverage", "Margin"])
def ask_for_value(message):
user_id = message.chat.id
key = message.text.lower().replace(" ", "_")
data = load_user_data(user_id)
current_value = data.get(key, "(не задано)")
user_states[user_id] = key
bot.send_message(user_id, f"Текущее значение {message.text}: {current_value}\n\nВведите новое значение:")
@bot.message_handler(func=lambda msg: msg.text == "Назад")
def back_to_menu(message):
user_id = message.chat.id
bot.send_message(user_id, "Вы вернулись в меню", reply_markup=get_menu_markup(user_id))
# ОБРАБОТЧИК ДЛЯ ВВОДА ЗНАЧЕНИЙ
@bot.message_handler(func=lambda msg: msg.chat.id in user_states)
def catch_input(message):
user_id = message.chat.id
if user_id in user_states:
key = user_states.pop(user_id)
data = load_user_data(user_id)
data[key] = message.text.strip()
save_user_data(user_id, data)
bot.send_message(user_id, f"{key} сохранено. Выберите следующий параметр:", reply_markup=settings_markup)Вы видите кнопки:
🟢 Бот включён | ℹ️ Инфо | ⚙️ НастройкиНажали "Настройки" → выбираете "Leverage" → вводите 10.
Бот сохраняет и говорит: "Leverage сохранено".
Это как личный кабинет, но в чате.
Парсинг и планирование
async def fetch_new_tokens(client):
new_tokens = []
try:
logging.debug("[Telethon] Получаем последние 50 сообщений из канала...")
messages = await client.get_messages(CHANNEL, limit=50)
logging.info(f"[Telethon] Всего сообщений получено: {len(messages)}")
for msg in messages:
if msg.text is None:
continue
match = re.search(POST_REGEX, msg.text, re.DOTALL)
if not match:
continue
symbol = match.group('token').strip()
result_date_str = match.group('result_date').strip()
try:
result_dt = datetime.strptime(result_date_str, "%d.%m.%Y %H:%M")
result_dt = pytz.utc.localize(result_dt)
except ValueError:
logging.error(f"[Telethon] Невозможно разобрать дату: {result_date_str}")
continue
now = datetime.utcnow().replace(tzinfo=pytz.utc)
if result_dt > now:
logging.info(f"[Telethon] Найден токен для планирования: {symbol} (Result: {result_dt})")
new_tokens.append({
"symbol": symbol,
"result_time": result_dt
})
except Exception as e:
logging.error(f"[Telethon] Ошибка при получении токенов: {e}")
raise
return new_tokens
scheduler = BackgroundScheduler(timezone=pytz.utc)
def start_scheduler():
logging.info("[Scheduler] Запуск планировщика...")
if not scheduler.running:
try:
scheduler.start()
except Exception as e:
logging.error(f"[Scheduler] Failed to start scheduler: {e}")
def start_telebot():
logging.info("[Telebot] Запуск Telegram-бота...")
while True:
try:
bot.polling(none_stop=True, timeout=30, long_polling_timeout=50)
except Exception as e:
logging.error(f"[Telebot] Polling error: {e}. Restarting in 10 sec...")
time.sleep(10)
def format_symbol(symbol):
return symbol.upper() + "USDT"
async def start_telethon():
logging.info("[Telethon] Запуск клиента...")
client = TelegramClient(SESSION_NAME, API_ID, API_HASH)
try:
await client.start()
logging.info("[Telethon] Client successfully started.")
except Exception as e:
logging.error(f"[Telethon] Failed to start client: {e}")
raise
# Setup event handler
@client.on(events.NewMessage(chats=CHANNEL))
async def new_message_handler(event):
text = event.message.message
if text is None:
return
match = re.search(POST_REGEX, text, re.DOTALL)
if not match:
return
raw_symbol = match.group('token').strip()
symbol = format_symbol(raw_symbol)
result_date_str = match.group('result_date').strip()
try:
result_dt = datetime.strptime(result_date_str, "%d.%m.%Y %H:%M")
result_dt = pytz.utc.localize(result_dt)
except ValueError:
logging.error(f"[Telethon] Невозможно разобрать дату из нового сообщения: {result_date_str}")
return
now = datetime.utcnow().replace(tzinfo=pytz.utc)
if result_dt > now:
logging.info(f"[Telethon] Новый токен из чата для планирования: {symbol} (Result: {result_dt})")
schedule_long({"symbol": symbol, "result_time": result_dt})
# Fetch initial tokens after start
try:
tokens = await fetch_new_tokens(client)
for token in tokens:
schedule_long(token)
logging.info("[Telethon] Initial tokens fetched and scheduled.")
except Exception as e:
logging.error(f"[Telethon] Failed to fetch initial tokens: {e}")
# Run the client with reconnect loop
while True:
try:
await client.run_until_disconnected()
except Exception as e:
logging.error(f"[Telethon] Disconnected: {e}. Reconnecting in 10 sec...")
await asyncio.sleep(10)
def schedule_long(token_info):
symbol = token_info["symbol"]
run_time = token_info["result_time"]
now = datetime.utcnow().replace(tzinfo=pytz.utc)
if run_time <= now:
logging.info(f"[Scheduler] {symbol} уже в прошлом, запускаем немедленно")
notify_all_enabled_users(symbol)
return
job_id = f"long_{symbol}_{run_time.strftime('%Y%m%d%H%M')}"
if scheduler.get_job(job_id):
logging.info(f"[Scheduler] Задача {job_id} уже запланирована, пропускаем.")
return
logging.info(f"[Scheduler] Планируем лонг для {symbol} на {run_time} (UTC)")
scheduler.add_job(notify_all_enabled_users, trigger=DateTrigger(run_date=run_time), args=[symbol], id=job_id, misfire_grace_time=60)Бот:
Подключается к Telegram через ваш аккаунт (нужен API ID и Hash).
Читает последние 50 сообщений из @TokenSplashBybit.
Находит новые анонсы.
Если время Result в будущем — ставит задачу в календарь.
Новое сообщение? → Бот сразу реагирует и планирует сделку.
Планировщик (APScheduler)
Это "будильник". Если Result в 14:30 — бот ставит задачу:
"В 14:30:00 открыть лонг по LAUSDT"
В назначенное время задача сработает. Удобно и не заставляет нас постоянно проверять время в скрипте.
Торговля
def get_valid_qty(session, symbol, raw_qty):
try:
info = session.get_instruments_info(category="linear", symbol=symbol)
if 'result' not in info or 'list' not in info['result'] or not info['result']['list']:
logging.error(f"❌ Пара {symbol} не найдена в get_instruments_info.")
return None
lot_filter = info['result']['list'][0].get('lotSizeFilter', {})
step = safe_float(lot_filter.get('qtyStep', 0))
min_qty = safe_float(lot_filter.get('minOrderQty', 0))
if step == 0:
logging.error(f"❌ qtyStep равен 0 для {symbol}")
return None
qty = max(raw_qty, min_qty)
valid_qty = (qty // step) * step
logging.info(f"qty: {valid_qty}, step: {step}, min_qty: {min_qty}")
return valid_qty
except Exception as e:
logging.error(f"⚠️ Ошибка получения допустимого объема для {symbol}: {e}")
return None
def step_qty(qty, qty_step):
if qty_step == 0:
return qty
return (qty // qty_step) * qty_step
def long_token(user_id, token_symbol):
logging.info(f"▶️ Лонг {token_symbol} for user {user_id}")
token_symbol = token_symbol + 'USDT'
try:
data = load_user_data(user_id)
if not all(key in data for key in ['api_key', 'secret', 'leverage', 'margin']):
logging.warning(f"Пользователь {user_id} не настроил все параметры. Пропуск.")
bot.send_message(user_id, "❌ Не все параметры настроены. Настройте в ⚙️ Настройки.")
return
session = HTTP(api_key=data['api_key'], api_secret=data['secret'], recv_window=60000)
res = session.get_tickers(category="linear", symbol=token_symbol)
if not res['result']['list']:
logging.error(f"❌ {token_symbol} пока не торгуется.")
bot.send_message(user_id, f"❌ {token_symbol} пока не торгуется.")
return
info = session.get_instruments_info(category="linear", symbol=token_symbol)
if not info['result']['list']:
logging.error(f"❌ {token_symbol} не имеет linear futures.")
bot.send_message(user_id, f"Фьючерс на токен {token_symbol} не существует! Ошибка сделки.")
return
# Безопасное получение параметров с проверкой на пустые строки
price_filter = info['result']['list'][0]['priceFilter']
tick_size = safe_float(price_filter['tickSize'], 0.01) # дефолтное значение 0.01 если пусто
price_precision = abs(Decimal(str(tick_size)).as_tuple().exponent)
lot_filter = info['result']['list'][0]['lotSizeFilter']
qty_step = safe_float(lot_filter.get('qtyStep', ''), 0.001) # дефолтное значение 0.001 если пусто
min_order_qty = safe_float(lot_filter.get('minOrderQty', ''), 0.001)
logging.info(f"Параметры инструмента: tick_size={tick_size}, qty_step={qty_step}, min_order_qty={min_order_qty}")
price = safe_float(res['result']['list'][0]['lastPrice'], 0)
if price == 0:
logging.error(f"❌ Цена для {token_symbol} равна 0")
bot.send_message(user_id, f"❌ Не удалось получить цену для {token_symbol}")
return
leverage = safe_int(data.get("leverage", 5))
margin = safe_float(data.get("margin", 10))
raw_qty = leverage * margin / price
full_qty = get_valid_qty(session, token_symbol, raw_qty)
if full_qty is None or full_qty == 0:
bot.send_message(user_id, f"❌ Не удалось определить объем для {token_symbol}")
return
# Balance check
balance_res = session.get_wallet_balance(accountType="UNIFIED")
if balance_res['retCode'] != 0:
logging.error(f"Ошибка баланса: {balance_res['retMsg']}")
bot.send_message(user_id, "❌ Ошибка проверки баланса на Bybit.")
return
# Безопасное получение баланса
usdt_balance = 0
try:
coins = balance_res['result']['list'][0]['coin']
for coin in coins:
if coin['coin'] == 'USDT':
usdt_balance = safe_float(coin.get('availableToWithdraw', 0))
break
except (KeyError, IndexError, TypeError) as e:
logging.error(f"Ошибка парсинга баланса: {e}")
bot.send_message(user_id, "❌ Ошибка получения баланса USDT.")
return
required_margin = full_qty * price / leverage
# Расчет объемов с безопасным округлением
buy_qty = step_qty(full_qty * 0.7, qty_step)
tp1_qty = step_qty(full_qty * 0.4, qty_step)
tp2_qty = step_qty(full_qty * 0.3, qty_step)
sl_qty = step_qty(full_qty * 0.3, qty_step)
# Проверка что объемы не нулевые
if buy_qty == 0 or tp1_qty == 0 or tp2_qty == 0 or sl_qty == 0:
logging.error(f"❌ Один из объемов равен 0: buy={buy_qty}, tp1={tp1_qty}, tp2={tp2_qty}, sl={sl_qty}")
bot.send_message(user_id, f"❌ Рассчитанные объемы слишком малы для {token_symbol}")
return
tp1 = round(price * (1 + TP1_PCT) / tick_size) * tick_size
tp2 = round(price * (1 + TP2_PCT) / tick_size) * tick_size
sl = round(price * (1 - STOP_LOSS_PCT) / tick_size) * tick_size
logging.info(f"Размещение ордеров: buy={buy_qty}, tp1={tp1_qty}@{tp1}, tp2={tp2_qty}@{tp2}, sl={sl_qty}@{sl}")
# Buy 70% with position SL
session.place_order(
category="linear",
symbol=token_symbol,
side="Buy",
order_type="Market",
qty=round(buy_qty, 0),
reduce_only=False,
time_in_force="GoodTillCancel",
stopLoss=str(sl) # явное преобразование в строку
)
# TP1 limit sell 40%
session.place_order(
category="linear",
symbol=token_symbol,
side="Sell",
order_type="Limit",
qty=round(buy_qty, 0),
price=str(tp1), # явное преобразование в строку
reduce_only=True,
time_in_force="GoodTillCancel"
)
# TP2 limit sell 30%
session.place_order(
category="linear",
symbol=token_symbol,
side="Sell",
order_type="Limit",
qty=float(tp2_qty),
price=str(tp2), # явное преобразование в строку
reduce_only=True,
time_in_force="GoodTillCancel"
)
bot.send_message(user_id, f"✅ Лонг по {token_symbol} выполнен по цене {price:.2f}")
except Exception as err:
logging.error(f'Error while placing order for {token_symbol}: {err}')
bot.send_message(user_id, f"❌ Ошибка выполнения ордера для {token_symbol}: {str(err)}")Когда время пришло, бот:
Заходит в ваш аккаунт Bybit (через API-ключ).
Смотрит текущую цену.
Считает объём:
raw_qty = leverage * margin / priceНапример: 10 × 20 ÷ 0.1 = 2000 токенов.
Корректирует под правила Bybit (шаг, минимум).
Открывает 4 ордера:
70% — рыночный Buy.
40% — Limit Sell на +3% (TP1).
30% — Limit Sell на +6% (TP2).
30% — Stop Market на -2% (SL).
Пишет вам в бота:
Лонг по LAUSDT выполнен по цене 0.10
Запуск
def main():
logging.info("[Main] Starting bot...")
try:
# Запускаем планировщик
threading.Thread(target=start_scheduler, daemon=True).start()
# Запускаем telebot
threading.Thread(target=start_telebot, daemon=True).start()
# Telethon в отдельном потоке
def run_telethon():
logging.info("[Telethon Thread] Starting...")
try:
loop = asyncio.new_event_loop()
asyncio.set_event_loop(loop)
loop.run_until_complete(start_telethon())
except Exception as e:
logging.error(f"[Telethon Thread] Failed: {e}")
run_telethon()
threading.Thread(target=run_telethon, daemon=True).start()
# Keep main thread alive
while True:
time.sleep(1)
except Exception as e:
logging.error(f"[Main] Fatal error: {e}")
raise
if __name__ == "__main__":
main()Бот запускает 3 потока:
Telegram-бот (кнопки).
Чтение канала (Telethon).
Планировщик (APScheduler).
Всё работает параллельно и не падает при ошибках — есть перезапуск.
Для запуска:
Установите Python и библиотеки:
pip install telebot telethon pybit apschedulerВставьте свои:
BOT TOKEN (от @BotFather)
API_ID, API_HASH (my.telegram.org)
API-ключ Bybit
Запустите — и бот начнёт работать.
Заключение
Этот бот — пример, как автоматизировать криптотрейдинг с использованием Telegram-API и биржевых инструментов. Он упрощает участие в token splash, ловя airdrop и потенциальный рост. Но помните: торговля рискованна, тестируйте на демо, и не используйте реальные деньги без понимания. Если улучшите код — делитесь в комментариях! Код открыт для экспериментов, но заполните свои API-ключи через телеграмм бота.
Если у вас есть вопросы — пишите. Удачных трейдов!
