Asyncio و معماری Trading System

۱. از مدل تا سیستم

تا اینجا هر چه ساختیم روی داده ذخیره‌شده اجرا می‌شد: مدل قیمت‌گذاری درس ۱۱، Backtest درس ۱۷، محدودیت‌های موجودی درس ۲۴. Backtest یک حلقه ساده است: رویداد بعدی را بخوان، تصمیم بگیر، برو سراغ رویداد بعدی. زمان منتظر تو می‌ماند.

در اجرای زنده زمان منتظر نمی‌ماند. هم‌زمان چند اتفاق می‌افتد:

داده بازار از یک WebSocket با نرخ ده‌ها یا صدها پیام در ثانیه می‌رسد.
پاسخ سفارش‌ها (تأیید، رد، پر شدن) از یک کانال دیگر می‌آید.
تایمرها باید سر وقت اجرا شوند: بررسی کهنه شدن داده، لغو سفارش‌های قدیمی، ذخیره وضعیت.
و همه این‌ها باید بدون اینکه یکی منتظر دیگری بماند پیش بروند.

بیشتر وقت چنین برنامه‌ای صرف انتظار برای شبکه می‌شود، نه محاسبه. به این نوع کار I/O-bound می‌گویند و ابزار استاندارد پایتون برایش asyncio است. این درس اولین درس بخش «Production» است: یاد می‌گیریم مدل را در قالب سیستمی بچینیم که در محیط واقعی دوام بیاورد.

۲. Event Loop در یک نگاه

asyncio روی یک Event Loop کار می‌کند: یک حلقه روی یک ترد که فهرستی از کارهای آماده اجرا دارد و آن‌ها را یکی‌یکی جلو می‌برد.

Coroutine: تابعی که با async def تعریف می‌شود. صدا زدنش فقط یک شیء می‌سازد؛ تا وقتی await نشود یا به Task تبدیل نشود، اجرا نمی‌شود.
await: نقطه‌ای که Coroutine می‌گوید «تا این نتیجه برسد، کنترل را پس می‌دهم». Event Loop در این فاصله کارهای دیگر را اجرا می‌کند.
Task: یک Coroutine که به Event Loop سپرده شده و مستقل از بقیه پیش می‌رود (asyncio.create_task یا TaskGroup).

نکته کلیدی این است که زمان‌بندی همکارانه (Cooperative) است. Event Loop نمی‌تواند یک Task را وسط کار متوقف کند؛ هر Task فقط در await کنترل را پس می‌دهد. این هم نقطه قوت است و هم دام:

قوت: بین دو await، کد تو به‌تنهایی اجرا می‌شود. هیچ ترد دیگری وسط یک خط محاسبه، متغیر را عوض نمی‌کند.
دام: اگر یک Task بدون await کار طولانی انجام دهد، همه سیستم منتظر می‌ماند؛ داده بازار خوانده نمی‌شود و سفارش‌ها لغو نمی‌شوند.

۳. اجزای سیستم و مرزهایشان

در درس ۱ یک معماری ساده دیدیم. حالا همان را با نگاه مهندسی دقیق‌تر می‌کنیم. هر جزء یک Task است که فقط از راه صف (Queue) با بقیه حرف می‌زند:

Market Data Handler دریافت WebSocket، بازسازی دفتر سفارش (درس ۱۶)، نگه داشتن فقط آخرین وضعیت
Strategy Fair Value، Spread و Skew؛ خروجی آن قیمت‌های مطلوب است، نه سفارش
Risk Gate بررسی هر سفارش پیش از ارسال، Kill Switch (درس ۲۷)
Order Gateway ارسال و لغو سفارش، Rate Limit، دریافت تأیید و Fill (درس ۲۶)
Position Keeper تنها نویسنده موجودی و سفارش‌های باز؛ رویداد Fill از اینجا به Strategy برمی‌گردد
Monitoring متریک، لاگ و هشدار (درس ۲۸)

این جداسازی سه فایده دارد:

