Как создать торгового бота для ByBit на Python: полное руководство для начинающих

_00_config/        # Все настройки и конфигурационные файлы
_01_sources/       # Источники данных (API, WebSocket и пр.)
_02_etl/           # Модули для извлечения, трансформации и загрузки данных
_03_datasets/      # Управление и хранение подготовленных данных
_04_indicators/    # Реализация технических индикаторов
_05_patterns/      # Алгоритмы распознавания торговых паттернов
_06_visualization/ # Визуализация данных и сигналов
_07_management/    # Управление рисками и капиталом
_08_baktesting/    # Бэктестинг торговых стратегий
_09_dashboard/     # Интерфейс для мониторинга и контроля
_10_trading/       # Основная логика торговли и взаимодействие с биржей
_11_notebooks/     # Jupyter-ноутбуки для аналитики и экспериментов

# _00_config/config.py

BYBIT_API_KEY = "your_api_key"
BYBIT_API_SECRET = "your_api_secret"

TRADING_PAIR = "BTCUSDT"
TIMEFRAME = "15m"

# Параметры стратегии
EMA_SHORT = 9
EMA_LONG = 21

# Риск-менеджмент
MAX_TRADE_SIZE = 0.05  # В BTC
STOP_LOSS_PERCENT = 1.5
TAKE_PROFIT_PERCENT = 3.0

# Режим работы
MODE = "paper"  # Возможные значения: "paper" или "live"

# Пути
DATA_PATH = "./_03_datasets/"
LOG_PATH = "./logs/"

# .env файл
BYBIT_API_KEY=your_api_key
BYBIT_API_SECRET=your_api_secret

# config.py с использованием dotenv
from dotenv import load_dotenv
import os

load_dotenv()

BYBIT_API_KEY = os.getenv("BYBIT_API_KEY")
BYBIT_API_SECRET = os.getenv("BYBIT_API_SECRET")

# _01_sources/bybit_client.py

from pybit.unified_trading import HTTP
from _00_config import config

session = HTTP(
    api_key=config.BYBIT_API_KEY,
    api_secret=config.BYBIT_API_SECRET,
    testnet=(config.MODE == "paper")
)

def get_ohlcv(symbol, interval, limit=200):
    response = session.get_kline(
        category="linear",
        symbol=symbol,
        interval=interval,
        limit=limit
    )
    return response['result']['list']

# Пример вызова
if __name__ == "__main__":
    data = get_ohlcv(config.TRADING_PAIR, config.TIMEFRAME)
    print(data[:2])  # Выводим первые две свечи

# _02_etl/etl_processor.py

import pandas as pd
from _01_sources.bybit_client import get_ohlcv  # Импорт функции для получения данных с ByBit
from _00_config import config  # Импорт конфигурационных параметров

def run_etl():
    # Шаг 1: Извлечение данных с биржи (получаем "сырые" данные в виде списка)
    raw_data = get_ohlcv(config.TRADING_PAIR, config.TIMEFRAME)
    
    # Шаг 2: Преобразование данных в DataFrame для удобной работы
    df = pd.DataFrame(raw_data, columns=[
        'timestamp',  # Время открытия свечи (в миллисекундах)
        'open',       # Цена открытия
        'high',       # Максимальная цена
        'low',        # Минимальная цена
        'close',      # Цена закрытия
        'volume',     # Объём торгов
        'turnover'    # Дополнительные данные (оборот)
    ])
    
    # Шаг 3: Преобразуем timestamp в читаемый формат даты и времени
    df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
    
    # Шаг 4: Приводим числовые данные к типу float для корректных вычислений
    df[['open', 'high', 'low', 'close', 'volume']] = df[['open', 'high', 'low', 'close', 'volume']].astype(float)
    
    # Шаг 5: Сортируем данные по времени (на случай, если биржа вернула их неупорядоченными)
    df = df.sort_values('timestamp')
    
    # Шаг 6: Сохраняем итоговый датасет в формате CSV для дальнейшего использования
    df.to_csv(f"{config.DATA_PATH}{config.TRADING_PAIR}_{config.TIMEFRAME}.csv", index=False)
    
    # Сообщаем пользователю о завершении процесса
    print(f"Данные сохранены: {config.DATA_PATH}{config.TRADING_PAIR}_{config.TIMEFRAME}.csv")

# Запускаем процесс ETL, если файл выполняется напрямую
if __name__ == "__main__":
    run_etl()

