Market Data Pipeline: دریافت، پاک‌سازی، ذخیره و بازسازی Order Book

۱. همه چیز از داده شروع می‌شود

در سه درس قبل فرض کردیم داده تمیز و کامل روی دیسک منتظر ماست. در واقعیت، کسی باید آن را جمع کند؛ و تقریباً هر نتیجه غلطی در Backtest را می‌شود تا یک ایراد در همین مرحله ردیابی کرد: پیامی که گم شد و کسی نفهمید، دفتر سفارشی که چند دقیقه کهنه بود، یا Timestampی که از ساعت اشتباه آمد.

Market Data Pipeline مسیری است که داده بازار از صرافی تا ابزارهای تحقیق و استراتژی طی می‌کند. این درس بخش چهارم را با طراحی همین مسیر تمام می‌کند:

1. Exchange Feed WebSocket: Snapshot دفتر، به‌روزرسانی‌های تدریجی و معامله‌ها
2. Ingestor دریافت پیام، ثبت زمان دریافت، بدون هیچ منطق دیگر
3. Raw Store ذخیره دقیقاً همان چیزی که رسید؛ فقط اضافه‌شونده
4. Normalizer تبدیل به ستون‌های نوع‌دار، حذف تکراری، بررسی شماره ترتیب
5. Book Builder Snapshot به‌علاوه تغییرات، تشخیص گپ و همگام‌سازی دوباره
6. Normalized Store Parquet پارتیشن‌بندی‌شده، به‌علاوه Snapshotهای دوره‌ای
7. Consumers تحقیق (درس ۱۴ و ۱۵)، Backtest (درس ۱۷) و استراتژی زنده

نکته کلیدی این طراحی، جدا بودن Ingestor از بقیه است. Ingestor فقط یک کار دارد: پیام را بگیرد، زمان دریافت را کنارش بگذارد و ذخیره کند. هر منطق دیگری (Parse، اعتبارسنجی، ساختن دفتر) ممکن است باگ داشته باشد؛ اگر داده خام را داشته باشی، بعد از رفع باگ همه چیز را دوباره می‌سازی. اگر نداشته باشی، آن داده برای همیشه از دست رفته است.

۲. دریافت: Snapshot و به‌روزرسانی تدریجی

فرستادن کل دفتر سفارش در هر تغییر، پهنای باند زیادی می‌خواهد. برای همین فیدهای دفتر سفارش معمولاً دو نوع پیام دارند (قالب دقیق در هر صرافی فرق دارد؛ اینجا یک قالب فرضی و عمومی را به کار می‌بریم):

• Snapshot: کل دفتر (یا چند ده سطح بالای آن) در یک لحظه، همراه با یک شماره ترتیب (Sequence Number).
• Diff: فقط سطح‌هایی که تغییر کرده‌اند، هر کدام با شماره ترتیب بعدی.

یک قرارداد مهم که باید در مستندات هر فید چک کنی: در بیشتر فیدهای دفتر سطحی، عدد هر سطح در Diff حجم جدید کل آن سطح است، نه مقدار تغییر. پیام «قیمت 100,030، حجم 700» یعنی حالا ۷۰۰ Lot آنجاست، نه اینکه ۷۰۰ Lot اضافه شده. حجم صفر یعنی آن سطح حذف شده است.

همگام‌سازی اولیه

Snapshot معمولاً از یک درخواست جداگانه (مثلاً REST) می‌آید و Diffها از WebSocket. چون این دو هم‌زمان نیستند، ترتیب کار این است:

1. به جریان Diff وصل شو و پیام‌ها را در یک بافر نگه دار، ولی هنوز اعمال نکن.
2. Snapshot را بگیر. فرض کن شماره ترتیبش S است.
3. همه Diffهای بافر با شماره کمتر یا مساوی S را دور بریز؛ اثرشان در Snapshot هست.
4. اولین Diff اعمال‌شده باید شماره S+1 داشته باشد. از آن به بعد، هر پیام باید دقیقاً یکی بیشتر از قبلی باشد.