هر جزء جداگانه قابل تست است. Strategy فقط یک تابع است که داده بازار و موجودی می‌گیرد و قیمت مطلوب برمی‌گرداند؛ پس همان منطقی که در Backtest درس ۱۷ اجرا می‌شود، می‌تواند بدون تغییر در اجرای زنده هم اجرا شود.
هر جزء سرعت خودش را دارد. داده بازار هر میلی‌ثانیه می‌رسد، Strategy شاید هر ۳ میلی‌ثانیه یک بار قیمت می‌دهد، و صرافی به ازای هر ثانیه تعداد محدودی سفارش قبول می‌کند. صف‌ها این سرعت‌های متفاوت را از هم جدا می‌کنند.
مسیر ریسک دور زده نمی‌شود. Strategy به Order Gateway دسترسی مستقیم ندارد. هر سفارش باید از Risk Gate رد شود.

۴. صف‌ها و Backpressure

وقتی تولیدکننده سریع‌تر از مصرف‌کننده است، صف بزرگ می‌شود. اگر صف حد نداشته باشد، تأخیر بی‌صدا بالا می‌رود: Strategy روی قیمتی تصمیم می‌گیرد که چند ثانیه پیش معتبر بوده. به سازوکاری که سرعت تولیدکننده را با ظرفیت مصرف‌کننده هماهنگ می‌کند Backpressure می‌گویند.

قاعده طلایی این است که سیاست صف به معنای پیام بستگی دارد:

نوع پیام	اگر مصرف‌کننده عقب بماند	چرا
وضعیت دفتر سفارش	فقط آخرین نسخه نگه داشته شود (Conflation)	قیمت قدیمی ارزشی ندارد؛ فقط وضعیت فعلی مهم است
قیمت مطلوب Strategy	فقط آخرین نسخه، یا صف کوچک با انتظار	ارسال ده قیمت قدیمی پشت سر هم فقط Rate Limit را مصرف می‌کند
معاملات بازار (برای آمار)	صف محدود؛ شمارش پیام‌های دورریخته و هشدار	برای آمار مهم‌اند ولی نه حیاتی
تأیید سفارش و Fill	هرگز دور ریخته نشود؛ هشدار روی طول صف	هر Fill یعنی تغییر واقعی موجودی و پول

Conflation یعنی اگر پیام تازه‌ای رسید و پیام قبلی هنوز خوانده نشده، پیام قبلی را با تازه جایگزین کنی. برای داده بازار تقریباً همیشه درست است. برای Fill همیشه غلط است.

۵. یک سیستم کوچک با asyncio

کد زیر سه جزء را با صف به هم وصل می‌کند: یک فید داده بازار مصنوعی که هر حدود ۱ میلی‌ثانیه یک Tick می‌سازد، یک Strategy که هر قیمت‌دهی‌اش حدود ۳ میلی‌ثانیه طول می‌کشد، و یک Order Gateway که رفت‌وبرگشت صرافی را با ۲ میلی‌ثانیه تأخیر شبیه‌سازی می‌کند. همه اعداد فرضی‌اند و فقط برای نشان دادن رفتار سیستم انتخاب شده‌اند.

import asyncio
import random

class Latest:
    """Conflating mailbox: keeps only the newest value (good for market data)."""
    def __init__(self):
        self._value = None
        self._ready = asyncio.Event()
        self.dropped = 0

    def put(self, value):
        if self._ready.is_set():
            self.dropped += 1          # previous tick was never read: overwrite it
        self._value = value
        self._ready.set()

    async def get(self):
        await self._ready.wait()
        self._ready.clear()
        return self._value

async def market_data(book: Latest, n_ticks: int, rng: random.Random):
    mid = 100_000.0
    for _ in range(n_ticks):
        mid += rng.choice([-10, 0, 10])   # toy random walk in 10 USDT steps
        book.put(mid)
        await asyncio.sleep(0.001)        # a new tick every ~1 ms
    book.put(None)                        # end of stream

