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
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252

import numpy as np
from collections import Counter, defaultdict
from typing import List, Dict, Tuple
import math
 
class MultinomialNaiveBayes:
    """
    Complete implementation of Multinomial Naive Bayes from scratch.
    
    This implementation prioritizes clarity and understanding over
    computational efficiency. Every step is explicitly documented.
    """
    
    def __init__(self, alpha: float = 1.0):
        """
        Initialize the classifier.
        
        Args:
            alpha: Laplace smoothing parameter. 
                   α = 1 is add-one smoothing (Laplace)
                   α = 0 is no smoothing (pure MLE)
                   α < 1 is Lidstone smoothing
        """
        self.alpha = alpha
        
        # Model parameters (learned during fit)
        self.classes_: List[str] = []
        self.log_priors_: Dict[str, float] = {}        # log P(c)
        self.log_likelihoods_: Dict[str, Dict[str, float]] = {}  # log P(w|c)
        self.vocabulary_: set = set()
        self.vocab_size_: int = 0
        
        # Statistics for interpretability
        self.class_word_counts_: Dict[str, int] = {}
        
    def _tokenize(self, document: str) -> List[str]:
        """Simple whitespace tokenization with lowercasing."""
        return document.lower().split()
    
    def fit(self, documents: List[str], labels: List[str]) -> 'MultinomialNaiveBayes':
        """
        Fit the model using Maximum Likelihood Estimation with smoothing.
        
        The fitting process:
        1. Build vocabulary from all training documents
        2. Count class frequencies → compute log P(c)
        3. Count word-class co-occurrences → compute log P(w|c)
        
        Args:
            documents: List of text documents
            labels: List of class labels (same length as documents)
            
        Returns:
            self (fitted model)
        """
        n_docs = len(documents)
        
        # Step 1: Tokenize all documents and build vocabulary
        tokenized_docs = [self._tokenize(doc) for doc in documents]
        for tokens in tokenized_docs:
            self.vocabulary_.update(tokens)
        self.vocab_size_ = len(self.vocabulary_)
        
        # Step 2: Count class frequencies
        self.classes_ = sorted(set(labels))
        class_doc_counts = Counter(labels)
        
        # Compute log priors: log P(c) = log(N_c / N)
        self.log_priors_ = {
            c: math.log(class_doc_counts[c] / n_docs)
            for c in self.classes_
        }
        
        # Step 3: Count word occurrences per class
        # word_counts[class][word] = count of word in all docs of class
        word_counts = {c: defaultdict(int) for c in self.classes_}
        
        for tokens, label in zip(tokenized_docs, labels):
            for token in tokens:
                word_counts[label][token] += 1
        
        # Total words per class (for normalization)
        self.class_word_counts_ = {
            c: sum(word_counts[c].values())
            for c in self.classes_
        }
        
        # Step 4: Compute log-likelihoods with Laplace smoothing
        # log P(w|c) = log[(count(w,c) + α) / (total_c + α|V|)]
        self.log_likelihoods_ = {}
        
        for c in self.classes_:
            total_words_c = self.class_word_counts_[c]
            denominator = total_words_c + self.alpha * self.vocab_size_
            
            self.log_likelihoods_[c] = {
                word: math.log((word_counts[c][word] + self.alpha) / denominator)
                for word in self.vocabulary_
            }
        
        return self
    
    def predict_log_proba(self, document: str) -> Dict[str, float]:
        """
        Compute log-probabilities for each class.
        
        For each class c:
        log P(c|d) ∝ log P(c) + Σ_w count(w,d) * log P(w|c)
        
        Args:
            document: Text document to classify
            
        Returns:
            Dictionary mapping class names to log-probabilities
        """
        tokens = self._tokenize(document)
        word_counts = Counter(tokens)
        
        log_probs = {}
        
        for c in self.classes_:
            # Start with log prior
            log_prob = self.log_priors_[c]
            
            # Add log-likelihood for each word (weighted by count)
            for word, count in word_counts.items():
                if word in self.log_likelihoods_[c]:
                    log_prob += count * self.log_likelihoods_[c][word]
                else:
                    # Unknown word: use smoothing estimate
                    # P(unknown|c) ≈ α / (total_c + α|V|)
                    log_unknown = math.log(
                        self.alpha / 
                        (self.class_word_counts_[c] + self.alpha * self.vocab_size_)
                    )
                    log_prob += count * log_unknown
                    
            log_probs[c] = log_prob
            
        return log_probs
    
    def predict_proba(self, document: str) -> Dict[str, float]:
        """
        Compute normalized probabilities for each class.
        
        Uses the log-sum-exp trick for numerical stability:
        P(c|d) = exp(log P(c|d) - log Σ_c' exp(log P(c'|d)))
        """
        log_probs = self.predict_log_proba(document)
        
        # Log-sum-exp for numerical stability
        max_log_prob = max(log_probs.values())
        log_sum = max_log_prob + math.log(
            sum(math.exp(lp - max_log_prob) for lp in log_probs.values())
        )
        
        return {
            c: math.exp(lp - log_sum)
            for c, lp in log_probs.items()
        }
    
    def predict(self, document: str) -> str:
        """Predict the most likely class for a document."""
        log_probs = self.predict_log_proba(document)
        return max(log_probs, key=log_probs.get)
    
    def get_feature_importance(self, class_name: str, top_n: int = 10) -> List[Tuple[str, float]]:
        """
        Get the most important features (words) for a class.
        
        Returns words with highest P(w|c), which are most indicative
        of that class.
        """
        if class_name not in self.classes_:
            raise ValueError(f"Unknown class: {class_name}")
            
        word_probs = [
            (word, math.exp(log_prob))
            for word, log_prob in self.log_likelihoods_[class_name].items()
        ]
        return sorted(word_probs, key=lambda x: x[1], reverse=True)[:top_n]
 
 
