Бот для торговли на бирже на Python: от простых до сложных решений

Расширенные примеры кода торгового бота

Полный бот с websocket и управлением капиталом

Пример использует Websocket для получения цен в реальном времени, логирование и динамический размер позиции.

import ccxt.pro as ccxtpro
import asyncio
import logging

logging.basicConfig(level=logging.INFO, format='%(asctime)s - %(message)s')

exchange = ccxtpro.binance()
symbol = 'BTC/USDT'
position = None
capital = 1000.0
fee = 0.001

async def trade():
    global position
    async for ticker in exchange.watch_ticker(symbol):
        price = ticker['last']
        # Простая стратегия: если цена выше SMA50 -> long
        # Здесь для примера используем фиктивные условия
        if price > 50000 and position != 'long':
            size = capital * 0.01 / price
            order = await exchange.create_market_buy_order(symbol, size)
            position = 'long'
            logging.info(f"Buy {size} BTC at {price}")
        elif price < 45000 and position == 'long':
            order = await exchange.create_market_sell_order(symbol, None)
            position = None
            logging.info(f"Sell at {price}")

asyncio.run(trade())

2025-04-01 12:00:05 - Buy 0.0002 BTC at 50123.4
2025-04-01 14:32:11 - Sell at 44890.1

Пояснение: используется ccxt.pro для асинхронного WebSocket, что снижает задержки. Размер позиции рассчитывается как 1% от капитала. Логирование помогает отслеживать действия.

Бот с машинным обучением (LSTM) для прогноза

Пример обучения простой LSTM на исторических ценах и генерации сигналов.

import numpy as np
import pandas as pd
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import LSTM, Dense
from sklearn.preprocessing import MinMaxScaler

data = pd.read_csv('btc_usd_1h.csv')['close'].values
scaler = MinMaxScaler()
data_scaled = scaler.fit_transform(data.reshape(-1,1))

# Создание последовательностей
def create_sequences(d, seq_len=60):
    X, y = [], []
    for i in range(len(d)-seq_len-1):
        X.append(d[i:i+seq_len])
        y.append(d[i+seq_len])
    return np.array(X), np.array(y)

X, y = create_sequences(data_scaled)
model = Sequential([
    LSTM(50, return_sequences=True, input_shape=(60,1)),
    LSTM(50),
    Dense(1)
])
model.compile(optimizer='adam', loss='mse')
model.fit(X, y, epochs=10, batch_size=32)

# Предсказание
last_60 = data_scaled[-60:].reshape(1,60,1)
pred = model.predict(last_60)
price_pred = scaler.inverse_transform(pred)
print(f"Предсказание следующей цены: {price_pred[0][0]:.2f}")

Предсказание следующей цены: 54321.67

Цель: генерация торговых сигналов на основе прогноза. Проблема: качество прогноза зависит от данных и архитектуры сети.

Backtesting с оптимизацией параметров

Используется backtrader с поиском лучших периодов SMA.

import backtrader as bt
import itertools

class SmaOpt(bt.Strategy):
    params = (('pfast', 10), ('pslow', 30))
    def __init__(self):
        self.sma_fast = bt.indicators.SMA(self.data.close, period=self.params.pfast)
        self.sma_slow = bt.indicators.SMA(self.data.close, period=self.params.pslow)
    def next(self):
        if self.sma_fast > self.sma_slow and not self.position:
            self.buy()
        elif self.sma_fast < self.sma_slow and self.position:
            self.sell()

cerebro = bt.Cerebro()
data = bt.feeds.YahooFinanceData(dataname='BTC-USD', fromdate='2023-01-01', todate='2024-01-01')
cerebro.adddata(data)

# Оптимизация
fast_range = range(5, 20, 5)
slow_range = range(20, 60, 10)
opt_results = []
for fast, slow in itertools.product(fast_range, slow_range):
    cerebro.addstrategy(SmaOpt, pfast=fast, pslow=slow)
    result = cerebro.run()
    opt_results.append((fast, slow, result[0].analyzers.getbyname(???).get_analysis()['total']['pnl']))

Результат: таблица с доходностью для каждой комбинации. Помогает выбрать лучшие параметры.