Automatyzacja handlu
Spójrzmy prawdzie w oczy i przeanalizujmy mechanikę automatyzacji handlu bez zbędnych iluzji. Sprawdzimy, które typy botów faktycznie generują zysk na rynku crypto, gdzie kryją się ukryte ryzyka i jak napisać dla nich odpowiednią logikę.
1. Architektura strategii: co działa w praktyce
Na rynku przeżywają tylko trzy główne rodzaje systemów zautomatyzowanych. Cała reszta to czysty marketing giełdowy mający na celu zbieranie prowizji.
- Siatka (Grid Trading): Idealna do rynków w trendzie bocznym (konsolidacji). Bot wystawia drabinkę zleceń oczekujących (limit orders) poniżej obecnej ceny na zakup i powyżej na sprzedaż. Zarabia na wahaniach wewnątrz wyznaczonego korytarza.
- Uśrednianie kosztów (Boty DCA): Strategia akumulacji. Bot dokupuje aktywa w równych częściach w stałych odstępach czasu lub w momencie spadku ceny o zadany procent. Działa na rynku spot w ramach portfela długoterminowego.
- Arbitraż: Wyłapywanie różnic cenowych między rynkami spot na różnych giełdach lub spreadu między rynkiem kontraktów terminowych a spot (funding rate arbitrage). Wymaga niskich opóźnień (low latency) oraz zoptymalizowanego API.
2. Matematyka botów Grid: dlaczego 80% użytkowników traci kapitał
Większość uruchamia siatkę na samych górkach (ath) zmiennego altcoina przy domyślnych ustawieniach. Finał zawsze jest taki sam: gwałtowne przebicie zakresu w dół, bot wyprzedaje całą bazową walutę po groszowych cenach, a portfel zostaje zapchany „martwym” aktywem z 60% obsunięciem kapitału (drawdown).
Aby tego uniknąć, kalkulacja parametrów opiera się na żelaznych zasadach:
Porównanie podejść do konfiguracji siatki:
- Wybór aktywa: Błąd nowicjusza – hajpowy memecoin o szalonej zmienności. Podejście profesjonalne – monety z top 20 o wysokiej płynności i silnych poziomach wsparcia (BTC, ETH).
- Typ siatki: Błąd nowicjusza – arytmetyczna (równe kroki w USD). Podejście profesjonalne – geometryczna (równe kroki procentowe, co jest kluczowe dla stabilnego procentu zysku).
- Zakres: Błąd nowicjusza – wąski korytarz na 5% od aktualnej ceny. Podejście profesjonalne – szeroki zakres pokrywający historyczną konsolidację z ostatnich 3–6 miesięcy.
3. Praktyczna implementacja: kod modułu handlowego siatki
Poniżej znajduje się działający fragment skryptu w Pythonie z wykorzystaniem biblioteki ccxt. To rdzeń logiki obliczeniowej do generowania zleceń siatki z krokiem geometrycznym. W komentarzach wyjaśniam co i do czego służy.
"""
Planista siatki zleceń limit dla Binance Spot (tryb symulacji).
Model siatki
------------
Poziomy L0 < L1 < ... < Ln tworzą n komórek [L_k, L_{k+1}].
W każdej komórce znajduje się jednocześnie dokładnie jedno zlecenie:
- L_k < cena -> BUY na L_k. Po realizacji wystawiane jest SELL na L_{k+1}
na ilość otrzymaną pomniejszoną o prowizję w walucie bazowej.
- L_k >= cena -> SELL на L_{k+1}. Wymaga początkowego zapasu waluty
bazowej (należy ją wcześniej dokupić po cenie rynkowej).
Poziom równy cenie bieżącej należy do komórki SELL (nie jest pomijany).
Skrypt NICzego nie wysyła na giełdę.
"""
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")
# --------------------------------------------------------------------------
# Narzędzia Decimal
# --------------------------------------------------------------------------
def dec(value: Any, name: str) -> Decimal:
try:
result = Decimal(str(value))
except (InvalidOperation, ValueError, TypeError) as exc:
raise ValueError(f"{name}: niepoprawna wartość numeryczna") from exc
if not result.is_finite():
raise ValueError(f"{name}: wartość musi być skończona")
return result
def text(value: Decimal) -> str:
"""Decimal -> łańcuch znaków bez notacji wykładniczej i zbędnych zer."""
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
# --------------------------------------------------------------------------
# Zasady (filtry) instrumentu 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]]
# Okno uśrednianя dla filtrów procentowych; 0 = używana jest cena ostatnia.
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:
"""W filtrach Binance 0 oznacza „ograniczenie wyłączone”."""
if value is None:
return None
result = dec(value, name)
return result if result > 0 else None
def load_rules(market: dict) -> SymbolRules:
"""Odczytuje filtry jednorazowo. Każdy problem tutaj to błąd konfiguracji."""
info = market.get("info") or {}
status = info.get("status")
if status is not None and status != "TRADING":
raise RuntimeError(f"Status pary to {status}, oczekiwano TRADING")
if info.get("isSpotTradingAllowed") is False:
raise RuntimeError("Handel Spot dla tej pary jest zablokowany")
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("Nie znaleziono Binance PRICE_FILTER")
if not lot_filter:
raise RuntimeError("Nie znaleziono 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 i stepSize muszą być większe od zera")
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"Różne wartości avgPriceMins w filtrach: {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,
)
# --------------------------------------------------------------------------
# Siatka
# --------------------------------------------------------------------------
def align_bounds(
lower: Decimal, upper: Decimal, tick: Decimal
) -> tuple[Decimal, Decimal]:
"""Granice są wyrównywane do wnętrza zakresu."""
lower = ceil_to_step(lower, tick)
upper = floor_to_step(upper, tick)
if lower >= upper:
raise ValueError("Po wyrównaniu do tickSize zakres jest pusty")
return lower, upper
def calculate_grid_levels(
lower_bound: Decimal,
upper_bound: Decimal,
grid_count: int,
tick: Decimal,
) -> list[Decimal]:
"""Siatka geometryczna: grid_count interwałów, grid_count + 1 poziomów."""
if not isinstance(grid_count, int) or isinstance(grid_count, bool):
raise ValueError("grid_count musi być liczbą całkowitą")
if grid_count < 1:
raise ValueError("grid_count musi wynosić co najmniej 1")
if min(lower_bound, upper_bound, tick) <= 0:
raise ValueError("Granice oraz tickSize muszą być większe od zera")
if lower_bound >= upper_bound:
raise ValueError("lower_bound musi być mniejsze niż 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))
# Każdy poziom jest liczony od dolnej granicy: błąd nie kumuluje się.
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(
"Poziomy pokryły się lub naruszyły porządek po zaokrągleniu do ticku. "
"Zmniejsz grid_count lub poszerz zakres."
)
return levels
def check_grid_economics(
levels: list[Decimal],
fee_rate: Decimal,
extra_cost_rate: Decimal,
) -> None:
"""
Filtr przybliżony: czysta rentowność komórki z uwzględnieniem prowizji
na obu nogach musi przekraczać bufor kosztów.
To nie jest gwarancją zyskowności.
"""
if fee_rate < 0 or extra_cost_rate < 0:
raise ValueError("Prowizja i bufor kosztów nie mogą być < 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"Krok siatki jest zbyt mały: {text(low)} -> {text(high)}; "
f"czysta rentowność {net_return:.4%}, "
f"wymagane więcej niż {extra_cost_rate:.4%}"
)
# --------------------------------------------------------------------------
# Walidacja zlecenia według filtrów giełdy
# --------------------------------------------------------------------------
def validate_order(
rules: SymbolRules,
side: str,
price: Decimal,
amount: Decimal,
avg_price: Decimal | None = None,
) -> Decimal:
"""Zwraca koszt nominalny zlecenia lub zgłasza ValueError."""
if side not in ("buy", "sell"):
raise ValueError("side musi być buy lub sell")
if price <= 0 or amount <= 0:
raise ValueError(
f"{side.upper()} @ {text(price)}: cena i ilość muszą być > 0 "
f"(amount={text(amount)})"
)
if price % rules.tick != 0:
raise ValueError(f"Cena {text(price)} nie jest wielokrotnością tickSize")
if amount % rules.step != 0:
raise ValueError(f"Ilość {text(amount)} nie jest wielokrotnością stepSize")
if price < rules.min_price:
raise ValueError(f"Cena {text(price)} < minPrice {text(rules.min_price)}")
if rules.max_price is not None and price > rules.max_price:
raise ValueError(f"Cena {text(price)} > maxPrice {text(rules.max_price)}")
if amount < rules.min_qty:
raise ValueError(
f"Ilość {text(amount)} < minQty {text(rules.min_qty)}"
)
if rules.max_qty is not None and amount > rules.max_qty:
raise ValueError(
f"Ilość {text(amount)} > maxQty {text(rules.max_qty)}"
)
cost = price * amount
if cost < rules.min_notional:
raise ValueError(
f"Koszt nominalny {text(cost)} < minNotional {text(rules.min_notional)}"
)
if rules.max_notional is not None and cost > rules.max_notional:
raise ValueError(
f"Koszt nominalny {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}: cena {text(price)} poza zakresem "
f"{text(minimum)}–{text(maximum)}"
)
return cost
# --------------------------------------------------------------------------
# Plan siatki
# --------------------------------------------------------------------------
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]]:
"""
Buduje zlecenia w komórkach. Zwraca (zlecenia, pominięte komórki).
Prowizja w walucie bazowej zmniejsza ilość sparowanego SELL.
"""
if budget_per_level <= 0:
raise ValueError("Budżet na poziom musi być większy od zera")
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:
# Komórka BUY: kupujemy na low, sprzedajemy na 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)
# Sparowany SELL zostanie wystawiony później, filtr procentowy zweryfikuje
# wówczas sama giełda, dlatego nie przekazujemy tutaj 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:
# Komórka SELL: ilość jest obliczana od ceny samego SELL,
# aby nominał nie przekraczał budżetu na poziom.
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"komórka {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:
"""
Szacowanie zapotrzebowania na środki. Jest to SZACOWE zapotrzebowanie z buforem na poślizg,
a nie gwarancja: rzeczywista cena i prowizje przy zakupie początkowym mogą
się różnić.
"""
buys = [o for o in orders if o["side"] == "buy"]
sells = [o for o in orders if o["side"] == "sell"]
# Zakup zlecenia BUY pobiera nie więcej niż notional: prowizja jest pobierana
# z otrzymywanego aktywa (BTC) lub z BNB, a nie z USDT.
quote_for_buys = sum((o["notional"] for o in buys), ZERO)
base_net_needed = sum((o["amount"] for o in sells), ZERO)
# Zakup początkowy rynkowy (taker). Przy prowizji w walucie bazowej kupować
# trzeba więcej, aby po prowizji pozostało 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:
# Zlecenia limit (BUY, sparowane i początkowe SELL) liczone są po maker,
# początkowy zakup rynkowy — po taker. Zlecenie limit, które
# od razu przecina arkusz, wykona się jako taker: szacunki tego nie uwzględniają.
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,
}
# --------------------------------------------------------------------------
# Obsługa giełdy
# --------------------------------------------------------------------------
def restrict_to_spot(exchange) -> None:
"""Prosimy ccxt o ładowanie tylko rynków spot (format opcji zależy od wersji)."""
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]:
"""Zwraca (bid, ask, last)."""
ticker = exchange.fetch_ticker(symbol)
if ticker.get("bid") is None or ticker.get("ask") is None:
raise RuntimeError(f"{symbol}: giełda nie zwróciła 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}: niepoprawne 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:
"""
Cena referencyjna dla PERCENT_PRICE*. Odtwarza regułę Binance:
avgPriceMins == 0 -> cena ostatnia, w przeciwnym razie średnia z avgPriceMins minut.
Okno zwrócone przez /avgPrice jest porównywane z oknem z filtra.
Ostateczną weryfikację i tak przeprowadza giełda przy składaniu zlecenia.
"""
if rules.avg_price_mins == 0:
if last is None:
raise RuntimeError("Dla filtra wymagana jest cena last, giełda jej nie zwróciła")
return last
response = exchange.publicGetAvgPrice({"symbol": market["id"]})
price = dec(response["price"], "avgPrice")
if price <= 0:
raise RuntimeError("Binance zwróciła niepoprawną 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"Okno avgPrice ({mins} min) nie zgadza się z avgPriceMins "
f"filtra ({rules.avg_price_mins} min): nie można odtworzyć weryfikacji"
)
return price
def resolve_rate(
market: dict,
key: str,
explicit: Decimal | None,
fee_in_bnb: bool,
) -> Decimal:
"""
Stawka prowizji: jawna, w przeciwnym razie z rynku (maker/taker), domyślnie 0.1%.
Zniżka za płatność w BNB NIE jest zakładana: faktyczną stawkę konta
należy przekazać jawnie, w przeciwnym razie szacunek jest konserwatywny (bez zniżki).
"""
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(
"Stawka %s nie została podana jawnie: użyto %s bez zniżki BNB. "
"Przekaż rzeczywistą stawkę konta za pomocą "
"--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:
"""
Dla przyszłego modułu wykonawczego: ilość sparowanego SELL po
rzeczywistym wykonaniu BUY.
executed_qty — BRUTTO kupiona ilość (executedQty zlecenia,
przed potrąceniem prowizji).
fills — wykonania zlecenia w formacie ccxt: każde posiada
fill["fee"] = {"cost": ..., "currency": ...} lub None.
Odlicza się tylko prowizje pobrane w walucie bazowej: prowizja w BNB
lub w walucie wyceny nie zmniejsza ilości aktywa.
free_balance — wolne saldo waluty bazowej, niezarezerwowane
pod inne zlecenia (odpowiada za to kod wywołujący).
"""
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("Po odliczeniu prowizji ilość nie jest dodatnia")
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="Maksymalny nominał pojedynczego zlecenia w walucie wyceny",
)
parser.add_argument(
"--fee-rate", default=None,
help="Prowizja maker (ułamek), domyślnie maker z rynku",
)
parser.add_argument(
"--taker-fee-rate", default=None,
help="Prowizja taker dla zakupu początkowego, domyślnie taker z rynku",
)
parser.add_argument(
"--fee-in-bnb", action="store_true",
help="Prowizja opłacana jest w BNB, a nie w otrzymywanym aktywie",
)
parser.add_argument("--extra-cost-rate", default="0.0005")
parser.add_argument(
"--slippage", default="0.002",
help="Bufor na poślizg cenowy zakupu początkowego (ułamek)",
)
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} musi być w zakresie [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"Nie znaleziono pary handlowej: {symbol}")
market = exchange.market(symbol)
if market.get("spot") is False:
raise RuntimeError(f"{symbol} nie jest rynkiem spot")
if market.get("active") is False:
raise RuntimeError(f"Para handlowa jest nieaktywna: {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"Bieżąca cena {text(mid)} poza zakresem siatki "
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("Pominięto: %s", message)
if not orders:
raise RuntimeError("Żadne zlecenie nie przeszło walidacji")
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("Nie znaleziono pary %s: nie można oszacować kosztu BNB", bnb_symbol)
summary = summarize_plan(
orders, fee_rate, taker_rate, fee_in_base, ask, slippage, bnb_price
)
print(f"Para: {symbol}")
print(f"Bid / Ask: {text(bid)} / {text(ask)} (mid {text(mid)})")
print(f"Tick / Step: {text(rules.tick)} / {text(rules.step)}")
print(f"Siatka: {text(levels[0])} – {text(levels[-1])}, "
f"komórek: {len(levels) - 1}")
print(f"Prowizja: maker {fee_rate:.4%}, taker {taker_rate:.4%} "
f"({'w otrzymywanym aktywie' if fee_in_base else 'w BNB'})")
if avg_price is not None:
print(f"Cena referencyjna filtrów: {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" -> po realizacji 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"Zleceń BUY: {summary['buy_orders']}, "
f"wymaga ~{money(summary['quote_for_buys'])} {quote}")
print(f"Zleceń SELL: {summary['sell_orders']}, "
f"potrzebny zapas {text(summary['base_needed_for_sells'])} {base}")
print(f"Zakup początkowy ~{money(summary['initial_buy_cost_estimate'])} "
f"{quote} (po ask, prowizja taker, poślizg {slippage:.2%})")
print(f"Razem ~{money(summary['total_quote_estimate'])} {quote} "
f"(szacunek, brak gwarancji wystarczającego salda)")
if not fee_in_base:
line = (f"Prowizje w BNB: ~{money(summary['bnb_fee_quote_equivalent'])} "
f"{quote} w ekwiwalencie")
if summary["bnb_needed"] is not None:
line += f", ~{summary['bnb_needed']:.4f} BNB"
print(line + " (pełny cykl wszystkich zleceń)")
print()
print("Tryb symulacji: zlecenia NIE zostały wysłane na giełdę.")
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("Błąd giełdy/sieci: %s", exc)
return 1
except (ValueError, RuntimeError) as exc:
logger.error("Błąd obliczeń: %s", exc)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())Zapisz i nazwij plik, na przykład grid_planner.py. Uruchomienie: python grid_planner.py --grid-count 20 --budget-per-level 100 --fee-in-bnb --fee-rate 0.00075 --taker-fee-rate 0.00075
4. Ukryte ryzyka i pułapki automatyzacji, o których giełdy wolą milczeć
Marketerzy giełdowi uwielbiają pokazywać ładne wykresy zysków botów z historycznych backtestów. W rzeczywistości handel algorytmiczny zderza się z barierami technicznymi, które potrafią wyczyścić depozyt szybciej niż ludzki błąd.
- Poślizg cenowy (Slippage) przy dużej zmienności: Kiedy rynek przebija poziom wsparcia, orderbook pustoszeje w milisekundę. Bot próbuje wystawić lub wykonać zlecenie rynkowe, ale transakcja łapie się po najgorszej cenie. Na cienkich altcoinach poślizg może w jeden impuls pochłonąć tygodniowy zysk z siatki.
- Toksyczność prowizji (Maker vs Taker): Siatka opiera się na zlecenie limitowanych — to tryb Makera, gdzie giełdy często płacą rebaty lub pobierają minimalny procent. Ale jeśli logika bota wpada w awaryjną wyprzedaż po cenie rynkowej (Stop-Loss lub panika przy przebiciu), włączają się zabójcze prowizje Takera, które niszczą matematykę całej strategii.
- Iluzja backtestów: Dane historyczne nie uwzględniają realnego opóźnienia sieci (ping do API giełdy) oraz kolejek dopasowywania zleceń (matching engine). Bot, który pokazał +300% w testach historycznych z lat 2024–2025, w realnej walce na boczniaku może polecieć na minus przez opóźnienia wykonania.
5. Jak chronić kapitał: check-list zarzadzania ryzykiem dla botów
Nie można odpalić algorytmu i jechać na urlop. Automatyzacja wymaga twardej infrastruktury monitoringowej:
Zasady bezpieczeństwa i kontroli:
- Twardy Kill-Switch: W kodzie lub ustawieniach terminala musi być zapisany bezwzględny stop-loss na cały depozyt bota. Jeśli łączna wartość wyceny waluty bazowej i kwotowanej spada o powiedziane 15%, bot ściąga wszystkie limity i zamyka pozycję na stratę.
- Izolacja depozytu: Nigdy nie trzymaj kapitału handlowego bota na tym samym subkoncie co główne, długoterminowe oszczędności. Używaj izolowanych kluczy API z wyłączoną funkcją wypłat — dozwolony wyłącznie handel.
- Monitoring przez alerty Telegram: Skonfiguruj wysyłanie krytycznych powiadomień nie do interfejsu giełdy, ale bezpośrednio do komunikatora. Każda przerwana sesja, błąd API czy zrealizowany awaryjny order muszą być raportowane natychmiast.
Podsumowując krótko.
Bot handlowy to skalpel w rękach chirurga, a nie żaden autopilot w samolocie. Automatyzuje dyscyplinę, ale nie zastępuje myślenia. Zaczynaj od małych kwot na siatkach spotowych dla płynnych instrumentów, testuj logikę pod kątem realnych opóźnień i pamiętaj: największym wrogim zysku nie jest rynek, lecz własne ustawienia bez wdrożonego risk managementu.
Akademia EXMON przypomina: niezawodna infrastruktura i kontrola ryzyka są zawsze na pierwszym miejscu. Zarabia ten, kto potrafi w porę zatrzymać algorytm.
