Phát hiện gian lận giao dịch với machine learning

Giới thiệu

Gian lận giao dịch tài chính là một vấn đề ngày càng nghiêm trọng đối với các tổ chức tài chính, sàn giao dịch, và doanh nghiệp trên toàn cầu. Thiệt hại do gian lận có thể lên đến hàng tỷ đô la mỗi năm. Trước tình hình đó, việc ứng dụng Machine Learning (ML) vào phát hiện gian lận đã trở thành một giải pháp hiệu quả, giúp phát hiện và ngăn chặn các hoạt động gian lận kịp thời.

Bài viết này sẽ trình bày cách xây dựng một hệ thống phát hiện gian lận dựa trên Machine Learning, từ việc thu thập và xử lý dữ liệu đến việc huấn luyện mô hình và đánh giá hiệu quả.

Các loại gian lận phổ biến

Trước khi đi vào chi tiết kỹ thuật, hãy tìm hiểu một số loại gian lận tài chính phổ biến:

1. Gian lận thẻ tín dụng: Sử dụng thông tin thẻ trái phép, thẻ giả.
2. Gian lận danh tính: Mạo danh người khác để thực hiện giao dịch.
3. Gian lận trên sàn giao dịch tiền điện tử: Thao túng thị trường, pump and dump, wash trading.
4. Gian lận thanh toán: Sử dụng nguồn tiền bất hợp pháp hoặc tranh chấp thanh toán.
5. Gian lận người trong: Nhân viên lợi dụng quyền truy cập để thực hiện gian lận.

Quy trình xây dựng hệ thống phát hiện gian lận

1. Thu thập và chuẩn bị dữ liệu

Dữ liệu là nền tảng của mọi hệ thống ML. Trong phát hiện gian lận, dữ liệu thường bao gồm:

• Lịch sử giao dịch
• Thông tin khách hàng
• Dữ liệu hành vi người dùng
• Dữ liệu vị trí địa lý
• Dữ liệu thiết bị

import pandas as pd
import numpy as np
from sklearn.model_selection import train_test_split
from sklearn.preprocessing import StandardScaler

# Đọc dữ liệu (ví dụ với tập dữ liệu giao dịch thẻ tín dụng)
data = pd.read_csv('transactions.csv')

# Kiểm tra dữ liệu thiếu
print(f"Số lượng giá trị thiếu trong mỗi cột:\n{data.isnull().sum()}")

# Kiểm tra sự mất cân bằng dữ liệu (thường gặp trong bài toán phát hiện gian lận)
fraud_count = len(data[data['is_fraud'] == 1])
non_fraud_count = len(data[data['is_fraud'] == 0])
print(f"Số lượng giao dịch gian lận: {fraud_count} ({fraud_count/len(data)*100:.2f}%)")
print(f"Số lượng giao dịch hợp lệ: {non_fraud_count} ({non_fraud_count/len(data)*100:.2f}%)")

Xử lý mất cân bằng dữ liệu

Dữ liệu gian lận thường rất mất cân bằng, với tỷ lệ gian lận chỉ chiếm khoảng 0.1% - 2% tổng số giao dịch. Để xử lý, ta có thể sử dụng:

• Oversampling: Tăng số lượng mẫu gian lận
• Undersampling: Giảm số lượng mẫu không gian lận
• SMOTE (Synthetic Minority Over-sampling Technique): Tạo mẫu gian lận tổng hợp

from imblearn.over_sampling import SMOTE
from imblearn.under_sampling import RandomUnderSampler
from imblearn.pipeline import Pipeline

# Tách đặc trưng và nhãn
X = data.drop('is_fraud', axis=1)
y = data['is_fraud']

# Tạo pipeline kết hợp SMOTE và undersampling
pipeline = Pipeline([
    ('smote', SMOTE(sampling_strategy=0.1)),  # Tăng tỷ lệ gian lận lên 10%
    ('under', RandomUnderSampler(sampling_strategy=0.5))  # Cân bằng tỷ lệ 1:1
])

# Áp dụng pipeline
X_resampled, y_resampled = pipeline.fit_resample(X, y)

print(f"Số lượng mẫu gian lận sau xử lý: {sum(y_resampled == 1)}")
print(f"Số lượng mẫu không gian lận sau xử lý: {sum(y_resampled == 0)}")

