ml spam detector - ghdrako/doc_snipets GitHub Wiki

• https://github.com/ghdrako/doc_snipets/wiki/ml-nlp

Simple spam detector end2end

EDA

import pandas as pd
# Load spam dataset
emails = pd.read_csv("spam.csv")
# Display first records
print(emails.head())
# Display dataset dimensions
print(emails.shape)
# Display dataset information
print(emails.info())
# Check missing values
print(emails.isnull().sum())
# Display label counts
print(
    emails["label"].value_counts()
)
# Display percentages
print(
    emails["label"]
    .value_counts(normalize=True) * 100
)
import matplotlib.pyplot as plt
emails["label"].value_counts().plot(
    kind="bar"
)
plt.xlabel("Email Category")
plt.ylabel("Number of Emails")
plt.title("Spam and Ham Distribution")
plt.show()

Preprocesing

import re
def clean_text(text):
    text = text.lower()  # Converting Text to Lowercase
    text = re.sub(
        r"http\S+",
        "",
        text
    )
    text = re.sub(
        r"[^a-z\s]",
        "",
        text
    )
    # Removing Extra Spaces
    text = re.sub(
        r"\s+",
        " ",
        text
    ).strip()
    return text

emails["clean_message"] = (
    emails["message"]
    .apply(clean_text)
)
print(
    emails[
        ["message", "clean_message"]
    ].head()
)

Zbyt agresywne usuwanie cech spamu w preprocessingu:

• Wyrażenie re.sub(r"http\S+", "", text) całkowicie usuwa linki. W wykrywaniu spamu sama obecność linku jest jedną z najsilniejszych cech. Zamiast go kasować, lepiej zamienić go na token, np. re.sub(r"http\S+", " httplink ", text).
• Podobnie wyrażenie re.sub(r"[^a-z\s]", "", text) usuwa znaki walut ($, €, £) oraz wykrzykniki (!), które są typowymi wyznacznikami wiadomości phishingowych/spamowych.

Split

from sklearn.model_selection import train_test_split
X = emails["clean_message"]
y = emails["label"]
X_train, X_test, y_train, y_test = train_test_split(
    X,
    y,
    test_size=0.20,
    random_state=42,
    stratify=y
)

Feature extraction

from sklearn.feature_extraction.text import TfidfVectorizer
vectorizer = TfidfVectorizer(
    stop_words="english",
    max_features=5000
)
X_train_tfidf = vectorizer.fit_transform(
    X_train
)
X_test_tfidf = vectorizer.transform(
    X_test)

Treain

from sklearn.naive_bayes import MultinomialNB
model = MultinomialNB()
model.fit(
    X_train_tfidf,
    y_train
)
print("Training completed.")

Prediction

predictions = model.predict(
    X_test_tfidf
)
print(predictions[:10])

Evaluation

from sklearn.metrics import (
    accuracy_score,
    precision_score,
    recall_score,
    f1_score
)
print(
    "Accuracy:",
    accuracy_score(
        y_test,
        predictions
    )
)
print(
    "Precision:",
    precision_score(
        y_test,
        predictions,
        pos_label="spam"
    )
)
print(
    "Recall:",
    recall_score(
        y_test,
        predictions,
        pos_label="spam"
    )
)
print(
    "F1 Score:",
    f1_score(
        y_test,
        predictions,
        pos_label="spam"
    )
)


from sklearn.metrics import confusion_matrix
matrix = confusion_matrix(
    y_test,
    predictions,
    labels=[
        "ham",
        "spam"
    ]
)
print(matrix)


from sklearn.metrics import classification_report
print(
    classification_report(
        y_test,
        predictions
    )
)

# Although TF-IDF does not directly indicate feature importance, examining the vocabulary provides insight into the terms learned during training.
feature_names = (
    vectorizer.get_feature_names_out()
)
print(feature_names[:50])

Predicting New Emails

