Order Book مقابل AMM: البنية التحتية للسوق والتحكيم HFT

إذا كنت لغاية الحين تعتقد أن أي صفقة في المنصة هي مجرد ضغطة زر "Buy" وتصير المعجزة، فخاطئ جداً. على مستوى الميكروثانية والميليثانية، السوق عبارة عن ساحة حرب طاحنة، السيولة فيها تنتقل من حالة لحالة بلمح البصر، وسرعة التنفيذ مع تشريح المسبح (Pool) هم اللي يحددون مين يخرج بالربح ومين يأكل الانزلاق السعري (slippage) والتفق المسموم (Toxic Flow).

في التداول المركزي (TradFi و CEX)، الترتيب لدفتر الأوامر (Order Book). بينما في الـ DeFi، أجمع الكل على الـ AMM كمعيار أساسي. بس لما ننزل لمستوى التداول عالي التردد (HFT)، الفرق بينهم ما يصير مجرد تقنية وخلاص، هذا يغير ميكانيكية تسعير السوق والإنعاش (Arbitrage) وتنفيذ الأوامر جذرياً.

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

1. تشريح دفتر الأوامر L2/L3: كيف شكل السيولة الحقيقي

معظم المتداولين الأفراد يشوفون دفتر L2 فقط: أحجام مجمعة عند كل مستوى سعر. بس خوارزميات الـ HFT والمؤسسات الكبيرة تلعب في الـ L3 (Order-by-Order) — تدفق بيانات خام، كل حدث فيه له معرف خاص (order_id)، وختم زمني فائق الدقة، وموقع محدد في الطابور.

مبدأ أولوية السعر والزمن (Price-Time Priority - FIFO)

في محرك المطابقة (Matching Engine) الكلاسيكي، التنفيذ يمشي بنظام صارم:

• السعر (أفضل سعر له الأولوية دائماً).
• الزمن (إذا تساوى السعر، اللي دخل الطابور أول ينفذ أول).

طابور الانتظار هذا يخلق أثرين حاسمين في بنية السوق الدقيقة:

Adverse Selection (الانتخاب المعاكس): إذا تنفذ أمر الحد (Limit Order) الخاص بك للشراء بسرعة قياسية، غالباً في بائع كبير دخل الماركت بتدفق عدائي (Toxic Flow)، والسعر بيكمل هبوط.

Queue Position Value: وجودك في مقدمة الطابور عند مستوى سعر معين له قيمة مالية قائمة بذاتها. عشان كذا بوتات الـ HFT تلغي الأوامر وترجع تحطها بشكل مستمر (تكتيكات الـ Spoofing/Layering أو إدارة السيولة التكيفية) بس عشان تحافظ على الأولية بدون ما تتحمل مخاطر مخزون مالية زائدة.

2. الـ AMM: من الـ Constant Product الكلاسيكي إلى السيولة المركزة

الـ AMM يختلف تماماً عن دفتر الأوامر: ما في أي طابور. السعر يتحدد بدالة رياضية ثابتة.

Constant Product (v2): x · y = k

في Uniswap v2 التقليدي، السيولة موزعة من الصفر إلى المالانهاية.

لما يسوي المتداول صفقة بحجم Δx، هو يسحب من المسبح كمية Δy، محسوبة بعد خصم الرسوم γ = (1 - fee):

Δy = (y · Δx · γ) / (x + Δx · γ)

سعر التنفيذ مو ثابت؛ هو ينزلق على طول المنحنى في لحظة التنفيذ الفعلية.

Concentrated Liquidity (v3 / قائمة على الـ Ticks)

Uniswap v3 قرب الـ AMM لفكره دفتر الأوامر لما سمح لمزودي السيولة (LPs) يحددون نطاقات سعرية خاصة [P_a, P_b].

السعر مقسم إلى تيكات (Ticks)، وكل تغيير في التيك يمثل حركة سعر بـ 0.01% (1 bps):

P(i) = 1.0001ⁱ

السيولة L داخل النطاق تشتغل كثابت "افتراضي"، بس لما السعر يتجاوز حد التيك، السيولة النشطة تتغير فجأة. بالنسبة للمراجح (Arbitrageur) في الـ HFT، هذا يعني أن دفتر AMM v3 منفصل وعمقه يتغير على شكل درجات كل ما انكسرت التيكات.

3. التحليل المقارن: مصفوفة البنية الدقيقة (Microstructure Matrix)

