Crypto Алготрейдинг

Як створити криптовалютний торговий бот на Python

20 лютого 2026 | 35 хв читання

У 2010 році програміст Ласло Ханьєц заплатив 10 000 біткоїнів за дві піци. Сьогодні це понад 500 мільйонів доларів. Криптовалютний ринок — це машина волатильності, де ціни можуть змінюватися на 20% за годину. Люди не можуть торгувати 24/7, але боти — можуть. У цій статті ми розберемо, як створити власного торгового бота від архітектури до deployment, і чому більшість початківців втрачають гроші.

Що таке алгоритмічна торгівля?

Алготрейдинг (algorithmic trading) — це використання комп'ютерних програм для автоматичного виконання торгових операцій за заздалегідь визначеними правилами. За даними різних досліджень, понад 70% обсягу торгів на традиційних біржах здійснюється алгоритмами.

Архітектура торгового бота

Професійний торговий бот складається з кількох ключових компонентів:

trading_bot/
├── config/
│   ├── __init__.py
│   ├── settings.py         # Конфігурація (НЕ API ключі!)
│   └── pairs.json           # Торгові пари та їх параметри
├── core/
│   ├── __init__.py
│   ├── exchange.py          # Клас для роботи з біржею
│   ├── data_feed.py         # Отримання ринкових даних
│   └── order_manager.py     # Управління ордерами
├── strategies/
│   ├── __init__.py
│   ├── base_strategy.py     # Базовий клас стратегії
│   ├── sma_crossover.py     # SMA стратегія
│   ├── rsi_strategy.py      # RSI стратегія
│   └── grid_trading.py      # Grid trading
├── risk/
│   ├── __init__.py
│   ├── position_sizing.py   # Розрахунок розміру позиції
│   └── stop_loss.py         # Stop-loss логіка
├── backtesting/
│   ├── __init__.py
│   ├── backtest_engine.py   # Движок бектестингу
│   └── metrics.py           # Метрики ефективності
├── utils/
│   ├── __init__.py
│   ├── logger.py            # Логування
│   └── notifications.py     # Telegram/Email сповіщення
├── tests/
├── .env.example
├── requirements.txt
└── main.py                  # Точка входу

Підключення до біржі: Binance API

Binance — найбільша криптобіржа за обсягом торгів. Її API надає доступ до ринкових даних, розміщення ордерів та управління акаунтом.

# core/exchange.py
import os
from binance.client import Client
from binance.exceptions import BinanceAPIException
from dotenv import load_dotenv
import logging

load_dotenv()
logger = logging.getLogger(__name__)