new_email = [
    "Congratulations! You have won a free vacation. Click here to claim your prize."
]
new_email_clean = [
    clean_text(text)
    for text in new_email
]
new_email_vector = vectorizer.transform(
    new_email_clean
)
prediction = model.predict(
    new_email_vector
)
probability = model.predict_proba(
    new_email_vector
)
print("Prediction:", prediction[0])
print(
    "Confidence:",
    probability.max()
)

Storing model and vectorizer to later use

import joblib
joblib.dump(
    model,
    "spam_model.pkl"
)
joblib.dump(
    vectorizer,
    "tfidf_vectorizer.pkl"
)

Building a Complete Pipeline

from sklearn.pipeline import Pipeline
pipeline = Pipeline([
    (
        "tfidf",
        TfidfVectorizer(
            preprocessor=clean_text,
            stop_words="english",
            max_features=5000
        )
    ),
    (
        "classifier",
        MultinomialNB()
    )
])
pipeline.fit(
    X_train,
    y_train
)
pipeline_prediction = pipeline.predict(
    new_email
)
print(pipeline_prediction)

Aplication

import re
import joblib
import pandas as pd
import streamlit as st

# 1. Ładowanie modelu i wektoryzatora
@st.cache_resource
def load_artifacts():
    model = joblib.load("spam_model.pkl")
    vectorizer = joblib.load("tfidf_vectorizer.pkl")
    return model, vectorizer

model, vectorizer = load_artifacts()

