محرك التحليلات التدفقية

اللوحات التي تعرض “المستخدمين النشطين الآن” لا إعادة حساب قاعدة البيانات الكاملة في كل نبضة ، بل تستهلك تدفقًا لا نهائيًا من الأحداث وتحتفظ بنافذة صغيرة محدّثة باستمرار لما حدث للتو. يبني هذا المشروع ذلك المحرك بلغة بايثون النقية: مولّد يُصدر تدفق أحداث واقعيًا، ونافذة منزلقة تُحدّث المتوسطات، واكتشاف القمم مقارنةً بخط أساس متحرك، وانضمام يربط عمليات الشراء بالصفحات التي سبقتها، وأخيرًا ذاكرة مؤقتة محدودة تبطئ التدفق عند القمم بدلاً من انفجار الذاكرة.

يُفترض هنا أساسيات بايثون ومعرفة المولّدات ، لا حاجة لحزم خارجية أو أي شيء من تحليل البيانات. هذا اختياري وغير مُقيَّم؛ راجع المشاريع الواقعية للقائمة الكاملة المتنامية.

🎯 ما ستفعله

1. اكتب مولّدًا يُصدر تدفقًا غير محدود من الأحداث مُخمّلًا بالوقت.
2. حافظ على نافذة زمنية منزلقة وأخرج متوسطًا محدّثًا لكل حدود.
3. علّم الأحداث التي تتجاوز خطًا أساسيًا متحركًا، لا رقمًا ثابتًا.
4. اربط أحداث الشراء بكل عرض صفحات سابق للمستخدم.
5. حدد التدفق بذاكرة مؤقتة محدودة وشغّل كل المراحل من البداية إلى النهاية.

أين تُشغّل هذا

محليًا باستخدام uv هو المسار الأساسي. هذا المحرك مكتبة معيارية نقية ، uv init وستعمل فورًا ، وكل مرحلة هي دالة يمكنك استدعاؤها وفحصها وإعادة تشغيلها من الطرفية تمامًا كما هو مكتوب أدناه.

Google Colab و Kaggle Notebooks و Binder تشغّل كل الخطوات بشكل متماثل، لأن لا تبعيّات خارجية للتثبيت ولا ملفات تحتاج أن تبقى بين الخلايا. التنبيه الصادق: خلايا الدفتر تستبدل مخرجات الطرفية للمحرك بمخرجات الدفتر، فما تفقده هو إحساس “أعد تشغيل التدفق وشاهد التغيّر”. استخدم الشارات لرؤية خط التدفق الكامل بنقرة واحدة، وانتقل إلى uv المحلي حين تريد توجيه المولّد إلى ملف أو مأخذ حقيقي.

الإعداد

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

uv init streaming-analytics
cd streaming-analytics

uv run python -c "from collections import deque; import random, datetime; print('ok')"

الاستيرادات الثلاثة تغطي اعتمادية هذا المشروع بالكامل: deque للنوافذ المنزلقة (الخطوة 2 فصاعدًا)، random للتدفق الاصطناعي (الخطوة 1)، و datetime/timedelta لإطارات الأحداث الزمنية التي تُقاس كل نافذة ضدها.

✅ قائمة التحقق

• ✅ uv init streaming-analytics أنشئ مجلدًا بملف pyproject.toml.
• ✅ uv run python -c "from collections import deque; import random, datetime" طبع ok ، صفر حزم مُضافة.

الخطوة 1: ابني مولّد تدفق أحداث مباشر

كل محرك تحليلات يبدأ من نفس المكان: أحداث تصل واحدة تلو الأخرى، إلى الأبد. مولّدات بايثون هي الطريقة الصادقة لنمذجة ذلك ، دالة تُصدر الأحداث بكسل تبدو تمامًا كمباشرة حية لكل ما يليها، دون الحاجة فعلًا إلى خادم.

1.1 أصدر أحداثًا مُخمّلة بالوقت

👟 تلميح البداية : أنشئ فئة بيانات لشكل الحدث، ثم مولّدًا يُصدر حدثًا واحدًا في كل تكرار بإطار زمني متصاعد بشكل مفرط ومُfixed ببذرة.

