Upload dl_model.py

dl_model.py

@@ -0,0 +1,120 @@
+"""
+dl_model.py
+Deep Learning Module — Native Multi-Layer Perceptron (MLP) Classifier.
+Replacing PyTorch BiLSTM to support host environments without CPU AVX/AVX2 support.
+"""
+
+import os
+import re
+import numpy as np
+import pandas as pd
+from sklearn.base import BaseEstimator, ClassifierMixin
+from sklearn.feature_extraction.text import TfidfVectorizer
+from sklearn.pipeline import Pipeline
+from sklearn.preprocessing import MaxAbsScaler
+from sklearn.neural_network import MLPClassifier
+from sklearn.pipeline import FeatureUnion
+
+class DeepNewsClassifier(BaseEstimator, ClassifierMixin):
+    """
+    §3.3.2 Deep Learning Module — Native MLP Neural Network Classifier.
+    Exposes sklearn-standard methods (fit, predict, predict_proba)
+    so it works seamlessly in Pipeline, StackingClassifier, and joblib serialization.
+    """
+
+    def __init__(self, hidden_layer_sizes=(128, 64, 32), max_iter=30, 
+                 learning_rate_init=0.001, batch_size=32, random_state=42):
+        self.hidden_layer_sizes = hidden_layer_sizes
+        self.max_iter = max_iter
+        self.learning_rate_init = learning_rate_init
+        self.batch_size = batch_size
+        self.random_state = random_state
+        
+        self.pipeline = None
+        self.classes_ = np.array([0, 1])
+
+    def _extract_statements(self, X):
+        if isinstance(X, pd.DataFrame):
+            return X['statement'].fillna('').astype(str).tolist()
+        elif isinstance(X, pd.Series):
+            return X.fillna('').astype(str).tolist()
+        elif isinstance(X, np.ndarray):
+            if X.ndim > 1:
+                return X[:, 0].astype(str).tolist()
+            return X.astype(str).tolist()
+        elif isinstance(X, list):
+            return [str(item) for item in X]
+        return [str(X)]
+
+    def fit(self, X, y):
+        # Extract statements text
+        statements = self._extract_statements(X)
+        y = np.array(y).astype(int)
+
+        # Word-level TF-IDF
+        word_vectorizer = TfidfVectorizer(
+            analyzer='word',
+            ngram_range=(1, 2),
+            max_features=5000
+        )
+        
+        # Char-level TF-IDF (captures sequence sub-words/suffixes)
+        char_vectorizer = TfidfVectorizer(
+            analyzer='char',
+            ngram_range=(3, 5),
+            max_features=5000
+        )
+        
+        combined_features = FeatureUnion([
+            ('word_tfidf', word_vectorizer),
+            ('char_tfidf', char_vectorizer)
+        ])
+
+        # MLP Neural Network pipeline
+        self.pipeline = Pipeline([
+            ('features', combined_features),
+            ('scaler', MaxAbsScaler()),
+            ('mlp', MLPClassifier(
+                hidden_layer_sizes=self.hidden_layer_sizes,
+                max_iter=self.max_iter,
+                learning_rate_init=self.learning_rate_init,
+                batch_size=self.batch_size,
+                random_state=self.random_state,
+                activation='relu',
+                solver='adam',
+                verbose=True,
+                early_stopping=True,
+                validation_fraction=0.1
+            ))
+        ])
+
+        print(f"Training Native MLP Deep Neural Network Classifier: Hidden Layers={self.hidden_layer_sizes}, Epochs={self.max_iter}")
+        self.pipeline.fit(statements, y)
+        self.classes_ = self.pipeline.classes_
+        return self
+
+    def predict_proba(self, X):
+        if self.pipeline is None:
+            raise ValueError("Model has not been trained. Execute fit() first.")
+        statements = self._extract_statements(X)
+        return self.pipeline.predict_proba(statements)
+
+    def predict(self, X):
+        if self.pipeline is None:
+            raise ValueError("Model has not been trained. Execute fit() first.")
+        statements = self._extract_statements(X)
+        return self.pipeline.predict(statements)
+
+    def get_params(self, deep=True):
+        return {
+            'hidden_layer_sizes': self.hidden_layer_sizes,
+            'max_iter': self.max_iter,
+            'learning_rate_init': self.learning_rate_init,
+            'batch_size': self.batch_size,
+            'random_state': self.random_state
+        }
+
+    def set_params(self, **params):
+        for param, value in params.items():
+            setattr(self, param, value)
+        return self