4. الميكانيكيات الخفية: MEV و LVR التدفق المسموم

في دفتر CEX، جماعة الـ HFT يتصارعون على أجزاء النانوثانية من التأخير الشبكي الفيزيائي (سيرفرات Colocation بنفس مركز بيانات المنصة، ألياف ضوئية، كروت FPGA).

أما في الـ DeFi، مفهوم الوقت يتغير: اللعبة كلها تعتمد على مين ينسق ويبني البلوك.

MEV (القيمة القصوى المستخرجة - Maximal Extractable Value)

في الـ AMM، ترتيب المعاملات داخل البلوك يتحدد عن طريق الـ Searchers والـ Builders. هذا يخلق تكتيكات HFT خاصة ببيئة الكريبتو:

• Front-running / Back-running: خطف الصفقات من الـ Mempool عبر دفع رسوم أولوية أعلى (Priority Fee).
• Sandwich Attacks: الشراء المباشر قبل صفقة ضخمة لمستخدم عادي لرفع السعر، ثم البيع على رأسه فوراً بنفس البلوك.

LVR (Loss-Versus-Rebalancing)

الجميع متعود يحسب مخاطر الـ LP عن طريق الخسارة غير الدائمة (IL). بس على مستوى الـ HFT، المقياس الأهم هو الـ LVR، وهو مفهوم طوّره باحثون من جامعة كولومبيا و Paradigm.

LVR (Loss-Versus-Rebalancing) هو الخسارة المنهجية التي يتحملها مزود السيولة في الـ AMM مقارنة بمحفظة مطابقة يتم إعادة توازنها باستمرار في CEX بناءً على الأسعار الخارجية.

السبب الرئيسي لـ LVR هو مراجحة التقلبات (Volatility Arbitrage). لما يتغير السعر في CEX، مسبح الـ AMM يظل معلقاً على سعر قديم (stale price). بوت الـ HFT يلقط الفارق، يسحب السيولة الرخيصة من الـ AMM، ويصرفها في الـ CEX. وبكذا مزود السيولة في الـ AMM ينفذ دائماً بأسوء سعر ضد تدفق معلوماتي قوي (Toxic Flow).

LVR ≈ (σ² / 8) · ∫₀^T S_t · L_t dt

حيث σ هي تقلبات الأصل، S_t هو السعر، و L_t هي السيولة النشطة. الخلاصة: كل ما زادت تقلبات السوق، كل ما قدم الـ LPs أرباحاً أكثر لبوتات الـ HFT على طبق من ذهب.

5. الشغل العملي: سكربت Python لمقارنة تسعير AMM v3 مقابل عمق L2

خلونا ننتقل للكود. بنبني أداة شغال في Python تحسب سعر التنفيذ الحقيقي (شامل الانزلاق السعري) لأمر افتراضي، ونقارن عمق دفتر L2 مع مسبح AMM قائم على Constant Product.

الكود مكتوب بـ Python صافي تماماً بدون أي مكتبات ثقيلة أو تعقيدات.

from __future__ import annotations
from dataclasses import dataclass
from enum import Enum
from typing import List
class FeeMode(Enum):
    QUOTE = "quote"  # العمولة الخصم من الـ USDT الداخلي
    BASE = "base"    # العمولة الخصم من الـ ETH المستلم
@dataclass
class AskLevel:
    price: float
    volume: float
