File size: 4,419 Bytes
7a0a647
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
 
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
"""

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