# stream.py
import random
from dataclasses import dataclass
from datetime import datetime, timedelta

@dataclass
class Event:
    ts: datetime
    kind: str
    value: float

def event_stream(events: int = 50, seed: int = 3):
    """Yield events lazily, as if arriving from a live feed."""
    random.seed(seed)
    now = datetime(2026, 1, 1, 9, 0, 0)
    for _ in range(events):
        kind = random.choice(["view", "click", "purchase"])
        value = {"view": random.randint(5, 15),
                 "click": random.randint(1, 4),
                 "purchase": random.choice([0, 1])}[kind]
        now += timedelta(seconds=random.randint(1, 3))
        yield Event(ts=now, kind=kind, value=value)

for ev in event_stream(5):
    print(ev.ts.strftime("%H:%M:%S"), ev.kind, ev.value)

@dataclass تعطيك كائن Event مقروءًا وقابلًا للتماثل دون كتابة مُنشئ. المولّد هو الفكرة الحاملة للحمولة: event_stream لا تحسب شيئًا حتى يبدأ بالتكرار، وكل yield يُعلّقه في منتصف الحلقة ، تمامًا كشكل تدفق يستمر في الإنتاج بعد استهلاكك 50 حدثًا. تتقدم الطوابع الزمنية بثانية واحدة إلى ثلاث عشوائيًا لكل حدث، فتمنح النافذة والانضمامات توقيتًا واقعيًا و غير منتظمًا للعمل به بدلاً من نبض ثابت تمامًا.

🎯 الناتج المتوقع : خمسة أسطر مثل 09:00:00 click 3، كل منها بإطار زمني لاحق للسابق وأحد أنواع الأحداث الثلاثة.

🩹 إذا لم يعمل : إذا كانت كل الأطارات الزمنية متطابقة، ف now += مفقود فالساعة لا تتقدم أبدًا. إذا أدى التكرار مرتين إلى أنواع مختلفة، ف random.seed(seed) مفقود، مما يجعل التدفق غير قابل للتكرار. إذا بدا Event غير قابل للتغليف أو مُحطّمًا، فمزخرف @dataclass مفقود فاختصارات المساواة/__repr__ غير موجودة.

1.2 تحقق من التدفق

✅ قائمة التحقق

• ✅ event_stream(5) طبع 5 أحداث بأطارات زمنية صاعدة بشكل صارم.
• ✅ البذرة نفسها تُنتج تسلسل أحداث مطابق في عمليات التشغيل المتكررة.
• ✅ يمكنك تفسير لماذا يُمثّل المولّد تدفقًا مباشرًا أفضل من إرجاع قائمة جاهزة.

🤔 سؤال (أسئلة) سقراطي(ة)

• عند استدعاء event_stream(50)، لا أحداث موجودة بعد ، أين يُنفق كود الذاكرة قبل استدعاء next() الأول، ولماذا هذا بالضبط ما يريدته أداة استهلاك التدفق الحقيقية؟
• كل حدث يتقدم الساعة بثانية واحدة إلى ثلاث عشوائيًا. ماذا سيتغيّر في نوافذ الخطوة 2 إذا كان timedelta دائمًا بالضبط ثانيتين؟

الخطوة 2: أضف نافذة زمنية منزلقة

تدفق لا يمكنك تلخيصه هو ضوضاء فحسب. تبني هذه الخطوة نافذة منزلقة ، “آخر 10 ثوانٍ من الأحداث، مُحدّثة باستمرار” ، وتُصدر متوسطًا متحركًا في كل مرة تنزلق فيها النافذة إلى الأمام، وهو شكل رقم “النشاط الأخير” المباشر.

2.1ُجمّع آخر window_s ثوانٍ

👟 تلميح البداية : استخدم deque كنافذة، ادفع كل حدث على اليمين، popleft لكل ما هو أقدم من window_s على اليسار، وأخرج المتوسط عندما تتجاوز الساعة حدود الخطوة.

