Почему «написать бота» ≠ «написать рабочего бота»
Первый торговый бот многих разработчиков — это 300 строк в одном файле: получить цену, посчитать сигнал, отправить ордер. Он работает 15 минут на бумажном трейдинге, а потом ломается на первом же исключении от API, не восстанавливает позицию после перезапуска и теряет деньги из-за дублирования ордеров.
Production-бот — это иначе. Покажем, как его правильно структурировать.
Архитектура: 5 слоёв
Хороший бот состоит из пяти независимых компонентов, каждый отвечает за свою зону ответственности:
- Data Handler — получение и нормализация рыночных данных
- Strategy Engine — логика генерации сигналов
- Risk Manager — проверка лимитов и размера позиции
- Order Executor — отправка и отслеживание ордеров
- Logger / State Store — запись событий и восстановление состояния
1. Data Handler
Отвечает за соединение с биржей и доставку «чистых» данных остальным компонентам. Ничего не знает о стратегии.
from dataclasses import dataclass
from typing import Callable
import asyncio, json, websockets
@dataclass
class Tick:
symbol: str
price: float
volume: float
ts: int # unix ms
class DataHandler:
def __init__(self, symbols: list[str], on_tick: Callable):
self.symbols = symbols
self.on_tick = on_tick
async def _run(self):
streams = "/".join(f"{s.lower()}@trade" for s in self.symbols)
url = f"wss://stream.binance.com:9443/stream?streams={streams}"
async with websockets.connect(url, ping_interval=20) as ws:
async for raw in ws:
d = json.loads(raw)["data"]
tick = Tick(d["s"], float(d["p"]), float(d["q"]), d["T"])
await self.on_tick(tick)
async def run(self):
while True:
try:
await self._run()
except Exception as e:
print(f"DataHandler error: {e}, reconnecting in 5s")
await asyncio.sleep(5)
2. Strategy Engine
Получает нормализованные тики, считает индикаторы и генерирует сигналы. Не отправляет ордера сам — только возвращает решение.
from enum import Enum
class Signal(Enum):
BUY = "buy"
SELL = "sell"
HOLD = "hold"
class MomentumStrategy:
def __init__(self, window: int = 20):
self.prices: list[float] = []
self.window = window
def on_tick(self, tick: Tick) -> Signal:
self.prices.append(tick.price)
if len(self.prices) > self.window:
self.prices.pop(0)
if len(self.prices) < self.window:
return Signal.HOLD
avg = sum(self.prices) / self.window
if tick.price > avg * 1.005: # +0.5% выше MA
return Signal.BUY
if tick.price < avg * 0.995: # -0.5% ниже MA
return Signal.SELL
return Signal.HOLD
3. Risk Manager
Самый важный компонент. Проверяет, можно ли вообще открывать позицию, и сколько.
class RiskManager:
def __init__(self, max_position_usd: float, max_daily_loss_usd: float):
self.max_pos = max_position_usd
self.max_loss = max_daily_loss_usd
self.daily_pnl = 0.0
self.open_positions: dict = {}
def can_open(self, symbol: str, side: str, price: float, qty: float) -> bool:
cost = price * qty
if cost > self.max_pos:
return False
if self.daily_pnl < -self.max_loss:
return False # дневной стоп-лосс
if symbol in self.open_positions:
return False # уже есть позиция
return True
def calc_qty(self, balance: float, price: float, risk_pct: float = 0.02) -> float:
# Ставим не более 2% депозита на сделку
budget = min(balance * risk_pct, self.max_pos)
return round(budget / price, 6)
4. Order Executor
Отправляет ордера, отслеживает статус и обрабатывает ошибки API.
import hmac, hashlib, time
import aiohttp
class OrderExecutor:
BASE = "https://api.binance.com"
def __init__(self, api_key: str, secret: str):
self.key = api_key
self.secret = secret.encode()
def _sign(self, params: str) -> str:
return hmac.new(self.secret, params.encode(), hashlib.sha256).hexdigest()
async def market_buy(self, symbol: str, qty: float) -> dict:
ts = int(time.time() * 1000)
params = f"symbol={symbol}&side=BUY&type=MARKET&quantity={qty}×tamp={ts}"
sig = self._sign(params)
url = f"{self.BASE}/api/v3/order?{params}&signature={sig}"
headers = {"X-MBX-APIKEY": self.key}
async with aiohttp.ClientSession() as s:
async with s.post(url, headers=headers) as r:
data = await r.json()
if r.status != 200:
raise RuntimeError(f"Order error: {data}")
return data
.env) или секреты вашей платформы. Никогда не включайте разрешение «Вывод средств» в ключе.
5. Logger и хранение состояния
Бот должен уметь восстанавливать состояние после перезапуска. Минимум — записывать все ордера и позиции в файл или БД.
import json, pathlib, datetime
class StateStore:
def __init__(self, path: str = "state.json"):
self.path = pathlib.Path(path)
self.data = json.loads(self.path.read_text()) if self.path.exists() else {}
def save_order(self, order: dict):
oid = str(order["orderId"])
self.data[oid] = {**order, "saved_at": datetime.datetime.utcnow().isoformat()}
self.path.write_text(json.dumps(self.data, indent=2))
def open_positions(self) -> list:
return [v for v in self.data.values() if v.get("status") == "FILLED"]
Собираем всё вместе
import asyncio
async def main():
store = StateStore()
risk = RiskManager(max_position_usd=500, max_daily_loss_usd=100)
strategy = MomentumStrategy(window=20)
executor = OrderExecutor(
api_key=os.environ["BINANCE_KEY"],
secret=os.environ["BINANCE_SECRET"]
)
async def on_tick(tick: Tick):
signal = strategy.on_tick(tick)
if signal == Signal.BUY:
qty = risk.calc_qty(balance=1000, price=tick.price)
if risk.can_open(tick.symbol, "buy", tick.price, qty):
order = await executor.market_buy(tick.symbol, qty)
store.save_order(order)
handler = DataHandler(["BTCUSDT", "ETHUSDT"], on_tick)
await handler.run()
asyncio.run(main())
Типичные ошибки новичков
- Нет идемпотентности ордеров. При перезапуске бот отправляет ордер повторно. Используйте
newClientOrderIdс уникальным UUID — биржа отклонит дубликат. - Нет обработки частичного исполнения. Ордер может заполниться на 60%. Бот думает, что позиции нет, и открывает новую.
- Стратегия и исполнение в одной функции. Тяжело тестировать и дебажить.
- Нет дневного стоп-лосса. При серии убыточных сделок бот продолжает торговать, пока не сольёт депозит.
- Синхронный код на I/O. Блокирующий
requests.getв async-контексте замораживает весь event loop.
Итог
Торговый бот — это не скрипт, а набор компонентов с чёткими зонами ответственности. Чем более изолированы слои, тем проще их тестировать и улучшать по отдельности. Риск-менеджер — не опция, а обязательный элемент любой системы на реальные деньги.
Если нужен готовый, надёжный бот под конкретную стратегию — напишите нам. Работаем под NDA, архитектуру согласовываем с вами до старта.