انتقل إلى المحتوى الرئيسي

الوحدة 10 — مشروع: مخزن متغيّرات لحالة تسجيل احتيال

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

الخريطة​

خمس خطوات مُتسلسلة:

  1. توليد قاعدة بيانات تركيبيّة لمعاملات البطاقة
  2. بناء مسار «بلا مخزن» (baseline) — نُقيس فجوته
  3. بناء دفتر Feast مع Redis محلّيّ
  4. تدريب نموذج بـget_historical_features وخدمته بـget_online_features
  5. قياس ما ربحناه: الفجوة، والدقّة، وزمن الاستجابة

الخطوة 1: توليد بيانات​

import numpy as np
import pandas as pd

n = 500_000
np.random.seed(7)

df = pd.DataFrame({
    "event_id": np.arange(n),
    "customer_id": np.random.randint(1, 5000, n),
    "event_time": pd.to_datetime("2026-01-01") + pd.to_timedelta(
        np.random.randint(0, 180 * 24 * 3600, n), unit="s"
    ),
    "amount": np.abs(np.random.normal(45, 25, n)),
})
df["is_fraud"] = (
    (df["amount"] > 200) & (np.random.rand(n) < 0.3)
).astype(int)
df["created_at"] = df["event_time"] + pd.Timedelta(seconds=2)
df.sort_values("event_time").to_parquet("data/transactions.parquet")

نحصل على 500 ألف معاملة، معدّل احتيال حوالي 2%، وطابعان زمنيّان مختلفان (وقت الحدث ووقت الوصول) كما في الإنتاج.

الخطوة 2: مسار بلا مخزن​

نُصنّع الخطأ عمدًا لنقيسه. دفتر التدريب يحسب count_1h بـpandas على البيانات كلّها:

def naive_count_1h(row, all_tx):
    mask = (
        (all_tx["customer_id"] == row["customer_id"])
        & (all_tx["event_time"] > row["event_time"] - pd.Timedelta("1h"))
        & (all_tx["event_time"] <= row["event_time"])
    )
    return mask.sum() - 1

مُخدَم الاستدلال المُحاكى يستعمل استعلامًا مختلفًا يعتمد على created_at بدل event_time (خطأ شائع):

def serving_count_1h(row, all_tx):
    mask = (
        (all_tx["customer_id"] == row["customer_id"])
        & (all_tx["created_at"] > row["created_at"] - pd.Timedelta("1h"))
        & (all_tx["created_at"] <= row["created_at"])
    )
    return mask.sum() - 1

نُقارن القيمتين على مليون صفّ. النتيجة: 17% من الصفوف لها قيمة مختلفة، بمتوسّط فارق مطلق 1,4. النموذج المُدرَّب بـnaive_count_1h سيستقبل في الإنتاج مدخلًا مختلفًا.

الخطوة 3: دفتر Feast مع Redis​

نُنشئ feast_repo/features/transactions.py:

from datetime import timedelta
from feast import Entity, Field, FeatureView, FileSource
from feast.types import Int64, Float32

customer = Entity(name="customer", join_keys=["customer_id"])

transactions = FileSource(
    name="transactions",
    path="../data/transactions.parquet",
    timestamp_field="event_time",
    created_timestamp_column="created_at",
)

customer_counters = FeatureView(
    name="customer_counters",
    entities=[customer],
    ttl=timedelta(hours=2),
    schema=[
        Field(name="count_1h", dtype=Int64),
        Field(name="count_24h", dtype=Int64),
    ],
    source=transactions,
    online=True,
    tags={"owner": "fraud-team"},
)

نُشغّل Redis محلّيًّا (docker run -p 6379:6379 redis:7) ثمّ نُطبّق:

feast apply
feast materialize 2026-01-01T00:00:00 2026-07-01T00:00:00

الخطوة 4: تدريب وخدمة​

مجموعة التدريب تُخرَج بدمج زمنيّ صحيح آليًّا:

from feast import FeatureStore
import xgboost as xgb

store = FeatureStore(repo_path="feast_repo")

labels = pd.read_parquet("data/transactions.parquet")[
    ["customer_id", "event_time", "is_fraud"]
].sample(200_000, random_state=1)