# stream.py (continued)
from collections import deque

def windowed_average(stream, window_s: int = 10, step_s: int = 3):
    """Emit the average of the last window_s seconds at each step boundary."""
    window: deque[Event] = deque()
    boundary = None
    for ev in stream:
        window.append(ev)
        while (ev.ts - window[0].ts).total_seconds() > window_s:
            window.popleft()
        if boundary is None or ev.ts >= boundary:
            boundary = ev.ts + timedelta(seconds=step_s)
            avg = sum(e.value for e in window) / len(window)
            yield ev.ts, round(avg, 2)

for ts, avg in windowed_average(event_stream(30), window_s=10, step_s=4):
    print(ts.strftime("%H:%M:%S"), "window avg:", avg)

شيئان يجعلان هذا O(1)- تقريبًا لكل حدث بدلاً من إعادة مسح التاريخ: الـ deque ، الذي إضافته على اليمين وإزالته من اليسار كلاهما وقت ثابت ، وحلقة while التي تُزيل الأحداث منتهية الصلاحية بمقارنة window[0]، الناجي الأقدم. لأن الأحداث تصل بترتيب الأطارات الزمنية، يكفي فحص واحد من اليسار للحفاظ على نظافة النافذة بالكامل. منطق boundary هو ما يحوّل نافذة مستمرة إلى إخراج دوري: لا يُصدر إلا عندما يتجاوز الحدث الأخير حد الخطوة التالي، فيحصل على متوسط واحد مقروء لكل خطوة بدلاً من واحد لكل حدث.

🎯 الناتج المتوقع : بضعة أسطر مطبوعة، مثل 09:00:13 window avg: 6.67، واحد لكل حدود الخطوة، كل منها يغطي تقريبًا آخر 10 ثوانٍ محاكاة.

🩹 إذا لم يعمل : إذا كان متوسط كل صف ضخمًا، فحلقة الإزالة while مفقودة والنافذة تنمو إلى الأبد. إذا لم يُطبع شيء، فالتدفق الذي مرّره يحتوي على أحداث أقل من خطوة واحدة ، مرّر عدّ events أكبر. إذا بدت الأطارات الزمنية تتداخل بشكل غريب، ف window_s/step_s متبديلان، مما يجعل النافذة أطول من المدخلات.

2.2 تحقق من النافذة

✅ قائمة التحقق

• ✅ متوسط واحد يُطبع كل ~4 ثوانٍ محاكاة، كل منها يغطي ~10 ثوانٍ سابقة.
• ✅ حجم النافذة يبقى محدودًا: إعادة التشغيل بأحداث أكثر لا تزيد قط عدد الأحداث المُحتفظ بها في وقت واحد.
• ✅ يمكنك تفسير لماذا window[0] هو فحص انتهاء الصلاحية الوحيد المطلوب.

🤔 سؤال (أسئلة) سقراطي(ة)

• تُصدر النافذات المتوسطات على حدود خطوة ثابتة بدلاً من لكل حدث. في أي لوحة م control حقيقية سيُسبب العين على الحدود انحرافًا سيئًا، وماذا ستغيّر لإصدار واحد لكل حدث بالضبط؟
• تحفظ النافذة value وتُعيد حساب المجموع في كل إخراج. ما المتغيرات المتحركة المنفصلة التي ستجعل إصدار المتوسط وقتًا ثابتًا بالضبط بغض النظر عن طول النافذة؟

الخطوة 3: اكتشف القمم مقارنةً بخط أساس متحرك

كشف الشذوذ في تدفق لا يمكنه استخدام حد ثابت ، حركة المرور ترتفع طبيعًا في وقت الغداء وتنخفض في الساعة 3 صباحًا. تُعلّم هذه الخطوة الأحداث التي تتجاوز خطًا أساسيًا متحركًا، ف “مرتفع جدًا” يعني “مرتفعًا الآن”.

3.1 علّم الأحداث فوق المتوسط المباشر

👟 تلميح البداية : حافظ على deque من القيم الحديثة كخط أساسي، احسب متوسطه، وأصدر كل ما يتجاوز المتوسط بمعامل مُتحوّل مُعرّف.