async def strategy(book: Latest, orders: asyncio.Queue, fills: asyncio.Queue):
    position = 0.0                        # single writer: only this task changes it
    quotes = 0
    while (mid := await book.get()) is not None:
        while not fills.empty():          # apply fills before pricing
            side, qty = fills.get_nowait()
            position += qty if side == "buy" else -qty
        skew = -position * 50             # lean quotes against inventory
        bid, ask = mid - 10 + skew, mid + 10 + skew
        await orders.put((bid, ask))      # bounded queue: waits if gateway is behind
        quotes += 1
        await asyncio.sleep(0.003)        # pretend pricing takes ~3 ms
    await orders.put(None)                # tell the gateway we are done
    while (fill := await fills.get()) is not None:   # late fills still count
        side, qty = fill
        position += qty if side == "buy" else -qty
    return quotes, position

async def order_gateway(orders: asyncio.Queue, fills: asyncio.Queue, rng: random.Random):
    sent = 0
    while (quote := await orders.get()) is not None:
        await asyncio.sleep(0.002)        # simulated exchange round trip
        sent += 1
        if rng.random() < 0.2:            # sometimes one side gets filled
            fills.put_nowait((rng.choice(["buy", "sell"]), 0.1))
    fills.put_nowait(None)
    return sent

async def main():
    book, orders, fills = Latest(), asyncio.Queue(maxsize=10), asyncio.Queue()
    async with asyncio.TaskGroup() as tg:   # if one task crashes, all are cancelled
        tg.create_task(market_data(book, 300, random.Random(1)))
        s = tg.create_task(strategy(book, orders, fills))
        g = tg.create_task(order_gateway(orders, fills, random.Random(2)))
    quotes, position = s.result()
    print(f"ticks=300 dropped={book.dropped} quotes={quotes} sent={g.result()}")
    print(f"final position={position:+.1f} BTC")

asyncio.run(main())

خروجی یک اجرا روی سیستم من (به Python 3.11 یا جدیدتر نیاز دارد):

ticks=300 dropped=192 quotes=108 sent=108
final position=+0.3 BTC

چند نکته از همین خروجی:

حدود دو سوم Tickها دور ریخته شدند و این درست است. Strategy هر ۳ میلی‌ثانیه آماده قیمت‌دهی می‌شود و در آن لحظه فقط تازه‌ترین قیمت برایش مهم است. اگر به‌جای Latest یک صف بی‌حد گذاشته بودیم، Strategy همه ۳۰۰ Tick را به ترتیب پردازش می‌کرد و در انتها بیش از ۶۰۰ میلی‌ثانیه از بازار عقب بود (در اجرای من حدود ۶۴۰ میلی‌ثانیه).
اعداد از اجرایی به اجرای دیگر کمی فرق می‌کنند. دقت asyncio.sleep به سیستم‌عامل بستگی دارد. در Backtest همه چیز قطعی است؛ در اجرای زنده نه. این یکی از دلایل اختلاف بک‌تست با اجرای واقعی است که در درس ۳۱ بررسی می‌کنیم.
موجودی فقط یک نویسنده دارد. در این مثال کوچک Strategy نقش Position Keeper را هم دارد. Gateway موجودی را عوض نمی‌کند؛ فقط رویداد Fill را در صف می‌گذارد. Strategy قبل از هر قیمت‌دهی Fillها را اعمال می‌کند. با موجودی 0.3 BTC، مقدار Skew برابر ۱۵ دلار است و هر دو قیمت ۱۵ دلار پایین‌تر می‌روند تا احتمال فروش بیشتر شود (منطق Skew را در درس ۶ دیدیم).
TaskGroup جلوی شکست بی‌صدا را می‌گیرد. اگر یکی از Taskها Exception بدهد، بقیه لغو می‌شوند و خطا بالا می‌آید. با create_task تنها، Task مرده ممکن است فقط یک پیام هشدار در لاگ بگذارد و سیستم با یک جزء از کار افتاده به کار ادامه دهد؛ مثلاً Strategy همچنان قیمت بدهد در حالی که دیگر Fillی دریافت نمی‌شود.

۶. هرگز Event Loop را مسدود نکن

هر فراخوانی که بدون await منتظر بماند، کل سیستم را متوقف می‌کند. مقصرهای رایج:

time.sleep به‌جای asyncio.sleep
کتابخانه HTTP همگام (Synchronous) برای ارسال سفارش یا گرفتن موجودی
محاسبه سنگین روی مسیر داغ: بازآموزی مدل، یک groupby بزرگ روی DataFrame، سریال کردن یک شیء حجیم
نوشتن همگام روی دیسک یا پایگاه داده در هر رویداد