class OrderBookL2:
    """
    دفتر أوامر L2 (جانب الـ Ask).
    مثال زوج ETH/USDT:
        price  = USDT لكل 1 ETH
        volume = ETH
    """
    def __init__(
        self,
        asks: List[tuple[float, float]],
        taker_fee: float = 0.001,
    ) -> None:
        if not (0 <= taker_fee < 1):
            raise ValueError("taker_fee يجب أن يكون في نطاق [0, 1)")
        self.taker_fee = float(taker_fee)
        self.asks: List[AskLevel] = []
        for price, volume in asks:
            if price <= 0:
                raise ValueError("السعر يجب أن يكون > 0")
            if volume <= 0:
                raise ValueError("الحجم يجب أن يكون > 0")
            self.asks.append(
                AskLevel(
                    price=float(price),
                    volume=float(volume),
                )
            )
        self.asks.sort(key=lambda x: x.price)
    def best_ask(self) -> float:
        if not self.asks:
            raise ValueError("دفتر الأوامر فارغ")
        return self.asks[0].price
    def spot_price(self) -> float:
        return self.best_ask()
    def total_liquidity_quote(self) -> float:
        return sum(
            level.price * level.volume
            for level in self.asks
        )
    def total_liquidity_base(self) -> float:
        return sum(
            level.volume
            for level in self.asks
        )
    def simulate_buy(
        self,
        amount_in_quote: float,
        mutate: bool = True,
        fee_mode: FeeMode = FeeMode.QUOTE,
    ) -> tuple[float, float]:
        """
        محاكاة شراء الأصل الأساسي باستخدام amount_in_quote.
        يرجع:
            (
                الأصل_الأساسي_المستلم,
                متوسط_السعر_الفعلي
            )
        """
        if amount_in_quote <= 0:
            raise ValueError(
                "amount_in_quote يجب أن يكون > 0"
            )
        if fee_mode == FeeMode.QUOTE:
            quote_for_trade = (
                amount_in_quote *
                (1.0 - self.taker_fee)
            )
        else:
            quote_for_trade = amount_in_quote
        remaining_quote = quote_for_trade
        total_base_bought = 0.0
        new_levels: List[AskLevel] = []
        for level in self.asks:
            if remaining_quote <= 0:
                new_levels.append(level)
                continue
            level_cost = (
                level.price *
                level.volume
            )
            if remaining_quote >= level_cost:
                total_base_bought += level.volume
                remaining_quote -= level_cost
            else:
                base_from_level = (
                    remaining_quote /
                    level.price
                )
                total_base_bought += base_from_level
                remaining_volume = (
                    level.volume -
                    base_from_level
                )
                if remaining_volume > 1e-12:
                    new_levels.append(
                        AskLevel(
                            level.price,
                            remaining_volume,
                        )
                    )
                remaining_quote = 0.0
        if remaining_quote > 1e-9:
            raise ValueError(
                "السيولة غير كافية في دفتر الأوامر"
            )
        if fee_mode == FeeMode.BASE:
            total_base_bought *= (
                1.0 - self.taker_fee
            )
        if total_base_bought <= 0:
            raise ValueError(
                "الحجم المستلم كان صفراً"
            )
        avg_price = (
            amount_in_quote /
            total_base_bought
        )
        if mutate:
            self.asks = new_levels
        return total_base_bought, avg_price
class ConstantProductAMM:
    """
    AMM بنمط Uniswap V2
    x = ETH
    y = USDT
    spot = y / x
    """
    def __init__(
        self,
        reserve_x: float,
        reserve_y: float,
        fee: float = 0.003,
    ) -> None:
        if reserve_x <= 0:
            raise ValueError(
                "reserve_x يجب أن يكون > 0"
            )
        if reserve_y <= 0:
            raise ValueError(
                "reserve_y يجب أن يكون > 0"
            )
        if not (0 <= fee < 1):
            raise ValueError(
                "fee يجب أن يكون في نطاق [0, 1)"
            )
        self.x = float(reserve_x)
        self.y = float(reserve_y)
        self.fee = float(fee)
    def spot_price(self) -> float:
        return self.y / self.x
    def invariant(self) -> float:
        return self.x * self.y
    def simulate_buy(
        self,
        amount_y_in: float,
        mutate: bool = True,
    ) -> tuple[float, float]:
        """
        محاكاة شراء ETH باستخدام USDT.
        يرجع:
            (
                ETH_المستلم,
                متوسط_السعر
            )
        """
        if amount_y_in <= 0:
            raise ValueError(
                "amount_y_in يجب أن يكون > 0"
            )
        y_effective = (
            amount_y_in *
            (1.0 - self.fee)
        )
        x_out = (
            self.x *
            y_effective
        ) / (
            self.y +
            y_effective
        )
        if x_out <= 0:
            raise ValueError(
                "الحجم المستلم كان صفراً"
            )
        if x_out >= self.x:
            raise ValueError(
                "السيولة غير كافية في المسبح"
            )
        avg_price = (
            amount_y_in /
            x_out
        )
        if mutate:
            self.x -= x_out
            self.y += amount_y_in
        return x_out, avg_price
def print_trade_result(
    title: str,
    received: float,
    avg_price: float,
    spot_price: float,
) -> None:
    slippage = (
        (avg_price - spot_price)
        / spot_price
    ) * 100
    print(
        f"{title:<12}"
        f" المستلم: {received:.6f}"
        f" | متوسط السعر: {avg_price:,.4f}"
        f" | الانزلاق السعري: {slippage:.4f}%"
    )
