File size: 8,621 Bytes
e909e6d | 1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 76 77 78 79 80 81 82 83 84 85 86 87 88 89 90 91 92 93 94 95 96 97 98 99 100 101 102 103 104 105 106 107 108 109 110 111 112 113 114 115 116 117 118 119 120 121 122 123 124 125 126 127 128 129 130 131 132 133 134 135 136 137 138 139 140 141 142 143 144 145 146 147 148 149 150 151 152 153 154 155 156 157 158 159 160 161 162 163 164 165 166 167 168 169 170 171 172 173 174 175 176 177 178 179 180 181 182 183 184 185 186 187 188 189 190 191 192 193 194 195 196 197 198 199 200 201 202 203 204 205 206 207 208 209 210 211 | import numpy as np
from scipy.sparse import issparse
from sklearn.base import BaseEstimator, ClassifierMixin # Giúp cho gridsearchcv hiểu tham số của mô hình
class ManualLogisticRegression:
def __init__(self, learning_rate=0.01, n_iters=1000, lambda_param=0.01, class_weight=None):
"""
:param learning_rate: Tốc độ học (alpha)
:param n_iters: Số vòng lặp
:param lambda_param: Hệ số L2 Regularization
:param class_weight: 'balanced' hoặc None. Nếu 'balanced', tự động tăng trọng số cho lớp thiểu số.
"""
self.lr = learning_rate
self.n_iters = n_iters
self.lambda_param = lambda_param # L2 penalty
self.class_weight = class_weight
self.weights = None
self.bias = None
self.loss_history = []
def decision_function(self, X):
"""Trả về giá trị thô (z = w*x + b) chưa qua Sigmoid"""
return X.dot(self.weights) + self.bias
def _sigmoid(self, z):
# Clip z để tránh tràn số (overflow) với exp
z = np.clip(z, -250, 250)
return 1 / (1 + np.exp(-z))
def _compute_sample_weights(self, y):
"""Tính trọng số cho từng mẫu dựa trên sự mất cân bằng dữ liệu"""
if self.class_weight != 'balanced':
return np.ones(len(y))
classes = np.unique(y)
n_samples = len(y)
n_classes = len(classes)
# Đếm số lượng từng class
counts = np.bincount(y.astype(int))
# Tính weight cho từng class (0 và 1)
class_weights_dict = {cls: n_samples / (n_classes * counts[cls]) for cls in classes}
# Gán weight tương ứng cho từng mẫu trong y
sample_weights = np.array([class_weights_dict[label] for label in y])
return sample_weights
def fit(self, X, y):
n_samples, n_features = X.shape
self.weights = np.zeros(n_features)
self.bias = 0
# Tính toán trọng số mẫu (Sample Weights) để xử lý Imbalance
sample_weights = self._compute_sample_weights(y)
for i in range(self.n_iters):
# 1. Forward pass: z = w*x + b
linear_model = X.dot(self.weights) + self.bias
# 2. Activation: y_pred = sigmoid(z)
y_predicted = self._sigmoid(linear_model)
# 3. Tính Gradient (Đạo hàm)
# Error thông thường: y_pred - y
# Error có trọng số: (y_pred - y) * sample_weight
error = (y_predicted - y) * sample_weights
# Gradient cho weights (có thêm L2 Regularization term: lambda * w)
# X.T.dot(error) rất nhanh với ma trận thưa
dw = (1 / n_samples) * (X.T.dot(error) + (self.lambda_param * self.weights))
# Gradient cho bias (Bias thường không áp dụng L2)
db = (1 / n_samples) * np.sum(error)
# 4. Update parameters (Gradient Descent)
self.weights -= self.lr * dw
self.bias -= self.lr * db
# (Optional) Tính Loss để theo dõi
if i % 100 == 0:
# Binary Cross Entropy Loss + L2 Term
# Clip y_predicted để tránh log(0)
epsilon = 1e-15
y_pred_safe = np.clip(y_predicted, epsilon, 1 - epsilon)
base_loss = -np.mean(sample_weights * (y * np.log(y_pred_safe) + (1 - y) * np.log(1 - y_pred_safe)))
l2_loss = (self.lambda_param / (2 * n_samples)) * np.sum(self.weights ** 2)
self.loss_history.append(base_loss + l2_loss)
def predict_proba(self, X):
linear_model = X.dot(self.weights) + self.bias
return self._sigmoid(linear_model)
def predict(self, X, threshold=0.5):
y_predicted_cls = [1 if i > threshold else 0 for i in self.predict_proba(X)]
return np.array(y_predicted_cls)
def print_weights(self, feature_names=None, top_n=10):
"""
In ra Bias và Weights.
Nếu có feature_names, sẽ in ra top_n đặc trưng đóng góp tích cực/tiêu cực nhất.
"""
print(f" Bias (Intercept): {self.bias:.4f}")
if feature_names is None:
# Nếu không có tên đặc trưng, in dạng array rút gọn
print(f" Weights (shape {self.weights.shape}):")
print(f" {self.weights}")
else:
# Nếu có tên đặc trưng, in top features
if len(feature_names) != len(self.weights):
print(" [Cảnh báo] Độ dài feature_names không khớp với số lượng weights.")
return
# Sắp xếp index của weights
sorted_indices = np.argsort(self.weights)
print(f" Top {top_n} từ khóa quan trọng nhất (Positive - Cho lớp 1):")
for idx in sorted_indices[-top_n:][::-1]:
print(f" {feature_names[idx]}: {self.weights[idx]:.4f}")
print(f" Top {top_n} từ khóa tiêu cực nhất (Negative - Cho lớp 0):")
for idx in sorted_indices[:top_n]:
print(f" {feature_names[idx]}: {self.weights[idx]:.4f}")
# --- Class One Vs Rest để xử lý Multi-label ---
class ManualLogisticRegressionOneVsRest(BaseEstimator, ClassifierMixin):
def __init__(self, learning_rate=0.01, n_iters=1000, lambda_param=0.01):
"""
Khai báo tham số tường minh để GridSearchCV có thể nhận diện.
"""
self.learning_rate = learning_rate
self.n_iters = n_iters
self.lambda_param = lambda_param
self.models = []
self.labels = []
def fit(self, X, Y):
# Chuyển đổi Y thành numpy array nếu là DataFrame
if hasattr(Y, 'columns'):
self.labels = Y.columns
Y_values = Y.values
else:
self.labels = range(Y.shape[1])
Y_values = Y
n_labels = Y_values.shape[1]
self.models = []
# Gom nhóm tham số để truyền vào model con
# Tại đây ta dùng các tham số self.learning_rate v.v...
base_params = {
'learning_rate': self.learning_rate,
'n_iters': self.n_iters,
'lambda_param': self.lambda_param
}
print(f"Bắt đầu huấn luyện {n_labels} mô hình Binary với lr={self.learning_rate}, lambda={self.lambda_param}...")
for i in range(n_labels):
y_col = Y_values[:, i]
# Tạo model con với tham số hiện tại
model = ManualLogisticRegression(class_weight='balanced', **base_params)
model.fit(X, y_col)
self.models.append(model)
return self
def predict_proba(self, X):
"""Trả về ma trận xác suất (n_samples, n_labels)"""
preds = []
for model in self.models:
# Lấy xác suất của lớp Positive (1)
preds.append(model.predict_proba(X))
# Transpose để có dạng (n_samples, n_labels)
return np.array(preds).T
def predict(self, X, threshold=0.5):
probs = self.predict_proba(X)
return (probs > threshold).astype(int)
def inspect_all_models(self, feature_names=None, top_n=5):
"""
In thông tin trọng số cho từng nhãn.
:param feature_names: List tên các đặc trưng (từ TF-IDF hoặc CountVectorizer)
:param top_n: Số lượng đặc trưng quan trọng muốn hiển thị cho mỗi nhãn
"""
print("\n" + "="*40)
print("CHI TIẾT TRỌNG SỐ CÁC MÔ HÌNH CON")
print("="*40)
for i, model in enumerate(self.models):
label_name = self.labels[i]
print(f"\n NHÃN: {label_name}")
print("-" * 20)
model.print_weights(feature_names=feature_names, top_n=top_n)
def decision_function(self, X):
"""
Trả về ma trận Logits (n_samples, n_labels)
Mỗi cột tương ứng với điểm số thô của một nhãn.
"""
logits = []
for model in self.models:
# Gọi hàm decision_function của từng model con
logits.append(model.decision_function(X))
# Transpose để có dạng (n_samples, n_labels)
return np.array(logits).T |