# stream.py (continued)
def detect_spikes(stream, window_s: int = 15, multiplier: float = 3.0):
    """Yield events whose value exceeds `multiplier * recent-average`."""
    recent: deque[Event] = deque()
    for ev in stream:
        recent.append(ev)
        while (ev.ts - recent[0].ts).total_seconds() > window_s:
            recent.popleft()
        baseline = sum(e.value for e in recent) / len(recent)
        if baseline > 0 and ev.value > multiplier * baseline:
            yield ev.ts, ev.kind, ev.value, round(baseline, 2)

for ts, kind, value, baseline in detect_spikes(event_stream(200), window_s=15, multiplier=2.5):
    print(ts.strftime("%H:%M:%S"), f"{kind:>8} {value:>3} vs baseline {baseline}")

الفكرة هي المقارنة مع ما عليه التدفق الآن، لا مع متوسط عام. مع multiplier=2.5، يُشغّل view بقيمة 25 تنبيهًا عندما كان متوسط آخر 15 ثانية 10، لكن نفس القيمة تبقى صامتة إذا كان الخط الأساسي بالفعل 30 ، لأن حدثًا طبيعيًا لفترة مزدحمة هو قمة في فترة هادئة. حرس baseline > 0 مهم: نافذة تحتوي بالصدفة على أصفار فقط يجب ألا تحوّل المقارنة إلى 0 > 0 مرضيّ التقييم.

🎯 الناتج المتوقع : أسطر أقل من أحداث المدخلات (200 → بالكاد handful)، كل منها يعرض قيمة حدث أعلى بكثير من خطه الأساسي المتحرك ، ليس فيضًا من كل حدث.

🩹 إذا لم يعمل : إذا طبع كل حدث، فالمعامل منخفض جدًا أو نافذة الخط الأساسي قصيرة جدًا لا تحتوي سوى على الحدث الأكثر صخبًا. إذا شاركت مخرجتان متتاليتان نفس الإطار الزمني، فحلقة الإزالة while مفقودة فالخط الأساسي يُدرج أحداثًا مستقبلية… ماضية إلى الأبد. إذا لم يُطبع شيء على الإطلاق، ف multiplier=2.5 غير محتمل مع البذرة التي استخدمتها ، جرّب 1.5 لترى كشف fired.

3.2 تحقق من كشف القمم

✅ قائمة التحقق

• ✅ detect_spikes لا يطبع سوى جزء صغير من التدفق.
• ✅ كل حدث مُعلّم قيمته تتجاوز 2.5× خطه الأساسي المتحرك.
• ✅ يمكنك تفسير لماذا نفس القيمة المطلقة تكون قمة أحيانًا وأحيانًا لا.

🤔 سؤال (أسئلة) سقراطي(ة)

• أسبوع بطيء ومستقر يعني أن الخط الأساسي المتحرك هو القمة ، ارتداد تدريجي لا يتجاوز أبدًا 2.5×. ما الفحص الإضافي الذي سيلقط اتجاهًا ينتقل من 10 إلى 30 عبر ساعة؟
• المعامل ثابت. كيف سي behaved الكاشف على منصة عادة هادئة لديها ارتفاع سنوي مشروع، وماذا تحتاج للحفاظ على فائدة التنبيه أثناء ذلك الارتفاع؟

الخطوة 4: اربط تدفقتين مترابطتين

عرض صفحة منفرد غير ملحوظ؛ عرض صفحة يليه بسرعة شراء من نفس المستخدم هو القصة. الربط يُrelate أحداثًا تشير إلى نفس المفتاح (هنا، مستخدم) ضمن ميزانية زمنية ، قريب هادئ من JOIN في SQL، لكنه يُنفّذ عبر الوقت بدلاً من الجداول.

4.1 اربط عمليات الشراء بالعروض السابقة

👟 تلميح البداية : أعطِ التدفق مفتاح user، تذكّر وقت آخر عرض صفحات لكل مستخدم، وأصدر صف “محوّل” عندما يصل شراء ضمن نافذة النظر إلى الخلف.