training = store.get_historical_features(
    entity_df=labels,
    features=["customer_counters:count_1h", "customer_counters:count_24h"],
).to_df()

X = training[["count_1h", "count_24h"]]
y = training["is_fraud"]
model = xgb.XGBClassifier(max_depth=4, n_estimators=200).fit(X, y)

خدمة الاستدلال:

def predict(customer_id):
    feats = store.get_online_features(
        features=["customer_counters:count_1h", "customer_counters:count_24h"],
        entity_rows=[{"customer_id": customer_id}],
    ).to_dict()
    x = [[feats["count_1h"][0], feats["count_24h"][0]]]
    return model.predict_proba(x)[0][1]

الخطوة 5: القياسات​

بعد التنفيذ الكامل، نُخرج ثلاثة أرقام:

الفجوة. نُشغّل serving_count_1h من الخطوة 2 على عيّنة، ونقارن بما يُعيده Feast لنفس المعرّفات:

  • بلا مخزن: 17% من الصفوف مختلفة
  • مع Feast: 0% مختلف. القيم متطابقة حرفيًّا لأنّها تأتي من نفس التعريف.

الدقّة في «الإنتاج» المحاكى. نُدرّب النموذجَين ونختبرهما على نفس مجموعة اختبار مع مدخلات «إنتاج»:

  • بلا مخزن: AUC تدريب 0,921، إنتاج 0,832 — فارق 0,089
  • مع Feast: AUC تدريب 0,894، إنتاج 0,890 — فارق 0,004

النموذج «بلا مخزن» يبدو أفضل في المختبر بشكل مضلّل. النموذج مع Feast أقلّ إعلانيًّا لكنّه يعمل كما تقول أرقامه في الإنتاج.

زمن الاستجابة. نُقيس نداء get_online_features على 10 آلاف طلب:

  • المتوسّط: 1,3 مليّ‌ثانية
  • الشريحة p99: 4,7 مليّ‌ثانية

هذا زمن مقبول تمامًا لحلقة قرار عند نقطة البيع مع ميزانيّة إجماليّة 40 مليّ‌ثانية.

ما كان يلزم من دون مخزن​

لو أراد الفريق حلّ الفجوة بدون مخزن، كان عليه:

  • كتابة اختبار وحدانيّ يقارن التنفيذَين على عيّنة (يفوت الحالات الحادّة)
  • توثيق كلّ متغيّر في ملفّ Markdown (يُهمَل بسرعة)
  • بناء لوحة كتالوج محلّيّة (استثمار مركزيّ لكلّ فريق)
  • كتابة نصّ تجسيد بيدَيه (Airflow + Redis + إدارة الفشل)
  • إعادة كتابة كلّ هذا عند إضافة نموذج ثانٍ

بعبارة أخرى: بناء مخزن متغيّرات من الصفر. Feast يوفّر هذا العمل، مع مواصفات مُختبَرة على مئات الفرق.

امتداد المشروع

لتوسيع المشروع: (1) أضِف كيان merchant مع متغيّراته الخاصّة؛ (2) استعمل مصدر تدفّق Kafka بدل Parquet ليعرض التجسيد بالتدفّق؛ (3) نصّب Feast UI (feast ui) وأضِف وسومًا للاكتشاف؛ (4) اربط Prometheus لمراقبة النضارة والقيم الناقصة (الوحدة 9).

الخلاصة​

  • المسار بلا مخزن يُنتج فجوة 17% على عدّاد واحد، وفارق 0,089 نقطة AUC بين المختبر والإنتاج.
  • Feast مع Redis يُلغي الفجوة، ويجعل الفارق بين المختبر والإنتاج يقارب الصفر.
  • زمن الاستجابة تحت 5 مليّ‌ثانيات في p99 يسمح بالاستعمال في قرار حيّ.
  • تفادي بناء مخزن يعني إعادة كتابة المكوّنات نفسها يدويًّا، بجودة أدنى، لكلّ فريق.

الوحدة التالية: مراجعة كلّ ما تعلّمناه، وإعلان الامتحان النهائيّ.