Pipelines ومنع تسرب البيانات — تعلم الآلة

Pipelines ومنع تسرب البيانات

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

1. تسرب البيانات داخل Cross-Validation: إذا قيّست كل بيانات التدريب ثم قسمتها لـ folds، فالمقياس "رأى" بيانات التحقق.

2. نسيان خطوة عند التنبؤ على بيانات جديدة في الإنتاج.

3. كود متشعب يصعب ضبطه وحفظه.

Pipeline يحل الثلاث: سلسلة خطوات تُعامل كنموذج واحد له fit و predict.

Pipeline بسيط

from sklearn.datasets import load_breast_cancer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import cross_val_score
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler

X, y = load_breast_cancer(return_X_y=True)
pipe = Pipeline([
    ("scale", StandardScaler()),
    ("model", LogisticRegression(max_iter=1000)),
])
print(cross_val_score(pipe, X, y, cv=5).mean().round(3))

في كل fold يتعلم المقياس من جزء التدريب فقط، تلقائياً. make_pipeline(StandardScaler(), LogisticRegression()) اختصار يسمي الخطوات بنفسه.

أعمدة بأنواع مختلفة: ColumnTransformer

البيانات الحقيقية فيها أعمدة رقمية (تحتاج تعويضاً وتقييساً) وأعمدة فئوية (تحتاج تعويضاً و One-Hot):

import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.pipeline import make_pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

numeric = ["age", "income"]
categorical = ["city"]

preprocess = ColumnTransformer([
    ("num", make_pipeline(SimpleImputer(strategy="median"), StandardScaler()), numeric),
    ("cat", make_pipeline(SimpleImputer(strategy="most_frequent"), OneHotEncoder(handle_unknown="ignore")), categorical),
])

df = pd.DataFrame({"age": [25, None, 40], "income": [700, 1200, None], "city": ["بغداد", "البصرة", None]})
print(preprocess.fit_transform(df))

handle_unknown="ignore": مدينة جديدة لم تظهر في التدريب لا توقف النظام في الإنتاج.

الضبط عبر Pipeline

معاملات أي خطوة تُكتب بصيغة اسم_الخطوة__المعامل:

from sklearn.datasets import load_breast_cancer
from sklearn.model_selection import GridSearchCV
from sklearn.pipeline import Pipeline
from sklearn.preprocessing import StandardScaler
from sklearn.svm import SVC

X, y = load_breast_cancer(return_X_y=True)
pipe = Pipeline([("scale", StandardScaler()), ("svm", SVC())])
grid = GridSearchCV(pipe, {"svm__C": [0.1, 1, 10], "svm__gamma": ["scale", 0.01]}, cv=5)
grid.fit(X, y)
print(grid.best_params_, round(grid.best_score_, 3))

الحفظ والتحميل

import joblib
from sklearn.datasets import load_iris
from sklearn.ensemble import RandomForestClassifier

X, y = load_iris(return_X_y=True)
model = RandomForestClassifier(random_state=0).fit(X, y)
joblib.dump(model, "model.joblib")        # احفظ الـ Pipeline كاملاً لا النموذج وحده
loaded = joblib.load("model.joblib")
print(loaded.predict(X[:3]))

⚠️ ملفات joblib و pickle تنفّذ كوداً عند التحميل: لا تحمّل ملف نموذج من مصدر لا تثق به.

صور أخرى لتسرب البيانات

- ميزة تحتوي الإجابة بشكل غير مباشر (مثل "تاريخ إلغاء الاشتراك" عند التنبؤ بالإلغاء).

- بيانات زمنية مقسمة عشوائياً: النموذج يتعلم من المستقبل ليتنبأ بالماضي. استخدم TimeSeriesSplit.

- صفوف مكررة تظهر في التدريب والاختبار معاً.

إذا كانت النتيجة "أجمل من أن تُصدق"، فابحث عن تسرب أولاً.

تمرين

أعد كتابة مثال درس تحضير البيانات كـ Pipeline مع ColumnTransformer و LogisticRegression، وقيّمه بـ cross_val_score، ثم احفظه وحمّله وتنبأ لعميل جديد من مدينة لم تظهر في التدريب.

import numpy as np
import pandas as pd
from sklearn.compose import ColumnTransformer
from sklearn.impute import SimpleImputer
from sklearn.linear_model import LogisticRegression
from sklearn.model_selection import GridSearchCV, train_test_split
from sklearn.pipeline import Pipeline, make_pipeline
from sklearn.preprocessing import OneHotEncoder, StandardScaler

rng = np.random.default_rng(42)
n = 300
df = pd.DataFrame({
    "age": rng.integers(18, 65, n).astype(float),
    "income": rng.normal(1000, 300, n).round(),
    "city": rng.choice(["بغداد", "البصرة", "أربيل", "النجف"], n),
})
# الهدف يعتمد على الدخل والعمر والمدينة (مع ضوضاء)
logit = 0.004 * (df["income"] - 1000) + 0.03 * (df["age"] - 40) + (df["city"] == "أربيل") * 0.8
df["purchased"] = (logit + rng.normal(0, 0.8, n) > 0).astype(int)

# قيم ناقصة واقعية
df.loc[rng.choice(n, 20, replace=False), "age"] = np.nan
df.loc[rng.choice(n, 15, replace=False), "city"] = None

X = df.drop(columns="purchased")
y = df["purchased"]
X_train, X_test, y_train, y_test = train_test_split(X, y, test_size=0.2, random_state=42, stratify=y)

preprocess = ColumnTransformer([
    ("num", make_pipeline(SimpleImputer(strategy="median"), StandardScaler()), ["age", "income"]),
    ("cat", make_pipeline(SimpleImputer(strategy="most_frequent"), OneHotEncoder(handle_unknown="ignore")), ["city"]),
])

pipe = Pipeline([("prep", preprocess), ("model", LogisticRegression(max_iter=1000))])

# الضبط بـ Cross-Validation على التدريب فقط — بلا تسرب
grid = GridSearchCV(pipe, {"model__C": [0.01, 0.1, 1, 10]}, cv=5)
grid.fit(X_train, y_train)
print("أفضل C:", grid.best_params_["model__C"], "| دقة CV:", round(grid.best_score_, 3))

# التقييم النهائي مرة واحدة
print("دقة الاختبار:", round(grid.score(X_test, y_test), 3))

# عميل من مدينة لم تظهر في التدريب: handle_unknown يمنع التوقف
new = pd.DataFrame({"age": [33.0], "income": [1250.0], "city": ["الموصل"]})
print("احتمال الشراء لعميل جديد:", round(grid.predict_proba(new)[0, 1], 2))