# stream.py (continued)
def user_stream(events: int = 80, seed: int = 5):
    random.seed(seed)
    users = [f"u{i}" for i in range(8)]
    now = datetime(2026, 1, 1, 9, 0, 0)
    for _ in range(events):
        uid = random.choice(users)
        kind = random.choices(["page_view", "purchase"], weights=[80, 20])[0]
        now += timedelta(seconds=random.randint(1, 4))
        yield {"ts": now, "user": uid, "kind": kind}

def correlated_join(stream, lookback_s: int = 30):
    """Yield (user, seconds-after-view) for purchases within lookback_s of a view."""
    last_view: dict[str, datetime] = {}
    for ev in stream:
        if ev["kind"] == "page_view":
            last_view[ev["user"]] = ev["ts"]
        elif ev["kind"] == "purchase" and ev["user"] in last_view:
            age = (ev["ts"] - last_view[ev["user"]]).total_seconds()
            if age <= lookback_s:
                yield ev["user"], round(age, 1), "converted"

for user, age, label in correlated_join(user_stream(120)):
    print(f"{user} purchased {age}s after viewing -> {label}")

الانضمام هو قاموس بمفتاح الانضمام (user) مضافًا إليه ميزانية زمنية: last_view يتذكر فقط وقت آخر عرض لكل مستخدم، والاستشارة تقرأه بدلاً من إعادة مسح كل الأحداث السابقة. مقارنة age <= lookback_s هي ما يحوّل ارتباطًا غير مشروط إلى ارتباط محدّد بالوقت ، شراء بعد خمس دقائق من عرض قد لا يكون نفس الرحلة. لأن الذاكرة المؤقتة تحتفظ بإطار زمني واحد لكل مستخدم نشط، ذاكرتها متناسبة مع عدد المستخدمين الفريدين، لا عدد الأحداث ، السبب ذاته الذي يجعل المحركات الحقيقية تحافظ على حالة لكل مفتاح وتُزيل المفاتيح القديمة.

🎯 الناتج المتوقع : بضعة تحويلات مطبوعة (حوالي 20% من الأحداث عمليات شراء، وبعضها فقط لديه عرض خلال 30 ثانية)، كل منها مثل u3 purchased 12.3s after viewing -> converted.

🩹 إذا لم يعمل : إذا حوّل كل شراء، ففحص age <= lookback_s غير موجود أو lookback_s ضخم. إذا لم يحوّل شيء، فقيم kind في user_stream لا تتطابق مع النصوص التي يفحصها الانضمام. إذا كان عرض قديم لمستخدم يظل يطابق عمليات الشراء بعد دقائق، ف last_view[user] = ev["ts"] يُكتب فقط على العروض كما هو مقصود ، لكن المفاتيح القديمة لا تُزال أبدًا، وهو الانحراف الذي يجب مراقبته في التدفق الطويل.

4.2 تحقق من الانضمام

✅ قائمة التحقق

• ✅ كل تحويل صادر يُظهر شراءً وصل بعد عرض مستخدمه.
• ✅ عدد صفوف المخرج أقل بكثير من عدد عمليات الشراء (انضمام محدّد بالوقت).
• ✅ يمكنك تسمية مفتاح الانضمام (user) والميزانية الزمنية (lookback_s) دون النظر إلى الكود.

🤔 سؤال (أسئلة) سقراطي(ة)

• يُخزّن الانضمام فقط آخر عرض لكل مستخدم. ماذا سيتغيّر في التحويلات إذا أخزنت بدلاً من ذلك العرض الأول للمستخدم في اليوم؟
• الربط في التدفقات الحقيقية يجب أيضًا أن يُزيل المفاتيح التي لا يلمسها أحد. إذا حدّد lookback_s نافذة الانضمام، لماذا لم يُحدّد بالفعل قاموس last_view ، وماذا قد ينمو بلا حد في انضمام طويل العمر؟

الخطوة 5: حدد التدفق بالضغط الخلفي

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