# _03_datasets/dataset_manager.py

import pandas as pd
import os
from _00_config import config
from _02_etl import etl_processor  # Импортируем ETL для обновления данных

class DatasetManager:
    def __init__(self, symbol, timeframe):
        """
        Инициализация менеджера датасетов с указанием торговой пары и таймфрейма.
        """
        self.symbol = symbol
        self.timeframe = timeframe
        self.file_path = f"{config.DATA_PATH}{symbol}_{timeframe}.csv"
        self.data = None  # Здесь будет храниться DataFrame с данными

    def load_data(self):
        """
        Загружаем датасет из CSV-файла.
        Если файл отсутствует — запускаем ETL для его создания.
        """
        if not os.path.exists(self.file_path):
            print("Датасет не найден. Запуск ETL для получения данных...")
            etl_processor.run_etl()

        # Чтение данных из CSV в DataFrame
        self.data = pd.read_csv(self.file_path, parse_dates=['timestamp'])
        print(f"Данные успешно загружены: {self.file_path}")

    def get_latest_data(self, n=50):
        """
        Возвращает последние n свечей из датасета.
        """
        if self.data is None:
            self.load_data()
        return self.data.tail(n)

    def refresh_data(self):
        """
        Обновление датасета (перезапуск ETL).
        """
        print("Обновляем данные...")
        etl_processor.run_etl()
        self.load_data()

# Пример использования
if __name__ == "__main__":
    dm = DatasetManager(config.TRADING_PAIR, config.TIMEFRAME)
    dm.load_data()  # Загружаем данные
    print(dm.get_latest_data(5))  # Показываем последние 5 свечей

# _04_indicators/ema_calculator.py

import pandas as pd

def calculate_ema(df, period, column='close'):
    """
    Функция для расчета экспоненциальной скользящей средней (EMA).

    :param df: DataFrame с рыночными данными (должен содержать колонку 'close')
    :param period: Период для расчета EMA (например, 9 или 21)
    :param column: Колонка, по которой рассчитывается EMA (по умолчанию 'close')
    :return: DataFrame с добавленной колонкой EMA_{period}
    """
    ema_column_name = f'EMA_{period}'  # Имя новой колонки для хранения значений EMA

    # Используем встроенную функцию pandas для расчета EMA
    df[ema_column_name] = df[column].ewm(span=period, adjust=False).mean()

    return df

# Пример использования
if __name__ == "__main__":
    # Загружаем тестовый датасет (предполагаем, что он уже есть после работы DatasetManager)
    df = pd.read_csv('./_03_datasets/BTCUSDT_15m.csv', parse_dates=['timestamp'])

    # Рассчитываем две EMA с разными периодами (например, для стратегии пересечения EMA)
    df = calculate_ema(df, period=9)
    df = calculate_ema(df, period=21)

    # Выводим последние 5 строк для проверки результата
    print(df[['timestamp', 'close', 'EMA_9', 'EMA_21']].tail(5))

# _05_patterns/ema_crossover_strategy.py

import pandas as pd

def detect_ema_crossover_signals(df, short_period=9, long_period=21):
    """
    Функция для обнаружения сигналов на основе пересечения EMA.

    :param df: DataFrame с рассчитанными EMA (должны быть колонки EMA_short_period и EMA_long_period)
    :param short_period: Период короткой EMA
    :param long_period: Период длинной EMA
    :return: DataFrame с добавленной колонкой 'signal'
    """
    # Названия колонок с EMA
    short_ema_col = f'EMA_{short_period}'
    long_ema_col = f'EMA_{long_period}'

    # Инициализация новой колонки 'signal' значениями 0 (нет сигнала)
    df['signal'] = 0

    # Логика пересечения EMA:
    # Если короткая EMA была ниже длинной и стала выше — это сигнал на покупку (1)
    # Если короткая EMA была выше длинной и стала ниже — это сигнал на продажу (-1)

    for i in range(1, len(df)):
        if df[short_ema_col].iloc[i-1] < df[long_ema_col].iloc[i-1] and df[short_ema_col].iloc[i] > df[long_ema_col].iloc[i]:
            df.at[i, 'signal'] = 1   # Buy signal
        elif df[short_ema_col].iloc[i-1] > df[long_ema_col].iloc[i-1] and df[short_ema_col].iloc[i] < df[long_ema_col].iloc[i]:
            df.at[i, 'signal'] = -1  # Sell signal

    return df