2. Kỹ thuật xử lý đặc trưng (Feature Engineering)

Đây là bước quan trọng nhất trong xây dựng hệ thống phát hiện gian lận. Các đặc trưng hiệu quả sẽ giúp mô hình phân biệt tốt hơn giữa giao dịch hợp lệ và gian lận.

Các đặc trưng thường được sử dụng

1. Đặc trưng từ thời gian:

   • Giờ trong ngày, ngày trong tuần
   • Thời gian giữa các giao dịch liên tiếp
   • Tốc độ giao dịch (số lượng giao dịch trong khung thời gian)

2. Đặc trưng từ số tiền:

   • Tỷ lệ giữa số tiền giao dịch hiện tại và trung bình lịch sử
   • Biến động số tiền giao dịch
   • Ngưỡng số tiền bất thường

3. Đặc trưng từ vị trí địa lý:

   • Khoảng cách giữa các giao dịch liên tiếp
   • Tốc độ di chuyển giữa các giao dịch
   • Quốc gia/khu vực có rủi ro cao

4. Đặc trưng từ hành vi:

   • Mô hình mua sắm
   • Tần suất đăng nhập
   • Thời gian giữa các phiên hoạt động

# Tạo đặc trưng từ thời gian
data['hour'] = pd.to_datetime(data['transaction_time']).dt.hour
data['day_of_week'] = pd.to_datetime(data['transaction_time']).dt.dayofweek

# Tạo đặc trưng từ số tiền
data['amount_zscore'] = (data['amount'] - data.groupby('customer_id')['amount'].transform('mean')) / data.groupby('customer_id')['amount'].transform('std').fillna(1)

# Tạo đặc trưng "tốc độ giao dịch"
data = data.sort_values(['customer_id', 'transaction_time'])
data['time_since_last_tx'] = data.groupby('customer_id')['transaction_time'].diff().dt.total_seconds()

# Tạo đặc trưng từ vị trí
from sklearn.metrics.pairwise import haversine_distances
import math

# Chuyển đổi độ sang radian
def degrees_to_radians(degrees):
    return degrees * math.pi / 180

# Tính khoảng cách giữa hai điểm (km)
def calculate_distance(lat1, lon1, lat2, lon2):
    lat1_rad = degrees_to_radians(lat1)
    lon1_rad = degrees_to_radians(lon1)
    lat2_rad = degrees_to_radians(lat2)
    lon2_rad = degrees_to_radians(lon2)
    
    coords_1 = [[lat1_rad, lon1_rad]]
    coords_2 = [[lat2_rad, lon2_rad]]
    
    distance = haversine_distances(coords_1, coords_2) * 6371  # Nhân với bán kính Trái Đất (km)
    return distance[0][0]

# Áp dụng tính khoảng cách giữa các giao dịch liên tiếp
data['prev_lat'] = data.groupby('customer_id')['latitude'].shift(1)
data['prev_lon'] = data.groupby('customer_id')['longitude'].shift(1)

# Tính khoảng cách (km)
data['distance_from_last_tx'] = data.apply(
    lambda row: calculate_distance(
        row['latitude'], row['longitude'], row['prev_lat'], row['prev_lon']
    ) if not pd.isna(row['prev_lat']) else 0,
    axis=1
)

# Tính tốc độ di chuyển (km/h)
data['speed_kmh'] = data.apply(
    lambda row: row['distance_from_last_tx'] / (row['time_since_last_tx'] / 3600) 
    if row['time_since_last_tx'] > 0 else 0,
    axis=1
)

# Flag cho tốc độ di chuyển bất thường (> 1000 km/h)
data['impossible_travel'] = (data['speed_kmh'] > 1000).astype(int)

3. Chọn và huấn luyện mô hình

Có nhiều thuật toán ML phù hợp cho bài toán phát hiện gian lận. Dưới đây là một số thuật toán hiệu quả:

Random Forest

from sklearn.ensemble import RandomForestClassifier
from sklearn.metrics import classification_report, confusion_matrix, roc_auc_score

# Chia dữ liệu
X_train, X_test, y_train, y_test = train_test_split(X_resampled, y_resampled, test_size=0.3, random_state=42)

# Khởi tạo và huấn luyện mô hình
rf_model = RandomForestClassifier(
    n_estimators=100, 
    max_depth=10,
    min_samples_split=10,
    min_samples_leaf=5,
    class_weight='balanced',
    random_state=42
)