class BinanceExchange:
    """
    Клас-обгортка для роботи з Binance API.
    Інкапсулює всі операції з біржею.
    """

    def __init__(self, testnet: bool = True):
        """
        Args:
            testnet: True для тестової мережі (рекомендовано для розробки)
        """
        self.testnet = testnet

        if testnet:
            # Testnet API endpoints
            self.client = Client(
                api_key=os.getenv('BINANCE_TESTNET_KEY'),
                api_secret=os.getenv('BINANCE_TESTNET_SECRET'),
                testnet=True
            )
            logger.info("Connected to Binance TESTNET")
        else:
            self.client = Client(
                api_key=os.getenv('BINANCE_API_KEY'),
                api_secret=os.getenv('BINANCE_API_SECRET')
            )
            logger.info("Connected to Binance MAINNET")

    def get_ticker_price(self, symbol: str) -> float:
        """Отримати поточну ціну"""
        ticker = self.client.get_symbol_ticker(symbol=symbol)
        return float(ticker['price'])

    def get_klines(self, symbol: str, interval: str, limit: int = 500) -> list:
        """
        Отримати історичні свічки (OHLCV дані).

        Args:
            symbol: Торгова пара (напр. 'BTCUSDT')
            interval: Таймфрейм ('1m', '5m', '1h', '1d', тощо)
            limit: Кількість свічок (макс. 1000)

        Returns:
            Список свічок з форматом:
            [open_time, open, high, low, close, volume, ...]
        """
        klines = self.client.get_klines(
            symbol=symbol,
            interval=interval,
            limit=limit
        )
        return klines

    def get_account_balance(self, asset: str = 'USDT') -> float:
        """Отримати баланс активу"""
        account = self.client.get_account()
        for balance in account['balances']:
            if balance['asset'] == asset:
                return float(balance['free'])
        return 0.0

    def place_market_order(self, symbol: str, side: str, quantity: float) -> dict:
        """
        Розмістити ринковий ордер.

        Args:
            symbol: Торгова пара
            side: 'BUY' або 'SELL'
            quantity: Кількість базового активу

        Returns:
            Інформація про виконаний ордер
        """
        try:
            order = self.client.create_order(
                symbol=symbol,
                side=side,
                type='MARKET',
                quantity=quantity
            )
            logger.info(f"Market {side} order executed: {order['orderId']}")
            return order
        except BinanceAPIException as e:
            logger.error(f"Order failed: {e.message}")
            raise

    def place_limit_order(self, symbol: str, side: str,
                          quantity: float, price: float) -> dict:
        """Розмістити лімітний ордер"""
        try:
            order = self.client.create_order(
                symbol=symbol,
                side=side,
                type='LIMIT',
                timeInForce='GTC',  # Good Till Cancelled
                quantity=quantity,
                price=str(price)
            )
            logger.info(f"Limit {side} order placed: {order['orderId']} @ {price}")
            return order
        except BinanceAPIException as e:
            logger.error(f"Order failed: {e.message}")
            raise

Технічний аналіз: індикатори

Технічний аналіз — основа більшості торгових стратегій. Індикатори — це математичні формули, що аналізують ціну та об'єм для генерації сигналів.

1. Moving Averages (Ковзні середні)

import pandas as pd
import numpy as np

def calculate_sma(prices: pd.Series, period: int) -> pd.Series:
    """
    Simple Moving Average — проста ковзна середня.

    SMA = (P1 + P2 + ... + Pn) / n

    Де P — ціна закриття, n — період.
    """
    return prices.rolling(window=period).mean()

def calculate_ema(prices: pd.Series, period: int) -> pd.Series:
    """
    Exponential Moving Average — експоненційна ковзна середня.

    EMA надає більшу вагу недавнім цінам.
    Multiplier = 2 / (period + 1)
    EMA = (Price - EMA_prev) * Multiplier + EMA_prev
    """
    return prices.ewm(span=period, adjust=False).mean()

# Приклад використання
df = pd.DataFrame({'close': [100, 102, 101, 105, 110, 108, 112, 115, 113, 118]})
df['SMA_5'] = calculate_sma(df['close'], 5)
df['EMA_5'] = calculate_ema(df['close'], 5)
print(df)

2. RSI (Relative Strength Index)

def calculate_rsi(prices: pd.Series, period: int = 14) -> pd.Series:
    """
    RSI — індекс відносної сили.

    Вимірює швидкість та зміну цінових рухів.
    RSI = 100 - (100 / (1 + RS))
    RS = Average Gain / Average Loss

    Інтерпретація:
    - RSI > 70: перекупленість (можливе падіння)
    - RSI < 30: перепроданість (можливе зростання)
    """
    delta = prices.diff()

    # Розділяємо на прибутки та збитки
    gain = delta.where(delta > 0, 0)
    loss = (-delta).where(delta < 0, 0)

    # Середні прибутки та збитки (EMA)
    avg_gain = gain.ewm(span=period, min_periods=period).mean()
    avg_loss = loss.ewm(span=period, min_periods=period).mean()

    # Відносна сила
    rs = avg_gain / avg_loss

    # RSI
    rsi = 100 - (100 / (1 + rs))
    return rsi

# Приклад
df['RSI'] = calculate_rsi(df['close'])
print(f"RSI: {df['RSI'].iloc[-1]:.2f}")

3. Bollinger Bands