# ============================================================
# COMPLETE EXAMPLE: Email Classification
# ============================================================
 
# Training data: emails labeled as spam or ham
training_emails = [
    # Spam examples
    ("FREE FREE FREE! You have won $1000 lottery prize! Claim now!", "spam"),
    ("Congratulations! You've been selected for a free vacation", "spam"),
    ("URGENT: Your account will be suspended unless you verify", "spam"),
    ("Make money fast! Work from home and earn $5000 daily", "spam"),
    ("Limited time offer! Get 90% discount on luxury watches now", "spam"),
    ("You are a winner! Click here to claim your prize money", "spam"),
    
    # Ham examples
    ("Meeting scheduled for tomorrow at 3pm in conference room B", "ham"),
    ("Please review the attached quarterly report before Friday", "ham"),
    ("Can we reschedule our lunch meeting to next week?", "ham"),
    ("The project deadline has been extended to December 15th", "ham"),
    ("Your order has been shipped and will arrive on Tuesday", "ham"),
    ("Here are the meeting notes from today's discussion", "ham"),
]
 
# Separate features and labels
docs = [email for email, _ in training_emails]
labels = [label for _, label in training_emails]
 
# Train the model
model = MultinomialNaiveBayes(alpha=1.0)
model.fit(docs, labels)
 
# Print model statistics
print("=" * 60)
print("MULTINOMIAL NAIVE BAYES MODEL SUMMARY")
print("=" * 60)
print(f"\nVocabulary size: {model.vocab_size_} words")
print(f"Classes: {model.classes_}")
 
print("\nClass Priors:")
for c in model.classes_:
    print(f"  P({c}) = exp({model.log_priors_[c]:.4f}) = {math.exp(model.log_priors_[c]):.4f}")
 
print("\nTop 5 Words per Class:")
for c in model.classes_:
    top_words = model.get_feature_importance(c, top_n=5)
    print(f"\n  {c.upper()}:")
    for word, prob in top_words:
        print(f"    {word:15s} P(w|{c}) = {prob:.4f}")
 
# Test on new emails
print("\n" + "=" * 60)
print("CLASSIFICATION RESULTS")
print("=" * 60)
 
test_emails = [
    "Claim your free prize now limited time offer winner",
    "The meeting has been rescheduled to Monday afternoon",
    "URGENT winner notification verify your account immediately",
    "Please find attached the updated project schedule"
]
 
for email in test_emails:
    prediction = model.predict(email)
    probs = model.predict_proba(email)
    
    print(f"\nEmail: '{email[:50]}...'")
    print(f"  Prediction: {prediction.upper()}")
    print(f"  P(spam|email) = {probs['spam']:.4f}")
    print(f"  P(ham|email)  = {probs['ham']:.4f}")

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

import numpy as np
import math
from typing import List, Tuple, Dict
 