rf_model.fit(X_train, y_train)

# Dự đoán
y_pred = rf_model.predict(X_test)
y_pred_proba = rf_model.predict_proba(X_test)[:, 1]

# Đánh giá
print("Confusion Matrix:\n", confusion_matrix(y_test, y_pred))
print("\nClassification Report:\n", classification_report(y_test, y_pred))
print("\nROC AUC Score:", roc_auc_score(y_test, y_pred_proba))

# Đặc trưng quan trọng
feature_importance = pd.DataFrame({
    'Feature': X.columns,
    'Importance': rf_model.feature_importances_
}).sort_values('Importance', ascending=False)

print("\nĐặc trưng quan trọng nhất:")
print(feature_importance.head(10))

XGBoost

import xgboost as xgb
from sklearn.metrics import precision_recall_curve, auc

# Khởi tạo và huấn luyện mô hình
xgb_model = xgb.XGBClassifier(
    max_depth=6,
    learning_rate=0.05,
    n_estimators=100,
    objective='binary:logistic',
    scale_pos_weight=sum(y_train == 0) / sum(y_train == 1),  # Cân bằng lớp
    gamma=1,
    subsample=0.8,
    colsample_bytree=0.8,
    random_state=42
)

xgb_model.fit(X_train, y_train)

# Dự đoán
y_pred = xgb_model.predict(X_test)
y_pred_proba = xgb_model.predict_proba(X_test)[:, 1]

# Đánh giá
print("XGBoost Classification Report:\n", classification_report(y_test, y_pred))

# Tính Precision-Recall AUC (thường tốt hơn ROC AUC cho dữ liệu mất cân bằng)
precision, recall, thresholds = precision_recall_curve(y_test, y_pred_proba)
pr_auc = auc(recall, precision)
print(f"Precision-Recall AUC: {pr_auc:.4f}")

Isolation Forest (phát hiện bất thường)

from sklearn.ensemble import IsolationForest

# Khởi tạo và huấn luyện mô hình
iso_forest = IsolationForest(
    n_estimators=100,
    contamination=0.01,  # Tỷ lệ dự kiến của outlier
    max_samples='auto',
    random_state=42
)

# Huấn luyện trên dữ liệu không gian lận
X_normal = X[y == 0]
iso_forest.fit(X_normal)

# Dự đoán trên tập kiểm tra
# -1 là bất thường (có thể là gian lận), 1 là bình thường
anomaly_scores = iso_forest.decision_function(X_test)
predictions = iso_forest.predict(X_test)
anomalies = predictions == -1

# Chuyển đổi dự đoán để phù hợp với nhãn (0: bình thường, 1: gian lận)
y_pred_anomaly = np.where(predictions == -1, 1, 0)

print("Isolation Forest Performance:")
print(classification_report(y_test, y_pred_anomaly))

Mạng nơ-ron (Neural Network)

import tensorflow as tf
from tensorflow.keras.models import Sequential
from tensorflow.keras.layers import Dense, Dropout, BatchNormalization
from tensorflow.keras.callbacks import EarlyStopping

# Tạo mô hình
model = Sequential([
    Dense(64, activation='relu', input_shape=(X_train.shape[1],)),
    BatchNormalization(),
    Dropout(0.3),
    Dense(32, activation='relu'),
    BatchNormalization(),
    Dropout(0.3),
    Dense(16, activation='relu'),
    BatchNormalization(),
    Dense(1, activation='sigmoid')
])

# Biên dịch mô hình
model.compile(
    optimizer='adam',
    loss='binary_crossentropy',
    metrics=['accuracy', tf.keras.metrics.AUC(), tf.keras.metrics.Precision(), tf.keras.metrics.Recall()]
)

# Early stopping để tránh overfit
early_stopping = EarlyStopping(
    monitor='val_loss',
    patience=10,
    restore_best_weights=True
)

# Huấn luyện mô hình
history = model.fit(
    X_train, y_train,
    epochs=100,
    batch_size=32,
    validation_split=0.2,
    callbacks=[early_stopping],
    class_weight={0: 1, 1: sum(y_train == 0) / sum(y_train == 1)}  # Cân bằng lớp
)