این مشکل را می‌شود اندازه گرفت. یک Task کوچک که مرتب ۱۰ میلی‌ثانیه می‌خوابد و می‌سنجد چقدر دیرتر از موعد بیدار شده، Loop Lag را نشان می‌دهد:

import asyncio
import time

async def lag_monitor(interval=0.01, samples=50):
    """Sleep for `interval` and measure how late we wake up: that delay is loop lag."""
    worst = 0.0
    for _ in range(samples):
        start = time.perf_counter()
        await asyncio.sleep(interval)
        worst = max(worst, time.perf_counter() - start - interval)
    return worst

def heavy_calc():
    time.sleep(0.2)          # stands in for a blocking call or heavy CPU work

async def bad_handler():
    await asyncio.sleep(0.1)
    heavy_calc()             # runs ON the loop thread: everything freezes

async def good_handler():
    await asyncio.sleep(0.1)
    await asyncio.to_thread(heavy_calc)   # runs in a worker thread; loop stays free

async def run(handler):
    monitor = asyncio.create_task(lag_monitor())
    await handler()
    return await monitor

for h in (bad_handler, good_handler):
    print(f"{h.__name__}: worst loop lag = {asyncio.run(run(h)) * 1000:.0f} ms")

bad_handler: worst loop lag = 199 ms
good_handler: worst loop lag = 1 ms

در نسخه بد، ۲۰۰ میلی‌ثانیه هیچ چیز در سیستم اتفاق نیفتاد. اگر در همین ۲۰۰ میلی‌ثانیه قیمت ۵۰ دلار حرکت کرده باشد، سفارش‌های تو با قیمت کهنه در دفتر مانده‌اند و کسی که سریع‌تر است آن‌ها را برمی‌دارد؛ همان Adverse Selection درس ۸.

to_thread همه جا جواب نمی‌دهد: time.sleep و بیشتر فراخوانی‌های I/O قفل GIL را آزاد می‌کنند، پس در ترد کمکی کار می‌کنند. اما محاسبه سنگین با کد خالص پایتون GIL را نگه می‌دارد و حتی در ترد جدا هم Event Loop را کند می‌کند. برای آن کار از ProcessPoolExecutor استفاده کن یا محاسبه را به یک سرویس جدا ببر و فقط نتیجه را از صف بخوان.

همین مانیتور را در سیستم واقعی نگه دار و مقدارش را به‌عنوان متریک گزارش کن. Loop Lag یکی از اولین متریک‌هایی است که در درس ۲۸ روی داشبورد می‌گذاریم.

۷. هر وضعیت، فقط یک نویسنده

اینکه asyncio تک‌ترد است به این معنا نیست که Race Condition ندارد. هر await نقطه‌ای است که Task دیگری می‌تواند اجرا شود. این کد را ببین:

import asyncio

state = {"position": 0.0}

async def save_to_db(value):
    await asyncio.sleep(0.01)    # any await is a point where other tasks run

async def on_fill(qty):
    pos = state["position"]      # read
    await save_to_db(pos + qty)  # yield to the loop
    state["position"] = pos + qty  # write: may overwrite another task's update

async def main():
    await asyncio.gather(on_fill(0.1), on_fill(0.1))
    print(f"position = {state['position']}")   # expected 0.2

asyncio.run(main())

position = 0.1

دو Fill رسید ولی موجودی فقط یکی را نشان می‌دهد. هر دو Task مقدار صفر را خواندند، هر دو منتظر پایگاه داده ماندند، و دومی نتیجه اولی را بازنویسی کرد. به این خطا Lost Update می‌گویند و در سیستم معاملاتی یعنی موجودی واقعی تو با چیزی که فکر می‌کنی فرق دارد؛ همه محدودیت‌های ریسک درس ۲۴ روی عدد غلط اجرا می‌شوند.

راه‌حل قفل نیست؛ طراحی است. قاعده Single Writer می‌گوید هر بخش از وضعیت دقیقاً یک صاحب دارد:

موجودی و سفارش‌های باز را فقط Position Keeper تغییر می‌دهد. بقیه رویداد می‌فرستند، نه دستور «این مقدار را بنویس».
صاحب وضعیت، رویدادها را یکی‌یکی از صفش برمی‌دارد و اعمال می‌کند. اعمال یک رویداد هیچ awaitی وسطش ندارد.
بقیه اجزا فقط یک کپی فقط‌خواندنی (Snapshot) از وضعیت می‌گیرند.
ذخیره در پایگاه داده بعد از به‌روزرسانی حافظه و از طریق صف جداگانه انجام می‌شود، نه وسط آن.

این همان ایده‌ای است که در بسیاری از سیستم‌های پرسرعت مالی دیده می‌شود: یک رشته پردازش برای هر وضعیت، و ارتباط فقط با پیام. بسیاری از Matching Engineها هم، مثل نسخه ساده درس ۳، همین‌طور کار می‌کنند: یک نویسنده برای هر دفتر سفارش و یک صف ورودی.

۸. چک‌لیست معماری

پیش از اینکه کد را به محیط واقعی ببری، این سؤال‌ها را از خودت بپرس:

آیا هر صف حد دارد، و برای هر صف تصمیم گرفته‌ای وقتی پر شد چه اتفاقی بیفتد؟
آیا Fillها در هیچ شرایطی دور ریخته نمی‌شوند؟
آیا در مسیر داغ هیچ فراخوانی همگامی وجود ندارد؟ Loop Lag را اندازه می‌گیری؟
آیا هر وضعیت مشترک دقیقاً یک نویسنده دارد؟
اگر یک Task از کار بیفتد، سیستم متوجه می‌شود و متوقف می‌شود، یا با یک جزء مرده ادامه می‌دهد؟
آیا Strategy بدون عبور از Risk Gate راهی به صرافی دارد؟

۹. تمرین این درس

گزینه مناسب هر سؤال را انتخاب کن و در پایان نتیجه را ببین.

سؤال ۱: یک Task در asyncio یک حلقه محاسباتی ۵۰ میلی‌ثانیه‌ای بدون هیچ await اجرا می‌کند. در این مدت چه اتفاقی برای بقیه Taskها می‌افتد؟

هیچ‌کدام اجرا نمی‌شوند تا آن حلقه تمام شود به‌صورت موازی در ترد‌های دیگر ادامه می‌دهند Event Loop حلقه را وسط کار متوقف می‌کند و نوبت را به بقیه می‌دهد

سؤال ۲: برای کدام نوع پیام Conflation (نگه داشتن فقط آخرین پیام) مناسب نیست؟

وضعیت دفتر سفارش Fill سفارش‌ها قیمت مطلوب Strategy

سؤال ۳: فید داده هر ۱ میلی‌ثانیه یک Tick می‌فرستد و Strategy برای هر Tick ۴ میلی‌ثانیه وقت می‌گذارد. با صف بی‌حد، بعد از ۱ ثانیه Strategy تقریباً چقدر از بازار عقب است؟

حدود ۴ میلی‌ثانیه حدود ۷۵۰ میلی‌ثانیه حدود ۳ ثانیه

سؤال ۴: دو Coroutine هر دو موجودی را می‌خوانند، await می‌کنند و بعد مقدار جدید را می‌نویسند. بهترین راه‌حل معماری چیست؟

یک صاحب واحد برای موجودی که رویدادها را یکی‌یکی و بدون await وسطش اعمال کند تبدیل برنامه به چند ترد افزودن asyncio.sleep(0) قبل از نوشتن

درس بعدی

در درس ۲۶ سراغ Order Lifecycle می‌رویم: یک سفارش از لحظه ساخت تا پر شدن یا لغو از چه وضعیت‌هایی عبور می‌کند، چرا تأیید صرافی با پر شدن فرق دارد، با Timeout و قطع ارتباط چه کنیم، و چطور بعد از وصل شدن دوباره وضعیت خودمان را با صرافی تطبیق دهیم.

این مطلب صرفاً آموزشی است و توصیه خرید، فروش یا سرمایه‌گذاری نیست.