def main() -> None:
    asks = [
        (3000.0, 1.5),
        (3001.0, 2.0),
        (3002.5, 5.0),
        (3005.0, 10.0),
        (3010.0, 25.0),
    ]
    orderbook = OrderBookL2(
        asks=asks,
        taker_fee=0.001,
    )
    amm = ConstantProductAMM(
        reserve_x=1000.0,
        reserve_y=3_000_000.0,
        fee=0.003,
    )
    trade_size = 15_000.0
    ob_spot = orderbook.spot_price()
    amm_spot = amm.spot_price()
    ob_received, ob_avg = orderbook.simulate_buy(
        amount_in_quote=trade_size,
        mutate=True,
        fee_mode=FeeMode.QUOTE,
    )
    amm_received, amm_avg = amm.simulate_buy(
        amount_y_in=trade_size,
        mutate=True,
    )
    print(
        f"\n=== مقارنة تنفيذ أمر "
        f"بقيمة {trade_size:,.2f} USDT ===\n"
    )
    print_trade_result(
        "ORDERBOOK",
        ob_received,
        ob_avg,
        ob_spot,
    )
    print_trade_result(
        "AMM V2",
        amm_received,
        amm_avg,
        amm_spot,
    )
    print("\n=== الحالة بعد الصفقة ===\n")
    if orderbook.asks:
        print(
            f"Best Ask: "
            f"{orderbook.best_ask():,.4f}"
        )
    else:
        print("تم مسح دفتر الأوامر بالكامل")
    print(
        f"AMM Spot: "
        f"{amm.spot_price():,.4f}"
    )
    print(
        f"AMM Reserves: "
        f"{amm.x:.6f} ETH | "
        f"{amm.y:.2f} USDT"
    )
if __name__ == "__main__":
    main()

كيف تقرأ النتائج وتحللها

في مسابح الـ AMM العميقة، الأوامر الصغيرة تنفذ بانزلاق سعري شبه معدوم، بس كل ما كبر الحجم، السعر في الـ AMM يقفز على منحنى محدد بشكل حاد. بينما في دفتر الأوامر، الموضوع يعتمد كلياً على جدران السيولة (liquidity walls). إذا الدفتر ناشف ومسحوب منه السيولة، بتتجاوز مستويات سعرية كثيرة وبتأكل انزلاق سعري ألعن بكثير من الـ AMM.

6. النماذج الهجينة: كيف تدمج منصات DEX سجل الأوامر (Order Book) ولماذا تتجه منصات CEX نحو صانع السوق الآلي (AMM)؟

الخط الفاصل بين دفاتر الأوامر (Order Books) ونماذج AMM بدأ يتلاشى تدريجياً. كبار المتداولين والمؤسسات (Institutional Traders) اصطدموا سريعاً بسقف إمكانيات نموذج Uniswap v2 التقليدي؛ فكفاءة رأس المال المنخفضة واستحالة وضع أوامر حدية معقدة (مثل Stop-loss و Take-profit و Iceberg) جعلت منصات DEX غير عملية لصناع السوق (Market Makers) النشطين.

هذا النقص دفع نحو نقلة نوعية وظهور معماريات هجينة أكثر تطوراً.

1. سجلات الأوامر على الشبكة وعلى الطبقة الثانية On-Chain & L2 Order Books (مثل dYdX، Hyperliquid، Vertex)

مع ظهور شبكات L1/L2 عالية الأداء (مثل Aptos، Sui، Arbitrum) وسلاسل التطبيقات المتخصصة (AppChains)، أصبح تشغيل سجل أوامر كامل ومباشر على البلوكشين أمراً ممكناً وعملياً.

مطابقة خارج الشبكة + تسوية على الشبكة (Off-chain matching + On-chain settlement): تعمل محركات مطابقة الأوامر خارج السلسلة بسرعة فائقة (لتحقيق زمن استجابة أقل من 10 مللي ثانية)، بينما يتم تسجيل نتائج التنفيذ ومتطلبات الهامش (Margin Requirements) عبر العقود الذكية.