def analyze_log_odds(model: 'MultinomialNaiveBayes', 
                     class_pos: str = 'spam',
                     class_neg: str = 'ham') -> Dict[str, float]:
    """
    Compute log-odds ratios for all words.
    
    LOR(w) = log[P(w|pos) / P(w|neg)]
    
    Returns dictionary mapping words to their LOR.
    """
    log_odds = {}
    
    for word in model.vocabulary_:
        log_p_pos = model.log_likelihoods_[class_pos][word]
        log_p_neg = model.log_likelihoods_[class_neg][word]
        log_odds[word] = log_p_pos - log_p_neg
        
    return log_odds
 
 
def get_most_discriminative_words(log_odds: Dict[str, float], 
                                   n: int = 10) -> Tuple[List, List]:
    """
    Get most discriminative words for each class.
    
    Returns:
        (words_for_spam, words_for_ham) - each is list of (word, LOR)
    """
    sorted_words = sorted(log_odds.items(), key=lambda x: x[1], reverse=True)
    
    spam_words = sorted_words[:n]  # Highest LOR = most spam-indicative
    ham_words = sorted_words[-n:][::-1]  # Lowest LOR = most ham-indicative
    
    return spam_words, ham_words
 
 
def explain_prediction(model: 'MultinomialNaiveBayes',
                       document: str,
                       class_pos: str = 'spam',
                       class_neg: str = 'ham') -> None:
    """
    Explain a classification decision by showing word contributions.
    
    For each word, shows how much it pushes toward spam vs ham.
    """
    from collections import Counter
    
    tokens = document.lower().split()
    word_counts = Counter(tokens)
    log_odds = analyze_log_odds(model, class_pos, class_neg)
    
    print(f"\nAnalyzing: '{document[:60]}...'")
    print("-" * 60)
    
    # Prior contribution
    prior_contribution = model.log_priors_[class_pos] - model.log_priors_[class_neg]
    print(f"Prior (log-odds): {prior_contribution:+.4f}")
    
    # Word contributions
    contributions = []
    for word, count in word_counts.items():
        if word in log_odds:
            contribution = count * log_odds[word]
            contributions.append((word, count, log_odds[word], contribution))
    
    # Sort by absolute contribution
    contributions.sort(key=lambda x: abs(x[3]), reverse=True)
    
    print("\nWord contributions (sorted by impact):")
    print(f"{'Word':15s} {'Count':>6s} {'LOR':>8s} {'Impact':>10s}")
    print("-" * 45)
    
    total_word_contribution = 0
    for word, count, lor, impact in contributions[:10]:
        direction = "→spam" if impact > 0 else "→ham"
        print(f"{word:15s} {count:>6d} {lor:>+8.3f} {impact:>+10.3f} {direction}")
        total_word_contribution += impact
    
    if len(contributions) > 10:
        remaining = sum(c[3] for c in contributions[10:])
        print(f"{'(other words)':15s} {'':>6s} {'':>8s} {remaining:>+10.3f}")
        total_word_contribution += remaining
    
    total_log_odds = prior_contribution + total_word_contribution
    print("-" * 45)
    print(f"{'TOTAL':15s} {'':>6s} {'':>8s} {total_log_odds:>+10.3f}")
    
    # Convert to probability
    prob_spam = 1 / (1 + math.exp(-total_log_odds))
    print(f"\nP(spam|document) = {prob_spam:.4f}")
    print(f"Prediction: {'SPAM' if total_log_odds > 0 else 'HAM'}")
 
 
# Example analysis (continuing from previous code)
# Assuming 'model' is already trained...
 
print("=" * 60)
print("LOG-ODDS RATIO ANALYSIS")
print("=" * 60)
 
# Analyze word discriminability
log_odds = analyze_log_odds(model, 'spam', 'ham')
spam_words, ham_words = get_most_discriminative_words(log_odds, n=5)
 
print("\nMost Spam-Indicative Words:")
print(f"{'Word':15s} {'LOR':>10s} {'Odds Ratio':>12s}")
print("-" * 40)
for word, lor in spam_words:
    print(f"{word:15s} {lor:>+10.3f} {math.exp(lor):>12.2f}x")
 
print("\nMost Ham-Indicative Words:")
print(f"{'Word':15s} {'LOR':>10s} {'Odds Ratio':>12s}")
print("-" * 40)
for word, lor in ham_words:
    print(f"{word:15s} {lor:>+10.3f} {math.exp(lor):>12.2f}x")
 