# Đánh giá mô hình
y_pred_proba = model.predict(X_test)
y_pred = (y_pred_proba > 0.5).astype(int)

print("Neural Network Performance:")
print(classification_report(y_test, y_pred))

4. Tinh chỉnh ngưỡng phát hiện

Trong phát hiện gian lận, việc chọn ngưỡng quyết định (decision threshold) rất quan trọng. Thay vì sử dụng ngưỡng mặc định 0.5, ta có thể tinh chỉnh dựa trên chi phí của các loại lỗi:

• False Positive (FP): Giao dịch hợp lệ bị đánh dấu là gian lận, gây bất tiện cho khách hàng.
• False Negative (FN): Giao dịch gian lận không bị phát hiện, gây thiệt hại tài chính.

from sklearn.metrics import precision_recall_curve
import matplotlib.pyplot as plt

# Tính precision và recall cho nhiều ngưỡng khác nhau
precisions, recalls, thresholds = precision_recall_curve(y_test, y_pred_proba)

# Vẽ đường cong precision-recall
plt.figure(figsize=(10, 6))
plt.plot(recalls, precisions, 'b-', linewidth=2)
plt.xlabel('Recall')
plt.ylabel('Precision')
plt.title('Precision-Recall Curve')
plt.grid(True)

# Giả sử chi phí FN là 50 lần chi phí FP
# Tối ưu hóa: min(50*FN + FP)
costs = []
for threshold in thresholds:
    y_pred_t = (y_pred_proba >= threshold).astype(int)
    tn, fp, fn, tp = confusion_matrix(y_test, y_pred_t).ravel()
    
    # Chi phí tổng
    total_cost = 50 * fn + fp
    costs.append(total_cost)

# Tìm ngưỡng tối ưu
optimal_idx = np.argmin(costs)
optimal_threshold = thresholds[optimal_idx]

print(f"Ngưỡng tối ưu: {optimal_threshold:.4f}")
print(f"Tại ngưỡng này: Precision = {precisions[optimal_idx]:.4f}, Recall = {recalls[optimal_idx]:.4f}")

# Áp dụng ngưỡng tối ưu
y_pred_optimal = (y_pred_proba >= optimal_threshold).astype(int)
print("\nKết quả với ngưỡng tối ưu:")
print(classification_report(y_test, y_pred_optimal))

5. Triển khai hệ thống theo thời gian thực

Sau khi có mô hình hiệu quả, bước tiếp theo là triển khai hệ thống phát hiện gian lận theo thời gian thực. Dưới đây là một mẫu code đơn giản cho API đánh giá giao dịch:

from flask import Flask, request, jsonify
import joblib
import pandas as pd

app = Flask(__name__)

# Tải mô hình đã huấn luyện
model = joblib.load('fraud_detection_model.pkl')
scaler = joblib.load('feature_scaler.pkl')

@app.route('/predict', methods=['POST'])
def predict_fraud():
    # Lấy dữ liệu giao dịch
    transaction_data = request.json
    
    try:
        # Tiền xử lý dữ liệu
        df = pd.DataFrame([transaction_data])
        
        # Tạo đặc trưng (đơn giản hóa)
        df['hour'] = pd.to_datetime(df['transaction_time']).dt.hour
        df['day_of_week'] = pd.to_datetime(df['transaction_time']).dt.dayofweek
        
        # Chọn các cột đặc trưng
        features = ['amount', 'hour', 'day_of_week', 'merchant_category', 'latitude', 'longitude']
        X = df[features]
        
        # Chuẩn hóa
        X_scaled = scaler.transform(X)
        
        # Dự đoán
        fraud_probability = model.predict_proba(X_scaled)[0, 1]
        is_fraud = fraud_probability >= 0.8  # Ngưỡng cao để giảm false positives
        
        # Phân loại mức độ rủi ro
        risk_level = 'cao' if fraud_probability >= 0.8 else 'trung bình' if fraud_probability >= 0.3 else 'thấp'
        
        return jsonify({
            'transaction_id': transaction_data.get('transaction_id'),
            'fraud_probability': float(fraud_probability),
            'is_fraud': bool(is_fraud),
            'risk_level': risk_level
        })
        
    except Exception as e:
        return jsonify({'error': str(e)}), 400

if __name__ == '__main__':
    app.run(debug=True)

6. Kết hợp với quy tắc kinh doanh (Business Rules)

