Alım-Satım Otomasyonu
Hiçbir illüzyona kapılmadan alım-satım otomasyonunun mekaniğini masaya yatıralım; kripto piyasasında hangi bot türlerinin gerçekten kazandırdığını, gizli risklerin nerede saklandığını ve bunlar için mantık (kod) kurgusunun nasıl yapılması gerektiğini inceleyelim.
1. Strateji Mimarisi: Pratikte Ne Çalışır?
Piyasada hayatta kalan sadece üç temel otomatik sistem türü var. Geri kalanı tamamen borsaların komisyon toplamak için yarattığı pazarlamadan ibaret.
- Grid (Ağ) Ticareti: Yatay piyasalar (flat/range) için kusursuzdur. Bot, mevcut fiyatın altına limit alım emirleri, üstüne ise limit satım emirleri olmak üzere bir merdiven kurar. Koridor içerisindeki dalgalanmalardan kâr sağlar.
- Maliyet Ortalaması (DCA Botları): Bir biriktirme stratejisidir. Bot, varlığı sabit zaman aralıklarında veya fiyat belirlenen bir yüzde oranında düştüğünde eşit parçalar halinde satın alır. Uzun vadeli portföyler için spot piyasada çalışır.
- Arbitraj Fırsatları: Farklı platformların spot piyasaları arasındaki fiyat farklarını ya da vadeli (futures) ile spot arasındaki spredi (funding rate arbitrage) yakalamak. Düşük gecikme (low latency) ve optimize edilmiş bir API gerektirir.
2. Grid Botlarının Matematiği: Kullanıcıların %80'i Neden Parasını Kaybeder?
Çoğu insan, varsayılan ayarlarla yüksek volatiliteye sahip bir altcoin'in tepesinde (ATH) grid botunu çalıştırır. Sonuç hep aynıdır: aralığın aşağı yönlü sert kırılması, botun tüm baz varlığı yerel fiyatlardan elden çıkarması ve portföyün %60 düşüşle "ölü" bir varlıkla dolması.
Bunun önüne geçmek için parametre hesabı katı kurallara dayanmalıdır:
Grid Kurulum Yaklaşımlarının Karşılaştırılması:
- Varlık Seçimi: Acemi hatası — çılgın volatiliteye sahip, popüler meme coin'ler. Profesyonel yaklaşım — yüksek likiditeye ve güçlü destek seviyelerine sahip ilk 20'deki coin'ler (BTC, ETH).
- Grid Türü: Acemi hatası — aritmetik (dolar cinsinden eşit adımlar). Profesyonel yaklaşım — geometrik (eşit yüzde adımları, istikrarlı getiri yüzdesi için kritiktir).
- Aralık (Range): Acemi hatası — mevcut fiyattan %5'lik dar bir koridor. Profesyonel yaklaşım — son 3-6 aydaki tarihi yatay piyasayı kapsayan geniş bir aralık.
3. Pratik Uygulama: Grid Alım-Satım Modülü Kodu
Aşağıda ccxt kütüphanesini kullanan çalışan bir Python betik parçası yer almaktadır. Bu, geometrik adımlarla grid emirleri üretmek için hesaplama mantığının çekirdeğidir. Yorum satırlarında neyin ne olduğunu açıkladım.
"""
Binance Spot için Limit Emir Grid Planlayıcısı (dry-run modu).
Grid Modeli
-----------
L0 < L1 < ... < Ln seviyeleri n adet [L_k, L_{k+1}] hücresi oluşturur.
Her hücrede aynı anda tam olarak bir aktif emir bulunur:
- L_k < fiyat -> L_k üzerinde BUY. Gerçekleştiğinde, L_{k+1} üzerinde SELL verilir
(temel para birimindeki komisyon düşüldükten kalan miktar ile).
- L_k >= fiyat -> L_{k+1} üzerinde SELL. Bunun için temel para biriminin başlangıç
rezervine ihtiyaç vardır (önceden market fiyatından alınmalıdır).
Mevcut fiyata eşit olan seviye SELL hücresine aittir (atlanmaz).
Bu betik borsaya ASLA hiçbir şey göndermez.
"""
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 Araçları
# --------------------------------------------------------------------------
def dec(value: Any, name: str) -> Decimal:
try:
result = Decimal(str(value))
except (InvalidOperation, ValueError, TypeError) as exc:
raise ValueError(f"{name}: geçersiz sayısal değer") from exc
if not result.is_finite():
raise ValueError(f"{name}: değer sonlu olmalıdır")
return result
def text(value: Decimal) -> str:
"""Decimal -> üslü gösterim ve gereksiz sıfırlar olmadan string."""
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 Enstrüman Kuralları (Filtreleri)
# --------------------------------------------------------------------------
@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]]
# Yüzde filtreleri için ortalama penceresi; 0 = son fiyat kullanılır.
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 filtrelerinde 0, 'sınırlama devre dışı' anlamına gelir."""
if value is None:
return None
result = dec(value, name)
return result if result > 0 else None
def load_rules(market: dict) -> SymbolRules:
"""Filtreleri bir kez okur. Buradaki herhangi bir sorun yapılandırma hatasıdır."""
info = market.get("info") or {}
status = info.get("status")
if status is not None and status != "TRADING":
raise RuntimeError(f"Parite durumu {status}, TRADING bekleniyordu")
if info.get("isSpotTradingAllowed") is False:
raise RuntimeError("Bu parite için spot işlem yapılması yasaklanmış")
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 bulunamadı")
if not lot_filter:
raise RuntimeError("Binance LOT_SIZE bulunamadı")
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 ve stepSize sıfırdan büyük olmalıdır")
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"Filtrelerde farklı avgPriceMins değerleri var: {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
# --------------------------------------------------------------------------
def align_bounds(
lower: Decimal, upper: Decimal, tick: Decimal
) -> tuple[Decimal, Decimal]:
"""Sınırlar, aralığın içine doğru hizalanır."""
lower = ceil_to_step(lower, tick)
upper = floor_to_step(upper, tick)
if lower >= upper:
raise ValueError("tickSize ile hizalandıktan sonra aralık boş kaldı")
return lower, upper
def calculate_grid_levels(
lower_bound: Decimal,
upper_bound: Decimal,
grid_count: int,
tick: Decimal,
) -> list[Decimal]:
"""Geometrik grid: grid_count aralık, grid_count + 1 seviye."""
if not isinstance(grid_count, int) or isinstance(grid_count, bool):
raise ValueError("grid_count tam sayı olmalıdır")
if grid_count < 1:
raise ValueError("grid_count en az 1 olmalıdır")
if min(lower_bound, upper_bound, tick) <= 0:
raise ValueError("Sınırlar ve tickSize sıfırdan büyük olmalıdır")
if lower_bound >= upper_bound:
raise ValueError("lower_bound, upper_bound değerinden küçük olmalıdır")
lower, upper = align_bounds(lower_bound, upper_bound, tick)
with localcontext() as ctx:
ctx.prec = 50
ratio = (upper / lower) ** (ONE / Decimal(grid_count))
# Her seviye alt sınırdan hesaplanır: hata birikmesi yaşanmaz.
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(
"Tike yuvarlama sonrasında seviyeler çakıştı veya sıra bozuldu. "
"grid_count değerini azaltın veya aralığı genişletin."
)
return levels
def check_grid_economics(
levels: list[Decimal],
fee_rate: Decimal,
extra_cost_rate: Decimal,
) -> None:
"""
Yaklaşık filtre: Her iki bacakta da komisyon hesaba katıldığında,
hücrenin net getirisi maliyet tamponunu aşmalıdır.
Bu bir kârlılık garantisi değildir.
"""
if fee_rate < 0 or extra_cost_rate < 0:
raise ValueError("Komisyon ve maliyet tamponu < 0 olamaz")
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 adımı çok küçük: {text(low)} -> {text(high)}; "
f"net getiri {net_return:.4%}, "
f"{extra_cost_rate:.4%}'den fazla olmalı"
)
# --------------------------------------------------------------------------
# Borsa Filtrelerine Göre Emir Doğrulama
# --------------------------------------------------------------------------
def validate_order(
rules: SymbolRules,
side: str,
price: Decimal,
amount: Decimal,
avg_price: Decimal | None = None,
) -> Decimal:
"""Emrin nominal maliyetini döndürür veya ValueError fırlatır."""
if side not in ("buy", "sell"):
raise ValueError("side 'buy' veya 'sell' olmalıdır")
if price <= 0 or amount <= 0:
raise ValueError(
f"{side.upper()} @ {text(price)}: fiyat ve miktar > 0 olmalıdır "
f"(amount={text(amount)})"
)
if price % rules.tick != 0:
raise ValueError(f"Fiyat {text(price)} tickSize katı değil")
if amount % rules.step != 0:
raise ValueError(f"Miktar {text(amount)} stepSize katı değil")
if price < rules.min_price:
raise ValueError(f"Fiyat {text(price)} < minPrice {text(rules.min_price)}")
if rules.max_price is not None and price > rules.max_price:
raise ValueError(f"Fiyat {text(price)} > maxPrice {text(rules.max_price)}")
if amount < rules.min_qty:
raise ValueError(
f"Miktar {text(amount)} < minQty {text(rules.min_qty)}"
)
if rules.max_qty is not None and amount > rules.max_qty:
raise ValueError(
f"Miktar {text(amount)} > maxQty {text(rules.max_qty)}"
)
cost = price * amount
if cost < rules.min_notional:
raise ValueError(
f"Nominal tutar {text(cost)} < minNotional {text(rules.min_notional)}"
)
if rules.max_notional is not None and cost > rules.max_notional:
raise ValueError(
f"Nominal tutar {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}: fiyat {text(price)} aralık dışında "
f"{text(minimum)}–{text(maximum)}"
)
return cost
# --------------------------------------------------------------------------
# Grid Planı
# --------------------------------------------------------------------------
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]]:
"""
Hücrelere göre emirleri oluşturur. (emirler, atlanan_hücreler) döndürür.
Temel para birimindeki komisyon, eşleşen SELL miktarını azaltır.
"""
if budget_per_level <= 0:
raise ValueError("Seviye başına bütçe sıfırdan büyük olmalıdır")
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 hücresi: low'dan al, high'da sat.
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)
# Eşleşen SELL daha sonra verilecek, yüzde filtresini
# zaten borsa doğrulayacaktır; bu yüzden burada avg_price aktarılmıyor.
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 hücresi: miktar, nominal tutarın seviye bütçesini aşmaması için
# doğrudan SELL fiyatından hesaplanır.
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"hücre {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:
"""
Sermaye gereksinimi tahmini. Bu, kayma (slippage) payı içeren bir TAHMİNDİR,
bir garanti değildir: ilk alımdaki gerçek fiyat ve komisyonlar farklılık gösterebilir.
"""
buys = [o for o in orders if o["side"] == "buy"]
sells = [o for o in orders if o["side"] == "sell"]
# BUY emrinin gerçekleşmesi en fazla notional kadar harcar: komisyon
# alınan varlıktan (BTC) veya BNB'den kesilir, USDT'den kesilmez.
quote_for_buys = sum((o["notional"] for o in buys), ZERO)
base_net_needed = sum((o["amount"] for o in sells), ZERO)
# Piyasa üzerinden başlangıç alımı (taker). Komisyon baz varlıkta ise,
# komisyon kesildikten sonra base_net_needed kalması için daha fazla alınmalıdır.
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 emirler (BUY, eşleşen SELL'ler ve başlangıç SELL'leri) maker üzerinden hesaplanır,
# başlangıç piyasa alımı ise taker üzerinden hesaplanır. Emiri deftere doğrudan
# yazmak yerine tahtayı delen limit emir taker gibi çalışır: bu tahmin bunu yakalamaz.
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,
}
# --------------------------------------------------------------------------
# Borsa Entegrasyonu
# --------------------------------------------------------------------------
def restrict_to_spot(exchange) -> None:
"""ccxt'den yalnızca spot piyasaları yüklemesini rica ediyoruz (seçenek formatı sürüme göre değişir)."""
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) döndürür."""
ticker = exchange.fetch_ticker(symbol)
if ticker.get("bid") is None or ticker.get("ask") is None:
raise RuntimeError(f"{symbol}: borsa bid/ask değerlerini döndürmedi")
bid = dec(ticker["bid"], "bid")
ask = dec(ticker["ask"], "ask")
if bid <= 0 or ask <= 0 or bid > ask:
raise RuntimeError(f"{symbol}: geçersiz 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* için referans fiyat. Binance kuralını taklit eder:
avgPriceMins == 0 -> son fiyat, aksi takdirde avgPriceMins dakikalık ortalama.
/avgPrice tarafından dönen pencere, filtredeki pencereyle karşılaştırılır.
Nihai kontrol emir gönderilirken borsa tarafından zaten yapılacaktır.
"""
if rules.avg_price_mins == 0:
if last is None:
raise RuntimeError("Filtre için last fiyatı gerekli, borsa bunu döndürmedi")
return last
response = exchange.publicGetAvgPrice({"symbol": market["id"]})
price = dec(response["price"], "avgPrice")
if price <= 0:
raise RuntimeError("Binance geçersiz bir avgPrice döndürdü")
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 penceresi ({mins} dk), filtrenin avgPriceMins "
f"değeriyle ({rules.avg_price_mins} dk) eşleşmiyor: kontrol doğrulanamıyor"
)
return price
def resolve_rate(
market: dict,
key: str,
explicit: Decimal | None,
fee_in_bnb: bool,
) -> Decimal:
"""
Komisyon oranı: açıkça belirtilmişse o, aksi takdirde piyasadan (maker/taker), yoksa %0.1.
BNB ile ödeme indirimi VARSAYILMAZ: hesabın gerçek oranı açıkça
iletilmelidir, aksi takdirde tahmin muhafazakardır (indirimsiz).
"""
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 oranı açıkça belirtilmedi: BNB indirimi olmaksızın %s kullanılıyor. "
"Gerçek hesap oranınızı --fee-rate / --taker-fee-rate "
"aracılığıyla iletin.",
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:
"""
Gelecekteki yürütme modülü için: GERÇEKLEŞEN bir BUY sonrasındaki
eşleşen SELL miktarı.
executed_qty — BRÜT satın alınan miktar (emrin executedQty değeri,
komisyon düşülmeden önce).
fills — ccxt formatında emir gerçekleşmeleri: her birinin
fill["fee"] = {"cost": ..., "currency": ...} veya None değeri vardır.
Yalnızca temel para biriminde kesilen komisyonlar düşülür: BNB veya
kotasyon para birimindeki komisyon varlık miktarını azaltmaz.
free_balance — diğer emirler için bloke edilmemiş, temel para biriminin
serbest bakiyesi (bunu yönetmek çağrı yapan kodun sorumluluğundadır).
"""
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("Komisyon düşüldükten sonra miktar pozitif değil")
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="Kotasyon para biriminde tek bir emrin maksimum nominal değeri",
)
parser.add_argument(
"--fee-rate", default=None,
help="Maker komisyonu (oran), varsayılan olarak piyasa maker oranı",
)
parser.add_argument(
"--taker-fee-rate", default=None,
help="Başlangıç alımı için taker komisyonu, varsayılan olarak piyasa taker oranı",
)
parser.add_argument(
"--fee-in-bnb", action="store_true",
help="Komisyon, alınan varlık yerine BNB ile ödenir",
)
parser.add_argument("--extra-cost-rate", default="0.0005")
parser.add_argument(
"--slippage", default="0.002",
help="Başlangıç alımı için kayma payı (oran)",
)
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) aralığında olmalıdır")
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"İşlem çifti bulunamadı: {symbol}")
market = exchange.market(symbol)
if market.get("spot") is False:
raise RuntimeError(f"{symbol} bir spot piyasa değil")
if market.get("active") is False:
raise RuntimeError(f"İşlem çifti aktif değil: {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"Mevcut fiyat {text(mid)} grid aralığının dışında "
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("Atlandı: %s", message)
if not orders:
raise RuntimeError("Hiçbir emir doğrulamadan geçemedi")
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 çifti bulunamadı: BNB maliyeti tahmin edilemiyor", bnb_symbol)
summary = summarize_plan(
orders, fee_rate, taker_rate, fee_in_base, ask, slippage, bnb_price
)
print(f"Parite: {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"hücre sayısı: {len(levels) - 1}")
print(f"Komisyon: maker {fee_rate:.4%}, taker {taker_rate:.4%} "
f"({'alınan varlıkta' if fee_in_base else 'BNB cinsinden'})")
if avg_price is not None:
print(f"Filtre referans fiyatı: {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" -> gerçekleştiğinde 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 emirleri: {summary['buy_orders']}, "
f"gereken ~{money(summary['quote_for_buys'])} {quote}")
print(f"SELL emirleri: {summary['sell_orders']}, "
f"gerekli rezerv {text(summary['base_needed_for_sells'])} {base}")
print(f"Başlangıç alımı ~{money(summary['initial_buy_cost_estimate'])} "
f"{quote} (ask üzerinden, taker komisyonu, kayma {slippage:.2%})")
print(f"Toplam ~{money(summary['total_quote_estimate'])} {quote} "
f"(tahmini, yeterli bakiye garantisi yoktur)")
if not fee_in_base:
line = (f"BNB komisyonu: ~{money(summary['bnb_fee_quote_equivalent'])} "
f"{quote} eşdeğeri")
if summary["bnb_needed"] is not None:
line += f", ~{summary['bnb_needed']:.4f} BNB"
print(line + " (tüm emirlerin tam yaşam döngüsü)")
print()
print("Dry-run modu: Borsaya hiç emir gönderilmedi.")
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("Borsa/ağ hatası: %s", exc)
return 1
except (ValueError, RuntimeError) as exc:
logger.error("Hesaplama hatası: %s", exc)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())Dosyayı kaydedip adlandırın, örneğin grid_planner.py. Çalıştırma: python grid_planner.py --grid-count 20 --budget-per-level 100 --fee-in-bnb --fee-rate 0.00075 --taker-fee-rate 0.00075
4. Borsaların Bahsetmediği Gizli Riskler ve Otomasyon Tuzakları
Borsa pazarlamacıları geçmişe dönük (backtest) verilerle botların harika getiri grafiklerini sergilemeye bayılır. Ancak gerçek dünyada algoritmik ticaret, bakiyenizi sıradan bir insan hatasından çok daha hızlı eritebilecek teknik engellerle karşı karşıya kalır.
- Yüksek Volatilitede Kayma (Slippage): Piyasa destek seviyesini kırdığında, tahta (order book) milisaniyeler içinde boşalır. Bot bir piyasa emri (market order) girmeye veya gerçekleştirmeye çalışır ancak işlem en kötü fiyattan eşleşir. Düşük hacimli altcoinlerde slippage, grid stratejisinin bir haftalık kârını tek bir mum darbesiyle yutabilir.
- Komisyon Toksisitesi (Maker vs Taker): Grid stratejisi limit emirler üzerine kuruludur — bu bir Maker rejimdir ve borsalar genellikle size indirim (rebate) sunar veya çok küçük bir oran alır. Ancak botun mantığı acil bir piyasa satışına (Stop-Loss veya kırılma anındaki panik) yönelirse, devreye giren astronomik Taker komisyonları stratejinin matematiğini tamamen yerle bir eder.
- Backtest Yanılgısı: Geçmiş veriler, gerçek ağ gecikmesini (borsa API'sine ping süresi) ve eşleştirme motoru (matching engine) sırasını hesaba katmaz. 2024–2025 dönemi geçmiş testlerinde %300 kâr gösteren bir bot, yatay piyasadaki canlı işlemlerde sadece bu yürütme gecikmeleri yüzünden zarara girebilir.
5. Sermayeyi Korumak: Botlar İçin Risk Yönetimi Kontrol Listesi
Algoritmayı çalıştırıp tatile çıkamazsınız. Otomasyon, sıkı bir izleme altyapısı gerektirir:
Güvenlik ve Kontrol Kuralları:
Toparlayacak olursak.
Bir alım satım botu, uçaktaki otopilot değil, bir cerrahın elindeki neşterdir. Disiplini otomatize eder ancak beynin yerini almaz. Likit enstrümanlar üzerinde spot grid ile küçük miktarlarla başlayın, mantığı gerçek gecikmelerle test edin ve unutmayın: kârın en büyük düşmanı piyasa değil, risk yönetimini hesaba katmayan kendi ayarlarınızdır.
EXMON Akademi Hatırlatıyor: Sağlam bir altyapı ve risk kontrolü her zaman önceliklidir. Asıl kazanan, algoritmayı ne zaman durduracağını iyi bilendir.
