بث البيانات عبر websockets

اتصال يبقى مفتوحًا

الطلب العادي على الويب سؤال وجواب ثم يُغلق الاتصال. وللبيانات الحية كان ذلك يعني السؤال مرارًا: بطيء ومتأخر دائمًا قليلًا. أما websocket فيبدأ كطلب HTTP عادي ثم يترقّى إلى اتصال يبقى مفتوحًا في الاتجاهين، فيستطيع الخادم دفع كل صفقة لحظة حدوثها. وتعرض البورصات والوسطاء بثّها الحي هكذا: تتصل وترسل رسالة subscribe تسمّي ما تريد وتتلقى سيلًا من رسائل JSON صغيرة.

# A subscribe message and a trade message, both small JSON texts
import json

subscribe = json.dumps({"op": "subscribe", "channel": "trades", "symbols": ["AAPL", "MSFT"]})
print(subscribe)                                          # {"op": "subscribe", "channel": "trades", "symbols": ["AAPL", "MSFT"]}

raw = '{"s": "AAPL", "p": "184.25", "q": 100, "t": 1700000000123}'
message = json.loads(raw)
print(message["s"], float(message["p"]), message["q"])    # AAPL 184.25 100

قراءة الرسائل بأمان

كل ما يمر على الخط نص، وهذا النص ليس تحت سيطرتك. فكّ ترميزه بـ json.loads لا بـ eval التي تنفذ ما شاء المرسل كتابته. وتوقّع المفاجآت: كثيرًا ما تصل الأسعار نصوصًا لتجنب تقريب الفاصلة العائمة في الخادم، وليست كل رسالة صفقة. ترسل المصادر نبضات (heartbeats) كل بضع ثوانٍ لتميّز سوقًا هادئًا من اتصال ميت؛ تعرّف عليها وتجاوزها، واعتبر الصمت الطويل علامة على وجوب إعادة الاتصال.

# One function turns a raw message into a trade, or None for a heartbeat
import json

def parse_tick(raw):
    message = json.loads(raw)
    if message.get("type") == "hb":
        return None
    return {"symbol": message["s"], "price": float(message["p"]), "size": int(message["q"]), "ts": int(message["t"])}

print(parse_tick('{"s": "AAPL", "p": "184.25", "q": 100, "t": 1700000000123}'))
print(parse_tick('{"s": "MSFT", "p": 410.5, "q": 25, "t": 1700000000456}'))
print(parse_tick('{"type": "hb", "t": 1700000001000}'))      # None  a heartbeat carries no trade

الرسائل المفقودة وإعادة الاتصال

تنقطع الاتصالات، ولذلك يرقّم المصدر رسائله. إذا استلمت 7 ثم 11 فالرسائل 8 و9 و10 مفقودة وفي بياناتك ثغرة: اكتشفها ثم رمّمها مثلًا بطلب لقطة. وبعد الانقطاع تعيد الاتصال لكن لا فورًا ولا في اللحظة نفسها مع كل العملاء: انتظر base ثم ضعفها ثم ضعفها ثانية حتى حدٍّ أقصى. ويمنع هذا التراجع الأسي إغراق خادم متعثر. وفي الشيفرة الحقيقية تكون حلقة الاستقبال async for message in connection: داخل asyncio التي رأيتها المرة الماضية.

# Gaps in the sequence numbers show the messages that were lost
def find_gaps(seqs):
    gaps = []
    for previous, current in zip(seqs, seqs[1:]):
        if current != previous + 1:
            gaps.append((previous + 1, current - 1))
    return gaps

print(find_gaps([1, 2, 3, 4]))             # []                 nothing lost
print(find_gaps([1, 2, 3, 7, 8, 11]))      # [(4, 6), (9, 10)]  4-6 and 9-10 are missing

# Waiting twice as long before each reconnect, up to a cap
def backoff_delays(attempts, base, cap):
    return [min(base * 2 ** i, cap) for i in range(attempts)]

print(backoff_delays(6, 0.5, 8.0))         # [0.5, 1.0, 2.0, 4.0, 8.0, 8.0]
print(backoff_delays(3, 1.0, 30.0))        # [1.0, 2.0, 4.0]

# A feed consumed with async for; the websockets package gives a connection used the same way
import asyncio

class FakeConnection:
    """Stands in for a websocket: an async iterator of raw text messages."""
    def __init__(self, messages):
        self.messages = list(messages)
    def __aiter__(self):
        return self
    async def __anext__(self):
        if not self.messages:
            raise StopAsyncIteration
        await asyncio.sleep(0)
        return self.messages.pop(0)

async def consume(connection):
    trades = []
    async for raw in connection:
        if '"hb"' not in raw:
            trades.append(raw)
    return trades

feed = FakeConnection(['{"s": "AAPL", "p": 184.25}', '{"type": "hb"}', '{"s": "MSFT", "p": 410.5}'])
print(asyncio.run(consume(feed)))          # the two trades; the heartbeat is skipped

• الـ websocket اتصال يبقى مفتوحًا في الاتجاهين: تشترك مرة واحدة ثم تستقبل سيلًا من الرسائل.
• فكّ الترميز بـ json.loads لا eval؛ وتوقّع أسعارًا نصية وتجاوز النبضات.
• أرقام التسلسل تكشف الرسائل المفقودة؛ وبعد الانقطاع أعد الاتصال بانتظار يتضاعف حتى حدٍّ أقصى.

التالي: بروتوكول FIX، صيغة الرسائل التي يستخدمها الوسطاء والبورصات لتقديم الأوامر.

المختبر: ما الذي يُقيَّم

• اطلب مصدر بيانات باستخدام subscribe_message25 نقطة
• اقرأ رسالة باستخدام parse_tick25 نقطة
• اكتشف الرسائل المفقودة باستخدام find_gaps25 نقطة
• خطّط لإعادة الاتصال باستخدام backoff_delays25 نقطة