Ngoài mô hình ML, hệ thống phát hiện gian lận hiệu quả cũng cần kết hợp với các quy tắc kinh doanh được xác định trước:

def apply_business_rules(transaction, model_prediction, model_confidence):
    """
    Áp dụng các quy tắc kinh doanh để đưa ra quyết định cuối cùng
    """
    # 1. Nếu mô hình tự tin cao về việc gian lận
    if model_confidence > 0.9:
        return "reject"
    
    # 2. Các quy tắc cứng về số tiền và vị trí
    if transaction['amount'] > 10000 and transaction['country'] in HIGH_RISK_COUNTRIES:
        return "review"
    
    # 3. Quy tắc về tần suất giao dịch
    if transaction['tx_per_day'] > 20:
        return "review"
    
    # 4. Quy tắc về thời gian
    if transaction['hour'] >= 1 and transaction['hour'] <= 5 and transaction['amount'] > 1000:
        return "review"
    
    # 5. Nếu mô hình dự đoán gian lận nhưng độ tin cậy không cao
    if model_prediction == 1 and model_confidence > 0.6:
        return "review"
    
    # Mặc định: chấp nhận giao dịch
    return "accept"

Đánh giá hiệu quả

Để đánh giá hiệu quả của hệ thống phát hiện gian lận, ta không chỉ sử dụng các độ đo truyền thống mà còn cần tính đến các yếu tố kinh tế:

1. Tiết kiệm chi phí

def calculate_cost_savings(y_true, y_pred, cost_per_fraud=500, cost_per_false_alarm=10):
    """
    Tính toán chi phí tiết kiệm từ hệ thống phát hiện gian lận
    """
    tn, fp, fn, tp = confusion_matrix(y_true, y_pred).ravel()
    
    # Chi phí nếu không có hệ thống: tất cả gian lận đều không bị phát hiện
    cost_without_system = (tp + fn) * cost_per_fraud
    
    # Chi phí với hệ thống: gian lận không phát hiện + chi phí xem xét false alarms
    cost_with_system = fn * cost_per_fraud + fp * cost_per_false_alarm
    
    # Chi phí tiết kiệm
    cost_savings = cost_without_system - cost_with_system
    
    return {
        'cost_without_system': cost_without_system,
        'cost_with_system': cost_with_system,
        'cost_savings': cost_savings,
        'roi_percentage': (cost_savings / cost_with_system) * 100 if cost_with_system > 0 else float('inf')
    }

# Áp dụng
cost_analysis = calculate_cost_savings(y_test, y_pred_optimal)
print(f"Chi phí tiết kiệm: ${cost_analysis['cost_savings']:,.2f}")
print(f"ROI: {cost_analysis['roi_percentage']:.2f}%")

2. Phát hiện theo thời gian

Một yếu tố quan trọng khác là khả năng phát hiện gian lận theo thời gian - liệu mô hình có duy trì hiệu quả khi kiểu gian lận thay đổi?

def evaluate_over_time(model, data_by_month, feature_cols, target_col):
    """
    Đánh giá hiệu suất mô hình theo thời gian
    """
    results = []
    
    for month, month_data in data_by_month.items():
        X_month = month_data[feature_cols]
        y_month = month_data[target_col]
        
        y_pred = model.predict(X_month)
        
        precision = precision_score(y_month, y_pred)
        recall = recall_score(y_month, y_pred)
        f1 = f1_score(y_month, y_pred)
        
        results.append({
            'month': month,
            'precision': precision,
            'recall': recall,
            'f1_score': f1,
            'fraud_rate': y_month.mean() * 100  # Tỷ lệ gian lận
        })
    
    return pd.DataFrame(results)

# Vẽ biểu đồ theo thời gian
def plot_performance_over_time(time_results):
    plt.figure(figsize=(12, 6))
    
    plt.subplot(2, 1, 1)
    plt.plot(time_results['month'], time_results['precision'], 'b-', label='Precision')
    plt.plot(time_results['month'], time_results['recall'], 'r-', label='Recall')
    plt.plot(time_results['month'], time_results['f1_score'], 'g-', label='F1 Score')
    plt.ylabel('Score')
    plt.legend()
    plt.title('Model Performance Over Time')
    
    plt.subplot(2, 1, 2)
    plt.plot(time_results['month'], time_results['fraud_rate'], 'r-')
    plt.ylabel('Fraud Rate (%)')
    plt.xlabel('Month')
    plt.title('Fraud Rate Over Time')
    
    plt.tight_layout()
    plt.show()