# Explain specific predictions
print("\n" + "=" * 60)
print("PREDICTION EXPLANATIONS")
print("=" * 60)
 
explain_prediction(model, "Free prize winner claim your lottery money now")
explain_prediction(model, "Meeting tomorrow to discuss the project deadline")

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

import numpy as np
from collections import Counter
import math
 
def demonstrate_length_effect():
    """
    Show how document length affects Multinomial NB predictions.
    """
    # Simplified vocabulary and probabilities
    vocab = ['free', 'money', 'meeting', 'report']
    
    # Log probabilities (simplified)
    log_p_spam = {'free': -2.0, 'money': -2.5, 'meeting': -5.0, 'report': -5.0}
    log_p_ham = {'free': -5.0, 'money': -5.0, 'meeting': -2.0, 'report': -2.5}
    
    # Two documents with same PROPORTIONS but different lengths
    short_doc = {'free': 1, 'money': 1, 'meeting': 1, 'report': 1}  # 4 words
    long_doc = {'free': 10, 'money': 10, 'meeting': 10, 'report': 10}  # 40 words
    
    def compute_log_likelihood(doc, log_probs):
        return sum(count * log_probs[word] for word, count in doc.items())
    
    # Compute likelihoods
    short_spam = compute_log_likelihood(short_doc, log_p_spam)
    short_ham = compute_log_likelihood(short_doc, log_p_ham)
    long_spam = compute_log_likelihood(long_doc, log_p_spam)
    long_ham = compute_log_likelihood(long_doc, log_p_ham)
    
    print("Without normalization:")
    print(f"  Short doc: spam={short_spam:.2f}, ham={short_ham:.2f}, diff={short_spam-short_ham:+.2f}")
    print(f"  Long doc:  spam={long_spam:.2f}, ham={long_ham:.2f}, diff={long_spam-long_ham:+.2f}")
    
    # With length normalization
    short_len = sum(short_doc.values())
    long_len = sum(long_doc.values())
    
    print("\nWith length normalization:")
    print(f"  Short doc: spam={short_spam/short_len:.2f}, ham={short_ham/short_len:.2f}")
    print(f"  Long doc:  spam={long_spam/long_len:.2f}, ham={long_ham/long_len:.2f}")
    print("  (Now the per-word scores are identical!)")
 
 
class LengthNormalizedMultinomialNB:
    """
    Multinomial Naive Bayes with length normalization.
    """
    
    def __init__(self, alpha: float = 1.0, normalize: str = 'none'):
        """
        Args:
            alpha: Smoothing parameter
            normalize: 'none', 'length', 'l2', or 'sublinear'
        """
        self.alpha = alpha
        self.normalize = normalize
        self.classes_ = []
        self.log_priors_ = {}
        self.log_likelihoods_ = {}
        self.vocabulary_ = set()
        self.vocab_size_ = 0
        
    def _normalize_counts(self, word_counts: dict) -> dict:
        """Apply normalization to word counts."""
        if self.normalize == 'none':
            return word_counts
            
        elif self.normalize == 'length':
            # Divide by document length (TF normalization)
            total = sum(word_counts.values())
            if total == 0:
                return word_counts
            return {w: c/total for w, c in word_counts.items()}
            
        elif self.normalize == 'l2':
            # L2 normalization
            norm = math.sqrt(sum(c**2 for c in word_counts.values()))
            if norm == 0:
                return word_counts
            return {w: c/norm for w, c in word_counts.items()}
            
        elif self.normalize == 'sublinear':
            # Sublinear TF: 1 + log(count) for count > 0
            return {w: (1 + math.log(c)) if c > 0 else 0 
                    for w, c in word_counts.items()}
        
        return word_counts
    
    def predict_log_proba(self, document: str) -> dict:
        """Predict with normalization applied to document."""
        tokens = document.lower().split()
        word_counts = Counter(tokens)
        
        # Apply normalization
        normalized_counts = self._normalize_counts(word_counts)
        
        log_probs = {}
        for c in self.classes_:
            log_prob = self.log_priors_[c]
            
            for word, count in normalized_counts.items():
                if word in self.log_likelihoods_[c]:
                    log_prob += count * self.log_likelihoods_[c][word]
                    
            log_probs[c] = log_prob
            
        return log_probs
 
 
# Demonstrate the effect
demonstrate_length_effect()