5.1 أضف ذاكرة مؤقتة محدودة وشغّل كل شيء

👟 تلميح البداية : حدد الكومة المعلّقة عند max_pending أحداث، استدعِ hook عندما يُضرب الحد، ثم اجمع كل مرحلة ليعمل المحرك بالكامل من __main__ واحد.

# stream.py (continued)
def with_backpressure(stream, max_pending: int = 8, on_overflow=None):
    """Mirror a bounded queue: absorb up to max_pending events, drop the rest."""
    on_overflow = on_overflow or (lambda ev: None)
    pending: list = []
    for ev in stream:
        if len(pending) < max_pending:
            pending.append(ev)
        else:
            on_overflow(ev)
    return pending

def main() -> None:
    dropped: list = []
    def count_drop(ev): dropped.append(ev)

    feed = event_stream(300)
    buffered = with_backpressure(feed, max_pending=8, on_overflow=count_drop)

    windows = list(windowed_average(iter(buffered), window_s=10, step_s=4))
    spikes = list(detect_spikes(iter(buffered), window_s=15, multiplier=2.5))
    joins = list(correlated_join(user_stream(200)))

    print(f"buffered: {len(buffered)}  dropped: {len(dropped)}")
    print(f"windows emitted: {len(windows)}  spikes: {len(spikes)}  conversions: {len(joins)}")

if __name__ == "__main__":
    main()