Các thách thức khi triển khai hệ thống phát hiện gian lận

1. Dữ liệu mất cân bằng

Như đã đề cập, tỷ lệ gian lận thường rất thấp (khoảng 0.1-2%), gây khó khăn cho việc huấn luyện mô hình. Các giải pháp bao gồm:

• Sử dụng kỹ thuật lấy mẫu như SMOTE
• Sử dụng class_weight trong các thuật toán
• Sử dụng cost-sensitive learning
• Đánh giá bằng các thước đo phù hợp (Precision-Recall AUC thay vì ROC AUC)

2. Luôn thay đổi (Concept Drift)

Kẻ gian lận liên tục thích ứng và thay đổi chiến thuật. Để đối phó, hệ thống cần:

• Huấn luyện lại mô hình thường xuyên với dữ liệu mới
• Theo dõi hiệu suất mô hình liên tục
• Áp dụng các kỹ thuật học tăng cường (online learning)

def detect_concept_drift(model, new_data, reference_performance, threshold=0.1):
    """
    Phát hiện concept drift bằng cách so sánh hiệu suất
    """
    # Đánh giá trên dữ liệu mới
    X_new = new_data.drop('is_fraud', axis=1)
    y_new = new_data['is_fraud']
    
    y_pred = model.predict(X_new)
    new_performance = f1_score(y_new, y_pred)
    
    # Nếu hiệu suất giảm quá nhiều, có thể đã xảy ra concept drift
    performance_drop = reference_performance - new_performance
    
    if performance_drop > threshold:
        return True, performance_drop
    else:
        return False, performance_drop

3. Giải thích được (Explainability)

Các bên liên quan (khách hàng, nhân viên, cơ quan quản lý) cần hiểu tại sao một giao dịch bị đánh dấu là gian lận. Giải pháp:

• Sử dụng các mô hình có khả năng giải thích (cây quyết định, random forest)
• Áp dụng các kỹ thuật như SHAP (SHapley Additive exPlanations) hoặc LIME

import shap

# Khởi tạo SHAP explainer
explainer = shap.TreeExplainer(rf_model)

# Tính giá trị SHAP cho một giao dịch cụ thể
transaction_idx = 42  # Ví dụ
transaction = X_test.iloc[transaction_idx:transaction_idx+1]
shap_values = explainer.shap_values(transaction)

# Hiển thị lý do tại sao giao dịch này bị đánh dấu là gian lận
shap.force_plot(
    explainer.expected_value[1],  # giá trị cơ sở cho lớp gian lận
    shap_values[1],               # giá trị SHAP cho lớp gian lận
    transaction,
    feature_names=X.columns.tolist()
)

4. Tính riêng tư và bảo mật

Khi xử lý dữ liệu tài chính nhạy cảm, bảo mật và quyền riêng tư là yếu tố quan trọng:

• Mã hóa dữ liệu nhạy cảm
• Ẩn danh hóa thông tin cá nhân
• Tuân thủ các quy định về bảo vệ dữ liệu (GDPR, CCPA)
• Áp dụng các kỹ thuật học liên hợp (federated learning) khi cần thiết

Xu hướng mới trong phát hiện gian lận

1. Học sâu (Deep Learning)

Các kiến trúc học sâu như LSTM, GRU hiệu quả trong việc phát hiện các mẫu gian lận phức tạp trong dữ liệu theo chuỗi thời gian.

from tensorflow.keras.layers import LSTM, Dense, Input
from tensorflow.keras.models import Model

# Tạo dữ liệu chuỗi thời gian
def create_sequences(data, customer_ids, seq_length=10):
    sequences = []
    labels = []
    
    for customer_id in customer_ids:
        customer_data = data[data['customer_id'] == customer_id].sort_values('transaction_time')
        
        if len(customer_data) >= seq_length:
            for i in range(len(customer_data) - seq_length):
                seq = customer_data.iloc[i:i+seq_length]
                label = customer_data.iloc[i+seq_length]['is_fraud']
                
                # Chỉ lấy các cột số
                seq_features = seq.select_dtypes(include=np.number).drop('is_fraud', axis=1)
                
                sequences.append(seq_features.values)
                labels.append(label)
    
    return np.array(sequences), np.array(labels)

