مشروع عملي (1): بناء نظام كشف الاحتيال في البطاقات الائتمانية باستخدام تعلم الآلة

مشروع عملي (1): بناء نظام كشف الاحتيال في البطاقات الائتمانية باستخدام تعلم الآلة

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

هذا النوع من المشاريع يجمع فعلياً بين علوم البيانات، وهندسة البيانات، ومعالجة البيانات الضخمة، لأن المشكلة ليست في بناء النموذج فقط، بل في تصميم خط متكامل يبدأ من جمع البيانات وينتهي بتوليد تنبيه لحظي. وإذا كنت قد قرأت سابقاً مدخل إلى علوم البيانات: كيف تحول الأرقام العشوائية إلى قرارات استراتيجية؟ فسترى هنا كيف تتحول الأرقام الخام إلى قرار تشغيلي مباشر.

فهم طبيعة مشكلة الاحتيال

المعضلة الأساسية في بيانات الاحتيال هي عدم التوازن الحاد بين الفئات. غالباً تكون العمليات الاحتيالية نسبة صغيرة جداً مقارنة بالعمليات السليمة، ما يعني أن أي نموذج ساذج قد يحقق دقة ظاهرية مرتفعة فقط لأنه يتوقع أن جميع العمليات سليمة. لذلك لا يكفي قياس Accuracy وحدها.

هنا تظهر أهمية فهم كيفية التعامل مع عدم توازن البيانات (Imbalanced Data) في نماذج التصنيف، بالإضافة إلى الاعتماد على مقاييس مثل Precision وRecall وF1-Score وROC-AUC. الهدف العملي ليس فقط اكتشاف أكبر عدد من حالات الاحتيال، بل تقليل الإنذارات الكاذبة التي تزعج العملاء.

معمارية النظام من البيانات إلى القرار

في المشاريع الحقيقية، لا نبدأ مباشرة بتدريب النموذج. أولاً نحدد مسار البيانات: مصدر العمليات، آلية الاستخراج، قواعد التنظيف، مستودع التخزين، طريقة التدريب، ثم طبقة الاستدلال اللحظي. هذه الفكرة ترتبط بما شرحناه في بناء خطوط أنابيب البيانات (ETL – Extract, Transform, Load) باستخدام بايثون، لكن هنا نطبّقها على حالة مالية شديدة الحساسية.

• استخراج سجلات المعاملات من قواعد SQL أو واجهات الدفع.
• تنظيف القيم الشاذة والمفقودة وتوحيد الأنواع الزمنية.
• بناء ميزات سلوكية مثل عدد العمليات خلال ساعة ومتوسط المبلغ.
• تدريب نموذج تصنيف مثل Logistic Regression أو Random Forest.
• إرسال النتيجة إلى نظام مراقبة أو خدمة قرار فوري.

في الأنظمة المالية واسعة النطاق، من الأفضل فصل مسار التدريب عن مسار التنبؤ اللحظي. التدريب يمكن أن يعمل على دفعات كبيرة باستخدام Spark أو مستودع بيانات، بينما التنبؤ الفوري يجب أن يكون خفيفاً وسريع الاستجابة بزمن أقل من ثوانٍ معدودة.

استكشاف البيانات وتجهيزها

نفترض أن لدينا ملفاً يحتوي على أعمدة مثل وقت العملية، المبلغ، الدولة، نوع التاجر، وسِمة الهدف is_fraud. أول خطوة هي قراءة البيانات وفحص بنيتها، وهي مهارة تأسيسية تناولناها في مكتبة Pandas (1): قراءة واستدعاء البيانات من ملفات CSV و Excel برمجياً ومكتبة Pandas (2): استكشاف هيكل البيانات وفهم DataFrame و Series.

import pandas as pd
import numpy as np

df = pd.read_csv("credit_card_transactions.csv")

print(df.head())
print(df.info())
print(df["is_fraud"].value_counts(normalize=True))

بعد ذلك نتحقق من القيم المفقودة والتكرارات والقيم الشاذة. هذه المرحلة ليست تجميلية، بل تؤثر مباشرة على سلوك النموذج. إذا كانت بعض السجلات تفتقد المبلغ أو وقت العملية، فقد نفقد إشارات حاسمة تساعد في كشف النمط الاحتيالي.