def calculate_bollinger_bands(prices: pd.Series, period: int = 20,
                               std_dev: int = 2) -> tuple:
    """
    Bollinger Bands — смуги Боллінджера.

    Middle Band = SMA(period)
    Upper Band = Middle Band + (std_dev * Standard Deviation)
    Lower Band = Middle Band - (std_dev * Standard Deviation)

    Інтерпретація:
    - Ціна торкається Upper Band: можливий відкат вниз
    - Ціна торкається Lower Band: можливий відскок вгору
    - Звуження смуг: очікується сильний рух
    """
    middle = prices.rolling(window=period).mean()
    std = prices.rolling(window=period).std()

    upper = middle + (std_dev * std)
    lower = middle - (std_dev * std)

    return upper, middle, lower

upper, middle, lower = calculate_bollinger_bands(df['close'])
df['BB_Upper'] = upper
df['BB_Middle'] = middle
df['BB_Lower'] = lower

Реалізація торгової стратегії

# strategies/sma_crossover.py
from abc import ABC, abstractmethod
from enum import Enum
import pandas as pd

class Signal(Enum):
    BUY = "BUY"
    SELL = "SELL"
    HOLD = "HOLD"

class BaseStrategy(ABC):
    """Абстрактний базовий клас для всіх стратегій"""

    @abstractmethod
    def generate_signal(self, df: pd.DataFrame) -> Signal:
        """Генерує торговий сигнал на основі даних"""
        pass

class SMACrossover(BaseStrategy):
    """
    Стратегія перетину ковзних середніх.

    Логіка:
    - BUY: коротка SMA перетинає довгу SMA знизу вгору (golden cross)
    - SELL: коротка SMA перетинає довгу SMA зверху вниз (death cross)
    - HOLD: немає перетину
    """

    def __init__(self, short_period: int = 10, long_period: int = 50):
        self.short_period = short_period
        self.long_period = long_period

    def calculate_indicators(self, df: pd.DataFrame) -> pd.DataFrame:
        """Розраховує індикатори стратегії"""
        df = df.copy()
        df['SMA_short'] = df['close'].rolling(window=self.short_period).mean()
        df['SMA_long'] = df['close'].rolling(window=self.long_period).mean()

        # Позиція: 1 якщо short > long, інакше 0
        df['position'] = (df['SMA_short'] > df['SMA_long']).astype(int)

        # Сигнал: зміна позиції
        df['signal'] = df['position'].diff()

        return df

    def generate_signal(self, df: pd.DataFrame) -> Signal:
        """Генерує сигнал на основі останньої свічки"""
        df = self.calculate_indicators(df)
        last_signal = df['signal'].iloc[-1]

        if last_signal == 1:
            return Signal.BUY
        elif last_signal == -1:
            return Signal.SELL
        else:
            return Signal.HOLD

class RSIStrategy(BaseStrategy):
    """
    Стратегія на основі RSI.

    Логіка:
    - BUY: RSI < oversold_level (перепроданість)
    - SELL: RSI > overbought_level (перекупленість)
    """

    def __init__(self, period: int = 14,
                 oversold: int = 30, overbought: int = 70):
        self.period = period
        self.oversold = oversold
        self.overbought = overbought

    def generate_signal(self, df: pd.DataFrame) -> Signal:
        df = df.copy()
        df['RSI'] = calculate_rsi(df['close'], self.period)
        current_rsi = df['RSI'].iloc[-1]

        if current_rsi < self.oversold:
            return Signal.BUY
        elif current_rsi > self.overbought:
            return Signal.SELL
        else:
            return Signal.HOLD

Ризик-менеджмент: як не втратити все

Ризик-менеджмент — це те, що відрізняє професіоналів від аматорів. Навіть найкраща стратегія без управління ризиками призведе до втрати капіталу.

# risk/position_sizing.py