# Пример использования
if __name__ == "__main__":
    # Загружаем датасет с рассчитанными EMA (см. предыдущий модуль)
    df = pd.read_csv('./_03_datasets/BTCUSDT_15m.csv', parse_dates=['timestamp'])

    # Предполагаем, что EMA уже добавлены (либо добавить расчёт здесь)
    df = detect_ema_crossover_signals(df, short_period=9, long_period=21)

    # Показываем строки, где есть сигналы
    signals = df[df['signal'] != 0]
    print(signals[['timestamp', 'close', 'EMA_9', 'EMA_21', 'signal']])

# _06_visualization/plot_signals.py

import pandas as pd
import matplotlib.pyplot as plt

def plot_ema_signals(df, short_period=9, long_period=21):
    """
    Функция для построения графика цены с наложенными EMA и торговыми сигналами.

    :param df: DataFrame с колонками ['timestamp', 'close', 'EMA_xx', 'signal']
    :param short_period: Период короткой EMA
    :param long_period: Период длинной EMA
    """
    plt.figure(figsize=(14, 7))  # Размер графика

    # Строим линию цены (close)
    plt.plot(df['timestamp'], df['close'], label='Close Price', linewidth=1)

    # Строим EMA линии
    plt.plot(df['timestamp'], df[f'EMA_{short_period}'], label=f'EMA {short_period}', linestyle='--')
    plt.plot(df['timestamp'], df[f'EMA_{long_period}'], label=f'EMA {long_period}', linestyle='--')

    # Добавляем точки входа и выхода
    buy_signals = df[df['signal'] == 1]
    sell_signals = df[df['signal'] == -1]

    # Отмечаем buy сигналы зелёными треугольниками
    plt.scatter(buy_signals['timestamp'], buy_signals['close'], marker='^', color='green', label='Buy Signal', s=100)

    # Отмечаем sell сигналы красными перевёрнутыми треугольниками
    plt.scatter(sell_signals['timestamp'], sell_signals['close'], marker='v', color='red', label='Sell Signal', s=100)

    # Настройки графика
    plt.title(f"Торговые сигналы по стратегии EMA Crossover ({short_period}/{long_period})")
    plt.xlabel('Время')
    plt.ylabel('Цена')
    plt.legend()
    plt.grid(True)
    plt.tight_layout()

    # Показываем график
    plt.show()

# Пример использования
if __name__ == "__main__":
    # Загружаем датасет с рассчитанными EMA и сигналами
    df = pd.read_csv('./_03_datasets/BTCUSDT_15m.csv', parse_dates=['timestamp'])

    # Строим график
    plot_ema_signals(df, short_period=9, long_period=21)

# _07_management/risk_manager.py

from _00_config import config

def calculate_position_size(balance, price):
    """
    Рассчитываем размер позиции на основе баланса и настроек риска.

    :param balance: Текущий баланс в USDT
    :param price: Текущая цена актива
    :return: Размер позиции в BTC (или другом активе)
    """
    # Ограничение: максимальный процент от баланса в одной сделке
    risk_usdt = balance * (config.MAX_TRADE_SIZE_PERCENT / 100)

    # Конвертируем сумму в количество актива
    position_size = risk_usdt / price

    return round(position_size, 6)  # Округляем до 6 знаков (актуально для крипты)

def calculate_sl_tp(entry_price):
    """
    Рассчитываем уровни стоп-лосса и тейк-профита.

    :param entry_price: Цена входа в сделку
    :return: Кортеж (stop_loss_price, take_profit_price)
    """
    stop_loss = entry_price * (1 - config.STOP_LOSS_PERCENT / 100)
    take_profit = entry_price * (1 + config.TAKE_PROFIT_PERCENT / 100)

    return round(stop_loss, 2), round(take_profit, 2)

# Пример использования
if __name__ == "__main__":
    current_balance = 1000  # Допустим, у нас 1000 USDT
    current_price = 30000   # Цена BTC

    # Расчёт размера позиции
    size = calculate_position_size(current_balance, current_price)
    print(f"Рекомендованный размер позиции: {size} BTC")

    # Расчёт уровней SL и TP
    sl, tp = calculate_sl_tp(current_price)
    print(f"Stop Loss: {sl} USDT, Take Profit: {tp} USDT")

MAX_TRADE_SIZE_PERCENT = 2  # Не более 2% от депозита в одной сделке
STOP_LOSS_PERCENT = 1.5     # Стоп-лосс на 1.5% ниже входа
TAKE_PROFIT_PERCENT = 3.0   # Тейк-профит на 3% выше входа