اگر ترتیب را برعکس کنی (اول Snapshot، بعد اتصال به WebSocket)، تغییرهای بین این دو لحظه را از دست می‌دهی و دفترت از همان ابتدا غلط است؛ بدون اینکه بفهمی.

گپ و همگام‌سازی دوباره

اگر بعد از پیام 102 پیام 104 برسد، پیام 103 گم شده است. نمی‌دانی 103 چه چیزی را تغییر داده بود؛ پس دفترت از این لحظه قابل اعتماد نیست. تنها کار درست: دفتر را نامعتبر اعلام کن، اعمال Diff را متوقف کن و از اول همگام شو (Resync). حدس زدن یا نادیده گرفتن گپ، دفتری می‌سازد که ممکن است سطحی را نشان دهد که ساعت‌ها پیش حذف شده است.

۳. پاک‌سازی: داده خراب چه شکلی دارد؟

مشکل	از کجا می‌آید	برخورد
پیام تکراری	ارسال دوباره بعد از اتصال مجدد، یا دو اتصال موازی	حذف بر اساس شماره ترتیب یا شناسه معامله
پیام نامرتب	ادغام چند اتصال یا چند منبع	مرتب‌سازی بر اساس شماره ترتیب، نه Timestamp
گپ	قطع اتصال، پیام گم‌شده	Resync و ثبت صریح بازه گم‌شده
دفتر Crossed	گپ تشخیص‌داده‌نشده یا باگ در Parse	نامعتبر اعلام کردن دفتر و Resync
قیمت پرت	خطای Parse، واحد اشتباه	علامت زدن، نه حذف بی‌صدا
Timestamp عقب‌رونده	ساعت سرور یا کلاینت	تکیه بر شماره ترتیب برای ترتیب، و بررسی جداگانه ساعت

دو قاعده پشت این جدول است. اول، ترتیب را از شماره ترتیب بگیر، نه از زمان. در یک اتصال TCP پیام‌ها به ترتیب می‌رسند، ولی Timestampها ممکن است مساوی یا حتی عقب‌رونده باشند. دوم، بازه‌های گم‌شده را ثبت کن. اگر از 10:14 تا 10:16 داده نداری، Backtest باید این را بداند و در آن بازه معامله نکند؛ پر کردن جای خالی با آخرین دفتر معلوم یعنی شبیه‌سازی بازاری که وجود نداشت.

دفتر Crossed یعنی بهترین Bid بزرگ‌تر یا مساوی بهترین Ask باشد. در یک بازار با یک موتور معاملات این حالت نباید پیش بیاید، چون همان‌طور که در درس ۳ دیدیم و در درس ۱۳ تستش کردیم، موتور چنین سفارش‌هایی را فوراً معامله می‌کند. پس اگر دفتر بازسازی‌شده تو Crossed شد، تقریباً همیشه تقصیر داده یا کد توست. این بررسی، خط دفاع دوم پشت شماره ترتیب است.

دو زمان برای هر پیام

هر رویداد را با دو Timestamp ذخیره کن:

• exchange_ts: زمانی که صرافی می‌گوید رویداد رخ داده.
• recv_ts: زمانی که پیام به سیستم تو رسید، با ساعت خودت (که باید با NTP یا PTP همگام باشد).

اختلاف این دو، تأخیر تو از بازار است و در درس ۱۹ به آن برمی‌گردیم. ولی نکته مهم‌تر برای تحقیق این است: استراتژی تو در recv_ts از رویداد باخبر شد، نه در exchange_ts. Backtestی که تصمیم‌ها را بر اساس exchange_ts بگیرد، کمی از آینده خبر دارد؛ همان باگ نگاه به آینده درس ۱۳ در مقیاس میلی‌ثانیه، و برای بازارساز که روی همین مقیاس رقابت می‌کند، کاملاً جدی.

۴. چیدمان ذخیره‌سازی

داده را در دو لایه نگه دار:

data/
├── raw/                                   # exactly as received, compressed, append-only
│   └── venue=example/channel=depth/symbol=BTCUSDT/date=2027-01-18/hour=10.jsonl.zst
└── normalized/                            # typed, deduplicated, rebuildable from raw
    ├── book_delta/symbol=BTCUSDT/date=2027-01-18/part-000.parquet
    ├── book_snapshot/symbol=BTCUSDT/date=2027-01-18/part-000.parquet
    ├── trades/symbol=BTCUSDT/date=2027-01-18/part-000.parquet
    └── gaps/symbol=BTCUSDT/date=2027-01-18.parquet

• لایه خام (Raw): هر پیام همان‌طور که رسید، یک خط JSON با recv_ts کنارش، فشرده با الگوریتمی مثل zstd. هیچ‌وقت ویرایش نمی‌شود.
• لایه نرمال‌شده (Normalized): ستون‌های نوع‌دار (درس ۱۴) در Parquet. هر ردیف book_delta یک تغییر سطح است: recv_ts، exchange_ts، seq، side، price، size. این لایه را هر وقت لازم شد از لایه خام دوباره می‌سازی.
• book_snapshot: هر چند دقیقه، دفتر بازسازی‌شده کامل را ذخیره کن. برای ساختن دفتر در ساعت ۱۵ لازم نیست از نیمه‌شب همه Diffها را اعمال کنی؛ از نزدیک‌ترین Snapshot قبل از آن شروع می‌کنی.
• gaps: بازه‌های بی‌داده، تا مصرف‌کننده بداند کجا نباید به داده اعتماد کند.

مسیرهای key=value را Hive Partitioning می‌گویند. Polars با scan_parquet و بسیاری از موتورهای تحلیلی این ساختار را می‌شناسند و وقتی روی symbol یا date فیلتر کنی، پوشه‌های نامربوط را اصلاً باز نمی‌کنند؛ همان Predicate Pushdown درس ۱۴، این بار در سطح پوشه. اندازه فایل‌ها را هم معقول نگه دار: هزاران فایل چندکیلوبایتی به اندازه یک فایل غول‌پیکر دردسر دارند.

۵. بازسازی Order Book

در درس ۳ موتور معاملاتی ساختیم که تک‌تک سفارش‌ها و صفشان را نگه می‌داشت. به این سطح جزئیات L3 (Order-by-Order) می‌گویند. بیشتر فیدهای عمومی بازار رمزارز فقط L2 می‌دهند: حجم کل هر سطح قیمت، بدون اینکه بدانی پشتش چند سفارش است. پس دفتری که می‌سازیم ساده‌تر است: یک dict از قیمت به حجم برای هر طرف. همان قراردادهای درس ۳ را نگه می‌داریم: قیمت عدد صحیح بر حسب Tick و حجم عدد صحیح بر حسب Lot.

کد زیر Snapshot و Diffها را اعمال می‌کند و سه دفاع این درس را دارد: حذف تکراری و کهنه، تشخیص گپ، و بررسی Crossed.

from dataclasses import dataclass, field

@dataclass
class L2Book:
    """Price-level book: price (ticks) -> total size (lots). No individual orders."""
    seq: int = -1          # last applied sequence number
    synced: bool = False
    bids: dict[int, int] = field(default_factory=dict)
    asks: dict[int, int] = field(default_factory=dict)

    def on_snapshot(self, msg: dict) -> str:
        self.bids, self.asks = dict(msg["bids"]), dict(msg["asks"])
        self.seq, self.synced = msg["seq"], True
        return "snapshot loaded"

    def on_diff(self, msg: dict) -> str:
        if not self.synced:
            return "ignored: waiting for snapshot"
        if msg["seq"] <= self.seq:
            return "dropped: duplicate or stale"
        if msg["seq"] != self.seq + 1:
            self.synced = False
            return f"GAP: expected {self.seq + 1} -> resync"
        for levels, updates in ((self.bids, msg.get("bids", [])), (self.asks, msg.get("asks", []))):
            for price, size in updates:
                if size == 0:
                    levels.pop(price, None)   # size 0 removes the level
                else:
                    levels[price] = size      # new absolute size, not a delta
        self.seq = msg["seq"]
        if self.bids and self.asks and max(self.bids) >= min(self.asks):
            self.synced = False
            return "CROSSED book -> resync"
        return "applied"

    def top(self) -> str:
        if not (self.synced and self.bids and self.asks):
            return "-"
        return f"{max(self.bids)} / {min(self.asks)}"