class PositionSizer:
    """
    Розрахунок оптимального розміру позиції.

    Використовує Fixed Fractional Position Sizing:
    Position Size = (Account * Risk%) / (Entry - Stop Loss)
    """

    def __init__(self, account_balance: float, risk_per_trade: float = 0.01):
        """
        Args:
            account_balance: Баланс рахунку в USDT
            risk_per_trade: Відсоток ризику на угоду (0.01 = 1%)
        """
        self.account_balance = account_balance
        self.risk_per_trade = risk_per_trade

    def calculate_position_size(self, entry_price: float,
                                 stop_loss_price: float) -> float:
        """
        Розраховує розмір позиції на основі ризику.

        Приклад:
        - Баланс: $10,000
        - Ризик на угоду: 1% = $100
        - Вхід: $50,000 (BTC)
        - Стоп-лосс: $48,000
        - Ризик на одиницю: $2,000

        Розмір позиції = $100 / $2,000 = 0.05 BTC
        """
        risk_amount = self.account_balance * self.risk_per_trade
        risk_per_unit = abs(entry_price - stop_loss_price)

        if risk_per_unit == 0:
            return 0

        position_size = risk_amount / risk_per_unit
        return round(position_size, 8)

    def calculate_stop_loss(self, entry_price: float,
                            atr: float, multiplier: float = 2) -> float:
        """
        Розраховує стоп-лосс на основі ATR (Average True Range).

        ATR-based stop loss адаптується до волатильності ринку.
        """
        return entry_price - (atr * multiplier)

    def calculate_take_profit(self, entry_price: float,
                              stop_loss_price: float,
                              risk_reward_ratio: float = 2) -> float:
        """
        Розраховує тейк-профіт на основі Risk/Reward ratio.
        """
        risk = abs(entry_price - stop_loss_price)
        return entry_price + (risk * risk_reward_ratio)


# Приклад використання
sizer = PositionSizer(account_balance=10000, risk_per_trade=0.02)

entry = 50000  # BTC entry price
stop_loss = 48000  # 4% нижче
take_profit = sizer.calculate_take_profit(entry, stop_loss, 2)

position = sizer.calculate_position_size(entry, stop_loss)
print(f"Position size: {position} BTC")
print(f"Take profit: ${take_profit:,.0f}")
print(f"Risk/Reward: 1:{(take_profit - entry) / (entry - stop_loss):.1f}")

Backtesting: тестування на історичних даних

Backtesting — це симуляція стратегії на історичних даних. Це обов'язковий крок перед запуском бота на реальні гроші.

# backtesting/backtest_engine.py
import pandas as pd
import numpy as np
from typing import List, Tuple

