الگوریتم Isolation Forest: تشخیص ناهنجاری با یادگیری ماشین

s(x, n) = 2^(-E(h(x)) / c(n))

pip install scikit-learn numpy pandas matplotlib seaborn

import numpy as np
import pandas as pd
import matplotlib.pyplot as plt
from sklearn.ensemble import IsolationForest
from sklearn.preprocessing import StandardScaler

# تولید داده‌های نمونه
np.random.seed(42)
# داده‌های عادی
normal_data = np.random.randn(1000, 2) * 2
# داده‌های ناهنجار
anomaly_data = np.random.uniform(low=-8, high=8, size=(50, 2))
# ترکیب داده‌ها
X = np.vstack([normal_data, anomaly_data])
# نرمال‌سازی داده‌ها
scaler = StandardScaler()
X_scaled = scaler.fit_transform(X)
# ایجاد و آموزش مدل
isolation_forest = IsolationForest(
    n_estimators=100,
    max_samples='auto',
    contamination=0.05,
    random_state=42
)
# پیش‌بینی
predictions = isolation_forest.fit_predict(X_scaled)
scores = isolation_forest.score_samples(X_scaled)
# تبدیل نتایج (1 = عادی، -1 = ناهنجار)
anomalies = predictions == -1
# نمایش نتایج
plt.figure(figsize=(12, 5))
plt.subplot(1, 2, 1)
plt.scatter(X[~anomalies, 0], X[~anomalies, 1], 
           c='blue', label='Normal', alpha=0.6)
plt.scatter(X[anomalies, 0], X[anomalies, 1], 
           c='red', label='Anomaly', alpha=0.6)
plt.title('Isolation Forest Results')
plt.xlabel('Feature 1')
plt.ylabel('Feature 2')
plt.legend()
plt.subplot(1, 2, 2)
plt.hist(scores, bins=50, alpha=0.7)
plt.axvline(x=scores[anomalies].max(), 
           color='r', linestyle='--', 
           label='Anomaly Threshold')
plt.title('Anomaly Scores Distribution')
plt.xlabel('Anomaly Score')
plt.ylabel('Frequency')
plt.legend()
plt.tight_layout()
plt.show()
# نمایش آمار
print(f"بهترین پارامترها: {best_params}")
print(f"بهترین F1-Score: {best_score:.4f}")
# آموزش مدل نهایی با بهترین پارامترها
final_model = IsolationForest(**best_params, random_state=42)
final_predictions = final_model.fit_predict(X_scaled)
anomaly_scores = final_model.score_samples(X_scaled)
# تبدیل نتایج
pred_labels = (final_predictions == -1).astype(int)
# ارزیابی مدل
print("\nGزارش طبقه‌بندی:")
print(classification_report(true_labels, pred_labels, 
                          target_names=['Normal', 'Fraud']))
# ماتریس درهم‌ریختگی
cm = confusion_matrix(true_labels, pred_labels)
plt.figure(figsize=(8, 6))
sns.heatmap(cm, annot=True, fmt='d', cmap='Blues')
plt.title('Confusion Matrix')
plt.ylabel('True Label')
plt.xlabel('Predicted Label')
plt.show()
# تحلیل ویژگی‌ها
feature_importance = np.abs(final_model.decision_function(X_scaled))
plt.figure(figsize=(10, 6))
for i, col in enumerate(df.columns):
    plt.scatter(df[col], anomaly_scores, alpha=0.5, label=col)
plt.xlabel('Feature Value')
plt.ylabel('Anomaly Score')
plt.title('Feature vs Anomaly Score')
plt.legend()
plt.show()

from sklearn.model_selection import learning_curve

# بررسی تأثیر تعداد درخت
n_estimators_range = [10, 50, 100, 200, 500]
scores = []
for n_est in n_estimators_range:
    model = IsolationForest(n_estimators=n_est, random_state=42)
    model.fit(X_scaled)
    score = model.score_samples(X_scaled).mean()
    scores.append(score)
plt.plot(n_estimators_range, scores, marker='o')
plt.xlabel('Number of Estimators')
plt.ylabel('Average Score')
plt.title('Impact of n_estimators on Model Performance')
plt.grid(True)
plt.show()

from sklearn.decomposition import PCA
from sklearn.feature_selection import SelectKBest, f_classif

# کاهش ابعاد با PCA
pca = PCA(n_components=0.95)  # حفظ 95% واریانس
X_pca = pca.fit_transform(X_scaled)
# آموزش مدل با داده‌های کاهش‌یافته
model_pca = IsolationForest(n_estimators=100, random_state=42)
model_pca.fit(X_pca)

from sklearn.ensemble import VotingClassifier

# ایجاد چندین مدل با پارامترهای مختلف
models = [
    ('if1', IsolationForest(n_estimators=100, contamination=0.05)),
    ('if2', IsolationForest(n_estimators=200, contamination=0.03)),
    ('if3', IsolationForest(n_estimators=150, contamination=0.07))
]
# ترکیب نتایج
predictions_ensemble = []
for name, model in models:
    pred = model.fit_predict(X_scaled)
    predictions_ensemble.append(pred)
# رای‌گیری اکثریت
final_pred = np.array(predictions_ensemble).T
final_pred = np.apply_along_axis(
    lambda x: -1 if np.sum(x == -1) >= 2 else 1, 
    axis=1, 
    arr=final_pred
)

import joblib