with_backpressure يجعل المساواة مرئية: حتى max_pending أحداث تنتظر في الطابور، أي ما يتجاوزها *يُسقطويُبلغ عبر خطافon_overflowبدلاً من الضياع الصامت أو التخزين الصامت. تجميع كل مرحلة عبرiter(buffered)يُظهر الخاصية الأخرى التي تجربها ، كل دالة تابعة في الخطوات 2–4 تستهلك أي iterable بكسل، فيبقى التدفق سلسلة من القارئين الصغيرين بدلاً من حلقة ضخمة واحدة. عدّmain()` يعطيك إشارة من البداية إلى النهاية: النوافذ والقمم والتحويلات كلها محسوبة من نفس التدفق المحدد، والفيض مرئي كرقم بدلاً من انهيار.

🎯 الناتج المتوقع : ملخص واحد، مثل buffered: 300 dropped: 0 windows emitted: 54 spikes: 9 conversions: 4 ، كل مرحلة عملت، لم يُ raising أي شيء.

🩹 إذا لم يعمل : إذا ظهر TypeError حول معامل مفقود، فمرحلة تُمرّر لها نتيجة مرحلة بدلاً من iterable ، مرّر iter(buffered) بثبات. إذا كان dropped غير صفري على تدفق 300 حدث، ف max_pending=8 يُضرب في منتصف التدفق، وهو سلوك صحيح؛ تأكد من أن السقوط يطابق نواياك قبل الذعر. إذا احتاج detect_spikes تدفقًا أطول، زِد عدّ events لا المعامل.

5.2 تحقق من البداية إلى النهاية

✅ قائمة التحقق

• ✅ uv run python stream.py طبع ملخصًا بدون تتبع خطأ.
• ✅ كل مرحلة من القائمة استهلكت نفس التدفق المحدد buffered.
• ✅ main() محمية بـ if __name__ == "__main__": فاستيراد stream.py في اختبار لا يشغّل خط التدفق.

🤔 سؤال (أسئلة) سقراطي(ة)

• with_backpressure يُسقط الأحداث بدلاً من حظر المنتج. ماذا يخسر المستهلك عند السقوط أثناء الذروة، وماذا تسجّل بجانب كل حدث مُسقط لجعل الخسارة قابلة للتدقيق؟
• المراحل الأربع تقرأ نفس القائمة buffered بالتتابع، فيجب على خط التدفق بالكامل إنهاء المرحلة 1 قبل بدء المرحلة 2. ماذا سيتغيّر في زمن الاستجابة إذا عملت المراحل بالتوازي ، وما مشكلة المزامنة التي ستحتاج فجأة لحلها؟

⚠️ المآزق الشائعة

• إعادة توليد التدفق في منتصف خط التدفق. تستهلك المراحل المُكرّرات مرة واحدة؛ استدعاء event_stream() مرة ثانية يُنتج تدفقًا جديدًا (مُخمّلًا)، فتُحسب النوافذ والقمم فوق أحداث مختلفة وتخالف الأرقام. الإصلاح: ولّد مرة واحدة ومرّره (أو iter(buffered)) لكل مرحلة، كما في الخطوة 5.
• نوافذ لا تُزيل أبدًا. نسيان حلقة while … popleft يجعل النافذة تنمو إلى الأبد، فيصبح “متوسط آخر 10 ثوانٍ” صامتًا “متوسط كل شيء حتى الآن”. الإصلاح: دائمًا أزل من اليسار بعد الإضافة.
• مقارنة حد ثابت بدلاً من خط أساسي. value > 25 مُثبت ي fired باستمرار خلال ساعات العمل الحركي ولا يُ fired أبدًا في الأوقات الهادئة؛ multiplier * rolling_average من الخطوة 3 هو ما يحافظ على الكشوف نسبيةً مع حركة المرور الحالية.
• انضمام على قاموس لا ينتهي أبدًا. last_view ينمو بخانة واحدة لكل مستخدم فريد ولا يتناقص أبدًا، فيحتجز انضمام طويل العمر الذاكرة. الإصلاح: أزل المفاتيح القديمة التي لم تراها خلال نافذة النظر إلى الخلف.
• ذاكرة مؤقتة محدودة تُسقط بصمت. تصميم الضغط الخلفي الحقيقي لا يمكنه discard فحسب؛ يجب أن يُظهر الفيض. خطاف on_overflow في الخطوة 5 هو الفرق بين سقوط مؤدّت وضياع بيانات صامت.

ما بنيته للتو

محرك تحليلات تدفق يعمل: مولّد أحداث، مُلخّص نافذة منزلقة، كشف قمم بخط أساسي متحرك، انضمام محدّد بالوقت، وذاكرة مؤقتة محدودة بضغط خلفي مرئي ، كل مرحلة دالة صغيرة قابلة للتجميع في بايثون النقية، بدون حزم طرف ثالث. المهارة القابلة للنقل هي معالجة البيانات عند وصولها بدلاً من بعد تخزينها: بمجرد بناءك نافذة deque واحدة ومولّد واحد، تتوقف لوحات المراقبة الدقيقة وحلقات المراقبة ومعالجات الأحداث عن أن تكون غامضة وت aynı نفس الدوال الخمسة.

إلى أين تذهب من هنا

• وجّه المولّد إلى مصدر حقيقي ، ملف يُستكمل، أو مأخذ ، حتى يكون “التدفق” أحداثًا حقيقية بدلاً من عشوائية مُخمّلة.
• أضف تقسيم الجلسات للانضمام: اجمع عروض المستخدم في جلسة منطقية واحدة، ثم أسنِد الشراء للجلسة التي وقع فيها (بهذا الأسلوب تُبلّغ أدوات الربط الحقيقية “التحويلات لكل جلسة”).
• استبدل متوسط windowed_average المُعاد حسابه بمتغيرات count/sum تدريجية حتى يكون الإخراج وقتًا ثابتًا بأي طول نافذة.
• احفظ النوافذ والقمم في ملف JSONL بكاتب flush-for-batch، وحوّل خط التدفق المباشر إلى شيء تستطيع لوحة المراقبة قراءته.

شارك مشروعك مع الفصل

بنيت شيئًا تفتخر به؟ examples/student-projects/ معرض لمشاريع قدمها طلاب آخرون ، و README الخاص به يحتوي على دليل كامل ومناسب للمبتدئين لإضافة مشروعك عبر طلب سحب، حتى لو لم تستخدم git من قبل: تفرّع المستودع، وإنشاء فرع، وعمل commit لملفاتك، وفتح طلب السحب، خطوة بخطوة. لا يُفترض خبرة git مسبقة.

أهلاً بكتابة بايثون خارج المتصفح. 🎓