Создайте торгового бота на Python и Django: автоматизация трейдинга

# Создание виртуального окружения
python -m venv trading_bot_env

# Активация в Windows
trading_bot_env\Scripts\activate

# Активация в MacOS/Linux
source trading_bot_env/bin/activate

pip install django pandas numpy requests ccxt matplotlib plotly scikit-learn django-rest-framework celery redis

django-admin startproject trading_platform
cd trading_platform
python manage.py startapp bot
python manage.py startapp analytics

pip install python-dotenv

BINANCE_API_KEY=ваш_ключ
BINANCE_API_SECRET=ваш_секрет
DATABASE_URL=postgresql://user:password@localhost/dbname

# settings.py
LOGGING = {
'version': 1,
'disable_existing_loggers': False,
'formatters': {
'verbose': {
'format': '{levelname} {asctime} {module} {message}',
'style': '{',
},
},
'handlers': {
'file': {
'level': 'INFO',
'class': 'logging.FileHandler',
'filename': 'trading.log',
'formatter': 'verbose',
},
},
'loggers': {
'bot': {
'handlers': ['file'],
'level': 'INFO',
'propagate': True,
},
},
}

import ccxt
import os
from dotenv import load_dotenv

# Загружаем переменные окружения
load_dotenv()

# Инициализируем подключение к бирже
exchange = ccxt.binance({
'apiKey': os.getenv('BINANCE_API_KEY'),
'secret': os.getenv('BINANCE_API_SECRET'),
'enableRateLimit': True, # Важно для соблюдения ограничений API
})

# Получение баланса аккаунта
balance = exchange.fetch_balance()
print(f"BTC баланс: {balance['BTC']['free']}")

# Получение данных по текущим ценам
ticker = exchange.fetch_ticker('BTC/USDT')
print(f"Текущая цена BTC: {ticker['last']} USDT")

# Размещение рыночного ордера на покупку
order = exchange.create_market_buy_order('BTC/USDT', 0.001)
print(f"Ордер размещен: {order}")

# Получение открытых ордеров
open_orders = exchange.fetch_open_orders('BTC/USDT')
print(f"Открытые ордера: {open_orders}")

# bot/connectors/binance_connector.py
import ccxt
import logging
from django.conf import settings

logger = logging.getLogger('bot')

class BinanceConnector:
def __init__(self):
self.exchange = ccxt.binance({
'apiKey': settings.BINANCE_API_KEY,
'secret': settings.BINANCE_API_SECRET,
'enableRateLimit': True,
})

def get_market_data(self, symbol, timeframe='1h', limit=100):
"""Получает исторические данные по указанной паре"""
try:
ohlcv = self.exchange.fetch_ohlcv(symbol, timeframe, limit=limit)
logger.info(f"Получены данные {symbol} ({timeframe}): {len(ohlcv)} свечей")
return ohlcv
except Exception as e:
logger.error(f"Ошибка получения данных {symbol}: {str(e)}")
return None

def place_order(self, symbol, order_type, side, amount, price=None):
"""Размещает ордер на бирже"""
try:
params = {}
if order_type == 'limit' and price is None:
raise ValueError("Для лимитного ордера необходимо указать цену")

if order_type == 'market':
if side == 'buy':
order = self.exchange.create_market_buy_order(symbol, amount)
else:
order = self.exchange.create_market_sell_order(symbol, amount)
elif order_type == 'limit':
if side == 'buy':
order = self.exchange.create_limit_buy_order(symbol, amount, price)
else:
order = self.exchange.create_limit_sell_order(symbol, amount, price)

logger.info(f"Размещен ордер: {order}")
return order
except Exception as e:
logger.error(f"Ошибка размещения ордера {symbol}: {str(e)}")
return None

# bot/algorithms/moving_average.py
import pandas as pd
import numpy as np
import logging

logger = logging.getLogger('bot')

class MovingAverageStrategy:
def __init__(self, short_window=10, long_window=50):
self.short_window = short_window
self.long_window = long_window

def generate_signals(self, data):
"""
Генерирует торговые сигналы на основе пересечения скользящих средних.

Args:
data: список OHLCV-данных в формате [[timestamp, open, high, low, close, volume], ...]

Returns:
DataFrame с торговыми сигналами: 1 (покупка), -1 (продажа), 0 (ничего не делать)
"""
# Преобразуем данные в pandas DataFrame
df = pd.DataFrame(data, columns=['timestamp', 'open', 'high', 'low', 'close', 'volume'])
df['timestamp'] = pd.to_datetime(df['timestamp'], unit='ms')
df.set_index('timestamp', inplace=True)