# 2. Funkcja czyszcząca
def clean_input(text):
    text = text.lower()
    text = re.sub(r"http\S+", "", text)
    text = re.sub(r"[^a-z\s]", "", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text

# 3. Interfejs predykcji
st.title("Spam Analyzer")
st.write("Classifying email using machine learning.")

email_input = st.text_area("Enter email text:")

if st.button("Analyze"):
    if email_input.strip() == "":
        st.warning("Please enter some text first.")
    else:
        cleaned_text = clean_input(email_input)
        vectorized_text = vectorizer.transform([cleaned_text])
        
        prediction = model.predict(vectorized_text)[0]
        probability = model.predict_proba(vectorized_text).max()
        
        if prediction == "spam":
            st.error(f"Prediction: **SPAM** (Confidence: {probability:.2%})")
        else:
            st.success(f"Prediction: **HAM** (Confidence: {probability:.2%})")

# 4. Sekcja podglądu danych
st.divider()

@st.cache_data
def load_data():
    return pd.read_csv("spam.csv")

try:
    emails = load_data()
    
    st.subheader("Dataset Overview")
    st.write(emails["label"].value_counts())
    
    st.subheader("Class Distribution")
    st.bar_chart(emails["label"].value_counts())
    
    st.subheader("Sample Messages")
    st.dataframe(emails["message", "label"](/ghdrako/doc_snipets/wiki/"message",-"label").head(10))
except FileNotFoundError:
    st.info("Plik 'spam.csv' nie został znaleziony do wyświetlenia podglądu danych.")

lunch app

streamlit run app.py

Rozbudowa

Jeden plik model i transformer razem:

import joblib
import pandas as pd
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline

emails = pd.read_csv("spam.csv")

# Definiujemy czyszczenie bezpośrednio w TfidfVectorizer
pipeline = Pipeline([
    ("tfidf", TfidfVectorizer(
        preprocessor=clean_input, 
        stop_words="english", 
        max_features=5000
    )),
    ("classifier", MultinomialNB())
])

# Uczymy cały proces od surowego tekstu do etykiety
pipeline.fit(emails["message"], emails["label"])

# Zapisujemy tylko JEDEN artefakt
joblib.dump(pipeline, "spam_pipeline.pkl")

i uzycie

import streamlit as st
import joblib

# Ładujemy jeden plik
@st.cache_resource
def load_model():
    return joblib.load("spam_pipeline.pkl")

pipeline = load_model()

email_input = st.text_area("Enter email text:")

if st.button("Analyze") and email_input.strip():
    # Pipeline sam czyści tekst, tworzy TF-IDF i klasyfikuje:
    prediction = pipeline.predict([email_input])[0]
    confidence = pipeline.predict_proba([email_input]).max()

    st.write(f"Wynik: {prediction} ({confidence:.2%})")

Walidacja krzyzowa:

import re
import joblib
import pandas as pd
from sklearn.model_selection import train_test_split, GridSearchCV
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline
from sklearn.metrics import classification_report

# 1. Funkcja czyszcząca tekst
def clean_text(text):
    text = text.lower()
    text = re.sub(r"http\S+", " httplink ", text)
    text = re.sub(r"[^a-z\s]", "", text)
    text = re.sub(r"\s+", " ", text).strip()
    return text

# 2. Wczytanie i podział danych
emails = pd.read_csv("spam.csv")
X_train, X_test, y_train, y_test = train_test_split(
    emails["message"],
    emails["label"],
    test_size=0.20,
    random_state=42,
    stratify=emails["label"]
)

# 3. Definicja bazowego Pipeline
pipeline = Pipeline([
    ("tfidf", TfidfVectorizer(preprocessor=clean_text)),
    ("classifier", MultinomialNB())
])

# 4. Siatka hiperparametrów (składnia: krok__parametr)
param_grid = {
    # Parametry TF-IDF
    "tfidf__max_features": [3000, 5000, 10000],
    "tfidf__ngram_range": [(1, 1), (1, 2)],        # pojedyncze słowa lub unigramy + bigramy
    "tfidf__stop_words": [None, "english"],
    
    # Parametry klasyfikatora Naive Bayes
    "classifier__alpha": [0.1, 0.5, 1.0]          # wygładzanie Laplace'a
}

# 5. Konfiguracja i uruchomienie GridSearchCV
grid_search = GridSearchCV(
    estimator=pipeline,
    param_grid=param_grid,
    cv=5,                          # 5-krotna stratyfikowana walidacja krzyżowa
    scoring="f1_macro",            # metryka dopasowana do niezbalansowanych klas
    n_jobs=-1,                     # równoległe obliczenia na wszystkich rdzeniach
    verbose=1
)

grid_search.fit(X_train, y_train)

# 6. Wyniki optymalizacji
print("Najlepsze parametry:", grid_search.best_params_)
print(f"Najlepszy wynik F1 (CV): {grid_search.best_score_:.4f}")

# 7. Ewaluacja na zbiorze testowym przy użyciu najlepszego modelu
best_pipeline = grid_search.best_estimator_
y_pred = best_pipeline.predict(X_test)
print("\nRaport klasyfikacji na zbiorze testowym:")
print(classification_report(y_test, y_pred))

# 8. Zapis zoptymalizowanego pipeline'u
joblib.dump(best_pipeline, "best_spam_pipeline.pkl")

Niestandardowy transformer:

Niestandardowy transformator (ang. custom transformer) w scikit-learn to własna klasa, która zachowuje się dokładnie tak jak wbudowane narzędzia biblioteki (np. TfidfVectorizer czy StandardScaler).

Tworzy się go, dziedzicząc po dwóch klasach bazowych:

• BaseEstimator – dodaje obsługę parametrów, klonowanie oraz współpracę z GridSearchCV (get_params(), set_params()).
• TransformerMixin – automatycznie dodaje metodę fit_transform(), więc wystarczy zaimplementować jedynie fit() i transform().

Dlaczego warto zamknąć czyszczenie w klasie zamiast zwykłej funkcji?

• Możliwość strojenia w GridSearchCV: Możesz sprawdzić, czy model radzi sobie lepiej z usuwaniem liczb vs. zamianą ich na tokeny.
• Hermetyzacja: Cała logika preprocessingu żyje wewnątrz pipeline'u, bez zewnętrznych luźnych funkcji.
• Serializacja: Zapisując pipeline przez joblib, zapisujesz całą klasę ze stanem i parametrami.

import re
import joblib
import pandas as pd
from sklearn.base import BaseEstimator, TransformerMixin
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.naive_bayes import MultinomialNB
from sklearn.pipeline import Pipeline


class TextCleaner(BaseEstimator, TransformerMixin):
    def __init__(self, replace_links=True, remove_numbers=True, lowercase=True):
        # Parametry muszą być przypisane do self o dokładnie takich samych nazwach
        self.replace_links = replace_links
        self.remove_numbers = remove_numbers
        self.lowercase = lowercase

    def fit(self, X, y=None):
        # Transformator bezstanowy – niczego się nie uczy z danych, więc tylko zwraca self
        return self

    def _clean_single_text(self, text):
        if not isinstance(text, str):
            text = str(text)

        if self.lowercase:
            text = text.lower()

        if self.replace_links:
            # Zamiast usuwać, zamieniamy na słowo kluczowe
            text = re.sub(r"http\S+|www\.\S+", " httplink ", text)

        if self.remove_numbers:
            text = re.sub(r"\d+", " ", text)

        # Usunięcie znaków specjalnych poza spacjami i literami
        text = re.sub(r"[^a-z\s]", " ", text)
        
        # Redukcja wielokrotnych spacji
        text = re.sub(r"\s+", " ", text).strip()
        return text

    def transform(self, X):
        # X może być listą, serią pandas lub tablicą numpy
        if isinstance(X, pd.Series):
            return X.apply(self._clean_single_text)
        return [self._clean_single_text(text) for text in X]

Użycie w pełnym Pipeline i GridSearch

pipeline = Pipeline([
    ("cleaner", TextCleaner()),
    ("tfidf", TfidfVectorizer(stop_words="english")),
    ("classifier", MultinomialNB())
])

# W GridSearchCV możesz testować wpływ poszczególnych kroków czyszczenia:
param_grid = {
    "cleaner__replace_links": [True, False],
    "cleaner__remove_numbers": [True, False],
    "tfidf__max_features": [3000, 5000],
    "classifier__alpha": [0.1, 1.0]
}

# Całość przyjmuje surowy tekst bez wcześniejszego przygotowania
emails = pd.read_csv("spam.csv")
pipeline.fit(emails["message"], emails["label"])

# Zapis do pojedynczego pliku
joblib.dump(pipeline, "spam_custom_pipeline.pkl")

Wektoryzacja (Word vs. Char n-grams)

• TfidfVectorizer over word or character n-grams is a strong baseline. (analyzer='char' lub 'char_wb').
• W zastosowaniach takich jak wykrywanie spamu czy analiza recenzji e-commerce n-gramy znakowe bez problemu radzą sobie z literówkami, celowym omijaniem filtrów (np. v1agra, f.r.e.e) oraz językami fleksyjnymi bez potrzeby budowania złożonych słowników.
• Kontrola rozmiaru: min_df (np. odrzucenie słów występujących rzadziej niż 2–5 razy) i max_df (odrzucenie słów obecnych w ponad 80–90% dokumentów, co działa jak dynamiczna lista stop-words) to znacznie lepsze praktyki niż sztywne obcinanie słownika samym max_features.

Redukcja wymiarowości (TruncatedSVD / LSA):

• Zwykłe PCA wymaga gęstej macierzy (dense), co przy dużym słowniku natychmiast doprowadzi do błędu braku pamięci (MemoryError / OOM).
• TruncatedSVD operuje bezpośrednio na macierzach rzadkich (scipy.sparse.csr_matrix), redukując przestrzeń np. z 50 000 wymiarów do 100–300 składowych semantycznych (Latent Semantic Analysis), co pozwala na użycie modeli wymagających gęstych danych (np. CatBoost, LightGBM, sieci neuronowe czy regresje logistyczne).

Czyszczenie danych (Umiar w preprocessingu)

• Agresywne podejścia (agresywny stemmer Portera, usuwanie wszystkich znaków interpunkcyjnych, twarde usuwanie stop-words) często niszczą istotny sygnał (np. negacje not good, znaki !, $, wielkie litery).
• Lematyzacja (NLTK/spaCy) drastycznie spowalnia proces przetwarzania (nawet 50–100x dłuższy czas przygotowania danych), a zysk na metrykach klasyfikacji w porównaniu z dobrze skonfigurowanym n-gramem TF-IDF bywa marginalny lub żaden.

Guardrails (Izolacja w Pipeline):

• Keep the vectorizer and SVD in a text sub-pipeline. This way the vocabulary is learned on training folds only, and transforms are identical at validation and inference.
• Unika wycieku danych (data leakage). Uczenie słownika na całym zbiorze (łącznie z danymi walidacyjnymi/testowymi) zafałszowuje metryki walidacji i prowadzi do błędów podczas wdrożenia na produkcji.

from sklearn.pipeline import Pipeline
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.decomposition import TruncatedSVD
from sklearn.linear_model import LogisticRegression

# Realizacja wszystkich 4 punktów w jednym sub-pipeline
text_pipeline = Pipeline([
    ("tfidf", TfidfVectorizer(
        ngram_range=(1, 2),        # n-gramy słowne
        min_df=3,                  # odrzucenie rzadkich szumów
        max_df=0.9,                # odrzucenie powszechnych słów (zamiast sztywnej listy stop-words)
        sublinear_tf=True          # skalowanie logarytmiczne częstości terminów
    )),
    ("lsa", TruncatedSVD(
        n_components=100,          # redukcja wymiarowości do 100 gęstych cech semantycznych
        random_state=42
    )),
    ("classifier", LogisticRegression(class_weight="balanced"))
])

Aby połączyć n-gramy słowne (przenoszące semantykę) z n-gramami znakowymi (odpornymi na literówki i podszywanie się pod słowa), używa się FeatureUnion. Łączy on równolegle wyniki obu wektoryzatorów w jedną szeroką macierz rzadką (sparse matrix).

import pandas as pd
from sklearn.model_selection import train_test_split
from sklearn.pipeline import Pipeline, FeatureUnion
from sklearn.feature_extraction.text import TfidfVectorizer
from sklearn.linear_model import LogisticRegression
from sklearn.metrics import classification_report

# 1. Wczytanie i podział danych
emails = pd.read_csv("spam.csv")

X_train, X_test, y_train, y_test = train_test_split(
    emails["message"],
    emails["label"],
    test_size=0.20,
    random_state=42,
    stratify=emails["label"]
)

# 2. Równoległa ekstrakcja cech słownych i znakowych
text_features = FeatureUnion([
    # Gałąź 1: N-gramy słowne (unigramy + bigramy)
    ("word_tfidf", TfidfVectorizer(
        analyzer="word",
        ngram_range=(1, 2),
        min_df=3,
        max_df=0.9,
        sublinear_tf=True
    )),
    
    # Gałąź 2: N-gramy znakowe w granicach słów (char_wb)
    # Wyłapuje podszycia typu: "v1agra", "f.r.e.e", "w!n"
    ("char_tfidf", TfidfVectorizer(
        analyzer="char_wb",
        ngram_range=(3, 5),        # fragmenty od 3 do 5 znaków
        min_df=5,
        max_df=0.95,
        sublinear_tf=True
    ))
],
# transformer_weights pozwala nadać różną wagę gałęziom (opcjonalnie)
transformer_weights={
    "word_tfidf": 1.0,
    "char_tfidf": 0.8
})

# 3. Pełny Pipeline (Ekstrakcja -> Klasyfikator)
pipeline = Pipeline([
    ("features", text_features),
    ("classifier", LogisticRegression(class_weight="balanced", max_iter=1000, random_state=42))
])

# 4. Trening i ewaluacja
pipeline.fit(X_train, y_train)
y_pred = pipeline.predict(X_test)

print(classification_report(y_test, y_pred))

Dlaczego analyzer="char_wb" zamiast zwykłego "char"?

• char tworzy n-gramy przechodzące przez spacje pomiędzy słowami (np. dla "free cash" n-gramem 4-znakowym będzie "ee c").
• char_wb (character within word boundaries) generuje n-gramy wyłącznie w obrębie pojedynczych słów, dodając spację na początku i końcu słowa (np. [ #fr, fre, ree, ee# ]). Jest to skuteczniejszy wariant, ponieważ ogranicza liczbę kombinacji i lepiej izoluje rdzenie słów z literówkami.