df = df.drop_duplicates()

missing_ratio = df.isnull().mean().sort_values(ascending=False)
print(missing_ratio)

df["transaction_amount"] = df["transaction_amount"].fillna(df["transaction_amount"].median())
df["transaction_time"] = pd.to_datetime(df["transaction_time"])

للتعمق في هذه المرحلة يمكنك الرجوع إلى تنظيف البيانات (Data Cleaning): اكتشاف ومعالجة القيم المفقودة (Missing Values) ومعالجة البيانات المكررة والمشوهة (Duplicates & Outliers) باستخدام بايثون.

هندسة الميزات السلوكية

قوة نظام كشف الاحتيال تأتي غالباً من هندسة الميزات (Feature Engineering): كيف تستخرج بيانات جديدة من البيانات الحالية؟. فالمبلغ وحده لا يكفي، لكن عند ربطه بزمن العملية، وتكرار الاستخدام، وموقع العميل، تظهر إشارات أكثر عمقاً.

من الميزات المفيدة عملياً:

• عدد العمليات للبطاقة نفسها خلال آخر ساعة.
• متوسط مبلغ المعاملات اليومية لكل عميل.
• الفرق بين الدولة الحالية وآخر دولة استخدمت فيها البطاقة.
• تنفيذ العملية في ساعة غير معتادة بالنسبة للعميل.
• نسبة المبلغ الحالي إلى المتوسط التاريخي للبطاقة.

df = df.sort_values(["card_id", "transaction_time"])

df["hour"] = df["transaction_time"].dt.hour
df["day_of_week"] = df["transaction_time"].dt.dayofweek

customer_avg = df.groupby("card_id")["transaction_amount"].transform("mean")
df["amount_to_avg_ratio"] = df["transaction_amount"] / (customer_avg + 1e-6)

df["tx_count_24h"] = (
    df.groupby("card_id")["transaction_time"]
      .transform(lambda x: x.diff().dt.total_seconds().fillna(0))
)

df["rapid_transaction_flag"] = (df["tx_count_24h"] < 120).astype(int)

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

تجهيز البيانات للتدريب

البيانات الفئوية مثل الدولة أو نوع التاجر تحتاج إلى ترميز، كما أن بعض النماذج تستفيد من التحجيم العددي. هذه الخطوة مرتبطة بما ورد في إعداد البيانات للتدريب (Data Preprocessing): تحجيم البيانات (Scaling & Normalization) وتقسيم البيانات (Train/Test Split): لماذا يجب أن نختبر النموذج على بيانات لم يرها من قبل؟.

from sklearn.model_selection import train_test_split
from sklearn.compose import ColumnTransformer
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler
from sklearn.impute import SimpleImputer
from sklearn.ensemble import RandomForestClassifier

target = "is_fraud"
features = ["transaction_amount", "hour", "day_of_week", "amount_to_avg_ratio",
            "rapid_transaction_flag", "merchant_category", "country"]

X = df[features]
y = df[target]

numeric_features = ["transaction_amount", "hour", "day_of_week", "amount_to_avg_ratio"]
categorical_features = ["merchant_category", "country"]
binary_features = ["rapid_transaction_flag"]

preprocessor = ColumnTransformer(
    transformers=[
        ("num", Pipeline([
            ("imputer", SimpleImputer(strategy="median")),
            ("scaler", StandardScaler())
        ]), numeric_features),
        ("cat", Pipeline([
            ("imputer", SimpleImputer(strategy="most_frequent")),
            ("encoder", OneHotEncoder(handle_unknown="ignore"))
        ]), categorical_features),
        ("bin", "passthrough", binary_features)
    ]
)

X_train, X_test, y_train, y_test = train_test_split(
    X, y, test_size=0.2, random_state=42, stratify=y
)

model = Pipeline([
    ("preprocessor", preprocessor),
    ("classifier", RandomForestClassifier(
        n_estimators=200,
        max_depth=12,
        class_weight="balanced",
        random_state=42
    ))
])