# Добавляем скользящие средние
df['short_ma'] = df['close'].rolling(window=self.short_window).mean()
df['long_ma'] = df['close'].rolling(window=self.long_window).mean()

# Инициализируем сигналы
df['signal'] = 0.0

# Генерируем сигналы
# Сигнал покупки: короткая MA пересекает длинную MA снизу вверх
df['signal'] = np.where(df['short_ma'] > df['long_ma'], 1.0, 0.0)

# Создаем колонку с изменениями сигнала
df['position_change'] = df['signal'].diff()

# Логируем генерацию сигналов
buy_signals = df[df['position_change'] == 1.0]
sell_signals = df[df['position_change'] == -1.0]

logger.info(f"Сгенерировано сигналов покупки: {len(buy_signals)}")
logger.info(f"Сгенерировано сигналов продажи: {len(sell_signals)}")

return df

def execute_strategy(self, connector, symbol, amount, data=None):
"""
Выполняет стратегию на основе последних данных.

Args:
connector: объект коннектора биржи
symbol: торговая пара
amount: размер позиции
data: опциональные данные, если None – получаем с биржи

Returns:
dict с информацией о выполненном ордере или None
"""
if data is None:
# Получаем данные с биржи, если они не предоставлены
data = connector.get_market_data(symbol, '1h', 100)
if data is None:
logger.error("Не удалось получить данные для анализа")
return None

# Генерируем сигналы
signals_df = self.generate_signals(data)

# Проверяем последний сигнал
last_signal = signals_df['signal'].iloc[-1]
last_change = signals_df['position_change'].iloc[-1]

# Действуем только если есть изменение позиции
if last_change == 1.0: # Сигнал покупки
logger.info(f"Выполняем покупку {symbol}")
return connector.place_order(symbol, 'market', 'buy', amount)
elif last_change == -1.0: # Сигнал продажи
logger.info(f"Выполняем продажу {symbol}")
return connector.place_order(symbol, 'market', 'sell', amount)
else:
logger.info(f"Нет сигнала для действия по {symbol}")
return None

# bot/backtest/simple_backtest.py
import pandas as pd
import matplotlib.pyplot as plt
import numpy as np

def backtest_strategy(strategy, data, initial_capital=10000.0):
"""
Проводит бэктест стратегии на исторических данных.

Args:
strategy: объект стратегии с методом generate_signals
data: исторические данные
initial_capital: начальный капитал

Returns:
DataFrame с результатами бэктеста
"""
# Получаем сигналы
signals = strategy.generate_signals(data)

# Добавляем колонки для отслеживания позиции и капитала
signals['position'] = signals['signal']
signals['returns'] = signals['close'].pct_change()
signals['strategy_returns'] = signals['position'].shift(1) * signals['returns']
signals['equity'] = (1 + signals['strategy_returns']).cumprod() * initial_capital

# Вычисляем метрики эффективности
total_return = (signals['equity'].iloc[-1] / initial_capital – 1) * 100
annualized_return = (signals['equity'].iloc[-1] / initial_capital) ** (252 / len(signals)) – 1
annualized_return *= 100

sharpe_ratio = signals['strategy_returns'].mean() / signals['strategy_returns'].std() * np.sqrt(252)
max_drawdown = (signals['equity'] / signals['equity'].cummax() – 1).min() * 100

print(f"Общая доходность: {total_return:.2f}%")
print(f"Годовая доходность: {annualized_return:.2f}%")
print(f"Коэффициент Шарпа: {sharpe_ratio:.2f}")
print(f"Максимальная просадка: {max_drawdown:.2f}%")

# Визуализация результатов
plt.figure(figsize=(12, 8))

plt.subplot(2, 1, 1)
plt.plot(signals.index, signals['close'])
plt.plot(signals.index, signals['short_ma'], 'r--')
plt.plot(signals.index, signals['long_ma'], 'g--')
buy_signals = signals[signals['position_change'] == 1.0]
sell_signals = signals[signals['position_change'] == -1.0]
plt.plot(buy_signals.index, signals.loc[buy_signals.index, 'close'], '^', markersize=10, color='g')
plt.plot(sell_signals.index, signals.loc[sell_signals.index, 'close'], 'v', markersize=10, color='r')
plt.title('Цена и скользящие средние')