class BacktestEngine:
    """
    Простий движок для backtesting торгових стратегій.
    """

    def __init__(self, initial_capital: float = 10000,
                 commission: float = 0.001):
        """
        Args:
            initial_capital: Початковий капітал
            commission: Комісія біржі (0.001 = 0.1%)
        """
        self.initial_capital = initial_capital
        self.commission = commission

    def run(self, df: pd.DataFrame, strategy) -> dict:
        """
        Запускає бектест стратегії.

        Args:
            df: DataFrame з OHLCV даними
            strategy: Об'єкт стратегії з методом generate_signal()

        Returns:
            Словник з результатами та метриками
        """
        df = strategy.calculate_indicators(df.copy())

        capital = self.initial_capital
        position = 0  # Кількість базового активу
        trades: List[dict] = []

        for i in range(1, len(df)):
            signal = df['signal'].iloc[i]
            price = df['close'].iloc[i]

            if signal == 1 and position == 0:  # BUY
                # Купуємо на весь капітал
                position = (capital * (1 - self.commission)) / price
                capital = 0
                trades.append({
                    'type': 'BUY',
                    'price': price,
                    'date': df.index[i],
                    'position': position
                })

            elif signal == -1 and position > 0:  # SELL
                # Продаємо всю позицію
                capital = position * price * (1 - self.commission)
                trades.append({
                    'type': 'SELL',
                    'price': price,
                    'date': df.index[i],
                    'position': position,
                    'capital': capital
                })
                position = 0

        # Закриваємо позицію в кінці
        if position > 0:
            capital = position * df['close'].iloc[-1] * (1 - self.commission)

        # Розрахунок метрик
        final_capital = capital
        total_return = (final_capital - self.initial_capital) / self.initial_capital
        num_trades = len(trades)

        # Buy & Hold для порівняння
        buy_hold_return = (df['close'].iloc[-1] / df['close'].iloc[0]) - 1

        return {
            'initial_capital': self.initial_capital,
            'final_capital': round(final_capital, 2),
            'total_return': f"{total_return:.2%}",
            'buy_hold_return': f"{buy_hold_return:.2%}",
            'num_trades': num_trades,
            'trades': trades
        }

    def calculate_sharpe_ratio(self, returns: pd.Series,
                                risk_free_rate: float = 0.02) -> float:
        """
        Коефіцієнт Шарпа — міра ризик-скоригованої дохідності.

        Sharpe = (Return - Risk-Free Rate) / Std(Returns)
        """
        excess_return = returns.mean() * 252 - risk_free_rate  # Annualized
        volatility = returns.std() * np.sqrt(252)
        return excess_return / volatility if volatility != 0 else 0

    def calculate_max_drawdown(self, equity_curve: pd.Series) -> float:
        """
        Maximum Drawdown — найбільше падіння від піку.

        MDD = (Peak - Trough) / Peak
        """
        peak = equity_curve.expanding(min_periods=1).max()
        drawdown = (equity_curve - peak) / peak
        return drawdown.min()


# Приклад бектесту
from binance.client import Client

# Завантаження історичних даних
client = Client()
klines = client.get_historical_klines(
    "BTCUSDT", Client.KLINE_INTERVAL_1HOUR,
    "1 Jan, 2025", "1 Jan, 2026"
)

df = pd.DataFrame(klines, columns=[
    'timestamp', 'open', 'high', 'low', 'close', 'volume',
    'close_time', 'quote_volume', 'trades', 'taker_buy_base',
    'taker_buy_quote', 'ignore'
])
df['close'] = pd.to_numeric(df['close'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)

# Запуск бектесту
strategy = SMACrossover(short_period=20, long_period=50)
engine = BacktestEngine(initial_capital=10000)
results = engine.run(df, strategy)

print(f"Initial Capital: ${results['initial_capital']:,}")
print(f"Final Capital: ${results['final_capital']:,}")
print(f"Strategy Return: {results['total_return']}")
print(f"Buy & Hold Return: {results['buy_hold_return']}")
print(f"Number of Trades: {results['num_trades']}")

Порівняння торгових стратегій

Deployment: запуск на сервері

Для стабільної роботи бот повинен працювати на сервері 24/7. Популярні варіанти:

• AWS EC2 / DigitalOcean — VPS від $5/місяць
• Google Cloud Run — serverless, платите за використання
• Raspberry Pi — локальний міні-сервер вдома

# docker-compose.yml
version: '3.8'

services:
  trading-bot:
    build: .
    restart: always
    env_file:
      - .env
    volumes:
      - ./logs:/app/logs
      - ./data:/app/data
    logging:
      driver: "json-file"
      options:
        max-size: "10m"
        max-file: "3"

  redis:
    image: redis:alpine
    restart: always

  # Опціонально: Grafana для моніторингу
  grafana:
    image: grafana/grafana
    ports:
      - "3000:3000"
    volumes:
      - grafana-data:/var/lib/grafana

volumes:
  grafana-data:

Рекомендовані бібліотеки

• python-binance

  Офіційний клієнт Binance API
  pip install python-binance
• ccxt

  Універсальний клієнт для 100+ бірж
  pip install ccxt
• ta-lib

  Технічний аналіз (150+ індикаторів)
  pip install TA-Lib
• backtrader

  Професійний фреймворк для backtesting
  pip install backtrader
• pandas-ta

  Технічні індикатори для Pandas
  pip install pandas-ta