Торговая автоматизация
Давайте разберем механику работы торговой автоматизации без иллюзий и выясним, какие типы ботов реально приносят доход на крипторынке, где кроются скрытые риски и как писать под них логику.
1. Архитектура стратегий: что работает на практике
На рынке выживают три основных типа автоматизированных систем. Остальное - это маркетинг от бирж для сбора комиссий.
- Сетка (Grid Trading): Идеальна для боковых диапазонов (флэта). Бот выставляет лестницу лимитных ордеров на покупку ниже текущей цены и на продажу выше. Зарабатывает на колебаниях внутри коридора.
- Усреднение стоимости (DCA-боты): Стратегия накопления. Бот покупает актив равными частями через фиксированные промежутки времени или при падении цены на заданный процент. Работает на споте для долгосрочного портфеля.
- Арбитражные связки: Ловля ценовых разниц между спотовыми рынками разных площадок или спреда между фьючерсом и спотом (funding rate arbitrage). Требует низкой задержки (low latency) и оптимизированного API.
2. Математика Grid-ботов: почему 80% пользователей теряют депозит
Большинство запускает сетку на хаях волатильного альткоина с настройками по умолчанию. Итог всегда одинаков: резкий пробой диапазона вниз, бот распродал всю базовую валюту по копеечным ценам, а портфель забит «мертвым» активом с просадкой в 60%.
Чтобы этого избежать, расчет параметров строится на жестких правилах:
Сравнение подходов к настройке сетки:
- Выбор актива: Ошибка новичка - хайповый мемкоин с дикой волатильностью. Профессиональный подход - монеты из топ-20 с высокой ликвидностью и сильными уровнями поддержки (BTC, ETH).
- Тип сетки: Ошибка новичка - арифметическая (равные шаги в долларах). Профессиональный подход — геометрическая (равные процентные шаги, критично для стабильного процента доходности).
- Диапазон: Ошибка новичка - узкий коридор на 5% от текущей цены. Профессиональный подход — широкий диапазон, перекрывающий исторический флэт за последние 3–6 месяцев.
3. Практическая реализация: код торгового модуля сетки
Ниже представлен рабочий фрагмент скрипта на Python с использованием библиотеки ccxt. Это ядро расчетной логики для генерации ордеров сетки с геометрическим шагом. В комментариях поясняю что и для чего.
"""
Планировщик сетки лимитных ордеров для Binance Spot (режим проверки).
Модель сетки
------------
Уровни L0 < L1 < ... < Ln образуют n ячеек [L_k, L_{k+1}].
В каждой ячейке одновременно живёт ровно один ордер:
* L_k < цена -> BUY на L_k. После исполнения выставляется SELL на L_{k+1}
на количество, полученное за вычетом комиссии в базовой валюте.
* L_k >= цена -> SELL на L_{k+1}. Для него нужен начальный запас базовой
валюты (его надо купить по рынку заранее).
Уровень, равный текущей цене, относится к SELL-ячейке (не теряется).
Скрипт НИЧЕГО не отправляет на биржу.
"""
from __future__ import annotations
import argparse
import logging
import sys
from dataclasses import dataclass
from decimal import (
ROUND_CEILING,
ROUND_FLOOR,
ROUND_HALF_UP,
Decimal,
InvalidOperation,
localcontext,
)
from typing import Any
import ccxt
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("grid_planner")
ZERO = Decimal("0")
ONE = Decimal("1")
# --------------------------------------------------------------------------
# Утилиты Decimal
# --------------------------------------------------------------------------
def dec(value: Any, name: str) -> Decimal:
try:
result = Decimal(str(value))
except (InvalidOperation, ValueError, TypeError) as exc:
raise ValueError(f"{name}: некорректное числовое значение") from exc
if not result.is_finite():
raise ValueError(f"{name}: значение должно быть конечным")
return result
def text(value: Decimal) -> str:
"""Decimal -> строка без экспоненты и лишних нулей."""
return format(value.normalize(), "f")
def floor_to_step(value: Decimal, step: Decimal) -> Decimal:
return (value / step).to_integral_value(rounding=ROUND_FLOOR) * step
def ceil_to_step(value: Decimal, step: Decimal) -> Decimal:
return (value / step).to_integral_value(rounding=ROUND_CEILING) * step
def round_to_step(value: Decimal, step: Decimal) -> Decimal:
return (value / step).to_integral_value(rounding=ROUND_HALF_UP) * step
# --------------------------------------------------------------------------
# Правила (фильтры) инструмента Binance Spot
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class SymbolRules:
tick: Decimal
step: Decimal
min_price: Decimal
max_price: Decimal | None
min_qty: Decimal
max_qty: Decimal | None
min_notional: Decimal
max_notional: Decimal | None
percent_general: tuple[Decimal, Decimal] | None # (down, up)
percent_by_side: dict[str, tuple[Decimal, Decimal]]
# Окно усреднения для процентных фильтров; 0 = используется последняя цена.
avg_price_mins: int | None
@property
def has_percent_filters(self) -> bool:
return self.percent_general is not None or bool(self.percent_by_side)
def _positive_or_none(value: Any, name: str) -> Decimal | None:
"""В фильтрах Binance 0 означает «ограничение отключено»."""
if value is None:
return None
result = dec(value, name)
return result if result > 0 else None
def load_rules(market: dict) -> SymbolRules:
"""Читает фильтры один раз. Любая проблема здесь — ошибка конфигурации."""
info = market.get("info") or {}
status = info.get("status")
if status is not None and status != "TRADING":
raise RuntimeError(f"Статус пары {status}, ожидается TRADING")
if info.get("isSpotTradingAllowed") is False:
raise RuntimeError("Spot-торговля по паре запрещена")
filters = {
item.get("filterType"): item
for item in info.get("filters", [])
if item.get("filterType")
}
price_filter = filters.get("PRICE_FILTER")
lot_filter = filters.get("LOT_SIZE")
if not price_filter:
raise RuntimeError("Binance PRICE_FILTER не найден")
if not lot_filter:
raise RuntimeError("Binance LOT_SIZE не найден")
tick = dec(price_filter["tickSize"], "PRICE_FILTER.tickSize")
step = dec(lot_filter["stepSize"], "LOT_SIZE.stepSize")
if tick <= 0 or step <= 0:
raise RuntimeError("tickSize и stepSize должны быть больше нуля")
min_notional = ZERO
max_notional = None
for filter_type in ("MIN_NOTIONAL", "NOTIONAL"):
item = filters.get(filter_type)
if not item:
continue
if item.get("minNotional") is not None:
min_notional = max(
min_notional,
dec(item["minNotional"], f"{filter_type}.minNotional"),
)
if item.get("maxNotional") is not None:
max_notional = _positive_or_none(
item["maxNotional"], f"{filter_type}.maxNotional"
)
percent_general = None
item = filters.get("PERCENT_PRICE")
if item:
percent_general = (
dec(item["multiplierDown"], "PERCENT_PRICE.multiplierDown"),
dec(item["multiplierUp"], "PERCENT_PRICE.multiplierUp"),
)
percent_by_side: dict[str, tuple[Decimal, Decimal]] = {}
item = filters.get("PERCENT_PRICE_BY_SIDE")
if item:
percent_by_side["buy"] = (
dec(item["bidMultiplierDown"], "bidMultiplierDown"),
dec(item["bidMultiplierUp"], "bidMultiplierUp"),
)
percent_by_side["sell"] = (
dec(item["askMultiplierDown"], "askMultiplierDown"),
dec(item["askMultiplierUp"], "askMultiplierUp"),
)
avg_mins_values = set()
for key in ("PERCENT_PRICE", "PERCENT_PRICE_BY_SIDE"):
item = filters.get(key)
if item and item.get("avgPriceMins") is not None:
avg_mins_values.add(int(dec(item["avgPriceMins"], "avgPriceMins")))
if len(avg_mins_values) > 1:
raise RuntimeError(
f"Разные avgPriceMins в фильтрах: {sorted(avg_mins_values)}"
)
avg_price_mins = avg_mins_values.pop() if avg_mins_values else None
return SymbolRules(
tick=tick,
step=step,
min_price=dec(price_filter.get("minPrice", "0"), "minPrice"),
max_price=_positive_or_none(price_filter.get("maxPrice"), "maxPrice"),
min_qty=dec(lot_filter.get("minQty", "0"), "LOT_SIZE.minQty"),
max_qty=_positive_or_none(lot_filter.get("maxQty"), "LOT_SIZE.maxQty"),
min_notional=min_notional,
max_notional=max_notional,
percent_general=percent_general,
percent_by_side=percent_by_side,
avg_price_mins=avg_price_mins,
)
# --------------------------------------------------------------------------
# Сетка
# --------------------------------------------------------------------------
def align_bounds(
lower: Decimal, upper: Decimal, tick: Decimal
) -> tuple[Decimal, Decimal]:
"""Границы выравниваются внутрь диапазона."""
lower = ceil_to_step(lower, tick)
upper = floor_to_step(upper, tick)
if lower >= upper:
raise ValueError("После выравнивания по tickSize диапазон пуст")
return lower, upper
def calculate_grid_levels(
lower_bound: Decimal,
upper_bound: Decimal,
grid_count: int,
tick: Decimal,
) -> list[Decimal]:
"""Геометрическая сетка: grid_count интервалов, grid_count + 1 уровней."""
if not isinstance(grid_count, int) or isinstance(grid_count, bool):
raise ValueError("grid_count должен быть целым числом")
if grid_count < 1:
raise ValueError("grid_count должен быть не меньше 1")
if min(lower_bound, upper_bound, tick) <= 0:
raise ValueError("Границы и tickSize должны быть больше нуля")
if lower_bound >= upper_bound:
raise ValueError("lower_bound должен быть меньше upper_bound")
lower, upper = align_bounds(lower_bound, upper_bound, tick)
with localcontext() as ctx:
ctx.prec = 50
ratio = (upper / lower) ** (ONE / Decimal(grid_count))
# Каждый уровень считается от нижней границы: ошибка не накапливается.
levels = [
round_to_step(lower * (ratio ** i), tick)
for i in range(grid_count + 1)
]
levels[0] = lower
levels[-1] = upper
if any(a >= b for a, b in zip(levels, levels[1:])):
raise ValueError(
"Уровни совпали или нарушили порядок после округления к тику. "
"Уменьшите grid_count или расширьте диапазон."
)
return levels
def check_grid_economics(
levels: list[Decimal],
fee_rate: Decimal,
extra_cost_rate: Decimal,
) -> None:
"""
Приближённый фильтр: чистая доходность ячейки с учётом комиссии
на обеих ногах должна превышать запас расходов.
Это не гарантия прибыльности.
"""
if fee_rate < 0 or extra_cost_rate < 0:
raise ValueError("Комиссия и запас расходов не могут быть < 0")
keep = (ONE - fee_rate) ** 2
for low, high in zip(levels, levels[1:]):
net_return = (high / low) * keep - ONE
if net_return <= extra_cost_rate:
raise ValueError(
f"Шаг сетки слишком мал: {text(low)} -> {text(high)}; "
f"чистая доходность {net_return:.4%}, "
f"нужно больше {extra_cost_rate:.4%}"
)
# --------------------------------------------------------------------------
# Проверка ордера по фильтрам биржи
# --------------------------------------------------------------------------
def validate_order(
rules: SymbolRules,
side: str,
price: Decimal,
amount: Decimal,
avg_price: Decimal | None = None,
) -> Decimal:
"""Возвращает стоимость ордера или бросает ValueError."""
if side not in ("buy", "sell"):
raise ValueError("side должен быть buy или sell")
if price <= 0 or amount <= 0:
raise ValueError(
f"{side.upper()} @ {text(price)}: цена и количество должны быть > 0 "
f"(amount={text(amount)})"
)
if price % rules.tick != 0:
raise ValueError(f"Цена {text(price)} не кратна tickSize")
if amount % rules.step != 0:
raise ValueError(f"Количество {text(amount)} не кратно stepSize")
if price < rules.min_price:
raise ValueError(f"Цена {text(price)} < minPrice {text(rules.min_price)}")
if rules.max_price is not None and price > rules.max_price:
raise ValueError(f"Цена {text(price)} > maxPrice {text(rules.max_price)}")
if amount < rules.min_qty:
raise ValueError(
f"Количество {text(amount)} < minQty {text(rules.min_qty)}"
)
if rules.max_qty is not None and amount > rules.max_qty:
raise ValueError(
f"Количество {text(amount)} > maxQty {text(rules.max_qty)}"
)
cost = price * amount
if cost < rules.min_notional:
raise ValueError(
f"Стоимость {text(cost)} < minNotional {text(rules.min_notional)}"
)
if rules.max_notional is not None and cost > rules.max_notional:
raise ValueError(
f"Стоимость {text(cost)} > maxNotional {text(rules.max_notional)}"
)
if avg_price is not None:
bounds = []
if rules.percent_general:
bounds.append(("PERCENT_PRICE", *rules.percent_general))
if side in rules.percent_by_side:
bounds.append(("PERCENT_PRICE_BY_SIDE", *rules.percent_by_side[side]))
for name, down, up in bounds:
minimum, maximum = avg_price * down, avg_price * up
if not minimum <= price <= maximum:
raise ValueError(
f"{name}: цена {text(price)} вне диапазона "
f"{text(minimum)}–{text(maximum)}"
)
return cost
# --------------------------------------------------------------------------
# План сетки
# --------------------------------------------------------------------------
def build_grid_plan(
levels: list[Decimal],
current_price: Decimal,
rules: SymbolRules,
budget_per_level: Decimal,
fee_rate: Decimal,
fee_in_base: bool,
avg_price: Decimal | None,
) -> tuple[list[dict], list[str]]:
"""
Строит ордера по ячейкам. Возвращает (ордера, пропущенные ячейки).
Комиссия в базовой валюте уменьшает количество парного SELL.
"""
if budget_per_level <= 0:
raise ValueError("Бюджет на уровень должен быть больше нуля")
base_fee = fee_rate if fee_in_base else ZERO
orders: list[dict] = []
skipped: list[str] = []
for k, (low, high) in enumerate(zip(levels, levels[1:])):
try:
if low < current_price:
# BUY-ячейка: покупаем на low, продаём на high.
amount = floor_to_step(budget_per_level / low, rules.step)
cost = validate_order(rules, "buy", low, amount, avg_price)
sell_amount = floor_to_step(amount * (ONE - base_fee), rules.step)
# Парный SELL выставится позже, процентный фильтр тогда
# проверит сама биржа, поэтому avg_price здесь не передаём.
validate_order(rules, "sell", high, sell_amount)
orders.append({
"cell": k,
"side": "buy",
"price": low,
"amount": amount,
"notional": cost,
"on_fill_sell_price": high,
"on_fill_sell_amount": sell_amount,
})
else:
# SELL-ячейка: количество считается от цены самого SELL,
# чтобы номинал не превышал бюджет на уровень.
amount = floor_to_step(budget_per_level / high, rules.step)
sell_cost = validate_order(
rules, "sell", high, amount, avg_price
)
orders.append({
"cell": k,
"side": "sell",
"price": high,
"amount": amount,
"notional": sell_cost,
})
except ValueError as exc:
skipped.append(f"ячейка {k} [{text(low)}–{text(high)}]: {exc}")
return orders, skipped
def summarize_plan(
orders: list[dict],
fee_rate: Decimal,
taker_fee_rate: Decimal,
fee_in_base: bool,
ask: Decimal,
slippage: Decimal,
bnb_price: Decimal | None,
) -> dict:
"""
Оценка потребности в средствах. Это ОЦЕНКА с запасом на проскальзывание,
а не гарантия: реальная цена и комиссии при начальной закупке могут
отличаться.
"""
buys = [o for o in orders if o["side"] == "buy"]
sells = [o for o in orders if o["side"] == "sell"]
# Покупка BUY-ордера списывает не больше notional: комиссия берётся
# из получаемого актива (BTC) или из BNB, но не из USDT.
quote_for_buys = sum((o["notional"] for o in buys), ZERO)
base_net_needed = sum((o["amount"] for o in sells), ZERO)
# Начальная закупка по рынку (taker). При комиссии в базе покупать
# нужно больше, чтобы после комиссии осталось base_net_needed.
taker_base_fee = taker_fee_rate if fee_in_base else ZERO
base_gross = base_net_needed / (ONE - taker_base_fee)
initial_notional = base_gross * ask * (ONE + slippage)
bnb_fee_quote = ZERO
bnb_needed = None
if not fee_in_base:
# Лимитные ордера (BUY, парные и начальные SELL) считаются по maker,
# начальная рыночная закупка — по taker. Лимитный ордер, который
# сразу пересекает стакан, исполнится как taker: оценка это не ловит.
buy_leg = sum((o["notional"] for o in buys), ZERO)
paired_sell_leg = sum(
(o["on_fill_sell_price"] * o["on_fill_sell_amount"] for o in buys),
ZERO,
)
sell_leg = sum((o["notional"] for o in sells), ZERO)
bnb_fee_quote = (
(buy_leg + paired_sell_leg + sell_leg) * fee_rate
+ initial_notional * taker_fee_rate
)
if bnb_price is not None:
bnb_needed = bnb_fee_quote / bnb_price
return {
"buy_orders": len(buys),
"sell_orders": len(sells),
"quote_for_buys": quote_for_buys,
"base_needed_for_sells": base_net_needed,
"initial_buy_cost_estimate": initial_notional,
"total_quote_estimate": quote_for_buys + initial_notional,
"bnb_fee_quote_equivalent": bnb_fee_quote,
"bnb_needed": bnb_needed,
}
# --------------------------------------------------------------------------
# Работа с биржей
# --------------------------------------------------------------------------
def restrict_to_spot(exchange) -> None:
"""Просим ccxt грузить только spot-рынки (формат опции зависит от версии)."""
fetch_markets = exchange.options.get("fetchMarkets")
if isinstance(fetch_markets, list):
exchange.options["fetchMarkets"] = ["spot"]
elif isinstance(fetch_markets, dict) and "types" in fetch_markets:
exchange.options["fetchMarkets"] = {**fetch_markets, "types": ["spot"]}
def get_ticker_prices(
exchange, symbol: str
) -> tuple[Decimal, Decimal, Decimal | None]:
"""Возвращает (bid, ask, last)."""
ticker = exchange.fetch_ticker(symbol)
if ticker.get("bid") is None or ticker.get("ask") is None:
raise RuntimeError(f"{symbol}: биржа не вернула bid/ask")
bid = dec(ticker["bid"], "bid")
ask = dec(ticker["ask"], "ask")
if bid <= 0 or ask <= 0 or bid > ask:
raise RuntimeError(f"{symbol}: некорректные bid/ask {bid}/{ask}")
last = ticker.get("last")
return bid, ask, (dec(last, "last") if last is not None else None)
def fetch_percent_reference_price(
exchange, market: dict, rules: SymbolRules, last: Decimal | None
) -> Decimal:
"""
Опорная цена для PERCENT_PRICE*. Воспроизводит правило Binance:
avgPriceMins == 0 -> последняя цена, иначе средняя за avgPriceMins минут.
Окно, которое вернул /avgPrice, сверяется с окном из фильтра.
Итоговую проверку всё равно делает биржа при отправке ордера.
"""
if rules.avg_price_mins == 0:
if last is None:
raise RuntimeError("Для фильтра нужна last-цена, биржа её не вернула")
return last
response = exchange.publicGetAvgPrice({"symbol": market["id"]})
price = dec(response["price"], "avgPrice")
if price <= 0:
raise RuntimeError("Binance вернула некорректную avgPrice")
mins = response.get("mins")
if (
rules.avg_price_mins is not None
and mins is not None
and int(mins) != rules.avg_price_mins
):
raise RuntimeError(
f"Окно avgPrice ({mins} мин) не совпадает с avgPriceMins "
f"фильтра ({rules.avg_price_mins} мин): проверку не воспроизвести"
)
return price
def resolve_rate(
market: dict,
key: str,
explicit: Decimal | None,
fee_in_bnb: bool,
) -> Decimal:
"""
Ставка комиссии: явная, иначе из рынка (maker/taker), иначе 0.1%.
Скидка за оплату в BNB НЕ предполагается: фактическую ставку аккаунта
нужно передать явно, иначе оценка консервативна (без скидки).
"""
if explicit is not None:
return explicit
value = market.get(key)
rate = dec(value, key) if value is not None else Decimal("0.001")
if fee_in_bnb:
logger.warning(
"Ставка %s не задана явно: используется %s без скидки BNB. "
"Передайте реальную ставку аккаунта через "
"--fee-rate / --taker-fee-rate.",
key, rate,
)
return rate
def paired_sell_amount(
executed_qty: Decimal,
fills: list[dict],
base_asset: str,
free_balance: Decimal,
price: Decimal,
rules: SymbolRules,
) -> Decimal:
"""
Для будущего модуля исполнения: количество парного SELL после
реального исполнения BUY.
executed_qty — ВАЛОВОЕ купленное количество (executedQty ордера,
до вычета комиссии).
fills — исполнения ордера в формате ccxt: у каждого
fill["fee"] = {"cost": ..., "currency": ...} или None.
Вычитаются только комиссии, взятые в базовой валюте: комиссия в BNB
или в котируемой валюте количество актива не уменьшает.
free_balance — свободный баланс базовой валюты, не зарезервированный
под другие ордера (за это отвечает вызывающий код).
"""
fee_in_base = ZERO
for fill in fills:
fee = fill.get("fee")
if (
fee
and fee.get("currency") == base_asset
and fee.get("cost") is not None
):
fee_in_base += dec(fee["cost"], "fee.cost")
net = executed_qty - fee_in_base
if net <= 0:
raise ValueError("После вычета комиссии количество не положительно")
amount = floor_to_step(min(net, free_balance), rules.step)
validate_order(rules, "sell", price, amount)
return amount
# --------------------------------------------------------------------------
# main
# --------------------------------------------------------------------------
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
parser.add_argument("--symbol", default="BTC/USDT")
parser.add_argument("--lower", default="50000")
parser.add_argument("--upper", default="70000")
parser.add_argument("--grid-count", type=int, default=20)
parser.add_argument(
"--budget-per-level",
default="100",
help="Максимальный номинал одного ордера в котируемой валюте",
)
parser.add_argument(
"--fee-rate", default=None,
help="Maker-комиссия (доля), по умолчанию maker из рынка",
)
parser.add_argument(
"--taker-fee-rate", default=None,
help="Taker-комиссия для начальной закупки, по умолчанию taker из рынка",
)
parser.add_argument(
"--fee-in-bnb", action="store_true",
help="Комиссия оплачивается в BNB, а не в получаемом активе",
)
parser.add_argument("--extra-cost-rate", default="0.0005")
parser.add_argument(
"--slippage", default="0.002",
help="Запас на проскальзывание начальной закупки (доля)",
)
return parser.parse_args(argv)
def optional_rate(value: str | None, name: str) -> Decimal | None:
if value is None:
return None
result = dec(value, name)
if not ZERO <= result < ONE:
raise ValueError(f"{name} должен быть в диапазоне [0, 1)")
return result
def run(args: argparse.Namespace) -> None:
symbol = args.symbol
lower_bound = dec(args.lower, "lower")
upper_bound = dec(args.upper, "upper")
budget = dec(args.budget_per_level, "budget_per_level")
extra_cost_rate = dec(args.extra_cost_rate, "extra_cost_rate")
slippage = optional_rate(args.slippage, "slippage")
explicit_fee = optional_rate(args.fee_rate, "fee_rate")
explicit_taker = optional_rate(args.taker_fee_rate, "taker_fee_rate")
exchange = ccxt.binance({
"enableRateLimit": True,
"timeout": 15000,
"options": {"defaultType": "spot"},
})
def money(value: Decimal) -> str:
return text(value.quantize(Decimal("0.01")))
try:
restrict_to_spot(exchange)
exchange.load_markets()
if symbol not in exchange.markets:
raise ValueError(f"Торговая пара не найдена: {symbol}")
market = exchange.market(symbol)
if market.get("spot") is False:
raise RuntimeError(f"{symbol} не является spot-рынком")
if market.get("active") is False:
raise RuntimeError(f"Торговая пара неактивна: {symbol}")
rules = load_rules(market)
fee_in_base = not args.fee_in_bnb
fee_rate = resolve_rate(market, "maker", explicit_fee, args.fee_in_bnb)
taker_rate = resolve_rate(
market, "taker", explicit_taker, args.fee_in_bnb
)
bid, ask, last = get_ticker_prices(exchange, symbol)
mid = (bid + ask) / Decimal("2")
levels = calculate_grid_levels(
lower_bound, upper_bound, args.grid_count, rules.tick
)
if not levels[0] <= mid <= levels[-1]:
raise ValueError(
f"Текущая цена {text(mid)} вне диапазона сетки "
f"{text(levels[0])}–{text(levels[-1])}"
)
check_grid_economics(levels, fee_rate, extra_cost_rate)
avg_price = (
fetch_percent_reference_price(exchange, market, rules, last)
if rules.has_percent_filters
else None
)
orders, skipped = build_grid_plan(
levels=levels,
current_price=mid,
rules=rules,
budget_per_level=budget,
fee_rate=fee_rate,
fee_in_base=fee_in_base,
avg_price=avg_price,
)
for message in skipped:
logger.warning("Пропущено: %s", message)
if not orders:
raise RuntimeError("Ни один ордер не прошёл проверку")
bnb_price = None
if args.fee_in_bnb:
bnb_symbol = f"BNB/{market['quote']}"
if bnb_symbol in exchange.markets:
b, a, _ = get_ticker_prices(exchange, bnb_symbol)
bnb_price = (b + a) / Decimal("2")
else:
logger.warning("Пара %s не найдена: расход BNB не оценить", bnb_symbol)
summary = summarize_plan(
orders, fee_rate, taker_rate, fee_in_base, ask, slippage, bnb_price
)
print(f"Пара: {symbol}")
print(f"Bid / Ask: {text(bid)} / {text(ask)} (mid {text(mid)})")
print(f"Tick / Step: {text(rules.tick)} / {text(rules.step)}")
print(f"Сетка: {text(levels[0])} – {text(levels[-1])}, "
f"ячеек: {len(levels) - 1}")
print(f"Комиссия: maker {fee_rate:.4%}, taker {taker_rate:.4%} "
f"({'в получаемом активе' if fee_in_base else 'в BNB'})")
if avg_price is not None:
print(f"Опорная цена фильтров: {text(avg_price)} "
f"(avgPriceMins={rules.avg_price_mins})")
print()
for order in orders:
line = (
f"{order['side'].upper():4} {text(order['amount'])} "
f"@ {text(order['price'])} (~{money(order['notional'])})"
)
if order["side"] == "buy":
line += (
f" -> после исполнения SELL "
f"{text(order['on_fill_sell_amount'])} "
f"@ {text(order['on_fill_sell_price'])}"
)
print(line)
quote, base = market["quote"], market["base"]
print()
print(f"BUY-ордеров: {summary['buy_orders']}, "
f"нужно ~{money(summary['quote_for_buys'])} {quote}")
print(f"SELL-ордеров: {summary['sell_orders']}, "
f"нужен запас {text(summary['base_needed_for_sells'])} {base}")
print(f"Начальная закупка ~{money(summary['initial_buy_cost_estimate'])} "
f"{quote} (по ask, taker-комиссия, проскальзывание {slippage:.2%})")
print(f"Итого ~{money(summary['total_quote_estimate'])} {quote} "
"(оценка, не гарантия достаточного баланса)")
if not fee_in_base:
line = (f"Комиссии в BNB: ~{money(summary['bnb_fee_quote_equivalent'])} "
f"{quote} в эквиваленте")
if summary["bnb_needed"] is not None:
line += f", ~{summary['bnb_needed']:.4f} BNB"
print(line + " (полный цикл всех ордеров)")
print()
print("Режим проверки: ордера на биржу НЕ отправлялись.")
finally:
exchange.close()
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
try:
run(args)
except ccxt.BaseError as exc:
logger.error("Ошибка биржи/сети: %s", exc)
return 1
except (ValueError, RuntimeError) as exc:
logger.error("Ошибка расчёта: %s", exc)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())Сохраните и назовите файл, например, grid_planner.py. Запуск: python grid_planner.py --grid-count 20 --budget-per-level 100 --fee-in-bnb --fee-rate 0.00075 --taker-fee-rate 0.00075
4. Скрытые риски и ловушки автоматизации, о которых молчат биржи
Биржевые маркетологи любят показывать красивые графики доходности ботов на историческом бэктесте. В реальности алгоритмическая торговля сталкивается с техническими барьерами, способными сжечь депозит быстрее человеческой ошибки.
- Проскальзывание (Slippage) при высокой волатильности: Когда рынок пробивает уровень поддержки, стакан пустеет за миллисекунды. Бот пытается выставить или исполнить рыночный ордер, но сделка цепляется по худшей цене. На тонких альткоинах проскальзывание может сожрать недельную прибыль от сетки за один импульс.
- Токсичность комиссий (Maker vs Taker): Сетка построена на лимитных ордерах — это режим Maker, где биржи часто платят вам ребейты или берут минимальный процент. Но если логика бота срывается в экстренную распродажу по маркету (Stop-Loss или паника при пробое), включаются конские комиссии Taker, которые убивают математику стратегии.
- Иллюзия бэктестов: Исторические данные не учитывают реальную задержку сети (ping до API биржи) и очереди сопоставления ордеров (matching engine). Бот, показавший +300% на исторических тестах 2024–2025 годов, в реальном бою на флэте может уйти в минус из-за задержек исполнения.
5. Как защитить капитал: чек-лист риск-менеджмента для ботов
Нельзя запускать алгоритм и уезжать в отпуск. Автоматизация требует жесткой инфраструктуры мониторинга:
Правила безопасности и контроля:
- Жесткий Kill-Switch: В коде или настройках терминала должен быть прописан абсолютный стоп-лосс на весь депозит бота. Если общая провальная стоимость базовой и котируемой валюты падает на условные 15%, бот снимает все лимитники и фиксирует убыток.
- Изоляция депозита: Никогда не держите торговый капитал бота на одном субсчете с основными долгосрочными сбережениями. Используйте изолированные API-ключи с отключенной функцией выводов средств — разрешена только торговля.
- Мониторинг через Telegram-алерты: Настройте отправку критических уведомлений не в интерфейс биржи, а напрямую в мессенджер. Каждая отмененная сессия, ошибка API или исполненный экстренный ордер должны фиксироваться мгновенно.
Ну и подведем итог.
Торговый бот - это скальпель в руках хирурга, а не автопилот в самолете. Он автоматизирует дисциплину, но не заменяет мозги. Начинайте с малых сумм на спотовых сетках по ликвидным инструментам, тестируйте логику на реальной задержке и помните: главный враг профита не рынок, а ваши собственные настройки без учета риск-менеджмента.
Академия EXMON напоминает: надежная инфраструктура и контроль рисков всегда первичны. Зарабатывает тот, кто умеет вовремя останавливать алгоритм.
