كيف تستخدم بوتات التداول لتحقيق أرباح الكريبتو؟

أتمتة التداول

دعونا نفكك ميكانيكا أتمتة التداول بدون أي أوهام، ونكتشف أنواع البوتات التي تحقق أرباحاً حقيقية في سوق العملات الرقمية، وأين تكمن المخاطر الخفية، وكيف نكتب المنطق البرمجي الخاص بها.

1. هندسة الاستراتيجيات: ما الذي يعمل حقاً على أرض الواقع

ثلاثة أنواع أساسية فقط من الأنظمة الآلية هي التي تنجو في السوق. والباقي مجرد تسويق من المنصات لجمع رسوم التداول.

• شبكة التداول (Grid Trading): مثالية للأسواق الجانبية (نطاقات التذبذب). يضع البوت سلماً من أوامر الشراء المحددة دون السعر الحالي وأوامر البيع فوقه، ليحقق أرباحاً من التذبذبات داخل النطاق.
• معدلة تكلفة الدولار (بوتات DCA): استراتيجية تراكم. يقوم البوت بشراء الأصل بأجزاء متساوية على فترات زمنية ثابتة أو عند انخفاض السعر بنسبة محددة. تعمل على السوق الفوري (Spot) للمحافظ طويلة الأجل.
• استراتيجيات المراجحة (Arbitrage): اقتناص فروق الأسعار بين الأسواق الفورية في منصات مختلفة أو السبريد بين العقود الآجلة والسوق الفوري (funding rate arbitrage). تتطلب تأخيراً منخفضاً (low latency) وواجهات برمجة تطبيقات (API) محسّنة.

2. رياضيات بوتات الشبكة (Grid Bots): لماذا يخسر 80% من المستخدمين ودائعهم

معظم المستخدمين يشغلون بوت الشبكة عند قمم عملة بديلة متقلبة باستخدام الإعدادات الافتراضية. النتيجة دائماً واحدة: كسر حاد للنطاق نحو الهبوط، ويقوم البوت ببيع كامل العملة الأساسية بأسعار زهيدة، بينما يمتلئ المحفظة بأصل "ميّت" بخسارة عائمة تصل إلى 60%.

لتجنب ذلك، يتم حساب المعلمات بناءً على قواعد صارمة:

مقارنة نهج إعداد الشبكة:

• اختيار الأصل: خطأ المبتدئ — عملة ميم (memecoin) ترند ذات تقلبات جنونية. النهج الاحترافي — عملات ضمن قائمة الـ 20 الأوائل ذات السيولة العالية ومستويات الدعم القوية (مثل BTC و ETH).
• نوع الشبكة: خطأ المبتدئ — حسابية (خطوات متساوية بالدولار). النهج الاحترافي — هندسية (خطوات نسبية متساوية، وهو أمر بالغ الأهمية للحفاظ على نسبة عوائد مستقرة).
• النطاق (Range): خطأ المبتدئ — ممر ضيق بنسبة 5% من السعر الحالي. النهج الاحترافي — نطاق واسع يغطي الحركة الجانبية التاريخية لآخر 3 إلى 6 أشهر.

3. التطبيق العملي: كود وحدة شبكة التداول

فيما يلي جزء برمجي لسكريبت بايثون يعمل بكفاءة باستخدام مكتبة ccxt. هذا هو جوهر منطق الحساب لتوليد أوامر الشبكة بخطوات هندسية. لقد قمت بشرح دور كل جزء في التعليقات البرمجية.

"""
مخطط شبكة أوامر الشراء/البيع المحددة لـ Binance Spot (وضع التجربة Dry-run).
نموذج الشبكة
------------
المستويات L0 < L1 < ... < Ln تشكل n خلايا [L_k, L_{k+1}].
في كل خلية يوجد أمر واحد نشط في نفس الوقت:
- L_k <  السعر  -> BUY عند L_k. بعد التنفيذ، يتم وضع SELL عند L_{k+1}
                 بالكمية الناتجة بعد خصم الرسوم بالعملة الأساسية.
- L_k >= السعر -> SELL عند L_{k+1}. يتطلب ذلك مخزونًا أوليًا من العملة الأساسية
                 (يجب شراؤها من السوق مسبقًا).
المستوى المساوئ للسعر الحالي ينتمي إلى خلية 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("التداول الفوري (Spot) غير مسموح به لهذا الزوج")
    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("مرشح السعر PRICE_FILTER لـ Binance غير موجود")
    if not lot_filter:
        raise RuntimeError("حجم اللوت LOT_SIZE لـ 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 و 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,
    )
# --------------------------------------------------------------------------
# الشبكة (Grid)
# --------------------------------------------------------------------------
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 عددًا صحيحًا")
    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(
            "تطابقت المستويات أو اختل ترتيبها بعد التقريب إلى الـ tick. "
            "قلل 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("لا يمكن أن تكون الرسوم وهامش التكاليف أقل من 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"خطوة الشبكة صغيرة جداً: {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)}: يجب أن يكون السعر والكمية أكبر من 0 "
            f"(amount={text(amount)})"
        )
    if price % rules.tick != 0:
        raise ValueError(f"السعر {text(price)} ليس مضاعفًا لـ tickSize")
    if amount % rules.step != 0:
        raise ValueError(f"الكمية {text(amount)} ليست مضاعفة لـ stepSize")
    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]]:
    """
    يبني الأوامر لكل خلية. يُرجع (الأوامر، الخلايا المستبعدة).
    الرسوم بالعملة الأساسية تقلل كمية أمر 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:
    """
    تقدير رأس المال المطلوب. هذا تقدير مع هامش للانزلاق السعري (slippage)،
    وليس ضمانًا: السعر الفعلي والرسوم عند الشراء الأولي قد تختلف.
    """
    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 المقترنة والأولية) تُحتسب كـ maker،
        # والشراء الأولي بسعر السوق كـ taker. الأمر المحدد الذي يعبر
        # دفتر الطلبات فورًا سيتم تنفيذه كـ taker: هذا التقدير لا يرصد ذلك.
        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 تحميل الأسواق الفورية (spot) فقط (تنسيق الخيار يعتمد على الإصدار)."""
    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 غير محدد صراحةً: يتم استخدام %s بدون خصم BNB. "
            "يرجى تمرير معدل الحساب الفعلي عبر "
            "--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:
    """
    لوحدة التنفيذ المستقبلية: كمية SELL المقترنة بعد
    التنفيذ الفعلي لـ BUY.
    executed_qty — الكمية الإجمالية المشتراة (executedQty للأمر،
                 قبل خصم الرسوم).
    fills        — تنفيذات الأمر بتنسيق ccxt: كل fill لديه
                   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="رسوم الـ Maker (كنسبة)، الافتراضي هو الـ maker من السوق",
    )
    parser.add_argument(
        "--taker-fee-rate", default=None,
        help="رسوم الـ Taker للشراء الأولي، الافتراضي هو الـ taker من السوق",
    )
    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} ليس سوق تداول فوري (spot)")
        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("وضع التجربة (dry-run): لم يتم إرسال أي أوامر إلى المنصة.")
    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. المخاطر الخفية وفخاخ الأتمتة التي تسكت عنها المنصات

