ट्रेडिंग ऑटोमेशन
आइए बिना किसी भ्रम के ट्रेडिंग ऑटोमेशन की असली कार्यप्रणाली (mechanics) को समझें और यह पता लगाएं कि क्रिप्टो मार्केट में कौन से बॉट वाकई में प्रॉफिट देते हैं, कहां छिपे हुए जोखिम होते हैं, और उनके लिए लॉजिक कैसे लिखा जाता है।
1. स्ट्रेटेजी आर्किटेक्चर: प्रैक्टिकल में क्या काम करता है
मार्केट में केवल तीन मुख्य प्रकार के ऑटोमेटेड सिस्टम्स टिक पाते हैं। बाकी सब एक्सचेंज का सिर्फ मार्केटिंग ड्रामा है ताकि फीस बटोरी जा सके।
- ग्रिड ट्रेडिंग (Grid Trading): साइडवेज़ मार्केट (फ्लैट ट्रेंड) के लिए बेस्ट है। बॉट मौजूदा कीमत से नीचे लिमिट बाय ऑर्डर और ऊपर लिमिट सेल ऑर्डर की एक सीढ़ी लगा देता है। यह कॉरिडोर के अंदर होने वाले उतार-चढ़ाव से कमाई करता है।
- डॉलर-कॉस्ट एवरेजिंग (DCA बॉट्स): यह एक एक्यूमुलेशन (संचयन) स्ट्रेटेजी है। बॉट एक निश्चित समय अंतराल पर या कीमत में तय प्रतिशत की गिरावट आने पर बराबर हिस्सों में एसेट खरीदता है। लॉन्ग-टेर्म पोर्टफोलियो के लिए स्पॉट मार्केट पर काम करता है।
- आर्बिट्रेज स्ट्रैटेजीज़: अलग-अलग प्लेटफॉर्म के स्पॉट मार्केट के बीच कीमत के अंतर को भुनाना या फ्यूचर्स और स्पॉट के बीच के स्प्रेड (फंडिंग रेट आर्बिट्रेज) को पकड़ना। इसके लिए लो लेटेंसी (low latency) और ऑप्टिमाइज़्ड API की जरूरत होती है।
2. ग्रिड बॉट्स का गणित: क्यों 80% यूजर अपना पैसा डूबा देते हैं
ज़्यादातर लोग डिफ़ॉल्ट सेटिंग्स के साथ किसी वोलाटाइल altcoin के बिल्कुल टॉप (हाइपर-हाइ) पर ग्रिड बॉट चालू कर देते हैं। नतीजा हमेशा एक जैसा होता है: रेंज का नीचे की तरफ अचानक ब्रेकडाउन होता है, बॉट कौड़ियों के भाव पर सारी बेस करेंसी बेच देता है, और पोर्टफोलियो 60% के बड़े ड्राडाउन के साथ 'डेड' एसेट्स से भर जाता है।
इस स्थिति से बचने के लिए, पैरामीटर्स की गणना सख्त नियमों के आधार पर की जाती है:
ग्रिड कॉन्फ़िगरेशन के तरीकों की तुलना:
- एसेट का चयन: नौसिखिए की गलती — भयंकर वोलैलिटी वाले हाइपी memecoin को चुनना। प्रोफेशनल तरीका — हाई लिक्विडिटी और मजबूत सपोर्ट लेवल वाले टॉप-20 कॉइंस (BTC, ETH) को चुनना।
- ग्रिड का प्रकार: नौसिखिए की गलती — एरिथमेटिक ग्रिड (डॉलर में बराबर अंतर)। प्रोफेशनल तरीका — जियोमेट्रिक ग्रिड (बराबर प्रतिशत अंतर, जो यील्ड का एक स्थिर प्रतिशत बनाए रखने के लिए बेहद जरूरी है)।
- रेंज: नौसिखिए की गलती — मौजूदा कीमत से सिर्फ 5% का एक संकीर्ण गलियारा (corridor)। प्रोफेशनल तरीका — एक चौड़ी रेंज जो पिछले 3-6 महीनों के ऐतिहासिक फ्लैट ट्रेंड को कवर करे।
3. प्रैक्टिकल इम्प्लीमेंटेशन: ग्रिड ट्रेडिंग मॉड्यूल का कोड
नीचे ccxt लाइब्रेरी का उपयोग करके Python स्क्रिप्ट का एक काम करने वाला फ्रैगमेंट दिया गया है। यह जियोमेट्रिक स्टेप के साथ ग्रिड ऑर्डर जनरेट करने के लिए कैलकुलेशन लॉजिक का कोर है। मैंने कॉमेंट्स में समझाया है कि कौन सा हिस्सा किस लिए है।
"""
Binance Spot के लिए लिमिट ऑर्डर ग्रिड प्लानर (ड्राय-रन मोड)।
ग्रिड मॉडल
---------
लेवल्स L0 < L1 < ... < Ln मिलकर n सेल [L_k, L_{k+1}] बनाते हैं।
हर सेल में एक समय पर ठीक एक ही एक्टिव ऑर्डर रहता है:
- L_k < price -> L_k पर BUY। एग्जीक्यूट होने के बाद, L_{k+1} पर SELL प्लेस किया जाता है
बेस करेंसी में फीस काटकर बची हुई मात्रा के साथ।
- L_k >= price -> L_{k+1} पर SELL। इसके लिए बेस करेंसी के शुरुआती रिजर्व की
जरूरत होती है (जिसे पहले मार्केट प्राइस पर खरीदना पड़ता है)।
मौजूदा कीमत के बराबर वाला लेवल SELL सेल का हिस्सा होता है (इसे छोड़ा नहीं जाता)।
यह स्क्रिप्ट एक्सचेंज पर कुछ भी सेंड नहीं करती है।
"""
from __future__ import annotations
import argparse
import logging
import sys
from dataclasses import dataclass
from decimal import (
ROUND_CEILING,
ROUND_FLOOR,
ROUND_HALF_UP,
Decimal,
InvalidOperation,
localcontext,
)
from typing import Any
import ccxt
logging.basicConfig(
level=logging.INFO,
format="%(asctime)s [%(levelname)s] %(message)s",
)
logger = logging.getLogger("grid_planner")
ZERO = Decimal("0")
ONE = Decimal("1")
# --------------------------------------------------------------------------
# Decimal यूटिलिटीज
# --------------------------------------------------------------------------
def dec(value: Any, name: str) -> Decimal:
try:
result = Decimal(str(value))
except (InvalidOperation, ValueError, TypeError) as exc:
raise ValueError(f"{name}: अमान्य संख्यात्मक मान") from exc
if not result.is_finite():
raise ValueError(f"{name}: मान finite होना चाहिए")
return result
def text(value: Decimal) -> str:
"""Decimal -> एक्सपोनेंट और फालतू जीरोज़ के बिना स्ट्रिंग।"""
return format(value.normalize(), "f")
def floor_to_step(value: Decimal, step: Decimal) -> Decimal:
return (value / step).to_integral_value(rounding=ROUND_FLOOR) * step
def ceil_to_step(value: Decimal, step: Decimal) -> Decimal:
return (value / step).to_integral_value(rounding=ROUND_CEILING) * step
def round_to_step(value: Decimal, step: Decimal) -> Decimal:
return (value / step).to_integral_value(rounding=ROUND_HALF_UP) * step
# --------------------------------------------------------------------------
# Binance Spot इंस्ट्रूमेंट के नियम (फिल्टर)
# --------------------------------------------------------------------------
@dataclass(frozen=True)
class SymbolRules:
tick: Decimal
step: Decimal
min_price: Decimal
max_price: Decimal | None
min_qty: Decimal
max_qty: Decimal | None
min_notional: Decimal
max_notional: Decimal | None
percent_general: tuple[Decimal, Decimal] | None # (down, up)
percent_by_side: dict[str, tuple[Decimal, Decimal]]
# परसेंटेज फिल्टर के लिए एवरेजिंग विंडो; 0 = आखिरी कीमत का उपयोग किया जाता है।
avg_price_mins: int | None
@property
def has_percent_filters(self) -> bool:
return self.percent_general is not None or bool(self.percent_by_side)
def _positive_or_none(value: Any, name: str) -> Decimal | None:
"""Binance फिल्टर में 0 का मतलब है 'लिमिट डिसेबल है'."""
if value is None:
return None
result = dec(value, name)
return result if result > 0 else None
def load_rules(market: dict) -> SymbolRules:
"""फिल्टर को एक बार रीड करता है। यहाँ कोई भी समस्या कॉन्फ़िगरेशन एरर है।"""
info = market.get("info") or {}
status = info.get("status")
if status is not None and status != "TRADING":
raise RuntimeError(f"पैर का स्टेटस {status} है, TRADING अपेक्षित था")
if info.get("isSpotTradingAllowed") is False:
raise RuntimeError("इस पैर के लिए स्पॉट ट्रेडिंग प्रतिबंधित है")
filters = {
item.get("filterType"): item
for item in info.get("filters", [])
if item.get("filterType")
}
price_filter = filters.get("PRICE_FILTER")
lot_filter = filters.get("LOT_SIZE")
if not price_filter:
raise RuntimeError("Binance PRICE_FILTER नहीं मिला")
if not lot_filter:
raise RuntimeError("Binance LOT_SIZE नहीं मिला")
tick = dec(price_filter["tickSize"], "PRICE_FILTER.tickSize")
step = dec(lot_filter["stepSize"], "LOT_SIZE.stepSize")
if tick <= 0 or step <= 0:
raise RuntimeError("tickSize और stepSize शून्य से अधिक होने चाहिए")
min_notional = ZERO
max_notional = None
for filter_type in ("MIN_NOTIONAL", "NOTIONAL"):
item = filters.get(filter_type)
if not item:
continue
if item.get("minNotional") is not None:
min_notional = max(
min_notional,
dec(item["minNotional"], f"{filter_type}.minNotional"),
)
if item.get("maxNotional") is not None:
max_notional = _positive_or_none(
item["maxNotional"], f"{filter_type}.maxNotional"
)
percent_general = None
item = filters.get("PERCENT_PRICE")
if item:
percent_general = (
dec(item["multiplierDown"], "PERCENT_PRICE.multiplierDown"),
dec(item["multiplierUp"], "PERCENT_PRICE.multiplierUp"),
)
percent_by_side: dict[str, tuple[Decimal, Decimal]] = {}
item = filters.get("PERCENT_PRICE_BY_SIDE")
if item:
percent_by_side["buy"] = (
dec(item["bidMultiplierDown"], "bidMultiplierDown"),
dec(item["bidMultiplierUp"], "bidMultiplierUp"),
)
percent_by_side["sell"] = (
dec(item["askMultiplierDown"], "askMultiplierDown"),
dec(item["askMultiplierUp"], "askMultiplierUp"),
)
avg_mins_values = set()
for key in ("PERCENT_PRICE", "PERCENT_PRICE_BY_SIDE"):
item = filters.get(key)
if item and item.get("avgPriceMins") is not None:
avg_mins_values.add(int(dec(item["avgPriceMins"], "avgPriceMins")))
if len(avg_mins_values) > 1:
raise RuntimeError(
f"फिल्टर में अलग-अलग avgPriceMins हैं: {sorted(avg_mins_values)}"
)
avg_price_mins = avg_mins_values.pop() if avg_mins_values else None
return SymbolRules(
tick=tick,
step=step,
min_price=dec(price_filter.get("minPrice", "0"), "minPrice"),
max_price=_positive_or_none(price_filter.get("maxPrice"), "maxPrice"),
min_qty=dec(lot_filter.get("minQty", "0"), "LOT_SIZE.minQty"),
max_qty=_positive_or_none(lot_filter.get("maxQty"), "LOT_SIZE.maxQty"),
min_notional=min_notional,
max_notional=max_notional,
percent_general=percent_general,
percent_by_side=percent_by_side,
avg_price_mins=avg_price_mins,
)
# --------------------------------------------------------------------------
# ग्रिड
# --------------------------------------------------------------------------
def align_bounds(
lower: Decimal, upper: Decimal, tick: Decimal
) -> tuple[Decimal, Decimal]:
"""सीमाओं को रेंज के भीतर अलाइन किया जाता है।"""
lower = ceil_to_step(lower, tick)
upper = floor_to_step(upper, tick)
if lower >= upper:
raise ValueError("tickSize के साथ अलाइन करने के बाद रेंज खाली है")
return lower, upper
def calculate_grid_levels(
lower_bound: Decimal,
upper_bound: Decimal,
grid_count: int,
tick: Decimal,
) -> list[Decimal]:
"""ज्योमेट्रिक ग्रिड: grid_count अंतराल, grid_count + 1 लेवल्स।"""
if not isinstance(grid_count, int) or isinstance(grid_count, bool):
raise ValueError("grid_count एक पूर्णांक (integer) होना चाहिए")
if grid_count < 1:
raise ValueError("grid_count कम से कम 1 होना चाहिए")
if min(lower_bound, upper_bound, tick) <= 0:
raise ValueError("सीमाएं और tickSize शून्य से अधिक होने चाहिए")
if lower_bound >= upper_bound:
raise ValueError("lower_bound, upper_bound से कम होना चाहिए")
lower, upper = align_bounds(lower_bound, upper_bound, tick)
with localcontext() as ctx:
ctx.prec = 50
ratio = (upper / lower) ** (ONE / Decimal(grid_count))
# हर लेवल की गणना निचلی सीमा से की जाती है: एरर एक्यूमुलेट नहीं होती।
levels = [
round_to_step(lower * (ratio ** i), tick)
for i in range(grid_count + 1)
]
levels[0] = lower
levels[-1] = upper
if any(a >= b for a, b in zip(levels, levels[1:])):
raise ValueError(
"टिक पर राउंडिंग के बाद लेवल्स आपस में मिल गए या ऑर्डर टूट गया। "
"grid_count घटाएं या रेंज बढ़ाएं।"
)
return levels
def check_grid_economics(
levels: list[Decimal],
fee_rate: Decimal,
extra_cost_rate: Decimal,
) -> None:
"""
अप्रोक्सिमेट फिल्टर: दोनों लेग्स पर फीस को ध्यान में रखते हुए
सेल का नेट रिटर्न कॉस्ट बफर से अधिक होना चाहिए।
यह प्रॉफिटेबिलिटी की कोई गारंटी नहीं है।
"""
if fee_rate < 0 or extra_cost_rate < 0:
raise ValueError("फीस और कॉस्ट बफर शून्य से कम नहीं हो सकते")
keep = (ONE - fee_rate) ** 2
for low, high in zip(levels, levels[1:]):
net_return = (high / low) * keep - ONE
if net_return <= extra_cost_rate:
raise ValueError(
f"ग्रिड स्टेप बहुत छोटा है: {text(low)} -> {text(high)}; "
f"नेट रिटर्न {net_return:.4%}, "
f"{extra_cost_rate:.4%} से अधिक की आवश्यकता है"
)
# --------------------------------------------------------------------------
# एक्सचेंज फिल्टर के आधार पर ऑर्डर वैलिडेशन
# --------------------------------------------------------------------------
def validate_order(
rules: SymbolRules,
side: str,
price: Decimal,
amount: Decimal,
avg_price: Decimal | None = None,
) -> Decimal:
"""ऑर्डर की नोशनल कॉस्ट रिटर्न करता है या ValueError थ्रो करता है।"""
if side not in ("buy", "sell"):
raise ValueError("side, buy या sell होना चाहिए")
if price <= 0 or amount <= 0:
raise ValueError(
f"{side.upper()} @ {text(price)}: कीमत और मात्रा शून्य से अधिक होनी चाहिए "
f"(amount={text(amount)})"
)
if price % rules.tick != 0:
raise ValueError(f"कीमत {text(price)} tickSize का गुणज (multiple) नहीं है")
if amount % rules.step != 0:
raise ValueError(f"मात्रा {text(amount)} stepSize का गुणज (multiple) नहीं है")
if price < rules.min_price:
raise ValueError(f"कीमत {text(price)} < minPrice {text(rules.min_price)}")
if rules.max_price is not None and price > rules.max_price:
raise ValueError(f"कीमत {text(price)} > maxPrice {text(rules.max_price)}")
if amount < rules.min_qty:
raise ValueError(
f"मात्रा {text(amount)} < minQty {text(rules.min_qty)}"
)
if rules.max_qty is not None and amount > rules.max_qty:
raise ValueError(
f"मात्रा {text(amount)} > maxQty {text(rules.max_qty)}"
)
cost = price * amount
if cost < rules.min_notional:
raise ValueError(
f"नोशनल {text(cost)} < minNotional {text(rules.min_notional)}"
)
if rules.max_notional is not None and cost > rules.max_notional:
raise ValueError(
f"नोशनल {text(cost)} > maxNotional {text(rules.max_notional)}"
)
if avg_price is not None:
bounds = []
if rules.percent_general:
bounds.append(("PERCENT_PRICE", *rules.percent_general))
if side in rules.percent_by_side:
bounds.append(("PERCENT_PRICE_BY_SIDE", *rules.percent_by_side[side]))
for name, down, up in bounds:
minimum, maximum = avg_price * down, avg_price * up
if not minimum <= price <= maximum:
raise ValueError(
f"{name}: कीमत {text(price)} रेंज से बाहर है "
f"{text(minimum)}–{text(maximum)}"
)
return cost
# --------------------------------------------------------------------------
# ग्रिड प्लान
# --------------------------------------------------------------------------
def build_grid_plan(
levels: list[Decimal],
current_price: Decimal,
rules: SymbolRules,
budget_per_level: Decimal,
fee_rate: Decimal,
fee_in_base: bool,
avg_price: Decimal | None,
) -> tuple[list[dict], list[str]]:
"""
सेल के हिसाब से ऑर्डर बनाता है। (orders, skipped_cells) रिटर्न करता है।
बेस करेंसी में लगने वाली फीस पेयर किए गए SELL की मात्रा को कम करती है।
"""
if budget_per_level <= 0:
raise ValueError("प्रति लेवल बजट शून्य से अधिक होना चाहिए")
base_fee = fee_rate if fee_in_base else ZERO
orders: list[dict] = []
skipped: list[str] = []
for k, (low, high) in enumerate(zip(levels, levels[1:])):
try:
if low < current_price:
# BUY सेल: low पर खरीदें, high पर बेचें।
amount = floor_to_step(budget_per_level / low, rules.step)
cost = validate_order(rules, "buy", low, amount, avg_price)
sell_amount = floor_to_step(amount * (ONE - base_fee), rules.step)
# पेयर किया गया SELL बाद में प्लेस किया जाएगा, परसेंटेज फिल्टर को
# तब खुद एक्सचेंज वेरीफाई करेगा, इसलिए यहाँ avg_price पास नहीं किया जा रहा।
validate_order(rules, "sell", high, sell_amount)
orders.append({
"cell": k,
"side": "buy",
"price": low,
"amount": amount,
"notional": cost,
"on_fill_sell_price": high,
"on_fill_sell_amount": sell_amount,
})
else:
# SELL सेल: मात्रा की गणना खुद SELL की कीमत से की जाती है,
# ताकि नोशनल वैल्यू प्रति लेवल बजट से अधिक न हो।
amount = floor_to_step(budget_per_level / high, rules.step)
sell_cost = validate_order(
rules, "sell", high, amount, avg_price
)
orders.append({
"cell": k,
"side": "sell",
"price": high,
"amount": amount,
"notional": sell_cost,
})
except ValueError as exc:
skipped.append(f"सेल {k} [{text(low)}–{text(high)}]: {exc}")
return orders, skipped
def summarize_plan(
orders: list[dict],
fee_rate: Decimal,
taker_fee_rate: Decimal,
fee_in_base: bool,
ask: Decimal,
slippage: Decimal,
bnb_price: Decimal | None,
) -> dict:
"""
फंड्स की जरूरत का अनुमान। यह स्लिपेज बफर के साथ एक ESTIMATE है,
गारंटी नहीं: शुरुआती खरीदारी के समय वास्तविक कीमत और फीस अलग हो सकती हैं।
"""
buys = [o for o in orders if o["side"] == "buy"]
sells = [o for o in orders if o["side"] == "sell"]
# BUY ऑर्डर की खरीदारी में notional से ज़्यादा डेबिट नहीं होता: फीस
# मिलने वाले एसेट (BTC) या BNB से ली जाती है, USDT से नहीं।
quote_for_buys = sum((o["notional"] for o in buys), ZERO)
base_net_needed = sum((o["amount"] for o in sells), ZERO)
# शुरुआती मार्केट खरीदारी (taker)। बेस करेंसी में फीस होने पर अधिक
# खरीदना पड़ता है ताकि फीस कटने के बाद base_net_needed बचा रहे।
taker_base_fee = taker_fee_rate if fee_in_base else ZERO
base_gross = base_net_needed / (ONE - taker_base_fee)
initial_notional = base_gross * ask * (ONE + slippage)
bnb_fee_quote = ZERO
bnb_needed = None
if not fee_in_base:
# लिमिट ऑर्डर (BUY, पेयर किए गए और शुरुआती SELL) मेकर के हिसाब से गिने जाते हैं,
# शुरुआती मार्केट खरीदारी — टेकर के हिसाब से। जो लिमिट ऑर्डर तुरंत
# ऑर्डरबुक को पार करता है, वह टेकर के रूप में निष्पादित होगा: यह अनुमान इसे कैप्चर नहीं करता।
buy_leg = sum((o["notional"] for o in buys), ZERO)
paired_sell_leg = sum(
(o["on_fill_sell_price"] * o["on_fill_sell_amount"] for o in buys),
ZERO,
)
sell_leg = sum((o["notional"] for o in sells), ZERO)
bnb_fee_quote = (
(buy_leg + paired_sell_leg + sell_leg) * fee_rate
+ initial_notional * taker_fee_rate
)
if bnb_price is not None:
bnb_needed = bnb_fee_quote / bnb_price
return {
"buy_orders": len(buys),
"sell_orders": len(sells),
"quote_for_buys": quote_for_buys,
"base_needed_for_sells": base_net_needed,
"initial_buy_cost_estimate": initial_notional,
"total_quote_estimate": quote_for_buys + initial_notional,
"bnb_fee_quote_equivalent": bnb_fee_quote,
"bnb_needed": bnb_needed,
}
# --------------------------------------------------------------------------
# एक्सचेंज इंटीग्रेशन
# --------------------------------------------------------------------------
def restrict_to_spot(exchange) -> None:
"""ccxt से केवल स्पॉट मार्केट लोड करने का अनुरोध करते हैं (विकल्प का फॉर्मेट संस्करण पर निर्भर करता है)।"""
fetch_markets = exchange.options.get("fetchMarkets")
if isinstance(fetch_markets, list):
exchange.options["fetchMarkets"] = ["spot"]
elif isinstance(fetch_markets, dict) and "types" in fetch_markets:
exchange.options["fetchMarkets"] = {**fetch_markets, "types": ["spot"]}
def get_ticker_prices(
exchange, symbol: str
) -> tuple[Decimal, Decimal, Decimal | None]:
"""(bid, ask, last) रिटर्न करता है।"""
ticker = exchange.fetch_ticker(symbol)
if ticker.get("bid") is None or ticker.get("ask") is None:
raise RuntimeError(f"{symbol}: एक्सचेंज ने bid/ask रिटर्न नहीं किया")
bid = dec(ticker["bid"], "bid")
ask = dec(ticker["ask"], "ask")
if bid <= 0 or ask <= 0 or bid > ask:
raise RuntimeError(f"{symbol}: अमान्य bid/ask {bid}/{ask}")
last = ticker.get("last")
return bid, ask, (dec(last, "last") if last is not None else None)
def fetch_percent_reference_price(
exchange, market: dict, rules: SymbolRules, last: Decimal | None
) -> Decimal:
"""
PERCENT_PRICE* के लिए रेफरेंस प्राइस। Binance के नियम को दोहराता है:
avgPriceMins == 0 -> आखिरी कीमत, अन्यथा avgPriceMins मिनट का औसत।
/avgPrice द्वारा लौटाई गई विंडो को फिल्टर की विंडो से मिलाया जाता है।
अंतिम सत्यापन ऑर्डर सबमिट करते समय वैसे भी एक्सचेंज द्वारा किया जाता है।
"""
if rules.avg_price_mins == 0:
if last is None:
raise RuntimeError("फिल्टर के लिए last कीमत आवश्यक है, एक्सचेंज ने इसे रिटर्न नहीं किया")
return last
response = exchange.publicGetAvgPrice({"symbol": market["id"]})
price = dec(response["price"], "avgPrice")
if price <= 0:
raise RuntimeError("Binance ने अमान्य avgPrice रिटर्न किया")
mins = response.get("mins")
if (
rules.avg_price_mins is not None
and mins is not None
and int(mins) != rules.avg_price_mins
):
raise RuntimeError(
f"avgPrice विंडो ({mins} मिनट) फिल्टर के avgPriceMins "
f"({rules.avg_price_mins} मिनट) से मेल नहीं खाती: सत्यापन दोहराया नहीं जा सकता"
)
return price
def resolve_rate(
market: dict,
key: str,
explicit: Decimal | None,
fee_in_bnb: bool,
) -> Decimal:
"""
फीस रेट: स्पष्ट, अन्यथा मार्केट से (maker/taker), अन्यथा 0.1%।
BNB में भुगतान पर छूट का अनुमान नहीं लगाया गया है: अकाउंट की वास्तविक दर
स्पष्ट रूप से पास की जानी चाहिए, अन्यथा अनुमान कंजर्वेटिव (बिना छूट के) होता है।
"""
if explicit is not None:
return explicit
value = market.get(key)
rate = dec(value, key) if value is not None else Decimal("0.001")
if fee_in_bnb:
logger.warning(
"रेट %s स्पष्ट रूप से सेट नहीं किया गया है: BNB छूट के बिना %s का उपयोग किया जा रहा है। "
"कृपया अपने अकाउंट की वास्तविक दर "
"--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:
"""
भविष्य के एग्जीक्यूशन मॉड्यूल के लिए: वास्तविक BUY निष्पादन के बाद
पेयर किए गए SELL की मात्रा।
executed_qty — कुल खरीदी गई मात्रा (ऑर्डर की executedQty,
फीस काटने से पहले)।
fills — ccxt फॉर्मेट में ऑर्डर निष्पादन: प्रत्येक में
fill["fee"] = {"cost": ..., "currency": ...} या None होता है।
केवल बेस करेंसी में ली गई फीस घटाई जाती है: BNB या कोट करेंसी में
ली गई फीस एसेट की मात्रा को कम नहीं करती।
free_balance — बेस करेंसी का फ्री बैलेंस, जो अन्य ऑर्डर्स के लिए
रिजर्व नहीं है (इसके लिए कॉलिंग कोड जिम्मेदार है)।
"""
fee_in_base = ZERO
for fill in fills:
fee = fill.get("fee")
if (
fee
and fee.get("currency") == base_asset
and fee.get("cost") is not None
):
fee_in_base += dec(fee["cost"], "fee.cost")
net = executed_qty - fee_in_base
if net <= 0:
raise ValueError("फीस काटने के बाद मात्रा सकारात्मक नहीं है")
amount = floor_to_step(min(net, free_balance), rules.step)
validate_order(rules, "sell", price, amount)
return amount
# --------------------------------------------------------------------------
# मुख्य (main)
# --------------------------------------------------------------------------
def parse_args(argv: list[str] | None = None) -> argparse.Namespace:
parser = argparse.ArgumentParser(description=__doc__.split("\n\n")[0])
parser.add_argument("--symbol", default="BTC/USDT")
parser.add_argument("--lower", default="50000")
parser.add_argument("--upper", default="70000")
parser.add_argument("--grid-count", type=int, default=20)
parser.add_argument(
"--budget-per-level",
default="100",
help="कोट करेंसी में एक सिंगल ऑर्डर की अधिकतम नोशनल वैल्यू",
)
parser.add_argument(
"--fee-rate", default=None,
help="मेकर फीस रेट (फ्रैक्शन), डिफ़ॉल्ट मार्केट मेकर",
)
parser.add_argument(
"--taker-fee-rate", default=None,
help="शुरुआती खरीदारी के लिए टेकर फीस रेट, डिफ़ॉल्ट मार्केट टेकर",
)
parser.add_argument(
"--fee-in-bnb", action="store_true",
help="फीस का भुगतान मिलने वाले एसेट के बजाय BNB में किया जाता है",
)
parser.add_argument("--extra-cost-rate", default="0.0005")
parser.add_argument(
"--slippage", default="0.002",
help="शुरुआती खरीदारी के लिए स्लिपेज बफर (फ्रैक्शन)",
)
return parser.parse_args(argv)
def optional_rate(value: str | None, name: str) -> Decimal | None:
if value is None:
return None
result = dec(value, name)
if not ZERO <= result < ONE:
raise ValueError(f"{name} रेंज [0, 1) के भीतर होना चाहिए")
return result
def run(args: argparse.Namespace) -> None:
symbol = args.symbol
lower_bound = dec(args.lower, "lower")
upper_bound = dec(args.upper, "upper")
budget = dec(args.budget_per_level, "budget_per_level")
extra_cost_rate = dec(args.extra_cost_rate, "extra_cost_rate")
slippage = optional_rate(args.slippage, "slippage")
explicit_fee = optional_rate(args.fee_rate, "fee_rate")
explicit_taker = optional_rate(args.taker_fee_rate, "taker_fee_rate")
exchange = ccxt.binance({
"enableRateLimit": True,
"timeout": 15000,
"options": {"defaultType": "spot"},
})
def money(value: Decimal) -> str:
return text(value.quantize(Decimal("0.01")))
try:
restrict_to_spot(exchange)
exchange.load_markets()
if symbol not in exchange.markets:
raise ValueError(f"ट्रेडिंग पैर नहीं मिला: {symbol}")
market = exchange.market(symbol)
if market.get("spot") is False:
raise RuntimeError(f"{symbol} स्पॉट मार्केट नहीं है")
if market.get("active") is False:
raise RuntimeError(f"ट्रेडिंग पैर सक्रिय नहीं है: {symbol}")
rules = load_rules(market)
fee_in_base = not args.fee_in_bnb
fee_rate = resolve_rate(market, "maker", explicit_fee, args.fee_in_bnb)
taker_rate = resolve_rate(
market, "taker", explicit_taker, args.fee_in_bnb
)
bid, ask, last = get_ticker_prices(exchange, symbol)
mid = (bid + ask) / Decimal("2")
levels = calculate_grid_levels(
lower_bound, upper_bound, args.grid_count, rules.tick
)
if not levels[0] <= mid <= levels[-1]:
raise ValueError(
f"मौजूदा कीमत {text(mid)} ग्रिड रेंज से बाहर है "
f"{text(levels[0])}–{text(levels[-1])}"
)
check_grid_economics(levels, fee_rate, extra_cost_rate)
avg_price = (
fetch_percent_reference_price(exchange, market, rules, last)
if rules.has_percent_filters
else None
)
orders, skipped = build_grid_plan(
levels=levels,
current_price=mid,
rules=rules,
budget_per_level=budget,
fee_rate=fee_rate,
fee_in_base=fee_in_base,
avg_price=avg_price,
)
for message in skipped:
logger.warning("छोड़ा गया: %s", message)
if not orders:
raise RuntimeError("कोई भी ऑर्डर वैलिडेशन पास नहीं कर पाया")
bnb_price = None
if args.fee_in_bnb:
bnb_symbol = f"BNB/{market['quote']}"
if bnb_symbol in exchange.markets:
b, a, _ = get_ticker_prices(exchange, bnb_symbol)
bnb_price = (b + a) / Decimal("2")
else:
logger.warning("पैर %s नहीं मिला: BNB लागत का अनुमान नहीं लगाया जा सकता", bnb_symbol)
summary = summarize_plan(
orders, fee_rate, taker_rate, fee_in_base, ask, slippage, bnb_price
)
print(f"पैर: {symbol}")
print(f"Bid / Ask: {text(bid)} / {text(ask)} (mid {text(mid)})")
print(f"Tick / Step: {text(rules.tick)} / {text(rules.step)}")
print(f"ग्रिड: {text(levels[0])} – {text(levels[-1])}, "
f"सेल: {len(levels) - 1}")
print(f"फीस रेट: maker {fee_rate:.4%}, taker {taker_rate:.4%} "
f"({'मिलने वाले एसेट में' if fee_in_base else 'BNB में'})")
if avg_price is not None:
print(f"फिल्टर रेफरेंस प्राइस: {text(avg_price)} "
f"(avgPriceMins={rules.avg_price_mins})")
print()
for order in orders:
line = (
f"{order['side'].upper():4} {text(order['amount'])} "
f"@ {text(order['price'])} (~{money(order['notional'])})"
)
if order["side"] == "buy":
line += (
f" -> एग्जीक्यूशन पर SELL "
f"{text(order['on_fill_sell_amount'])} "
f"@ {text(order['on_fill_sell_price'])}"
)
print(line)
quote, base = market["quote"], market["base"]
print()
print(f"BUY ऑर्डर: {summary['buy_orders']}, "
f"आवश्यकता ~{money(summary['quote_for_buys'])} {quote}")
print(f"SELL ऑर्डर: {summary['sell_orders']}, "
f"आवश्यक रिजर्व {text(summary['base_needed_for_sells'])} {base}")
print(f"शुरुआती खरीदारी ~{money(summary['initial_buy_cost_estimate'])} "
f"{quote} (ask पर, taker फीस, स्लिपेज {slippage:.2%})")
print(f"कुल ~{money(summary['total_quote_estimate'])} {quote} "
f"(अनुमान, पर्याप्त बैलेंस की कोई गारंटी नहीं)")
if not fee_in_base:
line = (f"BNB फीस: ~{money(summary['bnb_fee_quote_equivalent'])} "
f"{quote} समतुल्य")
if summary["bnb_needed"] is not None:
line += f", ~{summary['bnb_needed']:.4f} BNB"
print(line + " (सभी ऑर्डर्स का पूर्ण चक्र)")
print()
print("ड्राय-रन मोड: एक्सचेंज पर कोई ऑर्डर नहीं भेजे गए।")
finally:
exchange.close()
def main(argv: list[str] | None = None) -> int:
args = parse_args(argv)
try:
run(args)
except ccxt.BaseError as exc:
logger.error("एक्सचेंज/नेटवर्क एरर: %s", exc)
return 1
except (ValueError, RuntimeError) as exc:
logger.error("कैलकुलेशन एरर: %s", exc)
return 1
return 0
if __name__ == "__main__":
sys.exit(main())फ़ाइल को सेव करें और नाम दें, उदाहरण के लिए grid_planner.py। चलाने के लिए: python grid_planner.py --grid-count 20 --budget-per-level 100 --fee-in-bnb --fee-rate 0.00075 --taker-fee-rate 0.00075
4. ऑटोमेशन के छिपे हुए जोखिम और जाल, जिनके बारे में एक्सचेंज खामोश रहते हैं
एक्सचेंज के मार्केटर्स बैकटेस्ट के ऐतिहासिक डेटा पर बोट्स के आकर्षक रिटर्न ग्राफ दिखाना खूब पसंद करते हैं। लेकिन हकीकत में, एल्गो ट्रेडिंग को ऐसी तकनीकी बाधाओं का सामना करना पड़ता है जो किसी इंसानी गलती से भी तेजी से आपका पूरा कैपिटल साफ कर सकती हैं।
- हाई वोलेटिलिटी में स्लिपेज (Slippage): जब मार्केट सपोर्ट लेवल को तोड़ता है, तो ऑर्डर बुक पलक झपकते ही खाली हो जाती है। बोट मार्केट ऑर्डर प्लेस या एक्जीक्यूट करने की कोशिश करता है, लेकिन ट्रेड सबसे खराब प्राइस पर फिल होती है। कम लिक्विडिटी वाले ऑल्टकॉइन्स में, स्लिपेज एक ही झटके में आपकी पूरे हफ्ते की ग्रिड प्रॉफिट को खा सकती है।
- फीस का टॉक्सिक असर (Maker vs Taker): ग्रिड स्ट्रैटेजी लिमिट ऑर्डर्स पर टिकी होती है — यह Maker मोड है, जहाँ एक्सचेंज अक्सर आपको रिबेट देते हैं या मामूली फीस लेते हैं। लेकिन अगर बोट का लॉजिक पैनिक होकर इमर्जेंसी मार्केट सेल (Stop-Loss या ब्रेकडाउन पर पैनिक) करने लगता है, तो भारी-भरकम Taker फीस लागू हो जाती है, जो आपकी पूरी स्ट्रैटेजी के गणित को बिगाड़ देती है।
- बैकटेस्ट का भ्रम: हिस्टोरिकल डेटा कभी भी असली नेटवर्क लेटेंसी (एक्सचेंज API तक का पिंग) और मैचिंग इंजन (matching engine) की कतारों को ध्यान में नहीं रखता। जिस बोट ने 2024–2025 के बैकटेस्ट में +300% का रिटर्न दिखाया था, वह लाइव मार्केट के साइडवेज़ फेज में सिर्फ एक्जीक्यूशन डिले की वजह से नुकसान में जा सकता है।
5. अपनी पूंजी की सुरक्षा कैसे करें: बोट्स के लिए रिस्क-मैनेजमेंट चेकलिस्ट
आप बस एल्गोरिदम चालू करके छुट्टियों पर नहीं जा सकते। ऑटोमेशन के लिए एक मजबूत मॉनिटरिंग इंफ्रास्ट्रक्चर की जरूरत होती है:
सुरक्षा और कंट्रोल के नियम:
- कड़ा Kill-Switch: कोड या टर्मिनल सेटिंग्स में बोट के पूरे डिपॉजिट पर एक सख्त स्टॉप-लॉस सेट होना चाहिए। अगर बेस और कोटेड करेंसी की कुल वैल्यू मान लीजिए 15% गिर जाती है, तो बोट तुरंत सारे लिमिट ऑर्डर्स कैंसिल कर दे और लॉस बुक कर ले।
- फंड्स का आइसोलेशन: बोट की ट्रेडिंग कैपिटल को कभी भी अपनी मुख्य लॉन्ग-टर्म सेविंग्स वाले सब-अकाउंट में न रखें। आइसोलेटेड API-कीज़ का इस्तेमाल करें जिनमें विड्रॉल का ऑप्शन डिसेबल हो — सिर्फ ट्रेडिंग की अनुमति होनी चाहिए।
- Telegram अलर्ट्स के जरिए मॉनिटरिंग: जरूरी नोटिफिकेशन्स को एक्सचेंज के इंटरफेस के बजाय सीधे मैसेजिंग ऐप पर सेट करें। हर कैंसिल हुआ सेशन, API एरर या एक्जीक्यूट हुआ इमर्जेंसी ऑर्डर तुरंत रिकॉर्ड और नोटिफाई होना चाहिए।
चलिए, अब निष्कर्ष पर आते हैं।
ट्रेडिंग बोट सर्जन के हाथ में सर्जिकल ब्लेड की तरह है, न कि प्लेन का ऑटोपायलट। यह अनुशासन को ऑटोमेट करता है, लेकिन दिमाग की जगह नहीं ले सकता। लिक्विड एसेट्स पर स्पॉट ग्रिड से छोटी रकम के साथ शुरुआत करें, असली लेटेंसी पर अपने लॉजिक का टेस्ट करें और याद रखें: प्रॉफिट का असली दुश्मन मार्केट नहीं, बल्कि रिस्क मैनेजमेंट के बिना की गई आपकी अपनी सेटिंग्स हैं।
EXMON अकादमी याद दिलाती है: मजबूत इंफ्रास्ट्रक्चर और रिस्क कंट्रोल हमेशा सबसे पहले आते हैं। मुनाफा वही कमाता है जो सही समय पर एल्गोरिदम को रोकना जानता है।