سجل أوامر كامل على السلسلة Fully On-chain (مثل Hyperliquid، Serum/Ellipsis): يعيش سجل الأوامر بالكامل داخل آليات الإجماع (Consensus) للبلوكشين. التحدي الأكبر هنا هو التعامل مع الأحجام الضخمة من المعاملات؛ حيث إن عمليات تعديل وإلغاء الأوامر (Cancel/Replace) في التداول عالي التردد (HFT) تشكل ما بين 95% إلى 98% من إجمالي المعاملات الواردة.

2. المعمارية القائمة على المقاصد (Intent-Based Architecture) وطلب عروض الأسعار (RFQ)

نموذج "المقاصد" (المستخدم في Uniswap X، 1inch Fusion، CoW Protocol) يُعيد في الواقع النهج المؤسسي لتنفيذ الصفقات.

بدلاً من أن يوقع المتداول على معاملة للتفاعل مع مجمع سيولة (Pool) معين، فإنه يوقع على "قصد" أو رغبة (مثال: «أريد استبدال 100 ETH بما لا يقل عن 300,000 USDC»). بعد ذلك، تتنافس شبكة من المنفذين الخارجيين (Fillers / Solvers) للفوز بحق تنفيذ هذا الأمر باستخدام أي مصدر سيولة متاح — سواء كان منصة CEX، أو مخزون صناع السوق الخواص، أو منصات DEX.

بالنسبة للمتداول اليومي، هذا يعني الانزلاق السعري الصفري (Zero Slippage) والحماية الكاملة من هجمات MEV (هجمات الساندويتش)؛ إذ يتحمل الـ Solver جميع مخاطر الانزلاق والتنفيذ، محققاً أرباحه عبر المراجحة (Arbitrage) بين الأصول.

7. التطبيق العلمي: سكريبت متقدم للمراجحة (Arbitrage) بين سجل أوامر CEX L2 ونموذج AMM v2

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

فيما يلي سكريبت بلغة Python يحسب بالوقت الفعلي الحجم الأمثل لصفقة المراجحة (dx) مع مراعاة رسوم AMM، وعمق سجل الأوامر في CEX، ورسوم Maker/Taker.

from __future__ import annotations
from dataclasses import dataclass
from typing import List, Dict, Any

@dataclass
class BidLevel:
    price: float
    volume: float

class OrderBookSide:
    """
    جانب الطلب (Bid) في سجل الأوامر.
    price  = سعر الشراء
    volume = حجم الأصل الأساسي
    """
    def __init__(self, bids: List[tuple[float, float]]):
        self.bids: List[BidLevel] = []
        for price, volume in bids:
            if price <= 0:
                raise ValueError("يجب أن يكون السعر > 0")
            if volume <= 0:
                raise ValueError("يجب أن يكون الحجم > 0")
            self.bids.append(
                BidLevel(
                    float(price),
                    float(volume)
                )
            )
        self.bids.sort(
            key=lambda x: x.price,
            reverse=True
        )

    def is_empty(self) -> bool:
        return len(self.bids) == 0

    def best_bid(self) -> float:
        if self.is_empty():
            raise ValueError("سجل الأوامر فارغ")
        return self.bids[0].price

    def cumulative_levels(self) -> List[tuple[float, float]]:
        """
        يرجع نقاط تغير السيولة التراكمية.
        [
            (2 ETH, VWAP حتى 2 ETH),
            (7 ETH, VWAP حتى 7 ETH),
            ...
        ]
        """
        result = []
        cumulative_x = 0.0
        cumulative_y = 0.0
        for level in self.bids:
            cumulative_x += level.volume
            cumulative_y += level.volume * level.price
            result.append(
                (
                    cumulative_x,
                    cumulative_y
                )
            )
        return result

    def execute_sell(
        self,
        amount_x: float
    ) -> tuple[float, float]:
        """
        بيع amount_x في سجل الأوامر.
        يرجع:
        (
            المقبوض_USDT,
            VWAP
        )
        """
        if amount_x <= 0:
            raise ValueError(
                "يجب أن يكون amount_x > 0"
            )
        remaining_x = amount_x
        received_y = 0.0
        for level in self.bids:
            if remaining_x <= 0:
                break
            executed = min(
                remaining_x,
                level.volume
            )
            received_y += (
                executed *
                level.price
            )
            remaining_x -= executed
        if remaining_x > 1e-9:
            raise ValueError(
                "سيولة سجل الأوامر غير كافية"
            )
        vwap = received_y / amount_x
        return received_y, vwap