# Tạo và huấn luyện mô hình LSTM
def train_lstm_model(X_seq, y_seq):
    # Kiến trúc mô hình
    input_layer = Input(shape=(X_seq.shape[1], X_seq.shape[2]))
    
    lstm1 = LSTM(64, return_sequences=True)(input_layer)
    lstm2 = LSTM(32)(lstm1)
    
    dense1 = Dense(16, activation='relu')(lstm2)
    output = Dense(1, activation='sigmoid')(dense1)
    
    model = Model(inputs=input_layer, outputs=output)
    
    # Biên dịch
    model.compile(
        optimizer='adam',
        loss='binary_crossentropy',
        metrics=['accuracy', tf.keras.metrics.AUC()]
    )
    
    # Huấn luyện
    history = model.fit(
        X_seq, y_seq,
        epochs=50,
        batch_size=32,
        validation_split=0.2,
        class_weight={0: 1, 1: sum(y_seq == 0) / sum(y_seq == 1)}
    )
    
    return model, history

2. Học tự giám sát (Self-supervised Learning)

Kỹ thuật này cho phép học từ dữ liệu không gán nhãn, một lợi thế lớn trong phát hiện gian lận khi việc gán nhãn đúng cho mọi giao dịch là không khả thi.

from sklearn.preprocessing import StandardScaler
from tensorflow.keras.layers import Dense, Input, Dropout
from tensorflow.keras.models import Model

# Xây dựng autoencoder để phát hiện bất thường
def build_autoencoder(input_dim):
    # Encoder
    input_layer = Input(shape=(input_dim,))
    
    encoded = Dense(64, activation='relu')(input_layer)
    encoded = Dropout(0.3)(encoded)
    encoded = Dense(32, activation='relu')(encoded)
    encoded = Dropout(0.3)(encoded)
    encoded = Dense(16, activation='relu')(encoded)
    
    # Decoder
    decoded = Dense(32, activation='relu')(encoded)
    decoded = Dropout(0.3)(decoded)
    decoded = Dense(64, activation='relu')(decoded)
    decoded = Dropout(0.3)(decoded)
    decoded = Dense(input_dim, activation='linear')(decoded)
    
    # Autoencoder
    autoencoder = Model(inputs=input_layer, outputs=decoded)
    autoencoder.compile(optimizer='adam', loss='mse')
    
    # Encoder riêng để trích xuất đặc trưng
    encoder = Model(inputs=input_layer, outputs=encoded)
    
    return autoencoder, encoder

# Huấn luyện trên dữ liệu không gian lận
X_normal = X[y == 0]
scaler = StandardScaler()
X_normal_scaled = scaler.fit_transform(X_normal)

# Xây dựng và huấn luyện mô hình
autoencoder, encoder = build_autoencoder(X_normal_scaled.shape[1])

autoencoder.fit(
    X_normal_scaled, X_normal_scaled,
    epochs=50,
    batch_size=32,
    validation_split=0.2,
    shuffle=True
)

# Tính lỗi tái tạo
def compute_reconstruction_error(autoencoder, data):
    predictions = autoencoder.predict(data)
    mse = np.mean(np.power(data - predictions, 2), axis=1)
    return mse

# Áp dụng cho tập kiểm tra
X_test_scaled = scaler.transform(X_test)
reconstruction_errors = compute_reconstruction_error(autoencoder, X_test_scaled)

# Xác định ngưỡng dựa trên phân vị
threshold = np.percentile(reconstruction_errors, 95)  # 5% cao nhất được coi là bất thường

# Dự đoán
y_pred_autoencoder = (reconstruction_errors > threshold).astype(int)

print("Autoencoder Performance:")
print(classification_report(y_test, y_pred_autoencoder))

3. Học tăng cường (Reinforcement Learning)

Học tăng cường cho phép hệ thống phát hiện gian lận liên tục cải thiện dựa trên phản hồi.