plt.subplot(2, 1, 2)
plt.plot(signals.index, signals['equity'])
plt.title('Кривая капитала')

plt.tight_layout()
plt.savefig('backtest_results.png')

return signals

# bot/models.py
from django.db import models
from django.contrib.auth.models import User

class TradingPair(models.Model):
"""Модель для хранения торговых пар"""
symbol = models.CharField(max_length=20, unique=True)
base_asset = models.CharField(max_length=10)
quote_asset = models.CharField(max_length=10)
min_lot_size = models.DecimalField(max_digits=18, decimal_places=8, default=0)

def __str__(self):
return self.symbol

class Strategy(models.Model):
"""Модель для хранения торговых стратегий"""
STRATEGY_TYPES = (
('ma_cross', 'Пересечение скользящих средних'),
('rsi', 'RSI'),
('macd', 'MACD'),
('grid', 'Сетка ордеров'),
('custom', 'Пользовательская'),
)

name = models.CharField(max_length=100)
type = models.CharField(max_length=20, choices=STRATEGY_TYPES)
description = models.TextField(blank=True)
parameters = models.JSONField(default=dict)

def __str__(self):
return self.name

class Bot(models.Model):
"""Модель торгового бота"""
STATUS_CHOICES = (
('active', 'Активен'),
('paused', 'Приостановлен'),
('stopped', 'Остановлен'),
)

name = models.CharField(max_length=100)
user = models.ForeignKey(User, on_delete=models.CASCADE)
pair = models.ForeignKey(TradingPair, on_delete=models.CASCADE)
strategy = models.ForeignKey(Strategy, on_delete=models.CASCADE)
status = models.CharField(max_length=20, choices=STATUS_CHOICES, default='stopped')
amount_per_trade = models.DecimalField(max_digits=18, decimal_places=8)
max_open_trades = models.IntegerField(default=1)
created_at = models.DateTimeField(auto_now_add=True)
updated_at = models.DateTimeField(auto_now=True)

def __str__(self):
return f"{self.name} ({self.user.username})"

class Trade(models.Model):
"""Модель для хранения информации о сделках"""
SIDE_CHOICES = (
('buy', 'Покупка'),
('sell', 'Продажа'),
)

STATUS_CHOICES = (
('open', 'Открыта'),
('closed', 'Закрыта'),
('canceled', 'Отменена'),
)

bot = models.ForeignKey(Bot, on_delete=models.CASCADE)
order_id = models.CharField(max_length=100)
symbol = models.CharField(max_length=20)
side = models.CharField(max_length=4, choices=SIDE_CHOICES)
quantity = models.DecimalField(max_digits=18, decimal_places=8)
price = models.DecimalField(max_digits=18, decimal_places=8)
commission = models.DecimalField(max_digits=18, decimal_places=8, default=0)
status = models.CharField(max_length=10, choices=STATUS_CHOICES)
pnl = models.DecimalField(max_digits=18, decimal_places=8, null=True, blank=True)
opened_at = models.DateTimeField()
closed_at = models.DateTimeField(null=True, blank=True)

def __str__(self):
return f"{self.side.capitalize()} {self.quantity} {self.symbol} at {self.price}"

# bot/views.py
from django.shortcuts import render, redirect, get_object_or_404
from django.contrib.auth.decorators import login_required
from django.contrib import messages
from django.http import JsonResponse
from .models import Bot, Strategy, TradingPair, Trade
from .forms import BotForm, StrategyForm
import json

@login_required
def dashboard(request):
"""Показывает дашборд пользователя с его ботами и статистикой"""
bots = Bot.objects.filter(user=request.user)

# Получаем статистику по активным ботам
active_bots = bots.filter(status='active').count()
total_trades = Trade.objects.filter(bot__user=request.user).count()

# Вычисляем общий P&L
trades = Trade.objects.filter(bot__user=request.user, status='closed')
total_pnl = sum(trade.pnl for trade in trades if trade.pnl)

context = {
'bots': bots,
'active_bots': active_bots,
'total_trades': total_trades,
'total_pnl': total_pnl
}

return render(request, 'bot/dashboard.html', context)

@login_required
def bot_detail(request, bot_id):
"""Детальная информация о конкретном боте"""
bot = get_object_or_404(Bot, id=bot_id, user=request.user)
trades = Trade.objects.filter(bot=bot).order_by('-opened_at')

