Trading Automation
Let's break down the actual mechanics of trading automation—no fluff, no illusions. We'll look at what types of bots actually generate returns in crypto, where the hidden risks lurk, and how to structure the core logic behind them.
1. Strategy Architecture: What Actually Works in Practice
Only three main types of automated systems actually survive in the wild. Everything else is just exchange marketing designed to farm trading fees.
- Grid Trading: Perfect for sideways markets (ranging price action). The bot places a ladder of limit buy orders below the current price and limit sell orders above it, profiting from oscillations within the established channel.
- Dollar-Cost Averaging (DCA Bots): A pure accumulation strategy. The bot buys equal portions of an asset at fixed time intervals or when the price drops by a specific percentage. Ideal for spot markets and long-term portfolio building.
- Arbitrage Strategies: Capitalizing on price discrepancies between spot markets across different exchanges, or capturing the spread between futures and spot (funding rate arbitrage). Requires low latency and highly optimized API integrations.
2. The Math Behind Grid Bots: Why 80% of Users Blow Their Accounts
Most people spin up a grid bot at the absolute local top of a volatile altcoin using default settings. The outcome is always the same: a sharp breakdown out of the range, the bot dumps all the base asset at rock-bottom prices, and the portfolio is left holding a heavy bag sitting at a 60% drawdown.
To avoid this trap, parameter calculation must follow strict rules:
Comparing Grid Configuration Approaches:
- Asset Selection: Rookie mistake—chasing a hyped-up, hyper-volatile memecoin. Pro approach—sticking to top-20 assets with deep liquidity and strong technical support levels (like BTC or ETH).
- Grid Type: Rookie mistake—using arithmetic grids (equal dollar spacing). Pro approach—using geometric grids (equal percentage spacing, which is critical for consistent yield percentages).
- Price Range: Rookie mistake—setting a tight corridor just 5% away from the current price. Pro approach—setting a wide range that covers historical consolidation over the past 3 to 6 months.
3. Practical Implementation: Grid Trading Module Code
Below is a working Python script snippet utilizing the ccxt library. This is the core calculation engine for generating geometric grid orders. Check the inline comments for breakdown details.
"""
Limit order grid planner for Binance Spot (dry-run mode).
Grid Model
----------
Levels L0 < L1 < ... < Ln form n cells [L_k, L_{k+1}].
Exactly one active order lives in each cell at any time:
- L_k < price -> BUY at L_k. Once filled, a SELL is placed at L_{k+1}
for the amount received minus the base currency fee.
- L_k >= price -> SELL at L_{k+1}. Requires an initial base currency
reserve (needs to be bought upfront via market order).
The level equal to the current price belongs to the SELL cell (not dropped).
The script sends NOTHING to the exchange.
"""
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 Utilities
# --------------------------------------------------------------------------
def dec(value: Any, name: str) -> Decimal:
try:
result = Decimal(str(value))
except (InvalidOperation, ValueError, TypeError) as exc:
raise ValueError(f"{name}: invalid numeric value") from exc
if not result.is_finite():
raise ValueError(f"{name}: value must be finite")
return result
def text(value: Decimal) -> str:
"""Decimal -> string without exponent and trailing zeros."""
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 Symbol Rules (Filters)
# --------------------------------------------------------------------------
@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]]
# Averaging window for percentage filters; 0 = uses last price.
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:
"""In Binance filters, 0 means 'restriction disabled'."""
if value is None:
return None
result = dec(value, name)
return result if result > 0 else None
def load_rules(market: dict) -> SymbolRules:
"""Reads filters once. Any issue here is a configuration error."""
info = market.get("info") or {}
status = info.get("status")
if status is not None and status != "TRADING":
raise RuntimeError(f"Pair status is {status}, expected TRADING")
if info.get("isSpotTradingAllowed") is False:
raise RuntimeError("Spot trading is disabled for this pair")
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 not found")
if not lot_filter:
raise RuntimeError("Binance LOT_SIZE not found")
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 and stepSize must be greater than zero")
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"Conflicting avgPriceMins across filters: {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,
)
# --------------------------------------------------------------------------
# Grid Generation
# --------------------------------------------------------------------------
def align_bounds(
lower: Decimal, upper: Decimal, tick: Decimal
) -> tuple[Decimal, Decimal]:
"""Bounds are aligned inward."""
lower = ceil_to_step(lower, tick)
upper = floor_to_step(upper, tick)
if lower >= upper:
raise ValueError("Range is empty after tickSize alignment")
return lower, upper
def calculate_grid_levels(
lower_bound: Decimal,
upper_bound: Decimal,
grid_count: int,
tick: Decimal,
) -> list[Decimal]:
"""Geometric grid: grid_count intervals, grid_count + 1 levels."""
if not isinstance(grid_count, int) or isinstance(grid_count, bool):
raise ValueError("grid_count must be an integer")
if grid_count < 1:
raise ValueError("grid_count must be at least 1")
if min(lower_bound, upper_bound, tick) <= 0:
raise ValueError("Bounds and tickSize must be greater than zero")
if lower_bound >= upper_bound:
raise ValueError("lower_bound must be less than 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))
# Each level is calculated from the lower bound: prevents error accumulation.
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(
"Levels overlapped or broke order after tick rounding. "
"Decrease grid_count or widen the range."
)
return levels
def check_grid_economics(
levels: list[Decimal],
fee_rate: Decimal,
extra_cost_rate: Decimal,
) -> None:
"""
Approximate check: net cell yield factoring in fees on both legs
must exceed the expense buffer. This is not a guarantee of profitability.
"""
if fee_rate < 0 or extra_cost_rate < 0:
raise ValueError("Fee and extra cost rates cannot be < 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"Grid step too tight: {text(low)} -> {text(high)}; "
f"net yield {net_return:.4%}, "
f"required > {extra_cost_rate:.4%}"
)
# --------------------------------------------------------------------------
# Exchange Order Validation
# --------------------------------------------------------------------------
def validate_order(
rules: SymbolRules,
side: str,
price: Decimal,
amount: Decimal,
avg_price: Decimal | None = None,
) -> Decimal:
"""Returns order notional cost or raises ValueError."""
if side not in ("buy", "sell"):
raise ValueError("side must be buy or sell")
if price <= 0 or amount <= 0:
raise ValueError(
f"{side.upper()} @ {text(price)}: price and amount must be > 0 "
f"(amount={text(amount)})"
)
if price % rules.tick != 0:
raise ValueError(f"Price {text(price)} is not a multiple of tickSize")
if amount % rules.step != 0:
raise ValueError(f"Amount {text(amount)} is not a multiple of stepSize")
if price < rules.min_price:
raise ValueError(f"Price {text(price)} < minPrice {text(rules.min_price)}")
if rules.max_price is not None and price > rules.max_price:
raise ValueError(f"Price {text(price)} > maxPrice {text(rules.max_price)}")
if amount < rules.min_qty:
raise ValueError(
f"Amount {text(amount)} < minQty {text(rules.min_qty)}"
)
if rules.max_qty is not None and amount > rules.max_qty:
raise ValueError(
f"Amount {text(amount)} > maxQty {text(rules.max_qty)}"
)
cost = price * amount
if cost < rules.min_notional:
raise ValueError(
f"Notional {text(cost)} < minNotional {text(rules.min_notional)}"
)
if rules.max_notional is not None and cost > rules.max_notional:
raise ValueError(
f"Notional {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}: price {text(price)} out of range "
f"{text(minimum)}–{text(maximum)}"
)
return cost
# --------------------------------------------------------------------------
# Grid Plan Builder
# --------------------------------------------------------------------------
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]]:
"""
Builds orders across cells. Returns (orders, skipped_cells).
Base currency fee reduces the paired SELL amount.
"""
if budget_per_level <= 0:
raise ValueError("Budget per level must be greater than zero")
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 cell: buy at low, sell at 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)
# Paired SELL will be placed later, percentage filter will be
# checked by the exchange itself, so avg_price is not passed here.
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 cell: amount calculated from the SELL price itself,
# ensuring the notional does not exceed the level budget.
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"cell {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:
"""
Capital requirement estimation. This is an ESTIMATE padded for slippage,
not a guarantee: actual execution prices and initial buy fees may vary.
"""
buys = [o for o in orders if o["side"] == "buy"]
sells = [o for o in orders if o["side"] == "sell"]
# BUY order execution consumes no more than notional: fee is taken
# from the received asset (BTC) or from BNB, not from USDT.
quote_for_buys = sum((o["notional"] for o in buys), ZERO)
base_net_needed = sum((o["amount"] for o in sells), ZERO)
# Initial market buy (taker). When fee is in base, we need to buy
# more so that base_net_needed remains after the fee.
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:
# Limit orders (BUY, paired and initial SELL) use maker fee,
# initial market buy uses taker fee. Limit orders that immediately
# cross the book will fill as taker: this estimate does not catch that.
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,
}
# --------------------------------------------------------------------------
# Exchange Integration
# --------------------------------------------------------------------------
def restrict_to_spot(exchange) -> None:
"""Instruct ccxt to fetch only spot markets (option format depends on version)."""
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]:
"""Returns (bid, ask, last)."""
ticker = exchange.fetch_ticker(symbol)
if ticker.get("bid") is None or ticker.get("ask") is None:
raise RuntimeError(f"{symbol}: exchange did not return 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}: invalid 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:
"""
Reference price for PERCENT_PRICE*. Replicates Binance logic:
avgPriceMins == 0 -> last price, otherwise average over avgPriceMins minutes.
The window returned by /avgPrice is checked against the filter window.
The final check is still enforced by the exchange upon order submission.
"""
if rules.avg_price_mins == 0:
if last is None:
raise RuntimeError("Filter requires last price, but exchange did not return it")
return last
response = exchange.publicGetAvgPrice({"symbol": market["id"]})
price = dec(response["price"], "avgPrice")
if price <= 0:
raise RuntimeError("Binance returned an invalid 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 window ({mins} min) does not match filter avgPriceMins "
f"({rules.avg_price_mins} min): unable to replicate check"
)
return price
def resolve_rate(
market: dict,
key: str,
explicit: Decimal | None,
fee_in_bnb: bool,
) -> Decimal:
"""
Fee rate: explicit, or from market (maker/taker), or defaults to 0.1%.
BNB fee discount is NOT assumed: the actual account tier rate must
be passed explicitly, otherwise the estimate is conservative (no discount).
"""
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(
"Fee rate %s not set explicitly: using %s without BNB discount. "
"Pass your actual account tier rate via "
"--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:
"""
For future execution module: paired SELL amount after actual BUY execution.
executed_qty — GROSS bought quantity (order's executedQty, pre-fee).
fills — ccxt order fills: each has
fill["fee"] = {"cost": ..., "currency": ...} or None.
Only fees charged in the base currency are deducted: BNB or quote
currency fees do not reduce the asset quantity.
free_balance — free base currency balance not reserved for other orders
(caller is responsible for ensuring this).
"""
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("Net quantity is not positive after fee deduction")
amount = floor_to_step(min(net, free_balance), rules.step)
validate_order(rules, "sell", price, amount)
return amount
# --------------------------------------------------------------------------
# Main Execution
# --------------------------------------------------------------------------
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="Max notional value of a single order in quote currency",
)
parser.add_argument(
"--fee-rate", default=None,
help="Maker fee rate (fraction), defaults to market maker rate",
)
parser.add_argument(
"--taker-fee-rate", default=None,
help="Taker fee rate for initial purchase, defaults to market taker rate",
)
parser.add_argument(
"--fee-in-bnb", action="store_true",
help="Fees paid in BNB rather than the received asset",
)
parser.add_argument("--extra-cost-rate", default="0.0005")
parser.add_argument(
"--slippage", default="0.002",
help="Initial purchase slippage buffer (fraction)",
)
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} must be within range [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"Trading pair not found: {symbol}")
market = exchange.market(symbol)
if market.get("spot") is False:
raise RuntimeError(f"{symbol} is not a spot market")
if market.get("active") is False:
raise RuntimeError(f"Trading pair is inactive: {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"Current price {text(mid)} is outside grid range "
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("Skipped: %s", message)
if not orders:
raise RuntimeError("No orders passed validation")
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("Pair %s not found: unable to estimate BNB usage", bnb_symbol)
summary = summarize_plan(
orders, fee_rate, taker_rate, fee_in_base, ask, slippage, bnb_price
)
print(f"Pair: {symbol}")
print(f"Bid / Ask: {text(bid)} / {text(ask)} (mid {text(mid)})")
print(f"Tick / Step: {text(rules.tick)} / {text(rules.step)}")
print(f"Grid: {text(levels[0])} – {text(levels[-1])}, "
f"cells: {len(levels) - 1}")
print(f"Fee Rate: maker {fee_rate:.4%}, taker {taker_rate:.4%} "
f"({'in received asset' if fee_in_base else 'in BNB'})")
if avg_price is not None:
print(f"Filter ref price: {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" -> on fill 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 orders: {summary['buy_orders']}, "
f"requires ~{money(summary['quote_for_buys'])} {quote}")
print(f"SELL orders: {summary['sell_orders']}, "
f"reserve needed {text(summary['base_needed_for_sells'])} {base}")
print(f"Initial buy ~{money(summary['initial_buy_cost_estimate'])} "
f"{quote} (at ask, taker fee, slippage {slippage:.2%})")
print(f"Total ~{money(summary['total_quote_estimate'])} {quote} "
f"(estimate, not a guarantee of sufficient balance)")
if not fee_in_base:
line = (f"BNB fees: ~{money(summary['bnb_fee_quote_equivalent'])} "
f"{quote} equivalent")
if summary["bnb_needed"] is not None:
line += f", ~{summary['bnb_needed']:.4f} BNB"
print(line + " (full lifecycle of all orders)")
print()
print("Dry-run mode: orders were NOT sent to the exchange.")
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("Exchange/Network error: %s", exc)
return 1
except (ValueError, RuntimeError) as exc:
logger.error("Calculation error: %s", exc)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())Save and name the file, for example, grid_planner.py. Run: python grid_planner.py --grid-count 20 --budget-per-level 100 --fee-in-bnb --fee-rate 0.00075 --taker-fee-rate 0.00075
4. Hidden Risks and Bot Traps Exchanges Never Talk About
Exchange marketers love flexing cherry-picked backtest gains for their trading bots. In the real world, algo trading hits technical brick walls that can blow up your account faster than fat-finger human error.
- Slippage During High Volatility Spikes: When the market nukes through support, order book liquidity vanishes in milliseconds. Your bot tries to place or execute a market order, but fills get caught at absolute worst-case prices. On illiquid low-cap altcoins, a single volatility impulse can wipe out a full week's worth of grid profits.
- Fee Toxicity (Maker vs. Taker): Grid bots rely on passive limit orders—that’s Maker territory, where exchanges offer rebates or tiny fees. But if your bot's logic panic-sells at market (Stop-Loss or breakdown panic), brutal Taker fees kick in, instantly destroying your strategy's expected value.
- Backtest Overfitting & Illusions: Historical data ignores real-world network latency (ping to the exchange API) and matching engine queue delay. A bot flexing +300% on historical 2024–2025 data can bleed out on a crab market in live trading purely due to execution lag.
5. Protecting Your Stack: Risk Management Checklist for Algo Traders
You can't just spin up a bot and head out on vacation. Automation demands bulletproof infrastructure and tight oversight:
Safety Rules & Guardrails:
- Hard Kill-Switch: Hardcode a global drawdown limit into your code or terminal settings. If total portfolio value drops by, say, 15%, the bot must pull all open limit orders and immediately lock in the loss.
- Capital Isolation: Never keep bot trading capital on the same sub-account as your main cold wallet or long-term holdings. Use isolated API keys with withdrawal permissions strictly disabled—trading permissions only.
- Real-Time Telegram Alerts: Route critical alerts straight to your messenger instead of relying on exchange UI notifications. Canceled sessions, API rate-limit errors, or emergency market fills must trigger instant pings.
The Bottom Line
A trading bot is a surgeon’s scalpel, not a flight autopilot. It automates discipline, but it doesn't replace edge or intelligence. Start small with spot grid strategies on highly liquid pairs, benchmark your logic against real execution latency, and remember: the market isn't your biggest PnL killer—reckless bot settings without risk management are.
EXMON Academy Note: Battle-tested infrastructure and strict risk management always come first. The real winners in quant trading are those who know when to hit the kill-switch.