model.fit(X_train, y_train)

تقييم النموذج بشكل صحيح

في كشف الاحتيال، قد يكون Recall مهماً جداً لأنك تريد التقاط أكبر عدد ممكن من العمليات الاحتيالية. لكن إذا ارتفع أكثر من اللازم على حساب Precision فقد تبدأ بحظر عملاء شرعيين. لذلك يجب موازنة المقاييس وفق سياسة المخاطر في المؤسسة.

from sklearn.metrics import classification_report, roc_auc_score, confusion_matrix

y_pred = model.predict(X_test)
y_prob = model.predict_proba(X_test)[:, 1]

print(confusion_matrix(y_test, y_pred))
print(classification_report(y_test, y_pred))
print("ROC-AUC:", roc_auc_score(y_test, y_prob))

وإذا أردت رفع موثوقية التقييم أكثر، فراجع التقييم المتقاطع (Cross-Validation): ضمان عدم حفظ النموذج للبيانات (Overfitting) وتحسين المعاملات الفائقة (Hyperparameter Tuning): رفع دقة النموذج إلى أقصى حد (GridSearch).

في التطبيقات البنكية الفعلية، العتبة threshold لا تُترك دائماً على القيمة الافتراضية 0.5. غالباً يتم تعديلها حسب تكلفة الخطأ، ونوع العميل، وقيمة المعاملة، وسياق المخاطر اللحظي.

التوسع إلى البيانات الضخمة باستخدام PySpark

عندما ترتفع أحجام البيانات إلى ملايين أو مئات الملايين من العمليات، تصبح أدوات مثل Pandas غير كافية في كثير من الحالات، وهنا ننتقل إلى ما هو Apache Spark؟ ولماذا تتوقف مكتبة Pandas عن العمل مع البيانات الضخمة (Big Data)؟ ثم إلى إعداد بيئة PySpark: معالجة البيانات الموزعة على عدة أجهزة في نفس الوقت.

from pyspark.sql import SparkSession
from pyspark.sql.functions import col, hour, dayofweek, avg

spark = SparkSession.builder.appName("FraudDetection").getOrCreate()

sdf = spark.read.csv("credit_card_transactions.csv", header=True, inferSchema=True)

sdf = sdf.withColumn("hour", hour(col("transaction_time")))
sdf = sdf.withColumn("day_of_week", dayofweek(col("transaction_time")))

avg_df = sdf.groupBy("card_id").agg(avg("transaction_amount").alias("customer_avg_amount"))

sdf = sdf.join(avg_df, on="card_id", how="left")
sdf = sdf.withColumn(
    "amount_to_avg_ratio",
    col("transaction_amount") / (col("customer_avg_amount") + 0.000001)
)

sdf.show(5)

يمكن لاحقاً حفظ النتائج في مخزن تحليلي أو قاعدة NoSQL سريعة القراءة، أو تشغيل تنبؤات لحظية عبر معالجة تدفق البيانات اللحظية (Spark Structured Streaming) إذا كان النظام يتعامل مع تدفق مباشر من بوابات الدفع.

النشر والاستخدام التشغيلي

بعد الوصول إلى نموذج جيد، يجب تصديره وربطه بخدمة تشغيلية. هذه المرحلة عادة تتضمن حفظ النموذج، بناء واجهة API، ثم مراقبة الأداء بعد النشر. يمكنك الاستفادة هنا من تصدير نماذج الذكاء الاصطناعي (Pickle & Joblib) لاستخدامها لاحقاً في الـ Backend والنشر كخدمة ويب (Deployment): تغليف نموذج تحليل البيانات الخاص بك في واجهة Flask/FastAPI.

أحد أكبر أخطاء المشاريع هو اعتبار النموذج منجزاً بعد التدريب. في الواقع، يجب مراقبة انجراف البيانات Data Drift وتغير أنماط الاحتيال بمرور الوقت، لأن المحتالين يكيّفون سلوكهم بسرعة، ما يستلزم إعادة تدريب دورية وخط مراقبة مستمر.

خاتمة

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

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