dl_model.py

"""
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