Spaces:

viethung21IT
/

toxic_comment_classification_NLP

Sleeping

toxic_comment_classification_NLP / model3.py

Nguyễn Việt Hùng

source

e909e6d verified 10 months ago

8.62 kB

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

	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