# _08_baktesting/backtester.py

import pandas as pd  # Библиотека для работы с таблицами данных (DataFrame)
from _07_management.risk_manager import calculate_sl_tp  # Импорт функции расчёта уровней стоп-лосса и тейк-профита

def run_backtest(df, initial_balance=1000):
    """
    Простой бэктестинг стратегии на основе сигналов.
    
    :param df: DataFrame с колонками ['timestamp', 'close', 'signal']
    :param initial_balance: Стартовый баланс в USDT
    :return: Список логов всех сделок
    """
    balance = initial_balance  # Устанавливаем начальный баланс
    position = 0               # Переменная для отслеживания позиции (0 = нет позиции, 1 = LONG, -1 = SHORT)
    entry_price = 0            # Цена входа в сделку
    trade_log = []             # Список для хранения логов всех совершённых сделок

    # Проходим по каждой строке DataFrame (по каждой свече)
    for index, row in df.iterrows():
        signal = row['signal']  # Получаем значение сигнала на текущей свече
        price = row['close']    # Получаем цену закрытия свечи

        # --- Открытие позиции ---
        if position == 0 and signal != 0:  # Если нет открытой позиции и есть торговый сигнал
            entry_price = price            # Фиксируем цену входа
            sl, tp = calculate_sl_tp(entry_price)  # Рассчитываем уровни стоп-лосса и тейк-профита
            position = signal              # Открываем позицию (1 для LONG, -1 для SHORT)
            trade_log.append(f"Открыта {'LONG' if signal == 1 else 'SHORT'} по цене {price} | SL: {sl}, TP: {tp}")

        # --- Проверка условий закрытия позиции ---
        if position != 0:  # Если позиция открыта
            if position == 1:  # Если это LONG-позиция
                if price <= sl or price >= tp:  # Если сработал SL или TP
                    result = tp - entry_price if price >= tp else sl - entry_price  # Расчёт прибыли/убытка
                    balance += result  # Корректируем баланс
                    trade_log.append(f"Закрыт LONG по цене {price} | Результат: {result:.2f} USDT")
                    position = 0  # Закрываем позицию

            elif position == -1:  # Если это SHORT-позиция
                if price >= sl or price <= tp:  # Если сработал SL или TP
                    result = entry_price - tp if price <= tp else entry_price -

# _09_dashboard/dash_dashboard.py

import dash  # Фреймворк для создания веб-дашбордов
from dash import html, dcc  # Компоненты интерфейса
import plotly.graph_objs as go  # Для построения графиков
import pandas as pd  # Работа с данными
from datetime import datetime, timedelta  # Для генерации временных меток
import numpy as np  # Для создания тестовых данных

# --- Генерация тестовых данных для примера ---
dates = [datetime.now() - timedelta(minutes=15*i) for i in range(20)][::-1]  # Последние 20 точек времени
balances = np.cumsum(np.random.randn(20)) + 1000  # Симуляция баланса

df = pd.DataFrame({'time': dates, 'balance': balances})  # Создаем DataFrame

# --- Инициализация Dash-приложения ---
app = dash.Dash(__name__)

# --- Описание интерфейса дашборда ---
app.layout = html.Div(children=[
    html.H1(children='Trading Bot Dashboard', style={'textAlign': 'center'}),  # Заголовок

    html.Div(children=f"Статус бота: Активен", style={'textAlign': 'center', 'fontSize':20}),  # Статус бота
    html.Div(children=f"Текущий баланс: {df['balance'].iloc[-1]:.2f} USDT", style={'textAlign': 'center', 'fontSize':20}),  # Баланс

    # График изменения баланса
    dcc.Graph(
        id='balance-graph',
        figure={
            'data': [
                go.Scatter(
                    x=df['time'],
                    y=df['balance'],
                    mode='lines+markers',
                    name='Equity Curve'
                )
            ],
            'layout': go.Layout(
                title='Динамика Баланса',
                xaxis={'title': 'Время'},
                yaxis={'title': 'Баланс (USDT)'},
                template='plotly_dark'
            )
        }
    )
])

# --- Запуск сервера ---
if __name__ == '__main__':
    app.run_server(debug=True)

pip install dash plotly pandas numpy

python _09_dashboard/dash_dashboard.py

# _10_trading/trading_bot.py