# Mô phỏng đơn giản về hệ thống phát hiện gian lận sử dụng học tăng cường
class FraudDetectionEnv:
    def __init__(self, transactions, fraud_cost=100, false_alarm_cost=1):
        self.transactions = transactions
        self.current_idx = 0
        self.fraud_cost = fraud_cost
        self.false_alarm_cost = false_alarm_cost
    
    def reset(self):
        self.current_idx = 0
        return self._get_state()
    
    def _get_state(self):
        return self.transactions.iloc[self.current_idx][features].values
    
    def step(self, action):  # action: 0 = accept, 1 = reject
        transaction = self.transactions.iloc[self.current_idx]
        is_fraud = transaction['is_fraud']
        
        # Tính reward
        if action == 1 and is_fraud == 1:  # Đúng khi từ chối giao dịch gian lận
            reward = self.fraud_cost  # Tiết kiệm chi phí gian lận
        elif action == 1 and is_fraud == 0:  # Sai khi từ chối giao dịch hợp lệ
            reward = -self.false_alarm_cost  # Chi phí false alarm
        elif action == 0 and is_fraud == 1:  # Sai khi chấp nhận giao dịch gian lận
            reward = -self.fraud_cost  # Chi phí gian lận
        else:  # action == 0 and is_fraud == 0, đúng khi chấp nhận giao dịch hợp lệ
            reward = 0  # Không có chi phí
        
        # Chuyển sang giao dịch tiếp theo
        self.current_idx += 1
        done = self.current_idx >= len(self.transactions)
        
        next_state = self._get_state() if not done else None
        
        return next_state, reward, done, {}

4. Học liên hợp (Federated Learning)

Cho phép các tổ chức tài chính cùng nhau xây dựng mô hình phát hiện gian lận mà không cần chia sẻ dữ liệu nhạy cảm.

import numpy as np
from sklearn.metrics import classification_report

# Mô phỏng các client
class Client:
    def __init__(self, client_id, data, target):
        self.client_id = client_id
        self.data = data
        self.target = target
        self.model = None
    
    def train_local_model(self, model_class, **model_args):
        self.model = model_class(**model_args)
        self.model.fit(self.data, self.target)
        return self.model
    
    def evaluate_model(self, test_data, test_target):
        predictions = self.model.predict(test_data)
        return classification_report(test_target, predictions, output_dict=True)

# Hàm cập nhật mô hình liên hợp - Cách đơn giản là trung bình các trọng số
def federated_averaging(client_models, global_model_architecture):
    # Giả sử tất cả các mô hình đều có cùng kiến trúc và tham số
    if not client_models:
        return None
    
    # Lấy trọng số từ mô hình của client đầu tiên
    first_model = client_models[0]
    
    if hasattr(first_model, 'coef_'):  # Mô hình tuyến tính
        num_models = len(client_models)
        federated_coef = np.zeros_like(first_model.coef_)
        federated_intercept = 0
        
        for model in client_models:
            federated_coef += model.coef_
            federated_intercept += model.intercept_
        
        federated_coef /= num_models
        federated_intercept /= num_models
        
        # Tạo mô hình toàn cục
        global_model = global_model_architecture()
        global_model.coef_ = federated_coef
        global_model.intercept_ = federated_intercept
        
        return global_model
    
    # Đối với mô hình phức tạp hơn, cần triển khai cụ thể
    return None

Kết luận

Phát hiện gian lận là một lĩnh vực liên tục phát triển, nơi machine learning đóng vai trò quan trọng. Hệ thống phát hiện gian lận hiệu quả cần kết hợp nhiều kỹ thuật, từ việc xử lý dữ liệu mất cân bằng đến các thuật toán học máy tiên tiến và quy tắc kinh doanh thông minh.

Khi xây dựng hệ thống phát hiện gian lận, hãy nhớ:

1. Dữ liệu là nền tảng: Đầu tư vào việc thu thập và xử lý dữ liệu chất lượng cao
2. Kỹ thuật đặc trưng là quan trọng: Tạo các đặc trưng có giá trị phân biệt cao
3. Đánh giá đúng: Sử dụng các thước đo phù hợp với dữ liệu mất cân bằng
4. Cân bằng giữa phát hiện và trải nghiệm: Giảm thiểu false positives để không làm phiền khách hàng
5. Liên tục cập nhật: Đào tạo lại mô hình thường xuyên để đối phó với kỹ thuật gian lận mới

Bằng cách áp dụng các nguyên tắc này, bạn có thể xây dựng một hệ thống phát hiện gian lận hiệu quả, tiết kiệm chi phí đáng kể cho tổ chức của mình.