يحب مسوقو المنصات استعراض رسوم بيانية جذابة لأرباح البوتات بناءً على الاختبارات التاريخية (Backtesting). ولكن في الواقع، يصطدم التداول الخوارزمي بعقبات تقنية قادرة على تصفير محفظتك أسرع بكثير من أي خطأ بشري.

• الانزلاق السعري (Slippage) أثناء التقلبات العالية: عندما يكسر السوق مستوى دعم قوي، يفرغ دفتر الطلبات (Order Book) في أجزاء من الثانية. يحاول البوت إرسال أو تنفيذ أمر بسعر السوق، فيتم تنفيذ الصفقة بأسوأ سعر ممكن. في عملات البديلة (Altcoins) ذات السيولة الضعيفة، يمكن للانزلاق السعري أن يلتهم أرباح أسبوع كامل من الشبكة خلال شمعة واحدة.
• سميّة الرسوم (Maker vs Taker): تعتمد استراتيجية الشبكة على الأوامر المحددة (Limit Orders) — وهذا هو وضع الـ Maker، حيث تمنحك المنصات غالبًا مكافآت (Rebates) أو تخصم نسبة ضئيلة. ولكن إذا تحولت منطقية البوت إلى بيع اضطراري بسعر السوق (Stop-Loss أو هلع عند الكسر)، تتفعل رسوم الـ Taker الباهظة التي تقضي تمامًا على الحسابات الرياضية للاستراتيجية.
• وهم الاختبارات التاريخية: البيانات التاريخية لا تأخذ في الحسبان تأخير الشبكة الحقيقي (Ping مع API المنصة) ولا طوابير معالجة الأوامر (Matching Engine). البوت الذي حقق +300% في اختبارات تاريخية لعامي 2024–2025 قد يتحول إلى خسارة في التداول الحقيقي أثناء الحركة العرضية لمجرد التأخير في التنفيذ.

5. كيف تحمي رأس مالك: قائمة مراجعة إدارة المخاطر للبوتات

لا يمكنك تشغيل الخوارزمية والذهاب في إجازة. تتطلب الأتمتة البنية التحتية الصارمة للمراقبة:

قواعد الأمان والتحكم:

• مفتاح الإيقاف الصارم (Kill-Switch): يجب أن يتضمن الكود أو إعدادات الشاشة إيقاف خسارة مطلقًا على كامل رأس مال البوت. إذا انخفضت القيمة الإجمالية للعملة الأساسية والمقابلة بنسبة 15% مثلاً، يلغي البوت جميع الأوامر المحددة ويثبت الخسارة فورًا.
• عزل رأس المال: لا تحتفظ أبدًا برأس مال التداول الخاص بالبوت في نفس الحساب الفرعي الذي تحتوي فيه على مدخراتك الرئيسية طويلة الأجل. استخدم مفاتيح API معزولة مع تعطيل خاصية السحب — يُسمح بالتداول فقط.
• المراقبة عبر تنبيهات Telegram: قم بإعداد إرسال التنبيهات الحرجة مباشرة إلى تطبيق المراسلة بدلاً من الاعتماد على واجهة المنصة. كل جلسة ملغاة، أو خطأ في الـ API، أو أمر طوارئ تم تنفيذه يجب أن يُسجل فورًا.

في النهاية، دعنا نلخص الأمر.

بوت التداول هو مشرط في يد جراح، وليس الطيار الآلي في الطائرة. فهو يؤتمت الانضباط، ولكنه لا يحل محل العقل والخبرة. ابدأ بمبالغ صغيرة على شبكات التداول الفوري (Spot Grids) في الأصول ذات السيولة العالية، واختبر المنطق مع وجود التأخير الحقيقي، وتذكر دائمًا: العدو الأول لأرباحك ليس السوق، بل إعداداتك الخاصة التي تتجاهل إدارة المخاطر.

تذكرك أكاديمية EXMON: البنية التحتية الموثوقة وإدارة المخاطر تأتي دائمًا في المقام الأول. والرابح الحقيقي هو من يعرف متى يوقف الخوارزمية في الوقت المناسب.