import time  # Импортируем модуль для создания задержек между итерациями цикла
from _03_datasets.dataset_manager import DatasetManager  # Импортируем менеджер для работы с датасетами
from _04_indicators.ema_calculator import calculate_ema  # Импорт функции расчёта EMA
from _05_patterns.ema_crossover_strategy import detect_ema_crossover_signals  # Импорт стратегии поиска сигналов
from _07_management.risk_manager import calculate_position_size, calculate_sl_tp  # Импорт функций риск-менеджмента
from _01_sources.bybit_client import session  # Импорт сессии подключения к API ByBit
from _00_config import config  # Импорт конфигурационных параметров проекта

def run_trading_bot():
    """
    Основной цикл работы торгового бота.
    Здесь происходит получение данных, анализ и отправка ордеров.
    """
    balance = 1000  # Устанавливаем стартовый баланс (в реальном боте получаем с биржи)

    # Инициализируем менеджер данных для выбранной торговой пары и таймфрейма
    dm = DatasetManager(config.TRADING_PAIR, config.TIMEFRAME)

    while True:  # Запускаем бесконечный цикл для постоянной работы бота
        dm.refresh_data()  # Обновляем данные (запуск ETL и загрузка актуальных свечей)
        df = dm.get_latest_data(50)  # Получаем последние 50 свечей для анализа

        # Расчитываем две экспоненциальные скользящие средние (EMA 9 и EMA 21)
        df = calculate_ema(df, period=9)
        df = calculate_ema(df, period=21)

        # Определяем торговые сигналы на основе пересечения EMA
        df = detect_ema_crossover_signals(df, short_period=9, long_period=21)

        # Получаем последний сигнал (на текущей свече)
        last_signal = df['signal'].iloc[-1]

        if last_signal != 0:  # Если есть сигнал (1 = Buy, -1 = Sell)
            current_price = df['close'].iloc[-1]  # Получаем текущую цену актива
            position_size = calculate_position_size(balance, current_price)  # Рассчитываем безопасный объем сделки
            sl, tp = calculate_sl_tp(current_price)  # Рассчитываем уровни стоп-лосса и тейк-профита

            # Отправляем рыночный ордер на биржу ByBit через API
            order = session.place_order(
                category="linear",  # Тип рынка (фьючерсы)
                symbol=config.TRADING_PAIR,  # Торговая пара
                side="Buy" if last_signal == 1 else "Sell",  # Направление сделки в зависимости от сигнала
                orderType="Market",  # Рыночный ордер
                qty=position_size,  # Количество актива
                stopLoss=str(sl),  # Уровень стоп-лосса
                takeProfit=str(tp)  # Уровень тейк-профита
            )

            # Выводим информацию об отправленном ордере
            print(f"Отправлен ордер: {order}")
        else:
            # Если сигнала нет, бот просто ждет следующей проверки
            print("Сигнала нет. Ожидание следующего цикла...")

        time.sleep(60)  # Пауза на 60 секунд перед следующей итерацией (зависит от таймфрейма стратегии)

# Точка входа для запуска бота
if __name__ == "__main__":
    run_trading_bot()  # Запуск основного цикла торгового бота

trading_bot_project/
├── _00_config/
├── _01_sources/
├── _02_etl/
├── _03_datasets/
├── _04_indicators/
├── _05_patterns/
├── _06_visualization/
├── _07_management/
├── _08_baktesting/
├── _09_dashboard/
├── _10_trading/
└── main.py   <-- Главный скрипт запуска бота

# main.py

"""
Главный скрипт для запуска торгового бота на ByBit.
Этот файл объединяет все модули и инициирует работу системы.
"""

import threading  # Для параллельного запуска дашборда и бота
from _09_dashboard.dash_dashboard import app  # Импорт Dash-приложения
from _10_trading.trading_bot import run_trading_bot  # Импорт функции запуска торгового бота

def start_dashboard():
    """
    Запуск веб-дашборда в отдельном потоке.
    """
    app.run_server(debug=False, use_reloader=False)  # Запуск Dash без режима перезагрузки

def start_bot():
    """
    Запуск торгового бота.
    """
    run_trading_bot()

if __name__ == "__main__":
    # Создаем поток для дашборда
    dashboard_thread = threading.Thread(target=start_dashboard)
    dashboard_thread.start()  # Запускаем дашборд

    # Запускаем бота в основном потоке
    start_bot()

pip install dash plotly pandas numpy pybit

python main.py