class AMMPool:
    """
    صانع سوق آلي ذو ناتج ثابت (Constant Product AMM)
    x = ETH
    y = USDT
    """
    def __init__(
        self,
        reserve_x: float,
        reserve_y: float,
        fee: float = 0.003
    ):
        if reserve_x <= 0:
            raise ValueError(
                "يجب أن يكون reserve_x > 0"
            )
        if reserve_y <= 0:
            raise ValueError(
                "يجب أن يكون reserve_y > 0"
            )
        if not (0 <= fee < 1):
            raise ValueError(
                "يجب أن تكون الرسوم fee في النطاق [0,1)"
            )
        self.x = float(reserve_x)
        self.y = float(reserve_y)
        self.fee = float(fee)

    def spot_price(self) -> float:
        return self.y / self.x

    def calculate_out_x(
        self,
        amount_y_in: float
    ) -> float:
        if amount_y_in <= 0:
            raise ValueError(
                "يجب أن يكون amount_y_in > 0"
            )
        y_effective = (
            amount_y_in *
            (1.0 - self.fee)
        )
        x_out = (
            self.x *
            y_effective
        ) / (
            self.y +
            y_effective
        )
        return x_out

    def calculate_required_y_for_x(
        self,
        target_x_out: float
    ) -> float:
        """
        المعكوس الرياضي لصيغة AMM.
        كمية USDT المطلوبة للحصول على target_x_out.
        """
        if target_x_out <= 0:
            raise ValueError(
                "يجب أن يكون target_x_out > 0"
            )
        if target_x_out >= self.x:
            raise ValueError(
                "لا يمكن سحب كامل الاحتياطي"
            )
        y_effective = (
            target_x_out *
            self.y
        ) / (
            self.x -
            target_x_out
        )
        amount_y_in = (
            y_effective /
            (1.0 - self.fee)
        )
        return amount_y_in

def find_optimal_arbitrage(
    amm: AMMPool,
    cex_bids: OrderBookSide,
    cex_fee: float = 0.001
) -> Dict[str, Any]:
    if not (0 <= cex_fee < 1):
        raise ValueError(
            "يجب أن تكون cex_fee في النطاق [0,1)"
        )
    if cex_bids.is_empty():
        return {
            "opportunity": False,
            "reason": "سجل أوامر CEX فارغ"
        }
    best_bid = cex_bids.best_bid()
    if (
        best_bid *
        (1.0 - cex_fee)
        <=
        amm.spot_price()
    ):
        return {
            "opportunity": False,
            "reason": "لا يوجد فارق سعري (سبريد)"
        }
    best_result = None
    best_profit = float("-inf")
    cumulative = cex_bids.cumulative_levels()
    for cumulative_x, cumulative_y in cumulative:
        try:
            required_usdt = (
                amm.calculate_required_y_for_x(
                    cumulative_x
                )
            )
            gross_usdt = cumulative_y
            net_usdt = (
                gross_usdt *
                (1.0 - cex_fee)
            )
            profit = (
                net_usdt -
                required_usdt
            )
            if profit > best_profit:
                avg_buy_price = (
                    required_usdt /
                    cumulative_x
                )
                avg_sell_price = (
                    gross_usdt /
                    cumulative_x
                )
                amm_spot = (
                    amm.spot_price()
                )
                impact_pct = (
                    (
                        avg_buy_price -
                        amm_spot
                    )
                    /
                    amm_spot
                ) * 100
                roi_pct = (
                    profit /
                    required_usdt
                ) * 100
                best_profit = profit
                best_result = {
                    "opportunity": True,
                    "input_usdt_amm": required_usdt,
                    "bought_x": cumulative_x,
                    "gross_usdt_cex": gross_usdt,
                    "net_usdt_cex": net_usdt,
                    "net_profit_usdt": profit,
                    "roi_pct": roi_pct,
                    "amm_spot_price": amm_spot,
                    "amm_avg_buy_price": avg_buy_price,
                    "amm_price_impact_pct": impact_pct,
                    "cex_vwap": avg_sell_price,
                }
        except ValueError:
            continue
    if best_result is None:
        return {
            "opportunity": False,
            "reason": "السيولة غير كافية"
        }
    if best_result["net_profit_usdt"] <= 0:
        return {
            "opportunity": False,
            "reason": "الانزلاق السعري يقضي على الأرباح"
        }
    return best_result