# ذخیره مدل
joblib.dump(final_model, 'isolation_forest_model.pkl')
joblib.dump(scaler, 'scaler.pkl')
# بارگذاری مدل
loaded_model = joblib.load('isolation_forest_model.pkl')
loaded_scaler = joblib.load('scaler.pkl')
# استفاده از مدل بارگذاری شده
new_data = np.array([[150, 14, 6, 12]])
new_data_scaled = loaded_scaler.transform(new_data)
prediction = loaded_model.predict(new_data_scaled)
score = loaded_model.score_samples(new_data_scaled)
print(f"Prediction: {'Anomaly' if prediction[0] == -1 else 'Normal'}")
print(f"Anomaly Score: {score[0]:.4f}")

import time
from datetime import datetime

class IsolationForestMonitor:
    def __init__(self, model, scaler):
        self.model = model
        self.scaler = scaler
        self.predictions_log = []
        self.scores_log = []
        
    def predict_and_log(self, data):
        # نرمال‌سازی
        data_scaled = self.scaler.transform(data)
        
        # پیش‌بینی
        start_time = time.time()
        prediction = self.model.predict(data_scaled)
        score = self.model.score_samples(data_scaled)
        end_time = time.time()
        
        # ثبت لاگ
        log_entry = {
            'timestamp': datetime.now(),
            'prediction': prediction[0],
            'score': score[0],
            'inference_time': end_time - start_time
        }
        self.predictions_log.append(log_entry)
        self.scores_log.append(score[0])
        
        return prediction[0], score[0]
    
    def get_statistics(self):
        anomaly_rate = sum(1 for log in self.predictions_log 
                          if log['prediction'] == -1) / len(self.predictions_log)
        avg_inference_time = np.mean([log['inference_time'] 
                                     for log in self.predictions_log])
        avg_score = np.mean(self.scores_log)
        
        return {
            'total_predictions': len(self.predictions_log),
            'anomaly_rate': anomaly_rate,
            'avg_inference_time': avg_inference_time,
            'avg_anomaly_score': avg_score
        }
# استفاده
monitor = IsolationForestMonitor(loaded_model, loaded_scaler)

class AdaptiveIsolationForest:
    def __init__(self, initial_model, retrain_threshold=1000):
        self.model = initial_model
        self.data_buffer = []
        self.retrain_threshold = retrain_threshold
        
    def predict(self, data):
        prediction = self.model.predict(data)
        self.data_buffer.append(data)
        
        # بررسی نیاز به آموزش مجدد
        if len(self.data_buffer) >= self.retrain_threshold:
            self.retrain()
            
        return prediction
    
    def retrain(self):
        # آموزش مجدد با داده‌های جدید
        X_new = np.vstack(self.data_buffer)
        self.model.fit(X_new)
        self.data_buffer = []
        print(f"مدل در {datetime.now()} بروزرسانی شد")

from sklearn.preprocessing import RobustScaler

# استفاده از RobustScaler برای داده‌های با outlier
robust_scaler = RobustScaler()
X_robust = robust_scaler.fit_transform(X)
# حذف ویژگی‌های با واریانس صفر
from sklearn.feature_selection import VarianceThreshold
selector = VarianceThreshold(threshold=0.01)
X_filtered = selector.fit_transform(X_robust)

from sklearn.model_selection import StratifiedKFold

# برای ارزیابی بهتر از cross-validation استفاده کنید
def evaluate_isolation_forest(X, y, n_splits=5):
    skf = StratifiedKFold(n_splits=n_splits, shuffle=True, random_state=42)
    scores = []
    
    for train_idx, test_idx in skf.split(X, y):
        X_train, X_test = X[train_idx], X[test_idx]
        
        model = IsolationForest(n_estimators=100, random_state=42)
        model.fit(X_train)
        
        pred = model.predict(X_test)
        pred_binary = (pred == -1).astype(int)
        
        from sklearn.metrics import f1_score
        score = f1_score(y[test_idx], pred_binary)
        scores.append(score)
    
    return np.mean(scores), np.std(scores)
mean_score, std_score = evaluate_isolation_forest(X_scaled, true_labels)
print(f"میانگین F1-Score: {mean_score:.4f} (±{std_score:.4f})")

from sklearn.metrics import precision_recall_curve

# محاسبه امتیازات
scores = final_model.score_samples(X_scaled)
# محاسبه منحنی precision-recall
precision, recall, thresholds = precision_recall_curve(
    true_labels, 
    -scores  # منفی کردن برای تطبیق با کنوانسیون
)
# یافتن آستانه بهینه (F1 حداکثر)
f1_scores = 2 * (precision * recall) / (precision + recall)
best_threshold_idx = np.argmax(f1_scores)
best_threshold = -thresholds[best_threshold_idx]
print(f"آستانه بهینه: {best_threshold:.4f}")
# استفاده از آستانه بهینه
custom_predictions = (scores < best_threshold).astype(int)

# شناسایی False Positives و False Negatives
fp_indices = np.where((pred_labels == 1) & (true_labels == 0))[0]
fn_indices = np.where((pred_labels == 0) & (true_labels == 1))[0]

print(f"تعداد False Positives: {len(fp_indices)}")
print(f"تعداد False Negatives: {len(fn_indices)}")
# تحلیل ویژگی‌های نمونه‌های اشتباه
if len(fp_indices) > 0:
    print("\nمشخصات False Positives:")
    print(df.iloc[fp_indices].describe())
if len(fn_indices) > 0:
    print("\nمشخصات False Negatives:")
    print(df.iloc[fn_indices].describe())