# Расчет статистики для конкретного бота
closed_trades = trades.filter(status='closed')
win_trades = closed_trades.filter(pnl__gt=0).count()
loss_trades = closed_trades.filter(pnl__lte=0).count()

if closed_trades:
win_rate = (win_trades / closed_trades.count()) * 100
else:
win_rate = 0

context = {
'bot': bot,
'trades': trades[:20], # Показываем последние 20 сделок
'win_rate': win_rate,
'win_trades': win_trades,
'loss_trades': loss_trades
}

return render(request, 'bot/bot_detail.html', context)

@login_required
def create_bot(request):
"""Создание нового бота"""
if request.method == 'POST':
form = BotForm(request.POST)
if form.is_valid():
bot = form.save(commit=False)
bot.user = request.user
bot.save()
messages.success(request, f"Бот {bot.name} успешно создан!")
return redirect('bot_detail', bot_id=bot.id)
else:
form = BotForm()

context = {
'form': form,
'trading_pairs': TradingPair.objects.all(),
'strategies': Strategy.objects.all()
}

return render(request, 'bot/create_bot.html', context)

@login_required
def toggle_bot(request, bot_id):
"""Включение/выключение бота"""
bot = get_object_or_404(Bot, id=bot_id, user=request.user)

if request.method == 'POST':
action = request.POST.get('action')

if action == 'start' and bot.status != 'active':
bot.status = 'active'
messages.success(request, f"Бот {bot.name} запущен!")

# Здесь мы бы запускали асинхронную задачу для бота
# from bot.tasks import start_bot_task
# start_bot_task.delay(bot.id)

elif action == 'pause' and bot.status == 'active':
bot.status = 'paused'
messages.success(request, f"Бот {bot.name} приостановлен!")

elif action == 'stop':
bot.status = 'stopped'
messages.success(request, f"Бот {bot.name} остановлен!")

bot.save()

return redirect('bot_detail', bot_id=bot.id)

<!-- templates/bot/dashboard.html -->
{% extends 'base.html' %}

{% block content %}
<div class="dashboard-header">
<h1>Торговый терминал</h1>
<div class="dashboard-stats">
<div class="stat-card">
<h3>{{ active_bots }}</h3>
<p>Активных ботов</p>
</div>
<div class="stat-card">
<h3>{{ total_trades }}</h3>
<p>Всего сделок</p>
</div>
<div class="stat-card {% if total_pnl > 0 %}profit{% else %}loss{% endif %}">
<h3>{{ total_pnl|floatformat:2 }} USDT</h3>
<p>Общий P&L</p>
</div>
</div>
<a href="{% url 'create_bot' %}" class="btn btn-primary">Создать бота</a>
</div>

<div class="bots-container">
{% for bot in bots %}
<div class="bot-card">
<div class="bot-header">
<h2>{{ bot.name }}</h2>
<span class="badge badge-{{ bot.status }}">{{ bot.get_status_display }}</span>
</div>
<div class="bot-details">
<p><strong>Пара:</strong> {{ bot.pair }}</p>
<p><strong>Стратегия:</strong> {{ bot.strategy }}</p>
<p><strong>Размер позиции:</strong> {{ bot.amount_per_trade }} {{ bot.pair.base_asset }}</p>
</div>
<div class="bot-actions">
<a href="{% url 'bot_detail' bot.id %}" class="btn btn-info">Подробнее</a>
{% if bot.status != 'active' %}
<form method="post" action="{% url 'toggle_bot' bot.id %}">
{% csrf_token %}
<input type="hidden" name="action" value="start">
<button type="submit" class="btn btn-success">Запустить</button>
</form>
{% else %}
<form method="post" action="{% url 'toggle_bot' bot.id %}">
{% csrf_token %}
<input type="hidden" name="action" value="pause">
<button type="submit" class="btn btn-warning">Пауза</button>
</form>
{% endif %}
<form method="post" action="{% url 'toggle_bot' bot.id %}">
{% csrf_token %}
<input type="hidden" name="action" value="stop">
<button type="submit" class="btn btn-danger">Стоп</button>
</form>
</div>
</div>
{% empty %}
<div class="no-bots">
<p>У вас пока нет ботов. Создайте первого бота для начала торговли!</p>
<a href="{% url 'create_bot' %}" class="btn btn-primary">Создать бота</a>
</div>
{% endfor %}
</div>
{% endblock %}

// templates/bot/bot_detail.html (JavaScript часть)
<script src="https://cdn.jsdelivr.net/npm/chart.js@3.7.1/dist/chart.min.js"></script>