def main():
    amm = AMMPool(
        reserve_x=1000.0,
        reserve_y=3_000_000.0,
        fee=0.003
    )
    cex = OrderBookSide([
        (3050.0, 2.0),
        (3048.0, 5.0),
        (3045.0, 10.0),
        (3040.0, 20.0),
        (3030.0, 50.0),
    ])
    result = find_optimal_arbitrage(
        amm,
        cex,
        cex_fee=0.0005
    )
    print(
        "\n=== نتيجة البحث عن المراجحة (Arbitrage) ===\n"
    )
    if not result["opportunity"]:
        print(
            f"الحالة: إلغاء\n"
            f"السبب: {result['reason']}"
        )
        return
    print("الحالة: تم العثور على فرصة مراجحة")
    print()
    print(
        f"المبلغ المدخل في AMM: "
        f"${result['input_usdt_amm']:,.2f}"
    )
    print(
        f"ETH المشتراة: "
        f"{result['bought_x']:.6f}"
    )
    print(
        f"إجمالي عائدات CEX: "
        f"${result['gross_usdt_cex']:,.2f}"
    )
    print(
        f"صافي عائدات CEX: "
        f"${result['net_usdt_cex']:,.2f}"
    )
    print(
        f"الربح الصافي: "
        f"${result['net_profit_usdt']:,.2f}"
    )
    print(
        f"عائد الاستثمار (ROI): "
        f"{result['roi_pct']:.4f}%"
    )
    print(
        f"سعر AMM الفوري: "
        f"{result['amm_spot_price']:.4f}"
    )
    print(
        f"متوسط سعر الشراء في AMM: "
        f"{result['amm_avg_buy_price']:.4f}"
    )
    print(
        f"تأثير السعر في AMM: "
        f"{result['amm_price_impact_pct']:.4f}%"
    )
    print(
        f"متوسط السعر المرجح بالحجم في CEX (VWAP): "
        f"{result['cex_vwap']:.4f}"
    )

if __name__ == "__main__":
    main()

8. توصيات عملية للمتداول: كيف تتجنب دفع "ضريبة الجهل" بالسوق؟

فهم البنية الدقيقة للسوق (Microstructure) ليس مجرد معلومات أكاديمية تنظيرية، بل هو أداة أساسية تحمي رأس مالك من النزيف والتسرب الخفي.

• اختر نوع الأمر بدقة حسب ظروف السوق: في منصات CEX، التنفيذ السريع عبر أمر السوق (Market Order) لأحجام كبيرة سيمتص السيولة من عدة مستويات في سجل الأوامر، مما يضمن لك انزلاقاً سعرياً سلبياً. إذا كنت بحاجة للتنفيذ السريع، استخدم أوامر IOC (Immediate-or-Cancel) أو FOK (Fill-or-Kill) مع تحديد الحد الأقصى للسعر المقبول.
• احسب العمق الفعلي للسيولة وليس الفارق السعري (Spread) فقط: الفارق الضيق على مستويات L2 (مثلاً $0.01) غالباً ما يكون مجرد وهم. إذا كان المستوى الأول يحتوي على 100 دولار فقط، فإن أمرك بقيمة 50,000 دولار سيلتهم السجل لاهثاً إلى مستويات أعمق بأسعار مختلفة تماماً.
• انتبه لخطورة السيولة السامة (Toxic Flow) في الـ AMM: إذا كنت توفر السيولة في Uniswap v3 ضمن نطاق ضيق أثناء صدور الأخبار القوية أو التقلبات العالية، فإنك تضع نفسك كهدف سهل لبوتات الـ HFT Arbitrageurs. معدل LVR (Loss-Versus-Rebalancing) سيلتهم جميع أرباحك من الرسوم في دقائق معدودة.
• احمِ معاملاتك من هجمات MEV على الـ DEX: عند التداول على منصات AMM، حدد دائماً نسبة الانزلاق السعري (Slippage Tolerance) بحيث لا تتجاوز 0.1% إلى 0.3%، أو استخدم نقاط اتصال RPC خاصة (مثل Flashbots Protect / MEV-Blocker) لتفادي توجيه معاملاتك إلى الـ Mempool العام وحمايتها من هجمات الساندويتش.

شاركونا في التعليقات: ما هي أكبر نسبة انزلاق سعري واجهتموها أثناء تنفيذ الأوامر الكبيرة؟ وما هو النموذج المفضل لديكم لتداولاتكم اليومية (CEX Order Book أم DEX AMM)؟