حالا یک جریان مصنوعی که از دفتر درس ۱ شروع می‌شود و همه مشکلات بخش ۳ را یک‌جا دارد:

stream = [
    {"type": "snapshot", "seq": 100,
     "bids": [[99990, 400], [99980, 700], [99970, 1200]],
     "asks": [[100010, 300], [100020, 500], [100030, 900]]},
    {"type": "diff", "seq": 100, "bids": [[99990, 400]]},          # buffered before the snapshot
    {"type": "diff", "seq": 101, "asks": [[100010, 0], [100020, 0], [100030, 700]]},  # lesson 3 sweep
    {"type": "diff", "seq": 101, "asks": [[100010, 0], [100020, 0], [100030, 700]]},  # re-sent
    {"type": "diff", "seq": 102, "bids": [[100000, 200]]},         # new bid inside the spread
    {"type": "diff", "seq": 104, "asks": [[100025, 150]]},         # 103 was lost
    {"type": "diff", "seq": 105, "bids": [[99990, 0]]},
    {"type": "snapshot", "seq": 105,                                # fresh snapshot after resync
     "bids": [[100000, 200], [99980, 700], [99970, 1200]],
     "asks": [[100025, 150], [100030, 400]]},
    {"type": "diff", "seq": 106, "bids": [[100035, 100]]},         # bid above best ask
]

book = L2Book()
for msg in stream:
    handler = book.on_snapshot if msg["type"] == "snapshot" else book.on_diff
    result = handler(msg)
    print(f"{msg['type']:8} seq={msg['seq']}  {result:32} top: {book.top()}")

snapshot seq=100  snapshot loaded                  top: 99990 / 100010
diff     seq=100  dropped: duplicate or stale      top: 99990 / 100010
diff     seq=101  applied                          top: 99990 / 100030
diff     seq=101  dropped: duplicate or stale      top: 99990 / 100030
diff     seq=102  applied                          top: 100000 / 100030
diff     seq=104  GAP: expected 103 -> resync      top: -
diff     seq=105  ignored: waiting for snapshot    top: -
snapshot seq=105  snapshot loaded                  top: 100000 / 100025
diff     seq=106  CROSSED book -> resync           top: -

خط به خط:

• seq=100 (Diff): در بافر همگام‌سازی اولیه بود و اثرش در Snapshot هست؛ دور ریخته می‌شود.
• seq=101: همان خرید Market یک‌بیت‌کوینی درس ۳: سطح‌های 100,010 و 100,020 حذف می‌شوند و 100,030 از ۹۰۰ به ۷۰۰ Lot می‌رسد. Spread از ۲۰ به ۴۰ دلار باز می‌شود.
• seq=101 دوباره: تکراری؛ اگر اعمال می‌شد اینجا ضرری نداشت (حجم مطلق است)، ولی در فیدهایی که تغییر نسبی می‌فرستند، حجم را دو بار کم می‌کرد.
• seq=104: پیام 103 نیامده. دفتر نامعتبر می‌شود و top دیگر چیزی گزارش نمی‌کند. این مهم است: دفتر نامعتبر باید خاموش باشد، نه اینکه آخرین مقدارش را نشان دهد. استراتژی‌ای که از این دفتر قیمت می‌گیرد باید بفهمد داده ندارد و سفارش‌هایش را مطابق قواعد ریسک (درس ۲۷) مدیریت کند.
• Snapshot جدید در seq=105: دفتر از نو ساخته می‌شود. Ingestor بازه بین گپ و این Snapshot را در gaps ثبت می‌کند.
• seq=106: Bid در 100,035، بالاتر از Ask در 100,025. شماره ترتیب درست است، ولی دفتر Crossed شده؛ پس داده یا Parse مشکل دارد و دوباره Resync لازم است.