<script>
document.addEventListener('DOMContentLoaded', function() {
// Получаем данные о сделках из шаблона
const tradesData = {{ trades_json|safe }};

// Подготавливаем данные для графика
const dates = tradesData.map(trade => new Date(trade.opened_at));
const prices = tradesData.map(trade => trade.price);
const sides = tradesData.map(trade => trade.side);

// Создаем график
const ctx = document.getElementById('tradesChart').getContext('2d');
const chart = new Chart(ctx, {
type: 'line',
data: {
labels: dates,
datasets: [{
label: 'Цена',
data: prices,
fill: false,
borderColor: 'rgba(75, 192, 192, 1)',
tension: 0.1
}]
},
options: {
scales: {
x: {
type: 'time',
time: {
unit: 'day'
}
}
},
plugins: {
tooltip: {
callbacks: {
label: function(context) {
const index = context.dataIndex;
return `Цена: ${prices[index]}, Сторона: ${sides[index]}`;
}
}
}
}
}
});

// Добавляем маркеры для сделок
chart.data.datasets.push({
label: 'Покупки',
data: tradesData.filter(trade => trade.side === 'buy').map(trade => ({
x: new Date(trade.opened_at),
y: trade.price
})),
backgroundColor: 'green',
pointStyle: 'triangle',
pointRadius: 8,
showLine: false
});

chart.data.datasets.push({
label: 'Продажи',
data: tradesData.filter(trade => trade.side === 'sell').map(trade => ({
x: new Date(trade.opened_at),
y: trade.price
})),
backgroundColor: 'red',
pointStyle: 'triangle',
pointRadius: 8,
rotation: 180,
showLine: false
});

chart.update();
});
</script>

# bot/tasks.py
from celery import shared_task
import logging
import time
from django.utils import timezone
from .models import Bot, Trade
from .connectors.binance_connector import BinanceConnector
from .algorithms.moving_average import MovingAverageStrategy
# Импортируйте другие стратегии по необходимости

logger = logging.getLogger('bot')

@shared_task
def start_bot_task(bot_id):
"""Задача Celery для запуска бота"""
try:
bot = Bot.objects.get(id=bot_id)
if bot.status != 'active':
logger.warning(f"Бот {bot.name} не активен, задача отменена")
return

logger.info(f"Запущен бот {bot.name} для пары {bot.pair.symbol}")

# Создаем коннектор к бирже
connector = BinanceConnector()

# Выбираем и инициализируем стратегию на основе типа
if bot.strategy.type == 'ma_cross':
params = bot.strategy.parameters
strategy = MovingAverageStrategy(
short_window=params.get('short_window', 10),
long_window=params.get('long_window', 50)
)
# Добавьте другие стратегии по аналогии

# Получаем данные для анализа
data = connector.get_market_data(bot.pair.symbol, timeframe='1h')
if data is None:
logger.error(f"Не удалось получить данные для {bot.pair.symbol}")
return

# Выполняем стратегию
result = strategy.execute_strategy(
connector=connector,
symbol=bot.pair.symbol,
amount=float(bot.amount_per_trade),
data=data
)

# Если была совершена сделка, сохраняем её
if result:
trade = Trade(
bot=bot,
order_id=result.get('id', ''),
symbol=bot.pair.symbol,
side=result.get('side', ''),
quantity=float(result.get('amount', 0)),
price=float(result.get('price', 0)),
commission=float(result.get('fee', {}).get('cost', 0)),
status='open' if result.get('status') != 'closed' else 'closed',
opened_at=timezone.now()
)
trade.save()
logger.info(f"Сохранена новая сделка: {trade}")

except Exception as e:
logger.error(f"Ошибка выполнения бота {bot_id}: {str(e)}")

@shared_task
def run_active_bots():
"""Периодическая задача для запуска всех активных ботов"""
active_bots = Bot.objects.filter(status='active')
logger.info(f"Запуск {active_bots.count()} активных ботов")

for bot in active_bots:
start_bot_task.delay(bot.id)

# bot/urls.py
from django.urls import path
from . import views

urlpatterns = [
path('dashboard/', views.dashboard, name='dashboard'),
path('bot/create/', views.create_bot, name='create_bot'),
path('bot/<int:bot_id>/', views.bot_detail, name='bot_detail'),
path('bot/<int:bot_id>/toggle/', views.toggle_bot, name='toggle_bot'),
]