Pipeline Phát hiện Giao dịch Gian lận & Dị biệt Đa biến

Mục tiêu kiến thức

• Xây dựng mô hình Phát hiện dị biệt đa biến (Multivariate Anomaly Detection) dựa trên Khoảng cách Mahalanobis (Mahalanobis Distance).
• Hiểu được sự vượt trội của khoảng cách Mahalanobis so với khoảng cách Euclidean khi các đặc trưng tài chính (như số tiền giao dịch và tần suất giao dịch) có tương quan mạnh mẽ với nhau.
• Tính toán và nghịch đảo ma trận hiệp phương sai có điều quy chống suy biến (Regularized Covariance Matrix): Sigma_reg = Sigma + ε I.
• Xác định ngưỡng phân vị (Quantile Threshold) theo tỷ lệ ngoại lai kỳ vọng (contamination).

Mô tả bài toán

Trong hệ thống ngân hàng số, một giao dịch chuyển 50,000,000 VNĐ có thể là bình thường vào ban ngày tại chi nhánh quen thuộc, nhưng sẽ trở thành bất thường nghiêm trọng nếu diễn ra vào 3 giờ sáng từ một thiết bị mới. Việc chỉ kiểm tra từng đặc trưng độc lập bằng ngưỡng một chiều (Z-Score đơn biến) sẽ bỏ sót các mối tương quan này.

Khoảng cách bình phương Mahalanobis của vector giao dịch x ∈ R^D: D²(x) = (x - μ)^T Sigma_reg^-1 (x - μ)

Hãy xây dựng lớp:

class MultivariateAnomalyDetector:
    def __init__(self, contamination: float = 0.05, eps: float = 1e-6):
        ...
    def fit(self, X: np.ndarray) -> "MultivariateAnomalyDetector":
        ...
    def score_samples(self, X: np.ndarray) -> np.ndarray:
        ...
    def predict(self, X: np.ndarray) -> np.ndarray:
        ...

Yêu cầu chi tiết:

• Khởi tạo (__init__):
• contamination: Tỷ lệ phần trăm dự kiến các điểm ngoại lai trong tập huấn luyện (0.0 < contamination < 0.5). Nếu không thỏa mãn, ném ValueError("contamination must be in (0.0, 0.5)").
• eps: Hằng số điều quy ma trận ε > 0.
• Huấn luyện (fit):
• Nhận mảng 2D X ∈ R^{N × D} (N > D). Nếu X.ndim ≠ 2 hoặc N ≤ D, ném ValueError("X must be 2D with N > D").
• Tính vector trung bình: μ = 1N ∑_i=1^N x_i.
• Tính ma trận hiệp phương sai mẫu không thiên lệch (ddof=1):

Sigma = 1N - 1 (X - μ)^T (X - μ)

• Thêm thành phần điều quy đường chéo để đảm bảo ma trận xác định dương và khả nghịch:

Sigma_reg = Sigma + ε I_D

• Tính ma trận nghịch đảo: Sigma_inv = inv(Sigma_reg).
• Tính điểm D² cho toàn bộ tập huấn luyện X.
• Xác định ngưỡng phân vị τ: Ngưỡng bằng phân vị thứ (1 - contamination) × 100 của tập điểm huấn luyện:

threshold_ = np.percentile(train_scores, (1.0 - contamination) * 100.0).

• Trả về self.
• Tính điểm bất thường (score_samples):
• Nhận vào X ∈ R^{M × D}.
• Tính vector hóa khoảng cách bình phương Mahalanobis D²(x) cho từng dòng:

Δ = X - μ D²(x_i) = ∑_j=1^D (Δ Sigma_inv)_ij Δ_ij

• Trả về mảng 1D np.ndarray float64 độ dài M.
• Gán nhãn dự đoán (predict):
• Trả về mảng 1D kiểu boolean (bool_):

True nếu điểm số ≥ threshold_ (bất thường/gian lận), False nếu bình thường.

Ràng buộc

• Kích thước ma trận: 10 ≤ N ≤ 10,000, 2 ≤ D ≤ 20.
• Không dùng scipy.spatial.distance.mahalanobis hay sklearn.covariance. Toàn bộ tính bằng numpy.