۶. Replay: پخش دوباره بازار

حلقه پایانی کد بالا در واقع یک Replay است: پیام‌های ذخیره‌شده را به ترتیب می‌خواند و همان کدی را اجرا می‌کند که روی داده زنده اجرا می‌شود. این ایده پایه همه چیزهایی است که در بخش بعد می‌سازیم. چند اصل:

• همان کد برای زنده و Replay. اگر Book Builder زنده و Book Builder تحقیق دو پیاده‌سازی جدا باشند، دیر یا زود از هم فاصله می‌گیرند و نتیجه Backtest با اجرای واقعی نمی‌خواند.
• از Checkpoint شروع کن. برای Replay یک ساعت مشخص، آخرین book_snapshot قبل از آن را بار کن و فقط Diffهای بعدش را اعمال کن.
• خودت را بسنج. وقتی در Replay به زمان یک book_snapshot ذخیره‌شده رسیدی، دفتر بازسازی‌شده باید دقیقاً با آن برابر باشد. هر اختلاف یعنی باگ در Pipeline.
• قطعی (Deterministic) باشد. دو اجرای Replay روی یک داده باید دقیقاً یک خروجی بدهند. اگر چند جریان داری (مثلاً دفتر و معامله‌ها)، آن‌ها را با یک قاعده ثابت ادغام کن: بر اساس recv_ts و با یک کلید یکتا برای شکستن تساوی؛ همان درس تساوی زمانی درس ۱۵.

Replay می‌تواند با سرعت حداکثر اجرا شود (برای تحقیق) یا با سرعت واقعی (برای آزمودن سیستم زنده روی داده گذشته). در هر دو حالت، ترتیب و زمان‌بندی رویدادها همان است که سیستم تو در واقعیت دید؛ نه نسخه تمیز و ایده‌آل آن.

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

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

سؤال ۱: Snapshot با شماره ترتیب 500 رسید. در بافر Diffهایی با شماره 498، 499، 500، 501 و 502 داری. اولین Diffی که باید اعمال شود کدام است؟

501 498 500

سؤال ۲: آخرین پیام اعمال‌شده شماره 102 دارد و پیام بعدی شماره 104 دارد. کار درست چیست؟

پیام 104 را اعمال کن؛ یک پیام مهم نیست دفتر را نامعتبر اعلام کن و با Snapshot جدید دوباره همگام شو منتظر بمان و هر وقت 103 رسید آن را بعد از 104 اعمال کن

سؤال ۳: سطح 100,020 در دفتر ۵۰۰ Lot دارد. Diff بعدی برای همین سطح حجم 350 می‌فرستد (فیدی که حجم کل جدید را می‌فرستد). حجم جدید این سطح چقدر است؟

850 Lot 150 Lot 350 Lot

سؤال ۴: چرا Backtest باید تصمیم‌ها را بر اساس recv_ts بگیرد، نه exchange_ts؟

چون استراتژی فقط وقتی از رویداد باخبر می‌شود که پیام به سیستمش برسد؛ استفاده از exchange_ts یعنی کمی نگاه به آینده چون exchange_ts همیشه غلط است چون recv_ts دقت بیشتری دارد

بررسی پاسخ‌ها

درس بعدی

با این درس بخش چهارم تمام می‌شود و حالا داده تمیز، دفتر سفارش قابل بازسازی و ابزار Replay داریم. در درس هفدهم، اولین درس بخش پنجم، سراغ طراحی Backtest واقع‌گرایانه می‌رویم: چرا Backtest ساده‌ای که فرض می‌کند هر سفارش Limit به محض لمس قیمت پر می‌شود، تقریباً همیشه خوش‌بینانه است، و یک Backtest رویدادمحور روی همین Replay چطور باید ساخته شود تا نتیجه‌اش به اجرای واقعی نزدیک باشد.