Automatización de trading
Vamos a desglosar la verdadera mecánica de la automatización de trading sin falsas ilusiones: qué tipos de bots realmente dejan ganancia en el criptomercado, dónde se esconden los riesgos ocultos y cómo estructurar la lógica de código para armarlos.
1. Arquitectura de estrategias: qué funciona de verdad en la cancha
En este mercado solo sobreviven tres tipos principales de sistemas automatizados. El resto es puro marketing de los exchanges para farmear comisiones.
- Grid Trading (Bots de rejilla): Ideal para mercados laterales (rangos o flet). El bot despliega una escalera de órdenes límite de compra por debajo del precio actual y de venta por encima, generando ganancias con las oscilaciones dentro del canal.
- Dollar-Cost Averaging (Bots DCA): Una estrategia clásica de acumulación. El bot compra el activo en partes iguales a intervalos fijos de tiempo o cada vez que el precio cae un porcentaje determinado. Funciona en spot para armar un portafolio a largo plazo.
- Arbitraje: Capturar diferencias de precios entre mercados spot de diferentes plataformas o el spread entre futuros y spot (funding rate arbitrage). Requiere baja latencia (low latency) y APIs ultra optimizadas.
2. La matemática de los Grid Bots: por qué el 80% de los usuarios queman su cuenta
La mayoría arranca un grid bot en el pico (ATH) de una altcoin recontra volátil y con la configuración por defecto. El desenlace siempre es el mismo: un quiebre bajista brutal del rango, el bot termina vendiendo toda la moneda base a precio de remate, y el portafolio queda cargado de un activo "muerto" con una pérdida flotante del 60%.
Para esquivar esta trampa, el cálculo de parámetros se rige por reglas estrictas:
Comparativa de enfoques para configurar una rejilla:
- Selección de activo: Error de novato — un memecoin reventado de HYPE con volatilidad salvaje. Enfoque pro — monedas del top 20 con alta liquidez y soportes sólidos (BTC, ETH).
- Tipo de grid: Error de novato — aritmética (pasos fijos en dólares). Enfoque pro — geométrica (pasos porcentuales iguales, clave para mantener un rendimiento porcentual estable).
- Rango: Error de novato — un corredor estrecho al 5% del precio actual. Enfoque pro — un rango amplio que cubra la consolidación histórica de los últimos 3 a 6 meses.
3. Implementación práctica: código del módulo de grid trading
Abajo te comparto un fragmento de script funcional en Python usando la librería ccxt. Este es el núcleo de la lógica de cálculo para generar las órdenes de la rejilla con paso geométrico. Te dejo explicado qué hace cada parte en los comentarios.
"""
Planificador de grilla de órdenes limitadas para Binance Spot (modo dry-run).
Modelo de grilla
----------------
Los niveles L0 < L1 < ... < Ln forman n celdas [L_k, L_{k+1}].
En cada celda vive exactamente una orden activa a la vez:
- L_k < precio -> BUY en L_k. Tras ejecutarse, se coloca un SELL en L_{k+1}
por la cantidad resultante menos la comisión en la moneda base.
- L_k >= precio -> SELL en L_{k+1}. Esto requiere un inventario inicial de moneda base
(hay que comprarlo a mercado por anticipado).
El nivel igual al precio actual pertenece a la celda SELL (no se salta).
Este script NO envía absolutamente nada al 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")
# --------------------------------------------------------------------------
# Utilidades Decimal
# --------------------------------------------------------------------------
def dec(value: Any, name: str) -> Decimal:
try:
result = Decimal(str(value))
except (InvalidOperation, ValueError, TypeError) as exc:
raise ValueError(f"{name}: valor numérico inválido") from exc
if not result.is_finite():
raise ValueError(f"{name}: el valor debe ser finito")
return result
def text(value: Decimal) -> str:
"""Decimal -> string sin notación científica ni ceros sobrantes."""
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
# --------------------------------------------------------------------------
# Reglas (filtros) del instrumento en 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]]
# Ventana de promedio para los filtros porcentuales; 0 = se usa el último precio.
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:
"""En los filtros de Binance, 0 significa 'restricción desactivada'."""
if value is None:
return None
result = dec(value, name)
return result if result > 0 else None
def load_rules(market: dict) -> SymbolRules:
"""Lee los filtros una sola vez. Cualquier problema acá es un error de config."""
info = market.get("info") or {}
status = info.get("status")
if status is not None and status != "TRADING":
raise RuntimeError(f"El status del par es {status}, se esperaba TRADING")
if info.get("isSpotTradingAllowed") is False:
raise RuntimeError("El trading Spot está desactivado para este par")
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("No se encontró el PRICE_FILTER de Binance")
if not lot_filter:
raise RuntimeError("No se encontró el LOT_SIZE de Binance")
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 y stepSize deben ser mayores a cero")
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"Hay avgPriceMins dispares en los filtros: {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,
)
# --------------------------------------------------------------------------
# Grilla
# --------------------------------------------------------------------------
def align_bounds(
lower: Decimal, upper: Decimal, tick: Decimal
) -> tuple[Decimal, Decimal]:
"""Los límites se alinean hacia adentro del rango."""
lower = ceil_to_step(lower, tick)
upper = floor_to_step(upper, tick)
if lower >= upper:
raise ValueError("El rango queda vacío después de alinear con el tickSize")
return lower, upper
def calculate_grid_levels(
lower_bound: Decimal,
upper_bound: Decimal,
grid_count: int,
tick: Decimal,
) -> list[Decimal]:
"""Grilla geométrica: grid_count intervalos, grid_count + 1 niveles."""
if not isinstance(grid_count, int) or isinstance(grid_count, bool):
raise ValueError("grid_count debe ser un número entero")
if grid_count < 1:
raise ValueError("grid_count debe ser al menos 1")
if min(lower_bound, upper_bound, tick) <= 0:
raise ValueError("Los límites y el tickSize deben ser mayores a cero")
if lower_bound >= upper_bound:
raise ValueError("lower_bound debe ser menor que 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))
# Cada nivel se calcula desde el límite inferior: no se acumula error.
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(
"Los niveles se pisaron o rompieron el orden al redondear al tick. "
"Bajá el grid_count o ampliá el rango."
)
return levels
def check_grid_economics(
levels: list[Decimal],
fee_rate: Decimal,
extra_cost_rate: Decimal,
) -> None:
"""
Filtro aproximado: el rendimiento neto de la celda, considerando comisiones
en ambas puntas, debe superar el reaseguro de costos.
Esto no es garantía de rentabilidad.
"""
if fee_rate < 0 or extra_cost_rate < 0:
raise ValueError("La comisión y el reaseguro de costos no pueden ser < 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"El paso de la grilla es muy chico: {text(low)} -> {text(high)}; "
f"retorno neto {net_return:.4%}, "
f"se requiere más de {extra_cost_rate:.4%}"
)
# --------------------------------------------------------------------------
# Validación de la orden contra los filtros del exchange
# --------------------------------------------------------------------------
def validate_order(
rules: SymbolRules,
side: str,
price: Decimal,
amount: Decimal,
avg_price: Decimal | None = None,
) -> Decimal:
"""Devuelve el notional de la orden o lanza un ValueError."""
if side not in ("buy", "sell"):
raise ValueError("side debe ser buy o sell")
if price <= 0 or amount <= 0:
raise ValueError(
f"{side.upper()} @ {text(price)}: precio y cantidad deben ser > 0 "
f"(amount={text(amount)})"
)
if price % rules.tick != 0:
raise ValueError(f"El precio {text(price)} no es múltiplo de tickSize")
if amount % rules.step != 0:
raise ValueError(f"La cantidad {text(amount)} no es múltipla de stepSize")
if price < rules.min_price:
raise ValueError(f"Precio {text(price)} < minPrice {text(rules.min_price)}")
if rules.max_price is not None and price > rules.max_price:
raise ValueError(f"Precio {text(price)} > maxPrice {text(rules.max_price)}")
if amount < rules.min_qty:
raise ValueError(
f"Cantidad {text(amount)} < minQty {text(rules.min_qty)}"
)
if rules.max_qty is not None and amount > rules.max_qty:
raise ValueError(
f"Cantidad {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}: precio {text(price)} fuera de rango "
f"{text(minimum)}–{text(maximum)}"
)
return cost
# --------------------------------------------------------------------------
# Plan de la grilla
# --------------------------------------------------------------------------
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]]:
"""
Arma las órdenes por celda. Devuelve (órdenes, celdas_saltadas).
La comisión en moneda base achica la cantidad del SELL gemelo.
"""
if budget_per_level <= 0:
raise ValueError("El presupuesto por nivel debe ser mayor a cero")
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:
# Celda BUY: compramos en low, vendemos en 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)
# El SELL gemelo se lanza después; el filtro porcentual
# lo va a chequear el propio exchange en ese momento, por eso acá no pasamos 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:
# Celda SELL: la cantidad se calcula a partir del precio del propio SELL,
# para que el notional no se pase del presupuesto por nivel.
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"celda {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:
"""
Estimación de capital necesario. Es una ESTIMACIÓN con resto para slippage,
no una garantía: el precio real y las comisiones de la compra inicial pueden variar.
"""
buys = [o for o in orders if o["side"] == "buy"]
sells = [o for o in orders if o["side"] == "sell"]
# La orden BUY gasta a lo sumo su notional: la comisión se cobra
# del activo obtenido (BTC) o de BNB, jamás de USDT.
quote_for_buys = sum((o["notional"] for o in buys), ZERO)
base_net_needed = sum((o["amount"] for o in sells), ZERO)
# Compra inicial a mercado (taker). Si la comisión es en la base hay que
# comprar más para que, tras la comisión, quede el base_net_needed exacto.
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:
# Las órdenes limit (BUY, SELLs gemelos e iniciales) corren como maker,
# y la compra inicial a mercado como taker. Una orden limit que se lleve
# puesta la punta al toque sale como taker: esta estimación no capta eso.
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,
}
# --------------------------------------------------------------------------
# Operación con el exchange
# --------------------------------------------------------------------------
def restrict_to_spot(exchange) -> None:
"""Le pedimos a ccxt que traiga solo mercados spot (el formato de opción varía según la versión)."""
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]:
"""Devuelve (bid, ask, last)."""
ticker = exchange.fetch_ticker(symbol)
if ticker.get("bid") is None or ticker.get("ask") is None:
raise RuntimeError(f"{symbol}: el exchange no devolvió 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 inválidos {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:
"""
Precio de referencia para PERCENT_PRICE*. Replica la regla de Binance:
avgPriceMins == 0 -> último precio, sino el promedio de los últimos avgPriceMins minutos.
La ventana que devolvió /avgPrice se coteja con la ventana del filtro.
El chequeo definitivo igual lo hace el exchange al enviar la orden.
"""
if rules.avg_price_mins == 0:
if last is None:
raise RuntimeError("El filtro requiere un precio last y el exchange no lo devolvió")
return last
response = exchange.publicGetAvgPrice({"symbol": market["id"]})
price = dec(response["price"], "avgPrice")
if price <= 0:
raise RuntimeError("Binance devolvió un avgPrice inválido")
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"La ventana avgPrice ({mins} min) no coincide con avgPriceMins "
f"del filtro ({rules.avg_price_mins} min): imposible replicar la validación"
)
return price
def resolve_rate(
market: dict,
key: str,
explicit: Decimal | None,
fee_in_bnb: bool,
) -> Decimal:
"""
Tasa de comisión: explícita, sino la del mercado (maker/taker), sino 0.1%.
No se asume descuento por pagar en BNB: la tasa real de la cuenta
debe pasarse explícitamente, de lo contrario la estimación es conservadora (sin descuento).
"""
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(
"La tasa %s no está definida explícitamente: usando %s sin descuento de BNB. "
"Pasá la tasa real de tu cuenta mediante "
"--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:
"""
Para el futuro módulo de ejecución: cantidad del SELL gemelo tras
la ejecución real de un BUY.
executed_qty — cantidad BRUTA comprada (executedQty de la orden,
antes de descontar la comisión).
fills — ejecuciones de la orden en formato ccxt: cada una trae
fill["fee"] = {"cost": ..., "currency": ...} o None.
Solo se descuentan comisiones cobradas en moneda base: comisiones en BNB
o en la moneda cotizada no achican la cantidad del activo.
free_balance — saldo libre de moneda base no reservado
para otras órdenes (de esto se encarga quien llama).
"""
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("La cantidad no es positiva tras descontar la comisión")
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="Notional máximo de una sola orden en la moneda cotizada",
)
parser.add_argument(
"--fee-rate", default=None,
help="Comisión de maker (fracción), por defecto el maker del mercado",
)
parser.add_argument(
"--taker-fee-rate", default=None,
help="Comisión de taker para la compra inicial, por defecto el taker del mercado",
)
parser.add_argument(
"--fee-in-bnb", action="store_true",
help="La comisión se paga en BNB en lugar del activo obtenido",
)
parser.add_argument("--extra-cost-rate", default="0.0005")
parser.add_argument(
"--slippage", default="0.002",
help="Reaseguro por deslizamiento (slippage) para la compra inicial (fracción)",
)
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} debe estar en el rango [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"Par de trading no encontrado: {symbol}")
market = exchange.market(symbol)
if market.get("spot") is False:
raise RuntimeError(f"{symbol} no es un mercado spot")
if market.get("active") is False:
raise RuntimeError(f"El par de trading está inactivo: {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"El precio actual {text(mid)} está fuera del rango de la grilla "
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("Saltada: %s", message)
if not orders:
raise RuntimeError("Ninguna orden pasó la validación")
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("Par %s no encontrado: imposible estimar gasto en BNB", bnb_symbol)
summary = summarize_plan(
orders, fee_rate, taker_rate, fee_in_base, ask, slippage, bnb_price
)
print(f"Par: {symbol}")
print(f"Bid / Ask: {text(bid)} / {text(ask)} (mid {text(mid)})")
print(f"Tick / Step: {text(rules.tick)} / {text(rules.step)}")
print(f"Grilla: {text(levels[0])} – {text(levels[-1])}, "
f"celdas: {len(levels) - 1}")
print(f"Comisiones: maker {fee_rate:.4%}, taker {taker_rate:.4%} "
f"({'en el activo obtenido' if fee_in_base else 'en BNB'})")
if avg_price is not None:
print(f"Precio de referencia de filtros: {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" -> tras ejecución 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"Órdenes BUY: {summary['buy_orders']}, "
f"se necesitan ~{money(summary['quote_for_buys'])} {quote}")
print(f"Órdenes SELL: {summary['sell_orders']}, "
f"se necesita reserva de {text(summary['base_needed_for_sells'])} {base}")
print(f"Compra inicial ~{money(summary['initial_buy_cost_estimate'])} "
f"{quote} (al ask, comisión taker, slippage {slippage:.2%})")
print(f"Total ~{money(summary['total_quote_estimate'])} {quote} "
f"(estimado, sin garantía de saldo suficiente)")
if not fee_in_base:
line = (f"Comisiones en BNB: ~{money(summary['bnb_fee_quote_equivalent'])} "
f"{quote} en equivalente")
if summary["bnb_needed"] is not None:
line += f", ~{summary['bnb_needed']:.4f} BNB"
print(line + " (ciclo completo de todas las órdenes)")
print()
print("Modo dry-run: NO se mandó ninguna orden al 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("Error de exchange/red: %s", exc)
return 1
except (ValueError, RuntimeError) as exc:
logger.error("Error de cálculo: %s", exc)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())Guardá y nombrá el archivo como, por ejemplo, grid_planner.py. Ejecución: python grid_planner.py --grid-count 20 --budget-per-level 100 --fee-in-bnb --fee-rate 0.00075 --taker-fee-rate 0.00075
4. Riesgos ocultos y trampas de la automatización que los exchanges no te cuentan
A los marketers de los exchanges les encanta mostrar gráficos hermosos de rentabilidad de bots en backtests históricos. En la realidad, el trading algorítmico choca contra paredes técnicas capaces de quemar tu cuenta más rápido que un error humano por dedo gordo.
- Slippage en alta volatilidad: Cuando el mercado rompe un soporte, el order book se vacía en milisegundos. El bot intenta colocar o ejecutar una orden a mercado, pero la transacción se engancha al peor precio posible. En altcoins de baja liquidez, un solo impulso de volatilidad puede devorar la ganancia de toda una semana de la grilla.
- Toxicidad de comisiones (Maker vs Taker): Una grilla se basa en órdenes limit — el territorio Maker, donde los exchanges seguido te dan rebates o cobran centavos. Pero si la lógica del bot se desboca en una venta de pánico a mercado (Stop-Loss o pánico ante un quiebre), se activan unas comisiones Taker monstruosas que matan por completo la matemática de la estrategia.
- La ilusión de los backtests: Los datos históricos ignoran la latencia de red real (el ping a la API del exchange) y las colas del motor de emparejamiento (matching engine). Un bot que marcó un +300% en pruebas históricas de 2024–2025 puede irse a pérdida en un mercado lateral en vivo pura y exclusivamente por los retrasos de ejecución.
5. Cómo proteger tu capital: checklist de gestión de riesgo para bots
No podés prender un algoritmo e irte a la playa. La automatización exige una infraestructura de monitoreo blindada:
Reglas de seguridad y control:
- Kill-switch estricto: En el código o en la configuración de la plataforma tiene que estar programado un stop-loss absoluto para todo el capital del bot. Si el valor total combinado de la moneda base y la cotizada cae un 15%, ponele, el bot cancela todas las órdenes limit y corta la pérdida al toque.
- Aislamiento de capital: Jamás guardes el capital de trading del bot en la misma subcuenta que tus ahorros pesados a largo plazo. Usá API keys aisladas con los retiros bloqueados — solo se permite operar.
- Monitoreo con alertas de Telegram: Configurá las alertas críticas directo a tu mensajero en lugar de depender de la interfaz del exchange. Cada sesión caída, error de API u orden de emergencia ejecutada tiene que llegar al instante.
En pocas palabras
Un bot de trading es un bisturí en las manos de un cirujano, no el piloto automático de un avión. Automatiza la disciplina, pero no piensa por vos. Arrancá con montos chicos en grillas spot con pares líquidos, probá la lógica frente a la latencia real y acordate: el peor enemigo de tu PnL no es el mercado, sino tus propias configuraciones colgadas sin gestión de riesgo.
La Academia EXMON te lo recuerda: la infraestructura sólida y el control de riesgos van siempre primero. El que realmente hace plata es el que sabe cuándo cortar el algoritmo a tiempo.
